package unread

import (
	"context"
	"fmt"
	"github.com/tidbcloud/AgentTeam/internal/observability/logging"
	"log/slog"
	"strings"
	"sync"
	"time"
)

// SnapshotStore is the bounded snapshot persistence port used by Service.
// Implementations must keep every read scoped to the supplied workspace,
// member, scope, page, or batch arguments.
type SnapshotStore interface {
	UpsertSnapshots(ctx context.Context, snapshots []Snapshot) error
	LoadSnapshotPage(ctx context.Context, workspaceID string, memberID string, scopeIDs []string, afterScopeID string, limit int) ([]Snapshot, error)
}

type snapshotFrontierMetadataStore interface {
	LoadFrontierMetadataPage(ctx context.Context, workspaceID string, memberID string, scopeIDs []string, afterScopeID string, limit int) ([]FrontierMetadata, error)
}

type visibleConversationScopeStore interface {
	LoadVisibleConversationScopeIDsPage(ctx context.Context, workspaceID string, memberID string, afterScopeID string, limit int) ([]string, error)
}

type scopeRosterStore interface {
	LoadScopeRosterPage(ctx context.Context, workspaceID string, scopeID string, afterMemberID string, limit int) ([]RosterMember, error)
}

// DeltaSink emits realtime unread deltas. It must not perform unread database
// reads.
type DeltaSink interface {
	EmitUnreadUpdated(workspaceID string, delta UnreadDelta)
}

// Service owns runtime lifecycle, background persistence, lazy snapshot
// hydrate, and bounded roster warmup. Warm reads are served from Runtime
// memory. V1 trusts restored snapshots only after bounded durable frontier
// validation; it does not run broad durable tail replay or legacy task counter
// backfill to top up unread counts.
type Service struct {
	runtime *Runtime
	store   SnapshotStore
	logger  *slog.Logger

	queue chan hydrateJob

	mu                sync.Mutex
	inflight          map[string]struct{}
	retry             map[string]hydrateRetryState
	deltaSink         DeltaSink
	snapshotPersistMu sync.Mutex

	memberSnapshotsHydrated map[memberKey]struct{}
}

type hydrateJobKind string

const (
	hydrateJobSnapshot hydrateJobKind = "snapshot"
	hydrateJobRoster   hydrateJobKind = "roster"
)

type hydrateJob struct {
	kind        hydrateJobKind
	workspaceID string
	memberID    string
	scopeID     string
	scopeType   ScopeType
	scopeIDs    []string
	memberIDs   []string
}

type hydrateRetryState struct {
	job      hydrateJob
	attempts int
	nextAt   time.Time
}

// NewService creates a runtime service around an already validated Runtime.
func NewService(runtime *Runtime, store SnapshotStore, logger *slog.Logger) (*Service, error) {
	if runtime == nil {
		return nil, fmt.Errorf("runtime is required")
	}
	cfg := runtime.Config()
	if cfg.EventQueueSize <= 0 {
		return nil, fmt.Errorf("event queue size must be positive")
	}
	if logger == nil {
		logger = slog.Default()
	}
	return &Service{
		runtime:                 runtime,
		store:                   store,
		logger:                  logger,
		queue:                   make(chan hydrateJob, cfg.EventQueueSize),
		inflight:                make(map[string]struct{}),
		retry:                   make(map[string]hydrateRetryState),
		memberSnapshotsHydrated: make(map[memberKey]struct{}),
	}, nil
}

// SetDeltaSink installs the realtime sink used for unread deltas.
func (s *Service) SetDeltaSink(sink DeltaSink) {
	if s == nil {
		return
	}
	s.mu.Lock()
	s.deltaSink = sink
	s.mu.Unlock()
}

// Start begins background flush and lazy hydrate workers.
func (s *Service) Start(ctx context.Context) {
	if s == nil {
		return
	}
	logging.Go(s.logger, func() { s.runSnapshotFlush(ctx) })
	logging.Go(s.logger, func() { s.runHydrateRetry(ctx) })
	for i := 0; i < s.runtime.Config().HydrateMaxConcurrency; i++ {
		logging.Go(s.logger, func() { s.runHydrateWorker(ctx) })
	}
}

