package triggersub

import (
	"context"
	"database/sql"
	"fmt"
	"os"
	"testing"
	"time"

	_ "github.com/go-sql-driver/mysql"
)

func TestBackfillWritesExpectedRowsAndIsIdempotent(t *testing.T) {
	ctx := context.Background()
	db, cleanup := openTriggerSubscriptionTestDB(t)
	defer cleanup()

	seedTriggerSubscriptionObject(t, ctx, db, "obj-probe", "ProbeObject", "a00")
	seedTriggerSubscriptionFlowDefinition(t, ctx, db, "300000000000001AAA", "301000000000001AAA", "AutoLaunchedFlow")
	seedTriggerSubscriptionFlowVersion(t, ctx, db, "301000000000001AAA", "300000000000001AAA", "AutoLaunchedFlow", modelsFlowAfterSaveCreateAndUpdate())

	if err := Backfill(ctx, db); err != nil {
		t.Fatalf("Backfill() first run error = %v", err)
	}
	assertTriggerSubscriptionRows(t, ctx, db, "300000000000001AAA", "301000000000001AAA", []string{"Create", "Update"})

	if err := Backfill(ctx, db); err != nil {
		t.Fatalf("Backfill() second run error = %v", err)
	}
	assertTriggerSubscriptionRows(t, ctx, db, "300000000000001AAA", "301000000000001AAA", []string{"Create", "Update"})
}

func TestBackfillUsesWorkflowProjectionResolver(t *testing.T) {
	ctx := context.Background()
	db, cleanup := openTriggerSubscriptionTestDB(t)
	defer cleanup()

	seedTriggerSubscriptionObject(t, ctx, db, "obj-probe", "ProbeObject", "a00")
	seedTriggerSubscriptionFlowDefinition(t, ctx, db, "300000000000001AAA", "301000000000001AAA", "Workflow")
	seedTriggerSubscriptionFlowVersion(t, ctx, db, "301000000000001AAA", "300000000000001AAA", "Workflow", modelsWorkflowMetadata())

	if err := Backfill(ctx, db); err != nil {
		t.Fatalf("Backfill() error = %v", err)
	}
	assertTriggerSubscriptionRows(t, ctx, db, "300000000000001AAA", "301000000000001AAA", []string{"Create", "Update"})
}

func TestBackfillSkipsUnsupportedAtomicTriggerShape(t *testing.T) {
	ctx := context.Background()
	db, cleanup := openTriggerSubscriptionTestDB(t)
	defer cleanup()

	seedTriggerSubscriptionObject(t, ctx, db, "obj-probe", "ProbeObject", "a00")
	seedTriggerSubscriptionFlowDefinition(t, ctx, db, "300000000000001AAA", "301000000000001AAA", "Workflow")
	seedTriggerSubscriptionFlowVersion(t, ctx, db, "301000000000001AAA", "300000000000001AAA", "Workflow", modelsAtomicWorkflowMetadata())

	if err := Backfill(ctx, db); err != nil {
		t.Fatalf("Backfill() error = %v, want nil skip path", err)
	}
	if got := countTriggerSubscriptionRows(t, ctx, db, "300000000000001AAA"); got != 0 {
		t.Fatalf("trigger subscription row count = %d, want 0 after skip", got)
	}
}

