package serving

import (
	"context"
	"database/sql"
	"errors"
	"testing"
	"time"

	"github.com/pingcap-inc/nexuscrm-connectors/internal/sqltx"
)

func TestAccountServingGateSchemaReadinessAgainstTiDB(t *testing.T) {
	db, cleanup := openServingEpochTestTiDB(t)
	defer cleanup()
	createServingEpochTestTables(t, db)
	createAccountServingGateTestTable(t, db)

	if err := verifyAccountServingGateSchema(t.Context(), db); err != nil {
		t.Fatalf("verify exact AccountServingGate schema: %v", err)
	}

	if _, err := db.ExecContext(t.Context(), `ALTER TABLE _System_ConnectorAccountServingGate
  DROP INDEX ux_connector_account_gate_scope_account`); err != nil {
		t.Fatalf("drop exact scoped-account unique index: %v", err)
	}
	if err := verifyAccountServingGateSchema(t.Context(), db); !errors.Is(err, errAccountServingGateSchemaUnavailable) {
		t.Fatalf("verify drifted AccountServingGate schema error = %v, want schema unavailable", err)
	}
}

func TestAccountServingGateExpectedFenceSerializationAgainstTiDB(t *testing.T) {
	const (
		gateID             = "account-serving-gate-hot"
		connectorAccountID = "connector-account-hot"
		expectedFence      = int64(41)
	)

	db, cleanup := openServingEpochTestTiDB(t)
	defer cleanup()
	createServingEpochTestTables(t, db)
	createAccountServingGateTestTable(t, db)
	createAccountServingGateEffectFixtureTable(t, db)
	db.SetMaxOpenConns(8)
	db.SetMaxIdleConns(8)

	candidate := seedActiveProfiledEpochForAccountGateTest(t, db)
	seedAccountServingGateTestRow(
		t,
		db,
		gateID,
		candidate.ServingScopeDigest(),
		connectorAccountID,
		expectedFence,
		candidate.ID(),
		candidate.EpochDigest(),
	)
	if err := verifyAccountServingGateSchema(t.Context(), db); err != nil {
		t.Fatalf("verify AccountServingGate schema before serialization proof: %v", err)
	}

	winnerTx, winnerCleanup, err := sqltx.BeginPessimisticTx(t.Context(), db)
	if err != nil {
		t.Fatalf("begin winning gate transaction: %v", err)
	}
	winnerOpen := true
	defer func() {
		if winnerOpen {
			_ = winnerTx.Rollback()
			_ = winnerCleanup()
		}
	}()
	if _, err := lockExpectedV2Serving(
		t.Context(),
		winnerTx,
		candidate.ServingScopeDigest(),
		connectorAccountID,
		expectedFence,
		candidate.ID(),
		candidate.EpochDigest(),
	); err != nil {
		t.Fatalf("lock exact v2_serving gate in winning transaction: %v", err)
	}

	waiterTx, waiterCleanup, err := sqltx.BeginPessimisticTx(t.Context(), db)
	if err != nil {
		t.Fatalf("begin waiting gate transaction: %v", err)
	}
	waiterContext, cancelWaiter := context.WithTimeout(context.Background(), 10*time.Second)
	defer cancelWaiter()
	type waiterResult struct {
		guardErr  error
		finishErr error
	}
	waiterStarted := make(chan struct{})
	waiterDone := make(chan waiterResult, 1)
	go func() {
		close(waiterStarted)
		_, guardErr := lockExpectedV2Serving(
			waiterContext,
			waiterTx,
			candidate.ServingScopeDigest(),
			connectorAccountID,
			expectedFence,
			candidate.ID(),
			candidate.EpochDigest(),
		)
		if guardErr != nil {
			waiterDone <- waiterResult{
				guardErr:  guardErr,
				finishErr: errors.Join(waiterTx.Rollback(), waiterCleanup()),
			}
			return
		}
		_, effectErr := waiterTx.ExecContext(waiterContext, `
INSERT INTO _Fixture_AccountServingGateEffect (Id, GateId, ObservedFenceToken)
VALUES ('stale-waiter-effect', ?, ?)`, gateID, expectedFence)
		waiterDone <- waiterResult{
			guardErr:  effectErr,
			finishErr: errors.Join(waiterTx.Commit(), waiterCleanup()),
		}
	}()
	<-waiterStarted

	select {
	case result := <-waiterDone:
		t.Fatalf("waiting gate transaction returned before the lock owner committed: %+v", result)
	case <-time.After(150 * time.Millisecond):
	}

	transitionResult, err := winnerTx.ExecContext(t.Context(), `
UPDATE _System_ConnectorAccountServingGate
   SET Mode = 'halting_v2',
       FenceToken = FenceToken + 1,
       LastModifiedDate = CURRENT_TIMESTAMP(6)
 WHERE ServingScopeDigest = ? AND BINARY ServingScopeDigest = BINARY ?
   AND ConnectorAccountId = ? AND BINARY ConnectorAccountId = BINARY ?
   AND Mode = 'v2_serving' AND BINARY Mode = BINARY 'v2_serving'
   AND FenceToken = ?`,
		candidate.ServingScopeDigest(), candidate.ServingScopeDigest(),
		connectorAccountID, connectorAccountID,
		expectedFence,
	)
	if err != nil {
		t.Fatalf("apply fixture-only fenced halt transition: %v", err)
	}
	if rows, err := transitionResult.RowsAffected(); err != nil || rows != 1 {
		t.Fatalf("fixture-only fenced halt rows = %d, %v, want 1", rows, err)
	}
	if err := winnerTx.Commit(); err != nil {
		t.Fatalf("commit winning fenced halt: %v", err)
	}
	winnerOpen = false
	if err := winnerCleanup(); err != nil {
		t.Fatalf("cleanup winning fenced halt connection: %v", err)
	}

	var waited waiterResult
	select {
	case waited = <-waiterDone:
	case <-time.After(10 * time.Second):
		t.Fatal("stale expected-fence waiter did not finish after the winning commit")
	}
	if !errors.Is(waited.guardErr, errAccountServingGateFenceConflict) {
		t.Fatalf("stale expected-fence waiter error = %v, want fence conflict", waited.guardErr)
	}
	if waited.finishErr != nil {
		t.Fatalf("finish stale expected-fence waiter: %v", waited.finishErr)
	}

	var effectCount int
	if err := db.QueryRowContext(t.Context(), `
SELECT COUNT(*) FROM _Fixture_AccountServingGateEffect`).Scan(&effectCount); err != nil {
		t.Fatalf("count stale waiter fixture effects: %v", err)
	}
	if effectCount != 0 {
		t.Fatalf("stale expected-fence waiter effects = %d, want 0", effectCount)
	}

	var mode string
	var fence int64
	if err := db.QueryRowContext(t.Context(), `
SELECT Mode, FenceToken
  FROM _System_ConnectorAccountServingGate
 WHERE ServingScopeDigest = ? AND BINARY ServingScopeDigest = BINARY ?
   AND ConnectorAccountId = ? AND BINARY ConnectorAccountId = BINARY ?`,
		candidate.ServingScopeDigest(), candidate.ServingScopeDigest(),
		connectorAccountID, connectorAccountID,
	).Scan(&mode, &fence); err != nil {
		t.Fatalf("read final AccountServingGate: %v", err)
	}
	if mode != "halting_v2" || fence != expectedFence+1 {
		t.Fatalf("final AccountServingGate = mode:%q fence:%d, want halting_v2/%d", mode, fence, expectedFence+1)
	}
}

