package connector_ingress_receipts

import (
	"bytes"
	"context"
	"database/sql"
	"database/sql/driver"
	"encoding/base64"
	"encoding/json"
	"errors"
	"net/http"
	"regexp"
	"strings"
	"testing"
	"time"

	"github.com/DATA-DOG/go-sqlmock"
	"github.com/rs/zerolog"

	"github.com/pingcap-inc/nexuscrm-connectors/features/namedcredential"
	"github.com/pingcap-inc/nexuscrm-contract/pkg/constants"
)

func TestVerifyReadyChecksReceiptShapeWithoutReadingBusinessRows(t *testing.T) {
	for _, probeErr := range []error{nil, errors.New("missing receipt column")} {
		db, mock := newReceiptSQLMock(t)
		service := newReceiptTestService(db, receiptTestNow())
		expectation := mock.ExpectQuery(regexp.QuoteMeta(connectorIngressReceiptSelectSQL + " WHERE 1 = 0"))
		if probeErr != nil {
			expectation.WillReturnError(probeErr)
		} else {
			expectation.WillReturnRows(sqlmock.NewRows([]string{"Id"})).RowsWillBeClosed()
		}
		if err := service.VerifyReady(t.Context()); !errors.Is(err, probeErr) {
			t.Fatalf("readiness=%v want %v", err, probeErr)
		}
		requireReceiptSQLExpectations(t, mock)
		db.Close()
	}
}

func TestCreateVerifiedReceiptEncryptsBodyAndAllowlistedHeaders(t *testing.T) {
	db, mock := newReceiptSQLMock(t)
	defer db.Close()

	now := receiptTestNow()
	in := receiptTestInput()
	bodyHash := BodyHash(in.RawBody)
	targetHash := RequestTargetHash(http.MethodPost, in.RequestURI, in.RequestURL)
	receiptKey := ReceiptKeyFor(in.Namespace, in.InstalledPackID, in.SecretName, in.DeliveryIdentity)
	purgeAfter := DefaultPurgeAfter(now)

	mock.ExpectExec(regexp.QuoteMeta("INSERT INTO _System_ConnectorIngressReceipt")).
		WithArgs(
			"receipt-1", "provider_ns", "install-1", "v1", "default", "hmac-sha256",
			"signatureVerifier", "envelopeParser", http.MethodPost, targetHash,
			notContainingArg{fragments: []string{"query-secret", in.RequestURI, in.RequestURL}},
			receiptKey, bodyHash, len(in.RawBody),
			notContainingArg{fragments: []string{`"secret":"body-token"`}},
			notContainingArg{fragments: []string{"Bearer raw-token", "sig-secret", "signed-event-id"}},
			StatusPending, purgeAfter, now, now,
		).
		WillReturnResult(sqlmock.NewResult(0, 1))

	receipt, created, err := newReceiptTestService(db, now).CreateVerifiedReceipt(context.Background(), in)
	if err != nil {
		t.Fatalf("CreateVerifiedReceipt: %v", err)
	}
	if !created {
		t.Fatal("created = false, want true")
	}
	if receipt.ReceiptKey != receiptKey || receipt.BodyHash != bodyHash {
		t.Fatalf("receipt key/hash = %q/%q, want %q/%q", receipt.ReceiptKey, receipt.BodyHash, receiptKey, bodyHash)
	}
	if receipt.RequestTargetHash != targetHash {
		t.Fatalf("request target hash = %q, want %q", receipt.RequestTargetHash, targetHash)
	}
	if receipt.Headers.Get("Content-Type") != "application/json" {
		t.Fatalf("stored content-type = %q", receipt.Headers.Get("Content-Type"))
	}
	if got := receipt.Headers.Get("Authorization"); got != "" {
		t.Fatalf("authorization leaked into receipt headers: %q", got)
	}
	if got := receipt.Headers.Get("X-Event-Id"); got != "" {
		t.Fatalf("blocked verifier header leaked into receipt headers: %q", got)
	}
	if receipt.PurgeAfter == nil || !receipt.PurgeAfter.Equal(purgeAfter) {
		t.Fatalf("purgeAfter = %v, want %s", receipt.PurgeAfter, purgeAfter)
	}
	requireReceiptSQLExpectations(t, mock)
}

func TestCreateVerifiedReceiptGeneratesTimeOrderedID(t *testing.T) {
	db, mock := newReceiptSQLMock(t)
	defer db.Close()

	now := time.Date(2026, 6, 19, 2, 0, 0, 123456789, time.UTC)
	in := receiptTestInput()
	in.ID = ""
	bodyHash := BodyHash(in.RawBody)
	targetHash := RequestTargetHash(http.MethodPost, in.RequestURI, in.RequestURL)
	receiptKey := ReceiptKeyFor(in.Namespace, in.InstalledPackID, in.SecretName, in.DeliveryIdentity)
	purgeAfter := DefaultPurgeAfter(now)
	wantID := "ir:20260619T020000.123456789Z:rand-a"

	mock.ExpectExec(regexp.QuoteMeta("INSERT INTO _System_ConnectorIngressReceipt")).
		WithArgs(
			wantID, "provider_ns", "install-1", "v1", "default", "hmac-sha256",
			"signatureVerifier", "envelopeParser", http.MethodPost, targetHash,
			sqlmock.AnyArg(), receiptKey, bodyHash, len(in.RawBody),
			sqlmock.AnyArg(), sqlmock.AnyArg(), StatusPending, purgeAfter, now, now,
		).
		WillReturnResult(sqlmock.NewResult(0, 1))

	receipt, created, err := NewService(db, receiptTestCipherNoT(), func() string { return "rand-a" }, func() time.Time { return now }).CreateVerifiedReceipt(context.Background(), in)
	if err != nil {
		t.Fatalf("CreateVerifiedReceipt: %v", err)
	}
	if !created || receipt.ID != wantID {
		t.Fatalf("CreateVerifiedReceipt ID = (%q, %v), want %q/true", receipt.ID, created, wantID)
	}
	requireReceiptSQLExpectations(t, mock)
}

