package connector_ingress_receipts

import (
	"bytes"
	"context"
	"crypto/sha256"
	"database/sql"
	"encoding/base64"
	"encoding/hex"
	"encoding/json"
	"errors"
	"fmt"
	"net/http"
	"sort"
	"strings"
	"time"

	"github.com/pingcap-inc/nexuscrm-connectors/features/namedcredential"
	"github.com/pingcap-inc/nexuscrm-connectors/internal/id"
	"github.com/pingcap-inc/nexuscrm-contract/pkg/constants"
	"github.com/rs/zerolog/log"
)

const (
	StatusPending        = "pending"
	StatusProcessing     = "processing"
	StatusRetry          = "retry"
	StatusPendingConfig  = "pending_config"
	StatusEnqueued       = "enqueued"
	StatusDead           = "dead"
	StatusConfigTerminal = "config_terminal"

	DefaultReceiptRetention = 30 * 24 * time.Hour
)

var (
	ErrCipherRequired     = errors.New("connector ingress receipts: cipher required")
	ErrReceiptNotFound    = errors.New("connector ingress receipt not found")
	ErrLeaseLost          = errors.New("connector ingress receipt lease no longer held")
	ErrReceiptTerminal    = errors.New("connector ingress receipt already terminal")
	ErrInvalidReceipt     = errors.New("connector ingress receipt invalid")
	ErrReceiptKeyConflict = errors.New("connector ingress receipt key conflict")
	ErrPermissionDenied   = errors.New("connector ingress receipt permission denied")
	ErrReceiptGuardFailed = errors.New("connector ingress receipt guard failed")
)

type Service struct {
	db                  *sql.DB
	cipher              *namedcredential.SecretCipher
	idgen               func() string
	now                 func() time.Time
	claimScopeEnabled   bool
	claimPackInstallIDs []string
}

type PermissionChecker interface {
	Has(key string) bool
}

type Page struct {
	Limit  int
	Offset int
}

type ReceiptMetadataFilter struct {
	Namespace string
	Status    string
	Page      Page
}

type ReceiptMetadata struct {
	ID                     string
	Namespace              string
	InstalledPackID        string
	PackVersion            string
	SecretName             string
	SecretAlgorithm        string
	VerifierCapabilityKind string
	ParserCapabilityKind   string
	RequestMethod          string
	RequestTargetHash      string
	BodyHash               string
	BodySizeBytes          int64
	Status                 string
	AttemptCount           int
	NextAttemptAt          *time.Time
	ClaimedBy              string
	ClaimedAt              *time.Time
	LeaseExpiresAt         *time.Time
	EnqueuedEventIDs       []string
	LastErrorClass         string
	LastErrorAt            *time.Time
	PurgeAfter             *time.Time
	CreatedDate            time.Time
	LastModifiedDate       time.Time
}

type PurgeRequest struct {
	Before    time.Time
	Namespace string
	Limit     int
}

type PurgeResult struct {
	Deleted int64
	Before  time.Time
}

type VerifiedReceiptInput struct {
	ID                     string
	Namespace              string
	InstalledPackID        string
	PackVersion            string
	PackPayloadHash        string
	SecretName             string
	SecretAlgorithm        string
	SecretValueHash        string
	VerifierCapabilityKind string
	ParserCapabilityKind   string
	RequestMethod          string
	RequestURI             string
	RequestURL             string
	DeliveryIdentity       string
	ReceiptKey             string
	RawBody                []byte
	Headers                http.Header
	HeaderAllowlist        []string
	BlockedHeaders         []string
	PurgeAfter             time.Time
}

type Receipt struct {
	ID                     string
	Namespace              string
	InstalledPackID        string
	PackVersion            string
	SecretName             string
	SecretAlgorithm        string
	VerifierCapabilityKind string
	ParserCapabilityKind   string
	RequestMethod          string
	RequestURI             string
	RequestURL             string
	RequestTargetHash      string
	ReceiptKey             string
	BodyHash               string
	BodySizeBytes          int64
	Headers                http.Header
	RawBody                []byte
	Status                 string
	AttemptCount           int
	NextAttemptAt          *time.Time
	ClaimedBy              string
	ClaimToken             string
	ClaimedAt              *time.Time
	LeaseExpiresAt         *time.Time
	EnqueuedEventIDs       []string
	EnqueueMetadata        map[string]string
	LastError              string
	LastErrorAt            *time.Time
	PurgeAfter             *time.Time
	CreatedDate            time.Time
	LastModifiedDate       time.Time
}

type EnqueuedReceiptUpdate struct {
	ReceiptID  string
	ClaimToken string
	EventIDs   []string
	Metadata   map[string]string
}

type MaterializingReceiptUpdate struct {
	ReceiptID  string
	ClaimToken string
	Metadata   map[string]string
}

func NewService(db *sql.DB, cipher *namedcredential.SecretCipher, idgen func() string, now func() time.Time) *Service {
	if idgen == nil {
		idgen = id.Generate
	}
	if now == nil {
		now = time.Now
	}
	return &Service{db: db, cipher: cipher, idgen: idgen, now: now}
}

// NewScopedClaimService creates a receipt store whose claim path can acquire
// only the caller-selected pack installations. An empty selection deliberately
// claims nothing. Receipt creation and all existing unscoped service behavior
// remain unchanged.
func NewScopedClaimService(db *sql.DB, cipher *namedcredential.SecretCipher, packInstallIDs []string) (*Service, error) {
	service := NewService(db, cipher, nil, nil)
	seen := make(map[string]struct{}, len(packInstallIDs))
	for _, raw := range packInstallIDs {
		packInstallID := strings.TrimSpace(raw)
		if packInstallID == "" {
			return nil, fmt.Errorf("connector ingress receipts: claim pack installation ID is required")
		}
		if _, duplicate := seen[packInstallID]; duplicate {
			return nil, fmt.Errorf("connector ingress receipts: duplicate claim pack installation ID %q", packInstallID)
		}
		seen[packInstallID] = struct{}{}
		service.claimPackInstallIDs = append(service.claimPackInstallIDs, packInstallID)
	}
	sort.Strings(service.claimPackInstallIDs)
	service.claimScopeEnabled = true
	return service, nil
}

// VerifyReady checks the same stored shape used to read claimed receipts,
// without claiming, decrypting or materializing any business input.
func (s *Service) VerifyReady(ctx context.Context) error {
	if s == nil || s.db == nil || ctx == nil {
		return fmt.Errorf("connector ingress receipt store is not wired")
	}
	if s.cipher == nil {
		return ErrCipherRequired
	}
	rows, err := s.db.QueryContext(ctx, connectorIngressReceiptSelectSQL+" WHERE 1 = 0")
	if err != nil {
		return err
	}
	defer rows.Close()
	return rows.Err()
}

