package controlplane

import (
	"context"
	"database/sql"
	"strings"
	"testing"

	"github.com/nexuscrm/automation/triggersub"
	"github.com/nexuscrm/metadata"
	"github.com/nexuscrm/db"
	"github.com/nexuscrm/shared/pkg/models"
)

type triggerSubscriptionSnapshotRecorder struct {
	exec          *triggerSubscriptionSnapshotExecutor
	beginCalls    int
	commitCalls   int
	rollbackCalls int
}

type triggerSubscriptionSnapshotExecutor struct{}

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

	metaSvc := seedVerifyTriggerSubscriptionCanonicalState(t, ctx, rawDB)

	conn := database.NewConnectionFromDB(rawDB)
	if err := runFlowRuntimeMaintenanceAfterOrphanRepair(ctx, conn, metaSvc, "verify-trigger-subscription"); err != nil {
		t.Fatalf("runFlowRuntimeMaintenanceAfterOrphanRepair() error = %v", err)
	}

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

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

	metaSvc := seedVerifyTriggerSubscriptionCanonicalState(t, ctx, rawDB)

	conn := database.NewConnectionFromDB(rawDB)
	if err := runFlowRuntimeMaintenanceAfterOrphanRepair(ctx, conn, metaSvc, "verify-trigger-subscription"); err != nil {
		t.Fatalf("runFlowRuntimeMaintenanceAfterOrphanRepair() error = %v", err)
	}

	if _, err := rawDB.ExecContext(ctx, `
		DELETE FROM _System_TriggerSubscription
		WHERE FlowDefinitionId = ? AND RecordTriggerType = ?`,
		"300000000000001AAA",
		"Update",
	); err != nil {
		t.Fatalf("delete canonical trigger subscription row: %v", err)
	}

	err := VerifyTriggerSubscriptionConsistency(ctx, conn)
	if err == nil {
		t.Fatal("VerifyTriggerSubscriptionConsistency() error = nil, want missing-row failure")
	}
	if !strings.Contains(err.Error(), "missing rows") {
		t.Fatalf("error = %q, want missing rows detail", err.Error())
	}
}

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

	metaSvc := seedVerifyTriggerSubscriptionCanonicalState(t, ctx, rawDB)

	conn := database.NewConnectionFromDB(rawDB)
	if err := runFlowRuntimeMaintenanceAfterOrphanRepair(ctx, conn, metaSvc, "verify-trigger-subscription"); err != nil {
		t.Fatalf("runFlowRuntimeMaintenanceAfterOrphanRepair() error = %v", err)
	}

	if _, err := rawDB.ExecContext(ctx, `
		INSERT INTO _System_TriggerSubscription
			(Id, KeyPrefix, EventType, RecordTriggerType, FlowDefinitionId, FlowVersionId, IsActive, ConditionShapeHash, CreatedDate, LastModifiedDate)
		VALUES (?, ?, ?, ?, ?, ?, 1, NULL, NOW(), NOW())`,
		"trigsub_stale_delete",
		"a00",
		"RecordAfterDelete",
		"Delete",
		"300000000000001AAA",
		"301000000000001AAA",
	); err != nil {
		t.Fatalf("insert stale trigger subscription row: %v", err)
	}

	err := VerifyTriggerSubscriptionConsistency(ctx, conn)
	if err == nil {
		t.Fatal("VerifyTriggerSubscriptionConsistency() error = nil, want stale-row failure")
	}
	if !strings.Contains(err.Error(), "stale rows") {
		t.Fatalf("error = %q, want stale rows detail", err.Error())
	}
}

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

	metaSvc := seedVerifyTriggerSubscriptionCanonicalState(t, ctx, rawDB)

	conn := database.NewConnectionFromDB(rawDB)
	if err := runFlowRuntimeMaintenanceAfterOrphanRepair(ctx, conn, metaSvc, "verify-trigger-subscription"); err != nil {
		t.Fatalf("runFlowRuntimeMaintenanceAfterOrphanRepair() error = %v", err)
	}

	if _, err := rawDB.ExecContext(ctx, `
		UPDATE _System_TriggerSubscription
		SET FlowVersionId = ?
		WHERE FlowDefinitionId = ? AND RecordTriggerType = ?`,
		"301000000000099AAA",
		"300000000000001AAA",
		"Create",
	); err != nil {
		t.Fatalf("update trigger subscription flow version id: %v", err)
	}

	err := VerifyTriggerSubscriptionConsistency(ctx, conn)
	if err == nil {
		t.Fatal("VerifyTriggerSubscriptionConsistency() error = nil, want tuple-mismatch failure")
	}
	if !strings.Contains(err.Error(), "tuple mismatches") {
		t.Fatalf("error = %q, want tuple mismatch detail", err.Error())
	}
}

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

	metaSvc := seedVerifyTriggerSubscriptionCanonicalState(t, ctx, rawDB)

	conn := database.NewConnectionFromDB(rawDB)
	if err := runFlowRuntimeMaintenanceAfterOrphanRepair(ctx, conn, metaSvc, "verify-trigger-subscription"); err != nil {
		t.Fatalf("runFlowRuntimeMaintenanceAfterOrphanRepair() error = %v", err)
	}

	if _, err := rawDB.ExecContext(ctx, `
		INSERT INTO _System_TriggerSubscription
			(Id, KeyPrefix, EventType, RecordTriggerType, FlowDefinitionId, FlowVersionId, IsActive, ConditionShapeHash, CreatedDate, LastModifiedDate)
		VALUES (?, ?, ?, ?, ?, ?, 0, NULL, NOW(), NOW())`,
		"trigsub_inactive_stale",
		"a00",
		"RecordAfterDelete",
		"Delete",
		"300000000000001AAA",
		"301000000000001AAA",
	); err != nil {
		t.Fatalf("insert inactive stale trigger subscription row: %v", err)
	}

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

