package automation

import (
	"context"
	"errors"
	"fmt"
	"strings"
	"sync"
	"sync/atomic"
	"testing"
	"time"

	"github.com/stretchr/testify/require"
	"github.com/tidbcloud/AgentTeam/internal/contentpayload"
	contentstoreservice "github.com/tidbcloud/AgentTeam/internal/service/content"
	contentmigrationservice "github.com/tidbcloud/AgentTeam/internal/service/content/migration"
	automationstore "github.com/tidbcloud/AgentTeam/internal/store/automation"
	automationcutoverstore "github.com/tidbcloud/AgentTeam/internal/store/automationcutover"
	contentstorestore "github.com/tidbcloud/AgentTeam/internal/store/content"
	contentmigrationstore "github.com/tidbcloud/AgentTeam/internal/store/content/migration"
	resourcerepostore "github.com/tidbcloud/AgentTeam/internal/store/resource/repos"
)

type fakeAutomationMigrationExporter struct{ exports atomic.Int64 }

func (exporter *fakeAutomationMigrationExporter) ExportExactAutomationRepository(_ context.Context, _ string, commitSHA string, _ contentpayload.Limits) (contentmigrationservice.Export, error) {
	exporter.exports.Add(1)
	return contentmigrationservice.Export{Files: []contentmigrationservice.File{{
		Path: ".agentteam/build.yml", Data: []byte("name: Build\non: push\njobs:\n  build:\n    steps:\n      - run: echo " + commitSHA + "\n"),
	}}}, nil
}

type blockingAutomationMigrationExporter struct {
	started chan struct{}
	release chan struct{}
	once    sync.Once
}

func (exporter *blockingAutomationMigrationExporter) ExportExactAutomationRepository(ctx context.Context, _ string, commitSHA string, _ contentpayload.Limits) (contentmigrationservice.Export, error) {
	exporter.once.Do(func() { close(exporter.started) })
	select {
	case <-ctx.Done():
		return contentmigrationservice.Export{}, ctx.Err()
	case <-exporter.release:
	}
	return contentmigrationservice.Export{Files: []contentmigrationservice.File{{
		Path: ".agentteam/build.yml", Data: []byte("name: Build\non: push\njobs:\n  build:\n    steps:\n      - run: echo " + commitSHA + "\n"),
	}}}, nil
}

type failingAutomationContentAuthority struct {
	store    *contentstorestore.Store
	failNext atomic.Bool
	loseNext atomic.Bool
}

func (authority *failingAutomationContentAuthority) GetAuthority(ctx context.Context, workspaceID, domain, bindingKey string) (contentstorestore.Authority, bool, error) {
	return authority.store.GetAuthority(ctx, workspaceID, domain, bindingKey)
}

func (authority *failingAutomationContentAuthority) AcquireLegacyAuthorityFence(ctx context.Context, params contentstorestore.AcquireLegacyAuthorityFenceParams) (contentstorestore.LegacyAuthorityFence, error) {
	return authority.store.AcquireLegacyAuthorityFence(ctx, params)
}

func (authority *failingAutomationContentAuthority) ReleaseLegacyAuthorityFence(ctx context.Context, fence contentstorestore.LegacyAuthorityFence, now time.Time) error {
	return authority.store.ReleaseLegacyAuthorityFence(ctx, fence, now)
}

func (authority *failingAutomationContentAuthority) ActivateNativeAuthority(ctx context.Context, params contentstorestore.ActivateNativeAuthorityParams) (contentstorestore.Authority, error) {
	if authority.failNext.Swap(false) {
		return contentstorestore.Authority{}, errors.New("injected activation failure")
	}
	activated, err := authority.store.ActivateNativeAuthority(ctx, params)
	if err == nil && authority.loseNext.Swap(false) {
		return contentstorestore.Authority{}, errors.New("injected response loss")
	}
	return activated, err
}