func (s *Service) CreateVerifiedReceipt(ctx context.Context, in VerifiedReceiptInput) (*Receipt, bool, error) {
	if s == nil || s.db == nil {
		return nil, false, fmt.Errorf("connector ingress receipt store is not wired")
	}
	if s.cipher == nil {
		return nil, false, ErrCipherRequired
	}
	in = normalizeVerifiedReceiptInput(in)
	if err := validateVerifiedReceiptInput(in); err != nil {
		return nil, false, err
	}
	now := s.now()
	if in.ID == "" {
		in.ID = orderedReceiptID(now, s.idgen())
	}
	if in.ReceiptKey == "" {
		in.ReceiptKey = ReceiptKeyFor(in.Namespace, in.InstalledPackID, in.SecretName, in.DeliveryIdentity)
	}
	bodyHash := BodyHash(in.RawBody)
	requestTargetHash := RequestTargetHash(in.RequestMethod, in.RequestURI, in.RequestURL)

	targetJSON, err := json.Marshal(map[string]string{
		"requestURI": in.RequestURI,
		"requestURL": in.RequestURL,
	})
	if err != nil {
		return nil, false, err
	}
	headerJSON, err := json.Marshal(AllowlistedHeaders(in.Headers, in.HeaderAllowlist, in.BlockedHeaders))
	if err != nil {
		return nil, false, err
	}
	targetCiphertext, err := s.cipher.Encrypt(string(targetJSON))
	if err != nil {
		return nil, false, err
	}
	bodyCiphertext, err := s.cipher.Encrypt(base64.StdEncoding.EncodeToString(in.RawBody))
	if err != nil {
		return nil, false, err
	}
	headerCiphertext, err := s.cipher.Encrypt(string(headerJSON))
	if err != nil {
		return nil, false, err
	}

	purgeAfter := in.PurgeAfter
	if purgeAfter.IsZero() {
		purgeAfter = DefaultPurgeAfter(now)
	}
	res, err := s.insertVerifiedReceipt(ctx, in, requestTargetHash, targetCiphertext, bodyHash, bodyCiphertext, headerCiphertext, purgeAfter, now)
	if err != nil {
		if isDuplicateKey(err) {
			existing, lookupErr := s.FindByReceiptKey(ctx, in.ReceiptKey)
			if lookupErr != nil {
				return nil, false, lookupErr
			}
			if existing == nil {
				return nil, false, ErrReceiptKeyConflict
			}
			if !sameVerifiedReceiptBody(existing, in.RawBody) {
				return nil, false, fmt.Errorf("%w: duplicate receipt key maps to a different verified body", ErrReceiptKeyConflict)
			}
			return existing, false, nil
		}
		return nil, false, err
	}
	if strings.TrimSpace(in.SecretValueHash) != "" {
		n, rowsErr := res.RowsAffected()
		if rowsErr != nil {
			return nil, false, rowsErr
		}
		if n == 0 {
			return nil, false, ErrReceiptGuardFailed
		}
	}

	receipt := &Receipt{
		ID:                     in.ID,
		Namespace:              in.Namespace,
		InstalledPackID:        in.InstalledPackID,
		PackVersion:            in.PackVersion,
		SecretName:             in.SecretName,
		SecretAlgorithm:        in.SecretAlgorithm,
		VerifierCapabilityKind: in.VerifierCapabilityKind,
		ParserCapabilityKind:   in.ParserCapabilityKind,
		RequestMethod:          in.RequestMethod,
		RequestURI:             in.RequestURI,
		RequestURL:             in.RequestURL,
		RequestTargetHash:      requestTargetHash,
		ReceiptKey:             in.ReceiptKey,
		BodyHash:               bodyHash,
		BodySizeBytes:          int64(len(in.RawBody)),
		Headers:                AllowlistedHeaders(in.Headers, in.HeaderAllowlist, in.BlockedHeaders),
		RawBody:                append([]byte(nil), in.RawBody...),
		Status:                 StatusPending,
		CreatedDate:            now,
		LastModifiedDate:       now,
	}
	receipt.PurgeAfter = &purgeAfter
	logReceiptTransition(ctx, "create", receipt.ID, receipt.Namespace, receipt.InstalledPackID, receipt.Status)
	return receipt, true, nil
}

func (s *Service) insertVerifiedReceipt(
	ctx context.Context,
	in VerifiedReceiptInput,
	requestTargetHash string,
	targetCiphertext string,
	bodyHash string,
	bodyCiphertext string,
	headerCiphertext string,
	purgeAfter time.Time,
	now time.Time,
) (sql.Result, error) {
	if strings.TrimSpace(in.SecretValueHash) == "" {
		return s.db.ExecContext(ctx, `
	INSERT INTO _System_ConnectorIngressReceipt
	  (Id, Namespace, InstalledPackId, PackVersion, SecretName, SecretAlgorithm,
	   VerifierCapabilityKind, ParserCapabilityKind, RequestMethod, RequestTargetHash, RequestTargetCiphertext,
	   ReceiptKey, BodyHash, BodySizeBytes, BodyCiphertext, AllowedHeadersCiphertext, Status, AttemptCount,
	   PurgeAfter, CreatedDate, LastModifiedDate)
	VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, 0, ?, ?, ?)`,
			in.ID, in.Namespace, in.InstalledPackID, in.PackVersion, in.SecretName, in.SecretAlgorithm,
			in.VerifierCapabilityKind, in.ParserCapabilityKind, in.RequestMethod, requestTargetHash, targetCiphertext,
			in.ReceiptKey, bodyHash, len(in.RawBody), bodyCiphertext, headerCiphertext, StatusPending,
			purgeAfter, now, now,
		)
	}

	return s.db.ExecContext(ctx, `
	INSERT INTO _System_ConnectorIngressReceipt
	  (Id, Namespace, InstalledPackId, PackVersion, SecretName, SecretAlgorithm,
	   VerifierCapabilityKind, ParserCapabilityKind, RequestMethod, RequestTargetHash, RequestTargetCiphertext,
	   ReceiptKey, BodyHash, BodySizeBytes, BodyCiphertext, AllowedHeadersCiphertext, Status, AttemptCount,
	   PurgeAfter, CreatedDate, LastModifiedDate)
	SELECT ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, 0, ?, ?, ?
	 WHERE EXISTS (
	       SELECT 1
	         FROM _System_InstalledPack p
	        WHERE p.Name = ?
	          AND p.PackInstallId = ?
	          AND p.Category = 'connector-pack'
	          AND p.Status = 'active'
	          AND COALESCE(p.IsDeleted, 0) = 0
	          AND COALESCE(p.PackInstallId, '') <> ''
	          AND p.AppliedPayloadHash IS NOT NULL
	          AND p.AppliedPayloadHash <> ''
	          AND p.DesiredPayloadHash IS NOT NULL
	          AND p.DesiredPayloadHash <> ''
	          AND p.AppliedPayloadHash = p.DesiredPayloadHash
	          AND (
	              ? = ''
	              OR p.AppliedPayloadHash = ?
	          )
	          AND EXISTS (
	              SELECT 1
	                FROM _System_WebhookSecret ws
	                JOIN _System_WebhookSecretValue wsv
	                  ON wsv.WebhookSecretId = ws.Id
	                 AND wsv.IsActive = 1
	               WHERE ws.OwnerPack = ?
	                 AND ws.Name = ?
	                 AND ws.Algorithm = ?
	                 AND COALESCE(ws.IsDeleted, 0) = 0
	                 AND LOWER(SHA2(COALESCE(wsv.SecretValue, ''), 256)) = ?
	          )
	          AND EXISTS (
	              SELECT 1
	                FROM _System_ConnectorCapabilityDeclaration d
	                JOIN _System_ConnectorInboundHandlerConfig h
	                  ON h.InstalledPackId = p.PackInstallId
	                 AND h.Namespace = d.Namespace
	                 AND h.CapabilityKind = d.Kind
	               WHERE d.InstalledPackName = p.Name
	                 AND d.Namespace = ?
	                 AND d.Kind = ?
	                 AND COALESCE(d.IsDeleted, 0) = 0
	                 AND COALESCE(h.IsDeleted, 0) = 0
	          )
	          AND EXISTS (
	              SELECT 1
	                FROM _System_ConnectorCapabilityDeclaration d
	                JOIN _System_ConnectorInboundHandlerConfig h
	                  ON h.InstalledPackId = p.PackInstallId
	                 AND h.Namespace = d.Namespace
	                 AND h.CapabilityKind = d.Kind
	               WHERE d.InstalledPackName = p.Name
	                 AND d.Namespace = ?
	                 AND d.Kind = ?
	                 AND COALESCE(d.IsDeleted, 0) = 0
	                 AND COALESCE(h.IsDeleted, 0) = 0
	          )
	 )`,
		in.ID, in.Namespace, in.InstalledPackID, in.PackVersion, in.SecretName, in.SecretAlgorithm,
		in.VerifierCapabilityKind, in.ParserCapabilityKind, in.RequestMethod, requestTargetHash, targetCiphertext,
		in.ReceiptKey, bodyHash, len(in.RawBody), bodyCiphertext, headerCiphertext, StatusPending,
		purgeAfter, now, now,
		in.Namespace, in.InstalledPackID, in.PackPayloadHash, in.PackPayloadHash,
		in.Namespace, in.SecretName, in.SecretAlgorithm, in.SecretValueHash,
		in.Namespace, in.VerifierCapabilityKind,
		in.Namespace, in.ParserCapabilityKind,
	)
}

