//go:build runtime_api_integration

package persistence

import (
	"context"
	"errors"
	"testing"
	"time"
)

func TestRuntimeSecurityRepositoryTiDBAuthoritativeProjection(t *testing.T) {
	db, cleanup := openSessionRepositoryTiDB(t)
	defer cleanup()
	for _, statement := range []string{
		`CREATE TABLE _System_RuntimeSecurityState (
            OrgId VARCHAR(255) PRIMARY KEY, RuntimeAudience VARCHAR(255) NOT NULL,
            SecurityEpoch DECIMAL(20,0) NOT NULL, CreatedDate DATETIME(6) NOT NULL,
            LastModifiedDate DATETIME(6) NOT NULL)`,
		`CREATE TABLE _System_RuntimeSecurityBinding (
            Id VARCHAR(255) PRIMARY KEY, OrgId VARCHAR(255) NOT NULL, AppInstallationId VARCHAR(255) NOT NULL,
            CallerWorkloadId VARCHAR(1024) NOT NULL, CallerWorkloadDigest CHAR(64) NOT NULL,
            Capabilities JSON NOT NULL, IsActive BOOL NOT NULL,
            CreatedDate DATETIME(6) NOT NULL, LastModifiedDate DATETIME(6) NOT NULL,
            INDEX idx_installation (OrgId, AppInstallationId, CallerWorkloadDigest))`,
		`CREATE TABLE _System_RuntimeSecurityLease (
            RuntimeReplicaId VARCHAR(255) PRIMARY KEY, OrgId VARCHAR(255) NOT NULL,
            SecurityEpoch DECIMAL(20,0) NOT NULL, ExpiresAt DATETIME(6) NOT NULL,
            CreatedDate DATETIME(6) NOT NULL, LastModifiedDate DATETIME(6) NOT NULL,
            INDEX idx_org_epoch (OrgId, SecurityEpoch))`,
		`CREATE TABLE _System_RuntimeDelegationRevocation (
            Id VARCHAR(255) PRIMARY KEY, OrgId VARCHAR(255) NOT NULL, DelegationId VARCHAR(255) NOT NULL,
            ExpiresAt DATETIME(6) NOT NULL, CreatedDate DATETIME(6) NOT NULL,
            LastModifiedDate DATETIME(6) NOT NULL, UNIQUE KEY ux_org_delegation (OrgId, DelegationId))`,
	} {
		if _, err := db.Exec(statement); err != nil {
			t.Fatal(err)
		}
	}

	repo := NewRuntimeSecurityRepository(db)
	ctx := context.Background()
	capability := "runtime.records.read-selected-root.v1"
	for _, projection := range []RuntimeSecurityProjection{
		{OrgID: "org-1", RuntimeAudience: "runtime", SecurityEpoch: 1, Bindings: []RuntimeSecurityBinding{{
			CallerWorkloadID: "workload-1", OrgID: "org-1",
			AppInstallationID: "installation-1", Capabilities: []string{capability}, Active: true,
		}}},
		{OrgID: "org-2", RuntimeAudience: "runtime", SecurityEpoch: 1, Bindings: []RuntimeSecurityBinding{{
			CallerWorkloadID: "workload-1", OrgID: "org-2",
			AppInstallationID: "installation-2", Capabilities: []string{capability}, Active: true,
		}}},
	} {
		if err := repo.ActivateProjection(ctx, 0, projection); err != nil {
			t.Fatal(err)
		}
	}
	leaseOne, found, err := repo.RenewReplicaLease(ctx, "org-1", "replica-1", 30*time.Second)
	if err != nil || !found || leaseOne.SecurityEpoch != 1 || leaseOne.LeaseExpiresAt.Sub(leaseOne.AuthorityNow) != 30*time.Second {
		t.Fatalf("renew org-1 lease = %#v, %v, %v", leaseOne, found, err)
	}
	if _, found, err := repo.RenewReplicaLease(ctx, "org-1", "replica-contended", 30*time.Second); err != nil || !found {
		t.Fatalf("seed contended lease: found=%v err=%v", found, err)
	}
	leaseLock, err := db.BeginTx(ctx, nil)
	if err != nil {
		t.Fatal(err)
	}
	defer leaseLock.Rollback()
	var lockedExpiry time.Time
	if err := leaseLock.QueryRowContext(ctx, "SELECT `ExpiresAt` FROM `_System_RuntimeSecurityLease` WHERE `RuntimeReplicaId` = ? FOR UPDATE", "replica-contended").Scan(&lockedExpiry); err != nil {
		t.Fatal(err)
	}
	type renewalResult struct {
		state RuntimeReplicaSecurityContext
		found bool
		err   error
	}
	contended := make(chan renewalResult, 1)
	contendedDuration := 2 * time.Second
	go func() {
		state, found, err := repo.RenewReplicaLease(ctx, "org-1", "replica-contended", contendedDuration)
		contended <- renewalResult{state: state, found: found, err: err}
	}()
	select {
	case result := <-contended:
		t.Fatalf("lease renewal bypassed the contended replica row: %#v", result)
	case <-time.After(contendedDuration + 500*time.Millisecond):
	}
	if err := leaseLock.Commit(); err != nil {
		t.Fatal(err)
	}
	var renewedAfterWait renewalResult
	select {
	case renewedAfterWait = <-contended:
	case <-time.After(5 * time.Second):
		t.Fatal("lease renewal did not finish after the contended row was released")
	}
	if renewedAfterWait.err != nil || !renewedAfterWait.found ||
		renewedAfterWait.state.LeaseExpiresAt.Sub(renewedAfterWait.state.AuthorityNow) != contendedDuration {
		t.Fatalf("contended lease renewal = %#v", renewedAfterWait)
	}
	var authorityNow time.Time
	if err := db.QueryRowContext(ctx, "SELECT CURRENT_TIMESTAMP(6)").Scan(&authorityNow); err != nil {
		t.Fatal(err)
	}
	if remaining := renewedAfterWait.state.LeaseExpiresAt.Sub(authorityNow); remaining <= 500*time.Millisecond {
		t.Fatalf("contended lease lost lifetime before return: remaining=%s state=%#v", remaining, renewedAfterWait.state)
	}
	if _, found, err := repo.RenewReplicaLease(ctx, "org-2", "replica-2", 30*time.Second); err != nil || !found {
		t.Fatalf("renew org-2 lease: found=%v err=%v", found, err)
	}
	expires := leaseOne.LeaseExpiresAt

	replica, found, err := repo.ReadReplicaSecurityContext(ctx, "replica-1")
	if err != nil || !found || replica.OrgID != "org-1" || replica.RuntimeAudience != "runtime" || replica.SecurityEpoch != 1 || replica.LeaseSecurityEpoch != 1 {
		t.Fatalf("ReadReplicaSecurityContext = %#v, %v, %v", replica, found, err)
	}

	state, err := repo.ReadAdmissionState(ctx, "replica-1", "org-1", "installation-1", "workload-1", "delegation-1", time.Now())
	if err != nil || !state.InstallationFound || !state.InstallationActive || !state.LeaseFound || state.DelegationRevoked {
		t.Fatalf("authorized state = %#v, %v", state, err)
	}
	crossOrg, err := repo.ReadAdmissionState(ctx, "replica-1", "org-1", "installation-2", "workload-1", "delegation-1", time.Now())
	if err != nil || crossOrg.InstallationFound {
		t.Fatalf("cross-org state = %#v, %v", crossOrg, err)
	}

	if err := repo.RevokeDelegation(ctx, "org-1", "delegation-1", expires); err != nil {
		t.Fatal(err)
	}
	revoked, err := repo.ReadAdmissionState(ctx, "replica-1", "org-1", "installation-1", "workload-1", "delegation-1", time.Now())
	if err != nil || !revoked.DelegationRevoked {
		t.Fatalf("revoked state = %#v, %v", revoked, err)
	}

	advanced := RuntimeSecurityProjection{OrgID: "org-1", RuntimeAudience: "runtime", SecurityEpoch: 2, Bindings: []RuntimeSecurityBinding{{
		CallerWorkloadID: "workload-1", OrgID: "org-1",
		AppInstallationID: "installation-1", Capabilities: []string{capability}, Active: true,
	}}}
	if err := repo.ActivateProjection(ctx, 1, advanced); err != nil {
		t.Fatal(err)
	}
	if _, found, err := repo.ReadReplicaSecurityContext(ctx, "replica-1"); err != nil || found {
		t.Fatalf("old lease survived activation: found=%v err=%v", found, err)
	}
	renewed, found, err := repo.RenewReplicaLease(ctx, "org-1", "replica-1", 30*time.Second)
	if err != nil || !found || renewed.SecurityEpoch != 2 || renewed.LeaseSecurityEpoch != 2 {
		t.Fatalf("renew after epoch activation = %#v, %v, %v", renewed, found, err)
	}
	if err := repo.ActivateProjection(ctx, 1, advanced); err != nil {
		t.Fatalf("identical activation retry = %v", err)
	}
	conflicting := advanced
	conflicting.RuntimeAudience = "other-runtime"
	if err := repo.ActivateProjection(ctx, 1, conflicting); !errors.Is(err, ErrRuntimeSecurityEpochConflict) {
		t.Fatalf("same-epoch conflicting activation = %v, want epoch conflict", err)
	}
}

