//go:build runtime_worker_integration

package persistence

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

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

func TestRuntimeWorkFenceSlotsDrainOnOwnershipTransferRealTiDB(t *testing.T) {
	if os.Getenv("TIDB_HOST") == "" {
		t.Skip("runtime_worker_integration requires the declared TiDB environment")
	}
	db, cleanup := openRuntimeWorkFenceScratchDB(t)
	defer cleanup()
	control, err := NewRuntimeWorkOwnershipControl(db)
	if err != nil {
		t.Fatal(err)
	}
	owner, err := control.Activate(t.Context(), RuntimeWorkOwnershipActivation{
		OperationID: "activate-record-fence-one", WorkFamily: "runtime.record-event-dispatch.v1",
		DeploymentID:        "runtime-record-fence-one",
		WorkerReleaseDigest: "sha256:aaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaa",
	})
	if err != nil {
		t.Fatal(err)
	}
	authority, err := NewRuntimeWorkOwnershipRepository(db, owner)
	if err != nil {
		t.Fatal(err)
	}
	if err := authority.CheckFencesReady(t.Context()); err != nil {
		t.Fatalf("seeded fence readiness: %v", err)
	}
	keyOne := "Issue/issue-fence-one"
	keyTwo := "Issue/issue-fence-two"
	for RuntimeWorkFenceSlot(keyTwo) == RuntimeWorkFenceSlot(keyOne) {
		keyTwo += "-next"
	}

	txOne, cleanupOne, err := BeginPessimisticTx(t.Context(), db)
	if err != nil {
		t.Fatal(err)
	}
	defer cleanupOne()
	if err := authority.LockActiveFence(t.Context(), txOne, keyOne); err != nil {
		t.Fatal(err)
	}
	txTwo, cleanupTwo, err := BeginPessimisticTx(t.Context(), db)
	if err != nil {
		t.Fatal(err)
	}
	defer cleanupTwo()
	secondLocked := make(chan error, 1)
	go func() { secondLocked <- authority.LockActiveFence(t.Context(), txTwo, keyTwo) }()
	select {
	case err := <-secondLocked:
		if err != nil {
			t.Fatal(err)
		}
	case <-time.After(time.Second):
		t.Fatal("different fixed fence slots serialized")
	}

	transfer := make(chan error, 1)
	go func() {
		_, err := control.Activate(t.Context(), RuntimeWorkOwnershipActivation{
			OperationID: "activate-record-fence-two", WorkFamily: owner.WorkFamily,
			DeploymentID: "runtime-record-fence-two", WorkerReleaseDigest: owner.WorkerReleaseDigest,
			ExpectedOwnershipEpoch: owner.OwnershipEpoch,
		})
		transfer <- err
	}()
	select {
	case err := <-transfer:
		t.Fatalf("ownership transfer crossed live effect slots: %v", err)
	case <-time.After(100 * time.Millisecond):
	}
	if err := txOne.Commit(); err != nil {
		t.Fatal(err)
	}
	select {
	case err := <-transfer:
		t.Fatalf("ownership transfer ignored second live effect slot: %v", err)
	case <-time.After(100 * time.Millisecond):
	}
	if err := txTwo.Commit(); err != nil {
		t.Fatal(err)
	}
	select {
	case err := <-transfer:
		if err != nil {
			t.Fatal(err)
		}
	case <-time.After(2 * time.Second):
		t.Fatal("ownership transfer did not complete after all effect slots drained")
	}

	staleTx, staleCleanup, err := BeginPessimisticTx(t.Context(), db)
	if err != nil {
		t.Fatal(err)
	}
	defer staleCleanup()
	if err := authority.LockActiveFence(t.Context(), staleTx, keyOne); !errors.Is(err, ErrRuntimeWorkOwnershipLost) {
		t.Fatalf("stale effect fence = %v, want ownership loss", err)
	}
	_ = staleTx.Rollback()
}

