package message

import (
	"context"
	"errors"
	"os"
	"testing"
	"time"

	"github.com/stretchr/testify/require"

	"github.com/tidbcloud/AgentTeam/internal/core/domain"
	"github.com/tidbcloud/AgentTeam/internal/coredb"
	"github.com/tidbcloud/AgentTeam/internal/store/testfixture"
)

type scriptedSubExecResult struct {
	tag coredb.Result
	err error
}

type scriptedSubExecOnly struct {
	queries []string
	args    [][]any
	results []scriptedSubExecResult
}

func (execer *scriptedSubExecOnly) Exec(_ context.Context, query string, args ...any) (coredb.Result, error) {
	execer.queries = append(execer.queries, query)
	execer.args = append(execer.args, append([]any(nil), args...))
	if len(execer.results) == 0 {
		return testSubExecResult{}, nil
	}
	result := execer.results[0]
	execer.results = execer.results[1:]
	return result.tag, result.err
}

type testSubExecResult struct {
	rowsAffected int64
}

func (result testSubExecResult) RowsAffected() int64 {
	return result.rowsAffected
}

func TestListSubscribedThreadsForUserScopedQueryUsesScopeTopology(t *testing.T) {
	t.Parallel()

	query, args, err := BuildListSubscribedThreadsForUserScopedQuery("wsp-1", "mem-1", "usr-1", 25)
	require.NoError(t, err)
	require.Equal(t, []any{"wsp-1", "mem-1", "usr-1", 25}, args)
	require.Contains(t, query, "subscribed_thread_authorized_candidates AS (")
	require.Contains(t, query, "FROM scope_notification_settings ss")
	require.Contains(t, query, "subscribed_thread_page AS (")
	require.Contains(t, query, "FROM subscribed_thread_authorized_candidates")
	require.Contains(t, query, "LIMIT $4")
	require.Contains(t, query, "JOIN scopes reply_scope")
	require.Contains(t, query, "LEFT JOIN scope_channel_profiles parent_cp")
	require.Contains(t, query, "AND (parent_scope.type <> 'channel' OR parent_cp.subsystem IN ('im', 'team', 'tutorial'))")
	require.Contains(t, query, "COALESCE(reply_scope.latest_seq, 0) AS reply_count")
	require.Contains(t, query, "reply_scope.latest_message_at AS last_reply_at")
	require.Contains(t, query, "AS last_activity_at")
	require.Contains(t, query, "mac.created_by_user_id <> $3")
	require.NotContains(t, query, "visible_thread_refs AS (")
	require.NotContains(t, query, "JOIN visible_thread_refs")
	require.NotContains(t, query, legacySQLTableRefForMessageTest("FROM", "reply", "threads"))
	require.NotContains(t, query, legacySQLTableRefForMessageTest("JOIN", "reply", "thread", "stats"))
	require.NotContains(t, query, legacySQLTableRefForMessageTest("JOIN", "threads"))
	require.NotContains(t, query, legacySQLTableRefForMessageTest("LEFT JOIN", "dm", "threads"))
	require.NotContains(t, query, legacySQLTableRefForMessageTest("LEFT JOIN", "channels"))
	require.NotContains(t, query, legacyTableNameForMessageTest("space", "subscriptions"))
	require.NotContains(t, query, legacyTableNameForMessageTest("space", "participants"))
}

func TestBuildListSubscribedThreadInboxPageForUserScopedQueryUsesScopeTopology(t *testing.T) {
	t.Parallel()

	cursor := ThreadInboxCursor{
		SortAt:          time.Date(2026, 5, 14, 4, 0, 0, 0, time.UTC),
		ThreadChannelID: "thr-9",
	}

	query, args, err := BuildListSubscribedThreadInboxPageForUserScopedQuery("wsp-1", "mem-1", "usr-1", 20, cursor)
	require.NoError(t, err)
	require.Equal(t, []any{"wsp-1", "mem-1", "usr-1", cursor.SortAt, "thr-9", 21}, args)
	require.Contains(t, query, "subscribed_thread_authorized_candidates AS (")
	require.Contains(t, query, "subscribed_thread_page AS (")
	require.Contains(t, query, "WHERE (sort_at < $4 OR (sort_at = $4 AND thread_channel_id > $5))")
	require.Contains(t, query, "LIMIT $6")
	require.Contains(t, query, "JOIN scopes reply_scope")
	require.Contains(t, query, "AND (parent_scope.type <> 'channel' OR parent_cp.subsystem IN ('im', 'team', 'tutorial'))")
	require.Contains(t, query, "last_activity_at")
	require.Contains(t, query, "sort_at")
	require.NotContains(t, query, "visible_thread_refs AS (")
	require.NotContains(t, query, "JOIN visible_thread_refs")
	require.NotContains(t, query, legacySQLTableRefForMessageTest("FROM", "reply", "threads"))
	require.NotContains(t, query, legacySQLTableRefForMessageTest("JOIN", "reply", "thread", "stats"))
	require.NotContains(t, query, legacyTableNameForMessageTest("space", "subscriptions"))
}

func TestBuildCountSubscribedThreadsForUserScopedQueryUsesScopeTopology(t *testing.T) {
	t.Parallel()

	query, args, err := BuildCountSubscribedThreadsForUserScopedQuery("wsp-1", "mem-1", "usr-1")
	require.NoError(t, err)
	require.Equal(t, []any{"wsp-1", "mem-1", "usr-1"}, args)
	require.Contains(t, query, "subscribed_thread_authorized_candidates AS (")
	require.Contains(t, query, "SELECT COUNT(*)")
	require.Contains(t, query, "FROM subscribed_thread_authorized_candidates")
	require.Contains(t, query, "JOIN scopes reply_scope")
	require.Contains(t, query, "FROM scope_notification_settings ss")
	require.Contains(t, query, "AND (parent_scope.type <> 'channel' OR parent_cp.subsystem IN ('im', 'team', 'tutorial'))")
	require.NotContains(t, query, "visible_thread_refs AS (")
	require.NotContains(t, query, "JOIN visible_thread_refs")
	require.NotContains(t, query, "newer_ss")
	require.NotContains(t, query, "parent_message_preview")
	require.NotContains(t, query, legacySQLTableRefForMessageTest("FROM", "reply", "threads"))
	require.NotContains(t, query, legacySQLTableRefForMessageTest("JOIN", "reply", "thread", "stats"))
	require.NotContains(t, query, legacyTableNameForMessageTest("space", "subscriptions"))
}

func TestRemoveReplyThreadParticipantsBatchesUpdate(t *testing.T) {
	t.Parallel()

	now := time.Date(2026, 6, 7, 20, 30, 0, 0, time.UTC)
	execer := &scriptedSubExecOnly{}
	err := removeReplyThreadParticipants(context.Background(), execer, " thr-1 ", []string{"mem-b", "", " mem-a ", "mem-b"}, now)
	require.NoError(t, err)
	require.Len(t, execer.queries, 1)
	require.Contains(t, execer.queries[0], "WITH removed_member_ids AS")
	require.Contains(t, execer.queries[0], "SELECT $3 AS member_id")
	require.Contains(t, execer.queries[0], "SELECT $4 AS member_id")
	require.Contains(t, execer.queries[0], "UPDATE scope_memberships")
	require.Contains(t, execer.queries[0], "member_id IN (SELECT member_id FROM removed_member_ids)")
	require.Equal(t, []any{"thr-1", now, "mem-b", "mem-a"}, execer.args[0])
}

func TestThreadInboxCursorRoundTrip(t *testing.T) {
	t.Parallel()

	cursor := ThreadInboxCursor{
		SortAt:          time.Date(2026, 5, 14, 4, 30, 0, 0, time.UTC),
		ThreadChannelID: "thr-1",
	}

	decoded, err := normalizeThreadInboxCursor(cursor)
	require.NoError(t, err)
	require.Equal(t, cursor, decoded)
}