func TestOrderedReceiptIDSortsWithinSameDatetimeSecond(t *testing.T) {
	base := time.Date(2026, 6, 19, 2, 0, 0, 0, time.UTC)
	first := orderedReceiptID(base.Add(100*time.Nanosecond), "z")
	second := orderedReceiptID(base.Add(200*time.Nanosecond), "a")
	if !(first < second) {
		t.Fatalf("orderedReceiptID same-second order = %q >= %q", first, second)
	}
}

func TestCreateVerifiedReceiptRequiresStableReceiptKey(t *testing.T) {
	db, mock := newReceiptSQLMock(t)
	defer db.Close()

	in := receiptTestInput()
	in.DeliveryIdentity = ""
	in.ReceiptKey = ""

	_, _, err := newReceiptTestService(db, receiptTestNow()).CreateVerifiedReceipt(context.Background(), in)
	if !errors.Is(err, ErrInvalidReceipt) {
		t.Fatalf("CreateVerifiedReceipt without stable key = %v, want ErrInvalidReceipt", err)
	}
	requireReceiptSQLExpectations(t, mock)
}

func TestCreateVerifiedReceiptAllowsLegacyBlankSecretAlgorithm(t *testing.T) {
	db, mock := newReceiptSQLMock(t)
	defer db.Close()

	in := receiptTestInput()
	in.SecretAlgorithm = ""

	mock.ExpectExec(regexp.QuoteMeta("INSERT INTO _System_ConnectorIngressReceipt")).
		WillReturnResult(sqlmock.NewResult(0, 1))

	receipt, created, err := newReceiptTestService(db, receiptTestNow()).CreateVerifiedReceipt(context.Background(), in)
	if err != nil {
		t.Fatalf("CreateVerifiedReceipt legacy blank algorithm: %v", err)
	}
	if !created || receipt == nil {
		t.Fatalf("CreateVerifiedReceipt legacy blank algorithm = (%#v, %v), want created receipt", receipt, created)
	}
	if receipt.SecretAlgorithm != "" {
		t.Fatalf("SecretAlgorithm = %q, want legacy blank preserved", receipt.SecretAlgorithm)
	}
	requireReceiptSQLExpectations(t, mock)
}

func TestCreateVerifiedReceiptDuplicateReturnsExisting(t *testing.T) {
	db, mock := newReceiptSQLMock(t)
	defer db.Close()

	now := receiptTestNow()
	in := receiptTestInput()
	in.DeliveryIdentity = "delivery-1"
	bodyHash := BodyHash(in.RawBody)
	targetHash := RequestTargetHash(http.MethodPost, in.RequestURI, in.RequestURL)
	receiptKey := ReceiptKeyFor(in.Namespace, in.InstalledPackID, in.SecretName, in.DeliveryIdentity)
	cipher := receiptTestCipher(t)
	targetCiphertext := encryptReceiptTarget(t, cipher, in.RequestURI, in.RequestURL)
	bodyCiphertext := encryptReceiptBody(t, cipher, in.RawBody)
	headerCiphertext := encryptReceiptHeaders(t, cipher, http.Header{"Content-Type": []string{"application/json"}})

	mock.ExpectExec(regexp.QuoteMeta("INSERT INTO _System_ConnectorIngressReceipt")).
		WillReturnError(errors.New("Error 1062 duplicate entry"))
	mock.ExpectQuery(regexp.QuoteMeta("FROM _System_ConnectorIngressReceipt")).
		WithArgs(receiptKey).
		WillReturnRows(receiptRows().AddRow(
			"receipt-existing", "provider_ns", "install-1", "v1", "default", "hmac-sha256",
			"signatureVerifier", "envelopeParser", http.MethodPost, targetHash, targetCiphertext,
			receiptKey, bodyHash, len(in.RawBody), bodyCiphertext, headerCiphertext, StatusPending, 0,
			nil, "", "", nil, nil, `[]`, `{}`, "", nil, nil, now, now,
		))

	receipt, created, err := NewService(db, cipher, func() string { return "receipt-1" }, func() time.Time { return now }).CreateVerifiedReceipt(context.Background(), in)
	if err != nil {
		t.Fatalf("CreateVerifiedReceipt duplicate: %v", err)
	}
	if created {
		t.Fatal("created = true, want false for duplicate")
	}
	if receipt.ID != "receipt-existing" || string(receipt.RawBody) != string(in.RawBody) {
		t.Fatalf("duplicate receipt = %#v", receipt)
	}
	if receipt.RequestURI != in.RequestURI || receipt.RequestURL != in.RequestURL {
		t.Fatalf("decrypted request target = %q/%q", receipt.RequestURI, receipt.RequestURL)
	}
	requireReceiptSQLExpectations(t, mock)
}