// WorkspaceHasUnread returns the owner-local red-dot state. On a cold member it
// performs one bounded member snapshot hydrate through the runtime store before
// answering, so workspace-level presence uses the same in-memory/snapshot
// truth as scope and realtime replay reads without falling back to broad
// durable-fact aggregation.
func (s *Service) WorkspaceHasUnread(workspaceID string, memberID string) bool {
	if s == nil || s.runtime == nil {
		return false
	}
	workspaceID = strings.TrimSpace(workspaceID)
	memberID = strings.TrimSpace(memberID)
	if workspaceID == "" || memberID == "" {
		return false
	}
	if s.runtime.WorkspaceHasUnread(workspaceID, memberID) {
		return true
	}
	s.hydrateMemberSnapshotsIfNeeded(context.Background(), workspaceID, memberID)
	return s.runtime.WorkspaceHasUnread(workspaceID, memberID)
}

// WorkspaceUnreadConversationCount returns the authoritative workspace-wide
// number of scopes currently contributing to the red-dot aggregate. On a cold
// member it performs one bounded snapshot hydrate so the count matches the
// snapshot published on workspace hydration. The hydrate always runs first:
// unlike WorkspaceHasUnread (where any positive runtime aggregate proves the
// boolean true), a partial aggregate that already counts one scope cannot
// prove the workspace total, since other scopes may still live in unhydrated
// snapshots.
func (s *Service) WorkspaceUnreadConversationCount(workspaceID string, memberID string) int64 {
	if s == nil || s.runtime == nil {
		return 0
	}
	workspaceID = strings.TrimSpace(workspaceID)
	memberID = strings.TrimSpace(memberID)
	if workspaceID == "" || memberID == "" {
		return 0
	}
	s.hydrateMemberSnapshotsIfNeeded(context.Background(), workspaceID, memberID)
	return s.runtime.WorkspaceUnreadConversationCount(workspaceID, memberID)
}

// ScopeUnreadCounts returns loaded scope counts and queues bounded snapshot
// hydrate work for missing cold-start entries.
func (s *Service) ScopeUnreadCounts(workspaceID string, memberID string, scopeIDs []string) map[string]int64 {
	if s == nil || s.runtime == nil {
		return map[string]int64{}
	}
	counts := s.runtime.ScopeUnreadCounts(workspaceID, memberID, scopeIDs)
	missingScopeIDs := s.runtime.missingScopeIDs(workspaceID, memberID, scopeIDs)
	if s.store != nil && len(missingScopeIDs) > 0 {
		s.enqueue(hydrateJob{
			kind:        hydrateJobSnapshot,
			workspaceID: workspaceID,
			memberID:    memberID,
			scopeIDs:    missingScopeIDs,
		})
	}
	return counts
}

// WorkspaceStates returns every externally-servable unread state for one
// member after the same bounded cold-member snapshot hydrate used by
// WorkspaceHasUnread. Normal clients should prefer snapshot + stream deltas
// over broad reads; this exists to build the initial workspace snapshot.
func (s *Service) WorkspaceStates(workspaceID string, memberID string) map[string]ScopeUnreadState {
	workspaceID = strings.TrimSpace(workspaceID)
	memberID = strings.TrimSpace(memberID)
	if s == nil || s.runtime == nil || workspaceID == "" || memberID == "" {
		return map[string]ScopeUnreadState{}
	}
	s.hydrateMemberSnapshotsIfNeeded(context.Background(), workspaceID, memberID)
	return s.runtime.ScopeStates(workspaceID, memberID, nil)
}

// ScopeStates returns loaded scope unread states and queues bounded snapshot
// hydrate work for missing cold-start entries.
func (s *Service) ScopeStates(workspaceID string, memberID string, scopeIDs []string) map[string]ScopeUnreadState {
	if s == nil || s.runtime == nil {
		return map[string]ScopeUnreadState{}
	}
	if len(scopeIDs) == 0 {
		workspaceID = strings.TrimSpace(workspaceID)
		memberID = strings.TrimSpace(memberID)
		if workspaceID != "" && memberID != "" {
			s.hydrateMemberSnapshotsIfNeeded(context.Background(), workspaceID, memberID)
		}
	}
	states := s.runtime.ScopeStates(workspaceID, memberID, scopeIDs)
	missingScopeIDs := s.runtime.missingScopeIDs(workspaceID, memberID, scopeIDs)
	if s.store != nil && len(missingScopeIDs) > 0 {
		s.enqueue(hydrateJob{
			kind:        hydrateJobSnapshot,
			workspaceID: workspaceID,
			memberID:    memberID,
			scopeIDs:    missingScopeIDs,
		})
	}
	return states
}

// CurrentUnreadDelta atomically snapshots one scope and its workspace
// aggregate for a live projection.
func (s *Service) CurrentUnreadDelta(workspaceID, userID, memberID, scopeID string, fallbackScopeType ScopeType) (UnreadDelta, bool) {
	if s == nil || s.runtime == nil {
		return UnreadDelta{}, false
	}
	return s.runtime.CurrentUnreadDelta(workspaceID, userID, memberID, scopeID, fallbackScopeType)
}