func TestNormalizeThreadInboxCursorRejectsInvalidCursor(t *testing.T) {
	t.Parallel()

	_, err := normalizeThreadInboxCursor(ThreadInboxCursor{ThreadChannelID: "thr-1"})
	require.ErrorContains(t, err, "invalid thread inbox cursor")
}

func TestListSubscribedThreadsUsesCurrentNotificationState(t *testing.T) {
	t.Parallel()

	ctx := context.Background()
	db := startMessagingTestPool(t, ctx)
	now := time.Date(2026, 5, 18, 9, 0, 0, 0, time.UTC)
	fixture := seedReplyThreadSubscriptionFixture(t, ctx, db, "current_notifications", now)
	insertMessagingHumanScopeMessage(t, ctx, db, "msg_sub_duplicate_reply", fixture.workspaceID, fixture.threadChannelID, fixture.replyThreadID, fixture.peerMemberID, "reply", 1, 2, now.Add(time.Minute))
	mustExecMessaging(t, ctx, db, `
INSERT INTO scope_notification_settings (
    scope_id, member_id, state, created_at, updated_at
)
VALUES ($1, $2, 'mute', $3, $3)
`, fixture.threadChannelID, fixture.ownerMemberID, now.Add(2*time.Minute))

	store := New(db, nil, 0, Options{})
	threads, err := store.ListSubscribedThreadsForUserWithLimit(ctx, fixture.ownerUserID, fixture.workspaceID, 0)
	require.NoError(t, err)
	require.Len(t, threads, 1)
	require.Equal(t, fixture.threadChannelID, threads[0].ThreadChannelID)
	require.Equal(t, "mute", threads[0].SubscriptionState)

	page, err := store.ListSubscribedThreadInboxPageForUser(ctx, fixture.ownerUserID, fixture.workspaceID, 10, ThreadInboxCursor{})
	require.NoError(t, err)
	require.Len(t, page.Threads, 1)
	require.Equal(t, fixture.threadChannelID, page.Threads[0].ThreadChannelID)
	require.Equal(t, "mute", page.Threads[0].SubscriptionState)

	count, err := store.CountSubscribedThreadsForUser(ctx, fixture.ownerUserID, fixture.workspaceID)
	require.NoError(t, err)
	require.Equal(t, int64(1), count.Count)
}

func TestUnsubscribeThreadForUserArchivesReplyThreadConversationIndexRow(t *testing.T) {
	t.Parallel()

	ctx := context.Background()
	db := startMessagingTestPool(t, ctx)
	now := time.Date(2026, 6, 23, 10, 0, 0, 0, time.UTC)
	fixture := seedReplyThreadSubscriptionFixture(t, ctx, db, "unsubscribe_archives_index", now)
	require.NoError(t, ActivateThreadSubscription(ctx, db, fixture.ownerMemberID, fixture.threadChannelID, now.Add(time.Minute)))

	mustExecMessaging(t, ctx, db, `
INSERT INTO im_conversation_index (
  workspace_id, member_id, conversation_id, scope_id, kind, sort_at,
  last_event_seq, pinned_at, archived_at, projection_json, created_at, updated_at
)
VALUES ($1, $2, $3, $4, 'thread', $5, 0, NULL, NULL, JSON_OBJECT('kind', 'replyThread', 'route', $6, 'title', 'thread row'), $5, $5)
`, fixture.workspaceID, fixture.ownerMemberID, "thread:"+fixture.threadChannelID, fixture.threadChannelID, now.Add(2*time.Minute), "/inbox/channel/"+fixture.channelID+"/thread/"+fixture.threadChannelID)

	store := New(db, nil, 0, Options{})
	require.NoError(t, store.UnsubscribeThreadForUser(ctx, fixture.ownerUserID, fixture.workspaceID, fixture.threadChannelID))

	var archivedAt time.Time
	require.NoError(t, db.QueryRow(ctx, `
SELECT archived_at
FROM im_conversation_index
WHERE workspace_id = $1
  AND member_id = $2
  AND scope_id = $3
  AND kind = 'thread'
`, fixture.workspaceID, fixture.ownerMemberID, fixture.threadChannelID).Scan(&archivedAt))
	require.False(t, archivedAt.IsZero())
}

func TestSubscribedThreadTitleUsesParentMessagePreviewForListPageAndGet(t *testing.T) {
	t.Parallel()

	ctx := context.Background()
	db := startMessagingTestPool(t, ctx)
	now := time.Date(2026, 6, 12, 9, 0, 0, 0, time.UTC)
	fixture := seedReplyThreadSubscriptionFixture(t, ctx, db, "title_preview", now)
	require.NoError(t, ActivateThreadSubscription(ctx, db, fixture.ownerMemberID, fixture.threadChannelID, now.Add(time.Minute)))

	store := New(db, nil, 0, Options{})
	threads, err := store.ListSubscribedThreadsForUserWithLimit(ctx, fixture.ownerUserID, fixture.workspaceID, 0)
	require.NoError(t, err)
	require.Len(t, threads, 1)
	require.Equal(t, "parent message", threads[0].Title)

	page, err := store.ListSubscribedThreadInboxPageForUser(ctx, fixture.ownerUserID, fixture.workspaceID, 10, ThreadInboxCursor{})
	require.NoError(t, err)
	require.Len(t, page.Threads, 1)
	require.Equal(t, "parent message", page.Threads[0].Title)

	thread, err := store.GetReplyThreadForUser(ctx, fixture.ownerUserID, fixture.workspaceID, fixture.threadChannelID)
	require.NoError(t, err)
	require.Equal(t, "parent message", thread.Title)
}

func TestSubscribedThreadTitleFallsBackToParentLabelNotAuthor(t *testing.T) {
	t.Parallel()

	ctx := context.Background()
	db := startMessagingTestPool(t, ctx)
	now := time.Date(2026, 6, 12, 9, 30, 0, 0, time.UTC)
	fixture := seedReplyThreadSubscriptionFixture(t, ctx, db, "title_fallback", now)
	require.NoError(t, ActivateThreadSubscription(ctx, db, fixture.ownerMemberID, fixture.threadChannelID, now.Add(time.Minute)))
	mustExecMessaging(t, ctx, db, `
UPDATE scope_messages
SET content = '   ', updated_at = $3
WHERE workspace_id = $1
  AND id = $2
`, fixture.workspaceID, fixture.parentMessageID, now.Add(2*time.Minute))

	store := New(db, nil, 0, Options{})
	thread, err := store.GetReplyThreadForUser(ctx, fixture.ownerUserID, fixture.workspaceID, fixture.threadChannelID)
	require.NoError(t, err)
	require.Equal(t, "sub-title_fallback", thread.Title)
	require.NotEqual(t, thread.ParentMessageAuthor, thread.Title)
	require.NotEmpty(t, thread.Title)
}

