package connector_events

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

	connector_admission "github.com/pingcap-inc/nexuscrm-connectors/features/connector/admission"
)

func TestService_NamespaceFenceArmWaitsForAlreadyAdmittedMaterialization(t *testing.T) {
	db, cleanup := openTiDB(t)
	defer cleanup()

	const (
		namespace = "ingress-fence-serialized-arm"
		packID    = "install-token-ingress-fence-serialized-arm"
	)
	createInstalledPackTable(t, db, namespace)
	createConnectorEventTable(t, db)
	createConnectorInboundRouteTable(t, db, namespace)
	createConnectorIngressControlTable(t, db)
	setNamespaceFence(t, db, namespace, connector_admission.ModeActive, 0)

	enteredFence := make(chan struct{})
	releaseEnqueue := make(chan struct{})
	svc := NewService(db, deterministicID("event-before-arm", "work-before-arm"), nil)
	svc.afterIngressFenceLock = func() {
		close(enteredFence)
		<-releaseEnqueue
	}

	enqueueDone := make(chan error, 1)
	go func() {
		_, _, err := svc.EnqueueBatch(context.Background(), namespace, packID, "delivery-before-arm", []BatchEvent{{
			EventKind:      "webhook.event",
			IdempotencyKey: "before-arm",
			RouteVersion:   1,
		}})
		enqueueDone <- err
	}()

	select {
	case <-enteredFence:
	case <-time.After(10 * time.Second):
		t.Fatal("enqueue did not acquire the namespace fence")
	}

	// This is the release controller's arm write. It must wait on the very same
	// row lock until the transaction that was admitted before the arm finishes.
	armStarted := make(chan struct{})
	armDone := make(chan error, 1)
	go func() {
		close(armStarted)
		_, err := connector_admission.NewSQLGate(db).TransitionNamespaceFence(context.Background(), connector_admission.NamespaceFenceTransition{
			Namespace: namespace, TargetMode: connector_admission.ModeQuiescing, Reason: "release",
		})
		armDone <- err
	}()
	<-armStarted
	select {
	case err := <-armDone:
		t.Fatalf("arm finished before admitted enqueue released its fence lock: %v", err)
	case <-time.After(250 * time.Millisecond):
	}

	close(releaseEnqueue)
	if err := <-enqueueDone; err != nil {
		t.Fatalf("already-admitted enqueue: %v", err)
	}
	if err := <-armDone; err != nil {
		t.Fatalf("arm namespace fence: %v", err)
	}
	requireIngressMaterializationCounts(t, db, namespace, 1, 1)
}

func TestService_NamespaceFenceRejectsPostArmEnqueueAndFastReadIsNotAuthority(t *testing.T) {
	db, cleanup := openTiDB(t)
	defer cleanup()

	const (
		namespace = "ingress-fence-post-arm"
		packID    = "install-token-ingress-fence-post-arm"
	)
	createInstalledPackTable(t, db, namespace)
	createConnectorEventTable(t, db)
	createConnectorInboundRouteTable(t, db, namespace)
	createConnectorIngressControlTable(t, db)

	// A pre-transaction snapshot is only a latency optimization. It observes
	// active/no-row here, but the arm that follows must still reject the later
	// materialization transaction.
	fastDecision, err := connector_admission.NewSQLGate(db).Decide(context.Background(), namespace, packID, "webhook:default")
	if err != nil || !fastDecision.Allow {
		t.Fatalf("fast Decide before arm = (%+v, %v), want allow", fastDecision, err)
	}
	arm, err := connector_admission.NewSQLGate(db).TransitionNamespaceFence(context.Background(), connector_admission.NamespaceFenceTransition{
		Namespace: namespace, TargetMode: connector_admission.ModeQuiescing, Reason: "release",
	})
	if err != nil || arm.Receipt == nil || arm.Receipt.Epoch != 1 {
		t.Fatalf("arm namespace fence = (%+v, %v), want committed receipt at epoch 1", arm, err)
	}

	_, _, err = NewService(db, deterministicID("event-after-arm", "work-after-arm"), nil).EnqueueBatch(
		context.Background(), namespace, packID, "delivery-after-arm", []BatchEvent{{
			EventKind:      "webhook.event",
			IdempotencyKey: "after-arm",
			RouteVersion:   1,
		}},
	)
	if !errors.Is(err, connector_admission.ErrIngressQuiesced) {
		t.Fatalf("post-arm EnqueueBatch error = %v, want ErrIngressQuiesced", err)
	}
	var quiesced *connector_admission.QuiescedError
	if !errors.As(err, &quiesced) || quiesced.Fence.Epoch != 1 {
		t.Fatalf("post-arm EnqueueBatch error = %v, want typed epoch 1", err)
	}
	requireIngressMaterializationCounts(t, db, namespace, 0, 0)
}

