package tenantoverlay

import (
	"context"
	"database/sql"
	"fmt"
	"os"
	"path/filepath"
	"strings"
	"testing"
	"time"

	_ "github.com/go-sql-driver/mysql"
	"github.com/pingcap-inc/nexuscrm-controlplane/packs"
	"github.com/nexuscrm/metadata"
	"github.com/nexuscrm/db"
	"github.com/nexuscrm/persistence"
	"github.com/nexuscrm/shared/pkg/models"
)

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

	seedTenantFlowRuntimeObject(t, ctx, db, "obj-flowdef", "FlowDefinition", "300")
	seedTenantFlowRuntimeObject(t, ctx, db, "obj-flowver", "Flow", "301")
	seedTenantFlowRuntimeObject(t, ctx, db, "obj-probe", "ProbeObject", "a00")

	overlayDir := t.TempDir()
	flowsDir := filepath.Join(overlayDir, FlowsDir)
	if err := os.MkdirAll(flowsDir, 0o755); err != nil {
		t.Fatalf("mkdir flows dir: %v", err)
	}
	const flowJSON = `{
  "label": "Tenant Flow",
  "processType": "Workflow",
  "variables": [
    {
      "name": "myVariable_current",
      "dataType": "SObject",
      "objectType": "ProbeObject"
    }
  ]
}`
	if err := os.WriteFile(filepath.Join(flowsDir, "Tenant_Flow.flow.json"), []byte(flowJSON), 0o644); err != nil {
		t.Fatalf("write flow file: %v", err)
	}

	repo := newTenantFlowRuntimeRepo(db)
	metaSvc := metadata.NewMetadataService(repo, nil, nil)
	pack := New(overlayDir)

	err := pack.seedTenantFlows(ctx, packs.ApplyDeps{
		DB:           db,
		DBConn:       database.NewConnectionFromDB(db),
		Metadata:     metaSvc,
		MetadataRepo: repo,
	})
	if err != nil {
		t.Fatalf("seedTenantFlows() error = %v", err)
	}

	rows, err := db.QueryContext(ctx, `
		SELECT RecordTriggerType, FlowVersionId
		FROM _System_TriggerSubscription
		WHERE FlowDefinitionId = (
			SELECT Id FROM FlowDefinition WHERE DeveloperName = ? LIMIT 1
		) AND IsActive = 1
		ORDER BY RecordTriggerType`, "Tenant_Flow")
	if err != nil {
		t.Fatalf("query trigger subscriptions: %v", err)
	}
	defer rows.Close()

	var gotTypes []string
	var versionIDs []string
	for rows.Next() {
		var recordTriggerType string
		var flowVersionID string
		if err := rows.Scan(&recordTriggerType, &flowVersionID); err != nil {
			t.Fatalf("scan trigger subscription row: %v", err)
		}
		gotTypes = append(gotTypes, recordTriggerType)
		versionIDs = append(versionIDs, flowVersionID)
	}
	if err := rows.Err(); err != nil {
		t.Fatalf("iterate trigger subscriptions: %v", err)
	}

	if want := []string{"Create", "Update"}; len(gotTypes) != len(want) ||
		gotTypes[0] != want[0] || gotTypes[1] != want[1] {
		t.Fatalf("active trigger subscription types = %v, want %v", gotTypes, want)
	}
	if len(versionIDs) != 2 || strings.TrimSpace(versionIDs[0]) == "" || versionIDs[0] != versionIDs[1] {
		t.Fatalf("trigger subscription FlowVersionIds = %v, want the same active version id on both rows", versionIDs)
	}
}

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

	seedTenantFlowRuntimeObject(t, ctx, db, "obj-flowdef", "FlowDefinition", "300")
	seedTenantFlowRuntimeObject(t, ctx, db, "obj-flowver", "Flow", "301")
	seedTenantFlowRuntimeObject(t, ctx, db, "obj-probe", "ProbeObject", "a00")

	if _, err := db.ExecContext(ctx, `
		INSERT INTO FlowDefinition
			(Id, DeveloperName, MasterLabel, ProcessType, ActiveVersionId, LatestVersionId, IsDeleted, IsCustomized, CreatedDate, LastModifiedDate)
		VALUES (?, ?, ?, ?, ?, ?, 0, 0, ?, ?)`,
		"300000000000099AAA", "Broken_Flow", "Broken Flow", "Workflow", "301000000000099AAA", "301000000000099AAA",
		time.Now().Format(time.RFC3339), time.Now().Format(time.RFC3339)); err != nil {
		t.Fatalf("seed broken FlowDefinition: %v", err)
	}

	overlayDir := t.TempDir()
	flowsDir := filepath.Join(overlayDir, FlowsDir)
	if err := os.MkdirAll(flowsDir, 0o755); err != nil {
		t.Fatalf("mkdir flows dir: %v", err)
	}
	const flowJSON = `{
  "label": "Tenant Flow",
  "processType": "Workflow",
  "variables": [
    {
      "name": "myVariable_current",
      "dataType": "SObject",
      "objectType": "ProbeObject"
    }
  ]
}`
	if err := os.WriteFile(filepath.Join(flowsDir, "Tenant_Flow.flow.json"), []byte(flowJSON), 0o644); err != nil {
		t.Fatalf("write flow file: %v", err)
	}

	repo := newTenantFlowRuntimeRepo(db)
	metaSvc := metadata.NewMetadataService(repo, nil, nil)
	pack := New(overlayDir)

	err := pack.seedTenantFlows(ctx, packs.ApplyDeps{
		DB:           db,
		DBConn:       database.NewConnectionFromDB(db),
		Metadata:     metaSvc,
		MetadataRepo: repo,
	})
	if err != nil {
		t.Fatalf("seedTenantFlows() error = %v", err)
	}

	var activeVersionID sql.NullString
	if err := db.QueryRowContext(ctx, `SELECT ActiveVersionId FROM FlowDefinition WHERE Id = ?`, "300000000000099AAA").Scan(&activeVersionID); err != nil {
		t.Fatalf("read repaired orphan FlowDefinition: %v", err)
	}
	if activeVersionID.Valid && strings.TrimSpace(activeVersionID.String) != "" {
		t.Fatalf("historical orphan ActiveVersionId = %q, want NULL after tenant runtime maintenance prepass", activeVersionID.String)
	}
}