func TestCreateVerifiedReceiptDuplicateRejectsBodyHashCollision(t *testing.T) {
	db, mock := newReceiptSQLMock(t)
	defer db.Close()

	now := receiptTestNow()
	in := receiptTestInput()
	in.DeliveryIdentity = "delivery-1"
	bodyHash := BodyHash(in.RawBody)
	targetHash := RequestTargetHash(http.MethodPost, in.RequestURI, in.RequestURL)
	receiptKey := ReceiptKeyFor(in.Namespace, in.InstalledPackID, in.SecretName, in.DeliveryIdentity)
	cipher := receiptTestCipher(t)
	targetCiphertext := encryptReceiptTarget(t, cipher, in.RequestURI, in.RequestURL)
	otherBody := bytes.Repeat([]byte("x"), len(in.RawBody))
	bodyCiphertext := encryptReceiptBody(t, cipher, otherBody)
	headerCiphertext := encryptReceiptHeaders(t, cipher, http.Header{})

	mock.ExpectExec(regexp.QuoteMeta("INSERT INTO _System_ConnectorIngressReceipt")).
		WillReturnError(errors.New("Error 1062 duplicate entry"))
	mock.ExpectQuery(regexp.QuoteMeta("FROM _System_ConnectorIngressReceipt")).
		WithArgs(receiptKey).
		WillReturnRows(receiptRows().AddRow(
			"receipt-existing", "provider_ns", "install-1", "v1", "default", "hmac-sha256",
			"signatureVerifier", "envelopeParser", http.MethodPost, targetHash, targetCiphertext,
			receiptKey, bodyHash, len(in.RawBody), bodyCiphertext, headerCiphertext, StatusPending, 0,
			nil, "", "", nil, nil, `[]`, `{}`, "", nil, nil, now, now,
		))

	receipt, created, err := NewService(db, cipher, func() string { return "receipt-1" }, func() time.Time { return now }).CreateVerifiedReceipt(context.Background(), in)
	if !errors.Is(err, ErrReceiptKeyConflict) {
		t.Fatalf("CreateVerifiedReceipt duplicate body collision err = %v, want ErrReceiptKeyConflict", err)
	}
	if created || receipt != nil {
		t.Fatalf("CreateVerifiedReceipt collision = (%#v, %v), want nil/false", receipt, created)
	}
	requireReceiptSQLExpectations(t, mock)
}

func TestAllowlistedHeadersStoresOnlyPackDeclaredNames(t *testing.T) {
	headers := http.Header{
		"Content-Type":        []string{"application/json"},
		"Authorization":       []string{"Bearer token"},
		"Cookie":              []string{"sid=secret"},
		"X-Test-Signature":    []string{"sig-secret"},
		"X-Upstream-Token":    []string{"upstream-token"},
		"X-Client-Secret":     []string{"client-secret"},
		"X-Foo-Hmac-Sha256":   []string{"pack-authored-signature"},
		"X-Correlation-Id":    []string{"corr-1"},
		"Proxy-Authorization": []string{"Basic secret"},
		"X-Not-In-Safe-Set":   []string{"not-sensitive-but-not-authoritative"},
	}
	got := AllowlistedHeaders(headers, []string{
		"Content-Type",
		"X-Correlation-Id",
	})
	if got.Get("Content-Type") != "application/json" || got.Get("X-Correlation-Id") != "corr-1" {
		t.Fatalf("declared allowlisted headers = %#v", got)
	}
	for _, name := range []string{"Authorization", "Cookie", "X-Test-Signature", "X-Upstream-Token", "X-Client-Secret", "X-Foo-Hmac-Sha256", "Proxy-Authorization", "X-Not-In-Safe-Set"} {
		if got.Get(name) != "" {
			t.Fatalf("undeclared header %s leaked into allowlisted headers: %#v", name, got)
		}
	}
}

func TestAllowlistedHeadersScrubsSensitiveNamesEvenWhenPackDeclared(t *testing.T) {
	headers := http.Header{
		"Content-Type":        []string{"application/json"},
		"Authorization":       []string{"Bearer token"},
		"Cookie":              []string{"sid=secret"},
		"X-Test-Signature":    []string{"sig-secret"},
		"X-Upstream-Token":    []string{"upstream-token"},
		"X-Client-Secret":     []string{"client-secret"},
		"X-Foo-Hmac-Sha256":   []string{"pack-authored-signature"},
		"X-Api-Credential":    []string{"credential"},
		"X-Session-Id":        []string{"session"},
		"Proxy-Authorization": []string{"Basic secret"},
		"X-Request-Id":        []string{"req-1"},
	}
	got := AllowlistedHeaders(headers, []string{
		"Content-Type",
		"Authorization",
		"Cookie",
		"X-Test-Signature",
		"X-Upstream-Token",
		"X-Client-Secret",
		"X-Foo-Hmac-Sha256",
		"X-Api-Credential",
		"X-Session-Id",
		"Proxy-Authorization",
		"X-Request-Id",
	})
	if got.Get("Content-Type") != "application/json" || got.Get("X-Request-Id") != "req-1" {
		t.Fatalf("safe declared headers = %#v", got)
	}
	for _, name := range []string{"Authorization", "Cookie", "X-Test-Signature", "X-Upstream-Token", "X-Client-Secret", "X-Foo-Hmac-Sha256", "X-Api-Credential", "X-Session-Id", "Proxy-Authorization"} {
		if got.Get(name) != "" {
			t.Fatalf("sensitive header %s leaked into allowlisted headers: %#v", name, got)
		}
	}
}

func TestAllowlistedHeadersStoresNoneWithoutPackAllowlist(t *testing.T) {
	headers := http.Header{
		"Content-Type":     []string{"application/json"},
		"User-Agent":       []string{"signature-agent"},
		"X-Correlation-Id": []string{"corr-1"},
		"X-Event-Id":       []string{"signature-event"},
		"X-Request-Id":     []string{"signature-request"},
	}
	got := AllowlistedHeaders(headers, nil)
	if len(got) != 0 {
		t.Fatalf("headers persisted without pack allowlist: %#v", got)
	}
}

