// Package unread persists and recovers owner-local unread runtime state.
package unread

import (
	"context"
	"database/sql"
	"fmt"
	"sort"
	"strings"

	"github.com/tidbcloud/AgentTeam/internal/coredb"
	"github.com/tidbcloud/AgentTeam/internal/store/message"
	unreadservice "github.com/tidbcloud/AgentTeam/internal/unread"
)

// Store owns bounded TiDB access for unread runtime snapshot and recovery.
type Store struct {
	db coredb.DB
}

// New creates an unread runtime store.
func New(db coredb.DB) *Store {
	return &Store{db: db}
}

// SnapshotPageRequest bounds snapshot reads to one member and either a visible
// scope set or a primary-key page.
type SnapshotPageRequest struct {
	WorkspaceID  string
	MemberID     string
	ScopeIDs     []string
	AfterScopeID string
	Limit        int
}

// MetadataPageRequest bounds frontier metadata reads to one member and either
// a visible scope set or a primary-key page.
type MetadataPageRequest struct {
	WorkspaceID  string
	MemberID     string
	ScopeIDs     []string
	AfterScopeID string
	Limit        int
}

// RosterPageRequest bounds scope roster hydration to one scope page.
type RosterPageRequest struct {
	WorkspaceID   string
	ScopeID       string
	AfterMemberID string
	Limit         int
}

// TailPageRequest bounds durable tail recovery to one scope and one seq range.
type TailPageRequest struct {
	WorkspaceID string
	ScopeID     string
	AfterSeq    int64
	Limit       int
}

func (s *Store) LoadActiveHumanUserIDForMember(ctx context.Context, workspaceID string, memberID string) (string, error) {
	if s == nil || s.db == nil {
		return "", fmt.Errorf("unread runtime store is not configured")
	}
	var userID string
	if err := s.db.QueryRow(ctx, `
SELECT user_id
FROM members
WHERE workspace_id = $1
  AND id = $2
  AND type = 'human'
  AND status = 'active'
`, strings.TrimSpace(workspaceID), strings.TrimSpace(memberID)).Scan(&userID); err != nil {
		return "", fmt.Errorf("load unread runtime member user id: %w", err)
	}
	return strings.TrimSpace(userID), nil
}

// UpsertSnapshots flushes dirty runtime snapshots. Callers must pass an already
// coalesced, configured-size batch.
func (s *Store) UpsertSnapshots(ctx context.Context, snapshots []unreadservice.Snapshot) error {
	return s.upsertSnapshots(ctx, snapshots, false)
}

// UpsertTaskBarrierSnapshots persists the exact Task entries guarded by one
// targeted-v2 completion barrier. Unlike the periodic writer, this path may
// advance the protected Task aggregate after proving it descends from the
// currently stored shadow.
func (s *Store) UpsertTaskBarrierSnapshots(ctx context.Context, snapshots []unreadservice.Snapshot) error {
	return s.upsertSnapshots(ctx, snapshots, true)
}

func (s *Store) upsertSnapshots(
	ctx context.Context,
	snapshots []unreadservice.Snapshot,
	taskBarrier bool,
) error {
	if len(snapshots) == 0 {
		return nil
	}
	if s == nil || s.db == nil {
		return fmt.Errorf("unread runtime store is not configured")
	}
	prepared, err := prepareSnapshotsForUpsert(snapshots, taskBarrier)
	if err != nil {
		return err
	}

	return coredb.WithTx(ctx, s.db, func(tx coredb.Tx) error {
		if taskBarrier {
			if err := lockTaskSnapshotRows(ctx, tx, prepared); err != nil {
				return err
			}
			stored, err := loadLockedTaskSnapshots(ctx, tx, prepared)
			if err != nil {
				return err
			}
			for index := range prepared {
				key := snapshotStorageKeyFor(prepared[index].snapshot)
				current, ok := stored[key]
				if !ok {
					return fmt.Errorf(
						"locked Task unread runtime snapshot is missing: %s/%s/%s",
						key.workspaceID,
						key.memberID,
						key.scopeID,
					)
				}
				var advanced unreadservice.Snapshot
				if current.hasTaskShadow {
					advanced, err = advanceTaskSnapshotBarrier(
						current.snapshot,
						prepared[index].snapshot,
					)
				} else {
					advanced, err = establishTaskSnapshotBarrier(
						current.snapshot,
						prepared[index].snapshot,
					)
				}
				if err != nil {
					return fmt.Errorf(
						"advance Task unread snapshot barrier %s/%s/%s: %w",
						key.workspaceID,
						key.memberID,
						key.scopeID,
						err,
					)
				}
				prepared[index].snapshot = advanced
				version, payload, err := encodeTaskSnapshotShadow(advanced)
				if err != nil {
					return err
				}
				prepared[index].shadowFormatVersion = version
				prepared[index].shadowPayload = payload
			}
		}
		for _, snapshot := range prepared {
			if err := upsertSnapshot(ctx, tx, snapshot); err != nil {
				return err
			}
		}
		return nil
	})
}