func TestRecordEventClaimAdmissionAndFIFORealTiDB(t *testing.T) {
	if os.Getenv("TIDB_HOST") == "" {
		t.Skip("runtime_worker_integration requires the declared TiDB environment")
	}
	db, cleanup := openRuntimeWorkFenceScratchDB(t)
	defer cleanup()
	db.SetMaxOpenConns(32)
	scope := RuntimeWorkScope{
		OrgID: "org-runtime-fifo", AppInstallationID: "install-runtime-fifo",
		RuntimeCapabilityID: "runtime.records.mutate.v1",
	}
	repo, err := NewScopedRecordEventOutboxRepository(db, scope, time.Second)
	if err != nil {
		t.Fatal(err)
	}
	created := time.Now().UTC().Add(-time.Second)
	for index, event := range []struct {
		id, recordKey string
	}{
		{id: "event-same-older", recordKey: "Issue/issue-one"},
		{id: "event-same-younger", recordKey: "Issue/issue-one"},
		{id: "event-independent", recordKey: "Issue/issue-two"},
	} {
		if _, err := db.Exec(`INSERT INTO _System_RecordEventOutbox
(Id, EventType, Payload, Status, RetryCount, RecordKey, ExecutionScopeDigest, OrgId, AppInstallationId, RuntimeCapabilityId, CreatedDate, LastModifiedDate)
VALUES (?, 'record.updated', '{}', ?, 0, ?, ?, ?, ?, ?, ?, ?)`, event.id, constants.OutboxStatusPending,
			event.recordKey, repo.scopeDigest, scope.OrgID, scope.AppInstallationID, scope.RuntimeCapabilityID,
			created.Add(time.Duration(index)*time.Second), created); err != nil {
			t.Fatal(err)
		}
	}

	pending, err := repo.GetPendingEventsWithRecordFIFO(t.Context(), 10)
	if err != nil {
		t.Fatal(err)
	}
	if len(pending) != 2 || pending[0].ID != "event-same-older" || pending[1].ID != "event-independent" {
		t.Fatalf("FIFO-eligible events = %#v", pending)
	}

	batchTx, batchCleanup, err := BeginPessimisticTx(t.Context(), db)
	if err != nil {
		t.Fatal(err)
	}
	batchCtx := context.WithValue(t.Context(), txContextKey{}, batchTx)
	batchCtx = context.WithValue(batchCtx, pessimisticTxMarkerKey{}, true)
	batchEvents, batchClaims, err := repo.ClaimEventBatchWithRecordFIFO(batchCtx, 3)
	if err != nil {
		t.Fatal(err)
	}
	if len(batchEvents) != 2 || len(batchClaims) != 2 ||
		batchEvents[0].ID != "event-same-older" || batchEvents[1].ID != "event-independent" ||
		batchClaims[0].EventID != batchEvents[0].ID || batchClaims[1].EventID != batchEvents[1].ID ||
		batchEvents[0].RecordKey == batchEvents[1].RecordKey {
		t.Fatalf("same-transaction FIFO set claim = events:%+v claims:%+v, want one head from each independent key", batchEvents, batchClaims)
	}
	if nextEvents, nextClaims, err := repo.ClaimEventBatchWithRecordFIFO(batchCtx, 1); err != nil || len(nextEvents) != 0 || len(nextClaims) != 0 {
		t.Fatalf("next same-transaction FIFO batch = events:%+v claims:%+v err:%v, younger same-key row must remain blocked", nextEvents, nextClaims, err)
	}
	if err := batchTx.Rollback(); err != nil {
		t.Fatal(err)
	}
	batchCleanup()

	lockedTx, lockedCleanup, err := BeginPessimisticTx(t.Context(), db)
	if err != nil {
		t.Fatal(err)
	}
	defer lockedCleanup()
	lockedCtx := context.WithValue(t.Context(), txContextKey{}, lockedTx)
	if err := repo.LockOutboxClaimCandidate(lockedCtx, "event-same-older"); err != nil {
		t.Fatal(err)
	}
	contenderTx, contenderCleanup, err := BeginPessimisticTx(t.Context(), db)
	if err != nil {
		t.Fatal(err)
	}
	defer contenderCleanup()
	contenderCtx := context.WithValue(t.Context(), txContextKey{}, contenderTx)
	started := time.Now()
	if err := repo.LockOutboxClaimCandidate(contenderCtx, "event-same-older"); !errors.Is(err, ErrOutboxEventAlreadyClaimed) {
		t.Fatalf("NOWAIT contender = %v", err)
	}
	if elapsed := time.Since(started); elapsed > time.Second {
		t.Fatalf("NOWAIT contender blocked for %s", elapsed)
	}
	_ = contenderTx.Rollback()
	_ = lockedTx.Rollback()

	if _, err := db.Exec(`UPDATE _System_RecordEventOutbox
SET Status = 'processing', LastModifiedDate = TIMESTAMPADD(SECOND, -2, CURRENT_TIMESTAMP(6))
WHERE Id = 'event-same-older'`); err != nil {
		t.Fatal(err)
	}
	effectTx, effectCleanup, err := BeginPessimisticTx(t.Context(), db)
	if err != nil {
		t.Fatal(err)
	}
	effectCtx := context.WithValue(t.Context(), txContextKey{}, effectTx)
	if err := repo.LockOutboxClaimCandidate(effectCtx, "event-same-older"); err != nil {
		t.Fatal(err)
	}

	claimTx, claimCleanup, err := BeginPessimisticTx(t.Context(), db)
	if err != nil {
		t.Fatal(err)
	}
	claimCtx := context.WithValue(t.Context(), txContextKey{}, claimTx)
	claimCtx = context.WithValue(claimCtx, pessimisticTxMarkerKey{}, true)
	event, claim, found, err := repo.ClaimNextEventWithRecordFIFO(claimCtx)
	if err != nil {
		t.Fatal(err)
	}
	if !found || event.ID != "event-independent" || claim.EventID != event.ID || claim.RecordKey != "Issue/issue-two" || claim.Token == "" {
		t.Fatalf("claim while FIFO head locked = found:%t event:%+v claim:%+v, want independent event", found, event, claim)
	}
	if err := claimTx.Commit(); err != nil {
		t.Fatal(err)
	}
	claimCleanup()

	blockedTx, blockedCleanup, err := BeginPessimisticTx(t.Context(), db)
	if err != nil {
		t.Fatal(err)
	}
	blockedCtx := context.WithValue(t.Context(), txContextKey{}, blockedTx)
	blockedCtx = context.WithValue(blockedCtx, pessimisticTxMarkerKey{}, true)
	started = time.Now()
	_, _, found, err = repo.ClaimNextEventWithRecordFIFO(blockedCtx)
	if err != nil {
		t.Fatal(err)
	}
	if found {
		t.Fatal("stale live FIFO head became claimable while its effect row remained locked")
	}
	if elapsed := time.Since(started); elapsed > time.Second {
		t.Fatalf("stale live FIFO claim blocked for %s", elapsed)
	}
	_ = blockedTx.Rollback()
	blockedCleanup()

	_ = effectTx.Rollback()
	effectCleanup()

	nextTx, nextCleanup, err := BeginPessimisticTx(t.Context(), db)
	if err != nil {
		t.Fatal(err)
	}
	nextCtx := context.WithValue(t.Context(), txContextKey{}, nextTx)
	nextCtx = context.WithValue(nextCtx, pessimisticTxMarkerKey{}, true)
	event, claim, found, err = repo.ClaimNextEventWithRecordFIFO(nextCtx)
	if err != nil {
		t.Fatal(err)
	}
	if !found || event.ID != "event-same-older" || claim.EventID != event.ID || claim.RecordKey != "Issue/issue-one" {
		t.Fatalf("claim after FIFO head unlock = found:%t event:%+v claim:%+v, want oldest same-record event", found, event, claim)
	}
	if err := nextTx.Commit(); err != nil {
		t.Fatal(err)
	}
	nextCleanup()

	control, err := NewRuntimeWorkOwnershipControl(db)
	if err != nil {
		t.Fatal(err)
	}
	owner, err := control.Activate(t.Context(), RuntimeWorkOwnershipActivation{
		OperationID: "activate-record-claim-lanes", WorkFamily: "runtime.record-event-dispatch.v1",
		DeploymentID:        "runtime-record-claim-lanes",
		WorkerReleaseDigest: "sha256:cccccccccccccccccccccccccccccccccccccccccccccccccccccccccccccccc",
	})
	if err != nil {
		t.Fatal(err)
	}
	authority, err := NewRuntimeWorkOwnershipRepository(db, owner)
	if err != nil {
		t.Fatal(err)
	}
	for index := 0; index < 20; index++ {
		id := fmt.Sprintf("event-lane-%02d", index)
		if _, err := db.Exec(`INSERT INTO _System_RecordEventOutbox
(Id, EventType, Payload, Status, RetryCount, RecordKey, ExecutionScopeDigest, OrgId, AppInstallationId, RuntimeCapabilityId, CreatedDate, LastModifiedDate)
VALUES (?, 'record.updated', '{}', ?, 0, ?, ?, ?, ?, ?, ?, ?)`, id, constants.OutboxStatusPending,
			"Issue/"+id, repo.scopeDigest, scope.OrgID, scope.AppInstallationID, scope.RuntimeCapabilityID,
			created.Add(time.Second+time.Duration(index)*time.Millisecond), created); err != nil {
			t.Fatal(err)
		}
	}

	type laneResult struct {
		eventID string
		err     error
	}
	results := make(chan laneResult, 15)
	start := make(chan struct{})
	var lanes sync.WaitGroup
	for lane := 0; lane < 15; lane++ {
		lanes.Add(1)
		go func() {
			defer lanes.Done()
			<-start
			tx, txCleanup, err := BeginPessimisticTx(t.Context(), db)
			if err != nil {
				results <- laneResult{err: err}
				return
			}
			defer txCleanup()
			txCtx := context.WithValue(t.Context(), txContextKey{}, tx)
			txCtx = context.WithValue(txCtx, pessimisticTxMarkerKey{}, true)
			if err := authority.LockActiveContext(txCtx); err != nil {
				_ = tx.Rollback()
				results <- laneResult{err: err}
				return
			}
			event, _, found, err := repo.ClaimNextEventWithRecordFIFO(txCtx)
			if err != nil {
				_ = tx.Rollback()
				results <- laneResult{err: err}
				return
			}
			if !found {
				_ = tx.Rollback()
				results <- laneResult{err: errors.New("lane found no FIFO event")}
				return
			}
			if err := tx.Commit(); err != nil {
				results <- laneResult{err: err}
				return
			}
			results <- laneResult{eventID: event.ID}
		}()
	}
	close(start)
	lanes.Wait()
	close(results)
	claimed := map[string]struct{}{}
	for result := range results {
		if result.err != nil {
			t.Fatal(result.err)
		}
		if _, duplicate := claimed[result.eventID]; duplicate {
			t.Fatalf("15 concurrent lanes claimed duplicate event %s", result.eventID)
		}
		claimed[result.eventID] = struct{}{}
	}
	if len(claimed) != 15 {
		t.Fatalf("unique lane claims = %d, want 15", len(claimed))
	}
}

