llink/go/internal/network/service.go

package network

import (
	"context"
	"errors"
	"fmt"
	"html"
	"log/slog"
	"strings"

	"cloud.google.com/go/firestore"
	pbaero "github.com/flowy-live/llink/genproto/aero"
	"github.com/flowy-live/llink/internal/billing"
	"github.com/flowy-live/llink/internal/utils"
	"github.com/jackc/pgx/v5"
	"github.com/jackc/pgx/v5/pgxpool"
	"slices"
)

var ErrNotFound = errors.New("network not found")
var ErrInvalidName = errors.New("name cannot be empty")
var ErrCapacityExceeded = errors.New("active stream capacity exceeded")
var ErrInvalidRetentionHours = errors.New("message retention hours must be between 24 and 336")

type Service interface {
	// Create creates a network and adds adminHumanId as the first member. Returns ErrInvalidName if name is empty.
	Create(ctx context.Context, name, adminHumanId string) (*Network, error)
	// GetByID returns ErrNotFound if network doesn't exist.
	GetByID(ctx context.Context, id string) (*Network, error)
	// SetName returns ErrNotFound or ErrInvalidName.
	SetName(ctx context.Context, id, name string) error
	// AddMembers inserts members and syncs the new seat count to billing
	// atomically; a Stripe failure rolls the insert back.
	AddMembers(ctx context.Context, networkID string, humanIds []string) error
	RemoveMember(ctx context.Context, networkID, humanId string) error
	CountSeats(ctx context.Context, networkID string) (int, error)
	ListForHuman(ctx context.Context, humanId string) ([]*Network, error)
	IsMember(ctx context.Context, networkID, humanId string) (bool, error)
	// ListAll returns all networks with their members
	ListAll(ctx context.Context) ([]*Network, error)
	// ListAllMemberships returns humanId -> networkIds for every human with at
	// least one membership. Humans with zero memberships are absent from the map.
	// Used by the membership reconciler to diff the Firestore mirror.
	ListAllMemberships(ctx context.Context) (map[string][]string, error)

	// Invitations (email-based, for users who haven't registered yet)
	InviteByEmail(ctx context.Context, networkID string, emails []string) error
	ListInvitationsForEmail(ctx context.Context, email string) ([]*Invitation, error)
	ListInvitationsForNetwork(ctx context.Context, networkID string) ([]*Invitation, error)
	AcceptInvitation(ctx context.Context, networkID, email, humanId string) error
	RevokeInvitation(ctx context.Context, networkID, email string) error
}

type serviceImpl struct {
	pool       *pgxpool.Pool
	repo       repository
	aeroSvc    pbaero.PrimaryClient
	billingSvc billing.Service
	// fs mirrors network membership into humans/{humanId}.networks so
	// Firestore security rules can check membership at rule-eval time.
	// May be nil in services that never mutate membership (pusher).
	fs *firestore.Client
}

func NewService(pool *pgxpool.Pool, aeroSvc pbaero.PrimaryClient, billingSvc billing.Service, fs *firestore.Client) Service {
	return &serviceImpl{
		pool:       pool,
		repo:       newRepository(pool),
		aeroSvc:    aeroSvc,
		billingSvc: billingSvc,
		fs:         fs,
	}
}

func (s *serviceImpl) Create(ctx context.Context, name, adminHumanId string) (*Network, error) {
	name = strings.TrimSpace(name)
	if name == "" {
		return nil, ErrInvalidName
	}

	network, err := s.repo.create(ctx, name, adminHumanId)
	if err != nil {
		return nil, err
	}

	if err := s.AddMembers(ctx, network.ID, []string{adminHumanId}); err != nil {
		return nil, err
	}

	return network, nil
}

func (s *serviceImpl) GetByID(ctx context.Context, id string) (*Network, error) {
	n, err := s.repo.getByID(ctx, id)
	if errors.Is(err, errNotFound) {
		return nil, ErrNotFound
	}
	return n, err
}

func (s *serviceImpl) SetName(ctx context.Context, id, name string) error {
	name = strings.TrimSpace(name)
	if name == "" {
		return ErrInvalidName
	}

	err := s.repo.updateName(ctx, id, name)
	if errors.Is(err, errNotFound) {
		return ErrNotFound
	}
	return err
}