type tenantFlowRuntimeRepo struct {
	stubMetaRepoTFR
	db *sql.DB
}

func newTenantFlowRuntimeRepo(db *sql.DB) *tenantFlowRuntimeRepo {
	return &tenantFlowRuntimeRepo{db: db}
}

func (r *tenantFlowRuntimeRepo) GetExecutor(context.Context) persistence.Executor {
	return r.db
}

func (r *tenantFlowRuntimeRepo) GetAllFlowDefinitions(ctx context.Context) ([]*models.FlowDefinition, error) {
	rows, err := r.db.QueryContext(ctx, `
		SELECT Id, DeveloperName, MasterLabel, Description, ProcessType, TriggerType,
		       TriggerObjectOrEvent, ActiveVersionId, LatestVersionId, IsDeleted, IsCustomized
		FROM FlowDefinition
		WHERE COALESCE(IsDeleted, 0) = 0
		ORDER BY DeveloperName`)
	if err != nil {
		return nil, err
	}
	defer rows.Close()

	var defs []*models.FlowDefinition
	for rows.Next() {
		def, err := scanTenantFlowDefinition(rows.Scan)
		if err != nil {
			return nil, err
		}
		defs = append(defs, def)
	}
	return defs, rows.Err()
}

func (r *tenantFlowRuntimeRepo) GetFlowDefinitionByName(ctx context.Context, name string) (*models.FlowDefinition, error) {
	row := r.db.QueryRowContext(ctx, `
		SELECT Id, DeveloperName, MasterLabel, Description, ProcessType, TriggerType,
		       TriggerObjectOrEvent, ActiveVersionId, LatestVersionId, IsDeleted, IsCustomized
		FROM FlowDefinition
		WHERE DeveloperName = ?
		LIMIT 1`, name)
	def, err := scanTenantFlowDefinition(row.Scan)
	if err == sql.ErrNoRows {
		return nil, nil
	}
	return def, err
}