func TestListSubscribedThreadsExcludesKBParentChannelThreads(t *testing.T) {
	t.Parallel()

	ctx := context.Background()
	db := startMessagingTestPool(t, ctx)
	now := time.Date(2026, 6, 9, 10, 0, 0, 0, time.UTC)
	fixture := seedReplyThreadSubscriptionFixture(t, ctx, db, "kb_parent_excluded", now)
	require.NoError(t, ActivateThreadSubscription(ctx, db, fixture.ownerMemberID, fixture.threadChannelID, now.Add(time.Minute)))

	const (
		kbChannelID       = "chn_sub_kb_carrier_excluded"
		kbParentMessageID = "msg_sub_kb_carrier_excluded"
		kbThreadChannelID = "chn_thread_sub_kb_carrier_excluded"
	)
	insertMessagingChannel(t, ctx, db, kbChannelID, fixture.workspaceID, "kb-parent-excluded", fixture.ownerMemberID, "", now.Add(2*time.Minute))
	mustExecMessaging(t, ctx, db, `
UPDATE scope_channel_profiles
SET subsystem = $2, updated_at = $3
WHERE scope_id = $1
`, kbChannelID, ChannelSubsystemKB, now.Add(2*time.Minute))
	insertMessagingHumanScopeMessage(t, ctx, db, kbParentMessageID, fixture.workspaceID, kbChannelID, "", fixture.ownerMemberID, "kb parent", 1, 3, now.Add(3*time.Minute))
	insertMessagingReplyThread(t, ctx, db, kbThreadChannelID, fixture.workspaceID, kbChannelID, kbParentMessageID, "kbsubex", "", now.Add(4*time.Minute))
	require.NoError(t, ActivateThreadSubscription(ctx, db, fixture.ownerMemberID, kbThreadChannelID, now.Add(5*time.Minute)))

	store := New(db, nil, 0, Options{})
	threads, err := store.ListSubscribedThreadsForUserWithLimit(ctx, fixture.ownerUserID, fixture.workspaceID, 0)
	require.NoError(t, err)
	require.Equal(t, []string{fixture.threadChannelID}, subscribedThreadIDsForTest(threads))

	page, err := store.ListSubscribedThreadInboxPageForUser(ctx, fixture.ownerUserID, fixture.workspaceID, 10, ThreadInboxCursor{})
	require.NoError(t, err)
	require.Equal(t, []string{fixture.threadChannelID}, subscribedThreadPageIDsForTest(page.Threads))

	count, err := store.CountSubscribedThreadsForUser(ctx, fixture.ownerUserID, fixture.workspaceID)
	require.NoError(t, err)
	require.Equal(t, int64(1), count.Count)

	thread, err := store.GetReplyThreadForUser(ctx, fixture.ownerUserID, fixture.workspaceID, kbThreadChannelID)
	require.NoError(t, err)
	require.Equal(t, ChannelSubsystemKB, thread.ParentChannelType)
}

func TestUpsertThreadSubscriptionWritesScopeNotification(t *testing.T) {
	t.Parallel()

	now := time.Date(2026, 4, 4, 17, 0, 0, 0, time.UTC)
	execer := &scriptedSubExecOnly{}

	err := UpsertThreadSubscription(context.Background(), execer, "mem-1", "thr-1", now, nil)
	require.NoError(t, err)
	require.Len(t, execer.queries, 1)
	require.Contains(t, execer.queries[0], "INSERT INTO scope_notification_settings")
	require.Contains(t, execer.queries[0], "ON DUPLICATE KEY UPDATE")
	require.Contains(t, execer.queries[0], "updated_at = CASE")
	require.Contains(t, execer.queries[0], "WHEN state <> VALUES(state)")
	require.Contains(t, execer.queries[0], "ELSE updated_at")
	require.NotContains(t, execer.queries[0], legacyTableNameForMessageTest("thread", "subscriptions"))
	require.NotContains(t, execer.queries[0], legacyTableNameForMessageTest("space", "subscriptions"))
	require.Equal(t, []any{"thr-1", "mem-1", "active", now}, execer.args[0])
}

func TestUpsertTaskWatchNotificationStateWritesV2AuthorityAndLegacyMirror(t *testing.T) {
	t.Parallel()

	now := time.Date(2026, 7, 29, 2, 20, 0, 0, time.UTC)
	execer := &scriptedSubExecOnly{}

	err := UpsertTaskWatchNotificationState(context.Background(), execer, " mem-1 ", " task-1 ", " auto-wake ", now)

	require.NoError(t, err)
	require.Len(t, execer.queries, 1)
	require.Contains(t, execer.queries[0], "state, task_watch_state")
	require.Contains(t, execer.queries[0], "state = VALUES(state)")
	require.Contains(t, execer.queries[0], "task_watch_state = VALUES(task_watch_state)")
	require.Equal(t, []any{"task-1", "mem-1", "auto-wake", now}, execer.args[0])
}

func subscribedThreadIDsForTest(threads []SubscribedThread) []string {
	ids := make([]string, 0, len(threads))
	for _, thread := range threads {
		ids = append(ids, thread.ThreadChannelID)
	}
	return ids
}

func subscribedThreadPageIDsForTest(threads []SubscribedThread) []string {
	return subscribedThreadIDsForTest(threads)
}

func TestUpsertThreadSubscriptionUnsubscribeRemovesScopeState(t *testing.T) {
	t.Parallel()

	now := time.Date(2026, 4, 4, 17, 0, 0, 0, time.UTC)
	execer := &scriptedSubExecOnly{}

	err := UpsertThreadSubscription(context.Background(), execer, "mem-1", "thr-1", now, &now)
	require.NoError(t, err)
	require.Len(t, execer.queries, 2)
	require.Contains(t, execer.queries[0], "DELETE FROM scope_notification_settings")
	require.Contains(t, execer.queries[1], "UPDATE scope_memberships")
	for _, query := range execer.queries {
		require.NotContains(t, query, legacyTableNameForMessageTest("thread", "subscriptions"))
		require.NotContains(t, query, legacyTableNameForMessageTest("space", "subscriptions"))
	}
}

func TestInsertThreadSubscriptionIfMissingWritesScopeNotification(t *testing.T) {
	t.Parallel()

	now := time.Date(2026, 4, 4, 17, 5, 0, 0, time.UTC)
	execer := &scriptedSubExecOnly{}

	err := InsertThreadSubscriptionIfMissing(context.Background(), execer, "mem-1", "thr-1", now)
	require.NoError(t, err)
	require.Len(t, execer.queries, 1)
	require.Contains(t, execer.queries[0], "INSERT IGNORE INTO scope_notification_settings")
	require.NotContains(t, execer.queries[0], legacyTableNameForMessageTest("thread", "subscriptions"))
	require.NotContains(t, execer.queries[0], legacyTableNameForMessageTest("space", "subscriptions"))
}

func TestActivateThreadSubscriptionAtSeqPreservesPreExistingOpenNotification(t *testing.T) {
	t.Parallel()

	ctx := context.Background()
	db := startMessagingTestPool(t, ctx)
	now := time.Date(2026, 5, 14, 6, 45, 0, 0, time.UTC)

	const (
		workspaceID = "wsp_sub_seq_existing"
		userID      = "usr_sub_seq_existing"
		memberID    = "mem_sub_seq_existing"
		scopeID     = "thr_sub_seq_existing"
	)
	seedThreadSubscriptionScope(t, ctx, db, workspaceID, userID, memberID, scopeID, now)
	mustExecMessaging(t, ctx, db, `
INSERT INTO scope_memberships (
    scope_id, member_id, visibility, owner_member_id, last_read_seq,
    last_read_global_seq, last_activity_global_seq, owner_context_dismissed_through_global_seq,
    effective_from_seq, effective_to_seq, created_at, updated_at, removed_at, archived_at
)
VALUES ($1, $2, 'public', NULL, 0, 0, 0, 0, 0, NULL, $3, $3, NULL, NULL)
`, scopeID, memberID, now)
	mustExecMessaging(t, ctx, db, `
INSERT INTO scope_notification_settings (
    scope_id, member_id, state, created_at, updated_at
)
VALUES ($1, $2, 'quiet', $3, $3)
`, scopeID, memberID, now)

	require.NoError(t, ActivateThreadSubscriptionAtSeq(ctx, db, memberID, scopeID, 5, now.Add(time.Minute)))

	var activeCount int64
	require.NoError(t, db.QueryRow(ctx, `
SELECT COUNT(*)
FROM scope_notification_settings
WHERE scope_id = $1
  AND member_id = $2
`, scopeID, memberID).Scan(&activeCount))
	require.Equal(t, int64(1), activeCount)

	var state string
	require.NoError(t, db.QueryRow(ctx, `
SELECT state
FROM scope_notification_settings
WHERE scope_id = $1
  AND member_id = $2
`, scopeID, memberID).Scan(&state))
	require.Equal(t, "active", state)

	var membershipEffectiveFromSeq int64
	require.NoError(t, db.QueryRow(ctx, `
SELECT effective_from_seq
FROM scope_memberships
WHERE scope_id = $1
  AND member_id = $2
  AND effective_to_seq IS NULL
  AND removed_at IS NULL
`, scopeID, memberID).Scan(&membershipEffectiveFromSeq))
	require.Equal(t, int64(0), membershipEffectiveFromSeq)
}