func TestOutboxLifecycleOutcomesOnTiDB(t *testing.T) {
	if os.Getenv("TIDB_HOST") == "" {
		t.Skip("runtime_worker_integration requires the declared TiDB environment")
	}
	db, cleanup := openRuntimeWorkFenceScratchDB(t)
	defer cleanup()

	legacy := NewOutboxRepository(db)
	savepointTx, err := db.BeginTx(t.Context(), nil)
	if err != nil {
		t.Fatal(err)
	}
	savepointContext := context.WithValue(t.Context(), txContextKey{}, savepointTx)
	if err := legacy.CreateOutboxAttemptSavepoint(savepointContext); err != nil {
		t.Fatalf("create outbox savepoint: %v", err)
	}
	if _, err := savepointTx.ExecContext(t.Context(), `INSERT INTO _System_OutboxEvent
(Id, EventType, Payload, Status, RetryCount, CreatedDate, LastModifiedDate)
VALUES ('savepoint-rolled-back', 'record.updated', '{}', 'pending', 0, CURRENT_TIMESTAMP(6), CURRENT_TIMESTAMP(6))`); err != nil {
		t.Fatal(err)
	}
	if err := legacy.RollbackToOutboxAttemptSavepoint(savepointContext); err != nil {
		t.Fatalf("rollback outbox savepoint: %v", err)
	}
	if err := legacy.ReleaseOutboxAttemptSavepoint(savepointContext); err != nil {
		t.Fatalf("release outbox savepoint: %v", err)
	}
	if err := savepointTx.Commit(); err != nil {
		t.Fatal(err)
	}
	var rolledBackRows int
	if err := db.QueryRow(`SELECT COUNT(*) FROM _System_OutboxEvent WHERE Id = 'savepoint-rolled-back'`).Scan(&rolledBackRows); err != nil {
		t.Fatal(err)
	}
	if rolledBackRows != 0 {
		t.Fatalf("savepoint rollback left %d row(s)", rolledBackRows)
	}

	firstID, err := legacy.Enqueue(t.Context(), "record.updated", map[string]interface{}{
		"public_payload": map[string]interface{}{
			"object_api_name": "Issue",
			"record":          models.SObject{"Id": "issue-retry"},
		},
	})
	if err != nil {
		t.Fatalf("enqueue retry event: %v", err)
	}
	secondID, err := legacy.Enqueue(t.Context(), "record.updated", map[string]interface{}{
		"public_payload": map[string]interface{}{
			"object_api_name": "Issue",
			"record":          models.SObject{"Id": "issue-complete"},
		},
	})
	if err != nil {
		t.Fatalf("enqueue completion event: %v", err)
	}

	var firstRecordKey, firstStatus string
	if err := db.QueryRow(`SELECT RecordKey, Status FROM _System_OutboxEvent WHERE Id = ?`, firstID).
		Scan(&firstRecordKey, &firstStatus); err != nil {
		t.Fatal(err)
	}
	if firstRecordKey != "Issue/issue-retry" || firstStatus != string(constants.OutboxStatusPending) {
		t.Fatalf("persisted retry event = key:%q status:%q", firstRecordKey, firstStatus)
	}
	pending, err := legacy.GetPendingEvents(t.Context(), 10)
	if err != nil {
		t.Fatalf("read pending events: %v", err)
	}
	pendingIDs := map[string]bool{}
	for _, event := range pending {
		pendingIDs[event.ID] = true
	}
	if len(pending) != 2 || !pendingIDs[firstID] || !pendingIDs[secondID] {
		t.Fatalf("pending events = %#v, want both enqueued events", pending)
	}

	claim, err := legacy.ClaimEvent(t.Context(), firstID)
	if err != nil {
		t.Fatalf("claim retry event: %v", err)
	}
	if err := legacy.IncrementRetry(t.Context(), claim, "temporary failure"); err != nil {
		t.Fatalf("increment retry: %v", err)
	}
	if _, err := legacy.ClaimEvent(t.Context(), firstID); !errors.Is(err, ErrOutboxEventAlreadyClaimed) {
		t.Fatalf("retry became visible before database-owned delay: %v", err)
	}
	if _, err := db.Exec(`UPDATE _System_OutboxEvent
SET LastModifiedDate = TIMESTAMPADD(SECOND, -2, CURRENT_TIMESTAMP(6)) WHERE Id = ?`, firstID); err != nil {
		t.Fatal(err)
	}
	claim, err = legacy.ClaimEvent(t.Context(), firstID)
	if err != nil {
		t.Fatalf("reclaim retry event: %v", err)
	}
	if claim.RetryCount != 1 {
		t.Fatalf("durable retry count = %d, want 1", claim.RetryCount)
	}
	if err := legacy.MarkFailed(t.Context(), claim, "terminal failure"); err != nil {
		t.Fatalf("mark failed: %v", err)
	}
	var failedStatus, failedMessage string
	var failedRetries int
	if err := db.QueryRow(`SELECT Status, RetryCount, ErrorMessage FROM _System_OutboxEvent WHERE Id = ?`, firstID).
		Scan(&failedStatus, &failedRetries, &failedMessage); err != nil {
		t.Fatal(err)
	}
	if failedStatus != string(constants.OutboxStatusFailed) || failedRetries != 2 || failedMessage != "terminal failure" {
		t.Fatalf("terminal outcome = status:%q retries:%d message:%q", failedStatus, failedRetries, failedMessage)
	}

	claim, err = legacy.ClaimEvent(t.Context(), secondID)
	if err != nil {
		t.Fatalf("claim completion event: %v", err)
	}
	if err := legacy.UpdateStatus(t.Context(), claim, string(constants.OutboxStatusProcessed), ""); err != nil {
		t.Fatalf("complete event: %v", err)
	}
	if _, err := db.Exec(`UPDATE _System_OutboxEvent
SET ProcessedDate = TIMESTAMPADD(DAY, -2, CURRENT_TIMESTAMP(6)) WHERE Id = ?`, secondID); err != nil {
		t.Fatal(err)
	}
	if _, err := db.Exec(`INSERT INTO _System_OutboxEvent
(Id, EventType, Payload, Status, RetryCount, ProcessedDate, CreatedDate, LastModifiedDate)
VALUES ('cleanup-younger', 'record.updated', '{}', 'processed', 0,
TIMESTAMPADD(HOUR, -36, CURRENT_TIMESTAMP(6)), CURRENT_TIMESTAMP(6), CURRENT_TIMESTAMP(6))`); err != nil {
		t.Fatal(err)
	}
	deleted, err := legacy.CleanupProcessedBatch(t.Context(), 24*time.Hour, 1)
	if err != nil || deleted != 1 {
		t.Fatalf("cleanup processed event = %d, %v; want 1, nil", deleted, err)
	}
	var olderRemaining, youngerRemaining int
	if err := db.QueryRow(`SELECT COUNT(*) FROM _System_OutboxEvent WHERE Id = ?`, secondID).Scan(&olderRemaining); err != nil {
		t.Fatal(err)
	}
	if err := db.QueryRow(`SELECT COUNT(*) FROM _System_OutboxEvent WHERE Id = 'cleanup-younger'`).Scan(&youngerRemaining); err != nil {
		t.Fatal(err)
	}
	if olderRemaining != 0 || youngerRemaining != 1 {
		t.Fatalf("bounded cleanup left older:%d younger:%d, want 0/1", olderRemaining, youngerRemaining)
	}

	scope := recordEventTestScope()
	scoped, err := NewScopedRecordEventOutboxRepository(db, scope, 3*time.Second)
	if err != nil {
		t.Fatal(err)
	}
	scopedContext, err := WithRuntimeWorkScope(t.Context(), scope)
	if err != nil {
		t.Fatal(err)
	}
	scopedID, err := scoped.Enqueue(scopedContext, "record.updated", map[string]interface{}{
		"public_payload": map[string]interface{}{
			"object_api_name": "Issue",
			"record":          models.SObject{"Id": "issue-scoped"},
		},
	})
	if err != nil {
		t.Fatalf("enqueue scoped event: %v", err)
	}
	var recordKey, scopeDigest, orgID, installationID, capabilityID string
	if err := db.QueryRow(`SELECT RecordKey, ExecutionScopeDigest, OrgId, AppInstallationId, RuntimeCapabilityId
FROM _System_RecordEventOutbox WHERE Id = ?`, scopedID).
		Scan(&recordKey, &scopeDigest, &orgID, &installationID, &capabilityID); err != nil {
		t.Fatal(err)
	}
	wantDigest, err := scope.Digest()
	if err != nil {
		t.Fatal(err)
	}
	if recordKey != "Issue/issue-scoped" || scopeDigest != wantDigest || orgID != scope.OrgID ||
		installationID != scope.AppInstallationID || capabilityID != scope.RuntimeCapabilityID {
		t.Fatalf("scoped persistence = key:%q digest:%q org:%q installation:%q capability:%q",
			recordKey, scopeDigest, orgID, installationID, capabilityID)
	}

	scopedClaim, err := scoped.ClaimEvent(t.Context(), scopedID)
	if err != nil {
		t.Fatalf("claim scoped event: %v", err)
	}
	if scopedClaim.Token == "" {
		t.Fatal("scoped claim did not persist a fencing token")
	}
	if err := scoped.RefreshOutboxProcessingClaim(t.Context(), scopedClaim); err != nil {
		t.Fatalf("refresh live scoped claim: %v", err)
	}
	if _, err := db.Exec(`UPDATE _System_RecordEventOutbox
SET LastModifiedDate = TIMESTAMPADD(SECOND, -4, CURRENT_TIMESTAMP(6)) WHERE Id = ?`, scopedID); err != nil {
		t.Fatal(err)
	}
	if err := scoped.RefreshOutboxProcessingClaim(t.Context(), scopedClaim); !errors.Is(err, ErrOutboxEventAlreadyClaimed) {
		t.Fatalf("refresh expired scoped claim = %v, want claim fencing", err)
	}
	if _, err := db.Exec(`UPDATE _System_RecordEventOutbox
SET LastModifiedDate = CURRENT_TIMESTAMP(6) WHERE Id = ?`, scopedID); err != nil {
		t.Fatal(err)
	}
	staleClaim := scopedClaim
	staleClaim.Token = "00000000-0000-4000-8000-000000000000"
	if err := scoped.UpdateStatus(t.Context(), staleClaim, string(constants.OutboxStatusProcessed), ""); !errors.Is(err, ErrOutboxEventAlreadyClaimed) {
		t.Fatalf("stale completion = %v, want claim fencing", err)
	}
	if err := scoped.IncrementRetry(t.Context(), scopedClaim, "retry scoped event"); err != nil {
		t.Fatalf("current scoped retry: %v", err)
	}
	if err := scoped.RefreshOutboxProcessingClaim(t.Context(), scopedClaim); !errors.Is(err, ErrOutboxEventAlreadyClaimed) {
		t.Fatalf("refresh pending scoped event = %v, want processing-status fence", err)
	}
	if _, err := db.Exec(`UPDATE _System_RecordEventOutbox
SET LastModifiedDate = TIMESTAMPADD(SECOND, -2, CURRENT_TIMESTAMP(6)) WHERE Id = ?`, scopedID); err != nil {
		t.Fatal(err)
	}
	reclaimed, err := scoped.ClaimEvent(t.Context(), scopedID)
	if err != nil {
		t.Fatalf("reclaim scoped event: %v", err)
	}
	if reclaimed.Token == scopedClaim.Token || reclaimed.RetryCount != 1 {
		t.Fatalf("reclaimed claim = %+v, want rotated token and durable retry count 1", reclaimed)
	}
	if err := scoped.MarkFailed(t.Context(), scopedClaim, "stale failure"); !errors.Is(err, ErrOutboxEventAlreadyClaimed) {
		t.Fatalf("stale terminal transition = %v, want claim fencing", err)
	}
	if err := scoped.MarkFailed(t.Context(), reclaimed, "current terminal failure"); err != nil {
		t.Fatalf("current terminal transition: %v", err)
	}
	var scopedStatus, scopedMessage string
	var scopedRetries int
	if err := db.QueryRow(`SELECT Status, RetryCount, ErrorMessage FROM _System_RecordEventOutbox WHERE Id = ?`, scopedID).
		Scan(&scopedStatus, &scopedRetries, &scopedMessage); err != nil {
		t.Fatal(err)
	}
	if scopedStatus != string(constants.OutboxStatusFailed) || scopedRetries != 2 || scopedMessage != "current terminal failure" {
		t.Fatalf("scoped terminal outcome = status:%q retries:%d message:%q", scopedStatus, scopedRetries, scopedMessage)
	}

	otherScope := RuntimeWorkScope{
		OrgID: "org-other", AppInstallationID: "install-other",
		RuntimeCapabilityID: scope.RuntimeCapabilityID,
	}
	otherDigest, err := otherScope.Digest()
	if err != nil {
		t.Fatal(err)
	}
	for _, row := range []struct {
		id, digest, org, installation, recordID string
	}{
		{id: "backfill-owned", digest: wantDigest, org: scope.OrgID, installation: scope.AppInstallationID, recordID: "issue-owned"},
		{id: "backfill-other", digest: otherDigest, org: otherScope.OrgID, installation: otherScope.AppInstallationID, recordID: "issue-other"},
	} {
		payload := fmt.Sprintf(`{"public_payload":{"object_api_name":"Issue","record":{"Id":%q}}}`, row.recordID)
		if _, err := db.Exec(`INSERT INTO _System_RecordEventOutbox
(Id, EventType, Payload, Status, RetryCount, RecordKey, ExecutionScopeDigest, OrgId, AppInstallationId, RuntimeCapabilityId, CreatedDate, LastModifiedDate)
VALUES (?, 'record.updated', ?, 'pending', 0, NULL, ?, ?, ?, ?, CURRENT_TIMESTAMP(6), CURRENT_TIMESTAMP(6))`,
			row.id, payload, row.digest, row.org, row.installation, scope.RuntimeCapabilityID); err != nil {
			t.Fatal(err)
		}
	}
	updated, err := scoped.BackfillRecordKeys(t.Context(), 10)
	if err != nil || updated != 1 {
		t.Fatalf("scoped backfill = %d, %v; want 1, nil", updated, err)
	}
	var ownedKey string
	var otherKey sql.NullString
	if err := db.QueryRow(`SELECT RecordKey FROM _System_RecordEventOutbox WHERE Id = 'backfill-owned'`).Scan(&ownedKey); err != nil {
		t.Fatal(err)
	}
	if err := db.QueryRow(`SELECT RecordKey FROM _System_RecordEventOutbox WHERE Id = 'backfill-other'`).Scan(&otherKey); err != nil {
		t.Fatal(err)
	}
	if ownedKey != "Issue/issue-owned" || otherKey.Valid {
		t.Fatalf("scoped backfill keys = owned:%q other:%+v", ownedKey, otherKey)
	}
}