func (r *tenantFlowRuntimeRepo) EnsureFlowDefinitionByDeveloperName(ctx context.Context, def *models.FlowDefinition) (*models.FlowDefinition, error) {
	now := time.Now().Format(time.RFC3339)
	if _, err := r.db.ExecContext(ctx, `
		INSERT IGNORE INTO FlowDefinition
			(Id, DeveloperName, MasterLabel, Description, ProcessType, TriggerType, TriggerObjectOrEvent,
			 ActiveVersionId, LatestVersionId, IsDeleted, IsCustomized, OwnerPack, CreatedDate, LastModifiedDate)
		VALUES (?, ?, ?, '', '', '', '', NULL, NULL, 0, 0, '', ?, ?)`,
		def.Id, def.DeveloperName, def.DeveloperName, now, now); err != nil {
		return nil, err
	}
	return r.GetFlowDefinitionByName(ctx, def.DeveloperName)
}

func (r *tenantFlowRuntimeRepo) LockFlowDefinitionByDeveloperNameForUpdate(ctx context.Context, name string) (*models.FlowDefinition, error) {
	return r.GetFlowDefinitionByName(ctx, name)
}

func (r *tenantFlowRuntimeRepo) GetFlowVersionsForUpdate(ctx context.Context, definitionID string) ([]*models.FlowVersion, error) {
	rows, err := r.db.QueryContext(ctx, `
		SELECT Id, DefinitionId, VersionNumber, Status, Description, ProcessType, Metadata, Source
		FROM Flow
		WHERE DefinitionId = ?
		ORDER BY VersionNumber`, definitionID)
	if err != nil {
		return nil, err
	}
	defer rows.Close()

	var versions []*models.FlowVersion
	for rows.Next() {
		version, err := scanTenantFlowVersion(rows.Scan)
		if err != nil {
			return nil, err
		}
		versions = append(versions, version)
	}
	return versions, rows.Err()
}

func (r *tenantFlowRuntimeRepo) GetFlowVersion(ctx context.Context, id string) (*models.FlowVersion, error) {
	row := r.db.QueryRowContext(ctx, `
		SELECT Id, DefinitionId, VersionNumber, Status, Description, ProcessType, Metadata, Source
		FROM Flow
		WHERE Id = ?
		LIMIT 1`, id)
	return scanTenantFlowVersion(row.Scan)
}

func (r *tenantFlowRuntimeRepo) GetFlowVersionsByIDs(ctx context.Context, ids []string) (map[string]*models.FlowVersion, error) {
	if len(ids) == 0 {
		return map[string]*models.FlowVersion{}, nil
	}

	query := `SELECT Id, DefinitionId, VersionNumber, Status, Description, ProcessType, Metadata, Source
		FROM Flow WHERE Id IN (?` + strings.Repeat(",?", len(ids)-1) + `)`
	args := make([]any, 0, len(ids))
	for _, id := range ids {
		args = append(args, id)
	}
	rows, err := r.db.QueryContext(ctx, query, args...)
	if err != nil {
		return nil, err
	}
	defer rows.Close()

	result := make(map[string]*models.FlowVersion, len(ids))
	for rows.Next() {
		version, err := scanTenantFlowVersion(rows.Scan)
		if err != nil {
			return nil, err
		}
		result[version.Id] = version
	}
	return result, rows.Err()
}

func (r *tenantFlowRuntimeRepo) CreateTenantFlowVersion(ctx context.Context, version *models.FlowVersion) error {
	metadataJSON := strings.TrimSpace(version.MetadataJSON)
	if metadataJSON == "" && len(version.MetadataRaw) > 0 {
		metadataJSON = string(version.MetadataRaw)
	}
	now := time.Now().Format(time.RFC3339)
	_, err := r.db.ExecContext(ctx, `
		INSERT INTO Flow
			(Id, DefinitionId, MasterLabel, VersionNumber, Status, Description, ProcessType, Metadata, Source, IsDeleted, CreatedDate, LastModifiedDate)
		VALUES (?, ?, ?, ?, ?, ?, ?, ?, 'custom', 0, ?, ?)`,
		version.Id, version.DefinitionId, "Tenant Flow", version.VersionNumber, version.Status, version.Description, version.ProcessType, metadataJSON, now, now)
	return err
}

