package persistence

import (
	"testing"
	"time"
)

func TestPermissionAuthoringCommandLifecycleBoundsHistoricalAndTerminalRowsOnTiDB(t *testing.T) {
	db, cleanup := openSessionRepositoryTiDB(t)
	defer cleanup()
	for _, statement := range []string{
		`CREATE TABLE _System_RuntimeWorkOwnership (
			WorkFamily VARCHAR(64) PRIMARY KEY,
			DeploymentId VARCHAR(255) NOT NULL,
			WorkerReleaseDigest CHAR(71) NOT NULL,
			OwnershipEpoch DECIMAL(20,0) NOT NULL,
			IsActive TINYINT(1) NOT NULL DEFAULT 1,
			OperationId VARCHAR(255) NOT NULL,
			CreatedDate DATETIME(6) NOT NULL DEFAULT CURRENT_TIMESTAMP(6),
			LastModifiedDate DATETIME(6) NOT NULL DEFAULT CURRENT_TIMESTAMP(6)
		)`,
		`CREATE TABLE _System_PermissionAuthoringCommand (
			Id VARCHAR(255) PRIMARY KEY,
			ScopeDigest CHAR(64) NOT NULL,
			CommandId VARCHAR(255) NOT NULL,
			RequestDigest CHAR(71) NOT NULL,
			TargetKind VARCHAR(32) NOT NULL,
			TargetId VARCHAR(255) NOT NULL,
			CanonicalPermissionSetId VARCHAR(255) NULL,
			ExpectedAuthoringToken VARCHAR(255) NOT NULL,
			State VARCHAR(16) NOT NULL,
			ResultPayload MEDIUMTEXT NULL,
			AdmissionExpiresAt DATETIME(6) NOT NULL,
			OutcomeExpiresAt DATETIME(6) NULL,
			OutcomeRetentionMicros BIGINT NOT NULL,
			CreatedDate DATETIME(6) NOT NULL DEFAULT CURRENT_TIMESTAMP(6),
			LastModifiedDate DATETIME(6) NOT NULL DEFAULT CURRENT_TIMESTAMP(6),
			UNIQUE KEY ux_permission_authoring_command_scope (ScopeDigest, CommandId),
			KEY idx_permission_authoring_command_admission_expiry (State, AdmissionExpiresAt, Id),
			KEY idx_permission_authoring_command_outcome_expiry (State, OutcomeExpiresAt, Id)
		)`,
	} {
		if _, err := db.Exec(statement); err != nil {
			t.Fatalf("create lifecycle fixture: %v", err)
		}
	}
	owner := permissionAuthoringLifecycleOwnership()
	if _, err := db.Exec(`INSERT INTO _System_RuntimeWorkOwnership
(WorkFamily, DeploymentId, WorkerReleaseDigest, OwnershipEpoch, IsActive, OperationId)
VALUES (?, ?, ?, ?, 1, 'activate-permission-authoring-lifecycle')`,
		owner.WorkFamily, owner.DeploymentID, owner.WorkerReleaseDigest, owner.OwnershipEpoch); err != nil {
		t.Fatal(err)
	}
	for _, row := range []struct {
		id        string
		state     string
		canonical any
		admission string
		outcome   any
	}{
		{"accepted-expired", PermissionAuthoringCommandAccepted, nil, "2000-01-01 00:00:00", nil},
		{"bound-accepted-expired", PermissionAuthoringCommandAccepted, "0PS-target", "2000-01-01 00:00:00", nil},
		{"completed-expired", PermissionAuthoringCommandCompleted, "0PS-target", "2000-01-01 00:00:00", "2000-01-02 00:00:00"},
		{"bound-accepted-live", PermissionAuthoringCommandAccepted, "0PS-target", "2999-01-01 00:00:00", nil},
		{"completed-live", PermissionAuthoringCommandCompleted, "0PS-target", "2000-01-01 00:00:00", "2999-01-01 00:00:00"},
	} {
		if _, err := db.Exec(`INSERT INTO _System_PermissionAuthoringCommand
(Id, ScopeDigest, CommandId, RequestDigest, TargetKind, TargetId, CanonicalPermissionSetId,
 ExpectedAuthoringToken, State, ResultPayload, AdmissionExpiresAt, OutcomeExpiresAt, OutcomeRetentionMicros)
VALUES (?, ?, ?, ?, 'PERMISSION_SET', '0PS-target', ?, 'ps:v1', ?, ?, ?, ?, 3600000000)`,
			row.id,
			"aaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaa",
			row.id,
			"sha256:bbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbb",
			row.canonical, row.state, func() any {
				if row.state == PermissionAuthoringCommandCompleted {
					return `{"outcome":"NOOP"}`
				}
				return nil
			}(), row.admission, row.outcome,
		); err != nil {
			t.Fatalf("insert %s: %v", row.id, err)
		}
	}

	store, err := NewPermissionAuthoringCommandLifecycleStore(db, owner)
	if err != nil {
		t.Fatal(err)
	}
	result, err := store.CleanupExpired(t.Context(), 10)
	if err != nil {
		t.Fatal(err)
	}
	if result.Abandoned != 2 || result.Retired != 1 {
		t.Fatalf("CleanupExpired() = %+v", result)
	}
	var remaining int
	if err := db.QueryRow(`SELECT COUNT(*) FROM _System_PermissionAuthoringCommand`).Scan(&remaining); err != nil {
		t.Fatal(err)
	}
	if remaining != 2 {
		t.Fatalf("remaining command rows = %d, want two live rows", remaining)
	}

	// An execution transaction owns the same command row lock as cleanup. Even
	// after its recovery deadline passes, cleanup must wait and then observe the
	// freshly committed terminal deadline instead of deleting the command.
	if _, err := db.Exec(`UPDATE _System_PermissionAuthoringCommand
SET AdmissionExpiresAt = '2000-01-01 00:00:00'
WHERE Id = 'bound-accepted-live'`); err != nil {
		t.Fatal(err)
	}
	executionTx, executionCleanup, err := BeginPessimisticTx(t.Context(), db)
	if err != nil {
		t.Fatal(err)
	}
	defer executionCleanup()
	var state string
	if err := executionTx.QueryRowContext(t.Context(), `SELECT State
FROM _System_PermissionAuthoringCommand WHERE Id = 'bound-accepted-live' FOR UPDATE`).Scan(&state); err != nil {
		t.Fatal(err)
	}
	if state != PermissionAuthoringCommandAccepted {
		t.Fatalf("locked command state = %s", state)
	}
	if _, err := executionTx.ExecContext(t.Context(), `UPDATE _System_PermissionAuthoringCommand
SET State = 'COMPLETED', ResultPayload = '{"outcome":"NOOP"}', OutcomeExpiresAt = '2999-01-01 00:00:00'
WHERE Id = 'bound-accepted-live'`); err != nil {
		t.Fatal(err)
	}
	type cleanupOutcome struct {
		result PermissionAuthoringCommandCleanupResult
		err    error
	}
	cleanupDone := make(chan cleanupOutcome, 1)
	go func() {
		result, cleanupErr := store.CleanupExpired(t.Context(), 10)
		cleanupDone <- cleanupOutcome{result: result, err: cleanupErr}
	}()
	select {
	case outcome := <-cleanupDone:
		t.Fatalf("cleanup crossed live command row lock: %+v, %v", outcome.result, outcome.err)
	case <-time.After(100 * time.Millisecond):
	}
	if err := executionTx.Commit(); err != nil {
		t.Fatal(err)
	}
	select {
	case outcome := <-cleanupDone:
		if outcome.err != nil || outcome.result.Processed() != 0 {
			t.Fatalf("cleanup after terminal commit = %+v, %v", outcome.result, outcome.err)
		}
	case <-time.After(3 * time.Second):
		t.Fatal("cleanup did not resume after command transaction committed")
	}
	if err := db.QueryRow(`SELECT COUNT(*) FROM _System_PermissionAuthoringCommand
WHERE Id = 'bound-accepted-live' AND State = 'COMPLETED' AND OutcomeExpiresAt > UTC_TIMESTAMP(6)`).Scan(&remaining); err != nil {
		t.Fatal(err)
	}
	if remaining != 1 {
		t.Fatal("cleanup removed a command that completed while its row lock was held")
	}
}