package packs

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

	_ "github.com/go-sql-driver/mysql"
	"github.com/nexuscrm/kernel/pkg/ports"
	"github.com/nexuscrm/metadata"
	"github.com/nexuscrm/persistence"
	"github.com/nexuscrm/shared/pkg/models"
)

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

	seedFlowRuntimeMaintenanceObject(t, ctx, db, "obj-probe", "ProbeObject", "a00")
	seedFlowRuntimeMaintenanceDefinition(t, ctx, db, "300000000000001AAA", "301000000000001AAA")
	seedFlowRuntimeMaintenanceVersion(t, ctx, db, "301000000000001AAA", "300000000000001AAA", flowRuntimeMaintenanceMetadata("RecordAfterSave", "CreateAndUpdate", "ProbeObject"))

	repo := newFlowRuntimeMaintenanceShadowRepo(db)
	metaSvc := metadata.NewMetadataService(repo, nil, nil)

	if err := ApplyFlowRuntimeMaintenance(ctx, metaSvc); err != nil {
		t.Fatalf("ApplyFlowRuntimeMaintenance() first run error = %v", err)
	}

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

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

	if _, err := db.ExecContext(ctx, `UPDATE FlowDefinition SET ActiveVersionId = NULL WHERE Id = ?`, "300000000000001AAA"); err != nil {
		t.Fatalf("deactivate authoritative flow definition: %v", err)
	}
	repo.flowDefinitions[0].ActiveVersionId = ""

	if err := ApplyFlowRuntimeMaintenance(ctx, metaSvc); err != nil {
		t.Fatalf("ApplyFlowRuntimeMaintenance() deactivate run error = %v", err)
	}
	assertFlowRuntimeMaintenanceRows(t, ctx, db, "300000000000001AAA", "301000000000001AAA", false, []string{"Create", "Update"})
}

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

	seedFlowRuntimeMaintenanceObject(t, ctx, db, "obj-probe", "ProbeObject", "a00")
	seedFlowRuntimeMaintenanceDefinition(t, ctx, db, "300000000000001AAA", "301000000000001AAA")
	seedFlowRuntimeMaintenanceVersion(t, ctx, db, "301000000000001AAA", "300000000000001AAA", flowRuntimeMaintenanceMetadata("RecordAfterSave", "CreateAndUpdate", "ProbeObject"))

	repo := newFlowRuntimeMaintenanceShadowRepo(db)
	metaSvc := metadata.NewMetadataService(repo, nil, nil)
	if err := ApplyFlowRuntimeMaintenance(ctx, metaSvc); err != nil {
		t.Fatalf("ApplyFlowRuntimeMaintenance() seed run error = %v", err)
	}
	assertFlowRuntimeMaintenanceRows(t, ctx, db, "300000000000001AAA", "301000000000001AAA", true, []string{"Create", "Update"})

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

	err := ApplyFlowRuntimeMaintenance(ctx, metaSvc)
	if err == nil {
		t.Fatal("ApplyFlowRuntimeMaintenance() error = nil, want malformed shadow metadata failure")
	}

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

func openFlowRuntimeMaintenanceTestDB(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_pack_flow_runtime_%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)
	}
	createFlowRuntimeMaintenanceTables(t, db)

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

func createFlowRuntimeMaintenanceTables(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 seedFlowRuntimeMaintenanceObject(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 seedFlowRuntimeMaintenanceDefinition(t *testing.T, ctx context.Context, db *sql.DB, id string, activeVersionID 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', '', 'AutoLaunchedFlow', '', '', ?, ?, 0, 0, '', ?, ?)`,
		id, activeVersionID, activeVersionID, now, now); err != nil {
		t.Fatalf("seed FlowDefinition: %v", err)
	}
}

func seedFlowRuntimeMaintenanceVersion(t *testing.T, ctx context.Context, db *sql.DB, id string, definitionID 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', '', 'AutoLaunchedFlow', ?, 'custom', 0, ?, ?)`,
		id, definitionID, metadata, now, now); err != nil {
		t.Fatalf("seed Flow: %v", err)
	}
}