type preparedSnapshot struct {
	snapshot            unreadservice.Snapshot
	shadowFormatVersion int64
	shadowPayload       string
}

type snapshotStorageKey struct {
	workspaceID string
	memberID    string
	scopeID     string
}

type storedTaskSnapshot struct {
	snapshot      unreadservice.Snapshot
	hasTaskShadow bool
}

func prepareSnapshotsForUpsert(
	snapshots []unreadservice.Snapshot,
	taskBarrier bool,
) ([]preparedSnapshot, error) {
	prepared := make([]preparedSnapshot, 0, len(snapshots))
	seen := make(map[snapshotStorageKey]struct{}, len(snapshots))
	for _, snapshot := range snapshots {
		key := snapshotStorageKeyFor(snapshot)
		if key.workspaceID == "" || key.memberID == "" || key.scopeID == "" {
			return nil, fmt.Errorf("unread runtime snapshot identity is required")
		}
		if _, ok := seen[key]; ok {
			return nil, fmt.Errorf(
				"unread runtime snapshots must be coalesced before persistence: %s/%s/%s",
				key.workspaceID,
				key.memberID,
				key.scopeID,
			)
		}
		seen[key] = struct{}{}
		version := int64(0)
		payload := ""
		if taskBarrier {
			if !isTaskSnapshot(snapshot) {
				return nil, fmt.Errorf(
					"targeted unread snapshot barrier requires a Task scope: %s/%s/%s",
					key.workspaceID,
					key.memberID,
					key.scopeID,
				)
			}
			var err error
			version, payload, err = encodeTaskSnapshotShadow(snapshot)
			if err != nil {
				return nil, err
			}
		}
		prepared = append(prepared, preparedSnapshot{
			snapshot:            snapshot,
			shadowFormatVersion: version,
			shadowPayload:       payload,
		})
	}
	sort.Slice(prepared, func(left, right int) bool {
		leftKey := snapshotStorageKeyFor(prepared[left].snapshot)
		rightKey := snapshotStorageKeyFor(prepared[right].snapshot)
		if leftKey.workspaceID != rightKey.workspaceID {
			return leftKey.workspaceID < rightKey.workspaceID
		}
		if leftKey.scopeID != rightKey.scopeID {
			return leftKey.scopeID < rightKey.scopeID
		}
		return leftKey.memberID < rightKey.memberID
	})
	return prepared, nil
}

func snapshotStorageKeyFor(snapshot unreadservice.Snapshot) snapshotStorageKey {
	return snapshotStorageKey{
		workspaceID: strings.TrimSpace(snapshot.WorkspaceID),
		memberID:    strings.TrimSpace(snapshot.MemberID),
		scopeID:     strings.TrimSpace(snapshot.ScopeID),
	}
}