func TestService_NamespaceFenceEpochDistinguishesPauseResumePauseABA(t *testing.T) {
	db, cleanup := openTiDB(t)
	defer cleanup()

	const (
		namespace = "ingress-fence-aba"
		packID    = "install-token-ingress-fence-aba"
	)
	createInstalledPackTable(t, db, namespace)
	createConnectorEventTable(t, db)
	createConnectorInboundRouteTable(t, db, namespace)
	createConnectorIngressControlTable(t, db)
	setNamespaceFence(t, db, namespace, connector_admission.ModeActive, 40)
	gate := connector_admission.NewSQLGate(db)
	arm1, err := gate.TransitionNamespaceFence(context.Background(), connector_admission.NamespaceFenceTransition{
		Namespace: namespace, TargetMode: connector_admission.ModeQuiescing, Reason: "release-1",
	})
	if err != nil || arm1.Receipt == nil || arm1.Receipt.Epoch != 41 {
		t.Fatalf("first arm = (%+v, %v), want committed epoch 41", arm1, err)
	}

	first := requireQuiescedEnqueue(t, db, namespace, packID, "first-pause")
	if _, err := gate.TransitionNamespaceFence(context.Background(), connector_admission.NamespaceFenceTransition{Namespace: namespace, TargetMode: connector_admission.ModeActive}); err != nil {
		t.Fatalf("resume namespace fence: %v", err)
	}
	arm2, err := gate.TransitionNamespaceFence(context.Background(), connector_admission.NamespaceFenceTransition{
		Namespace: namespace, TargetMode: connector_admission.ModeQuiescing, Reason: "release-2",
	})
	if err != nil || arm2.Receipt == nil || arm2.Receipt.Epoch != 42 {
		t.Fatalf("second arm = (%+v, %v), want committed epoch 42", arm2, err)
	}
	second := requireQuiescedEnqueue(t, db, namespace, packID, "second-pause")

	if first.Fence.Epoch != 41 || second.Fence.Epoch != 42 || second.Fence.Epoch <= first.Fence.Epoch {
		t.Fatalf("pause-resume-pause epochs = %d -> %d, want strictly increasing distinct epochs", first.Fence.Epoch, second.Fence.Epoch)
	}
	requireIngressMaterializationCounts(t, db, namespace, 0, 0)
}