func TestObserveOutboxWorkStatusRangesRealTiDB(t *testing.T) {
	if os.Getenv("TIDB_HOST") == "" {
		t.Skip("runtime_worker_integration requires the declared TiDB environment")
	}
	db, cleanup := openRuntimeWorkFenceScratchDB(t)
	defer cleanup()
	scope := RuntimeWorkScope{
		OrgID: "org-observe", AppInstallationID: "install-observe",
		RuntimeCapabilityID: "runtime.records.mutate.v1",
	}
	repo, err := NewScopedRecordEventOutboxRepository(db, scope, time.Second)
	if err != nil {
		t.Fatal(err)
	}
	created := time.Now().UTC().Add(-10 * time.Second).Truncate(time.Second)
	statuses := []string{
		string(constants.OutboxStatusPending), outboxStatusProcessing,
		string(constants.OutboxStatusPending), string(constants.OutboxStatusFailed),
		string(constants.OutboxStatusFailed), string(constants.OutboxStatusProcessed),
	}
	for index, status := range statuses {
		if _, err := db.Exec(`INSERT INTO _System_RecordEventOutbox
(Id, EventType, Payload, Status, RetryCount, RecordKey, ExecutionScopeDigest, OrgId, AppInstallationId, RuntimeCapabilityId, CreatedDate, LastModifiedDate)
VALUES (?, 'record.updated', '{}', ?, 0, ?, ?, ?, ?, ?, ?, ?)`,
			fmt.Sprintf("observe-%d", index), status, fmt.Sprintf("Issue/observe-%d", index),
			repo.scopeDigest, scope.OrgID, scope.AppInstallationID, scope.RuntimeCapabilityID,
			created.Add(time.Duration(index)*time.Second), created); err != nil {
			t.Fatal(err)
		}
	}
	if _, err := db.Exec(`INSERT INTO _System_RecordEventOutbox
(Id, EventType, Payload, Status, RetryCount, RecordKey, ExecutionScopeDigest, OrgId, AppInstallationId, RuntimeCapabilityId, CreatedDate, LastModifiedDate)
VALUES ('observe-other-scope', 'record.updated', '{}', 'pending', 0, 'Issue/other', ?, 'other-org', 'other-install', 'other-capability', ?, ?)`,
		strings.Repeat("f", 64), created.Add(-time.Minute), created); err != nil {
		t.Fatal(err)
	}

	depth, failed, oldest, err := repo.ObserveOutboxWork(t.Context())
	if err != nil {
		t.Fatal(err)
	}
	if depth != 3 || failed != 2 {
		t.Fatalf("observation = depth:%d failed:%d, want 3/2", depth, failed)
	}
	if oldest < 9*time.Second || oldest > 20*time.Second {
		t.Fatalf("oldest age = %s, want approximately 10s from database time", oldest)
	}
}