// lockTaskSnapshotRows performs one bounded write-lock acquisition for the
// Task rows in this barrier batch. Inserting an absent row first avoids
// the missing-row gap where two new owners could both observe no shadow and
// then overwrite each other. Existing rows receive a true no-op update.
func lockTaskSnapshotRows(ctx context.Context, tx coredb.Tx, snapshots []preparedSnapshot) error {
	taskSnapshots := make([]preparedSnapshot, 0, len(snapshots))
	for _, snapshot := range snapshots {
		if isTaskSnapshot(snapshot.snapshot) {
			taskSnapshots = append(taskSnapshots, snapshot)
		}
	}
	if len(taskSnapshots) == 0 {
		return nil
	}

	var query strings.Builder
	query.WriteString(`
INSERT INTO member_scope_unread_runtime_snapshots (
  workspace_id, member_id, scope_id, updated_at,
  task_snapshot_format_version, task_snapshot_json
) VALUES `)
	args := make([]any, 0, len(taskSnapshots)*6)
	param := 1
	for index, snapshot := range taskSnapshots {
		if index > 0 {
			query.WriteString(",")
		}
		query.WriteString(fmt.Sprintf(
			"\n  ($%d, $%d, $%d, $%d, $%d, $%d)",
			param,
			param+1,
			param+2,
			param+3,
			param+4,
			param+5,
		))
		args = append(args,
			snapshot.snapshot.WorkspaceID,
			snapshot.snapshot.MemberID,
			snapshot.snapshot.ScopeID,
			snapshot.snapshot.UpdatedAt,
			snapshot.shadowFormatVersion,
			snapshot.shadowPayload,
		)
		param += 6
	}
	query.WriteString(`
ON DUPLICATE KEY UPDATE
  scope_id = VALUES(scope_id)
`)
	if _, err := tx.Exec(ctx, query.String(), args...); err != nil {
		return fmt.Errorf("lock Task unread runtime snapshot rows: %w", err)
	}
	return nil
}

func loadLockedTaskSnapshots(
	ctx context.Context,
	tx coredb.Tx,
	snapshots []preparedSnapshot,
) (map[snapshotStorageKey]storedTaskSnapshot, error) {
	taskSnapshots := make([]preparedSnapshot, 0, len(snapshots))
	for _, snapshot := range snapshots {
		if isTaskSnapshot(snapshot.snapshot) {
			taskSnapshots = append(taskSnapshots, snapshot)
		}
	}
	if len(taskSnapshots) == 0 {
		return map[snapshotStorageKey]storedTaskSnapshot{}, nil
	}

	var query strings.Builder
	query.WriteString(snapshotSelectColumns)
	query.WriteString(`
FROM member_scope_unread_runtime_snapshots snap
WHERE `)
	args := make([]any, 0, len(taskSnapshots)*3)
	param := 1
	for index, snapshot := range taskSnapshots {
		if index > 0 {
			query.WriteString("\n   OR ")
		}
		query.WriteString(fmt.Sprintf(
			"(snap.workspace_id = $%d AND snap.member_id = $%d AND snap.scope_id = $%d)",
			param,
			param+1,
			param+2,
		))
		args = append(args,
			snapshot.snapshot.WorkspaceID,
			snapshot.snapshot.MemberID,
			snapshot.snapshot.ScopeID,
		)
		param += 3
	}
	query.WriteString(`
ORDER BY snap.workspace_id, snap.scope_id, snap.member_id
FOR UPDATE
`)
	rows, err := tx.Query(ctx, query.String(), args...)
	if err != nil {
		return nil, fmt.Errorf("load locked Task unread runtime snapshots: %w", err)
	}
	defer rows.Close()

	stored := make(map[snapshotStorageKey]storedTaskSnapshot, len(taskSnapshots))
	for rows.Next() {
		snapshot, shadowFormatVersion, err := scanStoredSnapshotWithFormat(rows)
		if err != nil {
			return nil, err
		}
		stored[snapshotStorageKeyFor(snapshot)] = storedTaskSnapshot{
			snapshot:      snapshot,
			hasTaskShadow: shadowFormatVersion > 0,
		}
	}
	if err := rows.Err(); err != nil {
		return nil, fmt.Errorf("iterate locked Task unread runtime snapshots: %w", err)
	}
	if len(stored) != len(taskSnapshots) {
		return nil, fmt.Errorf(
			"locked Task unread runtime snapshot rows = %d, want %d",
			len(stored),
			len(taskSnapshots),
		)
	}
	return stored, nil
}