func TestReconcileDefinitionSkipsUnsupportedAtomicTriggerShape(t *testing.T) {
	ctx := context.Background()
	db, cleanup := openTriggerSubscriptionTestDB(t)
	defer cleanup()

	seedTriggerSubscriptionObject(t, ctx, db, "obj-probe", "ProbeObject", "a00")
	seedTriggerSubscriptionFlowDefinition(t, ctx, db, "300000000000001AAA", "301000000000001AAA", "Workflow")
	seedTriggerSubscriptionFlowVersion(t, ctx, db, "301000000000001AAA", "300000000000001AAA", "Workflow", modelsWorkflowMetadata())

	if err := ReconcileDefinition(ctx, db, "300000000000001AAA"); err != nil {
		t.Fatalf("ReconcileDefinition() seed run error = %v", err)
	}
	assertTriggerSubscriptionRows(t, ctx, db, "300000000000001AAA", "301000000000001AAA", []string{"Create", "Update"})

	if _, err := db.ExecContext(ctx, `UPDATE Flow SET Metadata = ? WHERE Id = ?`, modelsAtomicWorkflowMetadata(), "301000000000001AAA"); err != nil {
		t.Fatalf("update Flow metadata: %v", err)
	}

	if err := ReconcileDefinition(ctx, db, "300000000000001AAA"); err != nil {
		t.Fatalf("ReconcileDefinition() error = %v, want nil skip path", err)
	}
	if got := countActiveTriggerSubscriptionRows(t, ctx, db, "300000000000001AAA"); got != 0 {
		t.Fatalf("active trigger subscription row count = %d, want 0 after skip", got)
	}
}

func TestBackfillLeavesExistingRowsActiveWhenReplacementFails(t *testing.T) {
	ctx := context.Background()
	db, cleanup := openTriggerSubscriptionTestDB(t)
	defer cleanup()

	seedTriggerSubscriptionObject(t, ctx, db, "obj-probe", "ProbeObject", "a00")
	seedTriggerSubscriptionFlowDefinition(t, ctx, db, "300000000000001AAA", "301000000000001AAA", "AutoLaunchedFlow")
	seedTriggerSubscriptionFlowVersion(t, ctx, db, "301000000000001AAA", "300000000000001AAA", "AutoLaunchedFlow", modelsFlowAfterSaveCreateAndUpdate())

	if err := Backfill(ctx, db); err != nil {
		t.Fatalf("Backfill() seed run error = %v", err)
	}
	assertTriggerSubscriptionRows(t, ctx, db, "300000000000001AAA", "301000000000001AAA", []string{"Create", "Update"})

	seedTriggerSubscriptionFlowDefinition(t, ctx, db, "300000000000002AAA", "301000000000002AAA", "AutoLaunchedFlow")
	seedTriggerSubscriptionFlowVersion(t, ctx, db, "301000000000002AAA", "300000000000002AAA", "AutoLaunchedFlow", modelsFlowAfterSaveMissingObject())

	if err := Backfill(ctx, db); err == nil {
		t.Fatalf("Backfill() error = nil, want missing object metadata error")
	}

	assertTriggerSubscriptionRows(t, ctx, db, "300000000000001AAA", "301000000000001AAA", []string{"Create", "Update"})
	if got := countTriggerSubscriptionRows(t, ctx, db, "300000000000002AAA"); got != 0 {
		t.Fatalf("replacement failure wrote %d rows for broken definition, want 0", got)
	}
}

func TestReconcileDefinitionLeavesExistingRowsActiveWhenReplacementFails(t *testing.T) {
	ctx := context.Background()
	db, cleanup := openTriggerSubscriptionTestDB(t)
	defer cleanup()

	seedTriggerSubscriptionObject(t, ctx, db, "obj-probe", "ProbeObject", "a00")
	seedTriggerSubscriptionFlowDefinition(t, ctx, db, "300000000000001AAA", "301000000000001AAA", "AutoLaunchedFlow")
	seedTriggerSubscriptionFlowVersion(t, ctx, db, "301000000000001AAA", "300000000000001AAA", "AutoLaunchedFlow", modelsFlowAfterSaveCreateAndUpdate())

	if err := ReconcileDefinition(ctx, db, "300000000000001AAA"); err != nil {
		t.Fatalf("ReconcileDefinition() seed run error = %v", err)
	}
	assertTriggerSubscriptionRows(t, ctx, db, "300000000000001AAA", "301000000000001AAA", []string{"Create", "Update"})

	if _, err := db.ExecContext(ctx, `UPDATE Flow SET Metadata = ? WHERE Id = ?`, modelsFlowAfterSaveInvalidRecordTriggerType(), "301000000000001AAA"); err != nil {
		t.Fatalf("update Flow metadata: %v", err)
	}

	if err := ReconcileDefinition(ctx, db, "300000000000001AAA"); err == nil {
		t.Fatalf("ReconcileDefinition() error = nil, want invalid record trigger type error")
	}

	assertTriggerSubscriptionRows(t, ctx, db, "300000000000001AAA", "301000000000001AAA", []string{"Create", "Update"})
}

