refactor: unit testable units and cleaner dep injection

go/internal/network/reader.go

@@ -0,0 +1,94 @@
+package network
+
+import (
+	"context"
+	"errors"
+	"fmt"
+
+	"github.com/flowy-live/llink/internal/utils"
+	"github.com/jackc/pgx/v5/pgxpool"
+)
+
+type Reader interface {
+	// GetByID returns ErrNotFound if network doesn't exist.
+	GetByID(ctx context.Context, id string) (*Network, error)
+	// ListForHuman returns ErrInvalidHumanId if humanId is empty.
+	ListForHuman(ctx context.Context, humanId string) ([]*Network, error)
+	// IsMember returns ErrInvalidHumanId if humanId is empty.
+	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)
+
+	CountSeats(ctx context.Context, networkID string) (int, error)
+	// ListInvitationsForEmail returns ErrInvalidEmail if normalization fails.
+	ListInvitationsForEmail(ctx context.Context, email string) ([]*Invitation, error)
+	ListInvitationsForNetwork(ctx context.Context, networkID string) ([]*Invitation, error)
+}
+
+type readerImpl struct {
+	pool *pgxpool.Pool
+	repo repository
+}
+
+// newReader returns the concrete reader. Used by NewService to embed without
+// going through the Reader interface (which would hide pool/repo).
+func newReader(pool *pgxpool.Pool) *readerImpl {
+	return &readerImpl{
+		pool: pool,
+		repo: newRepository(pool),
+	}
+}
+
+func NewReader(pool *pgxpool.Pool) Reader {
+	return newReader(pool)
+}
+
+func (r *readerImpl) GetByID(ctx context.Context, id string) (*Network, error) {
+	n, err := r.repo.getByID(ctx, id)
+	if errors.Is(err, errNotFound) {
+		return nil, ErrNotFound
+	}
+	return n, err
+}
+
+func (r *readerImpl) ListForHuman(ctx context.Context, humanId string) ([]*Network, error) {
+	if humanId == "" {
+		return nil, ErrInvalidHumanId
+	}
+	return r.repo.getNetworksForHuman(ctx, humanId)
+}
+
+func (r *readerImpl) IsMember(ctx context.Context, networkID, humanId string) (bool, error) {
+	if humanId == "" {
+		return false, ErrInvalidHumanId
+	}
+	return r.repo.isMember(ctx, networkID, humanId)
+}
+
+func (r *readerImpl) ListAll(ctx context.Context) ([]*Network, error) {
+	return r.repo.listAll(ctx)
+}
+
+func (r *readerImpl) ListAllMemberships(ctx context.Context) (map[string][]string, error) {
+	return r.repo.listAllMemberships(ctx)
+}
+
+func (r *readerImpl) CountSeats(ctx context.Context, networkID string) (int, error) {
+	return r.repo.countSeats(ctx, r.pool, networkID)
+}
+
+func (r *readerImpl) ListInvitationsForEmail(ctx context.Context, email string) ([]*Invitation, error) {
+	normalized, err := utils.NormalizeEmail(email)
+	if err != nil {
+		return nil, fmt.Errorf("%w: %w", ErrInvalidEmail, err)
+	}
+	return r.repo.getInvitationsByEmail(ctx, normalized)
+}
+
+func (r *readerImpl) ListInvitationsForNetwork(ctx context.Context, networkID string) ([]*Invitation, error) {
+	return r.repo.getInvitationsByNetwork(ctx, networkID)
+}

go/internal/network/repository.go

@@ -33,8 +33,6 @@ func newNetworkID() (networkID, error) {
 	return typeid.New[networkID]()
 }
 