func createAccountServingGateTestTable(t *testing.T, db *sql.DB) {
	t.Helper()
	if _, err := db.ExecContext(t.Context(), accountServingGateTestDDL); err != nil {
		t.Fatalf("create AccountServingGate fixture schema: %v", err)
	}
}

func createAccountServingGateEffectFixtureTable(t *testing.T, db *sql.DB) {
	t.Helper()
	if _, err := db.ExecContext(t.Context(), accountServingGateEffectFixtureTestDDL); err != nil {
		t.Fatalf("create AccountServingGate effect fixture schema: %v", err)
	}
}

func seedActiveProfiledEpochForAccountGateTest(t *testing.T, db *sql.DB) ProfiledPreparedEpoch {
	t.Helper()
	repository := newServingEpochTestRepository(t, db)
	fixture := newProfiledVerificationTiDBFixture(t, "epoch-account-serving-gate-hot")
	seedProfiledWriterTiDBOwnerRows(t, db, fixture.candidate, "")
	prepareProfiledWriterTiDBCandidate(t, repository, fixture.candidate)
	service, _ := newProfiledVerificationTiDBService(t, repository, fixture)
	result, err := service.Verify(t.Context(), fixture.candidate.ID())
	if err != nil || result.State() != ProfiledEpochVerified || !result.Applied() {
		t.Fatalf("verify AccountServingGate fixture epoch = %#v, %v", result, err)
	}

	activated, err := db.ExecContext(t.Context(), `
UPDATE _System_ConnectorServingEpoch
   SET State = 'active',
       ActiveScopeDigest = ServingScopeDigest,
       LastModifiedDate = CURRENT_TIMESTAMP(6)
 WHERE Id = ? AND BINARY Id = BINARY ?
   AND EpochDigest = ? AND BINARY EpochDigest = BINARY ?
   AND State = 'verified' AND BINARY State = BINARY 'verified'
   AND ActiveScopeDigest IS NULL`,
		fixture.candidate.ID(), fixture.candidate.ID(),
		fixture.candidate.EpochDigest(), fixture.candidate.EpochDigest(),
	)
	if err != nil {
		t.Fatalf("activate fixture-only AccountServingGate epoch: %v", err)
	}
	if rows, err := activated.RowsAffected(); err != nil || rows != 1 {
		t.Fatalf("fixture-only activation rows = %d, %v, want 1", rows, err)
	}
	return fixture.candidate
}