func upsertSnapshot(ctx context.Context, tx coredb.Tx, prepared preparedSnapshot) error {
	snapshot := prepared.snapshot
	var shadowPayload any
	if prepared.shadowFormatVersion > 0 {
		shadowPayload = prepared.shadowPayload
	}
	if _, err := tx.Exec(ctx, `
INSERT INTO member_scope_unread_runtime_snapshots (
  workspace_id, member_id, scope_id,
  message_unread_count, activity_unread_count, manual_unread,
  last_message_read_seq, last_activity_read_seq, effective_from_seq,
  latest_message_accounted_seq, latest_activity_accounted_seq,
  notification_state, membership_visibility,
  scope_message_generation, scope_activity_generation,
  membership_generation, notification_generation, manual_unread_generation, rewind_generation,
  task_snapshot_format_version, task_snapshot_json,
  updated_at
) VALUES (
  $1, $2, $3,
  $4, $5, $6,
  $7, $8, $9,
  $10, $11,
  $12, $13,
  $14, $15,
  $16, $17, $18, $19,
  $20, $21,
  $22
)
ON DUPLICATE KEY UPDATE
  message_unread_count = VALUES(message_unread_count),
  activity_unread_count = VALUES(activity_unread_count),
  manual_unread = VALUES(manual_unread),
  last_message_read_seq = VALUES(last_message_read_seq),
  last_activity_read_seq = VALUES(last_activity_read_seq),
  effective_from_seq = VALUES(effective_from_seq),
  latest_message_accounted_seq = VALUES(latest_message_accounted_seq),
  latest_activity_accounted_seq = VALUES(latest_activity_accounted_seq),
  notification_state = VALUES(notification_state),
  membership_visibility = VALUES(membership_visibility),
  scope_message_generation = VALUES(scope_message_generation),
  scope_activity_generation = VALUES(scope_activity_generation),
  membership_generation = VALUES(membership_generation),
  notification_generation = VALUES(notification_generation),
  manual_unread_generation = VALUES(manual_unread_generation),
  rewind_generation = VALUES(rewind_generation),
  task_snapshot_json = CASE
    WHEN VALUES(task_snapshot_format_version) > 0
      AND VALUES(task_snapshot_format_version) >= task_snapshot_format_version
    THEN VALUES(task_snapshot_json)
    ELSE task_snapshot_json
  END,
  task_snapshot_format_version = GREATEST(
    task_snapshot_format_version,
    VALUES(task_snapshot_format_version)
  ),
  updated_at = VALUES(updated_at)
`, snapshot.WorkspaceID, snapshot.MemberID, snapshot.ScopeID,
		snapshot.MessageUnreadCount, snapshot.ActivityUnreadCount, snapshot.ManualUnread,
		snapshot.LastMessageReadSeq, snapshot.LastActivityReadSeq, snapshot.EffectiveFromSeq,
		snapshot.LatestMessageAccountedSeq, snapshot.LatestActivityAccountedSeq,
		snapshot.NotificationState, snapshot.MembershipVisibility,
		snapshot.ScopeMessageGeneration, snapshot.ScopeActivityGeneration,
		snapshot.MembershipGeneration, snapshot.NotificationGeneration, snapshot.ManualUnreadGeneration, snapshot.RewindGeneration,
		prepared.shadowFormatVersion, shadowPayload,
		snapshot.UpdatedAt); err != nil {
		return fmt.Errorf("upsert unread runtime snapshot: %w", err)
	}
	return nil
}

// LoadSnapshotPage loads a bounded snapshot page for lazy hydrate.
func (s *Store) LoadSnapshotPage(ctx context.Context, request SnapshotPageRequest) ([]unreadservice.Snapshot, error) {
	if s == nil || s.db == nil {
		return nil, fmt.Errorf("unread runtime store is not configured")
	}
	request.normalize()
	if err := request.validate(); err != nil {
		return nil, err
	}

	query, args, err := buildSnapshotPageQuery(request)
	if err != nil {
		return nil, err
	}
	rows, err := s.db.Query(ctx, query, args...)
	if err != nil {
		return nil, fmt.Errorf("load unread runtime snapshots: %w", err)
	}
	defer rows.Close()

	snapshots := make([]unreadservice.Snapshot, 0, request.Limit)
	for rows.Next() {
		snapshot, err := scanStoredSnapshot(rows)
		if err != nil {
			return nil, err
		}
		snapshot.ServingState = unreadservice.ServingStateSnapshotServing
		snapshots = append(snapshots, snapshot)
	}
	if err := rows.Err(); err != nil {
		return nil, fmt.Errorf("iterate unread runtime snapshots: %w", err)
	}
	return snapshots, nil
}