func DefaultPurgeAfter(now time.Time) time.Time {
	return now.Add(DefaultReceiptRetention)
}

func (s *Service) FindByReceiptKey(ctx context.Context, receiptKey string) (*Receipt, error) {
	if s == nil || s.db == nil {
		return nil, fmt.Errorf("connector ingress receipt store is not wired")
	}
	if s.cipher == nil {
		return nil, ErrCipherRequired
	}
	receiptKey = strings.TrimSpace(receiptKey)
	if receiptKey == "" {
		return nil, nil
	}
	receipt, err := s.scanReceipt(s.db.QueryRowContext(ctx, connectorIngressReceiptSelectSQL+`
 WHERE ReceiptKey = ?
 LIMIT 1`, receiptKey))
	if errors.Is(err, sql.ErrNoRows) {
		return nil, nil
	}
	return receipt, err
}

func (s *Service) ClaimNext(ctx context.Context, workerID, claimToken string, leaseExpiresAt time.Time) (receipt *Receipt, claimed bool, retErr error) {
	timer := newReceiptClaimStageTimer()
	defer func() {
		outcome := "ok"
		if retErr != nil {
			outcome = "error"
		} else if !claimed {
			outcome = "none"
		}
		timer.finish(outcome)
	}()

	if s == nil || s.db == nil {
		return nil, false, fmt.Errorf("connector ingress receipt store is not wired")
	}
	if s.cipher == nil {
		return nil, false, ErrCipherRequired
	}
	workerID = strings.TrimSpace(workerID)
	claimToken = strings.TrimSpace(claimToken)
	if workerID == "" || claimToken == "" || leaseExpiresAt.IsZero() {
		return nil, false, fmt.Errorf("%w: claim requires worker, token, and lease expiry", ErrInvalidReceipt)
	}
	now := s.now()
	var id string
	includePendingConfig := true
	id, err := s.selectClaimableReceiptID(ctx, now, includePendingConfig)
	if errors.Is(err, sql.ErrNoRows) {
		timer.mark("select_id", "none")
		return nil, false, nil
	}
	if err != nil {
		timer.mark("select_id", "error")
		return nil, false, err
	}
	timer.mark("select_id", "ok")
	res, err := s.claimSelectedReceipt(ctx, id, workerID, claimToken, leaseExpiresAt, now, includePendingConfig)
	if err != nil {
		timer.mark("claim_update", "error")
		return nil, false, err
	}
	n, err := res.RowsAffected()
	if err != nil {
		timer.mark("claim_update", "error")
		return nil, false, err
	}
	if n == 0 {
		timer.mark("claim_update", "lost_race")
		return nil, false, nil
	}
	timer.mark("claim_update", "ok")
	receipt, err = s.findByID(ctx, id)
	if err != nil {
		timer.mark("read_receipt", "error")
		return nil, false, err
	}
	timer.mark("read_receipt", "ok")
	logReceiptTransition(ctx, "claim", receipt.ID, receipt.Namespace, receipt.InstalledPackID, receipt.Status)
	return receipt, true, nil
}

func (s *Service) ClaimBatch(ctx context.Context, workerID, claimToken string, leaseExpiresAt time.Time, limit int) (receipts []*Receipt, claimed bool, retErr error) {
	timer := newReceiptClaimStageTimer()
	defer func() {
		outcome := "ok"
		if retErr != nil {
			outcome = "error"
		} else if !claimed {
			outcome = "none"
		}
		timer.finish(outcome)
	}()

	if s == nil || s.db == nil {
		return nil, false, fmt.Errorf("connector ingress receipt store is not wired")
	}
	if s.cipher == nil {
		return nil, false, ErrCipherRequired
	}
	workerID = strings.TrimSpace(workerID)
	claimToken = strings.TrimSpace(claimToken)
	if workerID == "" || claimToken == "" || leaseExpiresAt.IsZero() {
		return nil, false, fmt.Errorf("%w: claim requires worker, token, and lease expiry", ErrInvalidReceipt)
	}
	if limit <= 0 {
		limit = 1
	}
	if limit > 100 {
		limit = 100
	}

	now := s.now()
	pendingConfigLimit := pendingConfigFairShareLimit(limit)
	configIDs, err := s.selectPendingConfigReceiptIDs(ctx, now, pendingConfigLimit, nil)
	if err != nil {
		timer.mark("select_batch_id", "error")
		return nil, false, err
	}
	hotLimit := limit - len(configIDs)
	hotIDs, err := s.selectClaimableReceiptIDs(ctx, now, false, hotLimit)
	if err != nil {
		timer.mark("select_batch_id", "error")
		return nil, false, err
	}
	extraConfigLimit := limit - len(configIDs) - len(hotIDs)
	// Only probe beyond the fair-share quota when the first query filled that
	// quota. A short first page proves there was no additional eligible config
	// work in that snapshot; repeating the same predicate with a larger LIMIT is
	// pure idle traffic. Concurrent inserts remain covered by the next bounded
	// poll (or the receipt-router wake path).
	if extraConfigLimit > 0 && len(configIDs) == pendingConfigLimit {
		extraConfigIDs, err := s.selectPendingConfigReceiptIDs(ctx, now, extraConfigLimit, configIDs)
		if err != nil {
			timer.mark("select_batch_id", "error")
			return nil, false, err
		}
		configIDs = append(configIDs, extraConfigIDs...)
	}
	ids := append(configIDs, hotIDs...)
	if len(ids) == 0 {
		timer.mark("select_batch_id", "none")
		return nil, false, nil
	}
	timer.mark("select_batch_id", "ok")

	res, err := s.claimSelectedReceipts(ctx, ids, workerID, claimToken, leaseExpiresAt, now, true)
	if err != nil {
		timer.mark("claim_batch_update", "error")
		return nil, false, err
	}
	n, err := res.RowsAffected()
	if err != nil {
		timer.mark("claim_batch_update", "error")
		return nil, false, err
	}
	if n == 0 {
		timer.mark("claim_batch_update", "lost_race")
		return nil, false, nil
	}
	timer.mark("claim_batch_update", "ok")

	receipts, err = s.findByIDsWithClaim(ctx, ids, claimToken)
	if err != nil {
		timer.mark("read_batch_receipts", "error")
		return nil, false, err
	}
	if len(receipts) == 0 {
		timer.mark("read_batch_receipts", "lost_race")
		return nil, false, nil
	}
	timer.mark("read_batch_receipts", "ok")
	for _, receipt := range receipts {
		logReceiptTransition(ctx, "claim", receipt.ID, receipt.Namespace, receipt.InstalledPackID, receipt.Status)
	}
	return receipts, true, nil
}

// NextClaimableAt returns the earliest persisted retry or expired-lease
// deadline. It is a scheduling hint only: callers must still use ClaimBatch's
// authoritative predicates and CAS before processing a receipt.
func (s *Service) NextClaimableAt(ctx context.Context) (time.Time, bool, error) {
	if s == nil || s.db == nil || ctx == nil {
		return time.Time{}, false, fmt.Errorf("connector ingress receipt store is not wired")
	}
	now := s.now()
	scopeSQL, scopeArgs := s.claimScopePredicate()
	args := []interface{}{now, StatusPending, StatusRetry, StatusPendingConfig}
	args = append(args, scopeArgs...)
	args = append(args, StatusProcessing)
	args = append(args, scopeArgs...)
	var next sql.NullTime
	// COALESCE with a bound clock can make TiDB infer a string aggregate.
	// Keep the result temporal, including the clock's microseconds.
	err := s.db.QueryRowContext(ctx, `
SELECT CAST(MIN(scheduled.WakeAt) AS DATETIME(6))
  FROM (
    SELECT COALESCE(NextAttemptAt, ?) AS WakeAt
      FROM _System_ConnectorIngressReceipt
     WHERE Status IN (?, ?, ?)`+scopeSQL+`
    UNION ALL
    SELECT LeaseExpiresAt AS WakeAt
      FROM _System_ConnectorIngressReceipt
     WHERE Status = ?
       AND LeaseExpiresAt IS NOT NULL`+scopeSQL+`
  ) AS scheduled`, args...).Scan(&next)
	if err != nil {
		return time.Time{}, false, fmt.Errorf("connector ingress receipt next claimable time: %w", err)
	}
	if !next.Valid {
		return time.Time{}, false, nil
	}
	return next.Time.UTC(), true, nil
}