func TestClaimNextUsesCASAndDecryptsReceipt(t *testing.T) {
	db, mock := newReceiptSQLMock(t)
	defer db.Close()

	now := receiptTestNow()
	lease := now.Add(5 * time.Minute)
	cipher := receiptTestCipher(t)
	in := receiptTestInput()
	targetHash := RequestTargetHash(http.MethodPost, in.RequestURI, in.RequestURL)
	targetCiphertext := encryptReceiptTarget(t, cipher, in.RequestURI, in.RequestURL)
	bodyCiphertext := encryptReceiptBody(t, cipher, in.RawBody)
	headerCiphertext := encryptReceiptHeaders(t, cipher, AllowlistedHeaders(in.Headers, nil, in.BlockedHeaders))
	receiptKey := "cingr:id:receipt-1"

	mock.ExpectQuery(regexp.QuoteMeta("SELECT Id")).
		WithArgs(StatusPending, StatusRetry, StatusPendingConfig, now, now, StatusProcessing, now).
		WillReturnRows(sqlmock.NewRows([]string{"Id"}).AddRow("receipt-1"))
	mock.ExpectExec(regexp.QuoteMeta("UPDATE _System_ConnectorIngressReceipt")).
		WithArgs(StatusProcessing, "worker-1", "claim-1", now, lease, now, "receipt-1", StatusPending, StatusRetry, StatusPendingConfig, now, now, StatusProcessing, now).
		WillReturnResult(sqlmock.NewResult(0, 1))
	mock.ExpectQuery(regexp.QuoteMeta("FROM _System_ConnectorIngressReceipt")).
		WithArgs("receipt-1").
		WillReturnRows(receiptRows().AddRow(
			"receipt-1", "provider_ns", "install-1", "v1", "default", "hmac-sha256",
			"signatureVerifier", "envelopeParser", http.MethodPost, targetHash, targetCiphertext,
			receiptKey, BodyHash(in.RawBody), len(in.RawBody), bodyCiphertext, headerCiphertext, StatusProcessing, 0,
			nil, "worker-1", "claim-1", now, lease, `[]`, `{}`, "", nil, nil, now, now,
		))

	receipt, ok, err := NewService(db, cipher, nil, func() time.Time { return now }).ClaimNext(context.Background(), "worker-1", "claim-1", lease)
	if err != nil {
		t.Fatalf("ClaimNext: %v", err)
	}
	if !ok {
		t.Fatal("ok = false, want true")
	}
	if got := string(receipt.RawBody); got != string(in.RawBody) {
		t.Fatalf("raw body = %q", got)
	}
	if got := receipt.Headers.Get("Authorization"); got != "" {
		t.Fatalf("authorization leaked after claim: %q", got)
	}
	if receipt.RequestURI != in.RequestURI || receipt.RequestURL != in.RequestURL {
		t.Fatalf("decrypted request target = %q/%q", receipt.RequestURI, receipt.RequestURL)
	}
	requireReceiptSQLExpectations(t, mock)
}

func TestScopedClaimServiceSelectsOnlyAuthorizedPackInstallations(t *testing.T) {
	db, mock := newReceiptSQLMock(t)
	defer db.Close()
	now := receiptTestNow()
	service, err := NewScopedClaimService(db, receiptTestCipher(t), []string{"install-b", " install-a "})
	if err != nil {
		t.Fatal(err)
	}
	mock.ExpectQuery(`SELECT Id[\s\S]*InstalledPackId IN \(\?, \?\)`).
		WithArgs(StatusPending, StatusRetry, StatusPendingConfig, now, now, StatusProcessing, now, "install-a", "install-b").
		WillReturnRows(sqlmock.NewRows([]string{"Id"}))
	if _, err := service.selectClaimableReceiptID(context.Background(), now, true); !errors.Is(err, sql.ErrNoRows) {
		t.Fatalf("selectClaimableReceiptID error = %v, want sql.ErrNoRows", err)
	}
	requireReceiptSQLExpectations(t, mock)
}

func TestScopedClaimServiceEmptySelectionClaimsNothing(t *testing.T) {
	db, mock := newReceiptSQLMock(t)
	defer db.Close()
	now := receiptTestNow()
	service, err := NewScopedClaimService(db, receiptTestCipher(t), nil)
	if err != nil {
		t.Fatal(err)
	}
	mock.ExpectQuery(`SELECT Id[\s\S]*AND 1 = 0`).
		WithArgs(StatusPending, StatusRetry, StatusPendingConfig, now, now, StatusProcessing, now).
		WillReturnRows(sqlmock.NewRows([]string{"Id"}))
	if _, err := service.selectClaimableReceiptID(context.Background(), now, true); !errors.Is(err, sql.ErrNoRows) {
		t.Fatalf("selectClaimableReceiptID error = %v, want sql.ErrNoRows", err)
	}
	requireReceiptSQLExpectations(t, mock)
}

func TestScopedClaimServiceRechecksScopeInSingleLeaseMutation(t *testing.T) {
	db, mock := newReceiptSQLMock(t)
	defer db.Close()
	now := receiptTestNow()
	lease := now.Add(5 * time.Minute)
	service, err := NewScopedClaimService(db, receiptTestCipher(t), []string{"install-a"})
	if err != nil {
		t.Fatal(err)
	}
	mock.ExpectExec(`UPDATE _System_ConnectorIngressReceipt[\s\S]*InstalledPackId IN \(\?\)`).
		WithArgs(
			StatusProcessing, "worker-a", "claim-a", now, lease, now, "receipt-a",
			StatusPending, StatusRetry, StatusPendingConfig, now, now, StatusProcessing, now,
			"install-a",
		).
		WillReturnResult(sqlmock.NewResult(0, 1))
	if _, err := service.claimSelectedReceipt(t.Context(), "receipt-a", "worker-a", "claim-a", lease, now, true); err != nil {
		t.Fatal(err)
	}
	requireReceiptSQLExpectations(t, mock)
}

func TestScopedClaimServiceRechecksEmptyScopeInBatchLeaseMutation(t *testing.T) {
	db, mock := newReceiptSQLMock(t)
	defer db.Close()
	now := receiptTestNow()
	lease := now.Add(5 * time.Minute)
	service, err := NewScopedClaimService(db, receiptTestCipher(t), nil)
	if err != nil {
		t.Fatal(err)
	}
	mock.ExpectExec(`UPDATE _System_ConnectorIngressReceipt[\s\S]*AND 1 = 0`).
		WithArgs(
			StatusProcessing, "worker-a", "claim-a", now, lease, now, "receipt-a", "receipt-b",
			StatusPending, StatusRetry, now, now, StatusProcessing, now,
		).
		WillReturnResult(sqlmock.NewResult(0, 0))
	if _, err := service.claimSelectedReceipts(t.Context(), []string{"receipt-a", "receipt-b"}, "worker-a", "claim-a", lease, now, false); err != nil {
		t.Fatal(err)
	}
	requireReceiptSQLExpectations(t, mock)
}