-var errCapacityExceeded = errors.New("capacity exceeded")
-
 type repository interface {
 	create(ctx context.Context, name, adminHumanId string) (*Network, error)
 	getByID(ctx context.Context, id string) (*Network, error)
@@ -47,6 +45,9 @@ type repository interface {
 	getNetworksForHuman(ctx context.Context, humanId string) ([]*Network, error)
 	isMember(ctx context.Context, networkID, humanId string) (bool, error)
 	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.
 	listAllMemberships(ctx context.Context) (map[string][]string, error)
 
 	// Invitations
@@ -233,9 +234,6 @@ func (r *repositoryImpl) isMember(ctx context.Context, networkID, humanId string
 	return isMember, err
 }
 
-// listAllMemberships returns humanId -> networkIds for every human with at
-// least one membership. Humans with zero memberships are absent from the map;
-// callers layer them in separately.
 func (r *repositoryImpl) listAllMemberships(ctx context.Context) (map[string][]string, error) {
 	rows, err := r.pool.Query(ctx, `SELECT human_id, network_id FROM network_members`)
 	if err != nil {

go/internal/network/service.go

@@ -9,10 +9,10 @@ import (
 	"slices"
 	"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/constants"
+	"github.com/flowy-live/llink/internal/livestore"
 	"github.com/flowy-live/llink/internal/utils"
 	"github.com/jackc/pgx/v5"
 	"github.com/jackc/pgx/v5/pgxpool"
@@ -20,53 +20,49 @@ import (
 
 var ErrNotFound = errors.New("network not found")
 var ErrInvalidName = errors.New("name cannot be empty")
+var ErrInvalidEmail = errors.New("invalid email")
+var ErrInvalidHumanId = errors.New("invalid humanId")
 var ErrCapacityExceeded = errors.New("active stream capacity exceeded")
 var ErrInvalidRetentionHours = errors.New("message retention hours must be between 24 and 336")
 
 type Service interface {
+	Reader
+
 	// 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.
+	// Returns ErrInvalidHumanId if any humanId is empty.
 	AddMembers(ctx context.Context, networkID string, humanIds []string) error
+	// RemoveMember returns ErrInvalidHumanId if humanId is empty.
 	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 returns ErrNotFound if the network doesn't exist
+	// or ErrInvalidEmail if any email fails normalization.
 	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 returns ErrInvalidEmail or ErrInvalidHumanId.
 	AcceptInvitation(ctx context.Context, networkID, email, humanId string) error
+	// RevokeInvitation returns ErrInvalidEmail.
 	RevokeInvitation(ctx context.Context, networkID, email string) error
 }
 
 type serviceImpl struct {
-	pool       *pgxpool.Pool
-	repo       repository
+	*readerImpl
 	aeroSvc    pbaero.PrimaryClient
 	billingSvc billing.Service
-	fs         *firestore.Client
+	pub        livestore.MembershipPublisher
 }
 
-func NewService(pool *pgxpool.Pool, aeroSvc pbaero.PrimaryClient, billingSvc billing.Service, fs *firestore.Client) Service {
+func NewService(pool *pgxpool.Pool, aeroSvc pbaero.PrimaryClient, billingSvc billing.Service, pub livestore.MembershipPublisher) Service {
 	return &serviceImpl{
-		pool:       pool,
-		repo:       newRepository(pool),
+		readerImpl: newReader(pool),
 		aeroSvc:    aeroSvc,
 		billingSvc: billingSvc,
-		fs:         fs,
+		pub:        pub,
 	}
 }
 
@@ -101,14 +97,6 @@ func (s *serviceImpl) Create(ctx context.Context, name, adminHumanId string) (*N
 	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 == "" {
@@ -124,7 +112,7 @@ func (s *serviceImpl) SetName(ctx context.Context, id, name string) error {
 
 func (s *serviceImpl) AddMembers(ctx context.Context, networkID string, humanIds []string) error {
 	if slices.Contains(humanIds, "") {
-		return fmt.Errorf("invalid humanId")
+		return ErrInvalidHumanId
 	}
 	if err := s.mutateMembers(ctx, networkID, func(tx pgx.Tx) error {
 		for _, humanId := range humanIds {
@@ -145,7 +133,7 @@ func (s *serviceImpl) AddMembers(ctx context.Context, networkID string, humanIds
 
 func (s *serviceImpl) RemoveMember(ctx context.Context, networkID, humanId string) error {
 	if humanId == "" {
-		return fmt.Errorf("invalid humanId")
+		return ErrInvalidHumanId
 	}
 	if err := s.mutateMembers(ctx, networkID, func(tx pgx.Tx) error {
 		return s.repo.removeMember(ctx, tx, networkID, humanId)
@@ -157,34 +145,19 @@ func (s *serviceImpl) RemoveMember(ctx context.Context, networkID, humanId strin
 	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.
+// mirrorAddMembership / mirrorRemoveMembership keep the live store membership
+// projection (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.
 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)
+	if err := s.pub.Add(ctx, humanId, networkID); err != nil {
+		slog.Error("membership publish 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)
+	if err := s.pub.Remove(ctx, humanId, networkID); err != nil {
+		slog.Error("membership publish remove failed", "error", err, "humanId", humanId, "networkID", networkID)
 	}
 }
 
@@ -216,42 +189,19 @@ func (s *serviceImpl) mutateMembers(ctx context.Context, networkID string, fn fu
 	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
+		if errors.Is(err, errNotFound) {
+			return ErrNotFound
+		}
+		return err
 	}
 
 	for _, email := range emails {
 		normalized, err := utils.NormalizeEmail(email)
 		if err != nil {
-			return fmt.Errorf("invalid email %q: %w", email, err)
+			return fmt.Errorf("%q: %w: %w", email, ErrInvalidEmail, err)
 		}
 		if err := s.repo.createInvitation(ctx, networkID, normalized); err != nil {
 			return err
@@ -273,25 +223,13 @@ func (s *serviceImpl) InviteByEmail(ctx context.Context, networkID string, email
 	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)
+		return fmt.Errorf("%w: %w", ErrInvalidEmail, err)
 	}
 	if humanId == "" {
-		return fmt.Errorf("invalid humanId")
+		return ErrInvalidHumanId
 	}
 
 	if err := s.mutateMembers(ctx, networkID, func(tx pgx.Tx) error {
@@ -311,7 +249,7 @@ func (s *serviceImpl) AcceptInvitation(ctx context.Context, networkID, email, hu
 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 fmt.Errorf("%w: %w", ErrInvalidEmail, err)
 	}
 	return s.repo.deleteInvitation(ctx, s.pool, networkID, normalized)
 }

go/internal/network/service_test.go

@@ -6,7 +6,8 @@ import (
 	"testing"
 
 	pbaero "github.com/flowy-live/llink/genproto/aero"
-	"github.com/flowy-live/llink/internal/billing"
+	mock_billing "github.com/flowy-live/llink/internal/billing/mocks"
+	mock_livestore "github.com/flowy-live/llink/internal/livestore/mocks"
 	"github.com/flowy-live/llink/internal/network"
 	"github.com/flowy-live/llink/internal/testhelper"
 	"github.com/flowy-live/llink/internal/testhelper/mocks/aero"
@@ -32,7 +33,15 @@ func newTestService(t *testing.T) network.Service {
 		ShootEmail(gomock.Any(), gomock.Any()).
 		Return(&pbaero.ShootEmailResponse{}, nil).
 		AnyTimes()
-	return network.NewService(dbPool, mockAero, billing.Noop(), nil)
+
+	mockBilling := mock_billing.NewMockService(ctrl)
+	mockBilling.EXPECT().SyncSeats(gomock.Any(), gomock.Any(), gomock.Any()).Return(nil).AnyTimes()
+
+	mockPub := mock_livestore.NewMockMembershipPublisher(ctrl)
+	mockPub.EXPECT().Add(gomock.Any(), gomock.Any(), gomock.Any()).Return(nil).AnyTimes()
+	mockPub.EXPECT().Remove(gomock.Any(), gomock.Any(), gomock.Any()).Return(nil).AnyTimes()
+
+	return network.NewService(dbPool, mockAero, mockBilling, mockPub)
 }
 
 func TestNetworkService(t *testing.T) {