package metadata

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

	_ "github.com/go-sql-driver/mysql"
	"github.com/nexuscrm/db"
	"github.com/nexuscrm/persistence"
	"github.com/nexuscrm/shared/pkg/contract"
	"github.com/nexuscrm/shared/pkg/models"
)

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

	service := newFlowLifecycleMetadataService(db)
	seedFlowDefinition(t, ctx, db, "300000000000001AAA", "301000000000001AAA", "301000000000001AAA")
	seedFlowVersion(t, ctx, db, "301000000000001AAA", "300000000000001AAA", 1, models.FlowStatusActive)

	next := &models.FlowVersion{
		Id:            "301000000000002AAA",
		DefinitionId:  "300000000000001AAA",
		VersionNumber: 2,
		Status:        models.FlowStatusActive,
		ProcessType:   "AutoLaunchedFlow",
		Metadata: models.FlowMetadata{
			Start: &models.FlowStart{TriggerType: "RecordAfterSave", Object: "ProbeObject"},
		},
	}
	if err := service.CreateFlowVersion(ctx, next); err != nil {
		t.Fatalf("CreateFlowVersion(active) error = %v", err)
	}

	def := readLifecycleDefinition(t, ctx, db, "300000000000001AAA")
	if def.ActiveVersionId != next.Id {
		t.Fatalf("ActiveVersionId = %q, want %q", def.ActiveVersionId, next.Id)
	}
	if def.LatestVersionId != next.Id {
		t.Fatalf("LatestVersionId = %q, want %q", def.LatestVersionId, next.Id)
	}
	if got := readLifecycleVersionStatus(t, ctx, db, "301000000000001AAA"); got != "Obsolete" {
		t.Fatalf("old active status = %q, want Obsolete", got)
	}
	if got := readLifecycleVersionStatus(t, ctx, db, next.Id); got != models.FlowStatusActive {
		t.Fatalf("new version status = %q, want %q", got, models.FlowStatusActive)
	}
}

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

	service := newFlowLifecycleMetadataService(db)
	seedFlowDefinition(t, ctx, db, "300000000000001AAA", "301000000000001AAA", "301000000000002AAA")
	seedFlowVersion(t, ctx, db, "301000000000001AAA", "300000000000001AAA", 1, models.FlowStatusActive)
	seedFlowVersion(t, ctx, db, "301000000000002AAA", "300000000000001AAA", 2, models.FlowStatusDraft)

	if err := service.DeleteFlowVersion(ctx, "301000000000002AAA"); err != nil {
		t.Fatalf("DeleteFlowVersion(latest inactive) error = %v", err)
	}
	def := readLifecycleDefinition(t, ctx, db, "300000000000001AAA")
	if def.ActiveVersionId != "301000000000001AAA" {
		t.Fatalf("after inactive latest delete ActiveVersionId = %q, want active sibling", def.ActiveVersionId)
	}
	if def.LatestVersionId != "301000000000001AAA" {
		t.Fatalf("after inactive latest delete LatestVersionId = %q, want remaining active sibling", def.LatestVersionId)
	}

	seedFlowVersion(t, ctx, db, "301000000000003AAA", "300000000000001AAA", 3, models.FlowStatusDraft)
	setLifecycleDefinitionPointers(t, ctx, db, "300000000000001AAA", "301000000000001AAA", "301000000000003AAA")
	if err := service.DeleteFlowVersion(ctx, "301000000000001AAA"); err != nil {
		t.Fatalf("DeleteFlowVersion(active) error = %v", err)
	}
	def = readLifecycleDefinition(t, ctx, db, "300000000000001AAA")
	if def.ActiveVersionId != "" {
		t.Fatalf("after active delete ActiveVersionId = %q, want empty", def.ActiveVersionId)
	}
	if def.LatestVersionId != "301000000000003AAA" {
		t.Fatalf("after active delete LatestVersionId = %q, want latest inactive sibling", def.LatestVersionId)
	}
}

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

	service := newFlowLifecycleMetadataService(db)
	seedFlowDefinition(t, ctx, db, "300000000000001AAA", "301000000000001AAA", "301000000000002AAA")
	seedFlowVersion(t, ctx, db, "301000000000001AAA", "300000000000001AAA", 1, models.FlowStatusActive)
	seedFlowVersion(t, ctx, db, "301000000000002AAA", "300000000000001AAA", 2, models.FlowStatusDraft)

	var wg sync.WaitGroup
	errs := make(chan error, 2)
	wg.Add(2)
	go func() {
		defer wg.Done()
		errs <- service.ActivateFlowVersion(ctx, "301000000000002AAA")
	}()
	go func() {
		defer wg.Done()
		errs <- service.DeactivateFlowVersion(ctx, "301000000000001AAA")
	}()
	wg.Wait()
	close(errs)
	for err := range errs {
		if err != nil {
			t.Fatalf("concurrent lifecycle operation error = %v", err)
		}
	}

	def := readLifecycleDefinition(t, ctx, db, "300000000000001AAA")
	if def.ActiveVersionId != "301000000000002AAA" {
		t.Fatalf("ActiveVersionId after concurrent activate/deactivate = %q, want new active", def.ActiveVersionId)
	}
	rows := readLifecycleTriggerSubscriptions(t, ctx, db, "300000000000001AAA")
	if len(rows) != 2 {
		t.Fatalf("trigger subscription rows after concurrent activate/deactivate = %d, want 2", len(rows))
	}
	for _, row := range rows {
		if !row.IsActive {
			t.Fatalf("trigger subscription row %+v inactive after concurrent activate/deactivate", row)
		}
		if row.FlowVersionId != "301000000000002AAA" {
			t.Fatalf("trigger subscription FlowVersionId = %q, want new active version", row.FlowVersionId)
		}
	}
	if got := readLifecycleVersionStatus(t, ctx, db, "301000000000002AAA"); got != models.FlowStatusActive {
		t.Fatalf("new version status = %q, want Active", got)
	}
	if got := readLifecycleVersionStatus(t, ctx, db, "301000000000001AAA"); got != models.FlowStatusDraft && got != "Obsolete" {
		t.Fatalf("old version status = %q, want Draft or Obsolete", got)
	}
}

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

	service := newFlowLifecycleMetadataService(db)
	seedFlowDefinition(t, ctx, db, "300000000000001AAA", "301000000000001AAA", "301000000000002AAA")
	seedFlowVersion(t, ctx, db, "301000000000001AAA", "300000000000001AAA", 1, models.FlowStatusActive)
	seedFlowVersion(t, ctx, db, "301000000000002AAA", "300000000000001AAA", 2, models.FlowStatusDraft)

	if err := service.ActivateFlowVersion(ctx, "301000000000001AAA"); err != nil {
		t.Fatalf("ActivateFlowVersion(already active) error = %v", err)
	}
	if err := service.DeactivateFlowVersion(ctx, "301000000000002AAA"); err != nil {
		t.Fatalf("DeactivateFlowVersion(inactive sibling) error = %v", err)
	}
	if err := service.DeactivateFlowVersion(ctx, "301000000000002AAA"); err != nil {
		t.Fatalf("DeactivateFlowVersion(inactive sibling repeated) error = %v", err)
	}
	if err := service.DeleteFlowVersion(ctx, "301000000000099AAA"); err != nil {
		t.Fatalf("DeleteFlowVersion(missing) error = %v", err)
	}

	def := readLifecycleDefinition(t, ctx, db, "300000000000001AAA")
	if def.ActiveVersionId != "301000000000001AAA" {
		t.Fatalf("ActiveVersionId after no-op operations = %q, want unchanged active", def.ActiveVersionId)
	}
	if def.LatestVersionId != "301000000000002AAA" {
		t.Fatalf("LatestVersionId after no-op operations = %q, want unchanged latest", def.LatestVersionId)
	}
}

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

	service := newFlowLifecycleMetadataService(db)
	seedFlowDefinition(t, ctx, db, "300000000000001AAA", "301000000000001AAA", "301000000000001AAA")
	seedFlowVersion(t, ctx, db, "301000000000001AAA", "300000000000001AAA", 1, models.FlowStatusActive)
	setLifecycleDefinitionOwnership(t, ctx, db, "300000000000001AAA", "salesforce-crm-standard", false)

	description := "admin overwrite"
	err := service.ApplyFlowDefinitionPatch(ctx, "300000000000001AAA", contract.FlowAdminDefinitionPatchIntent{
		Action:      contract.FlowAdminDefinitionPatchActionUpdate,
		Description: &description,
	})
	if err == nil {
		t.Fatal("ApplyFlowDefinitionPatch() error = nil, want pack-owned FlowDefinition admission rejection")
	}

	def := readLifecycleDefinition(t, ctx, db, "300000000000001AAA")
	if def.Description != "" {
		t.Fatalf("Description after rejected patch = %q, want unchanged empty", def.Description)
	}
	if def.IsCustomized {
		t.Fatal("IsCustomized after rejected patch = true, want false")
	}
	if def.OwnerPack != "salesforce-crm-standard" {
		t.Fatalf("OwnerPack after rejected patch = %q, want salesforce-crm-standard", def.OwnerPack)
	}
}

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

	service := newFlowLifecycleMetadataService(db)
	seedFlowDefinition(t, ctx, db, "300000000000001AAA", "301000000000001AAA", "301000000000001AAA")
	seedFlowVersion(t, ctx, db, "301000000000001AAA", "300000000000001AAA", 1, models.FlowStatusActive)
	setLifecycleDefinitionOwnership(t, ctx, db, "300000000000001AAA", "salesforce-crm-standard", false)

	err := service.CreateFlowVersion(ctx, &models.FlowVersion{
		Id:           "301000000000002AAA",
		DefinitionId: "300000000000001AAA",
		Status:       models.FlowStatusDraft,
		ProcessType:  "AutoLaunchedFlow",
		Metadata:     models.FlowMetadata{},
	})
	if err == nil {
		t.Fatal("CreateFlowVersion() error = nil, want pack-owned FlowDefinition admission rejection")
	}

	def := readLifecycleDefinition(t, ctx, db, "300000000000001AAA")
	if def.LatestVersionId != "301000000000001AAA" {
		t.Fatalf("LatestVersionId after rejected version create = %q, want unchanged active version", def.LatestVersionId)
	}
	if def.IsCustomized {
		t.Fatal("IsCustomized after rejected version create = true, want false")
	}
	assertLifecycleFlowVersionMissing(t, ctx, db, "301000000000002AAA")
}