func TestService_NamespaceFenceRejectsAmbiguousLegacyGlobalRows(t *testing.T) {
	tests := []struct {
		name            string
		withScopeUnique bool
		expectedRows    int
		insertRows      func(t *testing.T, db *sql.DB, namespace string)
	}{
		{
			name:            "nullable legacy scope beside canonical row",
			withScopeUnique: true,
			expectedRows:    2,
			insertRows: func(t *testing.T, db *sql.DB, namespace string) {
				t.Helper()
				_, err := db.Exec(`
INSERT INTO _System_ConnectorIngressControl
  (Id, Namespace, PackInstallId, Channel, Mode, FenceEpoch, RetryAfterSeconds, IsDeleted, CreatedDate, LastModifiedDate)
VALUES
  (?, ?, '', '*', 'active', 41, 120, 0, CURRENT_TIMESTAMP, CURRENT_TIMESTAMP),
  (?, ?, NULL, '*', 'quiescing', NULL, 120, 0, CURRENT_TIMESTAMP, CURRENT_TIMESTAMP)`,
					connector_admission.NamespaceFenceID(namespace), namespace,
					"legacy-null-scope", namespace)
				if err != nil {
					t.Fatalf("seed nullable legacy global fence rows: %v", err)
				}
			},
		},
		{
			name:            "duplicate canonical global rows",
			withScopeUnique: false,
			expectedRows:    2,
			insertRows: func(t *testing.T, db *sql.DB, namespace string) {
				t.Helper()
				_, err := db.Exec(`
INSERT INTO _System_ConnectorIngressControl
  (Id, Namespace, PackInstallId, Channel, Mode, FenceEpoch, RetryAfterSeconds, IsDeleted, CreatedDate, LastModifiedDate)
VALUES
  (?, ?, '', '*', 'active', 41, 120, 0, CURRENT_TIMESTAMP, CURRENT_TIMESTAMP),
  (?, ?, '', '*', 'quiescing', 40, 120, 0, CURRENT_TIMESTAMP, CURRENT_TIMESTAMP)`,
					connector_admission.NamespaceFenceID(namespace), namespace,
					"duplicate-canonical-scope", namespace)
				if err != nil {
					t.Fatalf("seed duplicate canonical global fence rows: %v", err)
				}
			},
		},
		{
			name:            "canonical global row without epoch",
			withScopeUnique: true,
			expectedRows:    1,
			insertRows: func(t *testing.T, db *sql.DB, namespace string) {
				t.Helper()
				_, err := db.Exec(`
INSERT INTO _System_ConnectorIngressControl
  (Id, Namespace, PackInstallId, Channel, Mode, FenceEpoch, RetryAfterSeconds, IsDeleted, CreatedDate, LastModifiedDate)
VALUES (?, ?, '', '*', 'active', NULL, 120, 0, CURRENT_TIMESTAMP, CURRENT_TIMESTAMP)`,
					connector_admission.NamespaceFenceID(namespace), namespace)
				if err != nil {
					t.Fatalf("seed canonical global fence without epoch: %v", err)
				}
			},
		},
	}

	for _, test := range tests {
		t.Run(test.name, func(t *testing.T) {
			db, cleanup := openTiDB(t)
			defer cleanup()

			const namespace = "ingress-fence-ambiguous-legacy"
			createLegacyConnectorIngressControlTable(t, db, test.withScopeUnique)
			test.insertRows(t, db, namespace)

			result, err := connector_admission.NewSQLGate(db).TransitionNamespaceFence(context.Background(), connector_admission.NamespaceFenceTransition{
				Namespace: namespace, TargetMode: connector_admission.ModeQuiescing, Reason: "release",
			})
			if !errors.Is(err, connector_admission.ErrIngressSchemaUnavailable) {
				t.Fatalf("ambiguous legacy arm = (%+v, %v), want ErrIngressSchemaUnavailable", result, err)
			}
			if result.Receipt != nil {
				t.Fatalf("ambiguous legacy arm receipt = %+v, want none", result.Receipt)
			}

			var rows int
			if err := db.QueryRow(`SELECT COUNT(*) FROM _System_ConnectorIngressControl WHERE Namespace = ? AND Channel = '*'`, namespace).Scan(&rows); err != nil {
				t.Fatalf("count legacy global fence rows: %v", err)
			}
			if rows != test.expectedRows {
				t.Fatalf("legacy global fence rows after rejected arm = %d, want %d (no normalization or selection)", rows, test.expectedRows)
			}
		})
	}
}