func TestRuntimeSecurityRepositoryTiDBIdenticalRetryUsesCurrentProjection(t *testing.T) {
	db, cleanup := openSessionRepositoryTiDB(t)
	defer cleanup()
	for _, statement := range []string{
		`CREATE TABLE _System_RuntimeSecurityState (
            OrgId VARCHAR(255) PRIMARY KEY, RuntimeAudience VARCHAR(255) NOT NULL,
            SecurityEpoch DECIMAL(20,0) NOT NULL, CreatedDate DATETIME(6) NOT NULL,
            LastModifiedDate DATETIME(6) NOT NULL)`,
		`CREATE TABLE _System_RuntimeSecurityBinding (
            Id VARCHAR(255) PRIMARY KEY, OrgId VARCHAR(255) NOT NULL, AppInstallationId VARCHAR(255) NOT NULL,
            CallerWorkloadId VARCHAR(1024) NOT NULL, CallerWorkloadDigest CHAR(64) NOT NULL,
            Capabilities JSON NOT NULL, IsActive BOOL NOT NULL,
            CreatedDate DATETIME(6) NOT NULL, LastModifiedDate DATETIME(6) NOT NULL,
            INDEX idx_installation (OrgId, AppInstallationId, CallerWorkloadDigest))`,
		`CREATE TABLE _System_RuntimeSecurityLease (
            RuntimeReplicaId VARCHAR(255) PRIMARY KEY, OrgId VARCHAR(255) NOT NULL,
            SecurityEpoch DECIMAL(20,0) NOT NULL, ExpiresAt DATETIME(6) NOT NULL,
            CreatedDate DATETIME(6) NOT NULL, LastModifiedDate DATETIME(6) NOT NULL,
            INDEX idx_org_epoch (OrgId, SecurityEpoch))`,
	} {
		if _, err := db.Exec(statement); err != nil {
			t.Fatal(err)
		}
	}

	repo := NewRuntimeSecurityRepository(db)
	initial := RuntimeSecurityProjection{OrgID: "org-retry", RuntimeAudience: "runtime", SecurityEpoch: 1, Bindings: []RuntimeSecurityBinding{{
		CallerWorkloadID: "workload-old", OrgID: "org-retry", AppInstallationID: "installation-old",
		Capabilities: []string{"runtime.old.v1"}, Active: true,
	}}}
	if err := repo.ActivateProjection(context.Background(), 0, initial); err != nil {
		t.Fatal(err)
	}
	target := RuntimeSecurityProjection{OrgID: "org-retry", RuntimeAudience: "runtime", SecurityEpoch: 2, Bindings: []RuntimeSecurityBinding{{
		CallerWorkloadID: "workload-new", OrgID: "org-retry", AppInstallationID: "installation-new",
		Capabilities: []string{"runtime.new.v1"}, Active: true,
	}}}

	// Hold an uncommitted winning activation while the retry transaction starts.
	// Its epoch lock must wait, then every equivalence read must observe the
	// winner's committed rows rather than the retry transaction's old snapshot.
	winner, err := db.BeginTx(context.Background(), nil)
	if err != nil {
		t.Fatal(err)
	}
	defer winner.Rollback()
	var epoch []byte
	if err := winner.QueryRow("SELECT `SecurityEpoch` FROM `_System_RuntimeSecurityState` WHERE `OrgId` = ? FOR UPDATE", target.OrgID).Scan(&epoch); err != nil {
		t.Fatal(err)
	}
	if _, err := winner.Exec("UPDATE `_System_RuntimeSecurityState` SET `RuntimeAudience` = ?, `SecurityEpoch` = ? WHERE `OrgId` = ?", target.RuntimeAudience, epochString(target.SecurityEpoch), target.OrgID); err != nil {
		t.Fatal(err)
	}
	if _, err := winner.Exec("DELETE FROM `_System_RuntimeSecurityBinding` WHERE `OrgId` = ?", target.OrgID); err != nil {
		t.Fatal(err)
	}
	binding := target.Bindings[0]
	if _, err := winner.Exec("INSERT INTO `_System_RuntimeSecurityBinding` (`Id`, `OrgId`, `AppInstallationId`, `CallerWorkloadId`, `CallerWorkloadDigest`, `Capabilities`, `IsActive`, `CreatedDate`, `LastModifiedDate`) VALUES (?, ?, ?, ?, ?, ?, ?, CURRENT_TIMESTAMP(6), CURRENT_TIMESTAMP(6))",
		securityBindingID(binding), target.OrgID, binding.AppInstallationID, binding.CallerWorkloadID, digestString(binding.CallerWorkloadID), []byte(`["runtime.new.v1"]`), true); err != nil {
		t.Fatal(err)
	}

	result := make(chan error, 1)
	ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
	defer cancel()
	go func() { result <- repo.ActivateProjection(ctx, 1, target) }()
	select {
	case err := <-result:
		t.Fatalf("retry returned before winning activation committed: %v", err)
	case <-time.After(100 * time.Millisecond):
	}
	if err := winner.Commit(); err != nil {
		t.Fatal(err)
	}
	select {
	case err := <-result:
		if err != nil {
			t.Fatalf("identical retry after overlapping commit = %v", err)
		}
	case <-ctx.Done():
		t.Fatalf("identical retry did not finish: %v", ctx.Err())
	}
}