func (s *Service) selectClaimableReceiptID(ctx context.Context, now time.Time, includePendingConfig bool) (string, error) {
	var id string
	predicateSQL, args := claimableReceiptPredicate(now, includePendingConfig)
	scopeSQL, scopeArgs := s.claimScopePredicate()
	args = append(args, scopeArgs...)
	err := s.db.QueryRowContext(ctx, `
SELECT Id
  FROM _System_ConnectorIngressReceipt
 WHERE `+predicateSQL+scopeSQL+`
 ORDER BY CreatedDate ASC, Id ASC
 LIMIT 1`, args...).Scan(&id)
	return id, err
}

func (s *Service) selectClaimableReceiptIDs(ctx context.Context, now time.Time, includePendingConfig bool, limit int) ([]string, error) {
	if limit <= 0 {
		return nil, nil
	}
	predicateSQL, args := claimableReceiptPredicate(now, includePendingConfig)
	scopeSQL, scopeArgs := s.claimScopePredicate()
	args = append(args, scopeArgs...)
	args = append(args, limit)
	rows, err := s.db.QueryContext(ctx, `
SELECT Id
  FROM _System_ConnectorIngressReceipt
 WHERE `+predicateSQL+scopeSQL+`
 ORDER BY CreatedDate ASC, Id ASC
 LIMIT ?`, args...)
	if err != nil {
		return nil, err
	}
	return scanReceiptIDRows(rows, limit)
}

func (s *Service) selectPendingConfigReceiptIDs(ctx context.Context, now time.Time, limit int, excludeIDs []string) ([]string, error) {
	if limit <= 0 {
		return nil, nil
	}
	excludeIDs = compactStrings(excludeIDs)
	args := []interface{}{StatusPendingConfig, now, now}
	scopeSQL, scopeArgs := s.claimScopePredicate()
	args = append(args, scopeArgs...)
	excludeSQL := ""
	if len(excludeIDs) > 0 {
		excludeSQL = "\n   AND Id NOT IN (" + sqlPlaceholders(len(excludeIDs)) + ")"
		for _, id := range excludeIDs {
			args = append(args, id)
		}
	}
	args = append(args, limit)
	rows, err := s.db.QueryContext(ctx, `
SELECT Id
  FROM _System_ConnectorIngressReceipt
 WHERE Status = ?
   AND (NextAttemptAt IS NULL OR NextAttemptAt <= ?)
   AND (LeaseExpiresAt IS NULL OR LeaseExpiresAt <= ?)`+scopeSQL+excludeSQL+`
 ORDER BY CreatedDate ASC, Id ASC
 LIMIT ?`, args...)
	if err != nil {
		return nil, err
	}
	return scanReceiptIDRows(rows, limit)
}

func (s *Service) claimScopePredicate() (string, []interface{}) {
	if s == nil || !s.claimScopeEnabled {
		return "", nil
	}
	if len(s.claimPackInstallIDs) == 0 {
		return "\n   AND 1 = 0", nil
	}
	args := make([]interface{}, len(s.claimPackInstallIDs))
	for index, packInstallID := range s.claimPackInstallIDs {
		args[index] = packInstallID
	}
	return "\n   AND InstalledPackId IN (" + sqlPlaceholders(len(args)) + ")", args
}

var (
	claimableReceiptWithoutPendingConfigPredicateSQL = makeClaimableReceiptPredicateSQL(2)
	claimableReceiptWithPendingConfigPredicateSQL    = makeClaimableReceiptPredicateSQL(3)
)

func makeClaimableReceiptPredicateSQL(statusCount int) string {
	return `((Status IN (` + sqlPlaceholders(statusCount) + `)
         AND (NextAttemptAt IS NULL OR NextAttemptAt <= ?)
         AND (LeaseExpiresAt IS NULL OR LeaseExpiresAt <= ?))
        OR (Status = ? AND LeaseExpiresAt <= ?))`
}

func claimableReceiptPredicate(now time.Time, includePendingConfig bool) (string, []interface{}) {
	predicateSQL := claimableReceiptWithoutPendingConfigPredicateSQL
	args := []interface{}{StatusPending, StatusRetry}
	if includePendingConfig {
		predicateSQL = claimableReceiptWithPendingConfigPredicateSQL
		args = append(args, StatusPendingConfig)
	}
	args = append(args, now, now, StatusProcessing, now)
	return predicateSQL, args
}

func scanReceiptIDRows(rows *sql.Rows, limit int) ([]string, error) {
	defer rows.Close()
	ids := make([]string, 0, limit)
	for rows.Next() {
		var id string
		if err := rows.Scan(&id); err != nil {
			return nil, err
		}
		id = strings.TrimSpace(id)
		if id != "" {
			ids = append(ids, id)
		}
	}
	if err := rows.Err(); err != nil {
		return nil, err
	}
	return ids, nil
}

func pendingConfigFairShareLimit(limit int) int {
	if limit <= 0 {
		return 0
	}
	if limit == 1 {
		return 1
	}
	quota := limit / 10
	if quota < 1 {
		quota = 1
	}
	if quota > limit {
		return limit
	}
	return quota
}

func (s *Service) claimSelectedReceipt(ctx context.Context, id, workerID, claimToken string, leaseExpiresAt, now time.Time, includePendingConfig bool) (sql.Result, error) {
	predicateSQL, predicateArgs := claimableReceiptPredicate(now, includePendingConfig)
	scopeSQL, scopeArgs := s.claimScopePredicate()
	args := []interface{}{StatusProcessing, workerID, claimToken, now, leaseExpiresAt, now, id}
	args = append(args, predicateArgs...)
	args = append(args, scopeArgs...)
	return s.db.ExecContext(ctx, `
UPDATE _System_ConnectorIngressReceipt
   SET Status = ?, ClaimedBy = ?, ClaimToken = ?, ClaimedAt = ?, LeaseExpiresAt = ?, LastModifiedDate = ?
 WHERE Id = ?
   AND `+predicateSQL+scopeSQL,
		args...)
}

func (s *Service) claimSelectedReceipts(ctx context.Context, ids []string, workerID, claimToken string, leaseExpiresAt, now time.Time, includePendingConfig bool) (sql.Result, error) {
	ids = compactStrings(ids)
	if len(ids) == 0 {
		return noRowsResult{}, nil
	}
	inSQL := sqlPlaceholders(len(ids))
	args := []interface{}{StatusProcessing, workerID, claimToken, now, leaseExpiresAt, now}
	for _, id := range ids {
		args = append(args, id)
	}
	predicateSQL, predicateArgs := claimableReceiptPredicate(now, includePendingConfig)
	args = append(args, predicateArgs...)
	scopeSQL, scopeArgs := s.claimScopePredicate()
	args = append(args, scopeArgs...)
	return s.db.ExecContext(ctx, `
UPDATE _System_ConnectorIngressReceipt
   SET Status = ?, ClaimedBy = ?, ClaimToken = ?, ClaimedAt = ?, LeaseExpiresAt = ?, LastModifiedDate = ?
 WHERE Id IN (`+inSQL+`)
   AND `+predicateSQL+scopeSQL,
		args...)
}