func openFlowLifecycleTestDB(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_flow_lifecycle_%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)
	}
	createFlowLifecycleTables(t, db)
	seedLifecycleObject(t, context.Background(), db, "obj-probe", "ProbeObject", "a00")

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

func createFlowLifecycleTables(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),
			UNIQUE KEY idx_flow_definition_version (DefinitionId, VersionNumber)
		)`,
	}
	for _, stmt := range stmts {
		if _, err := db.Exec(stmt); err != nil {
			t.Fatalf("create lifecycle test table: %v", err)
		}
	}
}

func newFlowLifecycleMetadataService(db *sql.DB) *MetadataService {
	conn := database.NewConnectionFromDB(db)
	service := NewMetadataService(persistence.NewMetadataRepository(db), nil, nil)
	service.SetTransactionManager(persistence.NewTransactionManager(conn))
	return service
}

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

func seedFlowVersion(t *testing.T, ctx context.Context, db *sql.DB, id string, definitionID string, versionNumber int, status string) {
	t.Helper()
	_, err := db.ExecContext(ctx, `INSERT INTO Flow
		(Id, DefinitionId, MasterLabel, VersionNumber, Status, Description, ProcessType, Metadata, Source, IsDeleted, CreatedDate, LastModifiedDate)
		VALUES (?, ?, 'Probe Flow', ?, ?, '', 'AutoLaunchedFlow', '{"start":{"triggerType":"RecordAfterSave","object":"ProbeObject"}}', 'custom', 0, ?, ?)`,
		id, definitionID, versionNumber, status, time.Now().Format(time.RFC3339), time.Now().Format(time.RFC3339))
	if err != nil {
		t.Fatalf("seed Flow: %v", err)
	}
}

func setLifecycleDefinitionPointers(t *testing.T, ctx context.Context, db *sql.DB, definitionID string, activeID string, latestID string) {
	t.Helper()
	_, err := db.ExecContext(ctx, `UPDATE FlowDefinition SET ActiveVersionId = ?, LatestVersionId = ? WHERE Id = ?`,
		nullableForTest(activeID), nullableForTest(latestID), definitionID)
	if err != nil {
		t.Fatalf("set FlowDefinition pointers: %v", err)
	}
}

func setLifecycleDefinitionOwnership(t *testing.T, ctx context.Context, db *sql.DB, definitionID string, ownerPack string, isCustomized bool) {
	t.Helper()
	_, err := db.ExecContext(ctx, `UPDATE FlowDefinition SET OwnerPack = ?, IsCustomized = ? WHERE Id = ?`,
		nullableForTest(ownerPack), isCustomized, definitionID)
	if err != nil {
		t.Fatalf("set FlowDefinition ownership: %v", err)
	}
}

func assertLifecycleFlowVersionMissing(t *testing.T, ctx context.Context, db *sql.DB, id string) {
	t.Helper()
	var count int
	if err := db.QueryRowContext(ctx, `SELECT COUNT(*) FROM Flow WHERE Id = ?`, id).Scan(&count); err != nil {
		t.Fatalf("count Flow(%s): %v", id, err)
	}
	if count != 0 {
		t.Fatalf("Flow rows for %s = %d, want 0", id, count)
	}
}

func readLifecycleDefinition(t *testing.T, ctx context.Context, db *sql.DB, id string) *models.FlowDefinition {
	t.Helper()
	repo := persistence.NewMetadataRepository(db)
	def, err := repo.GetFlowDefinition(ctx, id)
	if err != nil {
		t.Fatalf("GetFlowDefinition(%s): %v", id, err)
	}
	if def == nil {
		t.Fatalf("GetFlowDefinition(%s) returned nil", id)
	}
	return def
}

func readLifecycleVersionStatus(t *testing.T, ctx context.Context, db *sql.DB, id string) string {
	t.Helper()
	var status string
	if err := db.QueryRowContext(ctx, `SELECT Status FROM Flow WHERE Id = ?`, id).Scan(&status); err != nil {
		t.Fatalf("read Flow.Status(%s): %v", id, err)
	}
	return status
}

func nullableForTest(v string) any {
	if v == "" {
		return nil
	}
	return v
}

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