package unread

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

	"github.com/stretchr/testify/require"

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

func TestTaskSnapshotShadowFencesRollingLegacyWriterAndAdvancesOnlyAtBarrier(t *testing.T) {
	ctx := context.Background()
	dsn := testkit.StartInitializedTiDB(t)
	sqlDB := testkit.OpenTiDB(t, dsn)
	defer sqlDB.Close()
	seedTaskSnapshotShadowIntegrationRows(t, ctx, sqlDB)

	store := New(coredb.WrapSQLDB(sqlDB))
	now := time.Date(2026, 7, 29, 5, 0, 0, 0, time.UTC)

	// An old owner can persist a later independent V1 aggregate before the
	// pending targeted-v2 barrier reaches this row. Because format-0 carries no
	// proof that it contains the v2 event, its frontier/generation/rewind must
	// not permanently fence the first shadow.
	laterLegacyBeforeBarrier := taskSnapshotFixture(now.Add(time.Minute))
	laterLegacyBeforeBarrier.MessageUnreadCount = 4
	laterLegacyBeforeBarrier.LatestMessageAccountedSeq = 31
	laterLegacyBeforeBarrier.ScopeMessageGeneration = 31
	laterLegacyBeforeBarrier.ActivityUnreadCount = 3
	laterLegacyBeforeBarrier.LatestActivityAccountedSeq = 21
	laterLegacyBeforeBarrier.ScopeActivityGeneration = 21
	laterLegacyBeforeBarrier.RewindGeneration = 31
	legacyStoreUpsertUnreadRuntimeSnapshot(t, ctx, sqlDB, laterLegacyBeforeBarrier)

	barrier := taskSnapshotFixture(now)
	barrier.MessageUnreadCount = 3
	barrier.LatestMessageAccountedSeq = 30
	barrier.ScopeMessageGeneration = 30
	barrier.ActivityUnreadCount = 2
	barrier.LatestActivityAccountedSeq = 20
	barrier.ScopeActivityGeneration = 20
	barrier.RewindGeneration = 30
	barrier.ManualUnread = true
	barrier.ManualUnreadGeneration = 8
	barrier.MembershipGeneration = 8
	barrier.NotificationGeneration = 8
	require.NoError(t, store.UpsertTaskBarrierSnapshots(ctx, []unreadservice.Snapshot{barrier}))

	got := loadOneStoredSnapshot(t, ctx, store)
	require.Equal(t, int64(3), got.MessageUnreadCount)
	require.Equal(t, int64(30), got.LatestMessageAccountedSeq)
	require.Equal(t, int64(30), got.ScopeMessageGeneration)
	require.Equal(t, int64(2), got.ActivityUnreadCount)
	require.Equal(t, int64(20), got.LatestActivityAccountedSeq)
	require.Equal(t, int64(20), got.ScopeActivityGeneration)
	require.Equal(t, int64(30), got.RewindGeneration)

	staleLegacy := taskSnapshotFixture(now.Add(time.Minute))
	staleLegacy.MessageUnreadCount = 1
	staleLegacy.LatestMessageAccountedSeq = 5
	staleLegacy.ScopeMessageGeneration = 5
	staleLegacy.MembershipGeneration = 5
	staleLegacy.NotificationGeneration = 5
	legacyStoreUpsertUnreadRuntimeSnapshot(t, ctx, sqlDB, staleLegacy)

	got = loadOneStoredSnapshot(t, ctx, store)
	require.Equal(t, int64(3), got.MessageUnreadCount)
	require.Equal(t, int64(30), got.LatestMessageAccountedSeq)
	require.Equal(t, int64(2), got.ActivityUnreadCount)
	require.Equal(t, int64(20), got.LatestActivityAccountedSeq)
	require.Equal(t, int64(30), got.ScopeMessageGeneration)
	require.Equal(t, int64(20), got.ScopeActivityGeneration)
	require.Equal(t, int64(30), got.RewindGeneration)
	require.True(t, got.ManualUnread)

	newerLegacy := barrier
	newerLegacy.MessageUnreadCount = 4
	newerLegacy.LatestMessageAccountedSeq = 31
	newerLegacy.ScopeMessageGeneration = 31
	newerLegacy.ActivityUnreadCount = 3
	newerLegacy.LatestActivityAccountedSeq = 21
	newerLegacy.ScopeActivityGeneration = 21
	newerLegacy.RewindGeneration = 32
	newerLegacy.ManualUnread = false
	newerLegacy.ManualUnreadGeneration = 9
	newerLegacy.MembershipVisibility = "hidden"
	newerLegacy.MembershipGeneration = 9
	newerLegacy.NotificationState = "muted"
	newerLegacy.NotificationGeneration = 9
	newerLegacy.UpdatedAt = now.Add(2 * time.Minute)
	legacyStoreUpsertUnreadRuntimeSnapshot(t, ctx, sqlDB, newerLegacy)

	got = loadOneStoredSnapshot(t, ctx, store)
	// A larger legacy aggregate cannot prove it contains the protected event:
	// old and new owners may have independently counted disjoint seqs. Keep the
	// protected event domains, but retain monotonic control generations.
	require.Equal(t, int64(3), got.MessageUnreadCount)
	require.Equal(t, int64(30), got.LatestMessageAccountedSeq)
	require.Equal(t, int64(2), got.ActivityUnreadCount)
	require.Equal(t, int64(20), got.LatestActivityAccountedSeq)
	require.Equal(t, int64(30), got.ScopeMessageGeneration)
	require.Equal(t, int64(20), got.ScopeActivityGeneration)
	require.Equal(t, int64(30), got.RewindGeneration)
	require.False(t, got.ManualUnread)
	require.Equal(t, "hidden", got.MembershipVisibility)
	require.Equal(t, "muted", got.NotificationState)
	var protectedShadow string
	require.NoError(t, sqlDB.QueryRowContext(ctx, `
SELECT CAST(task_snapshot_json AS CHAR)
FROM member_scope_unread_runtime_snapshots
WHERE workspace_id = 'w1' AND member_id = 'm1' AND scope_id = 'tsk_1'
`).Scan(&protectedShadow))

	// The next new periodic flush only updates regular columns. Its full-cover
	// read cursor is safe for a new reader to overlay, while its unprotected
	// Activity aggregate remains ordinary V1 best-effort. An old reader can
	// still consume the regular columns after an application rollback.
	periodic := barrier
	periodic.MessageUnreadCount = 0
	periodic.LastMessageReadSeq = 31
	periodic.LatestMessageAccountedSeq = 31
	periodic.ScopeMessageGeneration = 31
	periodic.ActivityUnreadCount = 4
	periodic.LatestActivityAccountedSeq = 22
	periodic.ScopeActivityGeneration = 22
	periodic.ManualUnread = false
	periodic.ManualUnreadGeneration = 10
	periodic.MembershipVisibility = "hidden"
	periodic.MembershipGeneration = 10
	periodic.NotificationState = "muted"
	periodic.NotificationGeneration = 10
	periodic.UpdatedAt = now.Add(3 * time.Minute)
	require.NoError(t, store.UpsertSnapshots(ctx, []unreadservice.Snapshot{periodic}))

	var legacyMessageCount, legacyMessageFrontier, legacyReadCursor int64
	var legacyActivityCount, legacyActivityFrontier int64
	var shadowVersion int64
	var shadowPayload sql.NullString
	require.NoError(t, sqlDB.QueryRowContext(ctx, `
SELECT
  message_unread_count,
  latest_message_accounted_seq,
  last_message_read_seq,
  activity_unread_count,
  latest_activity_accounted_seq,
  task_snapshot_format_version,
  task_snapshot_json
FROM member_scope_unread_runtime_snapshots
WHERE workspace_id = 'w1' AND member_id = 'm1' AND scope_id = 'tsk_1'
`).Scan(
		&legacyMessageCount,
		&legacyMessageFrontier,
		&legacyReadCursor,
		&legacyActivityCount,
		&legacyActivityFrontier,
		&shadowVersion,
		&shadowPayload,
	))
	require.Zero(t, legacyMessageCount)
	require.Equal(t, int64(31), legacyMessageFrontier)
	require.Equal(t, int64(31), legacyReadCursor)
	require.Equal(t, int64(4), legacyActivityCount)
	require.Equal(t, int64(22), legacyActivityFrontier)
	require.Equal(t, taskSnapshotShadowFormatV1, shadowVersion)
	require.True(t, shadowPayload.Valid)
	require.JSONEq(t, protectedShadow, shadowPayload.String, "periodic flush must not replace protected aggregate")

	got = loadOneStoredSnapshot(t, ctx, store)
	require.Zero(t, got.MessageUnreadCount)
	require.Equal(t, int64(31), got.LastMessageReadSeq)
	require.Equal(t, int64(2), got.ActivityUnreadCount)
	require.Equal(t, int64(20), got.LatestActivityAccountedSeq)

	// A later completion barrier is the only writer allowed to advance the
	// protected event aggregate. This candidate contains the prior protected
	// state plus the new Activity frontier.
	require.NoError(t, store.UpsertTaskBarrierSnapshots(ctx, []unreadservice.Snapshot{periodic}))
	got = loadOneStoredSnapshot(t, ctx, store)
	require.Zero(t, got.MessageUnreadCount)
	require.Equal(t, int64(4), got.ActivityUnreadCount)
	require.Equal(t, int64(22), got.LatestActivityAccountedSeq)

	// Task identity is canonical metadata, not an ID-prefix contract. Persist a
	// barrier for a custom non-"tsk_" scope and hydrate it through a fresh
	// store/DB connection, as a restarted owner would.
	customBarrier := taskSnapshotFixture(now.Add(4 * time.Minute))
	customBarrier.ScopeID = "custom_task_scope"
	customBarrier.MessageUnreadCount = 2
	customBarrier.LatestMessageAccountedSeq = 40
	customBarrier.ScopeMessageGeneration = 40
	customBarrier.ActivityUnreadCount = 1
	customBarrier.LatestActivityAccountedSeq = 25
	customBarrier.ScopeActivityGeneration = 25
	customBarrier.RewindGeneration = 40
	require.NoError(t, store.UpsertTaskBarrierSnapshots(ctx, []unreadservice.Snapshot{customBarrier}))

	restartedSQLDB := testkit.OpenTiDB(t, dsn)
	defer restartedSQLDB.Close()
	restartedStore := New(coredb.WrapSQLDB(restartedSQLDB))
	restarted, err := restartedStore.LoadSnapshotPage(ctx, SnapshotPageRequest{
		WorkspaceID: "w1",
		MemberID:    "m1",
		ScopeIDs:    []string{"custom_task_scope"},
		Limit:       1,
	})
	require.NoError(t, err)
	require.Len(t, restarted, 1)
	require.Equal(t, unreadservice.ScopeTypeTask, restarted[0].ScopeType)
	require.Equal(t, int64(2), restarted[0].MessageUnreadCount)
	require.Equal(t, int64(40), restarted[0].LatestMessageAccountedSeq)
	require.Equal(t, int64(1), restarted[0].ActivityUnreadCount)
	require.Equal(t, int64(25), restarted[0].LatestActivityAccountedSeq)
}