// ApplyMessage applies a committed human-visible message event when the write-side
// roster is already warm. If the roster is cold, it queues bounded roster
// warmup and later replays only process-local pending events. It never reads
// durable message tails to repair unread counts.
func (s *Service) ApplyMessage(event MessageStored) (ApplyResult, error) {
	if s == nil || s.runtime == nil {
		return ApplyResult{}, nil
	}
	result, err := s.runtime.ApplyMessage(event)
	if err == nil && result.Pending && !s.runtime.ScopeRosterReady(event.WorkspaceID, event.ScopeID) {
		s.enqueueRosterWarm(event.WorkspaceID, event.ScopeID, event.ScopeType)
	}
	return result, err
}

// ApplyNonMessageActivity applies a committed durable activity event when the
// write-side roster is already warm. Cold rosters are warmed in the background
// for process-local pending replay; durable activity tails are not replayed.
func (s *Service) ApplyNonMessageActivity(event NonMessageActivityStored) (ApplyResult, error) {
	if s == nil || s.runtime == nil {
		return ApplyResult{}, nil
	}
	result, err := s.runtime.ApplyNonMessageActivity(event)
	if err == nil && result.Pending && !s.runtime.ScopeRosterReady(event.WorkspaceID, event.ScopeID) {
		s.enqueueRosterWarm(event.WorkspaceID, event.ScopeID, event.ScopeType)
	}
	return result, err
}

// ApplyMessageRewound applies committed rewind deltas to owner-local memory.
func (s *Service) ApplyMessageRewound(event MessageRewound) (ApplyResult, error) {
	if s == nil || s.runtime == nil {
		return ApplyResult{}, nil
	}
	return s.runtime.ApplyMessageRewound(event)
}

// ApplyReadCursor applies a committed message read cursor update.
func (s *Service) ApplyReadCursor(event ReadCursorAdvanced) error {
	if s == nil || s.runtime == nil {
		return nil
	}
	if err := s.runtime.ApplyReadCursor(event); err != nil {
		return err
	}
	s.enqueueTaskSnapshotHydrateIfNeeded(event.WorkspaceID, event.MemberID, event.ScopeID)
	return nil
}

// ApplyActivityReadCursor applies a committed activity read cursor update.
func (s *Service) ApplyActivityReadCursor(event ActivityReadCursorAdvanced) error {
	if s == nil || s.runtime == nil {
		return nil
	}
	if err := s.runtime.ApplyActivityReadCursor(event); err != nil {
		return err
	}
	s.enqueueTaskSnapshotHydrateIfNeeded(event.WorkspaceID, event.MemberID, event.ScopeID)
	return nil
}

// ApplyManualUnread applies a committed manual unread marker update.
func (s *Service) ApplyManualUnread(event ManualUnreadChanged) error {
	if s == nil || s.runtime == nil {
		return nil
	}
	if err := s.runtime.ApplyManualUnread(event); err != nil {
		return err
	}
	s.enqueueTaskSnapshotHydrateIfNeeded(event.WorkspaceID, event.MemberID, event.ScopeID)
	return nil
}

// ApplyScopeInvalidated suppresses loaded state after a committed mutation
// that changes unread eligibility without an exact runtime delta.
func (s *Service) ApplyScopeInvalidated(event ScopeInvalidated) error {
	if s == nil || s.runtime == nil {
		return nil
	}
	return s.runtime.ApplyScopeInvalidated(event)
}

// ApplyScopeInvalidatedWithDeltas applies a committed suppression mutation and
// returns zero-state deltas for loaded entries.
func (s *Service) ApplyScopeInvalidatedWithDeltas(event ScopeInvalidated) ([]UnreadDelta, error) {
	if s == nil || s.runtime == nil {
		return nil, nil
	}
	return s.runtime.ApplyScopeInvalidatedWithDeltas(event)
}

func (s *Service) runSnapshotFlush(ctx context.Context) {
	if s.store == nil {
		return
	}
	cfg := s.runtime.Config()
	ticker := time.NewTicker(cfg.SnapshotFlushInterval)
	defer ticker.Stop()
	for {
		select {
		case <-ctx.Done():
			s.flushOnce(ctx, cfg.SnapshotFlushBatchSize)
			return
		case <-ticker.C:
			s.flushOnce(ctx, cfg.SnapshotFlushBatchSize)
		}
	}
}

