package server

import (
	"bytes"
	"context"
	"crypto/sha256"
	"encoding/hex"
	"fmt"
	"sync"
	"sync/atomic"
	"testing"
	"time"

	"github.com/stretchr/testify/require"
	"github.com/tidbcloud/AgentTeam/internal/contentpayload"
	"github.com/tidbcloud/AgentTeam/internal/gormtest"
	automationservice "github.com/tidbcloud/AgentTeam/internal/service/automation"
	contentstoreservice "github.com/tidbcloud/AgentTeam/internal/service/content"
	resourceactivityservice "github.com/tidbcloud/AgentTeam/internal/service/resource/activity"
	automationstore "github.com/tidbcloud/AgentTeam/internal/store/automation"
	contentstorestore "github.com/tidbcloud/AgentTeam/internal/store/content"
	activitystore "github.com/tidbcloud/AgentTeam/internal/store/resource/activity"
	"gorm.io/gorm"
)

type contentOutboxBindingReader struct {
	content contentstoreservice.Content
	version contentstoreservice.Version
}

func (reader contentOutboxBindingReader) GetContent(context.Context, string, string) (contentstoreservice.Content, error) {
	return reader.content, nil
}

func (reader contentOutboxBindingReader) GetVersion(context.Context, string, string, string) (contentstoreservice.Version, error) {
	return reader.version, nil
}

type contentOutboxAuthorityReader struct{ authority contentstorestore.Authority }

func (reader contentOutboxAuthorityReader) GetAuthority(context.Context, string, string, string) (contentstorestore.Authority, bool, error) {
	if reader.authority.ID == "" {
		return contentstorestore.Authority{State: contentstorestore.AuthorityStateLegacyActive}, false, nil
	}
	return reader.authority, true, nil
}

type recordingVersionConsumer struct{ calls int }

func (consumer *recordingVersionConsumer) ConsumeVersionCommitted(context.Context, contentstoreservice.VersionCommittedEvent) error {
	consumer.calls++
	return nil
}

type cancelAwareContentOutboxRunner struct {
	once    sync.Once
	started chan struct{}
}

func (runner *cancelAwareContentOutboxRunner) RunOnce(ctx context.Context, _ int) (contentstoreservice.OutboxWorkerResult, error) {
	runner.once.Do(func() { close(runner.started) })
	<-ctx.Done()
	return contentstoreservice.OutboxWorkerResult{}, ctx.Err()
}

func TestContentOutboxWorkerStopsOnCancellation(t *testing.T) {
	runner := &cancelAwareContentOutboxRunner{started: make(chan struct{})}
	cancel, done := startContentOutboxWorker(runner)
	require.NotNil(t, cancel)
	require.NotNil(t, done)
	select {
	case <-runner.started:
	case <-time.After(time.Second):
		t.Fatal("content outbox worker did not start")
	}
	cancel()
	select {
	case <-done:
	case <-time.After(time.Second):
		t.Fatal("content outbox worker did not stop")
	}
}

func TestAutomationContentOutboxGroupWaitsForExactNativeGeneration(t *testing.T) {
	event := contentstoreservice.VersionCommittedEvent{ID: "evt_1", WorkspaceID: "wsp_1", ContentID: "ctn_1", VersionID: "ctv_1"}
	automationConsumer := &recordingVersionConsumer{}
	activityConsumer := &recordingVersionConsumer{}
	reader := contentOutboxBindingReader{
		content: contentstoreservice.Content{ID: "ctn_1", WorkspaceID: "wsp_1", Domain: "automation", BindingKey: "team/repo", Generation: 1},
		version: contentstoreservice.Version{ID: "ctv_1", CreatedBy: "usr_1"},
	}
	group := authorityGatedVersionCommittedConsumers{content: reader, authority: contentOutboxAuthorityReader{}, automationConsumers: versionCommittedConsumers{automationConsumer, activityConsumer}}
	require.ErrorContains(t, group.ConsumeVersionCommitted(t.Context(), event), "not native-active")
	require.Zero(t, automationConsumer.calls)
	require.Zero(t, activityConsumer.calls)

	group.authority = contentOutboxAuthorityReader{authority: contentstorestore.Authority{ID: "cta_1", State: contentstorestore.AuthorityStateNativeActive, Generation: 2}}
	require.ErrorContains(t, group.ConsumeVersionCommitted(t.Context(), event), "not native-active")
	require.Zero(t, automationConsumer.calls)
	require.Zero(t, activityConsumer.calls)

	group.authority = contentOutboxAuthorityReader{authority: contentstorestore.Authority{ID: "cta_1", State: contentstorestore.AuthorityStateNativeActive, Generation: 1}}
	require.NoError(t, group.ConsumeVersionCommitted(t.Context(), event))
	require.Equal(t, 1, automationConsumer.calls)
	require.Equal(t, 1, activityConsumer.calls)
}

