package runtimehandoff

import (
	"context"
	"database/sql/driver"
	"regexp"
	"strings"
	"testing"
	"time"

	"github.com/DATA-DOG/go-sqlmock"
	connector_inbound "github.com/pingcap-inc/nexuscrm-connectors/features/connector/inbound"
	commandv1 "github.com/pingcap-inc/nexuscrm-contract/pkg/runtime/commandv1"
)

func TestStorePrepareFreezesFirstRequestAcrossRetry(t *testing.T) {
	for _, test := range []struct {
		name          string
		affected      int64
		storedPayload []byte
		wantDuplicate bool
		wantConflict  bool
	}{
		{
			name: "fresh", affected: 1,
			storedPayload: []byte(`{"definition":"v1"}`),
		},
		{
			name: "identical retry", affected: 0,
			storedPayload: []byte(`{"definition":"v1"}`),
			wantDuplicate: true,
		},
		{
			name: "retry rebuilt different bytes", affected: 0,
			storedPayload: []byte(`{"definition":"older"}`),
			wantDuplicate: true, wantConflict: true,
		},
	} {
		t.Run(test.name, func(t *testing.T) {
			db, mock, err := sqlmock.New()
			if err != nil {
				t.Fatal(err)
			}
			defer db.Close()
			store, err := NewStore(db)
			if err != nil {
				t.Fatal(err)
			}
			scope := testScope()
			candidate := testPreparedCommand(
				[]byte(`{"definition":"v1"}`),
			)
			mock.ExpectExec(regexp.QuoteMeta(`
INSERT IGNORE INTO _System_ConnectorRuntimeHandoffReceipt
       (WorkId, EventId, Namespace, InstalledPackId, EventKind, RouteVersion,
        OrgId, AppInstallationId, AuthorizedWorkloadId,
        ScopeDigest, MinimumCatalogRevision, BindingViewDigest, CommandId, IdempotencyKey,
        CommandKind, PartitionKey, PayloadFormatVersion, Payload,
        PayloadDigest, CreatedDate)
SELECT ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, NULLIF(?, ''), ?, ?, ?, UTC_TIMESTAMP(6)
FROM _System_ConnectorWorkItem
WHERE WorkId = ? AND EventId = ? AND State = 'processing' AND LeaseToken = ?
  AND LeaseUntil > UTC_TIMESTAMP(6)
LIMIT 1`)).
				WithArgs(testPrepareArgs(scope, candidate)...).
				WillReturnResult(sqlmock.NewResult(0, test.affected))
			stored := testPreparedCommand(test.storedPayload)
			mock.ExpectQuery(
				"(?s)SELECT r.Namespace.*LeaseUntil > UTC_TIMESTAMP",
			).
				WithArgs(scope.LeaseToken, scope.WorkItemID, scope.EventID).
				WillReturnRows(testStoredRows(stored))
			ctx, cancel := context.WithTimeout(
				context.Background(),
				time.Second,
			)
			defer cancel()
			got, duplicate, err := store.Prepare(
				ctx,
				scope,
				candidate,
			)
			if test.wantConflict {
				if err != ErrConflict || !duplicate {
					t.Fatalf(
						"Prepare() = %#v duplicate=%t err=%v",
						got,
						duplicate,
						err,
					)
				}
			} else {
				if err != nil || duplicate != test.wantDuplicate ||
					got.Request.PayloadDigest !=
						candidate.Request.PayloadDigest {
					t.Fatalf(
						"Prepare() = %#v duplicate=%t err=%v",
						got,
						duplicate,
						err,
					)
				}
			}
			if err := mock.ExpectationsWereMet(); err != nil {
				t.Fatal(err)
			}
		})
	}
}

func TestStorePrepareRejectsLostLeaseWithoutCreatingAuthority(t *testing.T) {
	db, mock, err := sqlmock.New()
	if err != nil {
		t.Fatal(err)
	}
	defer db.Close()
	store, err := NewStore(db)
	if err != nil {
		t.Fatal(err)
	}
	scope := testScope()
	candidate := testPreparedCommand([]byte(`{"definition":"v1"}`))
	mock.ExpectExec(
		"INSERT IGNORE INTO _System_ConnectorRuntimeHandoffReceipt",
	).
		WithArgs(testPrepareArgs(scope, candidate)...).
		WillReturnResult(sqlmock.NewResult(0, 0))
	mock.ExpectQuery(
		"(?s)SELECT r.Namespace.*LeaseUntil > UTC_TIMESTAMP",
	).
		WithArgs(scope.LeaseToken, scope.WorkItemID, scope.EventID).
		WillReturnRows(testStoredRows(PreparedCommand{}))
	ctx, cancel := context.WithTimeout(context.Background(), time.Second)
	defer cancel()
	if got, duplicate, err := store.Prepare(
		ctx,
		scope,
		candidate,
	); err != ErrLeaseLost || duplicate || got.Request.CommandID != "" {
		t.Fatalf(
			"Prepare() = %#v duplicate=%t err=%v",
			got,
			duplicate,
			err,
		)
	}
	if err := mock.ExpectationsWereMet(); err != nil {
		t.Fatal(err)
	}
}

