package im

import (
	"context"
	"database/sql"
	"fmt"
	"strings"
	"testing"
	"time"

	"github.com/stretchr/testify/require"

	"github.com/tidbcloud/AgentTeam/internal/coredb"
)

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

	ctx := context.Background()
	db := startIMTestPool(t, ctx)
	insertIMWorkspaceMember(t, ctx, db, "wsp_im_events", "mem_im_events")
	store := New(db)
	now := time.Now().UTC().Truncate(time.Millisecond)
	expiredAt := now.Add(-time.Minute)

	require.NoError(t, store.AppendEvents(ctx, "wsp_im_events", "mem_im_events", []StreamEvent{
		{EventID: "evt_1", Type: "message.created", ScopeID: "chn_1", Payload: map[string]any{"messageId": "msg_1"}, CreatedAt: now},
		{EventID: "evt_2", Type: "reaction.changed", ScopeID: "chn_1", Payload: map[string]any{"messageId": "msg_1"}, CreatedAt: now.Add(time.Millisecond)},
		{EventID: "evt_expired", Type: "message.created", ScopeID: "chn_1", Payload: map[string]any{"messageId": "msg_old"}, CreatedAt: now, ExpiresAt: &expiredAt},
	}))

	events, err := store.ListEventsAfter(ctx, "wsp_im_events", "mem_im_events", 0, 10)
	require.NoError(t, err)
	require.Len(t, events, 2)
	require.Equal(t, int64(1), events[0].EventSeq)
	require.Equal(t, int64(2), events[1].EventSeq)
	require.Equal(t, "message.created", events[0].Type)
	require.Equal(t, map[string]any{"messageId": "msg_1"}, events[0].Payload)

	afterFirst, err := store.ListEventsAfter(ctx, "wsp_im_events", "mem_im_events", 1, 10)
	require.NoError(t, err)
	require.Len(t, afterFirst, 1)
	require.Equal(t, "evt_2", afterFirst[0].EventID)

	latest, err := store.GetLatestSeq(ctx, "wsp_im_events", "mem_im_events")
	require.NoError(t, err)
	require.Equal(t, int64(2), latest)
}

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

	ctx := context.Background()
	db := startIMTestPool(t, ctx)
	insertIMWorkspaceMember(t, ctx, db, "wsp_im_cursor", "mem_im_cursor")
	store := New(db)

	require.NoError(t, store.UpdateCursorWithEvent(ctx, "wsp_im_cursor", "mem_im_cursor", "web", 5, "evt_5", "gen_a"))
	require.NoError(t, store.UpdateCursorWithEvent(ctx, "wsp_im_cursor", "mem_im_cursor", "web", 3, "evt_3", "gen_b"))

	cursor, err := store.GetCursor(ctx, "wsp_im_cursor", "mem_im_cursor", "web")
	require.NoError(t, err)
	require.Equal(t, int64(5), cursor.AckedEventSeq)
	require.Equal(t, "evt_5", cursor.AckedEventID)
	require.Equal(t, "gen_a", cursor.CursorGeneration)

	require.NoError(t, store.UpdateCursorWithEvent(ctx, "wsp_im_cursor", "mem_im_cursor", "web", 8, "evt_8", "gen_c"))
	cursor, err = store.GetCursor(ctx, "wsp_im_cursor", "mem_im_cursor", "web")
	require.NoError(t, err)
	require.Equal(t, int64(8), cursor.AckedEventSeq)
	require.Equal(t, "evt_8", cursor.AckedEventID)
	require.Equal(t, "gen_c", cursor.CursorGeneration)
}

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

	ctx := context.Background()
	db := startIMTestPool(t, ctx)
	insertIMWorkspaceMember(t, ctx, db, "wsp_im_conversations", "mem_im_conversations")
	store := New(db)
	base := time.Date(2026, 6, 9, 12, 0, 0, 0, time.UTC)

	for _, item := range []ConversationIndexItem{
		{ConversationID: "channel:old", ScopeID: "chn_old", Kind: "channel", SortAt: base.Add(-time.Hour), Projection: map[string]any{"title": "old"}},
		{ConversationID: "channel:new", ScopeID: "chn_new", Kind: "channel", SortAt: base, Projection: map[string]any{"title": "new"}},
		{ConversationID: "thread:mid", ScopeID: "thr_mid", Kind: "thread", SortAt: base.Add(-time.Minute), Projection: map[string]any{"title": "mid"}},
	} {
		require.NoError(t, store.UpsertConversation(ctx, "wsp_im_conversations", "mem_im_conversations", item))
	}

	first, err := store.ListConversations(ctx, "wsp_im_conversations", "mem_im_conversations", 2, ConversationCursor{}, nil)
	require.NoError(t, err)
	require.True(t, first.HasMore)
	require.Equal(t, []string{"channel:new", "thread:mid"}, []string{first.Items[0].ConversationID, first.Items[1].ConversationID})
	require.Equal(t, "thread:mid", first.NextCursor.ConversationID)
	require.False(t, first.NextCursor.SortAt.IsZero())

	second, err := store.ListConversations(ctx, "wsp_im_conversations", "mem_im_conversations", 2, first.NextCursor, nil)
	require.NoError(t, err)
	require.False(t, second.HasMore)
	require.Equal(t, []string{"channel:old"}, []string{second.Items[0].ConversationID})

	archivedAt := base.Add(time.Minute)
	require.NoError(t, store.UpsertConversation(ctx, "wsp_im_conversations", "mem_im_conversations", ConversationIndexItem{ConversationID: "channel:new", ScopeID: "chn_new", Kind: "channel", SortAt: base, ArchivedAt: &archivedAt, Projection: map[string]any{"title": "new"}}))
	visible, err := store.ListConversations(ctx, "wsp_im_conversations", "mem_im_conversations", 10, ConversationCursor{}, nil)
	require.NoError(t, err)
	require.Equal(t, []string{"thread:mid", "channel:old"}, []string{visible.Items[0].ConversationID, visible.Items[1].ConversationID})

	threads, err := store.ListConversations(ctx, "wsp_im_conversations", "mem_im_conversations", 10, ConversationCursor{}, []string{"replyThread"})
	require.NoError(t, err)
	require.Equal(t, []string{"thread:mid"}, []string{threads.Items[0].ConversationID})
}

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

	ctx := context.Background()
	db := startIMTestPool(t, ctx)
	insertIMWorkspaceMember(t, ctx, db, "wsp_im_deleted_service", "mem_im_deleted_service")
	store := New(db)
	now := time.Date(2026, 7, 14, 12, 0, 0, 0, time.UTC)

	require.NoError(t, store.UpsertConversation(ctx, "wsp_im_deleted_service", "mem_im_deleted_service", ConversationIndexItem{
		ConversationID: "serviceAccount:sasc_deleted",
		ScopeID:        "sasc_deleted",
		Kind:           "service_account",
		SortAt:         now,
		Projection:     map[string]any{"title": "Deleted service account"},
	}))
	require.NoError(t, store.UpsertConversation(ctx, "wsp_im_deleted_service", "mem_im_deleted_service", ConversationIndexItem{
		ConversationID: "channel:chn_active",
		ScopeID:        "chn_active",
		Kind:           "channel",
		SortAt:         now.Add(-time.Minute),
		Projection:     map[string]any{"title": "Active channel"},
	}))

	page, err := store.ListConversations(ctx, "wsp_im_deleted_service", "mem_im_deleted_service", 10, ConversationCursor{}, nil)
	require.NoError(t, err)
	require.Len(t, page.Items, 1)
	require.Equal(t, "channel:chn_active", page.Items[0].ConversationID)

	_, err = store.GetConversation(ctx, "wsp_im_deleted_service", "mem_im_deleted_service", "serviceAccount:sasc_deleted")
	require.ErrorContains(t, err, "conversation not found")
}

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

	ctx := context.Background()
	db := startIMTestPool(t, ctx)
	insertIMWorkspaceMember(t, ctx, db, "wsp_im_task_excluded", "mem_im_task_excluded")
	insertIMScope(t, ctx, db, "wsp_im_task_excluded", "tsk_legacy", "task", "mem_im_task_excluded", 0)
	insertIMScope(t, ctx, db, "wsp_im_task_excluded", "chn_active", "channel", "mem_im_task_excluded", 0)
	insertIMScopeMembership(t, ctx, db, "tsk_legacy", "mem_im_task_excluded", 0, nil)
	insertIMScopeMembership(t, ctx, db, "chn_active", "mem_im_task_excluded", 0, nil)
	store := New(db)
	now := time.Date(2026, 7, 15, 12, 0, 0, 0, time.UTC)

	require.NoError(t, store.UpsertConversation(ctx, "wsp_im_task_excluded", "mem_im_task_excluded", ConversationIndexItem{
		ConversationID: "task:tsk_legacy",
		ScopeID:        "tsk_legacy",
		Kind:           "task",
		SortAt:         now,
		Projection:     map[string]any{"title": "Legacy task row"},
	}))
	require.NoError(t, store.UpsertConversation(ctx, "wsp_im_task_excluded", "mem_im_task_excluded", ConversationIndexItem{
		ConversationID: "channel:chn_active",
		ScopeID:        "chn_active",
		Kind:           "channel",
		SortAt:         now.Add(-time.Minute),
		Projection:     map[string]any{"title": "Active channel"},
	}))

	page, err := store.ListConversations(ctx, "wsp_im_task_excluded", "mem_im_task_excluded", 10, ConversationCursor{}, nil)
	require.NoError(t, err)
	require.Len(t, page.Items, 1)
	require.Equal(t, "channel:chn_active", page.Items[0].ConversationID)

	_, err = store.GetConversation(ctx, "wsp_im_task_excluded", "mem_im_task_excluded", "task:tsk_legacy")
	require.ErrorContains(t, err, "conversation not found")

	tasks, err := store.ListConversations(ctx, "wsp_im_task_excluded", "mem_im_task_excluded", 10, ConversationCursor{}, []string{"task"})
	require.NoError(t, err)
	require.Empty(t, tasks.Items)

	_, err = db.Exec(ctx, `
UPDATE scopes
SET status = 'deleted',
    archived_at = $2,
    updated_at = $2
WHERE id = $1
`, "tsk_legacy", now.Add(time.Minute))
	require.NoError(t, err)

	page, err = store.ListConversations(ctx, "wsp_im_task_excluded", "mem_im_task_excluded", 10, ConversationCursor{}, nil)
	require.NoError(t, err)
	require.Len(t, page.Items, 1)
	require.Equal(t, "channel:chn_active", page.Items[0].ConversationID)

	_, err = store.GetConversation(ctx, "wsp_im_task_excluded", "mem_im_task_excluded", "task:tsk_legacy")
	require.ErrorContains(t, err, "conversation not found")
}

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

	ctx := context.Background()
	db := startIMTestPool(t, ctx)
	insertIMWorkspaceMember(t, ctx, db, "wsp_im_conversations_pinned", "mem_im_conversations_pinned")
	store := New(db)
	base := time.Date(2026, 6, 10, 12, 0, 0, 0, time.UTC)
	recentPinnedAt := base.Add(-30 * time.Minute)
	olderPinnedAt := base.Add(-time.Hour)

	for _, item := range []ConversationIndexItem{
		{ConversationID: "channel:latest", ScopeID: "chn_latest", Kind: "channel", SortAt: base, Projection: map[string]any{"title": "latest"}},
		{ConversationID: "channel:recently-pinned", ScopeID: "chn_recently_pinned", Kind: "channel", SortAt: base.Add(-time.Hour), PinnedAt: &recentPinnedAt, Projection: map[string]any{"title": "recently pinned"}},
		{ConversationID: "channel:active-pinned", ScopeID: "chn_active_pinned", Kind: "channel", SortAt: base.Add(-30 * time.Minute), PinnedAt: &olderPinnedAt, Projection: map[string]any{"title": "active pinned"}},
		{ConversationID: "channel:older", ScopeID: "chn_older", Kind: "channel", SortAt: base.Add(-2 * time.Hour), Projection: map[string]any{"title": "older"}},
	} {
		require.NoError(t, store.UpsertConversation(ctx, "wsp_im_conversations_pinned", "mem_im_conversations_pinned", item))
	}

	first, err := store.ListConversations(ctx, "wsp_im_conversations_pinned", "mem_im_conversations_pinned", 1, ConversationCursor{}, nil)
	require.NoError(t, err)
	require.True(t, first.HasMore)
	require.Equal(t, "channel:active-pinned", first.Items[0].ConversationID)
	require.NotNil(t, first.NextCursor.PinnedAt)

	second, err := store.ListConversations(ctx, "wsp_im_conversations_pinned", "mem_im_conversations_pinned", 10, first.NextCursor, nil)
	require.NoError(t, err)
	require.False(t, second.HasMore)
	require.Equal(t, []string{"channel:recently-pinned", "channel:latest", "channel:older"}, []string{second.Items[0].ConversationID, second.Items[1].ConversationID, second.Items[2].ConversationID})
}

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

	ctx := context.Background()
	db := startIMTestPool(t, ctx)
	insertIMWorkspaceMember(t, ctx, db, "wsp_im_scope_event", "mem_scope_owner")
	insertIMMember(t, ctx, db, "wsp_im_scope_event", "mem_scope_peer")
	insertIMMember(t, ctx, db, "wsp_im_scope_event", "mem_scope_removed")
	insertIMScope(t, ctx, db, "wsp_im_scope_event", "chn_im_scope_event", "channel", "mem_scope_owner", 7)
	insertIMScopeMembership(t, ctx, db, "chn_im_scope_event", "mem_scope_owner", 7, nil)
	insertIMScopeMembership(t, ctx, db, "chn_im_scope_event", "mem_scope_peer", 2, nil)
	removedAt := time.Now().UTC().Truncate(time.Second)
	insertIMScopeMembership(t, ctx, db, "chn_im_scope_event", "mem_scope_removed", 0, &removedAt)

	store := New(db)
	entitySeq := int64(7)
	createdAt := time.Date(2026, 6, 10, 9, 0, 0, 0, time.UTC)
	appended, err := store.AppendScopeEvent(ctx, ScopeEventRequest{
		WorkspaceID: "wsp_im_scope_event",
		ScopeID:     "chn_im_scope_event",
		Event: StreamEvent{
			EventID:   "evt_scope_message_7",
			Type:      "message.created",
			EntityID:  "msg_scope_7",
			EntitySeq: &entitySeq,
			Payload:   map[string]any{"messageId": "msg_scope_7"},
			CreatedAt: createdAt,
		},
		Conversation: &ConversationIndexItem{
			Projection: map[string]any{"title": "general"},
		},
	})
	require.NoError(t, err)
	require.Len(t, appended, 2)
	require.Equal(t, []string{"mem_scope_owner", "mem_scope_peer"}, []string{appended[0].MemberID, appended[1].MemberID})
	require.Equal(t, int64(1), appended[0].EventSeq)
	require.Equal(t, int64(1), appended[1].EventSeq)
	require.Equal(t, "channel:chn_im_scope_event", appended[0].ConversationID)

	ownerEvents, err := store.ListEventsAfter(ctx, "wsp_im_scope_event", "mem_scope_owner", 0, 10)
	require.NoError(t, err)
	require.Len(t, ownerEvents, 1)
	require.Equal(t, "message.created", ownerEvents[0].Type)
	require.Equal(t, map[string]any{"messageId": "msg_scope_7"}, ownerEvents[0].Payload)

	removedLatest, err := store.GetLatestSeq(ctx, "wsp_im_scope_event", "mem_scope_removed")
	require.NoError(t, err)
	require.Zero(t, removedLatest)

	ownerConversation, err := store.GetConversation(ctx, "wsp_im_scope_event", "mem_scope_owner", "channel:chn_im_scope_event")
	require.NoError(t, err)
	require.Equal(t, int64(1), ownerConversation.LastEventSeq)
	require.Equal(t, "general", ownerConversation.Projection["title"])
	require.NotContains(t, ownerConversation.Projection, "unreadCount")
	require.NotContains(t, ownerConversation.Projection, "hasUnread")
	require.NotContains(t, ownerConversation.Projection, "manualUnread")

	peerConversation, err := store.GetConversation(ctx, "wsp_im_scope_event", "mem_scope_peer", "channel:chn_im_scope_event")
	require.NoError(t, err)
	require.NotContains(t, peerConversation.Projection, "unreadCount")
	require.NotContains(t, peerConversation.Projection, "hasUnread")
	require.NotContains(t, peerConversation.Projection, "manualUnread")
}

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

	testCases := []struct {
		name             string
		suffix           string
		latestSeq        int64
		hiddenThroughSeq int64
		wantArchived     bool
	}{
		{name: "hidden through current message", suffix: "current", latestSeq: 2, hiddenThroughSeq: 2, wantArchived: true},
		{name: "new message restores conversation", suffix: "new_message", latestSeq: 3, hiddenThroughSeq: 2},
	}
	for _, testCase := range testCases {
		t.Run(testCase.name, func(t *testing.T) {
			ctx := context.Background()
			db := startIMTestPool(t, ctx)
			workspaceID := "wsp_im_hidden_" + testCase.suffix
			memberID := "mem_im_hidden_" + testCase.suffix
			scopeID := "chn_im_hidden_" + testCase.suffix
			insertIMWorkspaceMember(t, ctx, db, workspaceID, memberID)
			insertIMScope(t, ctx, db, workspaceID, scopeID, "channel", memberID, testCase.latestSeq)
			insertIMScopeMembership(t, ctx, db, scopeID, memberID, 0, nil)

			hiddenAt := time.Date(2026, 7, 14, 3, 0, 0, 0, time.UTC)
			_, err := db.Exec(ctx, `
INSERT INTO conversation_member_states (
  scope_id, member_id, hidden_at, hidden_through_seq, created_at, updated_at
) VALUES ($1, $2, $3, $4, $3, $3)
`, scopeID, memberID, hiddenAt, testCase.hiddenThroughSeq)
			require.NoError(t, err)

			store := New(db)
			require.NoError(t, store.UpsertConversation(ctx, workspaceID, memberID, ConversationIndexItem{
				ConversationID: "channel:" + scopeID,
				ScopeID:        scopeID,
				Kind:           "channel",
				SortAt:         hiddenAt,
				LastEventSeq:   1,
				ArchivedAt:     &hiddenAt,
				Projection:     map[string]any{"title": "hidden state channel"},
			}))

			entitySeq := testCase.latestSeq
			_, err = store.AppendScopeEvent(ctx, ScopeEventRequest{
				WorkspaceID: workspaceID,
				ScopeID:     scopeID,
				Event: StreamEvent{
					EventID:   "message.created:msg_im_hidden_" + testCase.suffix,
					Type:      "message.created",
					EntityID:  "msg_im_hidden_" + testCase.suffix,
					EntitySeq: &entitySeq,
					CreatedAt: hiddenAt.Add(time.Minute),
				},
				Conversation: &ConversationIndexItem{
					Projection: map[string]any{"title": "hidden state channel"},
				},
			})
			require.NoError(t, err)

			conversation, err := store.GetConversation(ctx, workspaceID, memberID, "channel:"+scopeID)
			require.NoError(t, err)
			require.Equal(t, testCase.wantArchived, conversation.ArchivedAt != nil)
			require.Equal(t, testCase.wantArchived, conversation.Projection["archived"])

			page, err := store.ListConversations(ctx, workspaceID, memberID, 10, ConversationCursor{}, []string{"channel"})
			require.NoError(t, err)
			if testCase.wantArchived {
				require.Empty(t, page.Items)
				return
			}
			require.Len(t, page.Items, 1)
			require.Equal(t, "channel:"+scopeID, page.Items[0].ConversationID)
		})
	}
}

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

	tests := []struct {
		kind    string
		scopeID string
		want    string
	}{
		{kind: "channel", scopeID: "chn_general", want: "channel:chn_general"},
		{kind: "thread", scopeID: "thr_reply", want: "thread:thr_reply"},
		{kind: "dm", scopeID: "dm_direct", want: "dm:dm_direct"},
		{kind: "task", scopeID: "tsk_work", want: "task:tsk_work"},
		{kind: "service_account", scopeID: "sasc_news", want: "serviceAccount:sasc_news"},
		{kind: "unknown", scopeID: "scope_other", want: "scope_other"},
	}
	for _, tt := range tests {
		t.Run(tt.kind, func(t *testing.T) {
			t.Parallel()
			require.Equal(t, tt.want, conversationIDForScopeKind(tt.kind, tt.scopeID))
		})
	}
}

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

	require.Equal(t,
		[]string{"channel", "dm", "thread", "task", "service_account"},
		normalizeConversationIndexKinds([]string{
			"channel", "dm", "replyThread", "task", "serviceAccount",
			"service_account", "unknown", " ",
		}),
	)
}

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

	ctx := context.Background()
	createdAt := time.Date(2026, 6, 11, 9, 0, 0, 0, time.UTC)
	recipients := []scopeEventRecipient{
		{MemberID: "mem_scope_owner", ScopeKind: "channel", SortAt: createdAt},
		{MemberID: "mem_scope_peer", ScopeKind: "channel", SortAt: createdAt},
		{MemberID: "mem_scope_third", ScopeKind: "channel", SortAt: createdAt},
	}
	tx := &recordingIMTx{
		latestSeqRows: []memberLatestSeqRow{
			{memberID: "mem_scope_owner", eventSeq: 5},
			{memberID: "mem_scope_third", eventSeq: 2},
		},
		refreshedRecipients: recipients,
	}

	appended, err := appendEventsForRecipients(ctx, tx, "wsp_im_scope_event", "chn_im_scope_event", StreamEvent{
		EventID:   "evt_scope_message_8",
		Type:      "message.created",
		EntityID:  "msg_scope_8",
		Payload:   map[string]any{"messageId": "msg_scope_8"},
		CreatedAt: createdAt,
	}, &ConversationIndexItem{Projection: map[string]any{"title": "general"}}, recipients)

	require.NoError(t, err)
	require.Len(t, appended, 3)
	require.Equal(t, []int64{6, 1, 3}, []int64{appended[0].EventSeq, appended[1].EventSeq, appended[2].EventSeq})
	require.Equal(t, 2, tx.queryCalls)
	require.Len(t, tx.queryArgs, 2)
	require.Len(t, tx.queryArgs[0], 4)
	require.Len(t, tx.queryArgs[1], 5)
	require.Equal(t, 1, strings.Count(tx.querySQL[0], "FOR UPDATE"))
	require.Equal(t, 1, strings.Count(tx.querySQL[1], "FOR UPDATE"))
	require.NotContains(t, tx.querySQL[1], "JOIN members", "projection refresh must not lock every member row in a large fanout")
	require.Contains(t, tx.querySQL[1], "conversation_member_states")
	require.Contains(t, tx.querySQL[1], "parent_cp.subsystem = 'kb'")
	require.Len(t, tx.execSQL, 2)
	require.Contains(t, tx.execSQL[0], "INSERT INTO im_stream_events")
	require.Contains(t, tx.execSQL[1], "INSERT INTO im_conversation_index")
	require.Len(t, tx.execArgs[0], len(recipients)*streamEventInsertArgCount)
	require.Len(t, tx.execArgs[1], len(recipients)*conversationIndexUpsertArgCount)
}

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

	for _, recipientCount := range []int{1, 100} {
		recipients := make([]scopeEventRecipient, 0, recipientCount)
		for index := 0; index < recipientCount; index++ {
			recipients = append(recipients, scopeEventRecipient{
				MemberID:        fmt.Sprintf("mem_%03d", index),
				ScopeKind:       "thread",
				ParentScopeType: "kb",
				SortAt:          time.Date(2026, 7, 19, 15, 0, 0, 0, time.UTC),
			})
		}
		tx := &recordingIMTx{refreshedRecipients: recipients}

		_, err := appendEventsForRecipients(context.Background(), tx, "wsp_1", "thr_1", StreamEvent{
			EventID:   "evt_query_count",
			Type:      "message.created",
			CreatedAt: time.Date(2026, 7, 19, 15, 0, 0, 0, time.UTC),
		}, &ConversationIndexItem{ConversationID: "thread:thr_1", Kind: "thread"}, recipients)

		require.NoError(t, err)
		require.Equal(t, 2, tx.queryCalls, "recipient count must not add scope-fact queries")
		require.Equal(t, 1, strings.Count(tx.querySQL[1], "parent_cp.subsystem = 'kb'"))
	}
}

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

	// The initial recipient snapshot represents an older Watch commit. The
	// refresh rows represent a newer Unwatch that committed before this append
	// acquired the member stream lock. Policy events without a materialized
	// Inbox conversation must still take the refresh path.
	initial := []scopeEventRecipient{{
		MemberID:       "mem_task_watch_no_index",
		ScopeKind:      "task",
		TaskWatchState: "active",
	}}
	tx := &recordingIMTx{
		refreshedRecipients: []scopeEventRecipient{{
			MemberID:       "mem_task_watch_no_index",
			ScopeKind:      "task",
			TaskWatchState: "mute",
		}},
	}
	appended, err := appendEventsForRecipients(
		context.Background(),
		tx,
		"wsp_task_watch_no_index",
		"tsk_task_watch_no_index",
		StreamEvent{
			EventID: "evt_task_watch_no_index",
			Type:    "conversation.updated",
			Payload: map[string]any{
				"action":  "unmute",
				"kind":    "task",
				"scopeId": "tsk_task_watch_no_index",
				"taskId":  "task_watch_no_index",
			},
		},
		nil,
		initial,
	)
	require.NoError(t, err)
	require.Len(t, appended, 1)
	require.Equal(t, 2, tx.queryCalls, "Task self action must re-read member state after locking the stream even without an index projection")
	require.Equal(t, "mute", appended[0].Payload["action"])
	require.Equal(t, "mute", appended[0].Payload["subscriptionState"])
	require.Len(t, tx.execSQL, 1, "no conversation projection should be manufactured")
	require.Contains(t, tx.execSQL[0], "INSERT INTO im_stream_events")
}

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

	recipientPinnedAt := time.Date(2026, 6, 10, 10, 0, 0, 0, time.UTC)
	event := StreamEvent{
		WorkspaceID: "wsp_im_projection_pin",
		MemberID:    "mem_projection_pin",
		EventSeq:    7,
		Type:        "conversation.unpinned",
		CreatedAt:   time.Date(2026, 6, 10, 10, 5, 0, 0, time.UTC),
	}
	item := conversationIndexItemForRecipient(ConversationIndexItem{
		ConversationID: "channel:chn_projection_pin",
		ScopeID:        "chn_projection_pin",
		Kind:           "channel",
		SortAt:         event.CreatedAt,
		PinnedAtSet:    true,
		Projection: map[string]any{
			"id":          "channel:chn_projection_pin",
			"kind":        "channel",
			"scopeId":     "chn_projection_pin",
			"title":       "General",
			"pinnedAt":    recipientPinnedAt.Format(time.RFC3339Nano),
			"unreadCount": float64(3),
			"hasUnread":   true,
		},
	}, event, scopeEventRecipient{
		MemberID:  "mem_projection_pin",
		ScopeKind: "channel",
		PinnedAt:  &recipientPinnedAt,
	}, "chn_projection_pin")

	require.NotNil(t, item.PinnedAt, "append-time recipient state must supersede a stale explicit unpin projection")
	require.Equal(t, recipientPinnedAt, item.PinnedAt.UTC())
	require.Equal(t, recipientPinnedAt, item.Projection["pinnedAt"])
	require.NotContains(t, item.Projection, "unreadCount")
	require.NotContains(t, item.Projection, "hasUnread")
}

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

	ctx := context.Background()
	db := startIMTestPool(t, ctx)
	insertIMWorkspaceMember(t, ctx, db, "wsp_im_scope_event_idempotent", "mem_scope_idempotent")
	insertIMScope(t, ctx, db, "wsp_im_scope_event_idempotent", "chn_im_scope_event_idempotent", "channel", "mem_scope_idempotent", 3)
	insertIMScopeMembership(t, ctx, db, "chn_im_scope_event_idempotent", "mem_scope_idempotent", 0, nil)

	store := New(db)
	entitySeq := int64(3)
	event := StreamEvent{
		EventID:   "message.created:msg_scope_idempotent",
		Type:      "message.created",
		EntityID:  "msg_scope_idempotent",
		EntitySeq: &entitySeq,
		Payload:   map[string]any{"messageId": "msg_scope_idempotent"},
		CreatedAt: time.Date(2026, 6, 10, 10, 0, 0, 0, time.UTC),
	}
	appended, err := store.AppendScopeEvent(ctx, ScopeEventRequest{
		WorkspaceID: "wsp_im_scope_event_idempotent",
		ScopeID:     "chn_im_scope_event_idempotent",
		Event:       event,
		Conversation: &ConversationIndexItem{
			Projection: map[string]any{"title": "original"},
		},
	})
	require.NoError(t, err)
	require.Len(t, appended, 1)

	event.Payload = map[string]any{"messageId": "msg_scope_idempotent", "duplicate": true}
	duplicate, err := store.AppendScopeEvent(ctx, ScopeEventRequest{
		WorkspaceID: "wsp_im_scope_event_idempotent",
		ScopeID:     "chn_im_scope_event_idempotent",
		Event:       event,
		Conversation: &ConversationIndexItem{
			Projection: map[string]any{"title": "duplicate"},
		},
	})
	require.NoError(t, err)
	require.Empty(t, duplicate)

	events, err := store.ListEventsAfter(ctx, "wsp_im_scope_event_idempotent", "mem_scope_idempotent", 0, 10)
	require.NoError(t, err)
	require.Len(t, events, 1)
	require.Equal(t, int64(1), events[0].EventSeq)
	require.Equal(t, map[string]any{"messageId": "msg_scope_idempotent"}, events[0].Payload)

	conversation, err := store.GetConversation(ctx, "wsp_im_scope_event_idempotent", "mem_scope_idempotent", "channel:chn_im_scope_event_idempotent")
	require.NoError(t, err)
	require.Equal(t, int64(1), conversation.LastEventSeq)
	require.Equal(t, "original", conversation.Projection["title"])
}

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

	const (
		workspaceID = "wsp_im_scope_event_late"
		memberID    = "mem_scope_event_late"
		scopeID     = "chn_im_scope_event_late"
	)
	ctx := context.Background()
	db := startIMTestPool(t, ctx)
	insertIMWorkspaceMember(t, ctx, db, workspaceID, memberID)
	insertIMScope(t, ctx, db, workspaceID, scopeID, "channel", memberID, 8)
	insertIMScopeMembership(t, ctx, db, scopeID, memberID, 0, nil)
	store := New(db)
	newerAt := time.Date(2026, 8, 3, 4, 0, 0, 0, time.UTC)
	olderAt := newerAt.Add(-time.Minute)

	appendMessage := func(messageID, title string, seq int64, createdAt time.Time) {
		t.Helper()
		_, err := store.AppendScopeEvent(ctx, ScopeEventRequest{
			WorkspaceID: workspaceID,
			ScopeID:     scopeID,
			Event: StreamEvent{
				EventID: "message.created:" + messageID, Type: "message.created",
				EntityID: messageID, EntitySeq: &seq, CreatedAt: createdAt,
			},
			Conversation: &ConversationIndexItem{
				SortAt: createdAt, Projection: map[string]any{"title": title},
			},
		})
		require.NoError(t, err)
	}

	appendMessage("msg_newer", "newer", 8, newerAt)
	appendMessage("msg_older_replay", "older", 7, olderAt)

	conversation, err := store.GetConversation(ctx, workspaceID, memberID, "channel:"+scopeID)
	require.NoError(t, err)
	require.Equal(t, newerAt, conversation.SortAt)
	require.Equal(t, "newer", conversation.Projection["title"])
	require.Equal(t, int64(2), conversation.LastEventSeq, "the late stream event still advances the member cursor")
}

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

	ctx := context.Background()
	db := startIMTestPool(t, ctx)
	insertIMWorkspaceMember(t, ctx, db, "wsp_im_member_event", "mem_member_owner")
	insertIMMember(t, ctx, db, "wsp_im_member_event", "mem_member_peer")
	insertIMScope(t, ctx, db, "wsp_im_member_event", "chn_im_member_event", "channel", "mem_member_owner", 3)
	insertIMScopeMembership(t, ctx, db, "chn_im_member_event", "mem_member_owner", 3, nil)
	insertIMScopeMembership(t, ctx, db, "chn_im_member_event", "mem_member_peer", 0, nil)

	store := New(db)
	event, err := store.AppendMemberEvent(ctx, "wsp_im_member_event", "mem_member_owner", "chn_im_member_event", StreamEvent{
		EventID: "evt_member_read",
		Type:    "conversation.read",
		Payload: map[string]any{"conversationId": "channel:chn_im_member_event"},
	}, &ConversationIndexItem{ConversationID: "channel:chn_im_member_event", Kind: "channel", Projection: map[string]any{"title": "general"}})
	require.NoError(t, err)
	require.Equal(t, "mem_member_owner", event.MemberID)
	require.Equal(t, int64(1), event.EventSeq)

	ownerLatest, err := store.GetLatestSeq(ctx, "wsp_im_member_event", "mem_member_owner")
	require.NoError(t, err)
	require.Equal(t, int64(1), ownerLatest)
	peerLatest, err := store.GetLatestSeq(ctx, "wsp_im_member_event", "mem_member_peer")
	require.NoError(t, err)
	require.Zero(t, peerLatest)

	_, err = store.GetConversation(ctx, "wsp_im_member_event", "mem_member_peer", "channel:chn_im_member_event")
	require.Error(t, err)
	ownerConversation, err := store.GetConversation(ctx, "wsp_im_member_event", "mem_member_owner", "channel:chn_im_member_event")
	require.NoError(t, err)
	require.Equal(t, int64(1), ownerConversation.LastEventSeq)
}

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

	ctx := context.Background()
	db := startIMTestPool(t, ctx)
	const (
		workspaceID = "wsp_task_watch_reversed"
		memberID    = "mem_task_watch_reversed"
		scopeID     = "tsk_task_watch_reversed"
	)
	insertIMWorkspaceMember(t, ctx, db, workspaceID, memberID)
	insertIMScope(t, ctx, db, workspaceID, scopeID, "task", memberID, 1)
	insertIMScopeMembership(t, ctx, db, scopeID, memberID, 0, nil)
	insertIMTask(t, ctx, db, workspaceID, scopeID, memberID)
	_, err := db.Exec(ctx, `
INSERT INTO scope_notification_settings (
  scope_id, member_id, state, task_watch_state, created_at, updated_at
) VALUES ($1, $2, 'active', 'active', NOW(), NOW())
`, scopeID, memberID)
	require.NoError(t, err)

	// Capture the older Watch projection, then commit Unwatch before either
	// durable self event is appended. Appending the newer Unwatch event first
	// and the stale Watch event second reproduces commit-order != append-order.
	staleWatchEvent := StreamEvent{
		EventID: "evt_task_watch_stale",
		Type:    "conversation.updated",
		Payload: map[string]any{
			"action":  "unmute",
			"kind":    "task",
			"scopeId": scopeID,
			"taskId":  "task_watch_reversed",
			"conversation": map[string]any{
				"id":                "task:" + scopeID,
				"kind":              "task",
				"scopeId":           scopeID,
				"subscriptionState": "active",
			},
		},
	}
	staleWatchConversation := ConversationIndexItem{
		ConversationID: "task:" + scopeID,
		ScopeID:        scopeID,
		Kind:           "task",
		SortAt:         time.Date(2026, 7, 29, 4, 0, 0, 0, time.UTC),
		Projection: map[string]any{
			"id":                "task:" + scopeID,
			"kind":              "task",
			"scopeId":           scopeID,
			"subscriptionState": "active",
			"title":             "Watch ordering",
		},
	}
	_, err = db.Exec(ctx, `
UPDATE scope_notification_settings
SET state = 'mute', task_watch_state = 'mute', updated_at = NOW()
WHERE scope_id = $1 AND member_id = $2
`, scopeID, memberID)
	require.NoError(t, err)

	store := New(db)
	currentUnwatch, err := store.AppendMemberEvent(ctx, workspaceID, memberID, scopeID, StreamEvent{
		EventID: "evt_task_unwatch_current",
		Type:    "conversation.updated",
		Payload: map[string]any{
			"action":  "mute",
			"kind":    "task",
			"scopeId": scopeID,
			"taskId":  "task_watch_reversed",
			"conversation": map[string]any{
				"id":                "task:" + scopeID,
				"kind":              "task",
				"scopeId":           scopeID,
				"subscriptionState": "mute",
			},
		},
	}, &staleWatchConversation)
	require.NoError(t, err)
	require.Equal(t, "mute", currentUnwatch.Payload["action"])

	reversedStaleWatch, err := store.AppendMemberEvent(
		ctx,
		workspaceID,
		memberID,
		scopeID,
		staleWatchEvent,
		&staleWatchConversation,
	)
	require.NoError(t, err)
	require.Equal(t, "mute", reversedStaleWatch.Payload["action"])
	require.Equal(t, "mute", reversedStaleWatch.Payload["subscriptionState"])
	reconciledPayload := reversedStaleWatch.Payload["conversation"].(map[string]any)
	require.Equal(t, "mute", reconciledPayload["subscriptionState"])

	conversation, err := store.GetConversation(ctx, workspaceID, memberID, "task:"+scopeID)
	require.NoError(t, err)
	require.Equal(t, "mute", conversation.SubscriptionState)
	require.Equal(t, "mute", conversation.Projection["subscriptionState"])
	require.Equal(t, int64(2), conversation.LastEventSeq)
}

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

	ctx := context.Background()
	db := startIMTestPool(t, ctx)
	const (
		workspaceID = "wsp_task_pin_reversed"
		memberID    = "mem_task_pin_reversed"
		scopeID     = "tsk_task_pin_reversed"
	)
	insertIMWorkspaceMember(t, ctx, db, workspaceID, memberID)
	insertIMScope(t, ctx, db, workspaceID, scopeID, "task", memberID, 1)
	insertIMScopeMembership(t, ctx, db, scopeID, memberID, 0, nil)
	insertIMTask(t, ctx, db, workspaceID, scopeID, memberID)
	pinnedAt := time.Date(2026, 7, 29, 4, 10, 0, 0, time.UTC)
	_, err := db.Exec(ctx, `
INSERT INTO conversation_member_states (
  scope_id, member_id, pinned_at, created_at, updated_at
) VALUES ($1, $2, $3, $3, $3)
`, scopeID, memberID, pinnedAt)
	require.NoError(t, err)

	store := New(db)
	base := ConversationIndexItem{
		ConversationID: "task:" + scopeID,
		ScopeID:        scopeID,
		Kind:           "task",
		SortAt:         pinnedAt.Add(-time.Hour),
		Projection: map[string]any{
			"id":      "task:" + scopeID,
			"kind":    "task",
			"scopeId": scopeID,
			"title":   "Pin ordering",
		},
	}
	currentPin, err := store.AppendMemberEvent(ctx, workspaceID, memberID, scopeID, StreamEvent{
		EventID: "evt_task_pin_current",
		Type:    "conversation.updated",
		Payload: map[string]any{
			"action":  "pin",
			"kind":    "task",
			"scopeId": scopeID,
			"taskId":  "task_pin_reversed",
			"conversation": map[string]any{
				"id":       "task:" + scopeID,
				"kind":     "task",
				"scopeId":  scopeID,
				"pinnedAt": pinnedAt.Format(time.RFC3339Nano),
			},
		},
	}, &base)
	require.NoError(t, err)
	require.Equal(t, "pin", currentPin.Payload["action"])

	staleUnpinBase := base
	staleUnpinBase.PinnedAtSet = true
	reversedStaleUnpin, err := store.AppendMemberEvent(ctx, workspaceID, memberID, scopeID, StreamEvent{
		EventID: "evt_task_unpin_stale",
		Type:    "conversation.updated",
		Payload: map[string]any{
			"action":  "unpin",
			"kind":    "task",
			"scopeId": scopeID,
			"taskId":  "task_pin_reversed",
			"conversation": map[string]any{
				"id":       "task:" + scopeID,
				"kind":     "task",
				"scopeId":  scopeID,
				"pinnedAt": nil,
			},
		},
	}, &staleUnpinBase)
	require.NoError(t, err)
	require.Equal(t, "pin", reversedStaleUnpin.Payload["action"])
	require.Equal(t, pinnedAt.Format(time.RFC3339Nano), reversedStaleUnpin.Payload["pinnedAt"])

	conversation, err := store.GetConversation(ctx, workspaceID, memberID, "task:"+scopeID)
	require.NoError(t, err)
	require.NotNil(t, conversation.PinnedAt)
	require.Equal(t, pinnedAt, conversation.PinnedAt.UTC())
	require.Equal(t, int64(2), conversation.LastEventSeq)
}

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

	ctx := context.Background()
	db := startIMTestPool(t, ctx)
	const (
		workspaceID = "wsp_task_hide_reversed"
		memberID    = "mem_task_hide_reversed"
		scopeID     = "tsk_task_hide_reversed"
	)
	insertIMWorkspaceMember(t, ctx, db, workspaceID, memberID)
	insertIMScope(t, ctx, db, workspaceID, scopeID, "task", memberID, 5)
	insertIMScopeMembership(t, ctx, db, scopeID, memberID, 0, nil)
	insertIMTask(t, ctx, db, workspaceID, scopeID, memberID)
	_, err := db.Exec(ctx, `UPDATE scopes SET latest_activity_seq = 5 WHERE id = $1`, scopeID)
	require.NoError(t, err)
	hiddenAt := time.Date(2026, 7, 29, 4, 20, 0, 0, time.UTC)
	_, err = db.Exec(ctx, `
INSERT INTO conversation_member_states (
  scope_id, member_id, hidden_at, hidden_through_seq,
  hidden_through_activity_seq, created_at, updated_at
) VALUES ($1, $2, $3, 5, 5, $3, $3)
`, scopeID, memberID, hiddenAt)
	require.NoError(t, err)

	// A later Task Activity has crossed its independent hidden frontier while
	// the message sequence remains exactly at the hidden message frontier.
	// During the expand-only release that Activity must not reveal the Task.
	_, err = db.Exec(ctx, `
UPDATE scopes
SET latest_activity_seq = 6, updated_at = $2
WHERE id = $1
`, scopeID, hiddenAt.Add(time.Minute))
	require.NoError(t, err)

	store := New(db)
	// Deliberately use the same sort timestamp as the stale hide. This proves
	// correctness comes from explicitly skipping the stale projection, not from
	// the Task upsert's older-sort guard.
	laterSortAt := hiddenAt
	laterActivity, err := store.AppendMemberEvent(ctx, workspaceID, memberID, scopeID, StreamEvent{
		EventID:   "evt_task_later_attention",
		Type:      "task.activity",
		EntityID:  "act_task_later_attention",
		Payload:   map[string]any{"scopeId": scopeID, "activityId": "act_task_later_attention"},
		CreatedAt: laterSortAt,
	}, &ConversationIndexItem{
		ConversationID: "task:" + scopeID,
		ScopeID:        scopeID,
		Kind:           "task",
		SortAt:         laterSortAt,
		Projection: map[string]any{
			"id":      "task:" + scopeID,
			"kind":    "task",
			"scopeId": scopeID,
			"title":   "Later attention",
		},
	})
	require.NoError(t, err)
	require.Equal(t, "task.activity", laterActivity.Type)

	afterActivity, err := store.GetConversation(ctx, workspaceID, memberID, "task:"+scopeID)
	require.NoError(t, err)
	require.NotNil(t, afterActivity.ArchivedAt)
	require.Equal(t, true, afterActivity.Projection["archived"])
	require.Equal(t, true, afterActivity.Projection["sidebarHidden"])
	require.Equal(t, "Later attention", afterActivity.Projection["title"])
	visible, err := store.ListConversations(ctx, workspaceID, memberID, 10, ConversationCursor{}, []string{"task"})
	require.NoError(t, err)
	require.Empty(t, visible.Items, "Activity after hide must not restore the Task Inbox row")

	delayedHide, err := store.AppendMemberEvent(ctx, workspaceID, memberID, scopeID, StreamEvent{
		EventID: "evt_task_hide_stale",
		Type:    "conversation.archived",
		Payload: map[string]any{
			"action":  "hide",
			"kind":    "task",
			"scopeId": scopeID,
			"taskId":  "task_hide_reversed",
			"conversation": map[string]any{
				"id":            "task:" + scopeID,
				"kind":          "task",
				"scopeId":       scopeID,
				"archived":      true,
				"sidebarHidden": true,
			},
		},
		CreatedAt: hiddenAt,
	}, &ConversationIndexItem{
		ConversationID: "task:" + scopeID,
		ScopeID:        scopeID,
		Kind:           "task",
		SortAt:         hiddenAt,
		ArchivedAt:     &hiddenAt,
		Projection: map[string]any{
			"id":            "task:" + scopeID,
			"kind":          "task",
			"scopeId":       scopeID,
			"title":         "Stale hide",
			"archived":      true,
			"sidebarHidden": true,
		},
	})
	require.NoError(t, err)
	require.Equal(t, "conversation.archived", delayedHide.Type)
	require.Equal(t, int64(2), delayedHide.EventSeq)
	require.Equal(t, "hide", delayedHide.Payload["action"])
	delayedHideConversation := delayedHide.Payload["conversation"].(map[string]any)
	require.Equal(t, true, delayedHideConversation["archived"])
	require.Equal(t, true, delayedHideConversation["sidebarHidden"])

	conversation, err := store.GetConversation(ctx, workspaceID, memberID, "task:"+scopeID)
	require.NoError(t, err)
	require.NotNil(t, conversation.ArchivedAt)
	require.Equal(t, "Stale hide", conversation.Projection["title"])
	require.Equal(t, true, conversation.Projection["archived"])
	require.Equal(t, true, conversation.Projection["sidebarHidden"])
	require.Equal(t, int64(2), conversation.LastEventSeq)
	visible, err = store.ListConversations(ctx, workspaceID, memberID, 10, ConversationCursor{}, []string{"task"})
	require.NoError(t, err)
	require.Empty(t, visible.Items)
}

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

	ctx := context.Background()
	db := startIMTestPool(t, ctx)
	const (
		workspaceID = "wsp_task_hide_after_message"
		memberID    = "mem_task_hide_after_message"
		scopeID     = "tsk_task_hide_after_message"
	)
	insertIMWorkspaceMember(t, ctx, db, workspaceID, memberID)
	insertIMScope(t, ctx, db, workspaceID, scopeID, "task", memberID, 5)
	insertIMScopeMembership(t, ctx, db, scopeID, memberID, 0, nil)
	insertIMTask(t, ctx, db, workspaceID, scopeID, memberID)
	_, err := db.Exec(ctx, `UPDATE scopes SET latest_activity_seq = 5 WHERE id = $1`, scopeID)
	require.NoError(t, err)
	hiddenAt := time.Date(2026, 7, 29, 4, 30, 0, 0, time.UTC)
	_, err = db.Exec(ctx, `
INSERT INTO conversation_member_states (
  scope_id, member_id, hidden_at, hidden_through_seq,
  hidden_through_activity_seq, created_at, updated_at
) VALUES ($1, $2, $3, 5, 5, $3, $3)
`, scopeID, memberID, hiddenAt)
	require.NoError(t, err)

	// A later Task message crosses the resolved message hide frontier before
	// either its projection callback or the delayed hide callback arrives.
	_, err = db.Exec(ctx, `
UPDATE scopes
SET latest_seq = 6, updated_at = $2
WHERE id = $1
`, scopeID, hiddenAt.Add(time.Minute))
	require.NoError(t, err)

	store := New(db)
	laterSortAt := hiddenAt
	laterMessage, err := store.AppendMemberEvent(ctx, workspaceID, memberID, scopeID, StreamEvent{
		EventID:   "evt_task_later_message",
		Type:      "message.created",
		EntityID:  "msg_task_later_message",
		Payload:   map[string]any{"scopeId": scopeID, "messageId": "msg_task_later_message"},
		CreatedAt: laterSortAt,
	}, &ConversationIndexItem{
		ConversationID: "task:" + scopeID,
		ScopeID:        scopeID,
		Kind:           "task",
		SortAt:         laterSortAt,
		Projection: map[string]any{
			"id":      "task:" + scopeID,
			"kind":    "task",
			"scopeId": scopeID,
			"title":   "Later message",
		},
	})
	require.NoError(t, err)
	require.Equal(t, "message.created", laterMessage.Type)

	afterMessage, err := store.GetConversation(ctx, workspaceID, memberID, "task:"+scopeID)
	require.NoError(t, err)
	require.Nil(t, afterMessage.ArchivedAt)
	require.Equal(t, false, afterMessage.Projection["archived"])
	require.Equal(t, false, afterMessage.Projection["sidebarHidden"])
	require.Equal(t, "Later message", afterMessage.Projection["title"])
	visible, err := store.ListConversations(ctx, workspaceID, memberID, 10, ConversationCursor{}, []string{"task"})
	require.NoError(t, err)
	require.Len(t, visible.Items, 1, "message after hide must restore the Task Inbox row")
	require.Equal(t, "task:"+scopeID, visible.Items[0].ConversationID)

	delayedHide, err := store.AppendMemberEvent(ctx, workspaceID, memberID, scopeID, StreamEvent{
		EventID: "evt_task_hide_after_message",
		Type:    "conversation.archived",
		Payload: map[string]any{
			"action":  "hide",
			"kind":    "task",
			"scopeId": scopeID,
			"taskId":  "task_hide_after_message",
			"conversation": map[string]any{
				"id":            "task:" + scopeID,
				"kind":          "task",
				"scopeId":       scopeID,
				"archived":      true,
				"sidebarHidden": true,
			},
		},
		CreatedAt: hiddenAt,
	}, &ConversationIndexItem{
		ConversationID: "task:" + scopeID,
		ScopeID:        scopeID,
		Kind:           "task",
		SortAt:         hiddenAt,
		ArchivedAt:     &hiddenAt,
		Projection: map[string]any{
			"id":            "task:" + scopeID,
			"kind":          "task",
			"scopeId":       scopeID,
			"title":         "Stale hide",
			"archived":      true,
			"sidebarHidden": true,
		},
	})
	require.NoError(t, err)
	require.Equal(t, "conversation.updated", delayedHide.Type)
	require.Equal(t, int64(2), delayedHide.EventSeq, "the stale request still advances the durable member stream")
	require.NotContains(t, delayedHide.Payload, "action")
	require.NotContains(t, delayedHide.Payload, "conversation")

	conversation, err := store.GetConversation(ctx, workspaceID, memberID, "task:"+scopeID)
	require.NoError(t, err)
	require.Nil(t, conversation.ArchivedAt)
	require.Equal(t, "Later message", conversation.Projection["title"])
	require.Equal(t, false, conversation.Projection["archived"])
	require.Equal(t, false, conversation.Projection["sidebarHidden"])
	require.Equal(t, int64(1), conversation.LastEventSeq, "the no-op event must not rewrite any conversation index column")
	visible, err = store.ListConversations(ctx, workspaceID, memberID, 10, ConversationCursor{}, []string{"task"})
	require.NoError(t, err)
	require.Len(t, visible.Items, 1)
	require.Equal(t, "task:"+scopeID, visible.Items[0].ConversationID)
}

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

	ctx := context.Background()
	db := startIMTestPool(t, ctx)
	insertIMWorkspaceMember(t, ctx, db, "wsp_im_member_event_no_membership", "mem_member_no_membership")
	insertIMScope(t, ctx, db, "wsp_im_member_event_no_membership", "chn_im_member_event_no_membership", "channel", "mem_member_no_membership", 4)
	insertIMScope(t, ctx, db, "wsp_im_member_event_no_membership", "thr_im_member_event_no_membership", "thread", "mem_member_no_membership", 4)
	insertIMScopeMembership(t, ctx, db, "chn_im_member_event_no_membership", "mem_member_no_membership", 0, nil)
	insertIMScopeNotification(t, ctx, db, "thr_im_member_event_no_membership", "mem_member_no_membership")
	_, err := db.Exec(ctx, `UPDATE scopes SET parent_scope_id = $1 WHERE id = $2`, "chn_im_member_event_no_membership", "thr_im_member_event_no_membership")
	require.NoError(t, err)
	_, err = db.Exec(ctx, `
INSERT INTO scope_channel_profiles (
  scope_id, workspace_id, slug, display_name, description, visibility,
  subsystem, owner_member_id, is_all_channel, created_at, updated_at
) VALUES ($1, $3, 'knowledge', 'Knowledge', '', 'private', 'kb', $4, 0, NOW(), NOW());
`, "chn_im_member_event_no_membership", "thr_im_member_event_no_membership", "wsp_im_member_event_no_membership", "mem_member_no_membership")
	require.NoError(t, err)

	hiddenAt := time.Date(2026, 7, 14, 3, 0, 0, 0, time.UTC)
	_, err = db.Exec(ctx, `
INSERT INTO conversation_member_states (
  scope_id, member_id, hidden_at, hidden_through_seq, created_at, updated_at
) VALUES ($1, $2, $3, 3, $3, $3)
`, "thr_im_member_event_no_membership", "mem_member_no_membership", hiddenAt)
	require.NoError(t, err)

	store := New(db)
	require.NoError(t, store.UpsertConversation(ctx, "wsp_im_member_event_no_membership", "mem_member_no_membership", ConversationIndexItem{
		ConversationID: "thread:thr_im_member_event_no_membership",
		ScopeID:        "thr_im_member_event_no_membership",
		Kind:           "thread",
		SortAt:         hiddenAt,
		ArchivedAt:     &hiddenAt,
		Projection:     map[string]any{"title": "thread"},
	}))

	_, err = store.AppendMemberEvent(ctx, "wsp_im_member_event_no_membership", "mem_member_no_membership", "thr_im_member_event_no_membership", StreamEvent{
		EventID: "evt_thread_read_no_projection",
		Type:    "conversation.read",
	}, nil)
	require.ErrorContains(t, err, "member is not active in scope")

	event, err := store.AppendMemberEvent(ctx, "wsp_im_member_event_no_membership", "mem_member_no_membership", "thr_im_member_event_no_membership", StreamEvent{
		EventID: "evt_thread_read_no_membership",
		Type:    "conversation.read",
		Payload: map[string]any{"conversationId": "thread:thr_im_member_event_no_membership"},
	}, &ConversationIndexItem{
		ConversationID: "thread:thr_im_member_event_no_membership",
		ScopeID:        "thr_im_member_event_no_membership",
		Kind:           "thread",
		Projection:     map[string]any{"title": "thread"},
	})
	require.NoError(t, err)
	require.Equal(t, int64(1), event.EventSeq)

	conversation, err := store.GetConversation(ctx, "wsp_im_member_event_no_membership", "mem_member_no_membership", "thread:thr_im_member_event_no_membership")
	require.NoError(t, err)
	require.Equal(t, int64(1), conversation.LastEventSeq)
	require.Nil(t, conversation.ArchivedAt)
	require.Equal(t, false, conversation.Projection["archived"])
	require.NotContains(t, conversation.Projection, "unreadCount")
	require.NotContains(t, conversation.Projection, "hasUnread")
	require.NotContains(t, conversation.Projection, "manualUnread")
	require.Equal(t, "kb", conversation.Projection["parentScopeType"])

	var membershipCount int64
	require.NoError(t, db.QueryRow(ctx, `
SELECT COUNT(*)
FROM scope_memberships
WHERE scope_id = $1 AND member_id = $2
`, "thr_im_member_event_no_membership", "mem_member_no_membership").Scan(&membershipCount))
	require.Zero(t, membershipCount)
}

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

	ctx := context.Background()
	db := startIMTestPool(t, ctx)
	const (
		workspaceID = "wsp_im_parent_scope_type"
		ownerID     = "mem_im_parent_scope_owner"
		peerID      = "mem_im_parent_scope_peer"
		kbParentID  = "chn_im_parent_scope_kb"
		kbThreadID  = "thr_im_parent_scope_kb"
	)
	insertIMWorkspaceMember(t, ctx, db, workspaceID, ownerID)
	insertIMMember(t, ctx, db, workspaceID, peerID)
	insertIMScope(t, ctx, db, workspaceID, kbParentID, "channel", ownerID, 1)
	insertIMScope(t, ctx, db, workspaceID, kbThreadID, "thread", ownerID, 1)
	insertIMScopeMembership(t, ctx, db, kbParentID, ownerID, 0, nil)
	insertIMScopeMembership(t, ctx, db, kbParentID, peerID, 0, nil)
	insertIMScopeMembership(t, ctx, db, kbThreadID, ownerID, 0, nil)
	insertIMScopeMembership(t, ctx, db, kbThreadID, peerID, 0, nil)
	insertIMScopeNotification(t, ctx, db, kbThreadID, ownerID)
	insertIMScopeNotification(t, ctx, db, kbThreadID, peerID)
	_, err := db.Exec(ctx, `UPDATE scopes SET parent_scope_id = $1 WHERE id = $2`, kbParentID, kbThreadID)
	require.NoError(t, err)
	_, err = db.Exec(ctx, `
INSERT INTO scope_channel_profiles (
  scope_id, workspace_id, slug, display_name, description, visibility,
  subsystem, owner_member_id, is_all_channel, created_at, updated_at
) VALUES ($1, $3, 'knowledge-fanout', 'Knowledge', '', 'private', 'kb', $4, 0, NOW(), NOW());
`, kbParentID, kbThreadID, workspaceID, ownerID)
	require.NoError(t, err)

	store := New(db)
	_, err = store.AppendScopeEvent(ctx, ScopeEventRequest{
		WorkspaceID: workspaceID,
		ScopeID:     kbThreadID,
		Event:       StreamEvent{EventID: "evt_parent_scope_type", Type: "message.created"},
		Conversation: &ConversationIndexItem{
			ConversationID: "thread:" + kbThreadID,
			ScopeID:        kbThreadID,
			Kind:           "thread",
			Projection:     map[string]any{"title": "Knowledge thread", "parentScopeType": "channel"},
		},
	})
	require.NoError(t, err)

	for _, memberID := range []string{ownerID, peerID} {
		conversation, err := store.GetConversation(ctx, workspaceID, memberID, "thread:"+kbThreadID)
		require.NoError(t, err)
		require.Equal(t, "kb", conversation.Projection["parentScopeType"])
	}

	for _, testCase := range []struct {
		name       string
		parentKind string
		want       string
	}{
		{name: "channel", parentKind: "channel", want: "channel"},
		{name: "dm", parentKind: "dm", want: "dm"},
	} {
		parentID := testCase.parentKind + "_im_parent_scope_runtime"
		threadID := "thr_im_parent_scope_" + testCase.name
		insertIMScope(t, ctx, db, workspaceID, parentID, testCase.parentKind, ownerID, 1)
		insertIMScope(t, ctx, db, workspaceID, threadID, "thread", ownerID, 1)
		insertIMScopeMembership(t, ctx, db, parentID, ownerID, 0, nil)
		insertIMScopeMembership(t, ctx, db, threadID, ownerID, 0, nil)
		insertIMScopeNotification(t, ctx, db, threadID, ownerID)
		_, err := db.Exec(ctx, `UPDATE scopes SET parent_scope_id = $1 WHERE id = $2`, parentID, threadID)
		require.NoError(t, err)
		_, err = store.AppendScopeEvent(ctx, ScopeEventRequest{
			WorkspaceID: workspaceID,
			ScopeID:     threadID,
			Event:       StreamEvent{EventID: "evt_parent_scope_type_" + testCase.name, Type: "message.created"},
			Conversation: &ConversationIndexItem{
				ConversationID: "thread:" + threadID,
				ScopeID:        threadID,
				Kind:           "thread",
				Projection:     map[string]any{"title": testCase.name, "parentScopeType": "stale"},
			},
		})
		require.NoError(t, err)
		conversation, err := store.GetConversation(ctx, workspaceID, ownerID, "thread:"+threadID)
		require.NoError(t, err)
		require.Equal(t, testCase.want, conversation.Projection["parentScopeType"])
	}

	const noParentScopeID = "chn_im_parent_scope_none"
	insertIMScope(t, ctx, db, workspaceID, noParentScopeID, "channel", ownerID, 1)
	insertIMScopeMembership(t, ctx, db, noParentScopeID, ownerID, 0, nil)
	_, err = store.AppendScopeEvent(ctx, ScopeEventRequest{
		WorkspaceID: workspaceID,
		ScopeID:     noParentScopeID,
		Event:       StreamEvent{EventID: "evt_parent_scope_type_none", Type: "message.created"},
		Conversation: &ConversationIndexItem{
			ConversationID: "channel:" + noParentScopeID,
			ScopeID:        noParentScopeID,
			Kind:           "channel",
			Projection:     map[string]any{"title": "No parent", "parentScopeType": "stale"},
		},
	})
	require.NoError(t, err)
	conversation, err := store.GetConversation(ctx, workspaceID, ownerID, "channel:"+noParentScopeID)
	require.NoError(t, err)
	require.Equal(t, "", conversation.Projection["parentScopeType"])
}

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

	ctx := context.Background()
	db := startIMTestPool(t, ctx)
	const (
		workspaceID    = "wsp_im_deleted_service_conversation"
		memberID       = "mem_im_deleted_service_conversation"
		scopeID        = "sasc_im_deleted_service_conversation"
		conversationID = "serviceAccount:" + scopeID
	)
	insertIMWorkspaceMember(t, ctx, db, workspaceID, memberID)
	insertIMScope(t, ctx, db, workspaceID, scopeID, "service_account", memberID, 1)
	insertIMScopeMembership(t, ctx, db, scopeID, memberID, 0, nil)

	store := New(db)
	sortAt := time.Date(2026, 7, 14, 12, 0, 0, 0, time.UTC)
	require.NoError(t, store.UpsertConversation(ctx, workspaceID, memberID, ConversationIndexItem{
		ConversationID: conversationID,
		ScopeID:        scopeID,
		Kind:           "service_account",
		SortAt:         sortAt,
		Projection: map[string]any{
			"id":       conversationID,
			"scopeId":  scopeID,
			"kind":     "serviceAccount",
			"route":    "/inbox/service-accounts/sacct_deleted",
			"title":    "Deleted Service Account",
			"peerType": "service",
			"peerId":   "sacct_deleted",
		},
	}))

	_, err := db.Exec(ctx, `DELETE FROM scopes WHERE workspace_id = $1 AND id = $2`, workspaceID, scopeID)
	require.NoError(t, err)

	archivedAt := sortAt.Add(time.Hour)
	payload := map[string]any{"action": "hide"}
	event, archived, err := store.ArchiveIndexedConversation(ctx, IndexedConversationArchiveRequest{
		WorkspaceID:    workspaceID,
		MemberID:       memberID,
		ConversationID: conversationID,
		Event: StreamEvent{
			EventID:   "evt_hide_deleted_service_conversation",
			Type:      "conversation.archived",
			Payload:   payload,
			CreatedAt: archivedAt,
		},
	})
	require.NoError(t, err)
	require.Equal(t, conversationID, event.ConversationID)
	require.Equal(t, memberID, event.MemberID)
	require.NotContains(t, payload, "conversation", "archive must not mutate the caller's event payload")
	eventConversation, ok := event.Payload["conversation"].(map[string]any)
	require.True(t, ok)
	require.Equal(t, "serviceAccount", eventConversation["kind"])
	require.Equal(t, true, eventConversation["archived"])
	require.Equal(t, true, eventConversation["sidebarHidden"])
	require.EqualValues(t, 0, eventConversation["unreadCount"])
	require.Equal(t, false, eventConversation["manualUnread"])
	require.NotNil(t, archived.ArchivedAt)
	require.Equal(t, archivedAt, archived.ArchivedAt.UTC())
	require.Equal(t, event.EventSeq, archived.LastEventSeq)
	require.Equal(t, true, archived.Projection["archived"])
	require.Equal(t, true, archived.Projection["sidebarHidden"])
	require.Equal(t, "Deleted Service Account", archived.Projection["title"])

	visible, err := store.ListConversations(ctx, workspaceID, memberID, 10, ConversationCursor{}, nil)
	require.NoError(t, err)
	require.Empty(t, visible.Items)
}

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

	ctx := context.Background()
	db := startIMTestPool(t, ctx)
	const (
		workspaceID    = "wsp_im_deleted_peer_conversation"
		memberID       = "mem_im_deleted_peer_conversation"
		scopeID        = "dm_im_deleted_peer_conversation"
		conversationID = "dm:" + scopeID
	)
	insertIMWorkspaceMember(t, ctx, db, workspaceID, memberID)
	insertIMScope(t, ctx, db, workspaceID, scopeID, "dm", memberID, 1)
	insertIMScopeMembership(t, ctx, db, scopeID, memberID, 0, nil)

	store := New(db)
	base := time.Date(2026, 7, 14, 12, 0, 0, 0, time.UTC)
	require.NoError(t, store.UpsertConversation(ctx, workspaceID, memberID, ConversationIndexItem{
		ConversationID: conversationID,
		ScopeID:        scopeID,
		Kind:           "dm",
		SortAt:         base,
		Projection: map[string]any{
			"id":          conversationID,
			"scopeId":     scopeID,
			"kind":        "agentDm",
			"route":       "/inbox/agent/agt_deleted_peer",
			"title":       "Deleted Agent",
			"peerType":    "agent",
			"peerId":      "agt_deleted_peer",
			"unreadCount": 1,
		},
	}))

	archiveEvent, archived, err := store.ArchiveIndexedConversation(ctx, IndexedConversationArchiveRequest{
		WorkspaceID:    workspaceID,
		MemberID:       memberID,
		ConversationID: conversationID,
		Event:          StreamEvent{Type: "conversation.archived", CreatedAt: base.Add(time.Minute)},
	})
	require.NoError(t, err)
	require.NotNil(t, archived.ArchivedAt)
	archivePayload, ok := archiveEvent.Payload["conversation"].(map[string]any)
	require.True(t, ok)
	require.Equal(t, "agentDm", archivePayload["kind"])
	require.EqualValues(t, 0, archivePayload["unreadCount"])

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

	_, err = store.AppendScopeEvent(ctx, ScopeEventRequest{
		WorkspaceID: workspaceID,
		ScopeID:     scopeID,
		Event:       StreamEvent{Type: "reaction.changed", CreatedAt: base.Add(2 * time.Minute)},
		Conversation: &ConversationIndexItem{
			ConversationID: conversationID,
			ScopeID:        scopeID,
			Kind:           "dm",
			SortAt:         base.Add(2 * time.Minute),
			Projection:     map[string]any{"title": "Deleted Agent", "peerType": "agent", "peerId": "agt_deleted_peer"},
		},
	})
	require.NoError(t, err)
	afterActivity, err := store.GetConversation(ctx, workspaceID, memberID, conversationID)
	require.NoError(t, err)
	require.NotNil(t, afterActivity.ArchivedAt, "non-message activity at the hidden sequence must not revive the row")

	_, err = db.Exec(ctx, `UPDATE scopes SET latest_seq = 2 WHERE workspace_id = $1 AND id = $2`, workspaceID, scopeID)
	require.NoError(t, err)
	_, err = store.AppendScopeEvent(ctx, ScopeEventRequest{
		WorkspaceID: workspaceID,
		ScopeID:     scopeID,
		Event:       StreamEvent{Type: "message.created", CreatedAt: base.Add(3 * time.Minute)},
		Conversation: &ConversationIndexItem{
			ConversationID: conversationID,
			ScopeID:        scopeID,
			Kind:           "dm",
			SortAt:         base.Add(3 * time.Minute),
			Projection:     map[string]any{"title": "Deleted Agent", "peerType": "agent", "peerId": "agt_deleted_peer"},
		},
	})
	require.NoError(t, err)
	afterMessage, err := store.GetConversation(ctx, workspaceID, memberID, conversationID)
	require.NoError(t, err)
	require.Nil(t, afterMessage.ArchivedAt, "a newer message must unhide the row according to hidden_through_seq")
}

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

	ctx := context.Background()
	db := startIMTestPool(t, ctx)
	const (
		workspaceID    = "wsp_im_refresh_hidden_recipient"
		memberID       = "mem_im_refresh_hidden_recipient"
		scopeID        = "dm_im_refresh_hidden_recipient"
		conversationID = "dm:" + scopeID
	)
	insertIMWorkspaceMember(t, ctx, db, workspaceID, memberID)
	insertIMScope(t, ctx, db, workspaceID, scopeID, "dm", memberID, 1)
	insertIMScopeMembership(t, ctx, db, scopeID, memberID, 0, nil)

	store := New(db)
	base := time.Date(2026, 7, 14, 13, 0, 0, 0, time.UTC)
	require.NoError(t, store.UpsertConversation(ctx, workspaceID, memberID, ConversationIndexItem{
		ConversationID: conversationID,
		ScopeID:        scopeID,
		Kind:           "dm",
		SortAt:         base,
		Projection:     map[string]any{"title": "Deleted Agent", "peerType": "agent", "peerId": "agt_deleted_peer"},
	}))
	require.NoError(t, store.AppendEvents(ctx, workspaceID, memberID, []StreamEvent{{
		EventID:   "evt_im_refresh_hidden_baseline",
		Type:      "conversation.created",
		ScopeID:   scopeID,
		CreatedAt: base,
	}}))

	err := coredb.WithTx(ctx, db, func(tx coredb.Tx) error {
		// The source event sampled its recipients before the hide committed.
		staleRecipients, err := loadScopeEventRecipients(ctx, tx, workspaceID, scopeID)
		if err != nil {
			return err
		}
		require.Len(t, staleRecipients, 1)
		require.Nil(t, staleRecipients[0].ArchivedAt)

		// A concurrent hide commits while the source event is still waiting to
		// serialize its member-local stream event.
		hiddenAt := base.Add(time.Minute)
		_, err = db.Exec(ctx, `
INSERT INTO conversation_member_states (
  scope_id, member_id, hidden_at, hidden_through_seq, created_at, updated_at
) VALUES ($1, $2, $3, 1, $3, $3)
`, scopeID, memberID, hiddenAt)
		if err != nil {
			return err
		}

		_, err = appendEventsForRecipients(ctx, tx, workspaceID, scopeID, StreamEvent{
			EventID:   "evt_im_refresh_hidden_activity",
			Type:      "reaction.changed",
			CreatedAt: base.Add(2 * time.Minute),
		}, &ConversationIndexItem{
			ConversationID: conversationID,
			ScopeID:        scopeID,
			Kind:           "dm",
			SortAt:         base.Add(2 * time.Minute),
			Projection:     map[string]any{"title": "Deleted Agent", "peerType": "agent", "peerId": "agt_deleted_peer"},
		}, staleRecipients)
		return err
	})
	require.NoError(t, err)

	conversation, err := store.GetConversation(ctx, workspaceID, memberID, conversationID)
	require.NoError(t, err)
	require.NotNil(t, conversation.ArchivedAt, "the later projection write must refresh CMS instead of reviving from a stale recipient snapshot")
}

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

	item := conversationIndexItemForRecipient(
		ConversationIndexItem{
			ConversationID: "thread:thr_kb",
			ScopeID:        "thr_kb",
			Kind:           "thread",
			Projection:     map[string]any{"parentScopeType": "channel"},
		},
		StreamEvent{WorkspaceID: "wsp_1", EventSeq: 1, CreatedAt: time.Now()},
		scopeEventRecipient{MemberID: "mem_1", ScopeKind: "thread", ParentScopeType: "kb"},
		"thr_kb",
	)

	require.Equal(t, "kb", item.Projection["parentScopeType"])
	require.Equal(t, "", mergeConversationProjection(
		map[string]any{"parentScopeType": "stale"},
		ConversationIndexItem{ParentScopeTypeSet: true},
	)["parentScopeType"])
}

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

	ctx := context.Background()
	db := startIMTestPool(t, ctx)
	const (
		workspaceID    = "wsp_im_archive_authority"
		ownerMemberID  = "mem_im_archive_authority_owner"
		otherMemberID  = "mem_im_archive_authority_other"
		conversationID = "dm:dm_im_archive_authority"
	)
	insertIMWorkspaceMember(t, ctx, db, workspaceID, ownerMemberID)
	insertIMMember(t, ctx, db, workspaceID, otherMemberID)
	store := New(db)
	require.NoError(t, store.UpsertConversation(ctx, workspaceID, ownerMemberID, ConversationIndexItem{
		ConversationID: conversationID,
		ScopeID:        "dm_im_archive_authority",
		Kind:           "dm",
		SortAt:         time.Date(2026, 7, 14, 12, 0, 0, 0, time.UTC),
		Projection:     map[string]any{"title": "Owner-only conversation"},
	}))

	_, _, err := store.ArchiveIndexedConversation(ctx, IndexedConversationArchiveRequest{
		WorkspaceID:    workspaceID,
		MemberID:       otherMemberID,
		ConversationID: conversationID,
		Event:          StreamEvent{Type: "conversation.archived"},
	})
	require.Error(t, err)
	require.True(t, coredb.IsNotFound(err))

	ownerConversation, err := store.GetConversation(ctx, workspaceID, ownerMemberID, conversationID)
	require.NoError(t, err)
	require.Nil(t, ownerConversation.ArchivedAt)
	otherEvents, err := store.ListEventsAfter(ctx, workspaceID, otherMemberID, 0, 10)
	require.NoError(t, err)
	require.Empty(t, otherEvents)
}