func TestService_NamespaceFenceArmWaitsForAlreadyAdmittedReplay(t *testing.T) {
	db, cleanup := openTiDB(t)
	defer cleanup()

	const (
		namespace = "ingress-fence-replay"
		packID    = "install-token-ingress-fence-replay"
	)
	createInstalledPackTable(t, db, namespace)
	createConnectorEventTable(t, db)
	createConnectorInboundRouteTable(t, db, namespace)

	svc := NewService(db, deterministicID("replay-source", "replay-child"), nil)
	sourceID, _, err := svc.Enqueue(context.Background(), namespace, "webhook.event", "replay-source-key", nil, packID, 1, "", "", 0, "delivery-replay")
	if err != nil {
		t.Fatalf("seed replay source: %v", err)
	}
	if _, err := db.Exec(`UPDATE _System_ConnectorEvent SET Status = 'dead' WHERE Id = ?`, sourceID); err != nil {
		t.Fatalf("mark replay source dead: %v", err)
	}

	enteredFence := make(chan struct{})
	releaseReplay := make(chan struct{})
	svc.afterIngressFenceLock = func() {
		close(enteredFence)
		<-releaseReplay
	}
	replayDone := make(chan struct {
		id  string
		err error
	}, 1)
	go func() {
		id, replayErr := svc.ReplayDeadEvent(context.Background(), allowPerm("ReplayConnectorEvents"), sourceID, packID, 1)
		replayDone <- struct {
			id  string
			err error
		}{id: id, err: replayErr}
	}()

	select {
	case <-enteredFence:
	case <-time.After(10 * time.Second):
		t.Fatal("replay did not acquire the namespace admission fence")
	}

	armDone := make(chan error, 1)
	go func() {
		_, armErr := connector_admission.NewSQLGate(db).TransitionNamespaceFence(context.Background(), connector_admission.NamespaceFenceTransition{
			Namespace: namespace, TargetMode: connector_admission.ModeQuiescing, Reason: "release",
		})
		armDone <- armErr
	}()
	select {
	case err := <-armDone:
		t.Fatalf("arm completed before replay released the shared fence: %v", err)
	case <-time.After(250 * time.Millisecond):
	}

	close(releaseReplay)
	replay := <-replayDone
	if replay.err != nil || replay.id == "" {
		t.Fatalf("already-admitted ReplayDeadEvent = (%q, %v)", replay.id, replay.err)
	}
	if err := <-armDone; err != nil {
		t.Fatalf("arm after replay commit: %v", err)
	}

	var replayRows int
	if err := db.QueryRow(`SELECT COUNT(*) FROM _System_ConnectorEvent WHERE ReplayOfId = ?`, sourceID).Scan(&replayRows); err != nil {
		t.Fatalf("count replay children: %v", err)
	}
	if replayRows != 1 {
		t.Fatalf("replay children = %d, want exactly one committed before arm", replayRows)
	}
}

func createConnectorIngressControlTable(t *testing.T, db *sql.DB) {
	t.Helper()
	_, err := db.Exec(`CREATE TABLE IF NOT EXISTS _System_ConnectorIngressControl (
		Id VARCHAR(255) PRIMARY KEY,
		Namespace VARCHAR(255) NOT NULL,
		PackInstallId VARCHAR(255) NOT NULL DEFAULT '',
		Channel VARCHAR(255) NOT NULL,
		Mode VARCHAR(32) NOT NULL DEFAULT 'active',
		FenceEpoch BIGINT NOT NULL DEFAULT 0,
		RetryAfterSeconds INT NOT NULL DEFAULT 120,
		Reason VARCHAR(255),
		ExpiresAt DATETIME NULL,
		IsDeleted TINYINT(1) NOT NULL DEFAULT 0,
		CreatedDate DATETIME NOT NULL DEFAULT CURRENT_TIMESTAMP,
		LastModifiedDate DATETIME NOT NULL DEFAULT CURRENT_TIMESTAMP,
		UNIQUE KEY uniq_connector_ingress_control_scope (Namespace, PackInstallId, Channel)
	)`)
	if err != nil {
		t.Fatalf("create _System_ConnectorIngressControl: %v", err)
	}
}