type snapshotScanner interface {
	Scan(...any) error
}

func scanStoredSnapshot(scanner snapshotScanner) (unreadservice.Snapshot, error) {
	snapshot, _, err := scanStoredSnapshotWithFormat(scanner)
	return snapshot, err
}

func scanStoredSnapshotWithFormat(
	scanner snapshotScanner,
) (unreadservice.Snapshot, int64, error) {
	var snapshot unreadservice.Snapshot
	var shadowFormatVersion int64
	var shadowPayload []byte
	if err := scanner.Scan(
		&snapshot.WorkspaceID,
		&snapshot.MemberID,
		&snapshot.ScopeID,
		&snapshot.MessageUnreadCount,
		&snapshot.ActivityUnreadCount,
		&snapshot.ManualUnread,
		&snapshot.LastMessageReadSeq,
		&snapshot.LastActivityReadSeq,
		&snapshot.EffectiveFromSeq,
		&snapshot.LatestMessageAccountedSeq,
		&snapshot.LatestActivityAccountedSeq,
		&snapshot.NotificationState,
		&snapshot.MembershipVisibility,
		&snapshot.ScopeMessageGeneration,
		&snapshot.ScopeActivityGeneration,
		&snapshot.MembershipGeneration,
		&snapshot.NotificationGeneration,
		&snapshot.ManualUnreadGeneration,
		&snapshot.RewindGeneration,
		&snapshot.UpdatedAt,
		&shadowFormatVersion,
		&shadowPayload,
	); err != nil {
		return unreadservice.Snapshot{}, 0, fmt.Errorf("scan unread runtime snapshot: %w", err)
	}
	merged, err := mergeStoredTaskSnapshot(snapshot, shadowFormatVersion, shadowPayload)
	if err != nil {
		return unreadservice.Snapshot{}, 0, fmt.Errorf(
			"merge unread runtime snapshot %s/%s/%s: %w",
			snapshot.WorkspaceID,
			snapshot.MemberID,
			snapshot.ScopeID,
			err,
		)
	}
	return merged, shadowFormatVersion, nil
}

// LoadVisibleConversationScopeIDsPage returns one bounded page of scopes that
// participate in the member's Inbox. It reuses the exact visibility predicate
// used by conversation listing and bulk mark-read.
func (s *Store) LoadVisibleConversationScopeIDsPage(ctx context.Context, workspaceID string, memberID string, userID string, afterScopeID string, limit int) ([]string, error) {
	if s == nil || s.db == nil {
		return nil, fmt.Errorf("unread runtime store is not configured")
	}
	query, args, err := message.BuildVisibleConversationScopeIDsForMemberPageQuery(
		strings.TrimSpace(workspaceID),
		strings.TrimSpace(memberID),
		strings.TrimSpace(userID),
		strings.TrimSpace(afterScopeID),
		limit,
	)
	if err != nil {
		return nil, err
	}
	rows, err := s.db.Query(ctx, query, args...)
	if err != nil {
		return nil, fmt.Errorf("load visible conversation scope ids: %w", err)
	}
	defer rows.Close()

	scopeIDs := make([]string, 0, limit)
	for rows.Next() {
		var scopeID string
		if err := rows.Scan(&scopeID); err != nil {
			return nil, fmt.Errorf("scan visible conversation scope id: %w", err)
		}
		if scopeID = strings.TrimSpace(scopeID); scopeID != "" {
			scopeIDs = append(scopeIDs, scopeID)
		}
	}
	if err := rows.Err(); err != nil {
		return nil, fmt.Errorf("iterate visible conversation scope ids: %w", err)
	}
	rows.Close()

	// Task Inbox rows are materialized in im_conversation_index rather than the
	// legacy message-conversation CTE above. Include their visible scope IDs in
	// the same lexical page so reconnect bootstrap can hydrate persisted Task
	// unread before runtime.unread_reset makes omission authoritative.
	taskRows, err := s.db.Query(ctx, buildVisibleTaskConversationScopeIDsPageQuery(),
		strings.TrimSpace(workspaceID),
		strings.TrimSpace(memberID),
		strings.TrimSpace(afterScopeID),
		limit,
	)
	if err != nil {
		return nil, fmt.Errorf("load visible Task conversation scope ids: %w", err)
	}
	defer taskRows.Close()
	for taskRows.Next() {
		var scopeID string
		if err := taskRows.Scan(&scopeID); err != nil {
			return nil, fmt.Errorf("scan visible Task conversation scope id: %w", err)
		}
		if scopeID = strings.TrimSpace(scopeID); scopeID != "" {
			scopeIDs = append(scopeIDs, scopeID)
		}
	}
	if err := taskRows.Err(); err != nil {
		return nil, fmt.Errorf("iterate visible Task conversation scope ids: %w", err)
	}

	sort.Strings(scopeIDs)
	compacted := scopeIDs[:0]
	for _, scopeID := range scopeIDs {
		if len(compacted) > 0 && compacted[len(compacted)-1] == scopeID {
			continue
		}
		compacted = append(compacted, scopeID)
		if len(compacted) == limit {
			break
		}
	}
	return compacted, nil
}