func seedAccountServingGateTestRow(
	t *testing.T,
	db *sql.DB,
	gateID, servingScopeDigest, connectorAccountID string,
	fenceToken int64,
	servingEpochID, servingEpochDigest string,
) {
	t.Helper()
	if _, err := db.ExecContext(t.Context(), `
INSERT INTO _System_ConnectorAccountServingGate
       (Id, ServingScopeDigest, ConnectorAccountId, Mode, FenceToken,
        ServingEpochId, ServingEpochDigest)
VALUES (?, ?, ?, 'v2_serving', ?, ?, ?)`,
		gateID,
		servingScopeDigest,
		connectorAccountID,
		fenceToken,
		servingEpochID,
		servingEpochDigest,
	); err != nil {
		t.Fatalf("seed exact AccountServingGate row: %v", err)
	}
}

const accountServingGateTestDDL = `CREATE TABLE _System_ConnectorAccountServingGate (
  Id VARCHAR(255) CHARACTER SET utf8mb4 COLLATE utf8mb4_bin NOT NULL,
  ServingScopeDigest CHAR(64) CHARACTER SET utf8mb4 COLLATE utf8mb4_bin NOT NULL,
  ConnectorAccountId VARCHAR(255) CHARACTER SET utf8mb4 COLLATE utf8mb4_bin NOT NULL,
  Mode VARCHAR(32) CHARACTER SET utf8mb4 COLLATE utf8mb4_bin NOT NULL DEFAULT 'v1_serving',
  FenceToken BIGINT NOT NULL DEFAULT 0,
  ServingEpochId VARCHAR(255) CHARACTER SET utf8mb4 COLLATE utf8mb4_bin NULL,
  ServingEpochDigest CHAR(71) CHARACTER SET utf8mb4 COLLATE utf8mb4_bin NULL,
  CreatedDate DATETIME(6) NOT NULL DEFAULT CURRENT_TIMESTAMP(6),
  LastModifiedDate DATETIME(6) NOT NULL DEFAULT CURRENT_TIMESTAMP(6),
  PRIMARY KEY (Id),
  UNIQUE KEY ux_connector_account_gate_scope_account (ServingScopeDigest, ConnectorAccountId),
  KEY idx_connector_account_gate_mode (ServingScopeDigest, Mode, ConnectorAccountId),
  KEY idx_connector_account_gate_epoch (ServingEpochId, Mode, ConnectorAccountId),
  CONSTRAINT fk_connector_account_serving_gate_epoch
    FOREIGN KEY (ServingEpochId) REFERENCES _System_ConnectorServingEpoch (Id)
    ON UPDATE RESTRICT ON DELETE RESTRICT
) DEFAULT CHARACTER SET utf8mb4 COLLATE utf8mb4_bin`

const accountServingGateEffectFixtureTestDDL = `CREATE TABLE _Fixture_AccountServingGateEffect (
  Id VARCHAR(255) CHARACTER SET utf8mb4 COLLATE utf8mb4_bin NOT NULL,
  GateId VARCHAR(255) CHARACTER SET utf8mb4 COLLATE utf8mb4_bin NOT NULL,
  ObservedFenceToken BIGINT NOT NULL,
  PRIMARY KEY (Id)
) DEFAULT CHARACTER SET utf8mb4 COLLATE utf8mb4_bin`