package network import ( "context" "errors" "fmt" "html" "log/slog" "strings" 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(ctx context.Context, name, adminHumanId string) (*Network, error) GetByID(ctx context.Context, id string) (*Network, error) 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(ctx context.Context) ([]*Network, 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 } func NewService(pool *pgxpool.Pool, aeroSvc pbaero.PrimaryClient, billingSvc billing.Service) Service { return &serviceImpl{ pool: pool, repo: newRepository(pool), aeroSvc: aeroSvc, billingSvc: billingSvc, } } 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") } return 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 }) } func (s *serviceImpl) RemoveMember(ctx context.Context, networkID, humanId string) error { if humanId == "" { return fmt.Errorf("invalid humanId") } return s.mutateMembers(ctx, networkID, func(tx pgx.Tx) error { return s.repo.removeMember(ctx, tx, networkID, humanId) }) } // 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) 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") } return 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) }) } 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(`
You've been invited to join %s on Flowy.llink
Download the app to accept your invitation and connect with your team.
Download Flowy.llink
`, safeName, downloadURL) }