func TestScopedClaimServiceKeepsScopeOnReturnedWork(t *testing.T) {
	db, mock := newReceiptSQLMock(t)
	defer db.Close()
	service, err := NewScopedClaimService(db, receiptTestCipher(t), []string{"install-a"})
	if err != nil {
		t.Fatal(err)
	}
	mock.ExpectQuery(`SELECT Id, Namespace, InstalledPackId[\s\S]*WHERE Id = \?[\s\S]*InstalledPackId IN \(\?\)`).
		WithArgs("receipt-a", "install-a").
		WillReturnRows(receiptRows())
	if _, err := service.findByID(t.Context(), "receipt-a"); !errors.Is(err, ErrReceiptNotFound) {
		t.Fatalf("findByID error = %v, want ErrReceiptNotFound", err)
	}
	mock.ExpectQuery(`SELECT Id, Namespace, InstalledPackId[\s\S]*WHERE ClaimToken = \?[\s\S]*InstalledPackId IN \(\?\)`).
		WithArgs("claim-a", "receipt-a", "receipt-b", "install-a").
		WillReturnRows(receiptRows())
	receipts, err := service.findByIDsWithClaim(t.Context(), []string{"receipt-a", "receipt-b"}, "claim-a")
	if err != nil || len(receipts) != 0 {
		t.Fatalf("findByIDsWithClaim = (%#v, %v)", receipts, err)
	}
	requireReceiptSQLExpectations(t, mock)
}

func TestScopedClaimServiceRejectsAmbiguousSelection(t *testing.T) {
	for _, values := range [][]string{{""}, {"install-a", "install-a"}} {
		if _, err := NewScopedClaimService(nil, nil, values); err == nil {
			t.Fatalf("NewScopedClaimService(%#v) accepted ambiguous selection", values)
		}
	}
}

func TestClaimNextReclaimsExpiredProcessingLease(t *testing.T) {
	db, mock := newReceiptSQLMock(t)
	defer db.Close()

	now := receiptTestNow()
	lease := now.Add(5 * time.Minute)
	cipher := receiptTestCipher(t)
	in := receiptTestInput()
	targetHash := RequestTargetHash(http.MethodPost, in.RequestURI, in.RequestURL)
	targetCiphertext := encryptReceiptTarget(t, cipher, in.RequestURI, in.RequestURL)
	bodyCiphertext := encryptReceiptBody(t, cipher, in.RawBody)
	headerCiphertext := encryptReceiptHeaders(t, cipher, AllowlistedHeaders(in.Headers, nil, in.BlockedHeaders))

	mock.ExpectQuery(`(?s)SELECT Id.*OR \(Status = \? AND LeaseExpiresAt <= \?\).*ORDER BY CreatedDate ASC, Id ASC`).
		WithArgs(StatusPending, StatusRetry, StatusPendingConfig, now, now, StatusProcessing, now).
		WillReturnRows(sqlmock.NewRows([]string{"Id"}).AddRow("receipt-1"))
	mock.ExpectExec(`(?s)UPDATE _System_ConnectorIngressReceipt.*OR \(Status = \? AND LeaseExpiresAt <= \?\)`).
		WithArgs(StatusProcessing, "worker-2", "claim-2", now, lease, now, "receipt-1", StatusPending, StatusRetry, StatusPendingConfig, now, now, StatusProcessing, now).
		WillReturnResult(sqlmock.NewResult(0, 1))
	mock.ExpectQuery(regexp.QuoteMeta("FROM _System_ConnectorIngressReceipt")).
		WithArgs("receipt-1").
		WillReturnRows(receiptRows().AddRow(
			"receipt-1", "provider_ns", "install-1", "v1", "default", "hmac-sha256",
			"signatureVerifier", "envelopeParser", http.MethodPost, targetHash, targetCiphertext,
			"cingr:id:receipt-1", BodyHash(in.RawBody), len(in.RawBody), bodyCiphertext, headerCiphertext, StatusProcessing, 0,
			nil, "worker-2", "claim-2", now, lease, `[]`, `{}`, "", nil, nil, now, now,
		))

	receipt, ok, err := NewService(db, cipher, nil, func() time.Time { return now }).ClaimNext(context.Background(), "worker-2", "claim-2", lease)
	if err != nil {
		t.Fatalf("ClaimNext: %v", err)
	}
	if !ok || receipt == nil || receipt.LeaseExpiresAt == nil || !receipt.LeaseExpiresAt.Equal(lease) {
		t.Fatalf("ClaimNext stale processing = (%#v, %v), want renewed lease %s", receipt, ok, lease)
	}
	requireReceiptSQLExpectations(t, mock)
}

func TestClaimBatchDoesNotRepeatShortPendingConfigPage(t *testing.T) {
	db, mock := newReceiptSQLMock(t)
	defer db.Close()

	now := receiptTestNow()
	mock.ExpectQuery(`(?s)SELECT Id.*Status = \?.*ORDER BY CreatedDate ASC, Id ASC.*LIMIT \?`).
		WithArgs(StatusPendingConfig, now, now, 5).
		WillReturnRows(sqlmock.NewRows([]string{"Id"}))
	mock.ExpectQuery(`(?s)SELECT Id.*Status IN.*ORDER BY CreatedDate ASC, Id ASC.*LIMIT \?`).
		WithArgs(StatusPending, StatusRetry, now, now, StatusProcessing, now, 50).
		WillReturnRows(sqlmock.NewRows([]string{"Id"}))

	receipts, claimed, err := newReceiptTestService(db, now).ClaimBatch(
		context.Background(), "worker-1", "claim-1", now.Add(time.Minute), 50,
	)
	if err != nil {
		t.Fatalf("ClaimBatch: %v", err)
	}
	if claimed || len(receipts) != 0 {
		t.Fatalf("ClaimBatch = (%#v, %v), want empty", receipts, claimed)
	}
	requireReceiptSQLExpectations(t, mock)
}