type memberLatestSeqRow struct {
	memberID string
	eventSeq int64
}

type recordingIMTx struct {
	queryCalls          int
	querySQL            []string
	queryArgs           [][]any
	execSQL             []string
	execArgs            [][]any
	latestSeqRows       []memberLatestSeqRow
	refreshedRecipients []scopeEventRecipient
}

func (tx *recordingIMTx) Query(_ context.Context, query string, args ...any) (coredb.Rows, error) {
	tx.queryCalls++
	tx.querySQL = append(tx.querySQL, query)
	tx.queryArgs = append(tx.queryArgs, append([]any(nil), args...))
	if strings.Contains(query, "FROM scopes s") {
		return &scopeEventRecipientRows{rows: tx.refreshedRecipients}, nil
	}
	return &memberLatestSeqRows{rows: tx.latestSeqRows}, nil
}

func (tx *recordingIMTx) QueryRow(context.Context, string, ...any) coredb.Row {
	return unexpectedIMRow{}
}

func (tx *recordingIMTx) Exec(_ context.Context, query string, args ...any) (coredb.Result, error) {
	tx.execSQL = append(tx.execSQL, query)
	tx.execArgs = append(tx.execArgs, append([]any(nil), args...))
	return imResult(0), nil
}

func (tx *recordingIMTx) Commit(context.Context) error {
	return nil
}