func TestActivateThreadSubscriptionAtSeqCollapsesDuplicateOpenNotifications(t *testing.T) {
	t.Parallel()

	ctx := context.Background()
	db := startMessagingTestPool(t, ctx)
	now := time.Date(2026, 5, 18, 9, 15, 0, 0, time.UTC)

	const (
		workspaceID = "wsp_sub_seq_duplicate_open"
		userID      = "usr_sub_seq_duplicate_open"
		memberID    = "mem_sub_seq_duplicate_open"
		scopeID     = "thr_sub_seq_duplicate_open"
	)
	seedThreadSubscriptionScope(t, ctx, db, workspaceID, userID, memberID, scopeID, now)
	mustExecMessaging(t, ctx, db, `
INSERT INTO scope_memberships (
    scope_id, member_id, visibility, owner_member_id, last_read_seq,
    last_read_global_seq, last_activity_global_seq, owner_context_dismissed_through_global_seq,
    effective_from_seq, effective_to_seq, created_at, updated_at, removed_at, archived_at
)
VALUES ($1, $2, 'public', NULL, 0, 0, 0, 0, 0, NULL, $3, $3, NULL, NULL)
`, scopeID, memberID, now)
	mustExecMessaging(t, ctx, db, `
INSERT INTO scope_notification_settings (
    scope_id, member_id, state, created_at, updated_at
)
VALUES ($1, $2, 'mute', $4, $4)
`, scopeID, memberID, now, now.Add(time.Minute))

	require.NoError(t, ActivateThreadSubscriptionAtSeq(ctx, db, memberID, scopeID, 7, now.Add(2*time.Minute)))

	var openCount int64
	require.NoError(t, db.QueryRow(ctx, `
SELECT COUNT(*)
FROM scope_notification_settings
WHERE scope_id = $1
  AND member_id = $2
`, scopeID, memberID).Scan(&openCount))
	require.Equal(t, int64(1), openCount)

	var state string
	require.NoError(t, db.QueryRow(ctx, `
SELECT state
FROM scope_notification_settings
WHERE scope_id = $1
  AND member_id = $2
`, scopeID, memberID).Scan(&state))
	require.Equal(t, "active", state)
}

func TestActivateThreadSubscriptionAtSeqReopensClosedNotificationAtMessageSeq(t *testing.T) {
	t.Parallel()

	ctx := context.Background()
	db := startMessagingTestPool(t, ctx)
	now := time.Date(2026, 5, 14, 6, 46, 0, 0, time.UTC)

	const (
		workspaceID = "wsp_sub_seq_reopen"
		userID      = "usr_sub_seq_reopen"
		memberID    = "mem_sub_seq_reopen"
		scopeID     = "thr_sub_seq_reopen"
	)
	seedThreadSubscriptionScope(t, ctx, db, workspaceID, userID, memberID, scopeID, now)
	mustExecMessaging(t, ctx, db, `
INSERT INTO scope_notification_settings (
    scope_id, member_id, state, created_at, updated_at
)
VALUES ($1, $2, 'active', $3, $3)
`, scopeID, memberID, now)

	require.NoError(t, ActivateThreadSubscriptionAtSeq(ctx, db, memberID, scopeID, 5, now.Add(time.Minute)))

	var membershipEffectiveFromSeq int64
	require.NoError(t, db.QueryRow(ctx, `
SELECT effective_from_seq
FROM scope_memberships
WHERE scope_id = $1
  AND member_id = $2
  AND effective_to_seq IS NULL
  AND removed_at IS NULL
`, scopeID, memberID).Scan(&membershipEffectiveFromSeq))
	require.Equal(t, int64(5), membershipEffectiveFromSeq)

	var notificationState string
	require.NoError(t, db.QueryRow(ctx, `
SELECT state
FROM scope_notification_settings
WHERE scope_id = $1
  AND member_id = $2
`, scopeID, memberID).Scan(&notificationState))
	require.Equal(t, "active", notificationState)

	mustExecMessaging(t, ctx, db, `
UPDATE scope_memberships
SET effective_to_seq = effective_from_seq
WHERE scope_id = $1
  AND member_id = $2
  AND effective_from_seq >= $3
  AND effective_to_seq IS NULL
`, scopeID, memberID, int64(5))

	var openCount int64
	require.NoError(t, db.QueryRow(ctx, `
SELECT COUNT(*)
FROM scope_notification_settings
WHERE scope_id = $1
  AND member_id = $2
`, scopeID, memberID).Scan(&openCount))
	require.Equal(t, int64(1), openCount)

	var openMembershipCount int64
	require.NoError(t, db.QueryRow(ctx, `
SELECT COUNT(*)
FROM scope_memberships
WHERE scope_id = $1
  AND member_id = $2
  AND effective_to_seq IS NULL
  AND removed_at IS NULL
`, scopeID, memberID).Scan(&openMembershipCount))
	require.Equal(t, int64(0), openMembershipCount)
}

func TestActivateThreadSubscriptionForAgentWritesAutoWakeScopeState(t *testing.T) {
	t.Parallel()

	ctx := context.Background()
	db := startMessagingTestPool(t, ctx)
	now := time.Date(2026, 5, 14, 7, 10, 0, 0, time.UTC)

	const (
		workspaceID = "wsp_sub_agent_auto_wake"
		userID      = "usr_sub_agent_auto_wake"
		ownerID     = "mem_sub_agent_auto_wake_owner"
		agentID     = "mem_sub_agent_auto_wake_agent"
		scopeID     = "thr_sub_agent_auto_wake"
	)
	seedThreadSubscriptionScope(t, ctx, db, workspaceID, userID, ownerID, scopeID, now)
	insertMessagingAgentMember(t, ctx, db, agentID, workspaceID, now)

	require.NoError(t, ActivateThreadSubscription(ctx, db, agentID, scopeID, now))

	var notificationState string
	require.NoError(t, db.QueryRow(ctx, `
SELECT state
FROM scope_notification_settings
WHERE scope_id = $1
  AND member_id = $2
`, scopeID, agentID).Scan(&notificationState))
	require.Equal(t, "auto-wake", notificationState)
}