func (s *serviceImpl) AddMembers(ctx context.Context, networkID string, humanIds []string) error {
	if slices.Contains(humanIds, "") {
		return fmt.Errorf("invalid humanId")
	}
	if err := s.mutateMembers(ctx, networkID, func(tx pgx.Tx) error {
		for _, humanId := range humanIds {
			if err := s.repo.addMember(ctx, tx, networkID, humanId); err != nil {
				return err
			}
		}
		return nil
	}); err != nil {
		return err
	}

	for _, humanId := range humanIds {
		s.mirrorAddMembership(ctx, humanId, networkID)
	}
	return nil
}

func (s *serviceImpl) RemoveMember(ctx context.Context, networkID, humanId string) error {
	if humanId == "" {
		return fmt.Errorf("invalid humanId")
	}
	if err := s.mutateMembers(ctx, networkID, func(tx pgx.Tx) error {
		return s.repo.removeMember(ctx, tx, networkID, humanId)
	}); err != nil {
		return err
	}

	s.mirrorRemoveMembership(ctx, humanId, networkID)
	return nil
}

// mirrorAddMembership / mirrorRemoveMembership keep the Firestore membership
// mirror (humans/{humanId}.networks) in sync with Postgres. Called after the
// Postgres transaction commits. Failures are logged but not returned: Postgres
// is the source of truth and the reconciler will heal drift. A missing fs
// client (pusher) no-ops.
func (s *serviceImpl) mirrorAddMembership(ctx context.Context, humanId, networkID string) {
	if s.fs == nil {
		return
	}
	_, err := s.fs.Collection("humans").Doc(humanId).Set(ctx, map[string]any{
		"networks":   firestore.ArrayUnion(networkID),
		"updated_at": firestore.ServerTimestamp,
	}, firestore.MergeAll)
	if err != nil {
		slog.Error("firestore mirror add failed", "error", err, "humanId", humanId, "networkID", networkID)
	}
}

func (s *serviceImpl) mirrorRemoveMembership(ctx context.Context, humanId, networkID string) {
	if s.fs == nil {
		return
	}
	_, err := s.fs.Collection("humans").Doc(humanId).Set(ctx, map[string]any{
		"networks":   firestore.ArrayRemove(networkID),
		"updated_at": firestore.ServerTimestamp,
	}, firestore.MergeAll)
	if err != nil {
		slog.Error("firestore mirror remove failed", "error", err, "humanId", humanId, "networkID", networkID)
	}
}

// mutateMembers runs fn in a tx, recounts seats, calls billing.SyncSeats,
// and commits. Any error rolls the membership change back.
func (s *serviceImpl) mutateMembers(ctx context.Context, networkID string, fn func(pgx.Tx) error) error {
	tx, err := s.pool.Begin(ctx)
	if err != nil {
		return fmt.Errorf("begin tx: %w", err)
	}
	defer tx.Rollback(ctx)

	if err := fn(tx); err != nil {
		return err
	}

	seats, err := s.repo.countSeats(ctx, tx, networkID)
	if err != nil {
		return fmt.Errorf("count seats: %w", err)
	}

	if err := s.billingSvc.SyncSeats(ctx, networkID, seats); err != nil {
		return fmt.Errorf("sync billing seats: %w", err)
	}

	if err := tx.Commit(ctx); err != nil {
		return fmt.Errorf("commit tx: %w", err)
	}
	return nil
}

func (s *serviceImpl) CountSeats(ctx context.Context, networkID string) (int, error) {
	return s.repo.countSeats(ctx, s.pool, networkID)
}

func (s *serviceImpl) ListForHuman(ctx context.Context, humanId string) ([]*Network, error) {
	if humanId == "" {
		return nil, fmt.Errorf("invalid humanId")
	}
	return s.repo.getNetworksForHuman(ctx, humanId)
}

func (s *serviceImpl) IsMember(ctx context.Context, networkID, humanId string) (bool, error) {
	if humanId == "" {
		return false, fmt.Errorf("invalid humanId")
	}
	return s.repo.isMember(ctx, networkID, humanId)
}

func (s *serviceImpl) ListAll(ctx context.Context) ([]*Network, error) {
	return s.repo.listAll(ctx)
}