func openTriggerSubscriptionTestDB(t *testing.T) (*sql.DB, func()) {
	t.Helper()

	host := getenvDefault("TIDB_HOST", "127.0.0.1")
	port := getenvDefault("TIDB_PORT", "4001")
	user := getenvDefault("TIDB_USER", "root")
	password := os.Getenv("TIDB_PASSWORD")
	adminDSN := fmt.Sprintf("%s:%s@tcp(%s:%s)/?charset=utf8mb4&parseTime=True&loc=Local", user, password, host, port)

	adminDB, err := sql.Open("mysql", adminDSN)
	if err != nil {
		t.Skipf("skipping integration test: open admin TiDB: %v", err)
	}
	if err := adminDB.Ping(); err != nil {
		_ = adminDB.Close()
		t.Skipf("skipping integration test: TiDB not reachable at %s:%s: %v", host, port, err)
	}

	dbName := fmt.Sprintf("nexuscrm_triggersub_%d", time.Now().UnixNano())
	if _, err := adminDB.Exec("CREATE DATABASE " + dbName); err != nil {
		_ = adminDB.Close()
		t.Skipf("skipping integration test: create temp database: %v", err)
	}

	dsn := fmt.Sprintf("%s:%s@tcp(%s:%s)/%s?charset=utf8mb4&parseTime=True&loc=Local", user, password, host, port, dbName)
	db, err := sql.Open("mysql", dsn)
	if err != nil {
		_, _ = adminDB.Exec("DROP DATABASE " + dbName)
		_ = adminDB.Close()
		t.Fatalf("open temp database: %v", err)
	}
	createTriggerSubscriptionTables(t, db)

	cleanup := func() {
		_ = db.Close()
		_, _ = adminDB.Exec("DROP DATABASE " + dbName)
		_ = adminDB.Close()
	}
	return db, cleanup
}

func createTriggerSubscriptionTables(t *testing.T, db *sql.DB) {
	t.Helper()
	stmts := []string{
		`CREATE TABLE _System_Object (
			Id VARCHAR(64) PRIMARY KEY,
			ApiName VARCHAR(255) NOT NULL,
			KeyPrefix VARCHAR(3) NULL,
			IsDeleted BOOL DEFAULT FALSE,
			DisplayNameField VARCHAR(255) NULL
		)`,
		`CREATE TABLE _System_TriggerSubscription (
			Id VARCHAR(255) PRIMARY KEY,
			KeyPrefix CHAR(3) COLLATE utf8mb4_bin NOT NULL,
			EventType VARCHAR(32) NOT NULL,
			RecordTriggerType VARCHAR(32) NOT NULL,
			FlowDefinitionId VARCHAR(64) NOT NULL,
			FlowVersionId VARCHAR(64) NOT NULL,
			IsActive BOOL NOT NULL DEFAULT TRUE,
			ConditionShapeHash CHAR(64) NULL,
			CreatedDate DATETIME NOT NULL DEFAULT CURRENT_TIMESTAMP,
			LastModifiedDate DATETIME NOT NULL DEFAULT CURRENT_TIMESTAMP,
			UNIQUE KEY uq_trigger_subscription_identity (FlowDefinitionId, KeyPrefix, EventType, RecordTriggerType),
			KEY idx_trigger_subscription_dispatch (KeyPrefix, EventType, RecordTriggerType, IsActive)
		)`,
		`CREATE TABLE FlowDefinition (
			Id VARCHAR(64) PRIMARY KEY,
			DeveloperName VARCHAR(255),
			MasterLabel VARCHAR(255),
			Description TEXT,
			ProcessType VARCHAR(64),
			TriggerType VARCHAR(64),
			TriggerObjectOrEvent VARCHAR(255),
			ActiveVersionId VARCHAR(64) NULL,
			LatestVersionId VARCHAR(64) NULL,
			IsDeleted BOOL DEFAULT FALSE,
			IsCustomized BOOL DEFAULT FALSE,
			OwnerPack VARCHAR(255) NULL,
			CreatedDate VARCHAR(64),
			LastModifiedDate VARCHAR(64)
		)`,
		`CREATE TABLE Flow (
			Id VARCHAR(64) PRIMARY KEY,
			DefinitionId VARCHAR(64),
			MasterLabel VARCHAR(255),
			VersionNumber INT,
			Status VARCHAR(64),
			Description TEXT,
			ProcessType VARCHAR(64),
			Metadata TEXT,
			Source VARCHAR(64),
			IsDeleted BOOL DEFAULT FALSE,
			CreatedDate VARCHAR(64),
			LastModifiedDate VARCHAR(64),
			INDEX idx_flow_definition_version (DefinitionId, VersionNumber)
		)`,
	}
	for _, stmt := range stmts {
		if _, err := db.Exec(stmt); err != nil {
			t.Fatalf("create test table: %v", err)
		}
	}
}