func TestActivateThreadSubscriptionForAgentPreservesNoWakeScopeState(t *testing.T) {
	t.Parallel()

	ctx := context.Background()
	db := startMessagingTestPool(t, ctx)
	now := time.Date(2026, 5, 14, 7, 15, 0, 0, time.UTC)

	const (
		workspaceID = "wsp_sub_agent_no_wake"
		userID      = "usr_sub_agent_no_wake"
		ownerID     = "mem_sub_agent_no_wake_owner"
		agentID     = "mem_sub_agent_no_wake_agent"
		scopeID     = "thr_sub_agent_no_wake"
	)
	seedThreadSubscriptionScope(t, ctx, db, workspaceID, userID, ownerID, scopeID, now)
	insertMessagingAgentMember(t, ctx, db, agentID, workspaceID, now)
	mustExecMessaging(t, ctx, db, `
INSERT INTO scope_memberships (
    scope_id, member_id, visibility, owner_member_id, last_read_seq,
    last_read_global_seq, last_activity_global_seq, owner_context_dismissed_through_global_seq,
    effective_from_seq, effective_to_seq, created_at, updated_at, removed_at, archived_at
)
VALUES ($1, $2, 'public', NULL, 0, 0, 0, 0, 0, NULL, $3, $3, NULL, NULL)
`, scopeID, agentID, now)
	mustExecMessaging(t, ctx, db, `
INSERT INTO scope_notification_settings (
    scope_id, member_id, state, created_at, updated_at
)
VALUES ($1, $2, 'no-wake', $3, $3)
`, scopeID, agentID, now)

	require.NoError(t, ActivateThreadSubscriptionAtSeq(ctx, db, agentID, scopeID, 5, now.Add(time.Minute)))

	var notificationState string
	require.NoError(t, db.QueryRow(ctx, `
SELECT state
FROM scope_notification_settings
WHERE scope_id = $1
  AND member_id = $2
`, scopeID, agentID).Scan(&notificationState))
	require.Equal(t, "no-wake", notificationState)
}

func TestActivateThreadSubscriptionIfNotExplicitlyUnsubscribedWritesScopeState(t *testing.T) {
	t.Parallel()

	ctx := context.Background()
	db := startMessagingTestPool(t, ctx)
	now := time.Date(2026, 5, 14, 7, 20, 0, 0, time.UTC)

	const (
		workspaceID = "wsp_sub_skip_missing"
		userID      = "usr_sub_skip_missing"
		memberID    = "mem_sub_skip_missing"
		scopeID     = "thr_sub_skip_missing"
	)
	seedThreadSubscriptionScope(t, ctx, db, workspaceID, userID, memberID, scopeID, now)

	require.NoError(t, ActivateThreadSubscriptionIfNotExplicitlyUnsubscribed(ctx, db, memberID, scopeID, now))

	var membershipCount int64
	require.NoError(t, db.QueryRow(ctx, `
SELECT COUNT(*)
FROM scope_memberships
WHERE scope_id = $1
  AND member_id = $2
  AND removed_at IS NULL
  AND effective_to_seq IS NULL
`, scopeID, memberID).Scan(&membershipCount))
	require.Equal(t, int64(1), membershipCount)

	var notificationState string
	require.NoError(t, db.QueryRow(ctx, `
SELECT state
FROM scope_notification_settings
WHERE scope_id = $1
  AND member_id = $2
`, scopeID, memberID).Scan(&notificationState))
	require.Equal(t, "active", notificationState)
}

func TestActivateThreadSubscriptionIfNotExplicitlyUnsubscribedSkipsRemovedScopeMembership(t *testing.T) {
	t.Parallel()

	ctx := context.Background()
	db := startMessagingTestPool(t, ctx)
	now := time.Date(2026, 5, 14, 7, 25, 0, 0, time.UTC)
	removedAt := now.Add(time.Minute)

	const (
		workspaceID = "wsp_sub_skip_removed"
		userID      = "usr_sub_skip_removed"
		memberID    = "mem_sub_skip_removed"
		scopeID     = "thr_sub_skip_removed"
	)
	seedThreadSubscriptionScope(t, ctx, db, workspaceID, userID, memberID, scopeID, now)
	mustExecMessaging(t, ctx, db, `
INSERT INTO scope_memberships (
    scope_id, member_id, visibility, owner_member_id, last_read_seq,
    last_read_global_seq, last_activity_global_seq, owner_context_dismissed_through_global_seq,
    effective_from_seq, effective_to_seq, created_at, updated_at, removed_at, archived_at
)
VALUES ($1, $2, 'public', NULL, 0, 0, 0, 0, 0, NULL, $3, $4, $4, NULL)
`, scopeID, memberID, now, removedAt)

	require.NoError(t, ActivateThreadSubscriptionIfNotExplicitlyUnsubscribed(ctx, db, memberID, scopeID, removedAt.Add(time.Minute)))

	var notificationCount int64
	require.NoError(t, db.QueryRow(ctx, `
SELECT COUNT(*)
FROM scope_notification_settings
WHERE scope_id = $1
  AND member_id = $2
`, scopeID, memberID).Scan(&notificationCount))
	require.Equal(t, int64(0), notificationCount)

	var openMembershipCount int64
	require.NoError(t, db.QueryRow(ctx, `
SELECT COUNT(*)
FROM scope_memberships
WHERE scope_id = $1
  AND member_id = $2
  AND removed_at IS NULL
`, scopeID, memberID).Scan(&openMembershipCount))
	require.Equal(t, int64(0), openMembershipCount)
}

func TestSyncReplyThreadSubscriptionsSkipsExplicitlyUnsubscribedParentAuthor(t *testing.T) {
	t.Parallel()

	ctx := context.Background()
	db := startMessagingTestPool(t, ctx)
	now := time.Date(2026, 5, 15, 6, 0, 0, 0, time.UTC)
	fixture := seedReplyThreadSubscriptionFixture(t, ctx, db, "parent_author", now)
	markThreadExplicitlyUnsubscribed(t, ctx, db, fixture.ownerMemberID, fixture.threadChannelID, now.Add(time.Minute))

	store := New(db, nil, 0, Options{})
	require.NoError(t, coredb.WithTx(ctx, db, func(tx coredb.Tx) error {
		return store.InTx(tx).SyncReplyThreadSubscriptions(
			ctx,
			fixture.peerMemberID,
			domain.MemberTypeHuman,
			fixture.threadChannelID,
			fixture.parentMessageID,
			nil,
			nil,
			2,
			2,
			now.Add(2*time.Minute),
		)
	}))

	assertNoOpenThreadSubscription(t, ctx, db, fixture.ownerMemberID, fixture.threadChannelID)
	assertOpenThreadSubscriptionState(t, ctx, db, fixture.peerMemberID, fixture.threadChannelID, "active")
}

func TestSyncReplyThreadSubscriptionsDirectMentionBypassesExplicitUnsubscribe(t *testing.T) {
	t.Parallel()

	ctx := context.Background()
	db := startMessagingTestPool(t, ctx)
	now := time.Date(2026, 5, 15, 6, 5, 0, 0, time.UTC)
	fixture := seedReplyThreadSubscriptionFixture(t, ctx, db, "direct_mention_skip", now)
	markThreadExplicitlyUnsubscribed(t, ctx, db, fixture.ownerMemberID, fixture.threadChannelID, now.Add(time.Minute))

	store := New(db, nil, 0, Options{})
	require.NoError(t, coredb.WithTx(ctx, db, func(tx coredb.Tx) error {
		return store.InTx(tx).SyncReplyThreadSubscriptions(
			ctx,
			fixture.peerMemberID,
			domain.MemberTypeHuman,
			fixture.threadChannelID,
			fixture.parentMessageID,
			[]string{fixture.ownerMemberID},
			nil,
			2,
			2,
			now.Add(2*time.Minute),
		)
	}))

	assertOpenThreadSubscriptionState(t, ctx, db, fixture.ownerMemberID, fixture.threadChannelID, "active")
	assertOpenThreadSubscriptionState(t, ctx, db, fixture.peerMemberID, fixture.threadChannelID, "active")
}

func TestEnsureReplyThreadParentAuthorParticipantSkipsExplicitUnsubscribe(t *testing.T) {
	t.Parallel()

	ctx := context.Background()
	db := startMessagingTestPool(t, ctx)
	now := time.Date(2026, 5, 15, 6, 10, 0, 0, time.UTC)
	fixture := seedReplyThreadSubscriptionFixture(t, ctx, db, "parent_projection", now)
	markThreadExplicitlyUnsubscribed(t, ctx, db, fixture.ownerMemberID, fixture.threadChannelID, now.Add(time.Minute))

	store := New(db, nil, 0, Options{})
	require.NoError(t, coredb.WithTx(ctx, db, func(tx coredb.Tx) error {
		return store.InTx(tx).EnsureReplyThreadParentAuthorParticipant(ctx, fixture.threadChannelID, fixture.parentMessageID, now.Add(2*time.Minute))
	}))

	assertNoOpenThreadSubscription(t, ctx, db, fixture.ownerMemberID, fixture.threadChannelID)
}