func buildVisibleTaskConversationScopeIDsPageQuery() string {
	return `
SELECT idx.scope_id
FROM im_conversation_index idx
JOIN scopes task_scope
  ON task_scope.workspace_id = idx.workspace_id
 AND task_scope.id = idx.scope_id
 AND task_scope.type = 'task'
 AND task_scope.status = 'active'
 AND task_scope.archived_at IS NULL
JOIN scope_memberships self_sm
  ON self_sm.scope_id = task_scope.id
 AND self_sm.member_id = idx.member_id
 AND self_sm.removed_at IS NULL
 AND self_sm.archived_at IS NULL
 AND self_sm.effective_to_seq IS NULL
LEFT JOIN conversation_member_states cms
  ON cms.scope_id = task_scope.id
 AND cms.member_id = idx.member_id
WHERE idx.workspace_id = $1
  AND idx.member_id = $2
  AND idx.kind = 'task'
  AND idx.scope_id > $3
  AND idx.archived_at IS NULL
  AND NOT (
    cms.hidden_at IS NOT NULL
    AND COALESCE(task_scope.latest_seq, 0) <= COALESCE(cms.hidden_through_seq, 0)
  )
ORDER BY idx.scope_id
LIMIT $4
`
}

// LoadFrontierMetadataPage loads metadata-only freshness facts.
func (s *Store) LoadFrontierMetadataPage(ctx context.Context, request MetadataPageRequest) ([]unreadservice.FrontierMetadata, error) {
	if s == nil || s.db == nil {
		return nil, fmt.Errorf("unread runtime store is not configured")
	}
	request.normalize()
	if err := request.validate(); err != nil {
		return nil, err
	}

	query, args, err := buildMetadataPageQuery(request)
	if err != nil {
		return nil, err
	}
	rows, err := s.db.Query(ctx, query, args...)
	if err != nil {
		return nil, fmt.Errorf("load unread runtime metadata: %w", err)
	}
	defer rows.Close()

	metadata := make([]unreadservice.FrontierMetadata, 0, request.Limit)
	for rows.Next() {
		var item unreadservice.FrontierMetadata
		var notificationState sql.NullString
		if err := rows.Scan(
			&item.WorkspaceID,
			&item.MemberID,
			&item.ScopeID,
			&item.ScopeType,
			&item.LatestMessageSeq,
			&item.LatestActivitySeq,
			&item.RewindGeneration,
			&item.LastMessageReadSeq,
			&item.LastActivityReadSeq,
			&item.EffectiveFromSeq,
			&item.MembershipGeneration,
			&notificationState,
			&item.NotificationGeneration,
			&item.MembershipVisibility,
			&item.ManualUnread,
			&item.ManualUnreadGeneration,
		); err != nil {
			return nil, fmt.Errorf("scan unread runtime metadata: %w", err)
		}
		item.NotificationState = notificationState.String
		item.ScopeMessageGeneration = item.LatestMessageSeq
		item.ScopeActivityGeneration = item.LatestActivitySeq
		metadata = append(metadata, item)
	}
	if err := rows.Err(); err != nil {
		return nil, fmt.Errorf("iterate unread runtime metadata: %w", err)
	}
	return metadata, nil
}

