//go:build runtime_worker_integration

package persistence

import (
	"context"
	"database/sql"
	"errors"
	"fmt"
	"os"
	"strings"
	"testing"
	"time"

	_ "github.com/go-sql-driver/mysql"
	database "github.com/nexuscrm/db"
	"github.com/nexuscrm/kernel/pkg/schemacatalog"
	"github.com/nexuscrm/shared/pkg/models"
)

func TestBulkWorkRealTiDBClaimFenceAndAtomicCheckpoint(t *testing.T) {
	if os.Getenv("TIDB_HOST") == "" {
		t.Skip("runtime_worker_integration requires the declared TiDB environment")
	}
	db, cleanup := openBulkWorkScratchDB(t)
	defer cleanup()
	control, err := NewRuntimeWorkOwnershipControl(db)
	if err != nil {
		t.Fatal(err)
	}
	owner, err := control.Activate(t.Context(), RuntimeWorkOwnershipActivation{
		OperationID: "activate-bulk-work-one", WorkFamily: "runtime.bulk-ingest.v1",
		DeploymentID:        "bulk-worker-one",
		WorkerReleaseDigest: "sha256:aaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaa",
	})
	if err != nil {
		t.Fatal(err)
	}
	authority, err := NewRuntimeWorkOwnershipRepository(db, owner)
	if err != nil {
		t.Fatal(err)
	}
	transactions := NewTransactionManager(database.NewConnectionFromDB(db))
	first, err := NewBulkWorkStore(db, transactions, authority, "holder-one", owner.OwnershipEpoch, `{"schema":"candidate"}`, time.Minute, 3)
	if err != nil {
		t.Fatal(err)
	}
	second, err := NewBulkWorkStore(db, transactions, authority, "holder-two", owner.OwnershipEpoch, `{"schema":"candidate"}`, time.Minute, 3)
	if err != nil {
		t.Fatal(err)
	}
	if err := first.CheckReady(t.Context()); err != nil {
		t.Fatal(err)
	}
	now := time.Now().UTC()
	repository := NewBulkRepository(db, transactions)
	if err := repository.CreateJob(t.Context(), &models.BulkJob{
		ID: "bulk-job-one", Object: "Account", Operation: models.OperationInsert,
		State: models.JobStateOpen, ConcurrencyMode: models.ConcurrencyParallel,
		ContentType: "CSV", ColumnDelimiter: "COMMA", LineEnding: "LF", ApiVersion: 63,
		CreatedById: "user-one", CreatedDate: now, SystemModstamp: now,
	}); err != nil {
		t.Fatal(err)
	}
	if err := repository.insertBatch(t.Context(), db, &models.BulkBatch{
		ID: "batch-one", JobID: "bulk-job-one", State: "Queued",
		ObjectKey: "bulk/batch-one.csv", StorageProvider: "filesystem",
		StorageBucket: "local-filesystem", StorageAuthorityDigest: "sha256:" + strings.Repeat("a", 64),
		ContentSHA256: "aaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaa",
		CreatedDate:   now, SystemModstamp: now,
	}); err != nil {
		t.Fatal(err)
	}
	changed, err := repository.AcceptIngestJob(t.Context(), "bulk-job-one")
	if err != nil || !changed {
		t.Fatalf("accept job = %t, %v", changed, err)
	}

	claims, _, err := first.Claim(t.Context(), 1)
	if err != nil || len(claims) != 1 {
		t.Fatalf("first claim = %+v, %v", claims, err)
	}
	if claims[0].Job.WorkBinding != `{"schema":"candidate"}` {
		t.Fatalf("first claim binding = %q, want deployment candidate", claims[0].Job.WorkBinding)
	}
	if competing, _, err := second.Claim(t.Context(), 1); err != nil || len(competing) != 0 {
		t.Fatalf("competing claim = %+v, %v", competing, err)
	}
	claim := claims[0]
	injected := errors.New("crash after effect before commit")
	err = first.RunClaimTransaction(t.Context(), claim, func(txCtx context.Context, _ *BulkRepository) error {
		if _, err := first.repository.executor(txCtx).ExecContext(txCtx, `INSERT INTO bulk_effect (Id) VALUES ('effect-one')`); err != nil {
			return err
		}
		if err := first.UpdateBatchCheckpoint(txCtx, "batch-one", 100, 100, 0, 100, "InProgress", ""); err != nil {
			return err
		}
		return injected
	})
	if !errors.Is(err, injected) {
		t.Fatalf("injected transaction = %v", err)
	}
	if !errors.Is(err, ErrBulkClaimTransactionRolledBack) {
		t.Fatalf("injected transaction lacks proven-rollback classification: %v", err)
	}
	assertBulkEffectAndCursor(t, db, 0, 0)

	if err := first.RunClaimTransaction(t.Context(), claim, func(txCtx context.Context, _ *BulkRepository) error {
		if _, err := first.repository.executor(txCtx).ExecContext(txCtx, `INSERT INTO bulk_effect (Id) VALUES ('effect-one')`); err != nil {
			return err
		}
		return first.UpdateBatchCheckpoint(txCtx, "batch-one", 100, 100, 0, 100, "Completed", "")
	}); err != nil {
		t.Fatal(err)
	}
	assertBulkEffectAndCursor(t, db, 1, 100)

	if _, err := db.Exec(`UPDATE _System_BulkJob SET AvailableAt = TIMESTAMPADD(SECOND, -1, CURRENT_TIMESTAMP(6)), LeaseExpiresAt = TIMESTAMPADD(SECOND, -1, CURRENT_TIMESTAMP(6)) WHERE Id = 'bulk-job-one'`); err != nil {
		t.Fatal(err)
	}
	reclaimed, _, err := second.Claim(t.Context(), 1)
	if err != nil || len(reclaimed) != 1 || reclaimed[0].Token == claim.Token {
		t.Fatalf("lease reclaim = %+v, %v", reclaimed, err)
	}
	if err := first.RunClaimTransaction(t.Context(), claim, func(context.Context, *BulkRepository) error { return nil }); err == nil {
		t.Fatal("stale claim crossed replacement token")
	}

	// Abort owns the nonterminal state transition and invalidates the live
	// replacement claim. The worker treats this typed fence as normal operator
	// cancellation instead of retrying or making the controller unusable.
	aborted, err := repository.AbortIngestJob(t.Context(), "bulk-job-one")
	if err != nil || !aborted {
		t.Fatalf("abort claimed job = %t, %v", aborted, err)
	}
	callbackRan := false
	err = second.RunClaimTransaction(t.Context(), reclaimed[0], func(context.Context, *BulkRepository) error {
		callbackRan = true
		return nil
	})
	if callbackRan || !errors.Is(err, ErrBulkIngestJobAborted) || !errors.Is(err, ErrBulkClaimTransactionRolledBack) {
		t.Fatalf("aborted replacement claim callback=%t err=%v", callbackRan, err)
	}
	var state models.BulkJobState
	var ownerID, token sql.NullString
	if err := db.QueryRow(`SELECT State, ClaimOwner, ClaimToken FROM _System_BulkJob WHERE Id = 'bulk-job-one'`).Scan(&state, &ownerID, &token); err != nil {
		t.Fatal(err)
	}
	if state != models.JobStateAborted || ownerID.Valid || token.Valid {
		t.Fatalf("aborted durable state=%s owner=%v token=%v", state, ownerID, token)
	}
}