func (s *Service) flushOnce(ctx context.Context, limit int) {
	s.snapshotPersistMu.Lock()
	defer s.snapshotPersistMu.Unlock()

	rows := s.runtime.FlushDirtySnapshots(limit, time.Now().UTC())
	if len(rows) == 0 {
		return
	}
	if err := s.store.UpsertSnapshots(ctx, rows); err != nil {
		s.logger.Warn("flush unread runtime snapshots", slog.Int("rows", len(rows)), slog.Any("error", err))
		s.runtime.RequeueDirtySnapshots(rows)
	}
}

func (s *Service) runHydrateWorker(ctx context.Context) {
	if s.store == nil {
		return
	}
	for {
		select {
		case <-ctx.Done():
			return
		case job := <-s.queue:
			err := s.runHydrateJob(ctx, job)
			s.finish(job, err)
		}
	}
}

func (s *Service) runHydrateJob(ctx context.Context, job hydrateJob) error {
	switch job.kind {
	case hydrateJobSnapshot:
		return s.hydrateSnapshotJob(ctx, job)
	case hydrateJobRoster:
		return s.hydrateRosterJob(ctx, job)
	}
	return nil
}

func (s *Service) hydrateRosterJob(ctx context.Context, job hydrateJob) error {
	store, ok := s.store.(scopeRosterStore)
	if !ok {
		return nil
	}
	workspaceID := strings.TrimSpace(job.workspaceID)
	scopeID := strings.TrimSpace(job.scopeID)
	if workspaceID == "" || scopeID == "" {
		return nil
	}
	if s.runtime.scopePermanentlyInvalidated(workspaceID, scopeID) || s.runtime.ScopeRosterReady(workspaceID, scopeID) {
		return nil
	}
	if err := s.runtime.BeginScopeRosterWarm(workspaceID, scopeID, job.scopeType); err != nil {
		return err
	}
	if s.runtime.scopePermanentlyInvalidated(workspaceID, scopeID) {
		return nil
	}
	// A refresh may have bumped the roster generation between the ready check
	// and Begin; abandon this hydrate so the refresh's next generation runs.
	if s.runtime.ScopeRosterWarmStale(workspaceID, scopeID) {
		return nil
	}

	afterMemberID := ""
	limit := s.runtime.Config().HydrateBatchSize
	for {
		if s.runtime.scopePermanentlyInvalidated(workspaceID, scopeID) {
			return nil
		}
		rows, err := store.LoadScopeRosterPage(ctx, workspaceID, scopeID, afterMemberID, limit)
		if err != nil {
			if s.runtime.scopePermanentlyInvalidated(workspaceID, scopeID) {
				return nil
			}
			s.logger.Warn("hydrate unread runtime roster", slog.String("workspace_id", workspaceID), slog.String("scope_id", scopeID), slog.Any("error", err))
			return err
		}
		if s.runtime.scopePermanentlyInvalidated(workspaceID, scopeID) {
			return nil
		}
		if len(rows) > 0 {
			if err := s.runtime.WarmScopeRosterPage(workspaceID, scopeID, job.scopeType, rows); err != nil {
				return err
			}
			// The pages were read before the refresh linearization point; stop
			// loading and abandon the hydrate without publishing.
			if s.runtime.ScopeRosterWarmStale(workspaceID, scopeID) {
				return nil
			}
			afterMemberID = rows[len(rows)-1].MemberID
		}
		if len(rows) < limit {
			break
		}
	}
	for {
		missingMemberIDs, ready, stale, err := s.runtime.finishScopeRosterWarmAfterTaskSnapshots(
			workspaceID,
			scopeID,
			job.scopeType,
			limit,
		)
		if err != nil {
			return err
		}
		if stale {
			return nil
		}
		if ready {
			break
		}
		for _, memberID := range missingMemberIDs {
			// This is a scope-local recovery read for a frozen Task audience,
			// not a full member bootstrap. hydrateSnapshotPage deliberately
			// does not mark memberSnapshotsHydrated.
			if _, ok := s.hydrateSnapshotPage(ctx, workspaceID, memberID, []string{scopeID}, ""); !ok {
				return fmt.Errorf("hydrate unread runtime Task recipient snapshot")
			}
		}
	}
	result, err := s.runtime.ApplyPendingForScope(workspaceID, scopeID)
	if err != nil {
		return err
	}
	s.emitUnreadDeltas(workspaceID, result.Deltas)
	return nil
}