func TestClaimNextLostCASReturnsNoReceipt(t *testing.T) {
	db, mock := newReceiptSQLMock(t)
	defer db.Close()

	now := receiptTestNow()
	lease := now.Add(5 * time.Minute)
	mock.ExpectQuery(regexp.QuoteMeta("SELECT Id")).
		WithArgs(StatusPending, StatusRetry, StatusPendingConfig, now, now, StatusProcessing, now).
		WillReturnRows(sqlmock.NewRows([]string{"Id"}).AddRow("receipt-1"))
	mock.ExpectExec(regexp.QuoteMeta("UPDATE _System_ConnectorIngressReceipt")).
		WithArgs(StatusProcessing, "worker-1", "claim-1", now, lease, now, "receipt-1", StatusPending, StatusRetry, StatusPendingConfig, now, now, StatusProcessing, now).
		WillReturnResult(sqlmock.NewResult(0, 0))

	receipt, ok, err := newReceiptTestService(db, now).ClaimNext(context.Background(), "worker-1", "claim-1", lease)
	if err != nil {
		t.Fatalf("ClaimNext: %v", err)
	}
	if ok || receipt != nil {
		t.Fatalf("ClaimNext = (%#v, %v), want nil/false on lost CAS", receipt, ok)
	}
	requireReceiptSQLExpectations(t, mock)
}

func TestMarkRetryStoresNextAttemptAndClearsLease(t *testing.T) {
	db, mock := newReceiptSQLMock(t)
	defer db.Close()

	now := receiptTestNow()
	next := now.Add(time.Minute)
	mock.ExpectExec(regexp.QuoteMeta("UPDATE _System_ConnectorIngressReceipt")).
		WithArgs(StatusRetry, next, "parser offline", now, now, "receipt-1", StatusProcessing, "claim-1").
		WillReturnResult(sqlmock.NewResult(0, 1))

	if err := newReceiptTestService(db, now).MarkRetry(context.Background(), "receipt-1", "claim-1", next, errors.New("parser offline")); err != nil {
		t.Fatalf("MarkRetry: %v", err)
	}
	requireReceiptSQLExpectations(t, mock)
}

func TestMarkPendingConfigStoresNextAttemptAndClearsLease(t *testing.T) {
	db, mock := newReceiptSQLMock(t)
	defer db.Close()

	now := receiptTestNow()
	next := now.Add(2 * time.Minute)
	mock.ExpectExec(regexp.QuoteMeta("UPDATE _System_ConnectorIngressReceipt")).
		WithArgs(StatusPendingConfig, next, "route not declared", now, now, "receipt-1", StatusProcessing, "claim-1").
		WillReturnResult(sqlmock.NewResult(0, 1))

	if err := newReceiptTestService(db, now).MarkPendingConfig(context.Background(), "receipt-1", "claim-1", next, errors.New("route not declared")); err != nil {
		t.Fatalf("MarkPendingConfig: %v", err)
	}
	requireReceiptSQLExpectations(t, mock)
}

func TestMarkMaterializingStoresMetadataUnderClaim(t *testing.T) {
	db, mock := newReceiptSQLMock(t)
	defer db.Close()

	now := receiptTestNow()
	mock.ExpectExec(regexp.QuoteMeta("UPDATE _System_ConnectorIngressReceipt")).
		WithArgs(`{"deliveryId":"dlv-1","materializedPackId":"install-1"}`, now, "receipt-1", StatusProcessing, "claim-1").
		WillReturnResult(sqlmock.NewResult(0, 1))

	err := newReceiptTestService(db, now).MarkMaterializing(context.Background(), "receipt-1", "claim-1", map[string]string{
		"deliveryId":         "dlv-1",
		"materializedPackId": "install-1",
	})
	if err != nil {
		t.Fatalf("MarkMaterializing: %v", err)
	}
	requireReceiptSQLExpectations(t, mock)
}

func TestMarkTerminalStatesRequireClaim(t *testing.T) {
	for _, tt := range []struct {
		name   string
		status string
		call   func(*Service) error
	}{
		{
			name:   "enqueued",
			status: StatusEnqueued,
			call: func(s *Service) error {
				return s.MarkEnqueued(context.Background(), "receipt-1", "claim-1", []string{"evt-1"}, map[string]string{"router": "pr6"})
			},
		},
		{
			name:   "dead",
			status: StatusDead,
			call: func(s *Service) error {
				return s.MarkDead(context.Background(), "receipt-1", "claim-1", errors.New("exhausted"))
			},
		},
		{
			name:   "config terminal",
			status: StatusConfigTerminal,
			call: func(s *Service) error {
				return s.MarkConfigTerminal(context.Background(), "receipt-1", "claim-1", errors.New("no route"))
			},
		},
	} {
		t.Run(tt.name, func(t *testing.T) {
			db, mock := newReceiptSQLMock(t)
			defer db.Close()
			now := receiptTestNow()
			mock.ExpectExec(regexp.QuoteMeta("UPDATE _System_ConnectorIngressReceipt")).
				WithArgs(tt.status, sqlmock.AnyArg(), sqlmock.AnyArg(), sqlmock.AnyArg(), sqlmock.AnyArg(), now, "receipt-1", StatusProcessing, "claim-1").
				WillReturnResult(sqlmock.NewResult(0, 1))
			if err := tt.call(newReceiptTestService(db, now)); err != nil {
				t.Fatalf("%s: %v", tt.name, err)
			}
			requireReceiptSQLExpectations(t, mock)
		})
	}
}