func (tx *recordingIMTx) Rollback(context.Context) error {
	return nil
}

type memberLatestSeqRows struct {
	rows []memberLatestSeqRow
	idx  int
}

type scopeEventRecipientRows struct {
	rows []scopeEventRecipient
	idx  int
}

func (rows *scopeEventRecipientRows) Close() {}

func (rows *scopeEventRecipientRows) Err() error {
	return nil
}

func (rows *scopeEventRecipientRows) Next() bool {
	if rows.idx >= len(rows.rows) {
		return false
	}
	rows.idx++
	return true
}

func (rows *scopeEventRecipientRows) Scan(dest ...any) error {
	if rows.idx == 0 || rows.idx > len(rows.rows) {
		return fmt.Errorf("scan called before next")
	}
	if len(dest) != 7 {
		return fmt.Errorf("expected 7 scan destinations, got %d", len(dest))
	}
	row := rows.rows[rows.idx-1]
	*dest[0].(*string) = row.MemberID
	*dest[1].(*string) = row.ScopeKind
	*dest[2].(*string) = row.ParentScopeType
	*dest[3].(*sql.NullTime) = sql.NullTime{Time: row.SortAt, Valid: !row.SortAt.IsZero()}
	if row.PinnedAt != nil {
		*dest[4].(*sql.NullTime) = sql.NullTime{Time: row.PinnedAt.UTC(), Valid: true}
	}
	if row.ArchivedAt != nil {
		*dest[5].(*sql.NullTime) = sql.NullTime{Time: row.ArchivedAt.UTC(), Valid: true}
	}
	*dest[6].(*string) = row.TaskWatchState
	return nil
}