func TestListTreatsKeyPrefixCaseSensitively(t *testing.T) {
	ctx := context.Background()
	db, cleanup := openTriggerSubscriptionTestDB(t)
	defer cleanup()

	rows := []Row{
		{Id: "row-upper", KeyPrefix: "00K", EventType: "RecordAfterSave", RecordTriggerType: "Create", FlowDefinitionId: "fd-upper", FlowVersionId: "fv-upper", IsActive: true},
		{Id: "row-lower", KeyPrefix: "00k", EventType: "RecordAfterSave", RecordTriggerType: "Create", FlowDefinitionId: "fd-lower", FlowVersionId: "fv-lower", IsActive: true},
	}
	for _, row := range rows {
		if _, err := db.ExecContext(ctx, `INSERT INTO _System_TriggerSubscription
			(Id, KeyPrefix, EventType, RecordTriggerType, FlowDefinitionId, FlowVersionId, IsActive)
			VALUES (?, ?, ?, ?, ?, ?, ?)`,
			row.Id, row.KeyPrefix, row.EventType, row.RecordTriggerType, row.FlowDefinitionId, row.FlowVersionId, row.IsActive); err != nil {
			t.Fatalf("seed trigger subscription row %s: %v", row.Id, err)
		}
	}

	gotUpper, err := List(ctx, db, "00K", "RecordAfterSave", "Create")
	if err != nil {
		t.Fatalf("List(00K) error = %v", err)
	}
	if len(gotUpper) != 1 || gotUpper[0].Id != "row-upper" {
		t.Fatalf("List(00K) = %+v, want only row-upper", gotUpper)
	}

	gotLower, err := List(ctx, db, "00k", "RecordAfterSave", "Create")
	if err != nil {
		t.Fatalf("List(00k) error = %v", err)
	}
	if len(gotLower) != 1 || gotLower[0].Id != "row-lower" {
		t.Fatalf("List(00k) = %+v, want only row-lower", gotLower)
	}
}

func seedTriggerSubscriptionObject(t *testing.T, ctx context.Context, db *sql.DB, id string, apiName string, keyPrefix string) {
	t.Helper()
	if _, err := db.ExecContext(ctx, `INSERT INTO _System_Object (Id, ApiName, KeyPrefix, IsDeleted) VALUES (?, ?, ?, 0)`, id, apiName, keyPrefix); err != nil {
		t.Fatalf("seed _System_Object: %v", err)
	}
}

func seedTriggerSubscriptionFlowDefinition(t *testing.T, ctx context.Context, db *sql.DB, id string, activeVersionID string, processType string) {
	t.Helper()
	now := time.Now().Format(time.RFC3339)
	if _, err := db.ExecContext(ctx, `INSERT INTO FlowDefinition
		(Id, DeveloperName, MasterLabel, Description, ProcessType, TriggerType, TriggerObjectOrEvent,
		 ActiveVersionId, LatestVersionId, IsDeleted, IsCustomized, OwnerPack, CreatedDate, LastModifiedDate)
		VALUES (?, 'Probe_Flow', 'Probe Flow', '', ?, '', '', ?, ?, 0, 0, '', ?, ?)`,
		id, processType, activeVersionID, activeVersionID, now, now); err != nil {
		t.Fatalf("seed FlowDefinition: %v", err)
	}
}

