package extensionflow

import (
	"context"
	"fmt"
	"strings"
	"sync"
	"testing"
	"time"

	"github.com/stretchr/testify/assert"
	"github.com/stretchr/testify/require"

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

// convertGitHubToolsCapabilityToExplicitOnly rewrites the seeded GitHub
// mcp.tools capability into the modern explicit-only execution contract with
// the given durable tool allowlist, and grants agent-1 explicitly.
func convertGitHubToolsCapabilityToExplicitOnly(
	t *testing.T,
	db coredb.QueryExecer,
	fixture serviceFixture,
	installationID string,
	tools []string,
) {
	t.Helper()
	now := time.Now().UTC()
	quoted := make([]string, 0, len(tools))
	for _, tool := range tools {
		quoted = append(quoted, "'"+tool+"'")
	}
	mustExecService(t, db, `
UPDATE service_capabilities
SET config=JSON_OBJECT(
      'tools', JSON_ARRAY(`+strings.Join(quoted, ",")+`),
      'scopes', JSON_ARRAY('provider.watches.read', 'provider.watches.write'),
      'authorizationMode', 'explicit_only',
      'installMode', 'feature_owned',
      'requiresSubscriptions', FALSE
    ), updated_at=$1
WHERE service_definition_id='app_github' AND capability_key='mcp.tools'`,
		now)
	mustExecService(t, db, `
INSERT INTO service_principal_authorizations (
    id, workspace_id, service_definition_id, service_installation_id,
    service_kind, principal_kind, principal_ref, status, disabled_reason,
    scopes_json, authorized_by_member_id, policy_version, metadata,
    created_at, updated_at, active_authorization_key
) VALUES (
    'authz-app-authority', $1, 'app_github', $2,
    'app', 'agent', 'agent-1', 'active', '',
    '["provider.watches.read","provider.watches.write"]', $3, 1, '{}',
    $4, $4, 'key-app-authority'
)`, fixture.WorkspaceID, installationID, fixture.HumanMemberID, now)
}

// TestInvokeFailsClosedWhenDurableToolRevoked proves the durable mcp.tools
// allowlist is enforced by the live invocation resolver, not only the Catalog
// projection: removing a tool from the App's execution capability removes the
// action for invocations under the same explicit grant and scopes, while the
// retained tools keep working.
func TestInvokeFailsClosedWhenDurableToolRevoked(t *testing.T) {
	ctx := context.Background()
	db := startServiceRollbackDB(t, ctx)
	fixture := seedServiceFixture(t, ctx, db, "apptools-invoke")
	require.NoError(t, registerProviderWatchBuiltinsForTest(fixture.Registry))
	service := newService(db, fixture.Registry)
	installationID, _ := seedProductAppConnection(t, db, fixture, "toolsinvoke")
	convertGitHubToolsCapabilityToExplicitOnly(
		t,
		db,
		fixture,
		installationID,
		[]string{
			"list-watches",
			"pause-watch",
			"resume-watch",
			"unwatch",
		},
	)
	require.NoError(t, coredb.WithTx(ctx, db, func(tx coredb.Tx) error {
		return service.ApplyAgentCatalogChange(
			ctx,
			tx,
			fixture.WorkspaceID,
			"agent-1",
			time.Now().UTC(),
		)
	}))
	registered, ok := fixture.Registry.LookupByIdentity("github", 1)
	require.True(t, ok)
	version := readAgentCatalogVersion(
		t,
		ctx,
		db,
		fixture.WorkspaceID,
		"agent-1",
	)

	// A retained tool still executes through the explicit grant.
	result, err := service.Invoke(ctx, invocationParams(
		fixture,
		"agent-1",
		"github",
		"list-watches",
		registered.Definition.Digest,
		version,
	))
	require.NoError(t, err)
	require.JSONEq(t, `{"ok":true}`, string(result.Data))

	// The revoked tool is gone from the live allowlist even though the
	// scopes still cover it: fail closed, not scope-only.
	_, err = service.Invoke(ctx, invocationParams(
		fixture,
		"agent-1",
		"github",
		"watch-pull-request",
		registered.Definition.Digest,
		version,
	))
	requireInvocationCode(t, err, InvocationNotFound)
}

// TestInvokeUnderAuthorizationChurnNeverDeadlocks is the concurrency
// regression for the snapshot-read design: the invocation path takes NO
// authorization row locks (the catalog row is its single lock point, taken
// before the membership assertion whose agents-row lock the catalog-writer
// event INSERT needs), so a concurrent revoke/restore churn against a hot
// invocation agentteam must never deadlock or surface an internal DB abort;
// every invocation outcome is success or a classified stale/not-available
// error.
func TestInvokeUnderAuthorizationChurnNeverDeadlocks(t *testing.T) {
	ctx := context.Background()
	db := startServiceConcurrentDB(t, ctx)
	fixture := seedServiceFixture(t, ctx, db, "lockorder")
	require.NoError(t, registerProviderWatchBuiltinsForTest(fixture.Registry))
	service := newService(db, fixture.Registry)
	service.invocationAuditTimeout = 30 * time.Second
	installationID, _ := seedProductAppConnection(t, db, fixture, "lockorder")
	require.NoError(t, coredb.WithTx(ctx, db, func(tx coredb.Tx) error {
		return service.ApplyAgentCatalogChange(
			ctx,
			tx,
			fixture.WorkspaceID,
			"agent-1",
			time.Now().UTC(),
		)
	}))
	registered, ok := fixture.Registry.LookupByIdentity("github", 1)
	require.True(t, ok)

	const churns = 24
	var wg sync.WaitGroup
	invokeFailures := make(chan string, churns*4)
	churnErr := make(chan error, 1)
	wg.Add(2)
	go func() {
		defer wg.Done()
		for iteration := 0; iteration < churns*4; iteration++ {
			var version int64
			assert.NoError(t, db.QueryRow(ctx, `
SELECT version FROM agent_extension_catalogs
WHERE workspace_id=$1 AND agent_id=$2`,
				fixture.WorkspaceID,
				"agent-1",
			).Scan(&version))
			_, err := service.Invoke(ctx, invocationParams(
				fixture,
				"agent-1",
				"github",
				"list-watches",
				registered.Definition.Digest,
				version,
			))
			if err == nil {
				continue
			}
			// Every non-success outcome must carry a CLASSIFIED code in
			// the linearizable set: stale (the churn raced ahead of the
			// catalog pin) or not-available (the churn dropped the grant).
			// Deadlocks, internal errors, context deadlines, and raw SQL
			// aborts have no code and fail the regression immediately.
			code, classified := InvocationErrorCodeOf(err)
			switch {
			case classified &&
				(code == InvocationAgentCatalogStale ||
					code == InvocationNotFound ||
					code == InvocationProviderAuthUnavailable):
			default:
				invokeFailures <- fmt.Sprintf(
					"%v (cause: %v)",
					err,
					InvocationErrorCause(err),
				)
			}
		}
	}()
	go func() {
		defer wg.Done()
		var churn error
		for iteration := 0; iteration < churns && churn == nil; iteration++ {
			churn = coredb.WithTx(ctx, db, func(tx coredb.Tx) error {
				now := time.Now().UTC()
				if iteration == 0 {
					_, err := tx.Exec(ctx, `
INSERT INTO service_principal_authorizations (
    id, workspace_id, service_definition_id, service_installation_id,
    service_kind, principal_kind, principal_ref, status, disabled_reason,
    scopes_json, authorized_by_member_id, policy_version, metadata,
    created_at, updated_at, disabled_at, active_authorization_key
) VALUES (
    'authz-lockorder', $1, 'app_github', $2,
    'app', 'agent', 'agent-1', 'disabled', 'churn',
    '[]', $3, 1, '{}', $4, $4, $4, NULL
)`, fixture.WorkspaceID, installationID, fixture.HumanMemberID, now)
					if err != nil {
						return err
					}
				} else {
					_, err := tx.Exec(ctx, `
UPDATE service_principal_authorizations
SET status = IF(status='active', 'disabled', 'active'),
    disabled_reason = IF(status='active', 'churn', ''),
    disabled_at = IF(status='active', NOW(), NULL),
    updated_at=$1
WHERE id='authz-lockorder'`, now)
					if err != nil {
						return err
					}
				}
				_, err := service.applyAgentCatalogChange(
					ctx,
					tx,
					fixture.WorkspaceID,
					"agent-1",
					now,
				)
				return err
			})
		}
		churnErr <- churn
	}()
	wg.Wait()
	close(invokeFailures)
	require.NoError(t, <-churnErr)
	for failure := range invokeFailures {
		t.Fatalf("invocation acked an unclassified outcome under concurrent authorization churn: %s", failure)
	}
}

// TestEmptyToolsAllowlistRevokesEveryAction proves an explicit-only App whose
// durable mcp.tools list is emptied (scopes untouched) advertises and
// executes nothing: an empty allowlist is a full revocation, never a
// fallback to the scope-derived action set.
func TestEmptyToolsAllowlistRevokesEveryAction(t *testing.T) {
	ctx := context.Background()
	db := startServiceRollbackDB(t, ctx)
	fixture := seedServiceFixture(t, ctx, db, "apptools-empty")
	require.NoError(t, registerProviderWatchBuiltinsForTest(fixture.Registry))
	service := newService(db, fixture.Registry)
	installationID, _ := seedProductAppConnection(t, db, fixture, "toolsempty")
	convertGitHubToolsCapabilityToExplicitOnly(t, db, fixture, installationID, nil)
	require.NoError(t, coredb.WithTx(ctx, db, func(tx coredb.Tx) error {
		return service.ApplyAgentCatalogChange(
			ctx, tx, fixture.WorkspaceID, "agent-1", time.Now().UTC())
	}))

	entries, err := service.AgentCatalogEntries(ctx, fixture.WorkspaceID, "agent-1")
	require.NoError(t, err)
	for _, entry := range entries {
		require.NotEqual(t, "github", entry.Domain,
			"an emptied tools allowlist must revoke the whole App surface")
	}

	registered, ok := fixture.Registry.LookupByIdentity("github", 1)
	require.True(t, ok)
	// The emptied surface stays sparse: no durable catalog row exists for
	// the agent, and an invocation pinned to any version fails closed.
	var catalogRows int
	require.NoError(t, db.QueryRow(ctx, `
SELECT COUNT(*) FROM agent_extension_catalogs
WHERE workspace_id=$1 AND agent_id='agent-1'`,
		fixture.WorkspaceID).Scan(&catalogRows))
	require.Zero(t, catalogRows)
	_, err = service.Invoke(ctx, invocationParams(
		fixture, "agent-1", "github", "list-watches",
		registered.Definition.Digest, 0))
	require.Error(t, err, "invocation under an emptied allowlist must fail")
}

// TestAuthorizationDriftIsFencedByCatalogState is the deterministic
// linearization proof for the snapshot-read design: when the durable
// authorization state drifts from the durable catalog (a lifecycle update
// that never landed its catalog mutation), the recomputed state no longer
// matches and the invocation fails closed as stale — a drifted snapshot can
// never keep executing behind the durable catalog.
func TestAuthorizationDriftIsFencedByCatalogState(t *testing.T) {
	ctx := context.Background()
	db := startServiceRollbackDB(t, ctx)
	fixture := seedServiceFixture(t, ctx, db, "driftfence")
	require.NoError(t, registerProviderWatchBuiltinsForTest(fixture.Registry))
	service := newService(db, fixture.Registry)
	installationID, _ := seedProductAppConnection(t, db, fixture, "driftfence")
	require.NoError(t, coredb.WithTx(ctx, db, func(tx coredb.Tx) error {
		return service.ApplyAgentCatalogChange(
			ctx, tx, fixture.WorkspaceID, "agent-1", time.Now().UTC())
	}))
	registered, ok := fixture.Registry.LookupByIdentity("github", 1)
	require.True(t, ok)
	version := readAgentCatalogVersion(t, ctx, db, fixture.WorkspaceID, "agent-1")

	// A lifecycle mutation lands WITHOUT its catalog mutation: the durable
	// authorization now contradicts the durable catalog state.
	now := time.Now().UTC()
	mustExecService(t, db, `
INSERT INTO service_principal_authorizations (
    id, workspace_id, service_definition_id, service_installation_id,
    service_kind, principal_kind, principal_ref, status, disabled_reason,
    scopes_json, authorized_by_member_id, policy_version, metadata,
    created_at, updated_at, disabled_at, active_authorization_key
) VALUES (
    'authz-driftfence', $1, 'app_github', $2,
    'app', 'agent', 'agent-1', 'disabled', 'drift probe',
    '[]', $3, 1, '{}', $4, $4, $4, NULL
)`, fixture.WorkspaceID, installationID, fixture.HumanMemberID, now)

	_, err := service.Invoke(ctx, invocationParams(
		fixture, "agent-1", "github", "list-watches",
		registered.Definition.Digest, version))
	require.Error(t, err)
	requireInvocationCode(t, err, InvocationAgentCatalogStale)
}

// TestAuthorizerOnlyChangeDoesNotAdvanceCatalogVersion proves the durable
// revision tracks the Agent-visible command set, not server-only policy
// identity. Replacing the authorizer leaves catalog_state and the revision
// unchanged, while an explicit Builtin reauthorization still rejects an
// envelope carrying the superseded authorizer.
func TestAuthorizerOnlyChangeDoesNotAdvanceCatalogVersion(t *testing.T) {
	ctx := context.Background()
	db := startServiceRollbackDB(t, ctx)
	fixture := seedServiceFixture(t, ctx, db, "authorizerfence")
	require.NoError(t, registerProviderWatchBuiltinsForTest(fixture.Registry))
	service := newService(db, fixture.Registry)
	installationID, connectionID := seedProductAppConnection(
		t, db, fixture, "authorizerfence")
	convertGitHubToolsCapabilityToExplicitOnly(
		t, db, fixture, installationID,
		[]string{
			"list-watches", "pause-watch", "resume-watch", "unwatch",
		},
	)
	now := time.Now().UTC()
	require.NoError(t, coredb.WithTx(ctx, db, func(tx coredb.Tx) error {
		return service.ApplyAgentCatalogChange(
			ctx, tx, fixture.WorkspaceID, "agent-1", now)
	}))
	before := readCatalogStateRow(
		t,
		ctx,
		db,
		fixture.WorkspaceID,
		"agent-1",
	)
	require.Equal(t, int64(1), before.Version)

	// Authorizer-only change: a second human member takes over the grant
	// seeded by convertGitHubToolsCapabilityToExplicitOnly
	// (authz-app-authority); the agent-visible action surface is identical.
	secondAuthorizer := "mem_flow_human_authorizerfence2"
	mustExecService(t, db, `
INSERT INTO users (id, email, name, display_name, password_hash, created_at, updated_at)
VALUES ($1, $2, 'Second', 'Second', 'hash', $3, $3)
`, "usr_flow_authorizerfence2", "authorizerfence2@example.com", now)
	mustExecService(t, db, `
INSERT INTO members (
  id, workspace_id, type, display_name, user_id, workspace_role, status,
  created_at, updated_at
) VALUES ($1, $2, 'human', 'Second', $3, 'owner', 'active', $4, $4)`,
		secondAuthorizer, fixture.WorkspaceID, "usr_flow_authorizerfence2", now)
	mustExecService(t, db, `
UPDATE service_principal_authorizations
SET authorized_by_member_id = $1, updated_at = $2
WHERE id = 'authz-app-authority'`, secondAuthorizer, now)
	require.NoError(t, coredb.WithTx(ctx, db, func(tx coredb.Tx) error {
		_, err := service.applyAgentCatalogChange(
			ctx, tx, fixture.WorkspaceID, "agent-1", now)
		return err
	}))
	after := readCatalogStateRow(
		t,
		ctx,
		db,
		fixture.WorkspaceID,
		"agent-1",
	)
	require.Equal(t, before.Version, after.Version,
		"an authorizer-only change must not advance the durable catalog version")
	require.Equal(t, before.State, after.State,
		"server-only policy identity must not enter catalog_state")
	require.Equal(t, before.Enabled, after.Enabled)
	assertUpdateCount(t, ctx, db, fixture.WorkspaceID, "agent-1", 1)

	registered, ok := fixture.Registry.LookupByIdentity("github", 1)
	require.True(t, ok)
	base := BuiltinReauthorization{
		WorkspaceID:                    fixture.WorkspaceID,
		AgentID:                        "agent-1",
		AgentMemberID:                  "mem_flow_agent_agent-1",
		AuthorizedByMemberID:           fixture.HumanMemberID,
		ServiceInstallationID:          installationID,
		ServiceConnectionID:            connectionID,
		ServiceConnectionGeneration:    "stale-generation",
		ServiceInstallationExtensionID: "selection-github-authorizerfence",
		ExtensionName:                  "github",
		Action:                         "list-watches",
		DefinitionVersion:              1,
		DefinitionDigest:               registered.Definition.Digest,
		CatalogVersion:                 after.Version,
	}
	require.Error(t, service.ReauthorizeBuiltin(ctx, base),
		"the fresh identity compare must reject the superseded authorizer")
}

// TestAuthorizerOnlyInterleavingKeepsCatalogRevisionStable is the deterministic
// barrier proof for the snapshot-read design: an in-flight invocation holds a
// pre-change decision while a concurrent transaction changes only the
// authorizer. Because the Agent-visible command set is unchanged, catalog_state
// remains equal, no revision is published, and the invocation may continue.
func TestAuthorizerOnlyInterleavingKeepsCatalogRevisionStable(t *testing.T) {
	ctx := context.Background()
	db := startServiceConcurrentDB(t, ctx)
	fixture := seedServiceFixture(t, ctx, db, "interleave")
	require.NoError(t, registerProviderWatchBuiltinsForTest(fixture.Registry))
	service := newService(db, fixture.Registry)
	installationID, connectionID := seedProductAppConnection(
		t, db, fixture, "interleave")
	convertGitHubToolsCapabilityToExplicitOnly(
		t, db, fixture, installationID,
		[]string{
			"list-watches", "pause-watch", "resume-watch", "unwatch",
		},
	)
	now := time.Now().UTC()
	require.NoError(t, coredb.WithTx(ctx, db, func(tx coredb.Tx) error {
		return service.ApplyAgentCatalogChange(
			ctx, tx, fixture.WorkspaceID, "agent-1", now)
	}))
	before := readCatalogStateRow(
		t,
		ctx,
		db,
		fixture.WorkspaceID,
		"agent-1",
	)
	registered, ok := fixture.Registry.LookupByIdentity("github", 1)
	require.True(t, ok)

	snapshotTaken := make(chan struct{})
	releaseInvocation := make(chan struct{})
	service.decisionsSnapshotHook = func() {
		select {
		case <-snapshotTaken:
		default:
			close(snapshotTaken)
		}
		<-releaseInvocation
	}
	invokeErr := make(chan error, 1)
	go func() {
		_, err := service.Invoke(ctx, invocationParams(
			fixture, "agent-1", "github", "list-watches",
			registered.Definition.Digest, before.Version))
		invokeErr <- err
	}()
	<-snapshotTaken

	// The concurrent transaction changes ONLY the authorizer: the
	// agent-visible command set and canonical catalog state are identical.
	secondAuthorizer := "mem_flow_human_interleave2"
	mustExecService(t, db, `
INSERT INTO users (id, email, name, display_name, password_hash, created_at, updated_at)
VALUES ($1, $2, 'Second', 'Second', 'hash', $3, $3)
`, "usr_flow_interleave2", "interleave2@example.com", now)
	mustExecService(t, db, `
INSERT INTO members (
  id, workspace_id, type, display_name, user_id, workspace_role, status,
  created_at, updated_at
) VALUES ($1, $2, 'human', 'Second', $3, 'owner', 'active', $4, $4)`,
		secondAuthorizer, fixture.WorkspaceID, "usr_flow_interleave2", now)
	mustExecService(t, db, `
UPDATE service_principal_authorizations
SET authorized_by_member_id = $1, updated_at = $2
WHERE id = 'authz-app-authority'`, secondAuthorizer, now)
	require.NoError(t, coredb.WithTx(ctx, db, func(tx coredb.Tx) error {
		return service.ApplyAgentCatalogChange(
			ctx, tx, fixture.WorkspaceID, "agent-1", now)
	}))
	after := readCatalogStateRow(
		t,
		ctx,
		db,
		fixture.WorkspaceID,
		"agent-1",
	)
	require.Equal(t, before, after,
		"the authorizer-only change must preserve revision and catalog_state")
	assertUpdateCount(t, ctx, db, fixture.WorkspaceID, "agent-1", 1)

	close(releaseInvocation)
	select {
	case err := <-invokeErr:
		require.NoError(t, err)
	case <-time.After(30 * time.Second):
		t.Fatal("interleaved invocation did not finish")
	}
	_ = connectionID
}