func TestVerifyTriggerSubscriptionConsistencyPassesForWorkflowProjection(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", controlplaneWorkflowMetadata("ProbeObject"))

	repo := &controlplaneFlowRuntimeRepo{
		exec: rawDB,
		flowDefinitions: []*models.FlowDefinition{
			{
				Id:              "300000000000001AAA",
				DeveloperName:   "Workflow_Flow",
				ProcessType:     "Workflow",
				ActiveVersionId: "301000000000001AAA",
			},
		},
		flowVersions: map[string]*models.FlowVersion{
			"301000000000001AAA": {
				Id:           "301000000000001AAA",
				DefinitionId: "300000000000001AAA",
				ProcessType:  "Workflow",
				MetadataRaw:  []byte(controlplaneWorkflowMetadata("ProbeObject")),
			},
		},
	}
	metaSvc := metadata.NewMetadataService(repo, nil, nil)
	conn := database.NewConnectionFromDB(rawDB)

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

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

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

	metaSvc := seedVerifyTriggerSubscriptionCanonicalState(t, ctx, rawDB)

	conn := database.NewConnectionFromDB(rawDB)
	if err := runFlowRuntimeMaintenanceAfterOrphanRepair(ctx, conn, metaSvc, "verify-trigger-subscription"); err != nil {
		t.Fatalf("runFlowRuntimeMaintenanceAfterOrphanRepair() error = %v", err)
	}

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

	err := VerifyTriggerSubscriptionConsistency(ctx, conn)
	if err == nil {
		t.Fatal("VerifyTriggerSubscriptionConsistency() error = nil, want malformed metadata failure")
	}
	if !strings.Contains(err.Error(), "resolve canonical rows") {
		t.Fatalf("error = %q, want canonical-row resolution context", err.Error())
	}
}

func TestVerifyTriggerSubscriptionConsistencyUsesSingleSnapshotForBothReads(t *testing.T) {
	ctx := context.Background()
	recorder := &triggerSubscriptionSnapshotRecorder{
		exec: &triggerSubscriptionSnapshotExecutor{},
	}
	expectedRow := triggersub.Row{
		KeyPrefix:         "a00",
		EventType:         "RecordAfterSave",
		RecordTriggerType: "Create",
		FlowDefinitionId:  "300000000000001AAA",
		FlowVersionId:     "301000000000001AAA",
	}

	err := verifyTriggerSubscriptionConsistencyWithSnapshot(
		ctx,
		func(context.Context) (triggersub.Executor, func() error, func() error, error) {
			recorder.beginCalls++
			return recorder.exec, func() error {
					recorder.commitCalls++
					return nil
				}, func() error {
					recorder.rollbackCalls++
					return nil
				}, nil
		},
		func(_ context.Context, exec triggersub.Executor) ([]triggersub.Row, error) {
			if exec != recorder.exec {
				t.Fatalf("expected canonical loader to use snapshot executor %p, got %p", recorder.exec, exec)
			}
			return []triggersub.Row{expectedRow}, nil
		},
		func(_ context.Context, exec triggersub.Executor) ([]triggersub.Row, error) {
			if exec != recorder.exec {
				t.Fatalf("expected active-row loader to use snapshot executor %p, got %p", recorder.exec, exec)
			}
			return []triggersub.Row{expectedRow}, nil
		},
	)
	if err != nil {
		t.Fatalf("verifyTriggerSubscriptionConsistencyWithSnapshot() error = %v, want nil", err)
	}
	if recorder.beginCalls != 1 {
		t.Fatalf("beginCalls = %d, want 1", recorder.beginCalls)
	}
	if recorder.commitCalls != 1 {
		t.Fatalf("commitCalls = %d, want 1", recorder.commitCalls)
	}
	if recorder.rollbackCalls != 0 {
		t.Fatalf("rollbackCalls = %d, want 0 after successful commit", recorder.rollbackCalls)
	}
}

func seedVerifyTriggerSubscriptionCanonicalState(t *testing.T, ctx context.Context, rawDB *sql.DB) *metadata.MetadataService {
	t.Helper()

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

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

func controlplaneWorkflowMetadata(objectAPIName string) string {
	return `{"variables":[{"name":"myVariable_current","dataType":"SObject","objectType":"` + objectAPIName + `"}]}`
}

func (triggerSubscriptionSnapshotExecutor) ExecContext(context.Context, string, ...interface{}) (sql.Result, error) {
	return nil, nil
}

func (triggerSubscriptionSnapshotExecutor) QueryContext(context.Context, string, ...interface{}) (*sql.Rows, error) {
	return nil, nil
}

func (triggerSubscriptionSnapshotExecutor) QueryRowContext(context.Context, string, ...interface{}) *sql.Row {
	return nil
}