func seedTriggerSubscriptionFlowVersion(t *testing.T, ctx context.Context, db *sql.DB, id string, definitionID string, processType string, metadata string) {
	t.Helper()
	now := time.Now().Format(time.RFC3339)
	if _, err := db.ExecContext(ctx, `INSERT INTO Flow
		(Id, DefinitionId, MasterLabel, VersionNumber, Status, Description, ProcessType, Metadata, Source, IsDeleted, CreatedDate, LastModifiedDate)
		VALUES (?, ?, 'Probe Flow', 1, 'Active', '', ?, ?, 'custom', 0, ?, ?)`,
		id, definitionID, processType, metadata, now, now); err != nil {
		t.Fatalf("seed Flow: %v", err)
	}
}

func assertTriggerSubscriptionRows(t *testing.T, ctx context.Context, db *sql.DB, definitionID string, versionID string, wantRecordTriggerTypes []string) {
	t.Helper()
	rows, err := db.QueryContext(ctx, `
		SELECT EventType, RecordTriggerType, FlowVersionId, IsActive
		FROM _System_TriggerSubscription
		WHERE FlowDefinitionId = ?
		ORDER BY RecordTriggerType`, definitionID)
	if err != nil {
		t.Fatalf("query TriggerSubscription rows: %v", err)
	}
	defer rows.Close()

	var got []string
	for rows.Next() {
		var eventType string
		var recordTriggerType string
		var gotVersionID string
		var isActive bool
		if err := rows.Scan(&eventType, &recordTriggerType, &gotVersionID, &isActive); err != nil {
			t.Fatalf("scan TriggerSubscription row: %v", err)
		}
		if eventType != "RecordAfterSave" {
			t.Fatalf("EventType = %q, want RecordAfterSave", eventType)
		}
		if gotVersionID != versionID {
			t.Fatalf("FlowVersionId = %q, want %q", gotVersionID, versionID)
		}
		if !isActive {
			t.Fatalf("row %s inactive after backfill", recordTriggerType)
		}
		got = append(got, recordTriggerType)
	}
	if err := rows.Err(); err != nil {
		t.Fatalf("iterate TriggerSubscription rows: %v", err)
	}
	if len(got) != len(wantRecordTriggerTypes) {
		t.Fatalf("row count = %d, want %d (%v)", len(got), len(wantRecordTriggerTypes), wantRecordTriggerTypes)
	}
	for i := range got {
		if got[i] != wantRecordTriggerTypes[i] {
			t.Fatalf("RecordTriggerType[%d] = %q, want %q", i, got[i], wantRecordTriggerTypes[i])
		}
	}
}

func countTriggerSubscriptionRows(t *testing.T, ctx context.Context, db *sql.DB, definitionID string) int {
	t.Helper()
	var count int
	if err := db.QueryRowContext(ctx, `SELECT COUNT(*) FROM _System_TriggerSubscription WHERE FlowDefinitionId = ?`, definitionID).Scan(&count); err != nil {
		t.Fatalf("count TriggerSubscription rows: %v", err)
	}
	return count
}

func countActiveTriggerSubscriptionRows(t *testing.T, ctx context.Context, db *sql.DB, definitionID string) int {
	t.Helper()
	var count int
	if err := db.QueryRowContext(ctx, `SELECT COUNT(*) FROM _System_TriggerSubscription WHERE FlowDefinitionId = ? AND IsActive = 1`, definitionID).Scan(&count); err != nil {
		t.Fatalf("count active TriggerSubscription rows: %v", err)
	}
	return count
}