func (rows *memberLatestSeqRows) Close() {}

func (rows *memberLatestSeqRows) Err() error {
	return nil
}

func (rows *memberLatestSeqRows) Next() bool {
	if rows.idx >= len(rows.rows) {
		return false
	}
	rows.idx++
	return true
}

func (rows *memberLatestSeqRows) Scan(dest ...any) error {
	if rows.idx == 0 || rows.idx > len(rows.rows) {
		return fmt.Errorf("scan called before next")
	}
	if len(dest) != 2 {
		return fmt.Errorf("expected 2 scan destinations, got %d", len(dest))
	}
	*dest[0].(*string) = rows.rows[rows.idx-1].memberID
	*dest[1].(*int64) = rows.rows[rows.idx-1].eventSeq
	return nil
}

type unexpectedIMRow struct{}

func (unexpectedIMRow) Scan(...any) error {
	return fmt.Errorf("unexpected query row")
}

type imResult int64

func (result imResult) RowsAffected() int64 {
	return int64(result)
}

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

	ctx := context.Background()
	db := startIMTestPool(t, ctx)
	insertIMWorkspaceMember(t, ctx, db, "wsp_im_workspace_event", "mem_workspace_owner")
	store := New(db)

	event, err := store.AppendMemberWorkspaceEvent(ctx, "wsp_im_workspace_event", "mem_workspace_owner", StreamEvent{
		EventID: "evt_bulk_read",
		Type:    "conversation.bulk.read",
		Payload: map[string]any{"updatedCount": int64(3)},
	})
	require.NoError(t, err)
	require.Equal(t, "mem_workspace_owner", event.MemberID)
	require.Equal(t, int64(1), event.EventSeq)
	require.Empty(t, event.ScopeID)

	events, err := store.ListEventsAfter(ctx, "wsp_im_workspace_event", "mem_workspace_owner", 0, 10)
	require.NoError(t, err)
	require.Len(t, events, 1)
	require.Equal(t, "conversation.bulk.read", events[0].Type)
	require.Equal(t, map[string]any{"updatedCount": float64(3)}, events[0].Payload)
}

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

	ctx := context.Background()
	db := startIMTestPool(t, ctx)
	insertIMWorkspaceMember(t, ctx, db, "wsp_im_bulk_read", "mem_bulk_read")
	store := New(db)
	sortAt := time.Date(2026, 6, 10, 12, 0, 0, 0, time.UTC)

	require.NoError(t, store.UpsertConversation(ctx, "wsp_im_bulk_read", "mem_bulk_read", ConversationIndexItem{
		ConversationID: "channel:chn_bulk",
		ScopeID:        "chn_bulk",
		Kind:           "channel",
		SortAt:         sortAt,
		Projection:     map[string]any{"title": "channel"},
	}))
	require.NoError(t, store.UpsertConversation(ctx, "wsp_im_bulk_read", "mem_bulk_read", ConversationIndexItem{
		ConversationID: "thread:thr_bulk",
		ScopeID:        "thr_bulk",
		Kind:           "thread",
		SortAt:         sortAt.Add(-time.Minute),
		Projection:     map[string]any{"title": "thread"},
	}))

	event, err := store.AppendConversationBulkReadEvent(ctx, "wsp_im_bulk_read", "mem_bulk_read", StreamEvent{
		EventID: "evt_bulk_read_index",
		Type:    "conversation.bulk.read",
		Payload: map[string]any{"updatedCount": int64(1)},
	}, []string{"channel"})
	require.NoError(t, err)
	require.Equal(t, int64(1), event.EventSeq)

	channelConversation, err := store.GetConversation(ctx, "wsp_im_bulk_read", "mem_bulk_read", "channel:chn_bulk")
	require.NoError(t, err)
	require.Equal(t, "channel", channelConversation.Projection["title"])

	threadConversation, err := store.GetConversation(ctx, "wsp_im_bulk_read", "mem_bulk_read", "thread:thr_bulk")
	require.NoError(t, err)
	require.Equal(t, "thread", threadConversation.Projection["title"])
}

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

	ctx := context.Background()
	db := startIMTestPool(t, ctx)
	insertIMWorkspaceMember(t, ctx, db, "wsp_im_strip_unread", "mem_strip_unread")
	store := New(db)

	require.NoError(t, store.UpsertConversation(ctx, "wsp_im_strip_unread", "mem_strip_unread", ConversationIndexItem{
		ConversationID: "channel:chn_strip",
		ScopeID:        "chn_strip",
		Kind:           "channel",
		SortAt:         time.Date(2026, 6, 10, 12, 0, 0, 0, time.UTC),
		Projection: map[string]any{
			"title":        "strip",
			"pinnedAt":     "2026-06-10T11:00:00Z",
			"unreadCount":  9,
			"hasUnread":    true,
			"manualUnread": true,
		},
	}))

	conversation, err := store.GetConversation(ctx, "wsp_im_strip_unread", "mem_strip_unread", "channel:chn_strip")
	require.NoError(t, err)
	require.Equal(t, "strip", conversation.Projection["title"])
	require.NotContains(t, conversation.Projection, "pinnedAt")
	require.NotContains(t, conversation.Projection, "unreadCount")
	require.NotContains(t, conversation.Projection, "hasUnread")
	require.NotContains(t, conversation.Projection, "manualUnread")
}