func createLegacyConnectorIngressControlTable(t *testing.T, db *sql.DB, withScopeUnique bool) {
	t.Helper()
	ddl := `CREATE TABLE _System_ConnectorIngressControl (
		Id VARCHAR(255) PRIMARY KEY,
		Namespace VARCHAR(255) NOT NULL,
		PackInstallId VARCHAR(255) NULL,
		Channel VARCHAR(255) NOT NULL,
		Mode VARCHAR(32) NOT NULL DEFAULT 'active',
		FenceEpoch BIGINT NULL,
		RetryAfterSeconds INT NOT NULL DEFAULT 120,
		Reason VARCHAR(255),
		ExpiresAt DATETIME NULL,
		IsDeleted TINYINT(1) NOT NULL DEFAULT 0,
		CreatedDate DATETIME NOT NULL DEFAULT CURRENT_TIMESTAMP,
		LastModifiedDate DATETIME NOT NULL DEFAULT CURRENT_TIMESTAMP`
	if withScopeUnique {
		ddl += `, UNIQUE KEY uniq_connector_ingress_control_scope (Namespace, PackInstallId, Channel)`
	}
	ddl += `)`
	if _, err := db.Exec(ddl); err != nil {
		t.Fatalf("create legacy _System_ConnectorIngressControl: %v", err)
	}
}

func setNamespaceFence(t *testing.T, db *sql.DB, namespace, mode string, epoch int64) {
	t.Helper()
	if err := updateNamespaceFence(context.Background(), db, namespace, mode, epoch); err != nil {
		t.Fatalf("set namespace fence mode=%q epoch=%d: %v", mode, epoch, err)
	}
}

func updateNamespaceFence(ctx context.Context, db *sql.DB, namespace, mode string, epoch int64) error {
	_, err := db.ExecContext(ctx, `
INSERT INTO _System_ConnectorIngressControl
  (Id, Namespace, PackInstallId, Channel, Mode, FenceEpoch, RetryAfterSeconds, IsDeleted, CreatedDate, LastModifiedDate)
VALUES (?, ?, '', '*', ?, ?, 120, 0, CURRENT_TIMESTAMP, CURRENT_TIMESTAMP)
ON DUPLICATE KEY UPDATE
  Mode = VALUES(Mode),
  FenceEpoch = VALUES(FenceEpoch),
  IsDeleted = 0,
  LastModifiedDate = CURRENT_TIMESTAMP`,
		connector_admission.NamespaceFenceID(namespace), namespace, mode, epoch)
	return err
}

func requireQuiescedEnqueue(t *testing.T, db *sql.DB, namespace, packID, idempotencyKey string) *connector_admission.QuiescedError {
	t.Helper()
	_, _, err := NewService(db, deterministicID("event-"+idempotencyKey, "work-"+idempotencyKey), nil).EnqueueBatch(
		context.Background(), namespace, packID, "delivery-"+idempotencyKey, []BatchEvent{{
			EventKind:      "webhook.event",
			IdempotencyKey: idempotencyKey,
			RouteVersion:   1,
		}},
	)
	if !errors.Is(err, connector_admission.ErrIngressQuiesced) {
		t.Fatalf("EnqueueBatch error = %v, want ErrIngressQuiesced", err)
	}
	var quiesced *connector_admission.QuiescedError
	if !errors.As(err, &quiesced) {
		t.Fatalf("EnqueueBatch error = %T, want *QuiescedError", err)
	}
	return quiesced
}

func requireIngressMaterializationCounts(t *testing.T, db *sql.DB, namespace string, wantEvents, wantWorkItems int) {
	t.Helper()
	var events int
	if err := db.QueryRow(`SELECT COUNT(*) FROM _System_ConnectorEvent WHERE Namespace = ?`, namespace).Scan(&events); err != nil {
		t.Fatalf("count connector events: %v", err)
	}
	if events != wantEvents {
		t.Fatalf("connector event count = %d, want %d", events, wantEvents)
	}
	var workItems int
	if err := db.QueryRow(`SELECT COUNT(*) FROM _System_ConnectorWorkItem`).Scan(&workItems); err != nil {
		t.Fatalf("count connector work items: %v", err)
	}
	if workItems != wantWorkItems {
		t.Fatalf("connector work item count = %d, want %d", workItems, wantWorkItems)
	}
}