func TestBulkWorkRealTiDBCutoverRejectsFutureReachableOpenBatches(t *testing.T) {
	if os.Getenv("TIDB_HOST") == "" {
		t.Skip("runtime_worker_integration requires the declared TiDB environment")
	}
	db, cleanup := openBulkWorkScratchDB(t)
	defer cleanup()
	now := time.Now().UTC()
	repository := NewBulkRepository(db, NewTransactionManager(database.NewConnectionFromDB(db)))
	if err := repository.CreateJob(t.Context(), &models.BulkJob{
		ID: "bulk-open-before-cutover", Object: "Account", Operation: models.OperationInsert,
		State: models.JobStateOpen, ConcurrencyMode: models.ConcurrencyParallel,
		ContentType: "CSV", ColumnDelimiter: "COMMA", LineEnding: "LF", ApiVersion: 63,
		CreatedById: "user-one", CreatedDate: now, SystemModstamp: now,
	}); err != nil {
		t.Fatal(err)
	}
	if err := repository.insertBatch(t.Context(), db, &models.BulkBatch{
		ID: "legacy-open-batch", JobID: "bulk-open-before-cutover", State: "Queued",
		CreatedDate: now, SystemModstamp: now,
	}); err != nil {
		t.Fatal(err)
	}
	if err := VerifyNoLegacyPendingBulkWork(t.Context(), db); err == nil {
		t.Fatal("Open job with an uploaded legacy batch crossed the cutover gate")
	}

	if _, err := db.Exec(`UPDATE _System_BulkBatch SET
ObjectKey = 'bulk/old-authority.csv', StorageProvider = 'filesystem',
StorageBucket = 'old-bucket', StorageAuthorityDigest = ?, ContentSHA256 = ?
WHERE Id = 'legacy-open-batch'`,
		"sha256:"+strings.Repeat("a", 64), strings.Repeat("b", 64),
	); err != nil {
		t.Fatal(err)
	}
	if err := VerifyNoLegacyPendingBulkWork(t.Context(), db); err != nil {
		t.Fatalf("complete Open batch was misclassified as legacy: %v", err)
	}
	if err := VerifyPendingBulkStorageAuthority(
		t.Context(), db, "filesystem", "new-bucket", "sha256:"+strings.Repeat("c", 64),
	); err == nil {
		t.Fatal("Open batch bound to the old storage authority crossed the cutover gate")
	}
}