type receiptClaimStageTimer struct {
	start time.Time
	last  time.Time
}

func newReceiptClaimStageTimer() *receiptClaimStageTimer {
	now := time.Now()
	return &receiptClaimStageTimer{start: now, last: now}
}

func (t *receiptClaimStageTimer) mark(stage, outcome string) {
	if t == nil {
		return
	}
	now := time.Now()
	recordConnectorIngressReceiptClaimStageDuration(stage, outcome, now.Sub(t.last).Seconds())
	t.last = now
}

func (t *receiptClaimStageTimer) finish(outcome string) {
	if t == nil {
		return
	}
	recordConnectorIngressReceiptClaimStageDuration("total", outcome, time.Since(t.start).Seconds())
}

func (s *Service) MarkEnqueued(ctx context.Context, receiptID, claimToken string, eventIDs []string, metadata map[string]string) error {
	eventIDsJSON, err := json.Marshal(eventIDs)
	if err != nil {
		return err
	}
	metadataJSON, err := json.Marshal(metadata)
	if err != nil {
		return err
	}
	return s.updateClaimedTerminal(ctx, receiptID, claimToken, StatusEnqueued, string(eventIDsJSON), string(metadataJSON), nil)
}

func (s *Service) MarkEnqueuedBatch(ctx context.Context, updates []EnqueuedReceiptUpdate) error {
	if s == nil || s.db == nil {
		return fmt.Errorf("connector ingress receipt store is not wired")
	}
	updates = compactEnqueuedReceiptUpdates(updates)
	if len(updates) == 0 {
		return nil
	}
	now := s.now()
	statusCase := newCaseByID()
	eventsCase := newCaseByID()
	metadataCase := newCaseByID()
	whereSQL, whereArgs := receiptClaimPairsWhere(updates)
	for _, update := range updates {
		eventIDsJSON, err := json.Marshal(update.EventIDs)
		if err != nil {
			return err
		}
		metadataJSON, err := json.Marshal(update.Metadata)
		if err != nil {
			return err
		}
		statusCase.add(update.ReceiptID, StatusEnqueued)
		eventsCase.add(update.ReceiptID, string(eventIDsJSON))
		metadataCase.add(update.ReceiptID, string(metadataJSON))
	}
	args := []interface{}{}
	args = append(args, statusCase.args...)
	args = append(args, eventsCase.args...)
	args = append(args, metadataCase.args...)
	args = append(args, now)
	args = append(args, StatusProcessing)
	args = append(args, whereArgs...)
	res, err := s.db.ExecContext(ctx, `
UPDATE _System_ConnectorIngressReceipt
   SET Status = `+statusCase.sql("Status")+`,
       EnqueuedEventIds = `+eventsCase.sql("EnqueuedEventIds")+`,
       EnqueueMetadata = `+metadataCase.sql("EnqueueMetadata")+`,
       LastError = NULL,
       LastErrorAt = NULL,
       ClaimedBy = NULL,
       ClaimToken = NULL,
       ClaimedAt = NULL,
       LeaseExpiresAt = NULL,
       LastModifiedDate = ?
 WHERE Status = ?
   AND (`+whereSQL+`)`,
		args...)
	if err != nil {
		return err
	}
	return requireUpdatedCount(res, int64(len(updates)))
}

func (s *Service) MarkMaterializing(ctx context.Context, receiptID, claimToken string, metadata map[string]string) error {
	if s == nil || s.db == nil {
		return fmt.Errorf("connector ingress receipt store is not wired")
	}
	metadataJSON, err := json.Marshal(metadata)
	if err != nil {
		return err
	}
	now := s.now()
	res, err := s.db.ExecContext(ctx, `
UPDATE _System_ConnectorIngressReceipt
   SET EnqueueMetadata = ?, LastModifiedDate = ?
 WHERE Id = ?
   AND Status = ?
   AND ClaimToken = ?`,
		string(metadataJSON), now,
		strings.TrimSpace(receiptID), StatusProcessing, strings.TrimSpace(claimToken))
	if err != nil {
		return err
	}
	if err := requireUpdated(res); err != nil {
		return err
	}
	logReceiptTransition(ctx, "materializing", strings.TrimSpace(receiptID), "", "", StatusProcessing)
	return nil
}

func (s *Service) MarkMaterializingBatch(ctx context.Context, updates []MaterializingReceiptUpdate) error {
	if s == nil || s.db == nil {
		return fmt.Errorf("connector ingress receipt store is not wired")
	}
	updates = compactMaterializingReceiptUpdates(updates)
	if len(updates) == 0 {
		return nil
	}
	now := s.now()
	metadataCase := newCaseByID()
	whereSQL, whereArgs := materializingClaimPairsWhere(updates)
	for _, update := range updates {
		metadataJSON, err := json.Marshal(update.Metadata)
		if err != nil {
			return err
		}
		metadataCase.add(update.ReceiptID, string(metadataJSON))
	}
	args := []interface{}{}
	args = append(args, metadataCase.args...)
	args = append(args, now)
	args = append(args, StatusProcessing)
	args = append(args, whereArgs...)
	res, err := s.db.ExecContext(ctx, `
UPDATE _System_ConnectorIngressReceipt
   SET EnqueueMetadata = `+metadataCase.sql("EnqueueMetadata")+`,
       LastModifiedDate = ?
 WHERE Status = ?
   AND (`+whereSQL+`)`,
		args...)
	if err != nil {
		return err
	}
	return requireUpdatedCount(res, int64(len(updates)))
}

func (s *Service) MarkRetry(ctx context.Context, receiptID, claimToken string, nextAttemptAt time.Time, reason error) error {
	return s.markClaimedBackoff(ctx, receiptID, claimToken, StatusRetry, nextAttemptAt, reason)
}

func (s *Service) MarkPendingConfig(ctx context.Context, receiptID, claimToken string, nextAttemptAt time.Time, reason error) error {
	return s.markClaimedBackoff(ctx, receiptID, claimToken, StatusPendingConfig, nextAttemptAt, reason)
}

func (s *Service) markClaimedBackoff(ctx context.Context, receiptID, claimToken, status string, nextAttemptAt time.Time, reason error) error {
	if s == nil || s.db == nil {
		return fmt.Errorf("connector ingress receipt store is not wired")
	}
	if nextAttemptAt.IsZero() {
		return fmt.Errorf("%w: backoff requires NextAttemptAt", ErrInvalidReceipt)
	}
	now := s.now()
	res, err := s.db.ExecContext(ctx, `
UPDATE _System_ConnectorIngressReceipt
   SET Status = ?, AttemptCount = AttemptCount + 1, NextAttemptAt = ?, LastError = ?, LastErrorAt = ?,
       ClaimedBy = NULL, ClaimToken = NULL, ClaimedAt = NULL, LeaseExpiresAt = NULL,
       LastModifiedDate = ?
 WHERE Id = ?
   AND Status = ?
   AND ClaimToken = ?`,
		status, nextAttemptAt, truncateLastError(reason), now, now,
		strings.TrimSpace(receiptID), StatusProcessing, strings.TrimSpace(claimToken))
	if err != nil {
		return err
	}
	if err := requireUpdated(res); err != nil {
		return err
	}
	logReceiptTransition(ctx, status, strings.TrimSpace(receiptID), "", "", status)
	return nil
}

func (s *Service) MarkDead(ctx context.Context, receiptID, claimToken string, reason error) error {
	return s.updateClaimedTerminal(ctx, receiptID, claimToken, StatusDead, "", "", reason)
}

func (s *Service) MarkConfigTerminal(ctx context.Context, receiptID, claimToken string, reason error) error {
	return s.updateClaimedTerminal(ctx, receiptID, claimToken, StatusConfigTerminal, "", "", reason)
}