func TestStoreLoadRejectsReceiptAfterWorkItemLeaseLoss(t *testing.T) {
	db, mock, err := sqlmock.New()
	if err != nil {
		t.Fatal(err)
	}
	defer db.Close()
	store, err := NewStore(db)
	if err != nil {
		t.Fatal(err)
	}
	scope := testScope()
	mock.ExpectQuery(
		"(?s)SELECT r.Namespace.*LeaseUntil > UTC_TIMESTAMP",
	).
		WithArgs(scope.LeaseToken, scope.WorkItemID, scope.EventID).
		WillReturnRows(testStoredRowsWithLease(
			testPreparedCommand([]byte(`{"definition":"v1"}`)),
			false,
		))
	ctx, cancel := context.WithTimeout(context.Background(), time.Second)
	defer cancel()
	if got, found, err := store.LoadForScope(ctx, scope); err != ErrLeaseLost ||
		found || got.Request.CommandID != "" {
		t.Fatalf("LoadForScope() = %#v found=%t err=%v", got, found, err)
	}
	if err := mock.ExpectationsWereMet(); err != nil {
		t.Fatal(err)
	}
}

func TestStoreVerifySchemaChecksCompleteReceiptShape(t *testing.T) {
	db, mock, err := sqlmock.New()
	if err != nil {
		t.Fatal(err)
	}
	defer db.Close()
	store, err := NewStore(db)
	if err != nil {
		t.Fatal(err)
	}
	columns := []string{
		"WorkId", "EventId", "Namespace", "InstalledPackId", "EventKind",
		"RouteVersion", "OrgId", "AppInstallationId", "AuthorizedWorkloadId",
		"ScopeDigest", "MinimumCatalogRevision", "BindingViewDigest", "CommandId",
		"IdempotencyKey", "CommandKind", "PartitionKey", "PayloadFormatVersion",
		"Payload", "PayloadDigest", "CreatedDate", "LastModifiedDate",
	}
	mock.ExpectQuery("SELECT WorkId, EventId, Namespace").
		WillReturnRows(sqlmock.NewRows(columns))
	expectCompleteReceiptIndexes(mock)
	ctx, cancel := context.WithTimeout(context.Background(), time.Second)
	defer cancel()
	if err := store.VerifySchema(ctx); err != nil {
		t.Fatalf("VerifySchema() error = %v", err)
	}
	if err := mock.ExpectationsWereMet(); err != nil {
		t.Fatal(err)
	}
}

func TestStoreVerifySchemaRejectsMissingReceiptUniqueness(t *testing.T) {
	db, mock, err := sqlmock.New()
	if err != nil {
		t.Fatal(err)
	}
	defer db.Close()
	store, err := NewStore(db)
	if err != nil {
		t.Fatal(err)
	}
	columns := []string{
		"WorkId", "EventId", "Namespace", "InstalledPackId", "EventKind",
		"RouteVersion", "OrgId", "AppInstallationId", "AuthorizedWorkloadId",
		"ScopeDigest", "MinimumCatalogRevision", "BindingViewDigest", "CommandId",
		"IdempotencyKey", "CommandKind", "PartitionKey", "PayloadFormatVersion",
		"Payload", "PayloadDigest", "CreatedDate", "LastModifiedDate",
	}
	mock.ExpectQuery("SELECT WorkId, EventId, Namespace").
		WillReturnRows(sqlmock.NewRows(columns))
	mock.ExpectQuery("SELECT INDEX_NAME, NON_UNIQUE").
		WillReturnRows(sqlmock.NewRows([]string{
			"INDEX_NAME", "NON_UNIQUE", "SEQ_IN_INDEX", "COLUMN_NAME", "SUB_PART",
		}).AddRow("PRIMARY", 0, 1, "WorkId", nil))
	ctx, cancel := context.WithTimeout(context.Background(), time.Second)
	defer cancel()
	if err := store.VerifySchema(ctx); err == nil ||
		!strings.Contains(err.Error(), "uniqueness is unavailable") {
		t.Fatalf("VerifySchema() error = %v", err)
	}
	if err := mock.ExpectationsWereMet(); err != nil {
		t.Fatal(err)
	}
}

