package persistence

import (
	"bytes"
	"errors"
	"fmt"
	"strings"
	"testing"
)

func TestRuntimePlatformCommandTiDBPersistsRestartAndRevisionFences(t *testing.T) {
	db, cleanup := openSessionRepositoryTiDB(t)
	defer cleanup()
	createRuntimePlatformCommandTestTable(t, db, "MEDIUMTEXT")
	if err := VerifyRuntimePlatformCommandSchema(t.Context(), db); err == nil {
		t.Fatal("VerifyRuntimePlatformCommandSchema() accepted a text payload column")
	}
	if _, err := db.Exec(`DROP TABLE _System_RuntimePlatformCommand`); err != nil {
		t.Fatalf("drop stale runtime platform command table: %v", err)
	}
	createRuntimePlatformCommandTestTable(t, db, "MEDIUMBLOB")
	if err := VerifyRuntimePlatformCommandSchema(t.Context(), db); err != nil {
		t.Fatal(err)
	}
	for _, change := range []struct {
		name    string
		prepare []string
		restore []string
	}{
		{
			name: "missing scope identity",
			prepare: []string{`ALTER TABLE _System_RuntimePlatformCommand
				DROP INDEX ux_runtime_platform_command_scope`},
			restore: []string{`ALTER TABLE _System_RuntimePlatformCommand
				ADD UNIQUE KEY ux_runtime_platform_command_scope (ScopeDigest, IdempotencyKey)`},
		},
		{
			name: "non-unique scope identity",
			prepare: []string{
				`ALTER TABLE _System_RuntimePlatformCommand DROP INDEX ux_runtime_platform_command_scope`,
				`ALTER TABLE _System_RuntimePlatformCommand ADD KEY ux_runtime_platform_command_scope (ScopeDigest, IdempotencyKey)`,
			},
			restore: []string{
				`ALTER TABLE _System_RuntimePlatformCommand DROP INDEX ux_runtime_platform_command_scope`,
				`ALTER TABLE _System_RuntimePlatformCommand ADD UNIQUE KEY ux_runtime_platform_command_scope (ScopeDigest, IdempotencyKey)`,
			},
		},
		{
			name: "reversed scope identity",
			prepare: []string{
				`ALTER TABLE _System_RuntimePlatformCommand DROP INDEX ux_runtime_platform_command_scope`,
				`ALTER TABLE _System_RuntimePlatformCommand ADD UNIQUE KEY ux_runtime_platform_command_scope (IdempotencyKey, ScopeDigest)`,
			},
			restore: []string{
				`ALTER TABLE _System_RuntimePlatformCommand DROP INDEX ux_runtime_platform_command_scope`,
				`ALTER TABLE _System_RuntimePlatformCommand ADD UNIQUE KEY ux_runtime_platform_command_scope (ScopeDigest, IdempotencyKey)`,
			},
		},
		{
			name: "missing command identity",
			prepare: []string{`ALTER TABLE _System_RuntimePlatformCommand
				DROP INDEX idx_runtime_platform_command_id`},
			restore: []string{`ALTER TABLE _System_RuntimePlatformCommand
				ADD UNIQUE KEY idx_runtime_platform_command_id (ScopeDigest, CommandId)`},
		},
		{
			name: "non-unique command identity",
			prepare: []string{
				`ALTER TABLE _System_RuntimePlatformCommand DROP INDEX idx_runtime_platform_command_id`,
				`ALTER TABLE _System_RuntimePlatformCommand ADD KEY idx_runtime_platform_command_id (ScopeDigest, CommandId)`,
			},
			restore: []string{
				`ALTER TABLE _System_RuntimePlatformCommand DROP INDEX idx_runtime_platform_command_id`,
				`ALTER TABLE _System_RuntimePlatformCommand ADD UNIQUE KEY idx_runtime_platform_command_id (ScopeDigest, CommandId)`,
			},
		},
		{
			name: "reversed command identity",
			prepare: []string{
				`ALTER TABLE _System_RuntimePlatformCommand DROP INDEX idx_runtime_platform_command_id`,
				`ALTER TABLE _System_RuntimePlatformCommand ADD UNIQUE KEY idx_runtime_platform_command_id (CommandId, ScopeDigest)`,
			},
			restore: []string{
				`ALTER TABLE _System_RuntimePlatformCommand DROP INDEX idx_runtime_platform_command_id`,
				`ALTER TABLE _System_RuntimePlatformCommand ADD UNIQUE KEY idx_runtime_platform_command_id (ScopeDigest, CommandId)`,
			},
		},
		{
			name: "missing worker claim identity",
			prepare: []string{`ALTER TABLE _System_RuntimePlatformCommand
				DROP INDEX idx_runtime_platform_command_claim`},
			restore: []string{`ALTER TABLE _System_RuntimePlatformCommand
				ADD KEY idx_runtime_platform_command_claim
				(ExecutionScopeDigest, State, AvailableAt, LeaseExpiresAt, Id)`},
		},
		{
			name: "unique worker claim identity",
			prepare: []string{
				`ALTER TABLE _System_RuntimePlatformCommand DROP INDEX idx_runtime_platform_command_claim`,
				`ALTER TABLE _System_RuntimePlatformCommand ADD UNIQUE KEY idx_runtime_platform_command_claim
				(ExecutionScopeDigest, State, AvailableAt, LeaseExpiresAt, Id)`,
			},
			restore: []string{
				`ALTER TABLE _System_RuntimePlatformCommand DROP INDEX idx_runtime_platform_command_claim`,
				`ALTER TABLE _System_RuntimePlatformCommand ADD KEY idx_runtime_platform_command_claim
				(ExecutionScopeDigest, State, AvailableAt, LeaseExpiresAt, Id)`,
			},
		},
		{
			name: "reversed worker claim identity",
			prepare: []string{
				`ALTER TABLE _System_RuntimePlatformCommand DROP INDEX idx_runtime_platform_command_claim`,
				`ALTER TABLE _System_RuntimePlatformCommand ADD KEY idx_runtime_platform_command_claim
				(State, ExecutionScopeDigest, AvailableAt, LeaseExpiresAt, Id)`,
			},
			restore: []string{
				`ALTER TABLE _System_RuntimePlatformCommand DROP INDEX idx_runtime_platform_command_claim`,
				`ALTER TABLE _System_RuntimePlatformCommand ADD KEY idx_runtime_platform_command_claim
				(ExecutionScopeDigest, State, AvailableAt, LeaseExpiresAt, Id)`,
			},
		},
	} {
		t.Run(change.name, func(t *testing.T) {
			for _, statement := range change.prepare {
				if _, err := db.Exec(statement); err != nil {
					t.Fatal(err)
				}
			}
			if err := VerifyRuntimePlatformCommandSchema(t.Context(), db); err == nil {
				t.Fatal("VerifyRuntimePlatformCommandSchema() accepted a non-authoritative identity index")
			}
			for _, statement := range change.restore {
				if _, err := db.Exec(statement); err != nil {
					t.Fatal(err)
				}
			}
			if err := VerifyRuntimePlatformCommandSchema(t.Context(), db); err != nil {
				t.Fatalf("restored Runtime command schema: %v", err)
			}
		})
	}

	input := RuntimePlatformCommandAcceptance{
		RowID:                "row-1",
		OrgID:                "org-1",
		AppInstallationID:    "installation-1",
		CallerWorkloadID:     "integration-1",
		RuntimeCapabilityID:  "runtime.commands.accept.v1",
		MetadataBinding:      "binding-1",
		ScopeDigest:          "aaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaa",
		CommandID:            "command-1",
		IdempotencyKey:       "attempt-1",
		CommandKind:          "runtime.generic.sync",
		PartitionKey:         "partition-1",
		PayloadFormatVersion: 1,
		RequestDigest:        "sha256:bbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbb",
		Payload:              []byte{0xff, 0xfe, 0x00, 0x61},
		Receipts:             []byte(`{"CatalogRevision":7,"SecurityEpoch":11}`),
	}
	input.ExecutionScopeDigest = mustRuntimeWorkScopeDigest(t, input)
	repository := NewRuntimePlatformCommandRepository(db)
	row, duplicate, err := repository.Accept(t.Context(), input)
	if err != nil || duplicate || row.State != RuntimePlatformCommandAccepted ||
		!bytes.Equal(row.Payload, input.Payload) {
		t.Fatalf("first Accept() = %#v duplicate=%t err=%v", row, duplicate, err)
	}
	// A new repository represents a process restart. Opaque payload bytes must
	// round-trip exactly while TiDB may normalize the JSON receipts document.
	restarted := NewRuntimePlatformCommandRepository(db)
	replay := input
	replay.Receipts = []byte(`{"SecurityEpoch":11,"CatalogRevision":7}`)
	row, duplicate, err = restarted.Accept(t.Context(), replay)
	if err != nil || !duplicate || row.State != RuntimePlatformCommandAccepted ||
		len(row.ResultPayload) != 0 ||
		!bytes.Equal(row.Payload, input.Payload) {
		t.Fatalf("restart Accept() = %#v duplicate=%t err=%v", row, duplicate, err)
	}
	if _, err := db.Exec(
		`UPDATE _System_RuntimePlatformCommand SET ExecutionScopeDigest = ? WHERE Id = ?`,
		strings.Repeat("c", 64),
		input.RowID,
	); err != nil {
		t.Fatal(err)
	}
	err = VerifyRuntimePlatformCommandSchema(t.Context(), db)
	if !errors.Is(err, ErrRuntimePlatformCommandLegacyAuthority) {
		t.Fatalf("well-formed mismatched authority error = %v, want quarantine", err)
	}
	if _, err := db.Exec(
		`UPDATE _System_RuntimePlatformCommand SET ExecutionScopeDigest = ? WHERE Id = ?`,
		input.ExecutionScopeDigest,
		input.RowID,
	); err != nil {
		t.Fatal(err)
	}
	if err := VerifyRuntimePlatformCommandSchema(t.Context(), db); err != nil {
		t.Fatalf("restored Runtime command authority: %v", err)
	}

	changedRevision := replay
	changedRevision.Receipts = []byte(`{"CatalogRevision":8,"SecurityEpoch":11}`)
	if _, duplicate, err := restarted.Accept(t.Context(), changedRevision); !duplicate ||
		!errors.Is(err, ErrRuntimePlatformCommandServingBindingChanged) {
		t.Fatalf("changed revision duplicate=%t err=%v", duplicate, err)
	}

	changedContent := replay
	changedContent.RequestDigest = "sha256:cccccccccccccccccccccccccccccccccccccccccccccccccccccccccccccccc"
	if _, duplicate, err := restarted.Accept(t.Context(), changedContent); !duplicate ||
		!errors.Is(err, ErrRuntimePlatformCommandConflict) {
		t.Fatalf("changed content duplicate=%t err=%v", duplicate, err)
	}

	changedPayload := replay
	changedPayload.Payload = []byte{0xff, 0xfe, 0x00, 0x62}
	if _, duplicate, err := restarted.Accept(t.Context(), changedPayload); !duplicate ||
		!errors.Is(err, ErrRuntimePlatformCommandConflict) {
		t.Fatalf("changed payload duplicate=%t err=%v", duplicate, err)
	}

	changedIdempotencyKey := replay
	changedIdempotencyKey.RowID = "row-2"
	changedIdempotencyKey.IdempotencyKey = "attempt-2"
	changedIdempotencyKey.Receipts = []byte(`{"CatalogRevision":8,"SecurityEpoch":11}`)
	if _, duplicate, err := restarted.Accept(t.Context(), changedIdempotencyKey); !duplicate ||
		!errors.Is(err, ErrRuntimePlatformCommandConflict) {
		t.Fatalf("changed idempotency key duplicate=%t err=%v", duplicate, err)
	}

	changedCommandID := replay
	changedCommandID.RowID = "row-3"
	changedCommandID.CommandID = "command-2"
	if _, duplicate, err := restarted.Accept(t.Context(), changedCommandID); !duplicate ||
		!errors.Is(err, ErrRuntimePlatformCommandConflict) {
		t.Fatalf("changed command id duplicate=%t err=%v", duplicate, err)
	}
}