func (s *Service) updateClaimedTerminal(ctx context.Context, receiptID, claimToken, status, eventIDsJSON, metadataJSON string, reason error) error {
	if s == nil || s.db == nil {
		return fmt.Errorf("connector ingress receipt store is not wired")
	}
	now := s.now()
	var eventIDsArg interface{}
	if eventIDsJSON != "" {
		eventIDsArg = eventIDsJSON
	}
	var metadataArg interface{}
	if metadataJSON != "" {
		metadataArg = metadataJSON
	}
	res, err := s.db.ExecContext(ctx, `
UPDATE _System_ConnectorIngressReceipt
   SET Status = ?, EnqueuedEventIds = ?, EnqueueMetadata = ?, LastError = ?, LastErrorAt = ?,
       ClaimedBy = NULL, ClaimToken = NULL, ClaimedAt = NULL, LeaseExpiresAt = NULL,
       LastModifiedDate = ?
 WHERE Id = ?
   AND Status = ?
   AND ClaimToken = ?`,
		status, eventIDsArg, metadataArg, truncateLastError(reason), nullableTime(now, reason != nil), now,
		strings.TrimSpace(receiptID), StatusProcessing, strings.TrimSpace(claimToken))
	if err != nil {
		return err
	}
	if err := requireUpdated(res); err != nil {
		return err
	}
	logReceiptTransition(ctx, "terminal", strings.TrimSpace(receiptID), "", "", status)
	return nil
}

func (s *Service) findByID(ctx context.Context, id string) (*Receipt, error) {
	scopeSQL, scopeArgs := s.claimScopePredicate()
	args := []interface{}{id}
	args = append(args, scopeArgs...)
	receipt, err := s.scanReceipt(s.db.QueryRowContext(ctx, connectorIngressReceiptSelectSQL+`
 WHERE Id = ?
`+scopeSQL+`
 LIMIT 1`, args...))
	if errors.Is(err, sql.ErrNoRows) {
		return nil, ErrReceiptNotFound
	}
	return receipt, err
}

func (s *Service) findByIDsWithClaim(ctx context.Context, ids []string, claimToken string) ([]*Receipt, error) {
	ids = compactStrings(ids)
	claimToken = strings.TrimSpace(claimToken)
	if len(ids) == 0 || claimToken == "" {
		return nil, nil
	}
	args := []interface{}{claimToken}
	for _, id := range ids {
		args = append(args, id)
	}
	scopeSQL, scopeArgs := s.claimScopePredicate()
	args = append(args, scopeArgs...)
	rows, err := s.db.QueryContext(ctx, connectorIngressReceiptSelectSQL+`
 WHERE ClaimToken = ?
   AND Id IN (`+sqlPlaceholders(len(ids))+`)
`+scopeSQL+`
 ORDER BY CreatedDate ASC, Id ASC`, args...)
	if err != nil {
		return nil, err
	}
	defer rows.Close()
	receipts := make([]*Receipt, 0, len(ids))
	for rows.Next() {
		receipt, err := s.scanReceipt(rows)
		if err != nil {
			return nil, err
		}
		receipts = append(receipts, receipt)
	}
	if err := rows.Err(); err != nil {
		return nil, err
	}
	return receipts, nil
}

const connectorIngressReceiptSelectSQL = `
SELECT 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
  FROM _System_ConnectorIngressReceipt`

func (s *Service) scanReceipt(row interface {
	Scan(dest ...interface{}) error
}) (*Receipt, error) {
	var r Receipt
	var targetCiphertext, bodyCiphertext, headersCiphertext, eventIDsJSON, metadataJSON string
	var nextAttemptAt, claimedAt, leaseExpiresAt, lastErrorAt, purgeAfter sql.NullTime
	err := row.Scan(
		&r.ID, &r.Namespace, &r.InstalledPackID, &r.PackVersion, &r.SecretName, &r.SecretAlgorithm,
		&r.VerifierCapabilityKind, &r.ParserCapabilityKind, &r.RequestMethod, &r.RequestTargetHash, &targetCiphertext,
		&r.ReceiptKey, &r.BodyHash, &r.BodySizeBytes, &bodyCiphertext, &headersCiphertext, &r.Status, &r.AttemptCount,
		&nextAttemptAt, &r.ClaimedBy, &r.ClaimToken, &claimedAt, &leaseExpiresAt,
		&eventIDsJSON, &metadataJSON, &r.LastError, &lastErrorAt,
		&purgeAfter, &r.CreatedDate, &r.LastModifiedDate,
	)
	if err != nil {
		return nil, err
	}
	targetJSON, err := s.cipher.Decrypt(targetCiphertext)
	if err != nil {
		return nil, err
	}
	var target struct {
		RequestURI string `json:"requestURI"`
		RequestURL string `json:"requestURL"`
	}
	if err := json.Unmarshal([]byte(targetJSON), &target); err != nil {
		return nil, err
	}
	r.RequestURI = target.RequestURI
	r.RequestURL = target.RequestURL
	rawBase64, err := s.cipher.Decrypt(bodyCiphertext)
	if err != nil {
		return nil, err
	}
	r.RawBody, err = base64.StdEncoding.DecodeString(rawBase64)
	if err != nil {
		return nil, err
	}
	headerJSON, err := s.cipher.Decrypt(headersCiphertext)
	if err != nil {
		return nil, err
	}
	if err := json.Unmarshal([]byte(headerJSON), &r.Headers); err != nil {
		return nil, err
	}
	if err := json.Unmarshal([]byte(eventIDsJSON), &r.EnqueuedEventIDs); err != nil {
		return nil, err
	}
	if err := json.Unmarshal([]byte(metadataJSON), &r.EnqueueMetadata); err != nil {
		return nil, err
	}
	r.NextAttemptAt = timePtr(nextAttemptAt)
	r.ClaimedAt = timePtr(claimedAt)
	r.LeaseExpiresAt = timePtr(leaseExpiresAt)
	r.LastErrorAt = timePtr(lastErrorAt)
	r.PurgeAfter = timePtr(purgeAfter)
	return &r, nil
}

func sameVerifiedReceiptBody(existing *Receipt, rawBody []byte) bool {
	if existing == nil {
		return false
	}
	if existing.BodyHash != BodyHash(rawBody) || existing.BodySizeBytes != int64(len(rawBody)) {
		return false
	}
	return bytes.Equal(existing.RawBody, rawBody)
}

func (s *Service) ListMetadata(ctx context.Context, perms PermissionChecker, filter ReceiptMetadataFilter) ([]ReceiptMetadata, error) {
	if !hasReceiptPermission(perms, constants.PermViewConnectorEvents) {
		return nil, ErrPermissionDenied
	}
	if s == nil || s.db == nil {
		return nil, fmt.Errorf("connector ingress receipt store is not wired")
	}
	filter.Namespace = strings.ToLower(strings.TrimSpace(filter.Namespace))
	filter.Status = strings.TrimSpace(filter.Status)
	limit, offset := normalizePage(filter.Page)

	query := connectorIngressReceiptMetadataSelectSQL
	args := make([]interface{}, 0, 4)
	where := make([]string, 0, 2)
	if filter.Namespace != "" {
		where = append(where, "Namespace = ?")
		args = append(args, filter.Namespace)
	}
	if filter.Status != "" {
		where = append(where, "Status = ?")
		args = append(args, filter.Status)
	}
	if len(where) > 0 {
		query += "\n WHERE " + strings.Join(where, " AND ")
	}
	query += "\n ORDER BY CreatedDate DESC, Id DESC\n LIMIT ? OFFSET ?"
	args = append(args, limit, offset)

	rows, err := s.db.QueryContext(ctx, query, args...)
	if err != nil {
		return nil, err
	}
	defer rows.Close()

	out := make([]ReceiptMetadata, 0)
	for rows.Next() {
		item, err := scanReceiptMetadata(rows)
		if err != nil {
			return nil, err
		}
		out = append(out, item)
	}
	if err := rows.Err(); err != nil {
		return nil, err
	}
	return out, nil
}