func assertBulkEffectAndCursor(t *testing.T, db *sql.DB, effects int, cursor int64) {
	t.Helper()
	var gotEffects int
	if err := db.QueryRow(`SELECT COUNT(*) FROM bulk_effect`).Scan(&gotEffects); err != nil {
		t.Fatal(err)
	}
	var gotCursor int64
	if err := db.QueryRow(`SELECT NextRowOffset FROM _System_BulkBatch WHERE Id = 'batch-one'`).Scan(&gotCursor); err != nil {
		t.Fatal(err)
	}
	if gotEffects != effects || gotCursor != cursor {
		t.Fatalf("effect/cursor = %d/%d, want %d/%d", gotEffects, gotCursor, effects, cursor)
	}
}

func openBulkWorkScratchDB(t *testing.T) (*sql.DB, func()) {
	t.Helper()
	host, port := os.Getenv("TIDB_HOST"), os.Getenv("TIDB_PORT")
	if port == "" {
		port = "4000"
	}
	user := os.Getenv("TIDB_USER")
	if user == "" {
		user = "root"
	}
	password := os.Getenv("TIDB_PASSWORD")
	admin, err := sql.Open("mysql", fmt.Sprintf("%s:%s@tcp(%s:%s)/?parseTime=true&loc=UTC", user, password, host, port))
	if err != nil {
		t.Fatal(err)
	}
	name := fmt.Sprintf("nexuscrm_bulk_work_%d", time.Now().UnixNano())
	if _, err := admin.Exec("CREATE DATABASE `" + name + "`"); err != nil {
		t.Fatal(err)
	}
	db, err := sql.Open("mysql", fmt.Sprintf("%s:%s@tcp(%s:%s)/%s?parseTime=true&loc=UTC", user, password, host, port, name))
	if err != nil {
		t.Fatal(err)
	}
	db.SetMaxOpenConns(16)
	wanted := map[string]bool{
		"_System_BulkJob": false, "_System_BulkBatch": false,
		"_System_BulkWorkGeneration": false, "_System_RuntimeWorkFence": false,
		"_System_RuntimeWorkOwnership": false,
	}
	definitions, err := schemacatalog.SystemTableDefinitions()
	if err != nil {
		t.Fatal(err)
	}
	repository := NewSchemaRepository(db, nil)
	for _, definition := range definitions {
		if _, ok := wanted[definition.TableName]; !ok {
			continue
		}
		if err := repository.CreatePhysicalTable(t.Context(), definition); err != nil {
			t.Fatalf("create %s: %v", definition.TableName, err)
		}
		wanted[definition.TableName] = true
	}
	for table, found := range wanted {
		if !found {
			t.Fatalf("canonical catalog is missing %s", table)
		}
	}
	if _, err := db.Exec(`CREATE TABLE bulk_effect (Id VARCHAR(255) PRIMARY KEY)`); err != nil {
		t.Fatal(err)
	}
	return db, func() {
		_ = db.Close()
		_, _ = admin.Exec("DROP DATABASE IF EXISTS `" + name + "`")
		_ = admin.Close()
	}
}