// TestBackfillUsesWorkflowOnAllChangesWithKeyPrefixParity verifies that a
// Workflow FlowDefinition+FlowVersion with start.triggerType="onAllChanges"
// and a valid _System_Object KeyPrefix is backfilled into
// _System_TriggerSubscription with EventType="RecordAfterSave" and the
// correct KeyPrefix (issue #571 — onAllChanges alias).
func TestBackfillUsesWorkflowOnAllChangesWithKeyPrefixParity(t *testing.T) {
	ctx := context.Background()
	db, cleanup := openTriggerSubscriptionTestDB(t)
	defer cleanup()

	// Seed the object with a distinct key prefix to verify KeyPrefix parity.
	seedTriggerSubscriptionObject(t, ctx, db, "obj-xodml", "XODMLChild__c", "x02")

	// Seed a Workflow FlowDefinition+FlowVersion with onAllChanges.
	seedTriggerSubscriptionFlowDefinition(t, ctx, db, "300xodml0000001AAA", "301xodml0000001AAA", "Workflow")
	seedTriggerSubscriptionFlowVersionOnAllChanges(t, ctx, db, "301xodml0000001AAA", "300xodml0000001AAA")

	if err := Backfill(ctx, db); err != nil {
		t.Fatalf("Backfill() error = %v", err)
	}

	// Assert: a row exists with EventType="RecordAfterSave" and KeyPrefix="x02".
	var gotEventType, gotKeyPrefix string
	err := db.QueryRowContext(ctx, `
		SELECT EventType, KeyPrefix
		FROM _System_TriggerSubscription
		WHERE FlowDefinitionId = ?
		LIMIT 1`, "300xodml0000001AAA").Scan(&gotEventType, &gotKeyPrefix)
	if err != nil {
		t.Fatalf("query TriggerSubscription for onAllChanges flow: %v", err)
	}
	if gotEventType != "RecordAfterSave" {
		t.Errorf("EventType = %q, want RecordAfterSave", gotEventType)
	}
	if gotKeyPrefix != "x02" {
		t.Errorf("KeyPrefix = %q, want x02", gotKeyPrefix)
	}
}

// seedTriggerSubscriptionFlowVersionOnAllChanges inserts a Flow row with
// start.triggerType="onAllChanges" for the XODMLChild__c object.
func seedTriggerSubscriptionFlowVersionOnAllChanges(t *testing.T, ctx context.Context, db *sql.DB, id string, definitionID string) {
	t.Helper()
	now := time.Now().Format(time.RFC3339)
	metadata := `{"start":{"triggerType":"onAllChanges","object":"XODMLChild__c"}}`
	if _, err := db.ExecContext(ctx, `INSERT INTO Flow
		(Id, DefinitionId, MasterLabel, VersionNumber, Status, Description, ProcessType, Metadata, Source, IsDeleted, CreatedDate, LastModifiedDate)
		VALUES (?, ?, 'XODML OnAllChanges Flow', 1, 'Active', '', 'Workflow', ?, 'custom', 0, ?, ?)`,
		id, definitionID, metadata, now, now); err != nil {
		t.Fatalf("seed Flow (onAllChanges): %v", err)
	}
}

func modelsFlowAfterSaveCreateAndUpdate() string {
	return `{"start":{"triggerType":"RecordAfterSave","recordTriggerType":"CreateAndUpdate","object":"ProbeObject"}}`
}

func modelsFlowAfterSaveMissingObject() string {
	return `{"start":{"triggerType":"RecordAfterSave","recordTriggerType":"CreateAndUpdate","object":"MissingObject"}}`
}

func modelsFlowAfterSaveInvalidRecordTriggerType() string {
	return `{"start":{"triggerType":"RecordAfterSave","recordTriggerType":"Bogus","object":"ProbeObject"}}`
}

func modelsWorkflowMetadata() string {
	return `{"variables":[{"name":"myVariable_current","dataType":"SObject","objectType":"ProbeObject"}]}`
}

func modelsAtomicWorkflowMetadata() string {
	return `{"runAtomically":true,"variables":[{"name":"myVariable_current","dataType":"SObject","objectType":"ProbeObject"}]}`
}

func getenvDefault(key string, fallback string) string {
	if v := os.Getenv(key); v != "" {
		return v
	}
	return fallback
}