func flowRuntimeMaintenanceMetadata(eventType string, recordTriggerType string, objectAPIName string) string {
	return fmt.Sprintf(`{"start":{"triggerType":%q,"recordTriggerType":%q,"object":%q}}`, eventType, recordTriggerType, objectAPIName)
}

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

	var gotTriggerTypes []string
	for rows.Next() {
		var recordTriggerType string
		var flowVersionID string
		var isActive bool
		if err := rows.Scan(&recordTriggerType, &flowVersionID, &isActive); err != nil {
			t.Fatalf("scan trigger subscription row: %v", err)
		}
		if flowVersionID != versionID {
			t.Fatalf("FlowVersionId = %q, want %q", flowVersionID, versionID)
		}
		if isActive != wantActive {
			t.Fatalf("TriggerSubscription %s active=%t, want %t", recordTriggerType, isActive, wantActive)
		}
		gotTriggerTypes = append(gotTriggerTypes, recordTriggerType)
	}
	if err := rows.Err(); err != nil {
		t.Fatalf("iterate trigger subscriptions: %v", err)
	}

	if len(gotTriggerTypes) != len(wantRecordTriggerTypes) {
		t.Fatalf("RecordTriggerTypes = %v, want %v", gotTriggerTypes, wantRecordTriggerTypes)
	}
	for i := range wantRecordTriggerTypes {
		if gotTriggerTypes[i] != wantRecordTriggerTypes[i] {
			t.Fatalf("RecordTriggerTypes[%d] = %q, want %q (full=%v)", i, gotTriggerTypes[i], wantRecordTriggerTypes[i], gotTriggerTypes)
		}
	}
}

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

type flowRuntimeMaintenanceShadowRepo struct {
	ports.MetadataRepository

	exec            persistence.Executor
	flowDefinitions []*models.FlowDefinition
	flowVersions    map[string]*models.FlowVersion
}

func newFlowRuntimeMaintenanceShadowRepo(db *sql.DB) *flowRuntimeMaintenanceShadowRepo {
	return &flowRuntimeMaintenanceShadowRepo{
		exec: db,
		flowDefinitions: []*models.FlowDefinition{{
			Id:              "300000000000001AAA",
			DeveloperName:   "Probe_Flow",
			ProcessType:     "AutoLaunchedFlow",
			ActiveVersionId: "301000000000001AAA",
		}},
		flowVersions: map[string]*models.FlowVersion{
			"301000000000001AAA": {
				Id:           "301000000000001AAA",
				DefinitionId: "300000000000001AAA",
				ProcessType:  "AutoLaunchedFlow",
				MetadataRaw:  []byte(flowRuntimeMaintenanceMetadata("RecordAfterSave", "CreateAndUpdate", "ProbeObject")),
			},
		},
	}
}

func (r *flowRuntimeMaintenanceShadowRepo) GetExecutor(context.Context) persistence.Executor {
	return r.exec
}

func (r *flowRuntimeMaintenanceShadowRepo) GetAllSchemas(context.Context) ([]*models.ObjectMetadata, error) {
	return nil, nil
}

func (r *flowRuntimeMaintenanceShadowRepo) GetAllRelationships(context.Context) ([]*models.Relationship, error) {
	return nil, nil
}

func (r *flowRuntimeMaintenanceShadowRepo) GetAllFlowDefinitions(context.Context) ([]*models.FlowDefinition, error) {
	return r.flowDefinitions, nil
}

func (r *flowRuntimeMaintenanceShadowRepo) GetAllValidationRules(context.Context) ([]*models.ValidationRule, error) {
	return nil, nil
}

func (r *flowRuntimeMaintenanceShadowRepo) GetAllAutoNumbers(context.Context) ([]*models.AutoNumber, error) {
	return nil, nil
}

func (r *flowRuntimeMaintenanceShadowRepo) GetAllIdentifierRules(context.Context) ([]*models.IdentifierRule, error) {
	return nil, nil
}

func (r *flowRuntimeMaintenanceShadowRepo) GetFlowVersionsByIDs(_ context.Context, ids []string) (map[string]*models.FlowVersion, error) {
	result := make(map[string]*models.FlowVersion, len(ids))
	for _, id := range ids {
		if version := r.flowVersions[id]; version != nil {
			result[id] = version
		}
	}
	return result, nil
}

func (r *flowRuntimeMaintenanceShadowRepo) SetTriggerInfo(_ context.Context, definitionID string, triggerType string, objectOrEvent string) error {
	for _, def := range r.flowDefinitions {
		if def.Id == definitionID {
			def.TriggerType = triggerType
			def.TriggerObjectOrEvent = objectOrEvent
		}
	}
	return nil
}