migrate orion repo into monorepo structure

2026-02-21 08:48:34 -08:00
parent 144596fcaa
commit b5f90709de
91 changed files with 19775 additions and 0 deletions
@@ -0,0 +1,954 @@
+package particle
+
+import (
+	"context"
+	"encoding/json"
+	"errors"
+	"log/slog"
+	"time"
+
+	"github.com/flowy-live/llink/internal/network"
+	"github.com/flowy-live/llink/internal/utils"
+	"github.com/jackc/pgx/v5/pgxpool"
+)
+
+const defaultPageSize = 50
+
+type Service interface {
+	// Create creates a new particle. Caller must be a network member (verified by handler).
+	// Returns ErrInvalidType, ErrInvalidData, ErrMembersRequired, ErrInvalidParent, ErrAccessDenied, or ErrCapacityExceeded.
+	Create(ctx context.Context, input CreateInput, requesterEmail string) (*Particle, error)
+	// GetByID returns ErrNotFound or ErrAccessDenied.
+	GetByID(ctx context.Context, id, requesterEmail string) (*Particle, error)
+	// Update updates the particle's data. Returns ErrNotFound, ErrAccessDenied, or ErrInvalidData.
+	Update(ctx context.Context, id string, data json.RawMessage, requesterEmail string) (*Particle, error)
+	// Delete returns ErrNotFound or ErrAccessDenied.
+	Delete(ctx context.Context, id, requesterEmail string) error
+
+	// List returns particles in a network. Use parentID=nil for root particles.
+	// Returns ErrNotFound or ErrAccessDenied if parentID is specified and inaccessible.
+	List(ctx context.Context, networkID string, parentID *string, requesterEmail string, filter ListFilter, cursor *Cursor, limit int) (*ParticleList, error)
+
+	// OpenStream opens a closed stream. Returns ErrNotFound, ErrAccessDenied, ErrNotAStream, ErrStreamAlreadyOpen, or ErrCapacityExceeded.
+	OpenStream(ctx context.Context, id, requesterEmail string) error
+	// CloseStream closes an open stream. Returns ErrNotFound, ErrAccessDenied, ErrNotAStream, or ErrStreamAlreadyClosed.
+	CloseStream(ctx context.Context, id, requesterEmail string) error
+
+	// SetVisibility changes the particle's visibility mode. Returns ErrNotFound, ErrAccessDenied, or ErrAccessExpansion.
+	SetVisibility(ctx context.Context, id string, mode VisibilityMode, requesterEmail string) error
+	// AddMembers adds members to a custom visibility particle. Returns ErrNotFound or ErrAccessDenied.
+	AddMembers(ctx context.Context, id string, emails []string, requesterEmail string) error
+	// RemoveMembers removes members from a custom visibility particle. Returns ErrNotFound or ErrAccessDenied.
+	RemoveMembers(ctx context.Context, id string, emails []string, requesterEmail string) error
+
+	// Seen tracking (private)
+	MarkSeen(ctx context.Context, id, requesterEmail string) error
+	MarkSeenBatch(ctx context.Context, ids []string, requesterEmail string) error
+
+	// Ack tracking (public, permanent)
+	Ack(ctx context.Context, id, requesterEmail string) error
+
+	// Unseen counts for stream list view
+	GetUnseenCounts(ctx context.Context, networkID string, streamIDs []string, requesterEmail string) (map[string]int, error)
+
+	// Bulk lookups for handler enrichment
+	GetSeenMap(ctx context.Context, particleIDs []string, requesterEmail string) (map[string]bool, error)
+	GetAcksMap(ctx context.Context, particleIDs []string) (map[string][]AckInfo, error)
+	GetMembersMap(ctx context.Context, particleIDs []string) (map[string][]string, error)
+}
+
+type serviceImpl struct {
+	repo       repository
+	networkSvc network.Service
+}
+
+func NewService(pool *pgxpool.Pool, networkSvc network.Service) Service {
+	return &serviceImpl{
+		repo:       newRepository(pool),
+		networkSvc: networkSvc,
+	}
+}
+
+// checkAccess verifies that the email has access to the particle based on visibility.
+// Assumes the caller is already verified as a network member (handler responsibility).
+// Walks up the ancestor chain only when visibility is inherited, stopping at the first
+// network_all or custom node.
+func (s *serviceImpl) checkAccess(ctx context.Context, particleID, email string) (bool, error) {
+	ancestors, err := s.repo.getAncestorChain(ctx, particleID)
+	if err != nil {
+		return false, err
+	}
+
+	if len(ancestors) == 0 {
+		return false, errNotFound
+	}
+
+	// Build lookup map by ID
+	byID := make(map[string]*Particle, len(ancestors))
+	for _, p := range ancestors {
+		byID[p.ID] = p
+	}
+
+	// Start from the target particle (first in chain) and walk up on inherited
+	current := ancestors[0]
+	for {
+		switch current.Visibility {
+		case VisibilityNetworkAll:
+			return true, nil
+		case VisibilityCustom:
+			return s.repo.isMemberOf(ctx, current.ID, email)
+		case VisibilityInherited:
+			if current.ParentID == nil {
+				// inherited at root is invalid state, deny access
+				return false, nil
+			}
+			parent, ok := byID[*current.ParentID]
+			if !ok {
+				return false, nil
+			}
+			current = parent
+		default:
+			return false, nil
+		}
+	}
+}
+
+func (s *serviceImpl) Create(ctx context.Context, input CreateInput, requesterEmail string) (*Particle, error) {
+	requesterEmail, err := utils.NormalizeEmail(requesterEmail)
+	if err != nil {
+		return nil, err
+	}
+
+	// Validate particle type
+	if !isValidParticleType(input.Type) {
+		return nil, ErrInvalidType
+	}
+
+	// Validate data matches type requirements
+	if err := validateParticleData(input.Type, input.Data); err != nil {
+		return nil, err
+	}
+
+	// MVP visibility rules:
+	// - Child particles (have parent) → always inherited
+	// - Root particles (no parent) → cannot be inherited, default network_all
+	if input.ParentID != nil {
+		// Children always inherit from parent
+		input.Visibility = VisibilityInherited
+		input.Members = nil // no members on inherited particles
+
+		// Reject streams and folders as children (MVP: streams are root-level only)
+		if input.Type == TypeStream || input.Type == TypeFolder {
+			return nil, ErrInvalidParent
+		}
+	} else {
+		// Root particles cannot be inherited
+		if input.Visibility == VisibilityInherited {
+			return nil, ErrInheritedAtRoot
+		}
+		if input.Visibility == "" {
+			input.Visibility = VisibilityNetworkAll
+		}
+	}
+
+	// Custom visibility requires at least one member and must be a stream
+	if input.Visibility == VisibilityCustom {
+		if input.Type != TypeStream {
+			return nil, ErrNotAContainer
+		}
+		if len(input.Members) == 0 {
+			return nil, ErrMembersRequired
+		}
+	}
+
+	// Network membership is verified by handler - we only check particle visibility
+	// If parent specified, check parent access (visibility-based)
+	if input.ParentID != nil {
+		hasAccess, err := s.checkAccess(ctx, *input.ParentID, requesterEmail)
+		if err != nil {
+			if errors.Is(err, errNotFound) {
+				return nil, ErrInvalidParent
+			}
+			return nil, err
+		}
+		if !hasAccess {
+			return nil, ErrAccessDenied
+		}
+	}
+
+	// Build the particle
+	p := &Particle{
+		Type:           input.Type,
+		NetworkID:      input.NetworkID,
+		ParentID:       input.ParentID,
+		CreatedByEmail: requesterEmail,
+		Visibility:     input.Visibility,
+		Data:           input.Data,
+	}
+
+	if p.Data == nil {
+		p.Data = json.RawMessage("{}")
+	}
+
+	// For streams, set initial status to open and check capacity
+	if input.Type == TypeStream {
+		// Set status to open in the data JSON
+		data, err := setStreamStatus(p.Data, string(StreamStatusOpen))
+		if err != nil {
+			return nil, err
+		}
+		p.Data = data
+
+		// Check and increment capacity
+		err = s.networkSvc.IncrementOpenStreamCount(ctx, input.NetworkID)
+		if err != nil {
+			if errors.Is(err, network.ErrCapacityExceeded) {
+				return nil, ErrCapacityExceeded
+			}
+			return nil, err
+		}
+	}
+
+	// Create the particle
+	created, err := s.repo.create(ctx, p)
+	if err != nil {
+		// If we incremented the stream count but creation failed, decrement it
+		if input.Type == TypeStream {
+			if decErr := s.networkSvc.DecrementOpenStreamCount(ctx, input.NetworkID); decErr != nil {
+				slog.Warn("failed to rollback stream count after particle creation failure", "error", decErr, "network_id", input.NetworkID)
+			}
+		}
+		return nil, err
+	}
+
+	// Add members if custom visibility (only streams for MVP)
+	if input.Visibility == VisibilityCustom && len(input.Members) > 0 {
+		normalizedEmails := make([]string, 0, len(input.Members)+1)
+		// Always include the creator
+		normalizedEmails = append(normalizedEmails, requesterEmail)
+		for _, email := range input.Members {
+			normalized, err := utils.NormalizeEmail(email)
+			if err != nil {
+				continue // Skip invalid emails
+			}
+			if normalized == requesterEmail {
+				continue // Already added
+			}
+			normalizedEmails = append(normalizedEmails, normalized)
+		}
+		if err := s.repo.addMembers(ctx, created.ID, normalizedEmails); err != nil {
+			return nil, err
+		}
+	}
+
+	return created, nil
+}
+
+func (s *serviceImpl) GetByID(ctx context.Context, id, requesterEmail string) (*Particle, error) {
+	requesterEmail, err := utils.NormalizeEmail(requesterEmail)
+	if err != nil {
+		return nil, err
+	}
+
+	// Check access
+	hasAccess, err := s.checkAccess(ctx, id, requesterEmail)
+	if err != nil {
+		if errors.Is(err, errNotFound) {
+			return nil, ErrNotFound
+		}
+		return nil, err
+	}
+	if !hasAccess {
+		return nil, ErrAccessDenied
+	}
+
+	p, err := s.repo.getByID(ctx, id)
+	if err != nil {
+		if errors.Is(err, errNotFound) {
+			return nil, ErrNotFound
+		}
+		return nil, err
+	}
+	return p, nil
+}
+
+func (s *serviceImpl) Update(ctx context.Context, id string, data json.RawMessage, requesterEmail string) (*Particle, error) {
+	requesterEmail, err := utils.NormalizeEmail(requesterEmail)
+	if err != nil {
+		return nil, err
+	}
+
+	// Check access
+	hasAccess, err := s.checkAccess(ctx, id, requesterEmail)
+	if err != nil {
+		if errors.Is(err, errNotFound) {
+			return nil, ErrNotFound
+		}
+		return nil, err
+	}
+	if !hasAccess {
+		return nil, ErrAccessDenied
+	}
+
+	// Get the particle to validate data against its type
+	p, err := s.repo.getByID(ctx, id)
+	if err != nil {
+		if errors.Is(err, errNotFound) {
+			return nil, ErrNotFound
+		}
+		return nil, err
+	}
+
+	// Validate data matches type requirements
+	if err := validateParticleData(p.Type, data); err != nil {
+		return nil, err
+	}
+
+	err = s.repo.update(ctx, id, data, time.Now())
+	if err != nil {
+		if errors.Is(err, errNotFound) {
+			return nil, ErrNotFound
+		}
+		return nil, err
+	}
+
+	return s.repo.getByID(ctx, id)
+}
+
+func (s *serviceImpl) Delete(ctx context.Context, id, requesterEmail string) error {
+	requesterEmail, err := utils.NormalizeEmail(requesterEmail)
+	if err != nil {
+		return err
+	}
+
+	// Check access
+	hasAccess, err := s.checkAccess(ctx, id, requesterEmail)
+	if err != nil {
+		if errors.Is(err, errNotFound) {
+			return ErrNotFound
+		}
+		return err
+	}
+	if !hasAccess {
+		return ErrAccessDenied
+	}
+
+	// Get the particle to check if it's an open stream
+	p, err := s.repo.getByID(ctx, id)
+	if err != nil {
+		if errors.Is(err, errNotFound) {
+			return ErrNotFound
+		}
+		return err
+	}
+
+	// If it's an open stream, decrement the count
+	if p.Type == TypeStream && getStreamStatus(p.Data) == string(StreamStatusOpen) {
+		if err := s.networkSvc.DecrementOpenStreamCount(ctx, p.NetworkID); err != nil {
+			return err
+		}
+	}
+
+	err = s.repo.delete(ctx, id)
+	if errors.Is(err, errNotFound) {
+		return ErrNotFound
+	}
+	return err
+}
+
+func (s *serviceImpl) List(ctx context.Context, networkID string, parentID *string, requesterEmail string, filter ListFilter, cursor *Cursor, limit int) (*ParticleList, error) {
+	requesterEmail, err := utils.NormalizeEmail(requesterEmail)
+	if err != nil {
+		return nil, err
+	}
+
+	// Network membership is verified by handler - we only check particle visibility
+	// If parentID specified, check access to parent (visibility-based)
+	if parentID != nil {
+		hasAccess, err := s.checkAccess(ctx, *parentID, requesterEmail)
+		if err != nil {
+			if errors.Is(err, errNotFound) {
+				return nil, ErrNotFound
+			}
+			return nil, err
+		}
+		if !hasAccess {
+			return nil, ErrAccessDenied
+		}
+	}
+
+	// Fetch one extra to determine if there are more
+	// Access filtering is done in the query itself (network_all OR user is member)
+	if limit == 0 {
+		limit = defaultPageSize
+	}
+	extraLimit := limit + 1
+	particles, err := s.repo.list(ctx, networkID, parentID, requesterEmail, filter, extraLimit, cursor)
+	if err != nil {
+		return nil, err
+	}
+
+	hasMore := len(particles) > limit
+	result := &ParticleList{
+		HasMore: hasMore,
+	}
+
+	if hasMore {
+		particles = particles[:limit]
+	}
+	result.Particles = particles
+
+	// Bidirectional cursors
+	if len(particles) > 0 {
+		firstParticle := particles[0]
+		lastParticle := particles[len(particles)-1]
+
+		result.PrevCursor = &Cursor{
+			Position:  firstParticle.UpdatedAt.Format(time.RFC3339Nano),
+			Direction: "before",
+		}
+
+		if result.HasMore {
+			result.NextCursor = &Cursor{
+				Position:  lastParticle.UpdatedAt.Format(time.RFC3339Nano),
+				Direction: "after",
+			}
+		}
+	}
+
+	return result, nil
+}
+
+func (s *serviceImpl) OpenStream(ctx context.Context, id, requesterEmail string) error {
+	requesterEmail, err := utils.NormalizeEmail(requesterEmail)
+	if err != nil {
+		return err
+	}
+
+	// Check access
+	hasAccess, err := s.checkAccess(ctx, id, requesterEmail)
+	if err != nil {
+		if errors.Is(err, errNotFound) {
+			return ErrNotFound
+		}
+		return err
+	}
+	if !hasAccess {
+		return ErrAccessDenied
+	}
+
+	// Get the particle
+	p, err := s.repo.getByID(ctx, id)
+	if err != nil {
+		if errors.Is(err, errNotFound) {
+			return ErrNotFound
+		}
+		return err
+	}
+
+	if p.Type != TypeStream {
+		return ErrNotAStream
+	}
+
+	if getStreamStatus(p.Data) == string(StreamStatusOpen) {
+		return ErrStreamAlreadyOpen
+	}
+
+	// Check and increment capacity
+	err = s.networkSvc.IncrementOpenStreamCount(ctx, p.NetworkID)
+	if err != nil {
+		if errors.Is(err, network.ErrCapacityExceeded) {
+			return ErrCapacityExceeded
+		}
+		return err
+	}
+
+	// Update stream status in data
+	newData, err := setStreamStatus(p.Data, string(StreamStatusOpen))
+	if err != nil {
+		if decErr := s.networkSvc.DecrementOpenStreamCount(ctx, p.NetworkID); decErr != nil {
+			slog.Warn("failed to rollback stream count after status update failure", "error", decErr, "network_id", p.NetworkID, "particle_id", id)
+		}
+		return err
+	}
+
+	err = s.repo.update(ctx, id, newData, time.Now())
+	if err != nil {
+		// Rollback the capacity increment
+		if decErr := s.networkSvc.DecrementOpenStreamCount(ctx, p.NetworkID); decErr != nil {
+			slog.Warn("failed to rollback stream count after particle update failure", "error", decErr, "network_id", p.NetworkID, "particle_id", id)
+		}
+		if errors.Is(err, errNotFound) {
+			return ErrNotFound
+		}
+		return err
+	}
+
+	return nil
+}
+
+func (s *serviceImpl) CloseStream(ctx context.Context, id, requesterEmail string) error {
+	requesterEmail, err := utils.NormalizeEmail(requesterEmail)
+	if err != nil {
+		return err
+	}
+
+	// Check access
+	hasAccess, err := s.checkAccess(ctx, id, requesterEmail)
+	if err != nil {
+		if errors.Is(err, errNotFound) {
+			return ErrNotFound
+		}
+		return err
+	}
+	if !hasAccess {
+		return ErrAccessDenied
+	}
+
+	// Get the particle
+	p, err := s.repo.getByID(ctx, id)
+	if err != nil {
+		if errors.Is(err, errNotFound) {
+			return ErrNotFound
+		}
+		return err
+	}
+
+	if p.Type != TypeStream {
+		return ErrNotAStream
+	}
+
+	if getStreamStatus(p.Data) == string(StreamStatusClosed) {
+		return ErrStreamAlreadyClosed
+	}
+
+	// Update stream status in data
+	newData, err := setStreamStatus(p.Data, string(StreamStatusClosed))
+	if err != nil {
+		return err
+	}
+
+	err = s.repo.update(ctx, id, newData, time.Now())
+	if err != nil {
+		if errors.Is(err, errNotFound) {
+			return ErrNotFound
+		}
+		return err
+	}
+
+	// Decrement capacity
+	return s.networkSvc.DecrementOpenStreamCount(ctx, p.NetworkID)
+}
+
+func (s *serviceImpl) SetVisibility(ctx context.Context, id string, mode VisibilityMode, requesterEmail string) error {
+	requesterEmail, err := utils.NormalizeEmail(requesterEmail)
+	if err != nil {
+		return err
+	}
+
+	// Check access
+	hasAccess, err := s.checkAccess(ctx, id, requesterEmail)
+	if err != nil {
+		if errors.Is(err, errNotFound) {
+			return ErrNotFound
+		}
+		return err
+	}
+	if !hasAccess {
+		return ErrAccessDenied
+	}
+
+	// Get the particle to check constraints
+	p, err := s.repo.getByID(ctx, id)
+	if err != nil {
+		if errors.Is(err, errNotFound) {
+			return ErrNotFound
+		}
+		return err
+	}
+
+	// Root particles cannot be inherited
+	if mode == VisibilityInherited && p.ParentID == nil {
+		return ErrInheritedAtRoot
+	}
+
+	// If expanding to network_all, check that parent's effective visibility allows it
+	if mode == VisibilityNetworkAll && p.ParentID != nil {
+		parentVis, err := s.getEffectiveVisibility(ctx, *p.ParentID)
+		if err != nil {
+			return err
+		}
+		if parentVis == VisibilityCustom {
+			return ErrAccessExpansion
+		}
+	}
+
+	err = s.repo.setVisibility(ctx, id, mode)
+	if errors.Is(err, errNotFound) {
+		return ErrNotFound
+	}
+	return err
+}
+
+// getEffectiveVisibility walks up the inherited chain to find the concrete visibility mode.
+func (s *serviceImpl) getEffectiveVisibility(ctx context.Context, particleID string) (VisibilityMode, error) {
+	ancestors, err := s.repo.getAncestorChain(ctx, particleID)
+	if err != nil {
+		return "", err
+	}
+	if len(ancestors) == 0 {
+		return "", errNotFound
+	}
+
+	byID := make(map[string]*Particle, len(ancestors))
+	for _, p := range ancestors {
+		byID[p.ID] = p
+	}
+
+	current := ancestors[0]
+	for {
+		if current.Visibility != VisibilityInherited {
+			return current.Visibility, nil
+		}
+		if current.ParentID == nil {
+			return VisibilityNetworkAll, nil
+		}
+		parent, ok := byID[*current.ParentID]
+		if !ok {
+			return VisibilityNetworkAll, nil
+		}
+		current = parent
+	}
+}
+
+func (s *serviceImpl) AddMembers(ctx context.Context, id string, emails []string, requesterEmail string) error {
+	requesterEmail, err := utils.NormalizeEmail(requesterEmail)
+	if err != nil {
+		return err
+	}
+
+	// Check access
+	hasAccess, err := s.checkAccess(ctx, id, requesterEmail)
+	if err != nil {
+		if errors.Is(err, errNotFound) {
+			return ErrNotFound
+		}
+		return err
+	}
+	if !hasAccess {
+		return ErrAccessDenied
+	}
+
+	// Get the particle to check type and parent access
+	p, err := s.repo.getByID(ctx, id)
+	if err != nil {
+		if errors.Is(err, errNotFound) {
+			return ErrNotFound
+		}
+		return err
+	}
+
+	// Only streams can have members
+	if p.Type != TypeStream {
+		return ErrNotAContainer
+	}
+
+	// Validate and normalize emails, check network membership
+	normalizedEmails := make([]string, 0, len(emails))
+	for _, email := range emails {
+		normalized, err := utils.NormalizeEmail(email)
+		if err != nil {
+			continue
+		}
+
+		// Root stream - check network membership
+		isMember, err := s.networkSvc.IsMember(ctx, p.NetworkID, normalized)
+		if err != nil || !isMember {
+			continue
+		}
+
+		normalizedEmails = append(normalizedEmails, normalized)
+	}
+
+	if len(normalizedEmails) == 0 {
+		return nil
+	}
+
+	return s.repo.addMembers(ctx, id, normalizedEmails)
+}
+
+func (s *serviceImpl) RemoveMembers(ctx context.Context, id string, emails []string, requesterEmail string) error {
+	requesterEmail, err := utils.NormalizeEmail(requesterEmail)
+	if err != nil {
+		return err
+	}
+
+	// Check access
+	hasAccess, err := s.checkAccess(ctx, id, requesterEmail)
+	if err != nil {
+		if errors.Is(err, errNotFound) {
+			return ErrNotFound
+		}
+		return err
+	}
+	if !hasAccess {
+		return ErrAccessDenied
+	}
+
+	// Get the particle to check type
+	p, err := s.repo.getByID(ctx, id)
+	if err != nil {
+		if errors.Is(err, errNotFound) {
+			return ErrNotFound
+		}
+		return err
+	}
+
+	// Only streams can have members
+	if p.Type != TypeStream {
+		return ErrNotAContainer
+	}
+
+	normalizedEmails := make([]string, 0, len(emails))
+	for _, email := range emails {
+		normalized, err := utils.NormalizeEmail(email)
+		if err != nil {
+			continue
+		}
+		normalizedEmails = append(normalizedEmails, normalized)
+	}
+
+	if len(normalizedEmails) == 0 {
+		return nil
+	}
+
+	return s.repo.removeMembers(ctx, id, normalizedEmails)
+}
+
+func (s *serviceImpl) MarkSeen(ctx context.Context, id, requesterEmail string) error {
+	requesterEmail, err := utils.NormalizeEmail(requesterEmail)
+	if err != nil {
+		return err
+	}
+
+	// Check access
+	hasAccess, err := s.checkAccess(ctx, id, requesterEmail)
+	if err != nil {
+		if errors.Is(err, errNotFound) {
+			return ErrNotFound
+		}
+		return err
+	}
+	if !hasAccess {
+		return ErrAccessDenied
+	}
+
+	return s.repo.markSeen(ctx, id, requesterEmail)
+}
+
+func (s *serviceImpl) MarkSeenBatch(ctx context.Context, ids []string, requesterEmail string) error {
+	requesterEmail, err := utils.NormalizeEmail(requesterEmail)
+	if err != nil {
+		return err
+	}
+
+	// Check access for each particle and mark seen
+	for _, id := range ids {
+		hasAccess, err := s.checkAccess(ctx, id, requesterEmail)
+		if err != nil {
+			if errors.Is(err, errNotFound) {
+				continue // Skip non-existent particles
+			}
+			return err
+		}
+		if !hasAccess {
+			continue // Skip inaccessible particles
+		}
+
+		if err := s.repo.markSeen(ctx, id, requesterEmail); err != nil {
+			return err
+		}
+	}
+
+	return nil
+}
+
+func (s *serviceImpl) Ack(ctx context.Context, id, requesterEmail string) error {
+	requesterEmail, err := utils.NormalizeEmail(requesterEmail)
+	if err != nil {
+		return err
+	}
+
+	// Check access
+	hasAccess, err := s.checkAccess(ctx, id, requesterEmail)
+	if err != nil {
+		if errors.Is(err, errNotFound) {
+			return ErrNotFound
+		}
+		return err
+	}
+	if !hasAccess {
+		return ErrAccessDenied
+	}
+
+	// Ack also marks as seen
+	if err := s.repo.markSeen(ctx, id, requesterEmail); err != nil {
+		return err
+	}
+
+	return s.repo.ack(ctx, id, requesterEmail)
+}
+
+func (s *serviceImpl) GetUnseenCounts(ctx context.Context, networkID string, streamIDs []string, requesterEmail string) (map[string]int, error) {
+	requesterEmail, err := utils.NormalizeEmail(requesterEmail)
+	if err != nil {
+		return nil, err
+	}
+
+	return s.repo.getUnseenCounts(ctx, streamIDs, requesterEmail)
+}
+
+func (s *serviceImpl) GetSeenMap(ctx context.Context, particleIDs []string, requesterEmail string) (map[string]bool, error) {
+	requesterEmail, err := utils.NormalizeEmail(requesterEmail)
+	if err != nil {
+		return nil, err
+	}
+
+	return s.repo.getSeenMap(ctx, particleIDs, requesterEmail)
+}
+
+func (s *serviceImpl) GetAcksMap(ctx context.Context, particleIDs []string) (map[string][]AckInfo, error) {
+	return s.repo.getAcksMap(ctx, particleIDs)
+}
+
+func (s *serviceImpl) GetMembersMap(ctx context.Context, particleIDs []string) (map[string][]string, error) {
+	return s.repo.getMembersMap(ctx, particleIDs)
+}
+
+func isValidParticleType(t ParticleType) bool {
+	switch t {
+	case TypeStream, TypeFolder, TypeMedia, TypeFile, TypeText, TypeQuest, TypePaper:
+		return true
+	default:
+		return false
+	}
+}
+
+// getStreamStatus extracts the status from a stream particle's data
+func getStreamStatus(data json.RawMessage) string {
+	var d StreamData
+	if err := json.Unmarshal(data, &d); err != nil {
+		return ""
+	}
+	return d.Status
+}
+
+// setStreamStatus updates the status in a stream particle's data
+func setStreamStatus(data json.RawMessage, status string) (json.RawMessage, error) {
+	var d StreamData
+	if err := json.Unmarshal(data, &d); err != nil {
+		d = StreamData{}
+	}
+	d.Status = status
+	return json.Marshal(d)
+}
+
+// validateParticleData validates that the data field contains valid JSON
+// and has required fields for the given particle type.
+func validateParticleData(pType ParticleType, data json.RawMessage) error {
+	// Empty or null data is allowed - will default to {}
+	if len(data) == 0 || string(data) == "null" || string(data) == "{}" {
+		return nil
+	}
+
+	switch pType {
+	case TypeStream:
+		var d StreamData
+		if err := json.Unmarshal(data, &d); err != nil {
+			return errors.Join(ErrInvalidData, err)
+		}
+		if d.Name == "" {
+			return errors.Join(ErrInvalidData, errors.New("stream requires name"))
+		}
+		if d.Status != string(StreamStatusOpen) && d.Status != string(StreamStatusClosed) {
+			return errors.Join(ErrInvalidData, errors.New("stream requires valid status"))
+		}
+
+	case TypeFolder:
+		var d FolderData
+		if err := json.Unmarshal(data, &d); err != nil {
+			return errors.Join(ErrInvalidData, err)
+		}
+		if d.Name == "" {
+			return errors.Join(ErrInvalidData, errors.New("folder requires name"))
+		}
+
+	case TypeMedia:
+		var d MediaData
+		if err := json.Unmarshal(data, &d); err != nil {
+			return errors.Join(ErrInvalidData, err)
+		}
+		if d.ObjectID == "" {
+			return errors.Join(ErrInvalidData, errors.New("media requires object_id"))
+		}
+		if d.MimeType == "" {
+			return errors.Join(ErrInvalidData, errors.New("media requires mime_type"))
+		}
+		if d.DurationMs <= 0 {
+			return errors.Join(ErrInvalidData, errors.New("media requires positive duration_ms"))
+		}
+
+	case TypeFile:
+		var d FileData
+		if err := json.Unmarshal(data, &d); err != nil {
+			return errors.Join(ErrInvalidData, err)
+		}
+		if d.ObjectID == "" {
+			return errors.Join(ErrInvalidData, errors.New("file requires object_id"))
+		}
+		if d.Filename == "" {
+			return errors.Join(ErrInvalidData, errors.New("file requires filename"))
+		}
+		if d.MimeType == "" {
+			return errors.Join(ErrInvalidData, errors.New("file requires mime_type"))
+		}
+		if d.Size <= 0 {
+			return errors.Join(ErrInvalidData, errors.New("file requires non-negative size"))
+		}
+
+	case TypeText:
+		var d TextData
+		if err := json.Unmarshal(data, &d); err != nil {
+			return errors.Join(ErrInvalidData, err)
+		}
+		if d.Content == "" {
+			return errors.Join(ErrInvalidData, errors.New("text requires content"))
+		}
+
+	case TypeQuest:
+		var d QuestData
+		if err := json.Unmarshal(data, &d); err != nil {
+			return errors.Join(ErrInvalidData, err)
+		}
+		if d.Title == "" {
+			return errors.Join(ErrInvalidData, errors.New("quest requires title"))
+		}
+		if d.Description == "" {
+			return errors.Join(ErrInvalidData, errors.New("quest requires description"))
+		}
+
+	case TypePaper:
+		var d PaperData
+		if err := json.Unmarshal(data, &d); err != nil {
+			return errors.Join(ErrInvalidData, err)
+		}
+		if d.Title == "" {
+			return errors.Join(ErrInvalidData, errors.New("paper requires title"))
+		}
+		if d.Content == "" {
+			return errors.Join(ErrInvalidData, errors.New("paper requires content"))
+		}
+	}
+
+	return nil
+}