func (s *Service) enqueueRosterWarm(workspaceID string, scopeID string, scopeType ScopeType) {
	if s == nil || s.store == nil {
		return
	}
	if _, ok := s.store.(scopeRosterStore); !ok {
		return
	}
	workspaceID = strings.TrimSpace(workspaceID)
	scopeID = strings.TrimSpace(scopeID)
	if workspaceID == "" || scopeID == "" ||
		s.runtime.scopePermanentlyInvalidated(workspaceID, scopeID) ||
		s.runtime.ScopeRosterReady(workspaceID, scopeID) {
		return
	}
	s.enqueue(hydrateJob{
		kind:        hydrateJobRoster,
		workspaceID: workspaceID,
		scopeID:     scopeID,
		scopeType:   scopeType,
	})
}

func (s *Service) emitUnreadDeltas(workspaceID string, deltas []UnreadDelta) {
	if s == nil || len(deltas) == 0 {
		return
	}
	s.mu.Lock()
	sink := s.deltaSink
	s.mu.Unlock()
	if sink == nil {
		return
	}
	for _, delta := range deltas {
		sink.EmitUnreadUpdated(workspaceID, delta)
	}
}

func (s *Service) hydrateSnapshotJob(ctx context.Context, job hydrateJob) error {
	workspaceID := strings.TrimSpace(job.workspaceID)
	memberID := strings.TrimSpace(job.memberID)
	scopeIDs := compactJobStrings(job.scopeIDs)
	limit := s.runtime.Config().HydrateBatchSize
	if len(scopeIDs) > 0 {
		for start := 0; start < len(scopeIDs); start += limit {
			end := start + limit
			if end > len(scopeIDs) {
				end = len(scopeIDs)
			}
			if _, ok := s.hydrateSnapshotPage(ctx, workspaceID, memberID, scopeIDs[start:end], ""); !ok {
				return fmt.Errorf("hydrate unread runtime snapshot page")
			}
		}
		return nil
	}

	afterScopeID := ""
	for {
		rows, ok := s.hydrateSnapshotPage(ctx, workspaceID, memberID, nil, afterScopeID)
		if !ok || len(rows) < limit {
			if !ok {
				return fmt.Errorf("hydrate unread runtime snapshot page")
			}
			s.markMemberSnapshotsHydrated(workspaceID, memberID)
			return nil
		}
		afterScopeID = rows[len(rows)-1].ScopeID
		if strings.TrimSpace(afterScopeID) == "" {
			s.markMemberSnapshotsHydrated(workspaceID, memberID)
			return nil
		}
	}
}

func (s *Service) hydrateMemberSnapshotsIfNeeded(ctx context.Context, workspaceID string, memberID string) {
	if s == nil || s.store == nil || s.runtime == nil {
		return
	}
	workspaceID = strings.TrimSpace(workspaceID)
	memberID = strings.TrimSpace(memberID)
	if workspaceID == "" || memberID == "" || s.memberSnapshotsAlreadyHydrated(workspaceID, memberID) {
		return
	}
	if store, ok := s.store.(visibleConversationScopeStore); ok {
		if err := s.hydrateVisibleMemberSnapshots(ctx, store, workspaceID, memberID); err != nil {
			s.logger.Warn("hydrate visible unread runtime member snapshots", slog.String("workspace_id", workspaceID), slog.String("member_id", memberID), slog.Any("error", err))
			return
		}
		s.markMemberSnapshotsHydrated(workspaceID, memberID)
		return
	}
	job := hydrateJob{kind: hydrateJobSnapshot, workspaceID: workspaceID, memberID: memberID}
	if err := s.hydrateSnapshotJob(ctx, job); err != nil {
		s.logger.Warn("hydrate unread runtime member snapshots", slog.String("workspace_id", workspaceID), slog.String("member_id", memberID), slog.Any("error", err))
	}
}

func (s *Service) hydrateVisibleMemberSnapshots(ctx context.Context, store visibleConversationScopeStore, workspaceID string, memberID string) error {
	limit := s.runtime.Config().HydrateBatchSize
	afterScopeID := ""
	for {
		scopeIDs, err := store.LoadVisibleConversationScopeIDsPage(ctx, workspaceID, memberID, afterScopeID, limit)
		if err != nil {
			return err
		}
		scopeIDs = compactJobStrings(scopeIDs)
		if len(scopeIDs) == 0 {
			return nil
		}
		for start := 0; start < len(scopeIDs); start += limit {
			end := start + limit
			if end > len(scopeIDs) {
				end = len(scopeIDs)
			}
			if _, ok := s.hydrateSnapshotPage(ctx, workspaceID, memberID, scopeIDs[start:end], ""); !ok {
				return fmt.Errorf("hydrate visible unread runtime snapshot page")
			}
		}
		if len(scopeIDs) < limit {
			return nil
		}
		afterScopeID = scopeIDs[len(scopeIDs)-1]
	}
}