func (s *serviceImpl) ListAllMemberships(ctx context.Context) (map[string][]string, error) {
	return s.repo.listAllMemberships(ctx)
}

func (s *serviceImpl) InviteByEmail(ctx context.Context, networkID string, emails []string) error {
	network, err := s.repo.getByID(ctx, networkID)
	if err != nil {
		return ErrNotFound
	}

	for _, email := range emails {
		normalized, err := utils.NormalizeEmail(email)
		if err != nil {
			return fmt.Errorf("invalid email %q: %w", email, err)
		}
		if err := s.repo.createInvitation(ctx, networkID, normalized); err != nil {
			return err
		}

		_, err = s.aeroSvc.ShootEmail(ctx, &pbaero.ShootEmailRequest{
			ToEmails: []string{email},
			Subject:  fmt.Sprintf("Invitation to Join %s on Flowy.llink", network.Name),
			TemplateData: &pbaero.ShootEmailRequest_SimpleHtmlData{
				SimpleHtmlData: &pbaero.SimpleHtmlData{
					Html: buildInvitationHTML(network.Name),
				},
			},
		})
		if err != nil {
			slog.Warn("unable to send email notification", "email", email, "network", network.Name)
		}
	}
	return nil
}

func (s *serviceImpl) ListInvitationsForEmail(ctx context.Context, email string) ([]*Invitation, error) {
	normalized, err := utils.NormalizeEmail(email)
	if err != nil {
		return nil, fmt.Errorf("invalid email: %w", err)
	}
	return s.repo.getInvitationsByEmail(ctx, normalized)
}

func (s *serviceImpl) ListInvitationsForNetwork(ctx context.Context, networkID string) ([]*Invitation, error) {
	return s.repo.getInvitationsByNetwork(ctx, networkID)
}

func (s *serviceImpl) AcceptInvitation(ctx context.Context, networkID, email, humanId string) error {
	normalized, err := utils.NormalizeEmail(email)
	if err != nil {
		return fmt.Errorf("invalid email: %w", err)
	}
	if humanId == "" {
		return fmt.Errorf("invalid humanId")
	}

	if err := s.mutateMembers(ctx, networkID, func(tx pgx.Tx) error {
		err := s.repo.deleteInvitation(ctx, tx, networkID, normalized)
		if err != nil {
			return err
		}
		return s.repo.addMember(ctx, tx, networkID, humanId)
	}); err != nil {
		return err
	}

	s.mirrorAddMembership(ctx, humanId, networkID)
	return nil
}

func (s *serviceImpl) RevokeInvitation(ctx context.Context, networkID, email string) error {
	normalized, err := utils.NormalizeEmail(email)
	if err != nil {
		return fmt.Errorf("invalid email: %w", err)
	}
	return s.repo.deleteInvitation(ctx, s.pool, networkID, normalized)
}

func buildInvitationHTML(networkName string) string {
	const downloadURL = "https://flowylabs.ai/llink"
	safeName := html.EscapeString(networkName)
	return fmt.Sprintf(`<!DOCTYPE html>
<html>
<body style="margin:0;padding:0;background-color:#f5f5f7;font-family:-apple-system,BlinkMacSystemFont,'Segoe UI',Roboto,sans-serif;">
  <table role="presentation" width="100%%" cellspacing="0" cellpadding="0" border="0" style="background-color:#f5f5f7;padding:48px 16px;">
    <tr>
      <td align="center">
        <table role="presentation" width="100%%" cellspacing="0" cellpadding="0" border="0" style="max-width:480px;background-color:#ffffff;border-radius:12px;padding:40px;">
          <tr>
            <td style="font-size:22px;font-weight:600;color:#111111;padding-bottom:16px;">
              You've been invited to join %s on Flowy.llink
            </td>
          </tr>
          <tr>
            <td style="font-size:15px;line-height:1.5;color:#444444;padding-bottom:32px;">
              Download the app to accept your invitation and connect with your team.
            </td>
          </tr>
          <tr>
            <td>
              <a href="%s" style="display:inline-block;background-color:#111111;color:#ffffff;text-decoration:none;font-size:15px;font-weight:500;padding:12px 24px;border-radius:8px;">
                Download Flowy.llink
              </a>
            </td>
          </tr>
        </table>
      </td>
    </tr>
  </table>
</body>
</html>`, safeName, downloadURL)
}