func TestExactAutomationContentMigrationRequiresDeltaBeforeActivationAndReplays(t *testing.T) {
	db, err := openServiceTestSQLiteDB(t)
	require.NoError(t, err)
	require.NoError(t, db.AutoMigrate(
		&automationstore.AutomationContentPublication{}, &automationstore.AutomationContentRepository{}, &automationstore.AutomationContentDefinition{},
		&contentstorestore.Authority{}, &contentstorestore.VersionedContent{}, &contentstorestore.ContentVersion{},
		&contentstorestore.VersionOperation{}, &contentstorestore.VersionPath{}, &contentstorestore.VersionPin{},
		&contentstorestore.Outbox{}, &contentstorestore.StagedPayloadGC{},
		&contentmigrationstore.Mapping{}, &contentmigrationstore.Job{}, &resourcerepostore.ResourceRepository{},
	))
	tree, err := contentpayload.NewFilesystemTreeBackend(t.TempDir())
	require.NoError(t, err)
	t.Cleanup(func() { require.NoError(t, tree.Close()) })
	var ids atomic.Int64
	newID := func(prefix string) (string, error) { return fmt.Sprintf("%s_%d", prefix, ids.Add(1)), nil }
	contentDB := contentstorestore.New(db)
	authorityStore := &failingAutomationContentAuthority{store: contentDB}
	core, err := contentstoreservice.NewTree(contentDB, tree, newID)
	require.NoError(t, err)
	workspace := &mutableAutomationWorkspaceAccess{}
	workspace.allowed.Store(true)
	repositories := resourcerepostore.New(db)
	policy, err := NewAutomationContentAccessPolicy(workspace, repositories, contentDB)
	require.NoError(t, err)
	authorized, err := contentstoreservice.NewAuthorizedService(core, policy)
	require.NoError(t, err)
	worker, err := contentmigrationservice.NewWorker(contentmigrationstore.New(db), authorized, newID, "automation", "automation-test", 3*time.Minute)
	require.NoError(t, err)
	now := time.Now().UTC().Truncate(time.Second)
	require.NoError(t, db.Create(&resourcerepostore.ResourceRepository{
		ID: "rrp_1", WorkspaceID: "wsp_1", ResourceType: resourcerepostore.ResourceTypeAutomationRepository,
		ResourceID: resourcerepostore.AutomationRepositoryResourceID("team/repo"), RepoRole: resourcerepostore.RepoRolePrimary,
		RepoOwner: "team", RepoName: "repo", RepoFullName: "team/repo", AGSRepoID: "42", DefaultRef: "main",
		CreatedBy: "usr_1", UpdatedBy: "usr_1", CreatedAt: now, UpdatedAt: now,
	}).Error)
	store := automationstore.New(db)
	exporter := &fakeAutomationMigrationExporter{}
	service := NewAutomationsService(store, func() time.Time { return now }, nil)
	service.SetRepositoryStore(repositories)
	service.ConfigureContentMigration(AutomationContentMigrationOptions{
		Worker: worker, Exporter: exporter, Mappings: contentmigrationstore.New(db), Content: authorized, Authority: authorityStore,
		Cutover: automationcutoverstore.New(db),
	})
	request := AutomationContentMigrationRequest{RepositoryID: "42", CommitSHA: strings.Repeat("a", 40), Ref: "main"}

	full, err := service.RunExactAutomationContentMigration(t.Context(), "wsp_1", "usr_1", "team", "repo", request)
	require.NoError(t, err)
	require.NotEmpty(t, full.ContentID)
	require.NotEmpty(t, full.ContentVersionID)
	require.EqualValues(t, 1, exporter.exports.Load())
	authority, found, err := contentDB.GetAuthority(t.Context(), "wsp_1", "automation", "team/repo")
	require.NoError(t, err)
	require.True(t, found)
	require.Equal(t, contentstorestore.AuthorityStateLegacyActive, authority.State)
	require.Empty(t, authority.Checkpoint)
	current, err := store.GetCurrentContentRepository(t.Context(), "wsp_1", "team", "repo")
	require.NoError(t, err)
	require.Equal(t, full.ContentVersionID, current.ContentVersionID)
	_, err = service.ActivateNativeAutomationContent(t.Context(), "wsp_1", "usr_1", "team", "repo", AutomationContentActivationRequest{
		RepositoryID: "42", FullCommitSHA: request.CommitSHA, FullContentVersionID: full.ContentVersionID,
		DeltaCommitSHA: request.CommitSHA, DeltaContentVersionID: full.ContentVersionID, DeltaObservedAt: now,
	})
	require.Error(t, err, "a full import cannot be replayed as its own pseudo-delta")
	_, err = service.ActivateNativeAutomationContent(t.Context(), "wsp_1", "usr_1", "team", "repo", AutomationContentActivationRequest{
		RepositoryID: "42", FullCommitSHA: request.CommitSHA, FullContentVersionID: full.ContentVersionID,
		DeltaCommitSHA: strings.Repeat("b", 40), DeltaContentVersionID: "ctv_missing", DeltaObservedAt: now,
	})
	require.ErrorContains(t, err, "delta observation evidence is incomplete")
	_, found, err = contentDB.GetAuthority(t.Context(), "wsp_1", "automation", "team/repo")
	require.NoError(t, err)
	require.True(t, found, "full import may materialize only an explicit legacy fence row")
	require.Equal(t, contentstorestore.AuthorityStateLegacyActive, authority.State)

	deltaRequest := AutomationContentMigrationRequest{
		RepositoryID: "42", CommitSHA: strings.Repeat("b", 40), Ref: "main",
		ExpectedContentVersionID: full.ContentVersionID,
	}
	now = now.Add(time.Minute)
	delta, err := service.RunExactAutomationContentMigration(t.Context(), "wsp_1", "usr_1", "team", "repo", deltaRequest)
	require.NoError(t, err)
	require.NotEqual(t, full.ContentVersionID, delta.ContentVersionID)
	require.EqualValues(t, 2, exporter.exports.Load())
	authority, found, err = contentDB.GetAuthority(t.Context(), "wsp_1", "automation", "team/repo")
	require.NoError(t, err)
	require.True(t, found, "delta observation must not expose native routing")
	require.Equal(t, contentstorestore.AuthorityStateLegacyActive, authority.State)
	current, err = store.GetCurrentContentRepository(t.Context(), "wsp_1", "team", "repo")
	require.NoError(t, err)
	require.Equal(t, delta.ContentVersionID, current.ContentVersionID)
	nativeHead, err := core.GetContent(t.Context(), "wsp_1", delta.ContentID)
	require.NoError(t, err)
	require.Equal(t, delta.ContentVersionID, nativeHead.HeadVersionID)
	authorizedHead, authorizedFound, err := authorized.FindContentByBinding(
		WithAutomationContentAccess(t.Context(), "wsp_1", "team/repo", "", 0, "usr_1", contentstoreservice.AccessReadMetadata),
		"wsp_1", "automation", "team/repo",
	)
	require.NoError(t, err)
	require.True(t, authorizedFound)
	require.Equal(t, delta.ContentVersionID, authorizedHead.HeadVersionID)

	observedAt := now
	now = now.Add(time.Minute)
	activation := AutomationContentActivationRequest{
		RepositoryID: "42", FullCommitSHA: request.CommitSHA, FullContentVersionID: full.ContentVersionID,
		DeltaCommitSHA: deltaRequest.CommitSHA, DeltaContentVersionID: delta.ContentVersionID,
		DeltaObservedAt: observedAt,
	}
	canceled, cancel := context.WithCancel(t.Context())
	cancel()
	_, err = service.ActivateNativeAutomationContent(canceled, "wsp_1", "usr_1", "team", "repo", activation)
	require.Error(t, err)
	_, found, err = contentDB.GetAuthority(t.Context(), "wsp_1", "automation", "team/repo")
	require.NoError(t, err)
	require.True(t, found, "canceled cutoff must leave the binding legacy")
	require.Equal(t, contentstorestore.AuthorityStateLegacyActive, authority.State)

	// A second process may have passed its initial legacy read before cutoff.
	// Its durable binding fence must exclude activation until the exact delta
	// publication and projection finish; retrying the stale pair then fails
	// closed, while the newly observed adjacent pair can activate.
	blockingExporter := &blockingAutomationMigrationExporter{started: make(chan struct{}), release: make(chan struct{})}
	secondWorker, err := contentmigrationservice.NewWorker(contentmigrationstore.New(db), authorized, newID, "automation", "automation-test-2", 3*time.Minute)
	require.NoError(t, err)
	secondService := NewAutomationsService(store, func() time.Time { return now }, nil)
	secondService.SetRepositoryStore(repositories)
	secondService.ConfigureContentMigration(AutomationContentMigrationOptions{
		Worker: secondWorker, Exporter: blockingExporter, Mappings: contentmigrationstore.New(db), Content: authorized, Authority: authorityStore,
		Cutover: automationcutoverstore.New(db),
	})
	concurrentRequest := AutomationContentMigrationRequest{
		RepositoryID: "42", CommitSHA: strings.Repeat("c", 40), Ref: "main",
		ExpectedContentVersionID: delta.ContentVersionID,
	}
	type migrationOutcome struct {
		result AutomationContentMigrationResult
		err    error
	}
	outcome := make(chan migrationOutcome, 1)
	go func() {
		result, runErr := secondService.RunExactAutomationContentMigration(t.Context(), "wsp_1", "usr_1", "team", "repo", concurrentRequest)
		outcome <- migrationOutcome{result: result, err: runErr}
	}()
	<-blockingExporter.started
	_, err = service.ActivateNativeAutomationContent(t.Context(), "wsp_1", "usr_1", "team", "repo", activation)
	require.ErrorIs(t, err, contentstorestore.ErrAuthorityConflict)
	close(blockingExporter.release)
	concurrent := <-outcome
	require.NoError(t, concurrent.err)
	_, err = service.ActivateNativeAutomationContent(t.Context(), "wsp_1", "usr_1", "team", "repo", activation)
	require.Error(t, err, "stale delta evidence cannot activate after a newer exact projection")
	now = now.Add(time.Minute)
	activation = AutomationContentActivationRequest{
		RepositoryID: "42", FullCommitSHA: deltaRequest.CommitSHA, FullContentVersionID: delta.ContentVersionID,
		DeltaCommitSHA: concurrentRequest.CommitSHA, DeltaContentVersionID: concurrent.result.ContentVersionID,
		DeltaObservedAt: now,
	}

	authorityStore.failNext.Store(true)
	_, err = service.ActivateNativeAutomationContent(t.Context(), "wsp_1", "usr_1", "team", "repo", activation)
	require.ErrorContains(t, err, "injected activation failure")
	_, found, err = contentDB.GetAuthority(t.Context(), "wsp_1", "automation", "team/repo")
	require.NoError(t, err)
	require.True(t, found, "failed cutoff must leave the binding legacy and retryable")
	require.Equal(t, contentstorestore.AuthorityStateLegacyActive, authority.State)

	authorityStore.loseNext.Store(true)
	_, err = service.ActivateNativeAutomationContent(t.Context(), "wsp_1", "usr_1", "team", "repo", activation)
	require.ErrorContains(t, err, "injected response loss")
	authority, found, err = contentDB.GetAuthority(t.Context(), "wsp_1", "automation", "team/repo")
	require.NoError(t, err)
	require.True(t, found)
	require.Equal(t, contentstorestore.AuthorityStateNativeActive, authority.State)

	replayed, err := service.ActivateNativeAutomationContent(t.Context(), "wsp_1", "usr_1", "team", "repo", activation)
	require.NoError(t, err)
	require.True(t, replayed.Replayed)
	require.Equal(t, uint64(1), replayed.AuthorityGeneration)
	require.Equal(t, authority.UpdatedAt.UTC(), replayed.CutoffAt)
	require.EqualValues(t, 2, exporter.exports.Load(), "activation replay must not read AGS")
	wrongGeneration := activation
	wrongGeneration.ExpectedAuthorityGeneration = replayed.AuthorityGeneration
	_, err = service.ActivateNativeAutomationContent(t.Context(), "wsp_1", "usr_1", "team", "repo", wrongGeneration)
	require.ErrorIs(t, err, contentstorestore.ErrAuthorityConflict, "replay requires exactly ExpectedGeneration+1")

	replayedDelta, err := service.RunExactAutomationContentMigration(t.Context(), "wsp_1", "usr_1", "team", "repo", concurrentRequest)
	require.NoError(t, err)
	require.True(t, replayedDelta.Replayed)
	other := concurrentRequest
	other.CommitSHA = strings.Repeat("d", 40)
	other.ExpectedContentVersionID = concurrent.result.ContentVersionID
	_, err = service.RunExactAutomationContentMigration(t.Context(), "wsp_1", "usr_1", "team", "repo", other)
	require.ErrorIs(t, err, contentstorestore.ErrAuthorityConflict)
	require.EqualValues(t, 2, exporter.exports.Load(), "activated binding must reject a new AGS export")
	content, err := core.GetContent(t.Context(), "wsp_1", delta.ContentID)
	require.NoError(t, err)
	require.Equal(t, concurrent.result.ContentVersionID, content.HeadVersionID, "rejected post-cutoff import must not advance native head")
}

func TestAutomationContentMigrationPreflightsDefinitionsBeforePublication(t *testing.T) {
	service := NewAutomationsService(nil, time.Now, nil)
	require.NoError(t, service.validateAutomationMigrationExport(contentmigrationservice.Export{Files: []contentmigrationservice.File{{
		Path: ".agentteam/build.yml", Data: []byte("name: Build\non: push\njobs:\n  build:\n    steps:\n      - run: echo ok\n"),
	}}}))
	require.Error(t, service.validateAutomationMigrationExport(contentmigrationservice.Export{Files: []contentmigrationservice.File{{
		Path: ".agentteam/build.yml", Data: []byte("not: [valid"),
	}}}))
	require.Error(t, service.validateAutomationMigrationExport(contentmigrationservice.Export{Files: []contentmigrationservice.File{{
		Path: "README.md", Data: []byte("no automation definition"),
	}}}))
}