func (s *Service) GetMetadata(ctx context.Context, perms PermissionChecker, id string) (*ReceiptMetadata, error) {
	if !hasReceiptPermission(perms, constants.PermViewConnectorEvents) {
		return nil, ErrPermissionDenied
	}
	if s == nil || s.db == nil {
		return nil, fmt.Errorf("connector ingress receipt store is not wired")
	}
	id = strings.TrimSpace(id)
	if id == "" {
		return nil, ErrReceiptNotFound
	}
	item, err := scanReceiptMetadata(s.db.QueryRowContext(ctx, connectorIngressReceiptMetadataSelectSQL+`
 WHERE Id = ?
 LIMIT 1`, id))
	if errors.Is(err, sql.ErrNoRows) {
		return nil, ErrReceiptNotFound
	}
	if err != nil {
		return nil, err
	}
	return &item, nil
}

func (s *Service) PurgeTerminal(ctx context.Context, perms PermissionChecker, req PurgeRequest) (PurgeResult, error) {
	if !hasReceiptPermission(perms, constants.PermReplayConnectorEvents) {
		return PurgeResult{}, ErrPermissionDenied
	}
	if s == nil || s.db == nil {
		return PurgeResult{}, fmt.Errorf("connector ingress receipt store is not wired")
	}
	before := req.Before
	if before.IsZero() {
		before = s.now()
	}
	limit := req.Limit
	if limit <= 0 {
		limit = 100
	}
	if limit > 1000 {
		limit = 1000
	}
	namespace := strings.ToLower(strings.TrimSpace(req.Namespace))

	query := `
DELETE FROM _System_ConnectorIngressReceipt
 WHERE Status IN (?, ?, ?)
   AND PurgeAfter IS NOT NULL
   AND PurgeAfter <= ?`
	args := []interface{}{StatusEnqueued, StatusDead, StatusConfigTerminal, before}
	if namespace != "" {
		query += `
   AND Namespace = ?`
		args = append(args, namespace)
	}
	query += `
 ORDER BY PurgeAfter ASC, Id ASC
 LIMIT ?`
	args = append(args, limit)

	res, err := s.db.ExecContext(ctx, query, args...)
	if err != nil {
		return PurgeResult{}, err
	}
	deleted, err := res.RowsAffected()
	if err != nil {
		return PurgeResult{}, err
	}
	log.Ctx(ctx).Info().
		Str("component", "connector_ingress_receipts").
		Str("operation", "purge").
		Str("namespace", namespace).
		Time("before", before).
		Int64("deleted", deleted).
		Msg("connector ingress receipt purge completed")
	return PurgeResult{Deleted: deleted, Before: before}, nil
}

const connectorIngressReceiptMetadataSelectSQL = `
SELECT 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
  FROM _System_ConnectorIngressReceipt`

func scanReceiptMetadata(row interface {
	Scan(dest ...interface{}) error
}) (ReceiptMetadata, error) {
	var r ReceiptMetadata
	var nextAttemptAt, claimedAt, leaseExpiresAt, lastErrorAt, purgeAfter sql.NullTime
	var eventIDsJSON, lastError string
	err := row.Scan(
		&r.ID, &r.Namespace, &r.InstalledPackID, &r.PackVersion, &r.SecretName, &r.SecretAlgorithm,
		&r.VerifierCapabilityKind, &r.ParserCapabilityKind, &r.RequestMethod, &r.RequestTargetHash,
		&r.BodyHash, &r.BodySizeBytes, &r.Status, &r.AttemptCount,
		&nextAttemptAt, &r.ClaimedBy, &claimedAt, &leaseExpiresAt,
		&eventIDsJSON, &lastError, &lastErrorAt,
		&purgeAfter, &r.CreatedDate, &r.LastModifiedDate,
	)
	if err != nil {
		return ReceiptMetadata{}, err
	}
	if err := json.Unmarshal([]byte(eventIDsJSON), &r.EnqueuedEventIDs); err != nil {
		return ReceiptMetadata{}, err
	}
	r.NextAttemptAt = timePtr(nextAttemptAt)
	r.ClaimedAt = timePtr(claimedAt)
	r.LeaseExpiresAt = timePtr(leaseExpiresAt)
	r.LastErrorAt = timePtr(lastErrorAt)
	r.PurgeAfter = timePtr(purgeAfter)
	r.LastErrorClass = classifyLastError(r.Status, lastError)
	return r, nil
}

func normalizeVerifiedReceiptInput(in VerifiedReceiptInput) VerifiedReceiptInput {
	in.Namespace = strings.ToLower(strings.TrimSpace(in.Namespace))
	in.InstalledPackID = strings.TrimSpace(in.InstalledPackID)
	in.PackVersion = strings.TrimSpace(in.PackVersion)
	in.PackPayloadHash = strings.TrimSpace(in.PackPayloadHash)
	in.SecretName = strings.ToLower(strings.TrimSpace(in.SecretName))
	in.SecretAlgorithm = strings.TrimSpace(in.SecretAlgorithm)
	in.SecretValueHash = strings.ToLower(strings.TrimSpace(in.SecretValueHash))
	in.VerifierCapabilityKind = strings.TrimSpace(in.VerifierCapabilityKind)
	in.ParserCapabilityKind = strings.TrimSpace(in.ParserCapabilityKind)
	in.RequestMethod = strings.ToUpper(strings.TrimSpace(in.RequestMethod))
	in.RequestURI = strings.TrimSpace(in.RequestURI)
	in.RequestURL = strings.TrimSpace(in.RequestURL)
	in.DeliveryIdentity = strings.TrimSpace(in.DeliveryIdentity)
	in.ReceiptKey = strings.TrimSpace(in.ReceiptKey)
	return in
}

func validateVerifiedReceiptInput(in VerifiedReceiptInput) error {
	if in.Namespace == "" || in.InstalledPackID == "" || in.SecretName == "" || in.RequestMethod == "" || in.RequestURI == "" {
		return fmt.Errorf("%w: namespace, pack, secret, method, and URI are required", ErrInvalidReceipt)
	}
	if in.ReceiptKey == "" && in.DeliveryIdentity == "" {
		return fmt.Errorf("%w: delivery identity or receipt key is required", ErrInvalidReceipt)
	}
	return nil
}

func normalizePage(page Page) (limit int, offset int) {
	limit = page.Limit
	if limit <= 0 {
		limit = 50
	}
	if limit > 500 {
		limit = 500
	}
	offset = page.Offset
	if offset < 0 {
		offset = 0
	}
	return limit, offset
}

func hasReceiptPermission(perms PermissionChecker, perm constants.Permission) bool {
	return perms != nil && perms.Has(string(perm))
}

func classifyLastError(status, lastError string) string {
	if strings.TrimSpace(lastError) == "" {
		return ""
	}
	switch strings.TrimSpace(status) {
	case StatusRetry:
		return "retry_error"
	case StatusPendingConfig:
		return "config_pending_error"
	case StatusDead, StatusConfigTerminal:
		return "terminal_error"
	default:
		return "receipt_error"
	}
}

func logReceiptTransition(ctx context.Context, operation, receiptID, namespace, installedPackID, status string) {
	log.Ctx(ctx).Info().
		Str("component", "connector_ingress_receipts").
		Str("operation", operation).
		Str("receipt_id", receiptID).
		Str("namespace", namespace).
		Str("installed_pack_id", installedPackID).
		Str("status", status).
		Msg("connector ingress receipt transition")
}

func BodyHash(body []byte) string {
	sum := sha256.Sum256(body)
	return hex.EncodeToString(sum[:])
}

func RequestTargetHash(method, requestURI, requestURL string) string {
	parts := []string{
		strings.ToUpper(strings.TrimSpace(method)),
		strings.TrimSpace(requestURI),
		strings.TrimSpace(requestURL),
	}
	sum := sha256.Sum256([]byte(strings.Join(parts, "\x1f")))
	return hex.EncodeToString(sum[:])
}