func TestListMetadataRedactsRawReceiptFields(t *testing.T) {
	db, mock := newReceiptSQLMock(t)
	defer db.Close()

	now := receiptTestNow()
	lastErrorAt := now.Add(time.Minute)
	purgeAfter := now.Add(time.Hour)
	mock.ExpectQuery(regexp.QuoteMeta("FROM _System_ConnectorIngressReceipt")).
		WithArgs("provider_ns", StatusDead, 10, 5).
		WillReturnRows(receiptMetadataRows().AddRow(
			"receipt-1", "provider_ns", "install-1", "v1", "default", "hmac-sha256",
			"signatureVerifier", "envelopeParser", http.MethodPost, "target-hash",
			"body-hash", 36, StatusDead, 3,
			nil, "", nil, nil, `["evt-1"]`, "raw secret body token", lastErrorAt,
			purgeAfter, now, now,
		))

	rows, err := newReceiptTestService(db, now).ListMetadata(context.Background(), allowReceiptPerm(string(constants.PermViewConnectorEvents)), ReceiptMetadataFilter{
		Namespace: "Provider_NS",
		Status:    StatusDead,
		Page:      Page{Limit: 10, Offset: 5},
	})
	if err != nil {
		t.Fatalf("ListMetadata: %v", err)
	}
	if len(rows) != 1 {
		t.Fatalf("rows = %d, want 1", len(rows))
	}
	row := rows[0]
	if row.BodyHash != "body-hash" || row.BodySizeBytes != 36 || row.LastErrorClass != "terminal_error" {
		t.Fatalf("metadata projection = %#v", row)
	}
	if len(row.EnqueuedEventIDs) != 1 || row.EnqueuedEventIDs[0] != "evt-1" {
		t.Fatalf("enqueued event ids = %#v", row.EnqueuedEventIDs)
	}
	payload, err := json.Marshal(row)
	if err != nil {
		t.Fatalf("marshal metadata: %v", err)
	}
	for _, secret := range []string{"raw secret body token", "BodyCiphertext", "AllowedHeadersCiphertext", "RawBody", "Headers"} {
		if strings.Contains(string(payload), secret) {
			t.Fatalf("metadata leaked %q: %s", secret, payload)
		}
	}
	requireReceiptSQLExpectations(t, mock)
}

func TestGetMetadataPermissionDeniedBeforeSQL(t *testing.T) {
	db, mock := newReceiptSQLMock(t)
	defer db.Close()

	_, err := newReceiptTestService(db, receiptTestNow()).GetMetadata(context.Background(), allowReceiptPerm(), "receipt-1")
	if !errors.Is(err, ErrPermissionDenied) {
		t.Fatalf("GetMetadata err = %v, want ErrPermissionDenied", err)
	}
	requireReceiptSQLExpectations(t, mock)
}

func TestPurgeTerminalDeletesOnlyTTLTerminalRows(t *testing.T) {
	db, mock := newReceiptSQLMock(t)
	defer db.Close()

	now := receiptTestNow()
	before := now.Add(24 * time.Hour)
	mock.ExpectExec(`(?s)DELETE FROM _System_ConnectorIngressReceipt.*Status IN.*PurgeAfter IS NOT NULL.*PurgeAfter <=.*ORDER BY PurgeAfter ASC, Id ASC.*LIMIT`).
		WithArgs(StatusEnqueued, StatusDead, StatusConfigTerminal, before, 100).
		WillReturnResult(sqlmock.NewResult(0, 2))

	result, err := newReceiptTestService(db, now).PurgeTerminal(context.Background(), allowReceiptPerm(string(constants.PermReplayConnectorEvents)), PurgeRequest{Before: before})
	if err != nil {
		t.Fatalf("PurgeTerminal: %v", err)
	}
	if result.Deleted != 2 || !result.Before.Equal(before) {
		t.Fatalf("purge result = %#v, want deleted=2 before=%s", result, before)
	}
	requireReceiptSQLExpectations(t, mock)
}

func TestReceiptTransitionLogOmitsRawSecret(t *testing.T) {
	db, mock := newReceiptSQLMock(t)
	defer db.Close()

	now := receiptTestNow()
	next := now.Add(time.Minute)
	mock.ExpectExec(regexp.QuoteMeta("UPDATE _System_ConnectorIngressReceipt")).
		WithArgs(StatusRetry, next, "upstream token=raw-secret", now, now, "receipt-1", StatusProcessing, "claim-1").
		WillReturnResult(sqlmock.NewResult(0, 1))

	var buf bytes.Buffer
	ctx := zerolog.New(&buf).WithContext(context.Background())
	err := newReceiptTestService(db, now).MarkRetry(ctx, "receipt-1", "claim-1", next, errors.New("upstream token=raw-secret"))
	if err != nil {
		t.Fatalf("MarkRetry: %v", err)
	}
	logLine := buf.String()
	if !strings.Contains(logLine, `"operation":"retry"`) || !strings.Contains(logLine, `"receipt_id":"receipt-1"`) {
		t.Fatalf("missing retry audit fields in log: %s", logLine)
	}
	if strings.Contains(logLine, "raw-secret") || strings.Contains(logLine, "upstream token") {
		t.Fatalf("transition log leaked raw error secret: %s", logLine)
	}
	requireReceiptSQLExpectations(t, mock)
}