func seedTaskSnapshotShadowIntegrationRows(t *testing.T, ctx context.Context, db *sql.DB) {
	t.Helper()
	now := time.Date(2026, 7, 29, 4, 59, 0, 0, time.UTC)
	_, err := db.ExecContext(ctx, `
INSERT INTO users (
  id, email, name, display_name, password_hash, created_at, updated_at
) VALUES (
  'u1', 'task-shadow@example.com', 'task_shadow', 'Task Shadow', 'hash', ?, ?
)
`, now, now)
	require.NoError(t, err)
	_, err = db.ExecContext(ctx, `
INSERT INTO organizations (
  id, slug, name, created_by_user_id, created_at, updated_at
) VALUES (
  'org1', 'task-shadow', 'Task Shadow', 'u1', ?, ?
)
`, now, now)
	require.NoError(t, err)
	_, err = db.ExecContext(ctx, `
INSERT INTO workspaces (
  id, organization_id, slug, name, created_at, updated_at
) VALUES (
  'w1', 'org1', 'task-shadow', 'Task Shadow', ?, ?
)
`, now, now)
	require.NoError(t, err)
	_, err = db.ExecContext(ctx, `
INSERT INTO members (
  id, workspace_id, type, display_name, user_id, workspace_role,
  status, created_at, updated_at
) VALUES (
  'm1', 'w1', 'human', 'Task Shadow', 'u1', 'admin',
  'active', ?, ?
)
`, now, now)
	require.NoError(t, err)
	_, err = db.ExecContext(ctx, `
INSERT INTO scopes (
  id, workspace_id, type, status, created_by_member_id, created_at, updated_at
) VALUES (
  'tsk_1', 'w1', 'task', 'active', 'm1', ?, ?
)
`, now, now)
	require.NoError(t, err)
	_, err = db.ExecContext(ctx, `
INSERT INTO scopes (
  id, workspace_id, type, status, created_by_member_id, created_at, updated_at
) VALUES (
  'custom_task_scope', 'w1', 'task', 'active', 'm1', ?, ?
)
`, now, now)
	require.NoError(t, err)
}