func ReceiptKeyFor(namespace, installedPackID, secretName, deliveryIdentity string) string {
	parts := []string{
		strings.ToLower(strings.TrimSpace(namespace)),
		strings.TrimSpace(installedPackID),
		strings.ToLower(strings.TrimSpace(secretName)),
		strings.TrimSpace(deliveryIdentity),
	}
	sum := sha256.Sum256([]byte(strings.Join(parts, "\x1f")))
	return "cingr:" + hex.EncodeToString(sum[:])
}

func orderedReceiptID(now time.Time, entropy string) string {
	entropy = strings.TrimSpace(entropy)
	if entropy == "" {
		entropy = id.Generate()
	}
	if now.IsZero() {
		now = time.Now()
	}
	return "ir:" + now.UTC().Format("20060102T150405.000000000Z") + ":" + entropy
}

func AllowlistedHeaders(headers http.Header, allowlist []string, blockedHeaders ...[]string) http.Header {
	allowed := make(map[string]string, len(allowlist))
	for _, name := range allowlist {
		canonical := http.CanonicalHeaderKey(strings.TrimSpace(name))
		if canonical == "" {
			continue
		}
		allowed[strings.ToLower(canonical)] = canonical
	}
	blocked := make(map[string]struct{})
	for _, group := range blockedHeaders {
		for _, name := range group {
			canonical := http.CanonicalHeaderKey(strings.TrimSpace(name))
			if canonical != "" {
				blocked[strings.ToLower(canonical)] = struct{}{}
			}
		}
	}
	out := http.Header{}
	keys := make([]string, 0, len(headers))
	for key := range headers {
		canonical := http.CanonicalHeaderKey(key)
		if isSensitiveHeaderName(canonical) {
			continue
		}
		keys = append(keys, canonical)
	}
	sort.Strings(keys)
	for _, key := range keys {
		canonical := http.CanonicalHeaderKey(key)
		if _, ok := blocked[strings.ToLower(canonical)]; ok {
			continue
		}
		outKey, ok := allowed[strings.ToLower(canonical)]
		if !ok {
			continue
		}
		values := headers.Values(key)
		if len(values) == 0 {
			continue
		}
		out[outKey] = append([]string(nil), values...)
	}
	return out
}

var sensitiveHeaderNameFragments = []string{
	"auth",
	"credential",
	"cookie",
	"hmac",
	"password",
	"secret",
	"session",
	"signature",
	"token",
}

var sensitiveHeaderNames = map[string]struct{}{
	"Authorization":        {},
	"Cookie":               {},
	"Set-Cookie":           {},
	"X-Api-Key":            {},
	"X-Session-Token":      {},
	"Proxy-Authorization":  {},
	"X-Csrf-Token":         {},
	"X-Salesforce-Session": {},
}

func isSensitiveHeaderName(name string) bool {
	canonical := http.CanonicalHeaderKey(strings.TrimSpace(name))
	if canonical == "" {
		return true
	}
	if _, bad := sensitiveHeaderNames[canonical]; bad {
		return true
	}
	lower := strings.ToLower(canonical)
	for _, fragment := range sensitiveHeaderNameFragments {
		if strings.Contains(lower, fragment) {
			return true
		}
	}
	return false
}

func isDuplicateKey(err error) bool {
	if err == nil {
		return false
	}
	text := strings.ToLower(err.Error())
	return strings.Contains(text, "duplicate") || strings.Contains(text, "1062")
}

func requireUpdated(res sql.Result) error {
	n, err := res.RowsAffected()
	if err != nil {
		return err
	}
	if n == 0 {
		return ErrLeaseLost
	}
	return nil
}

func requireUpdatedCount(res sql.Result, want int64) error {
	n, err := res.RowsAffected()
	if err != nil {
		return err
	}
	if n != want {
		return ErrLeaseLost
	}
	return nil
}

type noRowsResult struct{}

func (noRowsResult) LastInsertId() (int64, error) { return 0, nil }
func (noRowsResult) RowsAffected() (int64, error) { return 0, nil }

func sqlPlaceholders(n int) string {
	if n <= 0 {
		return ""
	}
	parts := make([]string, n)
	for i := range parts {
		parts[i] = "?"
	}
	return strings.Join(parts, ", ")
}

func compactStrings(values []string) []string {
	out := make([]string, 0, len(values))
	seen := map[string]struct{}{}
	for _, value := range values {
		value = strings.TrimSpace(value)
		if value == "" {
			continue
		}
		if _, ok := seen[value]; ok {
			continue
		}
		seen[value] = struct{}{}
		out = append(out, value)
	}
	return out
}

type caseByID struct {
	clauses []string
	args    []interface{}
}

func newCaseByID() *caseByID {
	return &caseByID{}
}

func (c *caseByID) add(id string, value interface{}) {
	if c == nil {
		return
	}
	id = strings.TrimSpace(id)
	if id == "" {
		return
	}
	c.clauses = append(c.clauses, "WHEN ? THEN ?")
	c.args = append(c.args, id, value)
}

func (c *caseByID) sql(column string) string {
	if c == nil || len(c.clauses) == 0 {
		return column
	}
	return "CASE Id " + strings.Join(c.clauses, " ") + " ELSE " + column + " END"
}

func compactEnqueuedReceiptUpdates(updates []EnqueuedReceiptUpdate) []EnqueuedReceiptUpdate {
	out := make([]EnqueuedReceiptUpdate, 0, len(updates))
	seen := map[string]struct{}{}
	for _, update := range updates {
		update.ReceiptID = strings.TrimSpace(update.ReceiptID)
		update.ClaimToken = strings.TrimSpace(update.ClaimToken)
		if update.ReceiptID == "" || update.ClaimToken == "" {
			continue
		}
		if _, ok := seen[update.ReceiptID]; ok {
			continue
		}
		seen[update.ReceiptID] = struct{}{}
		out = append(out, update)
	}
	return out
}

func compactMaterializingReceiptUpdates(updates []MaterializingReceiptUpdate) []MaterializingReceiptUpdate {
	out := make([]MaterializingReceiptUpdate, 0, len(updates))
	seen := map[string]struct{}{}
	for _, update := range updates {
		update.ReceiptID = strings.TrimSpace(update.ReceiptID)
		update.ClaimToken = strings.TrimSpace(update.ClaimToken)
		if update.ReceiptID == "" || update.ClaimToken == "" {
			continue
		}
		if _, ok := seen[update.ReceiptID]; ok {
			continue
		}
		seen[update.ReceiptID] = struct{}{}
		out = append(out, update)
	}
	return out
}

func receiptClaimPairsWhere(updates []EnqueuedReceiptUpdate) (string, []interface{}) {
	parts := make([]string, 0, len(updates))
	args := make([]interface{}, 0, len(updates)*2)
	for _, update := range updates {
		parts = append(parts, "(Id = ? AND ClaimToken = ?)")
		args = append(args, strings.TrimSpace(update.ReceiptID), strings.TrimSpace(update.ClaimToken))
	}
	return strings.Join(parts, " OR "), args
}

func materializingClaimPairsWhere(updates []MaterializingReceiptUpdate) (string, []interface{}) {
	parts := make([]string, 0, len(updates))
	args := make([]interface{}, 0, len(updates)*2)
	for _, update := range updates {
		parts = append(parts, "(Id = ? AND ClaimToken = ?)")
		args = append(args, strings.TrimSpace(update.ReceiptID), strings.TrimSpace(update.ClaimToken))
	}
	return strings.Join(parts, " OR "), args
}

func truncateLastError(reason error) interface{} {
	if reason == nil {
		return nil
	}
	msg := reason.Error()
	const maxLastErrorLen = 2048
	if len(msg) > maxLastErrorLen {
		msg = msg[:maxLastErrorLen]
	}
	return msg
}

func nullableTime(t time.Time, ok bool) interface{} {
	if !ok {
		return nil
	}
	return t
}

func timePtr(v sql.NullTime) *time.Time {
	if !v.Valid {
		return nil
	}
	return &v.Time
}