func TestRecordEventWorkHintIgnoresProcessedHistoryRealTiDB(t *testing.T) {
	if os.Getenv("TIDB_HOST") == "" {
		t.Skip("runtime_worker_integration requires the declared TiDB environment")
	}
	db, cleanup := openRuntimeWorkFenceScratchDB(t)
	defer cleanup()
	scope := RuntimeWorkScope{
		OrgID: "org-hint-plan", AppInstallationID: "install-hint-plan",
		RuntimeCapabilityID: "runtime.records.mutate.v1",
	}
	repo, err := NewScopedRecordEventOutboxRepository(db, scope, time.Second)
	if err != nil {
		t.Fatal(err)
	}

	tx, err := db.BeginTx(t.Context(), nil)
	if err != nil {
		t.Fatal(err)
	}
	insert, err := tx.PrepareContext(t.Context(), `INSERT INTO _System_RecordEventOutbox
(Id, EventType, Payload, Status, RetryCount, RecordKey, ExecutionScopeDigest, OrgId, AppInstallationId, RuntimeCapabilityId, CreatedDate, LastModifiedDate)
VALUES (?, 'record.updated', '{}', 'processed', 0, ?, ?, ?, ?, ?, CURRENT_TIMESTAMP(6), CURRENT_TIMESTAMP(6))`)
	if err != nil {
		_ = tx.Rollback()
		t.Fatal(err)
	}
	for index := 0; index < 3000; index++ {
		id := fmt.Sprintf("processed-%04d", index)
		if _, err := insert.ExecContext(t.Context(), id, "Issue/"+id, repo.scopeDigest, scope.OrgID, scope.AppInstallationID, scope.RuntimeCapabilityID); err != nil {
			_ = insert.Close()
			_ = tx.Rollback()
			t.Fatal(err)
		}
	}
	if err := insert.Close(); err != nil {
		_ = tx.Rollback()
		t.Fatal(err)
	}
	if err := tx.Commit(); err != nil {
		t.Fatal(err)
	}

	present, err := repo.HasOutboxWorkHint(t.Context())
	if err != nil || present {
		t.Fatalf("HasOutboxWorkHint() = %t, %v; want false, nil", present, err)
	}
	queryText, queryArgs := repo.scopedRecordEventWorkHintQuery()
	if strings.Count(queryText, "FORCE INDEX (`"+recordEventClaimIndex+"`)") != 2 {
		t.Fatalf("work hint SQL does not require %s: %s", recordEventClaimIndex, queryText)
	}
	emptyPlan := explainAnalyzeText(t, db, queryText, queryArgs...)
	if strings.Contains(emptyPlan, "idx_record_event_outbox_runtime_scope") {
		t.Fatalf("empty work hint plan regressed to scope-history index:\n%s", emptyPlan)
	}
	if strings.Contains(emptyPlan, "TableFullScan") {
		t.Fatalf("empty work hint plan scans processed history:\n%s", emptyPlan)
	}

	if _, err := db.Exec(`INSERT INTO _System_RecordEventOutbox
(Id, EventType, Payload, Status, RetryCount, RecordKey, ExecutionScopeDigest, OrgId, AppInstallationId, RuntimeCapabilityId, CreatedDate, LastModifiedDate)
VALUES ('pending-one', 'record.updated', '{}', 'pending', 0, 'Issue/pending-one', ?, ?, ?, ?, CURRENT_TIMESTAMP(6), CURRENT_TIMESTAMP(6))`,
		repo.scopeDigest, scope.OrgID, scope.AppInstallationID, scope.RuntimeCapabilityID); err != nil {
		t.Fatal(err)
	}
	positivePlan := explainAnalyzeText(t, db, queryText, queryArgs...)
	if strings.Contains(positivePlan, "idx_record_event_outbox_runtime_scope") || strings.Contains(positivePlan, "TableFullScan") {
		t.Fatalf("positive work hint plan scans scope history:\n%s", positivePlan)
	}
}