// LoadScopeRosterPage loads one bounded scope roster page.
func (s *Store) LoadScopeRosterPage(ctx context.Context, request RosterPageRequest) ([]unreadservice.RosterMember, error) {
	if s == nil || s.db == nil {
		return nil, fmt.Errorf("unread runtime store is not configured")
	}
	request.normalize()
	if err := request.validate(); err != nil {
		return nil, err
	}

	rows, err := s.db.Query(ctx, `
SELECT
  sm.member_id,
  m.user_id,
  sm.last_read_seq,
  sm.last_activity_global_seq,
  sm.effective_from_seq,
  COALESCE(sns.state, ''),
  sm.visibility
FROM scope_memberships sm
JOIN scopes scope
  ON scope.workspace_id = $1
 AND scope.id = sm.scope_id
 AND scope.type IN ('channel', 'dm', 'thread', 'task', 'service_account')
JOIN members m
  ON m.id = sm.member_id
 AND m.workspace_id = $1
 AND m.status = 'active'
LEFT JOIN scope_notification_settings sns
  ON sns.scope_id = sm.scope_id
 AND sns.member_id = sm.member_id
WHERE sm.scope_id = $2
  AND sm.member_id > $3
  AND sm.removed_at IS NULL
  AND sm.archived_at IS NULL
  AND sm.effective_to_seq IS NULL
ORDER BY sm.member_id
LIMIT $4
`, request.WorkspaceID, request.ScopeID, request.AfterMemberID, request.Limit)
	if err != nil {
		return nil, fmt.Errorf("load unread runtime roster: %w", err)
	}
	defer rows.Close()

	members := make([]unreadservice.RosterMember, 0, request.Limit)
	for rows.Next() {
		var member unreadservice.RosterMember
		var userID sql.NullString
		if err := rows.Scan(
			&member.MemberID,
			&userID,
			&member.LastMessageReadSeq,
			&member.LastActivityReadSeq,
			&member.EffectiveFromSeq,
			&member.NotificationState,
			&member.Visibility,
		); err != nil {
			return nil, fmt.Errorf("scan unread runtime roster: %w", err)
		}
		member.UserID = userID.String
		members = append(members, member)
	}
	if err := rows.Err(); err != nil {
		return nil, fmt.Errorf("iterate unread runtime roster: %w", err)
	}
	return members, nil
}

// LoadMessageTailPage loads a bounded human-visible message tail for
// recovery.
func (s *Store) LoadMessageTailPage(ctx context.Context, request TailPageRequest) ([]unreadservice.MessageStored, error) {
	if s == nil || s.db == nil {
		return nil, fmt.Errorf("unread runtime store is not configured")
	}
	request.normalize()
	if err := request.validate(); err != nil {
		return nil, err
	}

	rows, err := s.db.Query(ctx, fmt.Sprintf(`
SELECT
  m.workspace_id,
  m.scope_id,
  s.type,
  m.seq,
  m.author_member_id,
  COALESCE(NULLIF(m.notify_principal_member_id, ''), m.author_member_id),
  CASE
    WHEN EXISTS (
      SELECT 1
      FROM workflow_run_effects wre
      WHERE wre.workspace_id = m.workspace_id
        AND wre.message_id = m.id
    ) OR EXISTS (
      SELECT 1
      FROM automation_runs ar
      WHERE ar.workspace_id = m.workspace_id
        AND ar.message_id = m.id
    ) OR COALESCE(JSON_UNQUOTE(JSON_EXTRACT(m.metadata, '$.workflow_run_id')), '') <> ''
    THEN TRUE
    ELSE FALSE
  END
FROM scope_messages m
JOIN scopes s
  ON s.id = m.scope_id
 AND s.workspace_id = m.workspace_id
WHERE m.workspace_id = $1
  AND m.scope_id = $2
  AND s.type IN ('channel', 'dm', 'thread', 'task', 'service_account')
  AND m.seq > $3
  AND %s
ORDER BY m.seq
LIMIT $4
`, message.HumanVisibleMessageTypeSQL("m")), request.WorkspaceID, request.ScopeID, request.AfterSeq, request.Limit)
	if err != nil {
		return nil, fmt.Errorf("load unread runtime message tail: %w", err)
	}
	defer rows.Close()

	items := make([]unreadservice.MessageStored, 0, request.Limit)
	for rows.Next() {
		var item unreadservice.MessageStored
		if err := rows.Scan(&item.WorkspaceID, &item.ScopeID, &item.ScopeType, &item.Seq, &item.AuthorMemberID, &item.NotifyPrincipalMemberID, &item.AllowAuthorUnread); err != nil {
			return nil, fmt.Errorf("scan unread runtime message tail: %w", err)
		}
		item.MessageGeneration = item.Seq
		items = append(items, item)
	}
	if err := rows.Err(); err != nil {
		return nil, fmt.Errorf("iterate unread runtime message tail: %w", err)
	}
	return items, nil
}