func TestReplaceReplyThreadParticipantsSkipsExplicitlyUnsubscribedMember(t *testing.T) {
	t.Parallel()

	ctx := context.Background()
	db := startMessagingTestPool(t, ctx)
	now := time.Date(2026, 5, 15, 6, 15, 0, 0, time.UTC)
	fixture := seedReplyThreadSubscriptionFixture(t, ctx, db, "participant_projection", now)
	markThreadExplicitlyUnsubscribed(t, ctx, db, fixture.ownerMemberID, fixture.threadChannelID, now.Add(time.Minute))

	require.NoError(t, coredb.WithTx(ctx, db, func(tx coredb.Tx) error {
		return ReplaceReplyThreadParticipants(ctx, tx, fixture.threadChannelID, []string{fixture.ownerUserID, fixture.peerUserID})
	}))

	assertNoOpenThreadSubscription(t, ctx, db, fixture.ownerMemberID, fixture.threadChannelID)
	assertOpenThreadSubscriptionState(t, ctx, db, fixture.peerMemberID, fixture.threadChannelID, "auto-wake")
}

func TestReplaceReplyThreadParticipantsSkipsParentInactiveDerivedMember(t *testing.T) {
	t.Parallel()

	ctx := context.Background()
	db := startMessagingTestPool(t, ctx)
	now := time.Date(2026, 8, 9, 2, 15, 0, 0, time.UTC)
	fixture := seedReplyThreadSubscriptionFixture(t, ctx, db, "participant_parent_inactive", now)
	require.NoError(t, EnsureReplyThreadParticipantProjection(
		ctx, db, fixture.threadChannelID, fixture.ownerMemberID, "auto-wake", now.Add(time.Minute),
	))
	unsubscribedAt := now.Add(2 * time.Minute)
	require.NoError(t, UpsertThreadSubscription(
		ctx, db, fixture.ownerMemberID, fixture.threadChannelID, unsubscribedAt, &unsubscribedAt,
	))
	require.NoError(t, ActivateThreadSubscriptionAtSeq(
		ctx, db, fixture.ownerMemberID, fixture.threadChannelID, 2, now.Add(3*time.Minute),
	))
	mustExecMessaging(t, ctx, db, `
DELETE FROM scope_memberships
WHERE scope_id = $1 AND member_id = $2
`, fixture.channelID, fixture.ownerMemberID)

	require.NoError(t, coredb.WithTx(ctx, db, func(tx coredb.Tx) error {
		return ReplaceReplyThreadParticipants(ctx, tx, fixture.threadChannelID, []string{fixture.ownerUserID, fixture.peerUserID})
	}))

	var inactiveParentChildMemberships int64
	require.NoError(t, db.QueryRow(ctx, `
SELECT COUNT(*)
FROM scope_memberships
WHERE scope_id = $1
  AND member_id = $2
  AND removed_at IS NULL
  AND archived_at IS NULL
  AND effective_to_seq IS NULL
`, fixture.threadChannelID, fixture.ownerMemberID).Scan(&inactiveParentChildMemberships))
	require.Zero(t, inactiveParentChildMemberships, "derived participant with no parent access must not retain an active child membership")
	assertOpenThreadSubscriptionState(t, ctx, db, fixture.peerMemberID, fixture.threadChannelID, "auto-wake")
}

func TestReplyThreadSummaryParticipantsUseExplicitHumanSubscriptions(t *testing.T) {
	t.Parallel()

	ctx := context.Background()
	db := startMessagingTestPool(t, ctx)
	now := time.Date(2026, 5, 15, 6, 20, 0, 0, time.UTC)
	fixture := seedReplyThreadSubscriptionFixture(t, ctx, db, "summary_subscribers", now)

	require.NoError(t, coredb.WithTx(ctx, db, func(tx coredb.Tx) error {
		return ReplaceReplyThreadParticipants(ctx, tx, fixture.threadChannelID, []string{fixture.ownerUserID, fixture.peerUserID})
	}))

	participantMap, err := LoadReplyThreadParticipantDetailsMap(ctx, db, []string{fixture.threadChannelID})
	require.NoError(t, err)
	require.Empty(t, participantMap[fixture.threadChannelID])
	subscriberParticipantMap, err := loadReplyThreadSubscriberParticipantMap(ctx, db, []string{fixture.threadChannelID})
	require.NoError(t, err)
	require.Empty(t, subscriberParticipantMap[fixture.threadChannelID])

	require.NoError(t, ActivateThreadSubscription(ctx, db, fixture.peerMemberID, fixture.threadChannelID, now.Add(time.Minute)))

	participantMap, err = LoadReplyThreadParticipantDetailsMap(ctx, db, []string{fixture.threadChannelID})
	require.NoError(t, err)
	require.Len(t, participantMap[fixture.threadChannelID], 1)
	require.Equal(t, fixture.peerUserID, participantMap[fixture.threadChannelID][0].ID)
	require.Equal(t, "human", participantMap[fixture.threadChannelID][0].Type)
	subscriberParticipantMap, err = loadReplyThreadSubscriberParticipantMap(ctx, db, []string{fixture.threadChannelID})
	require.NoError(t, err)
	require.Equal(t, []string{fixture.peerUserID}, subscriberParticipantMap[fixture.threadChannelID])
}

func TestReplyThreadParticipantDetailsIncludeAuthoritativeAgentOwnershipAndVisibility(t *testing.T) {
	t.Parallel()

	ctx := context.Background()
	db := startMessagingTestPool(t, ctx)
	now := time.Date(2026, 7, 15, 2, 20, 0, 0, time.UTC)
	fixture := seedReplyThreadSubscriptionFixture(t, ctx, db, "participant_metadata", now)

	const (
		machineID     = "mach_sub_participant_metadata"
		agentMemberID = "mem_sub_agent_participant_metadata"
		agentID       = "agt_sub_participant_metadata"
	)
	insertMessagingMachine(t, ctx, db, machineID, fixture.workspaceID, fixture.ownerUserID, now)
	insertMessagingAgentMember(t, ctx, db, agentMemberID, fixture.workspaceID, now)
	insertMessagingAgent(t, ctx, db, agentID, fixture.workspaceID, machineID, agentMemberID, "metadata-agent", now)
	// Seed the parent channel private Agent membership first. The channel
	// reply-thread fence requires the agent to be active in the parent channel
	// before its child reply-thread membership can be written.
	require.NoError(t, EnsureScopeMembership(
		ctx,
		db,
		fixture.channelID,
		agentMemberID,
		"private",
		fixture.ownerMemberID,
		"auto-wake",
		0,
		0,
		now,
	))
	require.NoError(t, EnsureScopeMembership(
		ctx,
		db,
		fixture.threadChannelID,
		agentMemberID,
		"private",
		fixture.ownerMemberID,
		"auto-wake",
		0,
		0,
		now.Add(time.Minute),
	))

	participantMap, err := LoadReplyThreadParticipantDetailsMap(ctx, db, []string{fixture.threadChannelID})
	require.NoError(t, err)
	require.Len(t, participantMap[fixture.threadChannelID], 1)
	agent := participantMap[fixture.threadChannelID][0]
	require.Equal(t, agentID, agent.ID)
	require.Equal(t, agentMemberID, agent.MemberID)
	require.Equal(t, fixture.ownerMemberID, agent.OwnerMemberID)
	require.Equal(t, fixture.ownerUserID, agent.OwnerUserID)
	require.Equal(t, "private", agent.Visibility)
}