func TestAutomationContentOutboxWorkerGroupsExactRunAndActivityByAuthorityGeneration(t *testing.T) {
	db, err := gormtest.OpenSQLiteDB(t)
	require.NoError(t, err)
	require.NoError(t, db.AutoMigrate(
		&automationstore.AutomationRun{}, &automationstore.AutomationRepoStatusOutbox{},
		&automationstore.AutomationContentPublication{}, &automationstore.AutomationContentDefinition{},
		&automationstore.AutomationContentRepository{}, &automationstore.AutomationRunSlot{},
		&automationstore.AutomationJobRun{}, &automationstore.AutomationEvent{},
		&automationstore.AutomationJob{}, &automationstore.AutomationWait{},
		&contentstorestore.Authority{}, &contentstorestore.VersionedContent{}, &contentstorestore.ContentVersion{},
		&contentstorestore.VersionOperation{}, &contentstorestore.VersionPath{}, &contentstorestore.VersionPin{},
		&contentstorestore.Outbox{}, &contentstorestore.StagedPayloadGC{},
		&activitystore.ResourceActivityEvent{}, &activitystore.ResourceActivityLatestCommit{},
	))
	tree, err := contentpayload.NewFilesystemTreeBackend(t.TempDir())
	require.NoError(t, err)
	t.Cleanup(func() { require.NoError(t, tree.Close()) })
	var ids atomic.Int64
	contentDB := contentstorestore.New(db)
	content, err := contentstoreservice.NewTree(contentDB, tree, func(prefix string) (string, error) {
		return fmt.Sprintf("%s_%d", prefix, ids.Add(1)), nil
	})
	require.NoError(t, err)
	lineage, err := content.CreateContent(t.Context(), contentstoreservice.CreateContentRequest{
		WorkspaceID: "wsp_1", Domain: "automation", BindingKey: "team/repo", Model: contentstorestore.ModelTree, Actor: "usr_1",
	})
	require.NoError(t, err)
	definition := "name: Build\non: push\njobs:\n  build:\n    steps:\n      - run: echo grouped\n"
	committed, err := content.CommitVersion(t.Context(), contentstoreservice.CommitVersionRequest{
		WorkspaceID: "wsp_1", ContentID: lineage.ID, ExpectedGeneration: lineage.Generation,
		Actor: "usr_1", Model: contentstorestore.ModelTree, TreeLimits: contentpayload.DefaultLimits,
		Files: []contentpayload.File{{Path: ".agentteam/build.yml", Size: int64(len(definition)), Content: bytes.NewBufferString(definition)}},
	})
	require.NoError(t, err)
	_, err = contentDB.ActivateNativeAuthority(t.Context(), contentstorestore.ActivateNativeAuthorityParams{
		WorkspaceID: "wsp_1", Domain: "automation", BindingKey: "team/repo", ExpectedGeneration: 0,
		Checkpoint: committed.Version.ID, Actor: "usr_1", Now: time.Now().UTC(),
	})
	require.NoError(t, err)
	digest := sha256.Sum256([]byte(definition))
	automations := automationstore.New(db)
	require.NoError(t, automations.RecordContentPublication(t.Context(), automationstore.AutomationContentPublication{
		WorkspaceID: "wsp_1", ContentID: lineage.ID, ContentVersionID: committed.Version.ID,
		RepositoryOwner: "team", RepositoryName: "repo", Ref: "main", ChangedPathsJSON: `[".agentteam/build.yml"]`,
		ActorID: "usr_1", CreatedAt: committed.Version.CreatedAt,
	}, []automationstore.AutomationContentDefinition{{
		DefinitionPath: ".agentteam/build.yml", Digest: hex.EncodeToString(digest[:]), SizeBytes: int64(len(definition)),
	}}, "", automationstore.ContentPublicationRecordStrict))
	service := automationservice.NewAutomationsService(automations, func() time.Time { return time.Now().UTC() }, nil)
	service.SetAuthorizer(automationservice.AllowAutomationAuthorizer{})
	automationConsumer, err := automationservice.NewContentVersionAutomationConsumer(content, service)
	require.NoError(t, err)
	activityConsumer, err := resourceactivityservice.NewContentVersionActivityConsumer(content, activitystore.New(db))
	require.NoError(t, err)
	group := &authorityGatedVersionCommittedConsumers{
		content: content, authority: contentOutboxAuthorityReader{authority: contentstorestore.Authority{
			ID: "cta_wrong", State: contentstorestore.AuthorityStateNativeActive, Generation: lineage.Generation + 1,
		}},
		automationConsumers: versionCommittedConsumers{automationConsumer, activityConsumer},
	}
	worker, err := contentstoreservice.NewOutboxWorker(contentDB, group, "grouped-worker", time.Minute)
	require.NoError(t, err)
	result, err := worker.RunOnce(t.Context(), 10)
	require.NoError(t, err)
	require.Equal(t, contentstoreservice.OutboxWorkerResult{Leased: 1, Retried: 1}, result)
	assertAutomationContentProjectionCounts(t, db, committed.Version.ID, 0, 0)
	var outbox contentstorestore.Outbox
	require.NoError(t, db.Where("version_id = ?", committed.Version.ID).First(&outbox).Error)
	require.Equal(t, contentstorestore.OutboxStatusPending, outbox.Status)
	require.Equal(t, 1, outbox.Attempts)
	require.Empty(t, outbox.LeaseOwner)
	require.Nil(t, outbox.LeaseUntil)
	require.Contains(t, outbox.LastError, "not native-active")

	group.authority = contentDB
	require.NoError(t, db.Model(&contentstorestore.Outbox{}).Where("id = ?", outbox.ID).
		Update("next_attempt_at", time.Unix(0, 0).UTC()).Error)
	result, err = worker.RunOnce(t.Context(), 10)
	require.NoError(t, err)
	require.Equal(t, contentstoreservice.OutboxWorkerResult{Leased: 1, Delivered: 1}, result)
	assertAutomationContentProjectionCounts(t, db, committed.Version.ID, 1, 1)

	// Simulate a lost delivery acknowledgement: the same durable event is
	// leased through the real worker again, while product projections remain
	// idempotent.
	require.NoError(t, db.Model(&contentstorestore.Outbox{}).Where("id = ?", outbox.ID).Updates(map[string]any{
		"status": contentstorestore.OutboxStatusPending, "next_attempt_at": time.Unix(0, 0).UTC(),
	}).Error)
	result, err = worker.RunOnce(t.Context(), 10)
	require.NoError(t, err)
	require.Equal(t, contentstoreservice.OutboxWorkerResult{Leased: 1, Delivered: 1}, result)
	assertAutomationContentProjectionCounts(t, db, committed.Version.ID, 1, 1)
}

