package controlplane

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/db"
	"github.com/nexuscrm/persistence"
	"github.com/nexuscrm/shared/pkg/models"
)

func TestRunFlowRuntimeMaintenanceAfterOrphanRepairRepairsKnownOrphansFirst(t *testing.T) {
	ctx := context.Background()
	rawDB, cleanup := openControlplaneFlowRuntimeTestDB(t)
	defer cleanup()

	seedControlplaneFlowRuntimeObject(t, ctx, rawDB, "obj-probe", "ProbeObject", "a00")
	seedControlplaneFlowRuntimeDefinition(t, ctx, rawDB, "300000000000001AAA", "301000000000001AAA")
	seedControlplaneFlowRuntimeVersion(t, ctx, rawDB, "301000000000001AAA", "300000000000001AAA", controlplaneFlowRuntimeMetadata("RecordAfterSave", "CreateAndUpdate", "ProbeObject"))
	seedControlplaneFlowRuntimeDefinition(t, ctx, rawDB, "300000000000002AAA", "301000000000099AAA")

	repo := &controlplaneFlowRuntimeRepo{
		exec: rawDB,
		flowDefinitions: []*models.FlowDefinition{
			{
				Id:              "300000000000001AAA",
				DeveloperName:   "Valid_Flow",
				ProcessType:     "AutoLaunchedFlow",
				ActiveVersionId: "301000000000001AAA",
			},
			{
				Id:              "300000000000002AAA",
				DeveloperName:   "Broken_Flow",
				ProcessType:     "AutoLaunchedFlow",
				ActiveVersionId: "301000000000099AAA",
			},
		},
		flowVersions: map[string]*models.FlowVersion{
			"301000000000001AAA": {
				Id:           "301000000000001AAA",
				DefinitionId: "300000000000001AAA",
				ProcessType:  "AutoLaunchedFlow",
				MetadataRaw:  []byte(controlplaneFlowRuntimeMetadata("RecordAfterSave", "CreateAndUpdate", "ProbeObject")),
			},
		},
	}
	metaSvc := metadata.NewMetadataService(repo, nil, nil)
	conn := database.NewConnectionFromDB(rawDB)

	if err := runFlowRuntimeMaintenanceAfterOrphanRepair(ctx, conn, metaSvc, "test"); err != nil {
		t.Fatalf("runFlowRuntimeMaintenanceAfterOrphanRepair() error = %v", err)
	}

	var brokenActiveVersionID sql.NullString
	if err := rawDB.QueryRowContext(ctx, `SELECT ActiveVersionId FROM FlowDefinition WHERE Id = ?`, "300000000000002AAA").Scan(&brokenActiveVersionID); err != nil {
		t.Fatalf("read broken FlowDefinition ActiveVersionId: %v", err)
	}
	if brokenActiveVersionID.Valid && brokenActiveVersionID.String != "" {
		t.Fatalf("broken FlowDefinition ActiveVersionId = %q, want NULL after orphan repair", brokenActiveVersionID.String)
	}

	assertControlplaneFlowRuntimeRows(t, ctx, rawDB, "300000000000001AAA", true, []string{"Create", "Update"})
}

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

	host := getenvOrDefault("TIDB_HOST", "127.0.0.1")
	port := getenvOrDefault("TIDB_PORT", "4001")
	user := getenvOrDefault("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_controlplane_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)
	}
	createControlplaneFlowRuntimeTables(t, db)

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

func createControlplaneFlowRuntimeTables(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 controlplane flow runtime table: %v", err)
		}
	}
}

func seedControlplaneFlowRuntimeObject(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 seedControlplaneFlowRuntimeDefinition(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 (?, ?, ?, '', 'AutoLaunchedFlow', '', '', ?, ?, 0, 0, '', ?, ?)`,
		id, id, id, nullableTestValue(activeVersionID), nullableTestValue(activeVersionID), now, now); err != nil {
		t.Fatalf("seed FlowDefinition: %v", err)
	}
}

func seedControlplaneFlowRuntimeVersion(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 controlplaneFlowRuntimeMetadata(eventType string, recordTriggerType string, objectAPIName string) string {
	return fmt.Sprintf(`{"start":{"triggerType":%q,"recordTriggerType":%q,"object":%q}}`, eventType, recordTriggerType, objectAPIName)
}

func assertControlplaneFlowRuntimeRows(t *testing.T, ctx context.Context, db *sql.DB, definitionID string, wantActive bool, wantRecordTriggerTypes []string) {
	t.Helper()
	rows, err := db.QueryContext(ctx, `
		SELECT RecordTriggerType, 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 recordTriggerType string
		var isActive bool
		if err := rows.Scan(&recordTriggerType, &isActive); err != nil {
			t.Fatalf("scan TriggerSubscription row: %v", err)
		}
		if isActive != wantActive {
			t.Fatalf("TriggerSubscription %s active=%t, want %t", recordTriggerType, isActive, wantActive)
		}
		got = append(got, recordTriggerType)
	}
	if err := rows.Err(); err != nil {
		t.Fatalf("iterate TriggerSubscription rows: %v", err)
	}
	if len(got) != len(wantRecordTriggerTypes) {
		t.Fatalf("RecordTriggerTypes = %v, want %v", got, wantRecordTriggerTypes)
	}
	for i := range wantRecordTriggerTypes {
		if got[i] != wantRecordTriggerTypes[i] {
			t.Fatalf("RecordTriggerTypes[%d] = %q, want %q", i, got[i], wantRecordTriggerTypes[i])
		}
	}
}

func nullableTestValue(v string) interface{} {
	if v == "" {
		return nil
	}
	return v
}

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

type controlplaneFlowRuntimeRepo struct {
	ports.MetadataRepository

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

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

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

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

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

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

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

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

func (r *controlplaneFlowRuntimeRepo) 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 *controlplaneFlowRuntimeRepo) 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
}