func expectCompleteReceiptIndexes(mock sqlmock.Sqlmock) {
	mock.ExpectQuery("SELECT INDEX_NAME, NON_UNIQUE").
		WillReturnRows(sqlmock.NewRows([]string{
			"INDEX_NAME", "NON_UNIQUE", "SEQ_IN_INDEX", "COLUMN_NAME", "SUB_PART",
		}).
			AddRow("PRIMARY", 0, 1, "WorkId", nil).
			AddRow("ux_connector_runtime_handoff_command", 0, 1, "ScopeDigest", nil).
			AddRow("ux_connector_runtime_handoff_command", 0, 2, "CommandId", nil).
			AddRow("ux_connector_runtime_handoff_idempotency", 0, 1, "ScopeDigest", nil).
			AddRow("ux_connector_runtime_handoff_idempotency", 0, 2, "IdempotencyKey", nil))
}

func testScope() connector_inbound.RuntimeCommandHandoffScope {
	return connector_inbound.RuntimeCommandHandoffScope{
		WorkItemID: "work-1", EventID: "event-1",
		LeaseToken: "lease-1", Namespace: "connector-pack",
		InstalledPackID: "install-1", EventKind: "contact.created",
		SourcePartition: "source-1", RouteVersion: 7,
	}
}

func testPreparedCommand(payload []byte) PreparedCommand {
	revision := uint64(7)
	return PreparedCommand{
		OrgID: "org-1", AppInstallationID: "app-install-1",
		AuthorizedWorkloadID:   "integration-1",
		ScopeDigest:            consumerScopeDigest("org-1", "app-install-1", "integration-1"),
		MinimumCatalogRevision: 7,
		BindingViewDigest:      "sha256:" + strings.Repeat("a", 64),
		Request: commandv1.Request{
			CommandID:            "connector-work:work-1",
			IdempotencyKey:       "connector-work:work-1",
			CommandKind:          "runtime.flow.execute",
			PartitionKey:         "source-1",
			PayloadFormatVersion: 1,
			Payload:              append([]byte(nil), payload...),
			PayloadDigest:        commandv1.ComputePayloadDigest(payload),
			MinCatalogRevision:   &revision,
		},
	}
}

func testPrepareArgs(
	scope connector_inbound.RuntimeCommandHandoffScope,
	candidate PreparedCommand,
) []driver.Value {
	request := candidate.Request
	return []driver.Value{
		scope.WorkItemID,
		scope.EventID,
		scope.Namespace,
		scope.InstalledPackID,
		scope.EventKind,
		"7",
		candidate.OrgID,
		candidate.AppInstallationID,
		candidate.AuthorizedWorkloadID,
		candidate.ScopeDigest,
		"7",
		candidate.BindingViewDigest,
		request.CommandID,
		request.IdempotencyKey,
		request.CommandKind,
		request.PartitionKey,
		request.PayloadFormatVersion,
		request.Payload,
		request.PayloadDigest,
		scope.WorkItemID,
		scope.EventID,
		scope.LeaseToken,
	}
}

func testStoredRows(value PreparedCommand) *sqlmock.Rows {
	return testStoredRowsWithLease(value, true)
}

func testStoredRowsWithLease(value PreparedCommand, leaseLive bool) *sqlmock.Rows {
	rows := sqlmock.NewRows([]string{
		"Namespace", "InstalledPackId", "EventKind", "RouteVersion",
		"OrgId", "AppInstallationId", "AuthorizedWorkloadId",
		"ScopeDigest", "MinimumCatalogRevision", "BindingViewDigest", "CommandId",
		"IdempotencyKey", "CommandKind", "PartitionKey",
		"PayloadFormatVersion", "Payload", "PayloadDigest", "LeaseLive",
	})
	if value.Request.CommandID == "" {
		return rows
	}
	var partition any
	if value.Request.PartitionKey != "" {
		partition = value.Request.PartitionKey
	}
	return rows.AddRow(
		"connector-pack",
		"install-1",
		"contact.created",
		[]byte("7"),
		value.OrgID,
		value.AppInstallationID,
		value.AuthorizedWorkloadID,
		value.ScopeDigest,
		[]byte("7"),
		value.BindingViewDigest,
		value.Request.CommandID,
		value.Request.IdempotencyKey,
		value.Request.CommandKind,
		partition,
		value.Request.PayloadFormatVersion,
		value.Request.Payload,
		value.Request.PayloadDigest,
		leaseLive,
	)
}