// TestReplyThreadParticipantProjectionAtomicWhenNotificationFails proves the
// child membership and its notification are written in the SAME transaction:
// with the parent membership active (so the fence allows the write), an illegal
// notification state trips scope_notification_settings_state_check during the
// notification stage, and the already-written child membership MUST roll back
// too. A split-tx implementation would commit the membership before the
// notification and leave it behind (membership count > 0).
func TestReplyThreadParticipantProjectionAtomicWhenNotificationFails(t *testing.T) {
	ctx := context.Background()
	db := startMessagingTestPool(t, ctx)
	now := time.Date(2026, 8, 8, 3, 0, 0, 0, time.UTC)
	fixture := seedReplyThreadSubscriptionFixture(t, ctx, db, "participant_atomic", now)

	const (
		machineID     = "mach_sub_participant_atomic"
		agentMemberID = "mem_sub_agent_participant_atomic"
		agentID       = "agt_sub_participant_atomic"
	)
	insertMessagingMachine(t, ctx, db, machineID, fixture.workspaceID, fixture.ownerUserID, now)
	insertMessagingAgentMember(t, ctx, db, agentMemberID, fixture.workspaceID, now)
	insertMessagingAgent(t, ctx, db, agentID, fixture.workspaceID, machineID, agentMemberID, "atomic-agent", now)
	// Parent channel membership is ACTIVE so the fence allows the write.
	require.NoError(t, EnsureScopeMembership(ctx, db, fixture.channelID, agentMemberID, "public", "", "auto-wake", 0, 0, now))

	// Force the notify step (which runs AFTER the membership write, inside the same
	// fence tx) to fail. The membership + notification must roll back together — a
	// split-tx implementation would have committed the membership already.
	forceReplyThreadParticipantNotifyErrorForTest = errors.New("forced notify failure")
	t.Cleanup(func() { forceReplyThreadParticipantNotifyErrorForTest = nil })

	projErr := EnsureReplyThreadParticipantProjection(ctx, db, fixture.threadChannelID, agentMemberID, "auto-wake", now.Add(time.Minute))
	require.ErrorContains(t, projErr, "forced notify failure")

	var membershipCount, notificationCount int
	require.NoError(t, db.QueryRow(ctx, `
SELECT COUNT(*) FROM scope_memberships WHERE scope_id = $1 AND member_id = $2
`, fixture.threadChannelID, agentMemberID).Scan(&membershipCount))
	require.Zero(t, membershipCount, "atomic: child membership must roll back when the notify step fails")
	require.NoError(t, db.QueryRow(ctx, `
SELECT COUNT(*) FROM scope_notification_settings WHERE scope_id = $1 AND member_id = $2
`, fixture.threadChannelID, agentMemberID).Scan(&notificationCount))
	require.Zero(t, notificationCount, "atomic: child notification must roll back with the membership")
}

func TestGetReplyThreadIncludesVisibleMessageAuthorFallbackParticipants(t *testing.T) {
	t.Parallel()

	ctx := context.Background()
	db := startMessagingTestPool(t, ctx)
	now := time.Date(2026, 7, 15, 11, 0, 0, 0, time.UTC)
	fixture := seedReplyThreadSubscriptionFixture(t, ctx, db, "visible_author_fallback", now)
	require.NoError(t, ActivateThreadSubscription(ctx, db, fixture.ownerMemberID, fixture.threadChannelID, now.Add(time.Minute)))

	insertMessagingHumanScopeMessage(
		t,
		ctx,
		db,
		"msg_sub_visible_author_fallback",
		fixture.workspaceID,
		fixture.threadChannelID,
		fixture.replyThreadID,
		fixture.peerMemberID,
		"visible fallback reply",
		1,
		2,
		now.Add(2*time.Minute),
	)

	store := New(db, nil, 0, Options{})
	thread, err := store.GetReplyThreadForUser(
		ctx,
		fixture.ownerUserID,
		fixture.workspaceID,
		fixture.threadChannelID,
	)
	require.NoError(t, err)
	require.Equal(t, 2, thread.ParticipantCount)

	participantsByMemberID := make(map[string]ReplyThreadParticipant, len(thread.Participants))
	for _, participant := range thread.Participants {
		participantsByMemberID[participant.MemberID] = participant
	}
	require.Equal(t, fixture.ownerUserID, participantsByMemberID[fixture.ownerMemberID].ID)
	require.Equal(t, fixture.peerUserID, participantsByMemberID[fixture.peerMemberID].ID)
	require.Equal(t, "human", participantsByMemberID[fixture.peerMemberID].Type)
}

func TestReplyThreadPreviewAuthorsAreIndependentFromSubscriptions(t *testing.T) {
	t.Parallel()

	ctx := context.Background()
	db := startMessagingTestPool(t, ctx)
	now := time.Date(2026, 8, 4, 16, 0, 0, 0, time.UTC)
	fixture := seedReplyThreadSubscriptionFixture(t, ctx, db, "preview_authors", now)

	require.NoError(t, ActivateThreadSubscription(ctx, db, fixture.ownerMemberID, fixture.threadChannelID, now.Add(time.Minute)))
	mustExecMessaging(t, ctx, db, `
UPDATE scopes
SET recent_author_member_ids = JSON_ARRAY($2), updated_at = $3
WHERE id = $1
`, fixture.threadChannelID, fixture.peerMemberID, now.Add(2*time.Minute))

	counting := &visibleScopeCountingQueryer{Queryer: db}
	participantMap, authorMap, err := loadReplyThreadParticipantAndAuthorDetailsMaps(
		ctx,
		counting,
		[]string{fixture.threadChannelID},
	)
	require.NoError(t, err)
	require.Equal(t, 1, counting.queryCount, "participants and preview authors must hydrate in one batch query")
	require.Equal(t, []string{fixture.ownerUserID}, replyThreadParticipantIDsFromDetails(participantMap[fixture.threadChannelID]))
	require.Equal(t, []string{fixture.peerUserID}, ReplyThreadRecentAuthorIDsFromDetails(authorMap[fixture.threadChannelID]))

	require.NoError(t, ActivateThreadSubscription(ctx, db, fixture.peerMemberID, fixture.threadChannelID, now.Add(150*time.Second)))
	unsubscribedAt := now.Add(3 * time.Minute)
	require.NoError(t, UpsertThreadSubscription(
		ctx,
		db,
		fixture.peerMemberID,
		fixture.threadChannelID,
		unsubscribedAt,
		&unsubscribedAt,
	))
	_, authorMap, err = loadReplyThreadParticipantAndAuthorDetailsMaps(ctx, db, []string{fixture.threadChannelID})
	require.NoError(t, err)
	require.Equal(t, []string{fixture.peerUserID}, ReplyThreadRecentAuthorIDsFromDetails(authorMap[fixture.threadChannelID]))
}

func TestThreadSubscriptionRuntimeDoesNotWriteLegacySubscriptionTables(t *testing.T) {
	t.Parallel()

	for _, filename := range []string{"thread_subscriptions.go", "thread_subscriptions_store.go"} {
		source, err := os.ReadFile(filename)
		require.NoError(t, err)
		require.NotContains(t, string(source), legacyTableNameForMessageTest("thread", "subscriptions"), filename)
		require.NotContains(t, string(source), legacyTableNameForMessageTest("space", "subscriptions"), filename)
	}
}

type replyThreadSubscriptionFixture struct {
	workspaceID     string
	ownerUserID     string
	ownerMemberID   string
	peerUserID      string
	peerMemberID    string
	channelID       string
	parentThreadID  string
	parentMessageID string
	threadChannelID string
	replyThreadID   string
}