func assertAutomationContentProjectionCounts(t *testing.T, db *gorm.DB, versionID string, wantRuns, wantActivities int64) {
	t.Helper()
	var runs int64
	require.NoError(t, db.Model(&automationstore.AutomationRun{}).
		Where("source_kind = ? AND content_version_id = ?", "content_version", versionID).Count(&runs).Error)
	require.Equal(t, wantRuns, runs)
	var activities int64
	require.NoError(t, db.Model(&activitystore.ResourceActivityEvent{}).
		Where("source = ? AND target_id = ?", activitystore.SourceContentStore, versionID).Count(&activities).Error)
	require.Equal(t, wantActivities, activities)
}

func TestContentOutboxGroupPreservesOtherNativeDomainActivity(t *testing.T) {
	otherConsumer := &recordingVersionConsumer{}
	group := authorityGatedVersionCommittedConsumers{
		content:        contentOutboxBindingReader{content: contentstoreservice.Content{ID: "ctn_1", WorkspaceID: "wsp_1", Domain: "skill", BindingKey: "skill_1", Generation: 1}},
		authority:      contentOutboxAuthorityReader{authority: contentstorestore.Authority{ID: "cta_1", State: contentstorestore.AuthorityStateNativeActive, Generation: 1}},
		otherConsumers: versionCommittedConsumers{otherConsumer},
	}
	require.NoError(t, group.ConsumeVersionCommitted(t.Context(), contentstoreservice.VersionCommittedEvent{ID: "evt_1", WorkspaceID: "wsp_1", ContentID: "ctn_1", VersionID: "ctv_1"}))
	require.Equal(t, 1, otherConsumer.calls)
}