func (s *Service) memberSnapshotsAlreadyHydrated(workspaceID string, memberID string) bool {
	if s == nil {
		return false
	}
	key := memberKey{workspaceID: strings.TrimSpace(workspaceID), memberID: strings.TrimSpace(memberID)}
	s.mu.Lock()
	defer s.mu.Unlock()
	_, ok := s.memberSnapshotsHydrated[key]
	return ok
}

func (s *Service) markMemberSnapshotsHydrated(workspaceID string, memberID string) {
	if s == nil {
		return
	}
	key := memberKey{workspaceID: strings.TrimSpace(workspaceID), memberID: strings.TrimSpace(memberID)}
	if key.workspaceID == "" || key.memberID == "" {
		return
	}
	s.mu.Lock()
	s.memberSnapshotsHydrated[key] = struct{}{}
	s.mu.Unlock()
}

func (s *Service) hydrateSnapshotPage(ctx context.Context, workspaceID string, memberID string, scopeIDs []string, afterScopeID string) ([]Snapshot, bool) {
	requestedScopeIDs := compactJobStrings(scopeIDs)
	rows, err := s.store.LoadSnapshotPage(
		ctx,
		workspaceID,
		memberID,
		requestedScopeIDs,
		afterScopeID,
		s.runtime.Config().HydrateBatchSize,
	)
	if err != nil {
		s.logger.Warn("hydrate unread runtime snapshots", slog.Any("error", err))
		return nil, false
	}
	hydrateRows, unavailableScopeIDs, ok := s.frontierValidatedSnapshotRows(ctx, workspaceID, memberID, rows)
	if !ok {
		return nil, false
	}
	for _, row := range hydrateRows {
		if err := s.runtime.HydrateSnapshotIfAbsent(row); err != nil {
			s.logger.Warn("apply unread runtime snapshot", slog.Any("error", err))
		}
	}
	s.runtime.MarkEntriesUnavailable(workspaceID, memberID, unavailableScopeIDs)
	s.markMissingSnapshotsUnavailable(workspaceID, memberID, requestedScopeIDs, rows)
	return rows, true
}

func (s *Service) frontierValidatedSnapshotRows(ctx context.Context, workspaceID string, memberID string, rows []Snapshot) ([]Snapshot, []string, bool) {
	if len(rows) == 0 {
		return rows, nil, true
	}
	store, ok := s.store.(snapshotFrontierMetadataStore)
	if !ok {
		return rows, nil, true
	}
	scopeIDs := make([]string, 0, len(rows))
	for _, row := range rows {
		if scopeID := strings.TrimSpace(row.ScopeID); scopeID != "" {
			scopeIDs = append(scopeIDs, scopeID)
		}
	}
	scopeIDs = compactJobStrings(scopeIDs)
	if len(scopeIDs) == 0 {
		return rows, nil, true
	}
	metadataRows, err := store.LoadFrontierMetadataPage(ctx, workspaceID, memberID, scopeIDs, "", s.runtime.Config().HydrateBatchSize)
	if err != nil {
		s.logger.Warn("hydrate unread runtime frontier metadata", slog.Any("error", err))
		return nil, nil, false
	}
	metadataByScopeID := make(map[string]FrontierMetadata, len(metadataRows))
	for _, metadata := range metadataRows {
		scopeID := strings.TrimSpace(metadata.ScopeID)
		if scopeID == "" {
			continue
		}
		metadataByScopeID[scopeID] = metadata
	}
	validated := make([]Snapshot, 0, len(rows))
	unavailableScopeIDs := make([]string, 0)
	for _, row := range rows {
		scopeID := strings.TrimSpace(row.ScopeID)
		metadata, ok := metadataByScopeID[scopeID]
		if !ok {
			unavailableScopeIDs = append(unavailableScopeIDs, scopeID)
			continue
		}
		validated = append(validated, reconcileSnapshotWithFrontier(row, metadata))
	}
	return validated, unavailableScopeIDs, true
}

