package database

import (
	"context"
	"strings"
	"testing"
	"time"

	"github.com/stretchr/testify/require"
	"github.com/tidbcloud/AgentTeam/internal/store/content"
	"gorm.io/gorm"
	"gorm.io/gorm/clause"
)

func TestTiDBContentVersionOperationConcurrentPublishConverges(t *testing.T) {
	db := openTiDBTestDB(t, func(db *gorm.DB) error {
		return db.AutoMigrate(
			&content.VersionedContent{}, &content.ContentVersion{},
			&content.VersionOperation{}, &content.Outbox{},
			&content.StagedPayloadGC{},
		)
	})
	sqlDB, err := db.DB()
	require.NoError(t, err)
	sqlDB.SetMaxOpenConns(8)
	sqlDB.SetMaxIdleConns(8)
	require.GreaterOrEqual(t, sqlDB.Stats().MaxOpenConnections, 3)

	ctx, cancel := context.WithTimeout(context.Background(), 10*time.Second)
	defer cancel()
	now := time.Date(2026, 8, 13, 8, 0, 0, 0, time.UTC)
	store := content.New(db)
	_, err = store.CreateVersionedContent(ctx, content.VersionedContent{
		ID: "ctn_1", WorkspaceID: "wsp_1", Domain: "kb_source", BindingKey: "src_1",
		Model: content.ModelTree, CreatedBy: "worker_1", CreatedAt: now, UpdatedAt: now,
	})
	require.NoError(t, err)

	for _, suffix := range []string{"1", "2"} {
		require.NoError(t, store.RegisterStagedPayloadGC(ctx, content.StagedPayloadGC{
			ID: "ctg_" + suffix, WorkspaceID: "wsp_1", ContentID: "ctn_1",
			VersionID: "ctv_" + suffix, Model: content.ModelTree,
			PayloadRef: "wsp_1/ctn_1/ctv_" + suffix, DeleteAfter: now.Add(time.Hour),
		}))
	}

	// Hold the lineage row so both PublishVersion transactions reach the real
	// TiDB row lock before either can win. No application mutex serializes them.
	blocker := db.WithContext(ctx).Begin()
	require.NoError(t, blocker.Error)
	var locked content.VersionedContent
	require.NoError(t, blocker.Clauses(clause.Locking{Strength: "UPDATE"}).Where("workspace_id = ? AND id = ?", "wsp_1", "ctn_1").First(&locked).Error)

	type publishResult struct {
		result content.PublishVersionResult
		err    error
	}
	results := make(chan publishResult, 2)
	for _, suffix := range []string{"1", "2"} {
		suffix := suffix
		go func() {
			result, publishErr := store.PublishVersion(ctx, content.PublishVersionParams{
				WorkspaceID: "wsp_1", ContentID: "ctn_1", ExpectedGeneration: 1,
				Actor: "worker_1", PublishedAt: now,
				Version: content.ContentVersion{
					ID: "ctv_" + suffix, Model: content.ModelTree,
					PayloadRef: "wsp_1/ctn_1/ctv_" + suffix,
					RootDigest: strings.Repeat("a", 64), SizeBytes: 5, FileCount: 1,
				},
				Outbox: content.Outbox{ID: "cto_" + suffix},
				GC:     content.StagedPayloadGC{ID: "ctg_" + suffix},
				Operation: &content.VersionOperation{
					ID: "ctop_" + suffix, OperationKey: "kb.source-mirror:job-1",
					RequestDigest: strings.Repeat("b", 64), OperationNamespace: "kb.source-mirror",
					OperationVersion: 1, BoundsPolicyNamespace: "content.tree", BoundsPolicyVersion: 1,
					MaxFiles: 10_000, MaxFileBytes: 128 << 20, MaxTotalBytes: 512 << 20, MaxManifestBytes: 8 << 20,
				},
			})
			results <- publishResult{result: result, err: publishErr}
		}()
	}
	select {
	case completed := <-results:
		t.Fatalf("publish returned while the lineage row was locked: %v", completed.err)
	case <-time.After(100 * time.Millisecond):
	}
	require.NoError(t, blocker.Commit().Error)

	first, second := <-results, <-results
	require.NoError(t, first.err)
	require.NoError(t, second.err)
	require.Equal(t, first.result.Version.ID, second.result.Version.ID)

	for model, want := range map[any]int64{
		&content.ContentVersion{}:   1,
		&content.VersionOperation{}: 1,
		&content.Outbox{}:           1,
	} {
		var count int64
		require.NoError(t, db.Model(model).Count(&count).Error)
		require.Equal(t, want, count)
	}
	var pendingGC int64
	require.NoError(t, db.Model(&content.StagedPayloadGC{}).Where("status = ?", content.GCStatusPending).Count(&pendingGC).Error)
	require.Equal(t, int64(1), pendingGC, "the replay loser remains independently reclaimable")
}