func TestRuntimePlatformCommandTiDBStoppedCutoverQuarantinesLegacyRows(t *testing.T) {
	db, cleanup := openSessionRepositoryTiDB(t)
	defer cleanup()
	createLegacyRuntimePlatformCommandTestTable(t, db)
	if _, err := db.Exec(`INSERT INTO _System_RuntimePlatformCommand
(Id, ScopeDigest, CommandId, IdempotencyKey, CommandKind, PayloadFormatVersion,
 RequestDigest, Payload, Receipts, State)
VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, 'ACCEPTED')`,
		"legacy-row", strings.Repeat("a", 64), "legacy-command", "legacy-attempt",
		"legacy.kind", 1, "sha256:"+strings.Repeat("b", 64), []byte{0xff}, []byte(`{}`),
	); err != nil {
		t.Fatalf("insert pre-B2 accepted row: %v", err)
	}

	// This is the exact stopped-upgrade shape emitted by the additive schema
	// reconciler. TiDB preserves the old row and fills the new required text
	// authority with empty values; those values cannot be reconstructed.
	for _, statement := range []string{
		`ALTER TABLE _System_RuntimePlatformCommand ADD COLUMN OrgId VARCHAR(255) NOT NULL`,
		`ALTER TABLE _System_RuntimePlatformCommand ADD COLUMN AppInstallationId VARCHAR(255) NOT NULL`,
		`ALTER TABLE _System_RuntimePlatformCommand ADD COLUMN CallerWorkloadId VARCHAR(1024) NOT NULL`,
		`ALTER TABLE _System_RuntimePlatformCommand ADD COLUMN RuntimeCapabilityId VARCHAR(255) NOT NULL`,
		`ALTER TABLE _System_RuntimePlatformCommand ADD COLUMN MetadataBinding MEDIUMTEXT NOT NULL`,
		`ALTER TABLE _System_RuntimePlatformCommand ADD COLUMN ExecutionScopeDigest CHAR(64) NOT NULL`,
		`ALTER TABLE _System_RuntimePlatformCommand ADD COLUMN Attempt INT NOT NULL DEFAULT 0`,
		`ALTER TABLE _System_RuntimePlatformCommand ADD COLUMN AvailableAt DATETIME(6) NOT NULL DEFAULT CURRENT_TIMESTAMP(6)`,
		`ALTER TABLE _System_RuntimePlatformCommand ADD COLUMN LeaseToken VARCHAR(64) NULL`,
		`ALTER TABLE _System_RuntimePlatformCommand ADD COLUMN LeaseExpiresAt DATETIME(6) NULL`,
		`ALTER TABLE _System_RuntimePlatformCommand ADD COLUMN OwnershipEpoch DECIMAL(20,0) NULL`,
		`ALTER TABLE _System_RuntimePlatformCommand ADD COLUMN DependencyKind VARCHAR(64) NULL`,
		`ALTER TABLE _System_RuntimePlatformCommand ADD COLUMN DependencyObjectAPIName VARCHAR(255) NULL`,
		`ALTER TABLE _System_RuntimePlatformCommand ADD COLUMN DependencyRecordId VARCHAR(255) NULL`,
		`ALTER TABLE _System_RuntimePlatformCommand ADD COLUMN DependencyKeyHash CHAR(64) NULL`,
		`ALTER TABLE _System_RuntimePlatformCommand ADD COLUMN DependencyConnectorOwnerKey VARCHAR(255) NULL`,
		`ALTER TABLE _System_RuntimePlatformCommand ADD COLUMN DependencyConnectorNamespace VARCHAR(255) NULL`,
		`ALTER TABLE _System_RuntimePlatformCommand ADD COLUMN DependencyConnectorSourceObject VARCHAR(255) NULL`,
		`ALTER TABLE _System_RuntimePlatformCommand ADD COLUMN DependencyDestinationSlotId VARCHAR(255) NULL`,
		`ALTER TABLE _System_RuntimePlatformCommand ADD COLUMN DependencyConnectorRemoteId VARCHAR(255) NULL`,
		`ALTER TABLE _System_RuntimePlatformCommand ADD COLUMN DependencyBlockedAt DATETIME(6) NULL`,
		`ALTER TABLE _System_RuntimePlatformCommand ADD COLUMN LastError LONGTEXT NULL`,
		`ALTER TABLE _System_RuntimePlatformCommand ADD COLUMN RepairId VARCHAR(255) NULL`,
		`ALTER TABLE _System_RuntimePlatformCommand ADD COLUMN RepairReason VARCHAR(1024) NULL`,
		`ALTER TABLE _System_RuntimePlatformCommand ADD COLUMN RepairCount INT NOT NULL DEFAULT 0`,
		`ALTER TABLE _System_RuntimePlatformCommand ADD COLUMN LastRepairAt DATETIME(6) NULL`,
		`ALTER TABLE _System_RuntimePlatformCommand ADD COLUMN PublicationIdentity CHAR(71) NULL`,
		`ALTER TABLE _System_RuntimePlatformCommand ADD COLUMN PublishAttempt INT NOT NULL DEFAULT 0`,
		`ALTER TABLE _System_RuntimePlatformCommand ADD COLUMN SerializationKeyDigest CHAR(64) NULL`,
		`ALTER TABLE _System_RuntimePlatformCommand ADD COLUMN RecoveryState VARCHAR(16) NULL`,
		`ALTER TABLE _System_RuntimePlatformCommand DROP INDEX idx_runtime_platform_command_claim`,
		`ALTER TABLE _System_RuntimePlatformCommand ADD KEY idx_runtime_platform_command_claim
			(ExecutionScopeDigest, State, AvailableAt, LeaseExpiresAt, Id)`,
		`ALTER TABLE _System_RuntimePlatformCommand ADD KEY idx_runtime_platform_command_partition
			(ExecutionScopeDigest, PartitionKey, State, LeaseExpiresAt, Id)`,
		`ALTER TABLE _System_RuntimePlatformCommand ADD KEY idx_runtime_platform_command_serialization
			(ExecutionScopeDigest, SerializationKeyDigest, State, LeaseExpiresAt, Id)`,
	} {
		if _, err := db.Exec(statement); err != nil {
			t.Fatalf("apply stopped-cutover statement %q: %v", statement, err)
		}
	}

	err := VerifyRuntimePlatformCommandSchema(t.Context(), db)
	if !errors.Is(err, ErrRuntimePlatformCommandLegacyAuthority) {
		t.Fatalf("cutover readiness error = %v, want legacy-authority quarantine", err)
	}
	var rows, emptyAuthorities int
	if err := db.QueryRow(`SELECT COUNT(*),
SUM(CASE WHEN OrgId = '' AND AppInstallationId = '' AND CallerWorkloadId = ''
              AND RuntimeCapabilityId = '' AND MetadataBinding = ''
              AND ExecutionScopeDigest = '' THEN 1 ELSE 0 END)
FROM _System_RuntimePlatformCommand`).Scan(&rows, &emptyAuthorities); err != nil {
		t.Fatal(err)
	}
	if rows != 1 || emptyAuthorities != 1 {
		t.Fatalf("quarantined legacy rows = %d/%d, want preserved 1/1", rows, emptyAuthorities)
	}

	// Production data disposition is an explicit stopped-window operator
	// decision. The runtime never guesses authority or silently deletes rows;
	// once the preserved legacy inbox has been drained or archived, the exact
	// same schema becomes ready for B2 activation.
	if _, err := db.Exec(`DELETE FROM _System_RuntimePlatformCommand WHERE Id = ?`, "legacy-row"); err != nil {
		t.Fatal(err)
	}
	if err := VerifyRuntimePlatformCommandSchema(t.Context(), db); err != nil {
		t.Fatalf("schema after explicit legacy disposition: %v", err)
	}
}

