package connector_workqueue

import (
	"context"
	"errors"
	"testing"
	"time"

	"github.com/DATA-DOG/go-sqlmock"
	contractmodels "github.com/pingcap-inc/nexuscrm-contract/pkg/models"
)

func TestRepositoryCompleteWorkItemWithAttemptFallsBackBeforeTerminalAtSchema(t *testing.T) {
	db, mock, err := sqlmock.New()
	if err != nil {
		t.Fatalf("sqlmock.New: %v", err)
	}
	defer db.Close()

	now := time.Date(2026, 7, 22, 14, 47, 0, 0, time.UTC)
	ctx := WithWorkItemAttemptOutcome(context.Background(), WorkItemAttemptOutcome{
		FlowExecutionID: "flow-execution-legacy-schema",
	})

	mock.ExpectExec("SET SESSION tidb_txn_mode = 'pessimistic'").
		WillReturnResult(sqlmock.NewResult(0, 0))
	mock.ExpectBegin()
	mock.ExpectQuery(`(?s)SELECT EventId, Attempt.*FROM _System_ConnectorWorkItem.*FOR UPDATE`).
		WithArgs("work-legacy-schema", WorkStateProcessing, "lease-legacy-schema").
		WillReturnRows(sqlmock.NewRows([]string{"EventId", "Attempt"}).AddRow("event-legacy-schema", 2))
	mock.ExpectExec(`(?s)UPDATE _System_ConnectorWorkItem.*TerminalAt = \?.*WHERE WorkId = \?`).
		WithArgs(WorkStateProcessed, now, now, "work-legacy-schema", WorkStateProcessing, "lease-legacy-schema").
		WillReturnError(errors.New("Error 1054 (42S22): Unknown column 'TerminalAt' in 'field list'"))
	mock.ExpectExec(`(?s)UPDATE _System_ConnectorWorkItem.*LeaseToken = NULL.*LastModifiedDate = \?.*WHERE WorkId = \?`).
		WithArgs(WorkStateProcessed, now, "work-legacy-schema", WorkStateProcessing, "lease-legacy-schema").
		WillReturnResult(sqlmock.NewResult(0, 1))
	mock.ExpectExec(`(?s)INSERT INTO _System_ConnectorWorkAttempt`).
		WithArgs(
			"work-legacy-schema", 2, "event-legacy-schema", "flow-execution-legacy-schema", WorkStateProcessed,
			"", "", "", "", "", false, now, now,
		).
		WillReturnResult(sqlmock.NewResult(1, 1))
	mock.ExpectQuery(`SELECT InstalledPackId,\s+Generation,\s+COALESCE\(SourceLaneKey, ''\),\s+COALESCE\(TargetMutationLaneKey, ''\)`).
		WithArgs("work-legacy-schema").
		WillReturnRows(sqlmock.NewRows([]string{"InstalledPackId", "Generation", "SourceLaneKey", "TargetMutationLaneKey"}).
			AddRow("pack-1", int64(1), "", ""))
	mock.ExpectCommit()
	mock.ExpectExec("SET SESSION tidb_txn_mode = DEFAULT").
		WillReturnResult(sqlmock.NewResult(0, 0))

	completed, err := NewRepository(db, nil, nil).CompleteWorkItem(
		ctx,
		"work-legacy-schema",
		"lease-legacy-schema",
		WorkStateProcessed,
		now,
	)
	if err != nil {
		t.Fatalf("CompleteWorkItem: %v", err)
	}
	if !completed {
		t.Fatal("CompleteWorkItem completed=false, want old-schema compatibility update")
	}
	if err := mock.ExpectationsWereMet(); err != nil {
		t.Fatalf("sqlmock expectations: %v", err)
	}
}

func TestRepositoryFailWorkItemAppendsFlowAttemptBeforeCallerTransactionCommit(t *testing.T) {
	db, mock, err := sqlmock.New()
	if err != nil {
		t.Fatalf("sqlmock.New: %v", err)
	}
	defer db.Close()

	now := time.Date(2026, 7, 22, 14, 50, 0, 0, time.UTC)
	ctx := WithWorkItemAttemptOutcome(context.Background(), WorkItemAttemptOutcome{
		FlowExecutionID:    "flow-execution-failed",
		PresentedErrorCode: "EMAIL_MALFORMED",
		Cause: &contractmodels.FlowExecutionCause{
			Code:      "DB_LOCK_WAIT_TIMEOUT",
			Category:  contractmodels.FlowExecutionCauseCategoryConcurrency,
			Retriable: true,
		},
	})

	mock.ExpectBegin()
	tx, err := db.Begin()
	if err != nil {
		t.Fatalf("Begin: %v", err)
	}
	repo := NewRepository(db, nil, nil).WithTx(tx)
	mock.ExpectQuery(`(?s)SELECT w.EventId, w.FailureCount.*FOR UPDATE`).
		WithArgs("work-fail", "event-fail", WorkStateProcessing, "lease-fail").
		WillReturnRows(sqlmock.NewRows([]string{"EventId", "FailureCount"}).AddRow("event-fail", 0))
	mock.ExpectQuery(`(?s)SELECT Attempt.*FROM _System_ConnectorWorkItem`).
		WithArgs("work-fail", "event-fail").
		WillReturnRows(sqlmock.NewRows([]string{"Attempt"}).AddRow(2))
	mock.ExpectExec(`(?s)UPDATE _System_ConnectorWorkItem.*FailureCount = \?.*FailureClass = \?`).
		WithArgs(
			WorkStateRetry, now.Add(5*time.Second), 1,
			FailureClassInboundExecution, FailureSummaryInbound, now, nil, now,
			"work-fail", "event-fail", WorkStateProcessing, "lease-fail", 0,
		).
		WillReturnResult(sqlmock.NewResult(0, 1))
	mock.ExpectExec(`(?s)INSERT INTO _System_ConnectorWorkAttempt`).
		WithArgs(
			"work-fail", 2, "event-fail", "flow-execution-failed", WorkStateRetry,
			"EMAIL_MALFORMED", "DB_LOCK_WAIT_TIMEOUT", "concurrency", "", "", true, now, now,
		).
		WillReturnResult(sqlmock.NewResult(1, 1))

	result, failed, err := repo.FailWorkItemDeferredLaneReconcile(ctx, FailWorkItemInput{
		WorkID:         "work-fail",
		EventID:        "event-fail",
		LeaseToken:     "lease-fail",
		FailureClass:   FailureClassInboundExecution,
		FailureSummary: FailureSummaryInbound,
		RetryThreshold: 5,
		RetryDelays:    []time.Duration{5 * time.Second},
		Now:            now,
	})
	if err != nil || !failed || result == nil || result.State != WorkStateRetry {
		t.Fatalf("FailWorkItem = (%#v, %v, %v), want retry", result, failed, err)
	}
	mock.ExpectCommit()
	if err := tx.Commit(); err != nil {
		t.Fatalf("Commit: %v", err)
	}
	if err := mock.ExpectationsWereMet(); err != nil {
		t.Fatalf("sqlmock expectations: %v", err)
	}
}