func (r *tenantFlowRuntimeRepo) ActivateFlowVersion(ctx context.Context, id string) error {
	var definitionID string
	if err := r.db.QueryRowContext(ctx, `SELECT DefinitionId FROM Flow WHERE Id = ?`, id).Scan(&definitionID); err != nil {
		return err
	}

	var latestVersionID string
	if err := r.db.QueryRowContext(ctx, `
		SELECT Id
		FROM Flow
		WHERE DefinitionId = ?
		ORDER BY VersionNumber DESC
		LIMIT 1`, definitionID).Scan(&latestVersionID); err != nil {
		return err
	}

	if _, err := r.db.ExecContext(ctx, `UPDATE Flow SET Status = 'Obsolete' WHERE DefinitionId = ? AND Status = 'Active' AND Id <> ?`, definitionID, id); err != nil {
		return err
	}
	if _, err := r.db.ExecContext(ctx, `UPDATE Flow SET Status = 'Active', IsDeleted = 0 WHERE Id = ?`, id); err != nil {
		return err
	}
	_, err := r.db.ExecContext(ctx, `
		UPDATE FlowDefinition
		SET ActiveVersionId = ?, LatestVersionId = ?, IsDeleted = 0
		WHERE Id = ?`, id, latestVersionID, definitionID)
	return err
}

func (r *tenantFlowRuntimeRepo) UpdateFlowDefinition(ctx context.Context, id string, def *models.FlowDefinition) error {
	now := time.Now().Format(time.RFC3339)
	_, err := r.db.ExecContext(ctx, `
		UPDATE FlowDefinition
		SET DeveloperName = ?, MasterLabel = ?, Description = ?, ProcessType = ?,
		    TriggerType = ?, TriggerObjectOrEvent = ?, ActiveVersionId = ?, LatestVersionId = ?,
		    IsDeleted = ?, LastModifiedDate = ?
		WHERE Id = ?`,
		def.DeveloperName, def.MasterLabel, def.Description, def.ProcessType,
		def.TriggerType, def.TriggerObjectOrEvent, nullableString(def.ActiveVersionId), nullableString(def.LatestVersionId),
		def.IsDeleted, now, id)
	return err
}

func openTenantFlowRuntimeTestDB(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_tenant_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)
	}
	createTenantFlowRuntimeTables(t, db)

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

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

func seedTenantFlowRuntimeObject(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 scanTenantFlowDefinition(scan func(dest ...any) error) (*models.FlowDefinition, error) {
	var def models.FlowDefinition
	var activeVersionID sql.NullString
	var latestVersionID sql.NullString
	if err := scan(
		&def.Id,
		&def.DeveloperName,
		&def.MasterLabel,
		&def.Description,
		&def.ProcessType,
		&def.TriggerType,
		&def.TriggerObjectOrEvent,
		&activeVersionID,
		&latestVersionID,
		&def.IsDeleted,
		&def.IsCustomized,
	); err != nil {
		return nil, err
	}
	def.ActiveVersionId = activeVersionID.String
	def.LatestVersionId = latestVersionID.String
	return &def, nil
}

func scanTenantFlowVersion(scan func(dest ...any) error) (*models.FlowVersion, error) {
	var version models.FlowVersion
	var metadataJSON string
	if err := scan(
		&version.Id,
		&version.DefinitionId,
		&version.VersionNumber,
		&version.Status,
		&version.Description,
		&version.ProcessType,
		&metadataJSON,
		&version.Source,
	); err != nil {
		return nil, err
	}
	version.MetadataJSON = metadataJSON
	version.MetadataRaw = []byte(metadataJSON)
	return &version, nil
}

func nullableString(value string) any {
	if strings.TrimSpace(value) == "" {
		return nil
	}
	return value
}

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