func createLegacyRuntimePlatformCommandTestTable(t *testing.T, db Executor) {
	t.Helper()
	if _, err := db.ExecContext(t.Context(), `CREATE TABLE _System_RuntimePlatformCommand (
		Id VARCHAR(255) NOT NULL PRIMARY KEY,
		ScopeDigest CHAR(64) NOT NULL,
		CommandId VARCHAR(255) NOT NULL,
		IdempotencyKey VARCHAR(255) NOT NULL,
		CommandKind VARCHAR(255) NOT NULL,
		PartitionKey VARCHAR(255) NULL,
		PayloadFormatVersion BIGINT NOT NULL,
		RequestDigest CHAR(71) NOT NULL,
		Payload MEDIUMBLOB NOT NULL,
		Receipts JSON NOT NULL,
		State VARCHAR(16) NOT NULL,
		ResultPayload MEDIUMTEXT NULL,
		CreatedDate DATETIME(6) NOT NULL DEFAULT CURRENT_TIMESTAMP(6),
		LastModifiedDate DATETIME(6) NOT NULL DEFAULT CURRENT_TIMESTAMP(6),
		UNIQUE KEY ux_runtime_platform_command_scope (ScopeDigest, IdempotencyKey),
		KEY idx_runtime_platform_command_claim (State, PartitionKey, CreatedDate, Id),
		UNIQUE KEY idx_runtime_platform_command_id (ScopeDigest, CommandId)
	)`); err != nil {
		t.Fatalf("create pre-B2 Runtime command table: %v", err)
	}
}