func seedReplyThreadSubscriptionFixture(t *testing.T, ctx context.Context, db coredb.QueryExecer, suffix string, now time.Time) replyThreadSubscriptionFixture {
	t.Helper()

	fixture := replyThreadSubscriptionFixture{
		workspaceID:     "wsp_sub_" + suffix,
		ownerUserID:     "usr_sub_owner_" + suffix,
		ownerMemberID:   "mem_sub_owner_" + suffix,
		peerUserID:      "usr_sub_peer_" + suffix,
		peerMemberID:    "mem_sub_peer_" + suffix,
		channelID:       "chn_sub_" + suffix,
		parentThreadID:  "thr_parent_sub_" + suffix,
		parentMessageID: "msg_parent_sub_" + suffix,
		threadChannelID: "chn_thread_sub_" + suffix,
		replyThreadID:   "thr_reply_sub_" + suffix,
	}

	insertMessagingWorkspace(t, ctx, db, fixture.workspaceID, now)
	insertMessagingUser(t, ctx, db, fixture.ownerUserID, fixture.ownerUserID+"@example.com", now)
	insertMessagingUser(t, ctx, db, fixture.peerUserID, fixture.peerUserID+"@example.com", now)
	insertMessagingHumanMember(t, ctx, db, fixture.ownerMemberID, fixture.workspaceID, fixture.ownerUserID, now)
	insertMessagingHumanMember(t, ctx, db, fixture.peerMemberID, fixture.workspaceID, fixture.peerUserID, now)
	insertMessagingChannel(t, ctx, db, fixture.channelID, fixture.workspaceID, "sub-"+suffix, fixture.ownerMemberID, fixture.parentThreadID, now)
	mustExecMessaging(t, ctx, db, `
INSERT INTO scope_memberships (
    scope_id, member_id, visibility, owner_member_id, last_read_seq,
    last_read_global_seq, last_activity_global_seq, owner_context_dismissed_through_global_seq,
    effective_from_seq, effective_to_seq, created_at, updated_at, removed_at, archived_at
)
VALUES ($1, $2, 'public', NULL, 0, 0, 0, 0, 0, NULL, $3, $3, NULL, NULL)
`, fixture.channelID, fixture.peerMemberID, now)
	mustExecMessaging(t, ctx, db, `
INSERT INTO scope_notification_settings (
    scope_id, member_id, state, created_at, updated_at
)
VALUES ($1, $2, 'active', $3, $3)
`, fixture.channelID, fixture.peerMemberID, now)
	insertMessagingHumanScopeMessage(t, ctx, db, fixture.parentMessageID, fixture.workspaceID, fixture.channelID, fixture.parentThreadID, fixture.ownerMemberID, "parent message", 1, 1, now)
	insertMessagingReplyThread(t, ctx, db, fixture.threadChannelID, fixture.workspaceID, fixture.channelID, fixture.parentMessageID, "short-"+suffix, fixture.replyThreadID, now)

	return fixture
}

func insertMessagingHumanScopeMessage(t *testing.T, ctx context.Context, db coredb.QueryExecer, messageID string, workspaceID string, scopeID string, _ string, authorMemberID string, content string, seq int64, globalSeq int64, now time.Time) {
	t.Helper()

	mustExecMessaging(t, ctx, db, `
INSERT INTO scope_messages (
    id, workspace_id, scope_id, seq, workspace_global_seq,
    author_member_id, author_type, message_type, content, metadata, created_at, updated_at
)
VALUES ($1, $2, $3, $4, $5, $6, 'human', '', $7, NULL, $8, $8)
`, messageID, workspaceID, scopeID, seq, globalSeq, authorMemberID, content, now)
	mustExecMessaging(t, ctx, db, `
UPDATE scopes
SET latest_seq = $2,
    latest_global_seq = $3,
    latest_message_at = $4,
    visible_message_count = visible_message_count + 1,
    visible_last_message_at = $4,
    updated_at = $4
WHERE id = $1
`, scopeID, seq, globalSeq, now)
}

func markThreadExplicitlyUnsubscribed(t *testing.T, ctx context.Context, db coredb.QueryExecer, memberID string, threadChannelID string, unsubscribedAt time.Time) {
	t.Helper()

	require.NoError(t, ActivateThreadSubscription(ctx, db, memberID, threadChannelID, unsubscribedAt.Add(-time.Minute)))
	require.NoError(t, UpsertThreadSubscription(ctx, db, memberID, threadChannelID, unsubscribedAt, &unsubscribedAt))
}

func assertNoOpenThreadSubscription(t *testing.T, ctx context.Context, db coredb.Queryer, memberID string, threadChannelID string) {
	t.Helper()

	var openNotificationCount int64
	require.NoError(t, db.QueryRow(ctx, `
SELECT COUNT(*)
FROM scope_notification_settings
WHERE scope_id = $1
  AND member_id = $2
`, threadChannelID, memberID).Scan(&openNotificationCount))
	require.Equal(t, int64(0), openNotificationCount)

	var openMembershipCount int64
	require.NoError(t, db.QueryRow(ctx, `
SELECT COUNT(*)
FROM scope_memberships
WHERE scope_id = $1
  AND member_id = $2
  AND removed_at IS NULL
  AND effective_to_seq IS NULL
`, threadChannelID, memberID).Scan(&openMembershipCount))
	require.Equal(t, int64(0), openMembershipCount)
}

func assertOpenThreadSubscriptionState(t *testing.T, ctx context.Context, db coredb.Queryer, memberID string, threadChannelID string, expectedState string) {
	t.Helper()

	var openMembershipCount int64
	require.NoError(t, db.QueryRow(ctx, `
SELECT COUNT(*)
FROM scope_memberships
WHERE scope_id = $1
  AND member_id = $2
  AND removed_at IS NULL
  AND effective_to_seq IS NULL
`, threadChannelID, memberID).Scan(&openMembershipCount))
	require.Equal(t, int64(1), openMembershipCount)

	var notificationState string
	require.NoError(t, db.QueryRow(ctx, `
SELECT state
FROM scope_notification_settings
WHERE scope_id = $1
  AND member_id = $2
`, threadChannelID, memberID).Scan(&notificationState))
	require.Equal(t, expectedState, notificationState)
}

func seedThreadSubscriptionScope(t *testing.T, ctx context.Context, db coredb.QueryExecer, workspaceID string, userID string, memberID string, scopeID string, now time.Time) {
	t.Helper()

	mustExecMessaging(t, ctx, db, `
INSERT INTO users (id, email, name, display_name, avatar_url, is_admin, password_hash, email_verified_at, created_at, updated_at)
VALUES ($1, CONCAT($1, '@example.com'), $1, $1, '', FALSE, 'hash', $2, $2, $2)
`, userID, now)
	mustExecMessaging(t, ctx, db, `
INSERT INTO workspaces (id, organization_id, slug, name, latest_message_seq, created_at, updated_at)
VALUES ($1, $2, $1, $1, 0, $3, $3)
`, workspaceID, testfixture.SeedWorkspaceOrganization(t, ctx, db, workspaceID, now), now)
	mustExecMessaging(t, ctx, db, `
INSERT INTO members (id, workspace_id, type, display_name, user_id, workspace_role, created_at, updated_at)
VALUES ($1, $2, 'human', $1, $3, 'member', $4, $4)
`, memberID, workspaceID, userID, now)
	mustExecMessaging(t, ctx, db, `
INSERT INTO scopes (
    id, workspace_id, type, identity_key, latest_seq, latest_global_seq,
    latest_message_at, archived_until_seq, status, created_at, updated_at
)
VALUES ($1, $2, 'thread', CONCAT('thread:', $1), 0, 0, NULL, 0, 'active', $3, $3)
`, scopeID, workspaceID, now)
}