package message

import (
	"context"
	"sync/atomic"
	"testing"
	"time"

	"github.com/stretchr/testify/require"

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

// These tests intentionally model the lock order of the Task delete path from
// before the lifecycle guard was introduced:
//
//   - delete locks the selected Task row first;
//   - parent delete reads children in task-number order;
//   - child scopes are tombstoned before the child Task rows are deleted;
//   - the selected Task scope and Task row are handled last.
//
// Keeping that older order in the test matters for rolling deploys: a new
// fanout guard must serialize correctly with a transaction from an older
// backend, not merely with another copy of the new guard.
func TestTaskScopeLifecycleGuardSerializesLegacyHierarchyDelete(t *testing.T) {
	setupCtx, setupCancel := context.WithTimeout(t.Context(), 30*time.Second)
	defer setupCancel()

	dsn := testkit.StartInitializedTiDBWithRootDSN(t, messagingTiDB.RootDSN(t))
	sqlDB, err := testkit.OpenTiDBContext(setupCtx, dsn)
	require.NoError(t, err)
	t.Cleanup(func() { _ = sqlDB.Close() })
	sqlDB.SetMaxOpenConns(8)
	sqlDB.SetMaxIdleConns(8)
	db := coredb.WrapSQLDB(sqlDB)

	t.Run("parent guard wins", func(t *testing.T) {
		ctx, cancel := context.WithTimeout(t.Context(), 10*time.Second)
		defer cancel()

		fixture := seedTaskScopeLifecycleHierarchy(t, ctx, db, "parent_guard_first")
		store := New(db, nil, 0, Options{})

		callbackStarted := make(chan struct{})
		releaseCallback := make(chan struct{})
		guardDone := make(chan hierarchyLifecycleGuardResult, 1)
		var callbackCount atomic.Int64
		go func() {
			active, guardErr := store.WithActiveTaskScopes(
				ctx,
				fixture.workspaceID,
				fixture.allScopeIDs(),
				func() error {
					close(callbackStarted)
					select {
					case <-releaseCallback:
					case <-ctx.Done():
						return ctx.Err()
					}
					callbackCount.Add(1)
					return nil
				},
			)
			guardDone <- hierarchyLifecycleGuardResult{active: active, err: guardErr}
		}()
		requireHierarchySignal(t, ctx, callbackStarted, "hierarchy guard callback did not start")

		deleteAttempted := make(chan struct{})
		deleteLocked := make(chan struct{})
		deleteDone := make(chan legacyHierarchyDeleteResult, 1)
		go func() {
			tombstoned, deleteErr := runLegacyHierarchyDelete(
				ctx,
				db,
				fixture.parent.taskID,
				deleteAttempted,
				deleteLocked,
				nil,
			)
			deleteDone <- legacyHierarchyDeleteResult{tombstonedScopeIDs: tombstoned, err: deleteErr}
		}()
		requireHierarchySignal(t, ctx, deleteAttempted, "legacy parent delete was not attempted")

		requireHierarchyBlocked(t, deleteLocked, deleteDone, "legacy parent delete acquired the parent Task lock while guard effects were active")

		close(releaseCallback)
		guardResult := requireHierarchyGuardResult(t, ctx, guardDone)
		require.NoError(t, guardResult.err)
		require.True(t, guardResult.active)
		require.Equal(t, int64(1), callbackCount.Load())

		deleteResult := requireLegacyHierarchyDeleteResult(t, ctx, deleteDone)
		require.NoError(t, deleteResult.err)
		require.Equal(t, []string{
			fixture.children[0].scopeID,
			fixture.children[1].scopeID,
			fixture.parent.scopeID,
		}, deleteResult.tombstonedScopeIDs)
		requireHierarchyDeleted(t, ctx, db, fixture)
	})

	t.Run("parent delete wins", func(t *testing.T) {
		ctx, cancel := context.WithTimeout(t.Context(), 10*time.Second)
		defer cancel()

		fixture := seedTaskScopeLifecycleHierarchy(t, ctx, db, "parent_delete_first")
		store := New(db, nil, 0, Options{})

		deleteAttempted := make(chan struct{})
		deleteLocked := make(chan struct{})
		releaseDelete := make(chan struct{})
		deleteDone := make(chan legacyHierarchyDeleteResult, 1)
		go func() {
			tombstoned, deleteErr := runLegacyHierarchyDelete(
				ctx,
				db,
				fixture.parent.taskID,
				deleteAttempted,
				deleteLocked,
				releaseDelete,
			)
			deleteDone <- legacyHierarchyDeleteResult{tombstonedScopeIDs: tombstoned, err: deleteErr}
		}()
		requireHierarchySignal(t, ctx, deleteAttempted, "legacy parent delete was not attempted")
		requireHierarchySignal(t, ctx, deleteLocked, "legacy parent delete did not acquire the parent Task lock")

		callbackStarted := make(chan struct{})
		guardDone := make(chan hierarchyLifecycleGuardResult, 1)
		var callbackCount atomic.Int64
		go func() {
			active, guardErr := store.WithActiveTaskScopes(
				ctx,
				fixture.workspaceID,
				fixture.allScopeIDs(),
				func() error {
					callbackCount.Add(1)
					close(callbackStarted)
					return nil
				},
			)
			guardDone <- hierarchyLifecycleGuardResult{active: active, err: guardErr}
		}()

		requireHierarchyGuardBlocked(t, callbackStarted, guardDone, "hierarchy guard passed a parent delete that already held the parent Task lock")
		close(releaseDelete)

		deleteResult := requireLegacyHierarchyDeleteResult(t, ctx, deleteDone)
		require.NoError(t, deleteResult.err)
		require.Equal(t, []string{
			fixture.children[0].scopeID,
			fixture.children[1].scopeID,
			fixture.parent.scopeID,
		}, deleteResult.tombstonedScopeIDs)

		guardResult := requireHierarchyGuardResult(t, ctx, guardDone)
		require.NoError(t, guardResult.err)
		require.False(t, guardResult.active)
		require.Zero(t, callbackCount.Load())
		requireHierarchyDeleted(t, ctx, db, fixture)
	})

	t.Run("direct child guard wins", func(t *testing.T) {
		ctx, cancel := context.WithTimeout(t.Context(), 10*time.Second)
		defer cancel()

		fixture := seedTaskScopeLifecycleHierarchy(t, ctx, db, "child_guard_first")
		store := New(db, nil, 0, Options{})
		child := fixture.children[0]

		callbackStarted := make(chan struct{})
		releaseCallback := make(chan struct{})
		guardDone := make(chan hierarchyLifecycleGuardResult, 1)
		var callbackCount atomic.Int64
		go func() {
			active, guardErr := store.WithActiveTaskScopes(
				ctx,
				fixture.workspaceID,
				[]string{child.scopeID},
				func() error {
					close(callbackStarted)
					select {
					case <-releaseCallback:
					case <-ctx.Done():
						return ctx.Err()
					}
					callbackCount.Add(1)
					return nil
				},
			)
			guardDone <- hierarchyLifecycleGuardResult{active: active, err: guardErr}
		}()
		requireHierarchySignal(t, ctx, callbackStarted, "child guard callback did not start")

		deleteAttempted := make(chan struct{})
		deleteLocked := make(chan struct{})
		deleteDone := make(chan legacyHierarchyDeleteResult, 1)
		go func() {
			tombstoned, deleteErr := runLegacyHierarchyDelete(
				ctx,
				db,
				child.taskID,
				deleteAttempted,
				deleteLocked,
				nil,
			)
			deleteDone <- legacyHierarchyDeleteResult{tombstonedScopeIDs: tombstoned, err: deleteErr}
		}()
		requireHierarchySignal(t, ctx, deleteAttempted, "legacy child delete was not attempted")
		requireHierarchyBlocked(t, deleteLocked, deleteDone, "legacy child delete acquired the child Task lock while guard effects were active")

		close(releaseCallback)
		guardResult := requireHierarchyGuardResult(t, ctx, guardDone)
		require.NoError(t, guardResult.err)
		require.True(t, guardResult.active)
		require.Equal(t, int64(1), callbackCount.Load())

		deleteResult := requireLegacyHierarchyDeleteResult(t, ctx, deleteDone)
		require.NoError(t, deleteResult.err)
		require.Equal(t, []string{child.scopeID}, deleteResult.tombstonedScopeIDs)
		requireDirectChildDeleteState(t, ctx, db, fixture, child)
	})

	t.Run("direct child delete wins", func(t *testing.T) {
		ctx, cancel := context.WithTimeout(t.Context(), 10*time.Second)
		defer cancel()

		fixture := seedTaskScopeLifecycleHierarchy(t, ctx, db, "child_delete_first")
		store := New(db, nil, 0, Options{})
		child := fixture.children[0]

		deleteAttempted := make(chan struct{})
		deleteLocked := make(chan struct{})
		releaseDelete := make(chan struct{})
		deleteDone := make(chan legacyHierarchyDeleteResult, 1)
		go func() {
			tombstoned, deleteErr := runLegacyHierarchyDelete(
				ctx,
				db,
				child.taskID,
				deleteAttempted,
				deleteLocked,
				releaseDelete,
			)
			deleteDone <- legacyHierarchyDeleteResult{tombstonedScopeIDs: tombstoned, err: deleteErr}
		}()
		requireHierarchySignal(t, ctx, deleteAttempted, "legacy child delete was not attempted")
		requireHierarchySignal(t, ctx, deleteLocked, "legacy child delete did not acquire the child Task lock")

		callbackStarted := make(chan struct{})
		guardDone := make(chan hierarchyLifecycleGuardResult, 1)
		var callbackCount atomic.Int64
		go func() {
			active, guardErr := store.WithActiveTaskScopes(
				ctx,
				fixture.workspaceID,
				[]string{child.scopeID},
				func() error {
					callbackCount.Add(1)
					close(callbackStarted)
					return nil
				},
			)
			guardDone <- hierarchyLifecycleGuardResult{active: active, err: guardErr}
		}()

		requireHierarchyGuardBlocked(t, callbackStarted, guardDone, "child guard passed a delete that already held the child Task lock")
		close(releaseDelete)

		deleteResult := requireLegacyHierarchyDeleteResult(t, ctx, deleteDone)
		require.NoError(t, deleteResult.err)
		require.Equal(t, []string{child.scopeID}, deleteResult.tombstonedScopeIDs)

		guardResult := requireHierarchyGuardResult(t, ctx, guardDone)
		require.NoError(t, guardResult.err)
		require.False(t, guardResult.active)
		require.Zero(t, callbackCount.Load())
		requireDirectChildDeleteState(t, ctx, db, fixture, child)
	})

	t.Run("child target retries after parent promote", func(t *testing.T) {
		ctx, cancel := context.WithTimeout(t.Context(), 10*time.Second)
		defer cancel()

		fixture := seedTaskScopeLifecycleHierarchy(t, ctx, db, "child_parent_promote")
		child := fixture.children[0]

		promoteAttempted := make(chan struct{})
		promoteLocked := make(chan struct{})
		releasePromote := make(chan struct{})
		promoteDone := make(chan error, 1)
		go func() {
			promoteDone <- runLegacyHierarchyPromote(
				ctx,
				db,
				fixture.parent.taskID,
				promoteAttempted,
				promoteLocked,
				releasePromote,
			)
		}()
		requireHierarchySignal(t, ctx, promoteAttempted, "legacy parent promote was not attempted")
		requireHierarchySignal(t, ctx, promoteLocked, "legacy parent promote did not acquire the parent Task lock")

		// Observe the guard's first topology read so the interleaving does not
		// depend on a sleep: the child was read with parent_task_ref=R while
		// the legacy transaction held R but had not promoted the child yet.
		topologyRead := make(chan struct{})
		guardDB := newHierarchyLifecycleTopologyObservedDB(db, topologyRead)
		store := New(guardDB, nil, 0, Options{})
		callbackStarted := make(chan struct{})
		guardDone := make(chan hierarchyLifecycleGuardResult, 1)
		var callbackCount atomic.Int64
		go func() {
			active, guardErr := store.WithActiveTaskScopes(
				ctx,
				fixture.workspaceID,
				[]string{child.scopeID},
				func() error {
					callbackCount.Add(1)
					close(callbackStarted)
					return nil
				},
			)
			guardDone <- hierarchyLifecycleGuardResult{active: active, err: guardErr}
		}()
		requireHierarchySignal(t, ctx, topologyRead, "child guard did not finish its pre-promote topology read")
		requireHierarchyGuardBlocked(t, callbackStarted, guardDone, "child guard passed a promote that held its hinted parent Task lock")

		close(releasePromote)
		require.NoError(t, requireHierarchyErrorResult(t, ctx, promoteDone, "legacy parent promote"))

		// R disappeared, but the requested child survived as a new top-level
		// Task. The guard must rollback the stale topology attempt, re-read,
		// lock the child as a root, and execute its callback exactly once.
		guardResult := requireHierarchyGuardResult(t, ctx, guardDone)
		require.NoError(t, guardResult.err)
		require.True(t, guardResult.active)
		require.Equal(t, int64(1), callbackCount.Load())
		requirePromotedHierarchyState(t, ctx, db, fixture)
	})

	t.Run("child target stays inactive after parent delete", func(t *testing.T) {
		ctx, cancel := context.WithTimeout(t.Context(), 10*time.Second)
		defer cancel()

		fixture := seedTaskScopeLifecycleHierarchy(t, ctx, db, "child_parent_delete")
		child := fixture.children[0]

		deleteAttempted := make(chan struct{})
		deleteLocked := make(chan struct{})
		releaseDelete := make(chan struct{})
		deleteDone := make(chan legacyHierarchyDeleteResult, 1)
		go func() {
			tombstoned, deleteErr := runLegacyHierarchyDelete(
				ctx,
				db,
				fixture.parent.taskID,
				deleteAttempted,
				deleteLocked,
				releaseDelete,
			)
			deleteDone <- legacyHierarchyDeleteResult{tombstonedScopeIDs: tombstoned, err: deleteErr}
		}()
		requireHierarchySignal(t, ctx, deleteAttempted, "legacy parent delete was not attempted")
		requireHierarchySignal(t, ctx, deleteLocked, "legacy parent delete did not acquire the parent Task lock")

		topologyRead := make(chan struct{})
		guardDB := newHierarchyLifecycleTopologyObservedDB(db, topologyRead)
		store := New(guardDB, nil, 0, Options{})
		callbackStarted := make(chan struct{})
		guardDone := make(chan hierarchyLifecycleGuardResult, 1)
		var callbackCount atomic.Int64
		go func() {
			active, guardErr := store.WithActiveTaskScopes(
				ctx,
				fixture.workspaceID,
				[]string{child.scopeID},
				func() error {
					callbackCount.Add(1)
					close(callbackStarted)
					return nil
				},
			)
			guardDone <- hierarchyLifecycleGuardResult{active: active, err: guardErr}
		}()
		requireHierarchySignal(t, ctx, topologyRead, "child guard did not finish its pre-delete topology read")
		requireHierarchyGuardBlocked(t, callbackStarted, guardDone, "child guard passed a delete that held its hinted parent Task lock")

		close(releaseDelete)
		deleteResult := requireLegacyHierarchyDeleteResult(t, ctx, deleteDone)
		require.NoError(t, deleteResult.err)
		require.Equal(t, []string{
			fixture.children[0].scopeID,
			fixture.children[1].scopeID,
			fixture.parent.scopeID,
		}, deleteResult.tombstonedScopeIDs)

		// Unlike promote, delete removed the requested child. A retry must
		// therefore finish inactive and must not execute the callback.
		guardResult := requireHierarchyGuardResult(t, ctx, guardDone)
		require.NoError(t, guardResult.err)
		require.False(t, guardResult.active)
		require.Zero(t, callbackCount.Load())
		requireHierarchyDeleted(t, ctx, db, fixture)
	})
}