package automation

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

	database "github.com/nexuscrm/db"
	eventfeature "github.com/nexuscrm/events"
	"github.com/nexuscrm/metadata/catalogsnapshot"
	"github.com/nexuscrm/persistence"
	flowsemantic "github.com/pingcap-inc/nexuscrm-contract/pkg/runtime/flowv1"
	mutationsemantic "github.com/pingcap-inc/nexuscrm-contract/pkg/runtime/recordmutationv1"
)

type timedFlowTxExtractor struct{ tx *sql.Tx }

func (e timedFlowTxExtractor) ExtractTx(context.Context) interface{} { return e.tx }

func testTimedFlowBinding() catalogsnapshot.RuntimeWorkBinding {
	return catalogsnapshot.RuntimeWorkBinding{
		OrgID: "org-timed", AppInstallationID: "installation-timed",
		RuntimeCapabilityID: flowsemantic.CapabilityName,
		CatalogRevision:     7, SchemaGeneration: 8, EngineGeneration: 9,
		CatalogSnapshotDigest:     catalogsnapshot.Digest("sha256:" + strings.Repeat("a", 64)),
		RuntimeMetadataViewDigest: catalogsnapshot.Digest("sha256:" + strings.Repeat("b", 64)),
	}
}

func testTimedFlowDueRequest() TimedFlowDueRequest {
	return TimedFlowDueRequest{
		Binding: testTimedFlowBinding(), FlowDefinitionID: "definition-timed",
		FlowVersionID: "version-timed", PathName: "OneMinuteLater",
		EntryElement: "RunLater", RecordID: "record-timed",
		Offset: time.Minute, CanonicalInputs: []byte(`{"$Record":{"Id":"record-timed"}}`),
	}
}

func testTimedFlowRecordEventScope(request TimedFlowDueRequest) persistence.RuntimeWorkScope {
	return persistence.RuntimeWorkScope{
		OrgID: request.Binding.OrgID, AppInstallationID: request.Binding.AppInstallationID,
		RuntimeCapabilityID: mutationsemantic.CapabilityName,
	}
}

func testTimedFlowEventContext(
	t *testing.T,
	eventID string,
	request TimedFlowDueRequest,
) context.Context {
	t.Helper()
	ctx := eventfeature.WithCurrentOutboxEventID(t.Context(), eventID)
	ctx, err := persistence.WithRuntimeWorkScope(ctx, testTimedFlowRecordEventScope(request))
	if err != nil {
		t.Fatal(err)
	}
	return ctx
}

func TestTimedFlowDueRequestAllowsZeroOffsetOnlyForAfterCommit(t *testing.T) {
	request := testTimedFlowDueRequest()
	request.AfterCommit = true
	request.Offset = 0
	if err := request.validateLaunchBound(); err != nil {
		t.Fatalf("zero-offset after-commit request rejected: %v", err)
	}

	request.Offset = time.Minute
	if err := request.validateLaunchBound(); err == nil {
		t.Fatal("positive-offset after-commit request was accepted")
	}

	request.AfterCommit = false
	request.Offset = 0
	if err := request.validateLaunchBound(); err == nil {
		t.Fatal("zero-offset timed request was accepted")
	}
}