func createRuntimePlatformCommandTestTable(t *testing.T, db Executor, payloadType string) {
	t.Helper()
	if payloadType != "MEDIUMTEXT" && payloadType != "MEDIUMBLOB" {
		t.Fatalf("unsupported runtime platform command test payload type %q", payloadType)
	}
	if _, err := db.ExecContext(t.Context(), fmt.Sprintf(`CREATE TABLE _System_RuntimePlatformCommand (
		Id VARCHAR(255) NOT NULL PRIMARY KEY,
		OrgId VARCHAR(255) NOT NULL,
		AppInstallationId VARCHAR(255) NOT NULL,
		CallerWorkloadId VARCHAR(1024) NOT NULL,
		RuntimeCapabilityId VARCHAR(255) NOT NULL,
		MetadataBinding MEDIUMTEXT NOT NULL,
		ExecutionScopeDigest CHAR(64) NOT NULL,
		ScopeDigest CHAR(64) NOT NULL,
		CommandId VARCHAR(255) NOT NULL,
		IdempotencyKey VARCHAR(255) NOT NULL,
		CommandKind VARCHAR(255) NOT NULL,
		PartitionKey VARCHAR(255) NULL,
		SerializationKeyDigest CHAR(64) NULL,
		PayloadFormatVersion BIGINT NOT NULL,
		RequestDigest CHAR(71) NOT NULL,
		Payload %s NOT NULL,
		Receipts JSON NOT NULL,
		State VARCHAR(16) NOT NULL,
		Attempt INT NOT NULL DEFAULT 0,
		AvailableAt DATETIME(6) NOT NULL DEFAULT CURRENT_TIMESTAMP(6),
		LeaseToken VARCHAR(64) NULL,
		LeaseExpiresAt DATETIME(6) NULL,
		OwnershipEpoch DECIMAL(20,0) NULL,
		DependencyKind VARCHAR(64) NULL,
		DependencyObjectAPIName VARCHAR(255) NULL,
		DependencyRecordId VARCHAR(255) NULL,
		DependencyKeyHash CHAR(64) NULL,
		DependencyConnectorOwnerKey VARCHAR(255) NULL,
		DependencyConnectorNamespace VARCHAR(255) NULL,
		DependencyConnectorSourceObject VARCHAR(255) NULL,
		DependencyDestinationSlotId VARCHAR(255) NULL,
		DependencyConnectorRemoteId VARCHAR(255) NULL,
		DependencyBlockedAt DATETIME(6) NULL,
		LastError LONGTEXT NULL,
		RepairId VARCHAR(255) NULL,
		RepairReason VARCHAR(1024) NULL,
		RepairCount INT NOT NULL DEFAULT 0,
		LastRepairAt DATETIME(6) NULL,
		ResultPayload MEDIUMTEXT NULL,
		PublicationIdentity CHAR(71) NULL,
		PublishAttempt INT NOT NULL DEFAULT 0,
		RecoveryState VARCHAR(16) NULL,
		CreatedDate DATETIME(6) NOT NULL DEFAULT CURRENT_TIMESTAMP(6),
		LastModifiedDate DATETIME(6) NOT NULL DEFAULT CURRENT_TIMESTAMP(6),
		UNIQUE KEY ux_runtime_platform_command_scope (ScopeDigest, IdempotencyKey),
		UNIQUE KEY idx_runtime_platform_command_id (ScopeDigest, CommandId),
		KEY idx_runtime_platform_command_claim (ExecutionScopeDigest, State, AvailableAt, LeaseExpiresAt, Id),
		KEY idx_runtime_platform_command_partition (ExecutionScopeDigest, PartitionKey, State, LeaseExpiresAt, Id),
		KEY idx_runtime_platform_command_serialization (ExecutionScopeDigest, SerializationKeyDigest, State, LeaseExpiresAt, Id)
	)`, payloadType)); err != nil {
		t.Fatalf("create runtime platform command table with %s payload: %v", payloadType, err)
	}
}