func receiptTestInput() VerifiedReceiptInput {
	return VerifiedReceiptInput{
		ID:                     "receipt-1",
		Namespace:              "Provider_NS",
		InstalledPackID:        "install-1",
		PackVersion:            "v1",
		SecretName:             "Default",
		SecretAlgorithm:        "hmac-sha256",
		VerifierCapabilityKind: "signatureVerifier",
		ParserCapabilityKind:   "envelopeParser",
		RequestMethod:          http.MethodPost,
		RequestURI:             "/webhooks/provider_ns/default?token=query-secret",
		RequestURL:             "https://example.test/webhooks/provider_ns/default?token=query-secret",
		DeliveryIdentity:       "delivery-1",
		RawBody:                []byte(`{"secret":"body-token","safe":true}`),
		Headers: http.Header{
			"Content-Type":     []string{"application/json"},
			"Authorization":    []string{"Bearer raw-token"},
			"X-Event-Id":       []string{"signed-event-id"},
			"X-Test-Signature": []string{"sig-secret"},
		},
		HeaderAllowlist: []string{"Content-Type"},
		BlockedHeaders:  []string{"X-Event-Id"},
	}
}

func receiptTestNow() time.Time {
	return time.Date(2026, 6, 19, 2, 0, 0, 0, time.UTC)
}

func newReceiptSQLMock(t *testing.T) (*sql.DB, sqlmock.Sqlmock) {
	t.Helper()
	db, mock, err := sqlmock.New()
	if err != nil {
		t.Fatalf("sqlmock.New: %v", err)
	}
	return db, mock
}

func newReceiptTestService(db *sql.DB, now time.Time) *Service {
	return NewService(db, receiptTestCipherNoT(), func() string { return "receipt-1" }, func() time.Time { return now })
}

func receiptTestCipher(t *testing.T) *namedcredential.SecretCipher {
	t.Helper()
	c, err := namedcredential.NewSecretCipherFromKey([]byte("0123456789abcdef0123456789abcdef"))
	if err != nil {
		t.Fatalf("NewSecretCipherFromKey: %v", err)
	}
	return c
}

func receiptTestCipherNoT() *namedcredential.SecretCipher {
	c, err := namedcredential.NewSecretCipherFromKey([]byte("0123456789abcdef0123456789abcdef"))
	if err != nil {
		panic(err)
	}
	return c
}

func encryptReceiptBody(t *testing.T, cipher *namedcredential.SecretCipher, body []byte) string {
	t.Helper()
	encoded := base64.StdEncoding.EncodeToString(body)
	ciphertext, err := cipher.Encrypt(encoded)
	if err != nil {
		t.Fatalf("Encrypt body: %v", err)
	}
	return ciphertext
}

func encryptReceiptTarget(t *testing.T, cipher *namedcredential.SecretCipher, requestURI, requestURL string) string {
	t.Helper()
	payload, err := json.Marshal(map[string]string{
		"requestURI": requestURI,
		"requestURL": requestURL,
	})
	if err != nil {
		t.Fatalf("Marshal target: %v", err)
	}
	ciphertext, err := cipher.Encrypt(string(payload))
	if err != nil {
		t.Fatalf("Encrypt target: %v", err)
	}
	return ciphertext
}

func encryptReceiptHeaders(t *testing.T, cipher *namedcredential.SecretCipher, headers http.Header) string {
	t.Helper()
	payload, err := json.Marshal(headers)
	if err != nil {
		t.Fatalf("Marshal headers: %v", err)
	}
	ciphertext, err := cipher.Encrypt(string(payload))
	if err != nil {
		t.Fatalf("Encrypt headers: %v", err)
	}
	return ciphertext
}

func receiptRows() *sqlmock.Rows {
	return sqlmock.NewRows([]string{
		"Id", "Namespace", "InstalledPackId", "COALESCE(PackVersion, '')", "SecretName", "SecretAlgorithm",
		"VerifierCapabilityKind", "ParserCapabilityKind", "RequestMethod", "RequestTargetHash", "RequestTargetCiphertext",
		"ReceiptKey", "BodyHash", "COALESCE(BodySizeBytes, 0)", "BodyCiphertext", "AllowedHeadersCiphertext", "Status", "AttemptCount",
		"NextAttemptAt", "COALESCE(ClaimedBy, '')", "COALESCE(ClaimToken, '')", "ClaimedAt", "LeaseExpiresAt",
		"COALESCE(EnqueuedEventIds, '[]')", "COALESCE(EnqueueMetadata, '{}')", "COALESCE(LastError, '')", "LastErrorAt",
		"PurgeAfter", "CreatedDate", "LastModifiedDate",
	})
}

func receiptMetadataRows() *sqlmock.Rows {
	return sqlmock.NewRows([]string{
		"Id", "Namespace", "InstalledPackId", "COALESCE(PackVersion, '')", "SecretName", "SecretAlgorithm",
		"VerifierCapabilityKind", "ParserCapabilityKind", "RequestMethod", "RequestTargetHash",
		"BodyHash", "COALESCE(BodySizeBytes, 0)", "Status", "AttemptCount",
		"NextAttemptAt", "COALESCE(ClaimedBy, '')", "ClaimedAt", "LeaseExpiresAt",
		"COALESCE(EnqueuedEventIds, '[]')", "COALESCE(LastError, '')", "LastErrorAt",
		"PurgeAfter", "CreatedDate", "LastModifiedDate",
	})
}

type receiptPerms map[string]bool

func allowReceiptPerm(keys ...string) receiptPerms {
	out := make(receiptPerms, len(keys))
	for _, key := range keys {
		out[key] = true
	}
	return out
}

func (p receiptPerms) Has(key string) bool {
	return p[key]
}

func requireReceiptSQLExpectations(t *testing.T, mock sqlmock.Sqlmock) {
	t.Helper()
	if err := mock.ExpectationsWereMet(); err != nil {
		t.Fatalf("unmet sql expectations: %v", err)
	}
}

type notContainingArg struct {
	fragments []string
}

func (a notContainingArg) Match(v driver.Value) bool {
	s, ok := v.(string)
	if !ok || strings.TrimSpace(s) == "" {
		return false
	}
	for _, fragment := range a.fragments {
		if strings.Contains(s, fragment) {
			return false
		}
	}
	return true
}