func TestTimedFlowDueProducerTiDBCommitsAndRollsBackWithRecordEventAttempt(t *testing.T) {
	db, cleanup := openTiDB(t)
	defer cleanup()
	request := testTimedFlowDueRequest()
	eventScope := testTimedFlowRecordEventScope(request)
	eventScopeDigest, err := eventScope.Digest()
	if err != nil {
		t.Fatal(err)
	}
	createFencedScheduledRunTables(t, db)
	if _, err := db.Exec(`CREATE TABLE _System_RecordEventOutbox (
Id VARCHAR(255) PRIMARY KEY,
OrgId VARCHAR(191) NOT NULL,
AppInstallationId VARCHAR(191) NOT NULL,
RuntimeCapabilityId VARCHAR(255) NOT NULL,
ExecutionScopeDigest CHAR(64) NOT NULL,
CreatedDate DATETIME(6) NOT NULL)`); err != nil {
		t.Fatal(err)
	}
	owner := persistence.RuntimeWorkOwnership{
		WorkFamily: RuntimeScheduledFlowWorkFamily, DeploymentID: "timed-worker",
		WorkerReleaseDigest: "sha256:" + strings.Repeat("c", 64), OwnershipEpoch: 1,
	}
	if _, err := db.Exec(`INSERT INTO _System_RuntimeWorkOwnership
(WorkFamily, DeploymentId, WorkerReleaseDigest, OwnershipEpoch, IsActive, OperationId)
VALUES (?, ?, ?, ?, 1, 'activate-timed-worker')`,
		owner.WorkFamily, owner.DeploymentID, owner.WorkerReleaseDigest, owner.EpochDecimal(),
	); err != nil {
		t.Fatal(err)
	}
	eventTime := time.Date(2026, 7, 29, 12, 0, 0, 123000000, time.Local)
	for _, eventID := range []string{"event-commit", "event-rollback", "event-after-transfer"} {
		if _, err := db.Exec(`INSERT INTO _System_RecordEventOutbox
(Id, OrgId, AppInstallationId, RuntimeCapabilityId, ExecutionScopeDigest, CreatedDate)
VALUES (?, 'org-timed', 'installation-timed', ?, ?, ?)`,
			eventID,
			eventScope.RuntimeCapabilityID,
			eventScopeDigest,
			eventTime,
		); err != nil {
			t.Fatal(err)
		}
	}
	transactions := persistence.NewTransactionManager(database.NewConnectionFromDB(db))
	producer, err := NewFencedTimedFlowDueProducer(db, transactions, owner)
	if err != nil {
		t.Fatal(err)
	}
	if err := producer.CheckReady(t.Context()); err != nil {
		t.Fatal(err)
	}
	ctx := testTimedFlowEventContext(t, "event-commit", request)
	if err := transactions.WithPessimisticTransaction(ctx, func(txCtx context.Context) error {
		return producer.Accept(txCtx, request)
	}); err != nil {
		t.Fatal(err)
	}
	var status, state, entry string
	var attempt int
	var dueAt time.Time
	if err := db.QueryRow(`SELECT Status, State, Attempt, EntryElement, DueAt
FROM _System_ScheduledRunReservation WHERE FlowDefinitionId = ? AND InputKey = ?`,
		request.FlowDefinitionID,
		request.RecordID,
	).Scan(&status, &state, &attempt, &entry, &dueAt); err != nil {
		t.Fatal(err)
	}
	if status != "accepted" || state != "ACCEPTED" || attempt != 0 ||
		entry != request.EntryElement || !dueAt.Equal(eventTime.Add(time.Minute)) {
		t.Fatalf("accepted due row = status=%q state=%q attempt=%d entry=%q due=%s",
			status, state, attempt, entry, dueAt)
	}

	rollbackRequest := request
	rollbackRequest.RecordID = "record-rollback"
	rollbackCtx := testTimedFlowEventContext(t, "event-rollback", request)
	rollbackSentinel := errors.New("roll back event attempt")
	err = transactions.WithPessimisticTransaction(rollbackCtx, func(txCtx context.Context) error {
		if err := producer.Accept(txCtx, rollbackRequest); err != nil {
			return err
		}
		return rollbackSentinel
	})
	if !errors.Is(err, rollbackSentinel) {
		t.Fatalf("rollback transaction error = %v", err)
	}
	var rolledBack int
	if err := db.QueryRow(`SELECT COUNT(*) FROM _System_ScheduledRunReservation
WHERE InputKey = 'record-rollback'`).Scan(&rolledBack); err != nil {
		t.Fatal(err)
	}
	if rolledBack != 0 {
		t.Fatalf("rolled-back timed occurrences = %d, want zero", rolledBack)
	}

	if _, err := db.Exec(`UPDATE _System_RuntimeWorkOwnership
SET DeploymentId = 'timed-worker-next', OwnershipEpoch = 2, OperationId = 'transfer-timed-worker'
WHERE WorkFamily = ?`, RuntimeScheduledFlowWorkFamily); err != nil {
		t.Fatal(err)
	}
	transferredRequest := request
	transferredRequest.RecordID = "record-after-transfer"
	transferredCtx := testTimedFlowEventContext(t, "event-after-transfer", request)
	err = transactions.WithPessimisticTransaction(transferredCtx, func(txCtx context.Context) error {
		return producer.Accept(txCtx, transferredRequest)
	})
	if !errors.Is(err, persistence.ErrRuntimeWorkOwnershipLost) {
		t.Fatalf("accept after scheduled ownership transfer = %v", err)
	}
	var transferred int
	if err := db.QueryRow(`SELECT COUNT(*) FROM _System_ScheduledRunReservation
WHERE InputKey = 'record-after-transfer'`).Scan(&transferred); err != nil {
		t.Fatal(err)
	}
	if transferred != 0 {
		t.Fatalf("post-transfer timed occurrences = %d, want zero", transferred)
	}
}