func explainAnalyzeText(t *testing.T, db *sql.DB, queryText string, args ...interface{}) string {
	t.Helper()
	rows, err := db.QueryContext(t.Context(), "EXPLAIN ANALYZE "+queryText, args...)
	if err != nil {
		t.Fatal(err)
	}
	defer rows.Close()
	columns, err := rows.Columns()
	if err != nil {
		t.Fatal(err)
	}
	var plan strings.Builder
	for rows.Next() {
		values := make([]sql.RawBytes, len(columns))
		destinations := make([]interface{}, len(columns))
		for index := range values {
			destinations[index] = &values[index]
		}
		if err := rows.Scan(destinations...); err != nil {
			t.Fatal(err)
		}
		for _, value := range values {
			plan.Write(value)
			plan.WriteByte('\t')
		}
		plan.WriteByte('\n')
	}
	if err := rows.Err(); err != nil {
		t.Fatal(err)
	}
	return plan.String()
}

func openRuntimeWorkFenceScratchDB(t *testing.T) (*sql.DB, func()) {
	t.Helper()
	host := os.Getenv("TIDB_HOST")
	port := 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_runtime_fence_%d", time.Now().UnixNano())
	if _, err := admin.Exec("CREATE DATABASE `" + name + "`"); err != nil {
		_ = admin.Close()
		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(8)
	definitions, err := schemacatalog.SystemTableDefinitions()
	if err != nil {
		t.Fatalf("load system table catalog: %v", err)
	}
	wanted := map[string]bool{
		"_System_OutboxEvent":          false,
		"_System_RecordEventOutbox":    false,
		"_System_RuntimeWorkFence":     false,
		"_System_RuntimeWorkOwnership": false,
	}
	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 from system table catalog: %v", definition.TableName, err)
		}
		wanted[definition.TableName] = true
	}
	for table, found := range wanted {
		if !found {
			t.Fatalf("system table catalog does not declare %s", table)
		}
	}
	return db, func() {
		_ = db.Close()
		_, _ = admin.Exec("DROP DATABASE IF EXISTS `" + name + "`")
		_ = admin.Close()
	}
}