func legacyStoreUpsertUnreadRuntimeSnapshot(
	t *testing.T,
	ctx context.Context,
	db *sql.DB,
	snapshot unreadservice.Snapshot,
) {
	t.Helper()
	_, err := db.ExecContext(ctx, `
INSERT INTO member_scope_unread_runtime_snapshots (
  workspace_id, member_id, scope_id,
  message_unread_count, activity_unread_count, manual_unread,
  last_message_read_seq, last_activity_read_seq, effective_from_seq,
  latest_message_accounted_seq, latest_activity_accounted_seq,
  notification_state, membership_visibility,
  scope_message_generation, scope_activity_generation,
  membership_generation, notification_generation, manual_unread_generation, rewind_generation,
  updated_at
) VALUES (
  ?, ?, ?,
  ?, ?, ?,
  ?, ?, ?,
  ?, ?,
  ?, ?,
  ?, ?,
  ?, ?, ?, ?,
  ?
)
ON DUPLICATE KEY UPDATE
  message_unread_count = VALUES(message_unread_count),
  activity_unread_count = VALUES(activity_unread_count),
  manual_unread = VALUES(manual_unread),
  last_message_read_seq = VALUES(last_message_read_seq),
  last_activity_read_seq = VALUES(last_activity_read_seq),
  effective_from_seq = VALUES(effective_from_seq),
  latest_message_accounted_seq = VALUES(latest_message_accounted_seq),
  latest_activity_accounted_seq = VALUES(latest_activity_accounted_seq),
  notification_state = VALUES(notification_state),
  membership_visibility = VALUES(membership_visibility),
  scope_message_generation = VALUES(scope_message_generation),
  scope_activity_generation = VALUES(scope_activity_generation),
  membership_generation = VALUES(membership_generation),
  notification_generation = VALUES(notification_generation),
  manual_unread_generation = VALUES(manual_unread_generation),
  rewind_generation = VALUES(rewind_generation),
  updated_at = VALUES(updated_at)
`, snapshot.WorkspaceID, snapshot.MemberID, snapshot.ScopeID,
		snapshot.MessageUnreadCount, snapshot.ActivityUnreadCount, snapshot.ManualUnread,
		snapshot.LastMessageReadSeq, snapshot.LastActivityReadSeq, snapshot.EffectiveFromSeq,
		snapshot.LatestMessageAccountedSeq, snapshot.LatestActivityAccountedSeq,
		snapshot.NotificationState, snapshot.MembershipVisibility,
		snapshot.ScopeMessageGeneration, snapshot.ScopeActivityGeneration,
		snapshot.MembershipGeneration, snapshot.NotificationGeneration, snapshot.ManualUnreadGeneration, snapshot.RewindGeneration,
		snapshot.UpdatedAt)
	require.NoError(t, err)
}

func loadOneStoredSnapshot(t *testing.T, ctx context.Context, store *Store) unreadservice.Snapshot {
	t.Helper()
	rows, err := store.LoadSnapshotPage(ctx, SnapshotPageRequest{
		WorkspaceID: "w1",
		MemberID:    "m1",
		ScopeIDs:    []string{"tsk_1"},
		Limit:       1,
	})
	require.NoError(t, err)
	require.Len(t, rows, 1)
	return rows[0]
}