package network import ( "context" "errors" "fmt" "strings" "github.com/flowy-live/llink/internal/utils" "github.com/jackc/pgx/v5/pgxpool" ) var ErrNotFound = errors.New("network not found") var ErrInvalidName = errors.New("name cannot be empty") var ErrCapacityExceeded = errors.New("active stream capacity exceeded") type Service interface { // Create creates a network and adds adminEmail as the first member. Returns ErrInvalidName if name is empty. Create(ctx context.Context, name, adminEmail 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(ctx context.Context, networkID string, emails []string) error RemoveMember(ctx context.Context, networkID, email string) error ListForEmail(ctx context.Context, email string) ([]*Network, error) IsMember(ctx context.Context, networkID, email string) (bool, error) // SetOpenStreamCapacity sets the max open streams for a network. Returns ErrNotFound. SetOpenStreamCapacity(ctx context.Context, networkID string, capacity int) error // IncrementOpenStreamCount returns ErrNotFound or ErrCapacityExceeded. IncrementOpenStreamCount(ctx context.Context, networkID string) error // DecrementOpenStreamCount returns ErrNotFound. DecrementOpenStreamCount(ctx context.Context, networkID string) error } type serviceImpl struct { repo repository } func NewService(pool *pgxpool.Pool) Service { return &serviceImpl{repo: newRepository(pool)} } func (s *serviceImpl) Create(ctx context.Context, name, adminEmail string) (*Network, error) { name = strings.TrimSpace(name) if name == "" { return nil, ErrInvalidName } adminEmail, err := utils.NormalizeEmail(adminEmail) if err != nil { return nil, err } network, err := s.repo.create(ctx, name, adminEmail) if err != nil { return nil, err } err = s.AddMembers(ctx, network.ID, []string{adminEmail}) if 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, emails []string) error { for _, email := range emails { normalized, err := utils.NormalizeEmail(email) if err != nil { return fmt.Errorf("invalid email: %w", err) } if err := s.repo.addMember(ctx, networkID, normalized); err != nil { return err } } return nil } func (s *serviceImpl) RemoveMember(ctx context.Context, networkID, email string) error { email, err := utils.NormalizeEmail(email) if err != nil { return fmt.Errorf("invalid email: %w", err) } return s.repo.removeMember(ctx, networkID, email) } func (s *serviceImpl) ListForEmail(ctx context.Context, email string) ([]*Network, error) { email, err := utils.NormalizeEmail(email) if err != nil { return nil, fmt.Errorf("invalid email: %w", err) } return s.repo.getNetworksForEmail(ctx, email) } func (s *serviceImpl) IsMember(ctx context.Context, networkID, email string) (bool, error) { email, err := utils.NormalizeEmail(email) if err != nil { return false, err } return s.repo.isMember(ctx, networkID, email) } func (s *serviceImpl) SetOpenStreamCapacity(ctx context.Context, networkID string, capacity int) error { if capacity < 0 { capacity = 0 } err := s.repo.setOpenStreamCapacity(ctx, networkID, capacity) if errors.Is(err, errNotFound) { return ErrNotFound } return err } func (s *serviceImpl) IncrementOpenStreamCount(ctx context.Context, networkID string) error { err := s.repo.incrementOpenStreamCount(ctx, networkID) if errors.Is(err, errNotFound) { return ErrNotFound } if errors.Is(err, errCapacityExceeded) { return ErrCapacityExceeded } return err } func (s *serviceImpl) DecrementOpenStreamCount(ctx context.Context, networkID string) error { err := s.repo.decrementOpenStreamCount(ctx, networkID) if errors.Is(err, errNotFound) { return ErrNotFound } return err }