func reconcileSnapshotWithFrontier(row Snapshot, metadata FrontierMetadata) Snapshot {
	row.ScopeType = metadata.ScopeType
	servingStateChanged := false
	if metadata.LastMessageReadSeq > row.LastMessageReadSeq {
		row.LastMessageReadSeq = metadata.LastMessageReadSeq
	}
	if metadata.LastActivityReadSeq > row.LastActivityReadSeq {
		row.LastActivityReadSeq = metadata.LastActivityReadSeq
	}
	if metadata.EffectiveFromSeq > row.EffectiveFromSeq {
		row.EffectiveFromSeq = metadata.EffectiveFromSeq
	}
	if metadata.ScopeMessageGeneration > row.ScopeMessageGeneration {
		row.ScopeMessageGeneration = metadata.ScopeMessageGeneration
	}
	if metadata.ScopeActivityGeneration > row.ScopeActivityGeneration {
		row.ScopeActivityGeneration = metadata.ScopeActivityGeneration
	}
	if metadata.MembershipGeneration > row.MembershipGeneration {
		row.MembershipGeneration = metadata.MembershipGeneration
	}
	if metadata.NotificationGeneration > row.NotificationGeneration {
		row.NotificationGeneration = metadata.NotificationGeneration
	}
	if metadata.ManualUnreadGeneration > row.ManualUnreadGeneration {
		row.ManualUnreadGeneration = metadata.ManualUnreadGeneration
	}
	if metadata.RewindGeneration > row.RewindGeneration {
		row.RewindGeneration = metadata.RewindGeneration
	}
	row.NotificationState = strings.TrimSpace(metadata.NotificationState)
	row.MembershipVisibility = strings.TrimSpace(metadata.MembershipVisibility)

	if !frontierMetadataEligibleForUnread(metadata) {
		if row.MessageUnreadCount != 0 || row.ActivityUnreadCount != 0 || row.ManualUnread {
			servingStateChanged = true
		}
		row.MessageUnreadCount = 0
		row.ActivityUnreadCount = 0
		row.ManualUnread = false
	} else {
		if row.MessageUnreadCount > 0 &&
			row.LatestMessageAccountedSeq > 0 &&
			metadata.LastMessageReadSeq >= row.LatestMessageAccountedSeq &&
			frontierMessageCursorCoversLatest(metadata) {
			row.MessageUnreadCount = 0
			servingStateChanged = true
		}
		if row.ActivityUnreadCount > 0 &&
			row.LatestActivityAccountedSeq > 0 &&
			metadata.LastActivityReadSeq >= row.LatestActivityAccountedSeq &&
			frontierActivityCursorCoversLatest(metadata) {
			row.ActivityUnreadCount = 0
			servingStateChanged = true
		}
		if row.ManualUnread != metadata.ManualUnread {
			row.ManualUnread = metadata.ManualUnread
			servingStateChanged = true
		}
	}
	if metadata.LastMessageReadSeq > row.LatestMessageAccountedSeq {
		row.LatestMessageAccountedSeq = metadata.LastMessageReadSeq
	}
	if metadata.LastActivityReadSeq > row.LatestActivityAccountedSeq {
		row.LatestActivityAccountedSeq = metadata.LastActivityReadSeq
	}
	if servingStateChanged && (!frontierMetadataEligibleForUnread(metadata) || frontierCursorsCoverLatest(metadata)) {
		row.ServingState = ServingStateReady
	}
	if row.ServingState == "" {
		row.ServingState = ServingStateSnapshotServing
	}
	return row
}

func (s *Service) markMissingSnapshotsUnavailable(workspaceID string, memberID string, requestedScopeIDs []string, rows []Snapshot) {
	if len(requestedScopeIDs) == 0 {
		return
	}
	found := make(map[string]struct{}, len(rows))
	for _, row := range rows {
		scopeID := strings.TrimSpace(row.ScopeID)
		if scopeID != "" {
			found[scopeID] = struct{}{}
		}
	}
	missing := make([]string, 0)
	for _, scopeID := range requestedScopeIDs {
		if _, ok := found[scopeID]; ok {
			continue
		}
		missing = append(missing, scopeID)
	}
	s.runtime.MarkEntriesUnavailable(workspaceID, memberID, missing)
}

func (s *Service) runHydrateRetry(ctx context.Context) {
	if s.store == nil {
		return
	}
	interval := s.runtime.Config().RecoveryRetryInterval
	ticker := time.NewTicker(interval)
	defer ticker.Stop()
	for {
		select {
		case <-ctx.Done():
			return
		case <-ticker.C:
			s.retryHydrateJobs()
		}
	}
}

func (s *Service) retryHydrateJobs() {
	for _, job := range s.dueRetryJobs(time.Now()) {
		if s.jobStillPending(job) {
			s.enqueue(job)
		} else {
			s.clearRetry(job)
		}
	}
}