// LoadActivityTailPage loads a bounded durable high-signal activity tail for
// diagnostics or explicit repair.
func (s *Store) LoadActivityTailPage(ctx context.Context, request TailPageRequest) ([]unreadservice.NonMessageActivityStored, error) {
	if s == nil || s.db == nil {
		return nil, fmt.Errorf("unread runtime store is not configured")
	}
	request.normalize()
	if err := request.validate(); err != nil {
		return nil, err
	}

	rows, err := s.db.Query(ctx, buildActivityTailPageQuery(), request.WorkspaceID, request.ScopeID, request.AfterSeq, request.Limit)
	if err != nil {
		return nil, fmt.Errorf("load unread runtime activity tail: %w", err)
	}
	defer rows.Close()

	items := make([]unreadservice.NonMessageActivityStored, 0, request.Limit)
	for rows.Next() {
		var item unreadservice.NonMessageActivityStored
		if err := rows.Scan(&item.WorkspaceID, &item.ScopeID, &item.ScopeType, &item.Seq, &item.ActorMemberID); err != nil {
			return nil, fmt.Errorf("scan unread runtime activity tail: %w", err)
		}
		item.ActivityGeneration = item.Seq
		items = append(items, item)
	}
	if err := rows.Err(); err != nil {
		return nil, fmt.Errorf("iterate unread runtime activity tail: %w", err)
	}
	return items, nil
}

func (request *SnapshotPageRequest) normalize() {
	request.WorkspaceID = strings.TrimSpace(request.WorkspaceID)
	request.MemberID = strings.TrimSpace(request.MemberID)
	request.AfterScopeID = strings.TrimSpace(request.AfterScopeID)
	request.ScopeIDs = compactStrings(request.ScopeIDs)
}

func (request SnapshotPageRequest) validate() error {
	if request.WorkspaceID == "" || request.MemberID == "" {
		return fmt.Errorf("workspace id and member id are required")
	}
	if request.Limit <= 0 {
		return fmt.Errorf("limit must be positive")
	}
	return nil
}

func (request *MetadataPageRequest) normalize() {
	request.WorkspaceID = strings.TrimSpace(request.WorkspaceID)
	request.MemberID = strings.TrimSpace(request.MemberID)
	request.AfterScopeID = strings.TrimSpace(request.AfterScopeID)
	request.ScopeIDs = compactStrings(request.ScopeIDs)
}

func (request MetadataPageRequest) validate() error {
	if request.WorkspaceID == "" || request.MemberID == "" {
		return fmt.Errorf("workspace id and member id are required")
	}
	if request.Limit <= 0 {
		return fmt.Errorf("limit must be positive")
	}
	return nil
}

func (request *RosterPageRequest) normalize() {
	request.WorkspaceID = strings.TrimSpace(request.WorkspaceID)
	request.ScopeID = strings.TrimSpace(request.ScopeID)
	request.AfterMemberID = strings.TrimSpace(request.AfterMemberID)
}

func (request RosterPageRequest) validate() error {
	if request.WorkspaceID == "" || request.ScopeID == "" {
		return fmt.Errorf("workspace id and scope id are required")
	}
	if request.Limit <= 0 {
		return fmt.Errorf("limit must be positive")
	}
	return nil
}

func (request *TailPageRequest) normalize() {
	request.WorkspaceID = strings.TrimSpace(request.WorkspaceID)
	request.ScopeID = strings.TrimSpace(request.ScopeID)
}

func (request TailPageRequest) validate() error {
	if request.WorkspaceID == "" || request.ScopeID == "" {
		return fmt.Errorf("workspace id and scope id are required")
	}
	if request.Limit <= 0 {
		return fmt.Errorf("limit must be positive")
	}
	return nil
}