func (s *Service) enqueue(job hydrateJob) {
	if s == nil || s.store == nil {
		return
	}
	job.workspaceID = strings.TrimSpace(job.workspaceID)
	job.memberID = strings.TrimSpace(job.memberID)
	job.scopeID = strings.TrimSpace(job.scopeID)
	job.scopeIDs = compactJobStrings(job.scopeIDs)
	job.memberIDs = compactJobStrings(job.memberIDs)
	if job.workspaceID == "" {
		return
	}
	if job.kind == hydrateJobSnapshot && job.memberID == "" {
		return
	}
	if job.kind == hydrateJobRoster && job.scopeID == "" {
		return
	}
	if job.kind == hydrateJobRoster && s.runtime != nil && s.runtime.scopePermanentlyInvalidated(job.workspaceID, job.scopeID) {
		return
	}
	key := job.key()
	s.mu.Lock()
	if _, ok := s.inflight[key]; ok {
		s.mu.Unlock()
		return
	}
	s.inflight[key] = struct{}{}
	s.mu.Unlock()

	select {
	case s.queue <- job:
	default:
		s.finish(job, fmt.Errorf("drop unread runtime hydrate job"))
		s.logger.Warn("drop unread runtime hydrate job", slog.String("kind", string(job.kind)))
	}
}

func (s *Service) finish(job hydrateJob, errs ...error) {
	var err error
	if len(errs) > 0 {
		err = errs[0]
	}
	key := job.key()
	s.mu.Lock()
	delete(s.inflight, key)
	if err == nil {
		delete(s.retry, key)
	}
	s.mu.Unlock()
	if err != nil {
		s.scheduleRetry(job)
		return
	}
	// A Task event can arrive after the snapshot barrier publishes the roster
	// but before this worker releases its in-flight key. If its frozen active
	// recipient is still cold, Runtime turns the roster cold again; enqueueing
	// from the live path is intentionally deduplicated against this job. Check
	// once after releasing the key so that race cannot strand pending work.
	if job.kind == hydrateJobRoster && s.jobStillPending(job) {
		s.enqueue(job)
	}
}

func (s *Service) scheduleRetry(job hydrateJob) {
	if s == nil {
		return
	}
	key := job.key()
	now := time.Now()
	s.mu.Lock()
	state := s.retry[key]
	state.job = job
	state.nextAt = now.Add(s.retryDelay(state.attempts))
	state.attempts++
	s.retry[key] = state
	s.mu.Unlock()
}

func (s *Service) clearRetry(job hydrateJob) {
	if s == nil {
		return
	}
	s.mu.Lock()
	delete(s.retry, job.key())
	s.mu.Unlock()
}

func (s *Service) dueRetryJobs(now time.Time) []hydrateJob {
	if s == nil {
		return nil
	}
	s.mu.Lock()
	defer s.mu.Unlock()
	jobs := make([]hydrateJob, 0)
	for key, state := range s.retry {
		if _, ok := s.inflight[key]; ok || state.nextAt.After(now) {
			continue
		}
		jobs = append(jobs, state.job)
	}
	return jobs
}

func (s *Service) retryDelay(attempts int) time.Duration {
	if attempts <= 0 || s == nil || s.runtime == nil {
		return 0
	}
	interval := s.runtime.Config().RecoveryRetryInterval
	multiplier := 1
	for i := 1; i < attempts && multiplier < 16; i++ {
		multiplier *= 2
	}
	if multiplier > 16 {
		multiplier = 16
	}
	return time.Duration(multiplier) * interval
}

func (s *Service) jobStillPending(job hydrateJob) bool {
	if s == nil || s.runtime == nil {
		return false
	}
	switch job.kind {
	case hydrateJobSnapshot:
		if len(job.scopeIDs) == 0 {
			return true
		}
		return len(s.runtime.missingScopeIDs(job.workspaceID, job.memberID, job.scopeIDs)) > 0
	case hydrateJobRoster:
		return !s.runtime.scopePermanentlyInvalidated(job.workspaceID, job.scopeID) &&
			!s.runtime.ScopeRosterReady(job.workspaceID, job.scopeID)
	default:
		return false
	}
}

func (job hydrateJob) key() string {
	return string(job.kind) + "\x00" + strings.TrimSpace(job.workspaceID) + "\x00" + strings.TrimSpace(job.memberID) + "\x00" + strings.TrimSpace(job.scopeID) + "\x00" + strings.Join(compactJobStrings(job.scopeIDs), ",") + "\x00" + strings.Join(compactJobStrings(job.memberIDs), ",")
}

func compactJobStrings(values []string) []string {
	if len(values) == 0 {
		return nil
	}
	seen := make(map[string]struct{}, len(values))
	compacted := make([]string, 0, len(values))
	for _, value := range values {
		value = strings.TrimSpace(value)
		if value == "" {
			continue
		}
		if _, ok := seen[value]; ok {
			continue
		}
		seen[value] = struct{}{}
		compacted = append(compacted, value)
	}
	return compacted
}