From 404eac8a37bffc277c136fa8c25fc067063b276e Mon Sep 17 00:00:00 2001 From: Phloraxx Date: Fri, 11 Sep 2026 18:26:53 +0000 Subject: [PATCH] feat: harden burst payment reliability --- internal/v4/httpapi/relay.go | 25 ++++- internal/v4/httpapi/relay_test.go | 39 ++++++++ internal/v4/payments/allocator.go | 4 +- internal/v4/payments/allocator_test.go | 41 ++++++++- internal/v4/relay/service.go | 96 +++++++++++++++++-- internal/v4/storage/db_test.go | 65 ++++++++++++- internal/v4/storage/restore.go | 2 +- internal/v4/storage/schema.go | 123 ++++++++++++++++++++++++- 8 files changed, 371 insertions(+), 24 deletions(-) diff --git a/internal/v4/httpapi/relay.go b/internal/v4/httpapi/relay.go index 33e8bc0..62bb663 100644 --- a/internal/v4/httpapi/relay.go +++ b/internal/v4/httpapi/relay.go @@ -9,7 +9,10 @@ import ( "github.com/Phloraxx/payment-api/internal/v4/relay" ) -const relayHTTPBodyLimit = 64 << 10 +const ( + relayHTTPBodyLimit = 64 << 10 + relayHTTPBatchBodyLimit = 1 << 20 +) const ( relayDeviceHeader = "X-PayGate-Relay-Device" @@ -27,6 +30,7 @@ func NewRelayHandler(service *relay.Service) *RelayHandler { h := &RelayHandler{Relay: service, mux: http.NewServeMux()} h.mux.HandleFunc("POST /api/v4/relay/pair", h.pair) h.mux.HandleFunc("POST "+relay.EventPath, h.event) + h.mux.HandleFunc("POST "+relay.EventBatchPath, h.eventBatch) h.mux.HandleFunc("POST "+relay.HeartbeatPath, h.heartbeat) h.mux.HandleFunc("GET "+relay.DevicePath, h.getThisDevice) return h @@ -90,6 +94,19 @@ func (h *RelayHandler) event(w http.ResponseWriter, r *http.Request) { writeJSON(w, http.StatusOK, result) } +func (h *RelayHandler) eventBatch(w http.ResponseWriter, r *http.Request) { + raw, ok := readRelayBodyLimit(w, r, relayHTTPBatchBodyLimit) + if !ok { + return + } + result, err := h.Relay.IngestBatchSigned(r.Context(), relayAuth(r, relay.EventBatchPath), raw) + if err != nil { + writeRelayError(w, err) + return + } + writeJSON(w, http.StatusOK, result) +} + func (h *RelayHandler) heartbeat(w http.ResponseWriter, r *http.Request) { raw, ok := readRelayBody(w, r) if !ok { @@ -133,11 +150,15 @@ func relayAuth(r *http.Request, path string) relay.RequestAuth { } func readRelayBody(w http.ResponseWriter, r *http.Request) ([]byte, bool) { + return readRelayBodyLimit(w, r, relayHTTPBodyLimit) +} + +func readRelayBodyLimit(w http.ResponseWriter, r *http.Request, limit int64) ([]byte, bool) { if !isJSON(r.Header.Get("Content-Type")) { writeError(w, http.StatusUnsupportedMediaType, "invalid_content_type", "Content-Type must be application/json") return nil, false } - r.Body = http.MaxBytesReader(w, r.Body, relayHTTPBodyLimit) + r.Body = http.MaxBytesReader(w, r.Body, limit) raw, err := io.ReadAll(r.Body) if err != nil { writeError(w, http.StatusBadRequest, "invalid_request", "Relay body is too large or unreadable") diff --git a/internal/v4/httpapi/relay_test.go b/internal/v4/httpapi/relay_test.go index 915903e..7866367 100644 --- a/internal/v4/httpapi/relay_test.go +++ b/internal/v4/httpapi/relay_test.go @@ -193,6 +193,45 @@ func TestRelaySignedEventAndSignatureFailure(t *testing.T) { } } +func TestRelaySignedBatchAcceptsAndDeduplicatesEvents(t *testing.T) { + f := newRelayHTTPFixture(t) + epoch := pairRelayHTTP(t, f) + eventA := map[string]any{ + "schema_version": 1, "event_id": strings.Repeat("a", 64), + "package_name": "com.paytm.business", "posted_at_ms": f.now.UnixMilli(), + "title": "Payment Received on Paytm for Business", "text": "Received Rs. 100.00 from Alice", + } + eventB := map[string]any{ + "schema_version": 1, "event_id": strings.Repeat("b", 64), + "package_name": "com.paytm.business", "posted_at_ms": f.now.UnixMilli(), + "title": "Payment Received on Paytm for Business", "text": "Received Rs. 100.00 from Bob", + } + body, _ := json.Marshal(map[string]any{"schema_version": 1, "events": []any{eventA, eventB}}) + + rr := httptest.NewRecorder() + f.handler.ServeHTTP(rr, signedRelayRequest(t, f, relay.EventBatchPath, body, epoch)) + if rr.Code != http.StatusOK { + t.Fatalf("batch status=%d body=%s", rr.Code, rr.Body.String()) + } + if !strings.Contains(rr.Body.String(), `"event_id":"`+strings.Repeat("a", 64)+`"`) || + !strings.Contains(rr.Body.String(), `"event_id":"`+strings.Repeat("b", 64)+`"`) { + t.Fatalf("batch response missing source ids: %s", rr.Body.String()) + } + var count int + if err := f.db.SQL.QueryRow(`SELECT COUNT(*) FROM relay_events`).Scan(&count); err != nil || count != 2 { + t.Fatalf("relay events=%d err=%v", count, err) + } + + rr = httptest.NewRecorder() + f.handler.ServeHTTP(rr, signedRelayRequest(t, f, relay.EventBatchPath, body, epoch)) + if rr.Code != http.StatusOK || strings.Count(rr.Body.String(), `"duplicate":true`) != 2 { + t.Fatalf("batch retry status=%d body=%s", rr.Code, rr.Body.String()) + } + if err := f.db.SQL.QueryRow(`SELECT COUNT(*) FROM relay_events`).Scan(&count); err != nil || count != 2 { + t.Fatalf("relay events after retry=%d err=%v", count, err) + } +} + func TestRelayRejectsQueryParameters(t *testing.T) { f := newRelayHTTPFixture(t) req := httptest.NewRequest(http.MethodPost, "/api/v4/relay/pair?token=leak", strings.NewReader(`{}`)) diff --git a/internal/v4/payments/allocator.go b/internal/v4/payments/allocator.go index a4c3388..2e0510f 100644 --- a/internal/v4/payments/allocator.go +++ b/internal/v4/payments/allocator.go @@ -28,7 +28,7 @@ func NewAllocator() Allocator { return Allocator{ Random: cryptoRandomIndex, SoftHorizon: defaultSoftHorizon, - Buckets: 2, + Buckets: 6, } } @@ -44,7 +44,7 @@ func (a Allocator) Select(ctx context.Context, tx *storage.ImmediateTx, profileI } buckets := a.Buckets if buckets <= 0 { - buckets = 2 + buckets = 6 } randomIndex := a.Random if randomIndex == nil { diff --git a/internal/v4/payments/allocator_test.go b/internal/v4/payments/allocator_test.go index 60d8855..2035e3e 100644 --- a/internal/v4/payments/allocator_test.go +++ b/internal/v4/payments/allocator_test.go @@ -199,12 +199,14 @@ func TestAllocatorReleasesDueReservationBeforeChoosing(t *testing.T) { } } -func TestAllocatorReturnsCapacityOnlyWhenBothBucketsFull(t *testing.T) { +func TestAllocatorReturnsCapacityOnlyWhenAllBucketsFull(t *testing.T) { db := openAllocatorDB(t) ctx := context.Background() now := time.UnixMilli(1_788_200_000_000) - fillActiveRange(t, db.SQL, 10001, 10099, -1, now) - fillActiveRange(t, db.SQL, 10101, 10199, -1, now) + for bucket := int64(0); bucket < 6; bucket++ { + start := int64(10001) + bucket*100 + fillActiveRange(t, db.SQL, start, start+98, -1, now) + } allocator := NewAllocator() err := db.WithImmediateTx(ctx, func(tx *storage.ImmediateTx) error { @@ -216,6 +218,37 @@ func TestAllocatorReturnsCapacityOnlyWhenBothBucketsFull(t *testing.T) { } } +func TestAllocatorSupportsThreeHundredConcurrentPayments(t *testing.T) { + db := openAllocatorDB(t) + ctx := context.Background() + now := time.UnixMilli(1_788_200_000_000) + service := NewService(db) + service.Now = func() time.Time { return now } + service.Allocator.Random = fixedIndex(0) + + seen := make(map[int64]struct{}, 300) + for i := 0; i < 300; i++ { + result, err := service.Create(ctx, CreateInput{ + RequestedAmountPaise: 10000, + Name: fmt.Sprintf("Person %d", i), + ExternalID: "evt_capacity_300", + Metadata: []byte(`{}`), + IdempotencyScope: "load-test", + IdempotencyKey: fmt.Sprintf("payment-%03d", i), + }) + if err != nil { + t.Fatalf("create %d failed: %v", i+1, err) + } + if _, exists := seen[result.Payment.PayableAmountPaise]; exists { + t.Fatalf("duplicate payable amount %d at create %d", result.Payment.PayableAmountPaise, i+1) + } + seen[result.Payment.PayableAmountPaise] = struct{}{} + } + if len(seen) != 300 { + t.Fatalf("unique payable amounts=%d, want 300", len(seen)) + } +} + func TestAllocatorRejectsNonWholeRequestedAmount(t *testing.T) { db := openAllocatorDB(t) ctx := context.Background() @@ -274,7 +307,7 @@ func insertTestPayment(t *testing.T, db *sql.DB, id string, payable int64, creat t.Helper() requested := int64(10000) adjustment := payable - requested - if adjustment <= 0 || adjustment > 199 { + if adjustment <= 0 || adjustment > 599 { t.Fatalf("invalid test payable %d", payable) } now := created.UnixMilli() diff --git a/internal/v4/relay/service.go b/internal/v4/relay/service.go index 982b191..1558bbf 100644 --- a/internal/v4/relay/service.go +++ b/internal/v4/relay/service.go @@ -23,7 +23,10 @@ import ( const ( SchemaVersion = 1 EventPath = "/api/v4/relay/events" + EventBatchPath = "/api/v4/relay/events/batch" maxRawBodyBytes = 64 << 10 + maxBatchRawBodyBytes = 1 << 20 + maxBatchEvents = 50 maxNotificationTextBytes = 32 << 10 ) @@ -47,6 +50,7 @@ type EventInput struct { } type IngestResult struct { + EventID string `json:"event_id,omitempty"` RelayEventID string `json:"relay_event_id"` Status string `json:"status"` PaymentID string `json:"payment_id,omitempty"` @@ -54,6 +58,15 @@ type IngestResult struct { Transitioned bool `json:"transitioned,omitempty"` } +type BatchInput struct { + SchemaVersion int `json:"schema_version"` + Events []json.RawMessage `json:"events"` +} + +type BatchIngestResult struct { + Results []IngestResult `json:"results"` +} + func NewService(db *storage.DB, paymentService *payments.Service) *Service { return &Service{ DB: db, Payments: paymentService, Now: time.Now, NewID: randomID, @@ -168,9 +181,70 @@ func (s *Service) IngestSigned(ctx context.Context, auth RequestAuth, rawBody [] if err := validateEventInput(&input); err != nil { return IngestResult{}, err } - payloadHash := sha256.Sum256(rawBody) + return s.ingestVerifiedEvent(ctx, device, input, rawBody, now) +} + +func (s *Service) IngestBatchSigned(ctx context.Context, auth RequestAuth, rawBody []byte) (BatchIngestResult, error) { + if s == nil || s.DB == nil || s.DB.SQL == nil || s.Payments == nil { + return BatchIngestResult{}, errors.New("relay storage and payment service are required") + } + if len(rawBody) == 0 || len(rawBody) > maxBatchRawBodyBytes { + return BatchIngestResult{}, relayError("RELAY_BATCH_TOO_LARGE", "relay batch body is empty or too large", 400) + } + if strings.ToUpper(strings.TrimSpace(auth.Method)) != "POST" || auth.Path != EventBatchPath { + return BatchIngestResult{}, relayError("INVALID_RELAY_ENDPOINT", "relay signature is not for the v4 batch endpoint", 401) + } + nowFn := s.Now + if nowFn == nil { + nowFn = time.Now + } + now := nowFn().UTC() + device, err := verifyRequest(ctx, s.DB, auth, rawBody, now) + if err != nil { + return BatchIngestResult{}, err + } + var batch BatchInput + if err := json.Unmarshal(rawBody, &batch); err != nil { + return BatchIngestResult{}, relayError("INVALID_RELAY_BATCH", "relay batch is not valid JSON", 400) + } + if batch.SchemaVersion != SchemaVersion { + return BatchIngestResult{}, relayError("UNSUPPORTED_RELAY_SCHEMA", "schema_version must be 1", 400) + } + if len(batch.Events) == 0 || len(batch.Events) > maxBatchEvents { + return BatchIngestResult{}, relayError("INVALID_RELAY_BATCH", "relay batch must contain between 1 and 50 events", 400) + } + + inputs := make([]EventInput, len(batch.Events)) + for i, rawEvent := range batch.Events { + if len(rawEvent) == 0 || len(rawEvent) > maxRawBodyBytes { + return BatchIngestResult{}, relayError("RELAY_EVENT_TOO_LARGE", "relay batch contains an empty or oversized event", 400) + } + if err := json.Unmarshal(rawEvent, &inputs[i]); err != nil { + return BatchIngestResult{}, relayError("INVALID_RELAY_EVENT", "relay batch contains invalid event JSON", 400) + } + if err := validateEventInput(&inputs[i]); err != nil { + return BatchIngestResult{}, err + } + } + + out := BatchIngestResult{Results: make([]IngestResult, 0, len(inputs))} + for i, input := range inputs { + result, err := s.ingestVerifiedEvent(ctx, device, input, batch.Events[i], now) + if err != nil { + // Earlier items may already be durable. Retrying the entire signed batch + // is safe because source event IDs are idempotent per relay device. + return BatchIngestResult{}, err + } + out.Results = append(out.Results, result) + } + return out, nil +} + +func (s *Service) ingestVerifiedEvent(ctx context.Context, device verifiedDevice, input EventInput, rawEvent []byte, now time.Time) (IngestResult, error) { + payloadHash := sha256.Sum256(rawEvent) postedAt, postedReliable := sanitizePostedAt(input.PostedAtMS, now) result, inserted, err := s.acceptEvent(ctx, device, input, postedAt, postedReliable, now, payloadHash[:]) + result.EventID = input.EventID if err != nil || !inserted { return result, err } @@ -219,8 +293,6 @@ func (s *Service) IngestSigned(ctx context.Context, auth RequestAuth, rawBody [] } matched, err := s.Payments.ApplyObservationForRelay(ctx, result.RelayEventID, obs, now, device.ID, device.EnrolledAt) if errors.Is(err, payments.ErrRelayEventNotFound) { - // A retention worker or device revocation may have finalized this - // stale event while parsing was in flight. Do not apply its payload. if finishErr := s.finishIgnored(ctx, result.RelayEventID, errors.New("relay event was no longer authorized for processing")); finishErr != nil { return IngestResult{}, finishErr } @@ -364,7 +436,7 @@ func existingRelayEvent(ctx context.Context, tx *storage.ImmediateTx, deviceID, result.PaymentID = paymentID.String same := false if len(storedHash) == sha256.Size { - same = bytes.Equal(storedHash, payloadHash) + same = bytes.Equal(storedHash, payloadHash) || relayEventFieldsEqual(packageName, storedPostedAt, amountHint, title, text, bigText, in, postedAt, postedReliable) } else if bytes.HasPrefix(storedHash, []byte("legacy:")) { comparisonPostedAt := postedAt if !postedReliable { @@ -374,15 +446,19 @@ func existingRelayEvent(ctx context.Context, tx *storage.ImmediateTx, deviceID, } else if len(storedHash) == 0 { // Legacy rows without a hash retain their fields until this service // redacts them, so retries remain idempotent during migration. - same = packageName == in.PackageName && - (!postedReliable || storedPostedAt == postedAt.UnixMilli()) && - nullableAmountEqual(amountHint, in.AmountHintPaise) && - nullableTextEqual(title, in.Title) && - nullableTextEqual(text, in.Text) && - nullableTextEqual(bigText, in.BigText) + same = relayEventFieldsEqual(packageName, storedPostedAt, amountHint, title, text, bigText, in, postedAt, postedReliable) } return result, true, same, nil } +func relayEventFieldsEqual(packageName string, storedPostedAt int64, amountHint sql.NullInt64, title, text, bigText sql.NullString, in EventInput, postedAt time.Time, postedReliable bool) bool { + return packageName == in.PackageName && + (!postedReliable || storedPostedAt == postedAt.UnixMilli()) && + nullableAmountEqual(amountHint, in.AmountHintPaise) && + nullableTextEqual(title, in.Title) && + nullableTextEqual(text, in.Text) && + nullableTextEqual(bigText, in.BigText) +} + func legacyPayloadFingerprint(packageName string, postedAt int64, amountHint sql.NullInt64, title, text, bigText sql.NullString) []byte { data := make([]byte, 0, 128) diff --git a/internal/v4/storage/db_test.go b/internal/v4/storage/db_test.go index 9a256a6..e6c6a6c 100644 --- a/internal/v4/storage/db_test.go +++ b/internal/v4/storage/db_test.go @@ -3,6 +3,7 @@ package storage import ( "context" "database/sql" + "fmt" "path/filepath" "strings" "testing" @@ -232,7 +233,7 @@ func TestPaymentAmountAndStatusConstraints(t *testing.T) { }{ {"dot-zero-payable", 10000, 10100, 100, "pending", nil}, {"requested-has-paise", 10001, 10037, 36, "pending", nil}, - {"adjustment-too-large", 10000, 10201, 201, "pending", nil}, + {"adjustment-too-large", 10000, 10601, 601, "pending", nil}, {"invalid-status", 10000, 10037, 37, "review", nil}, {"paid-without-paid-at", 10000, 10037, 37, "paid", nil}, } @@ -247,9 +248,9 @@ func TestPaymentAmountAndStatusConstraints(t *testing.T) { } if _, err := db.SQL.Exec(`INSERT INTO payments(id,name,requested_amount_paise,payable_amount_paise,adjustment_paise,collection_profile_id,upi_id_snapshot,status,created_at,expires_at,grace_until,reuse_after) - VALUES('overflow_ok','Person',10000,10199,199,'paytm','merchant@paytm','pending',?,?,?,?)`, + VALUES('overflow_ok','Person',10000,10599,599,'paytm','merchant@paytm','pending',?,?,?,?)`, now, now+300_000, now+600_000, now+900_000); err != nil { - t.Fatalf("valid second-bucket amount rejected: %v", err) + t.Fatalf("valid sixth-bucket amount rejected: %v", err) } } @@ -448,6 +449,64 @@ func TestOpenMigratesV4CrossProfileDuplicateReservationsWithoutRewritingAmounts( } } +func TestOpenWidensHistoricalPaymentAdjustmentConstraint(t *testing.T) { + ctx := context.Background() + path := filepath.Join(t.TempDir(), "paygate-old-adjustment.db") + db, err := Open(ctx, path) + if err != nil { + t.Fatal(err) + } + if err := db.Close(); err != nil { + t.Fatal(err) + } + + raw, err := sql.Open("sqlite", "file:"+filepath.ToSlash(path)) + if err != nil { + t.Fatal(err) + } + var createSQL string + if err := raw.QueryRowContext(ctx, `SELECT sql FROM sqlite_schema WHERE type='table' AND name='payments'`).Scan(&createSQL); err != nil { + t.Fatal(err) + } + oldSQL := strings.Replace(createSQL, + "adjustment_paise INTEGER NOT NULL CHECK(adjustment_paise BETWEEN 1 AND 599)", + "adjustment_paise INTEGER NOT NULL CHECK(adjustment_paise BETWEEN 1 AND 199)", 1) + if oldSQL == createSQL { + t.Fatal("current payments schema did not contain widened adjustment constraint") + } + if _, err := raw.ExecContext(ctx, "PRAGMA writable_schema=ON"); err != nil { + t.Fatal(err) + } + if _, err := raw.ExecContext(ctx, `UPDATE sqlite_schema SET sql=? WHERE type='table' AND name='payments'`, oldSQL); err != nil { + t.Fatal(err) + } + var schemaNumber int + if err := raw.QueryRowContext(ctx, "PRAGMA schema_version").Scan(&schemaNumber); err != nil { + t.Fatal(err) + } + if _, err := raw.ExecContext(ctx, fmt.Sprintf("PRAGMA schema_version=%d", schemaNumber+1)); err != nil { + t.Fatal(err) + } + if _, err := raw.ExecContext(ctx, "PRAGMA writable_schema=OFF"); err != nil { + t.Fatal(err) + } + if err := raw.Close(); err != nil { + t.Fatal(err) + } + + db, err = Open(ctx, path) + if err != nil { + t.Fatalf("reopen historical constraint database: %v", err) + } + defer db.Close() + if err := db.SQL.QueryRowContext(ctx, `SELECT sql FROM sqlite_schema WHERE type='table' AND name='payments'`).Scan(&createSQL); err != nil { + t.Fatal(err) + } + if !strings.Contains(createSQL, "adjustment_paise BETWEEN 1 AND 599") { + t.Fatalf("payments constraint was not widened: %s", createSQL) + } +} + func TestOpenMigratesV1DatabaseToRelayHealthSchema(t *testing.T) { ctx := context.Background() path := filepath.Join(t.TempDir(), "paygate-v1.db") diff --git a/internal/v4/storage/restore.go b/internal/v4/storage/restore.go index a5bdb7d..3a613b7 100644 --- a/internal/v4/storage/restore.go +++ b/internal/v4/storage/restore.go @@ -392,7 +392,7 @@ func validateRestoreDatabase(ctx context.Context, path string) error { "PAYABLE_AMOUNT_PAISE > REQUESTED_AMOUNT_PAISE", "PAYABLE_AMOUNT_PAISE % 100 BETWEEN 1 AND 99", "PAYABLE_AMOUNT_PAISE = REQUESTED_AMOUNT_PAISE + ADJUSTMENT_PAISE", - "ADJUSTMENT_PAISE BETWEEN 1 AND 199", + "ADJUSTMENT_PAISE BETWEEN 1 AND 599", "CURRENCY = 'INR'", "JSON_VALID(METADATA_JSON)", "STATUS IN ('PENDING','PAID','EXPIRED','CANCELLED')", diff --git a/internal/v4/storage/schema.go b/internal/v4/storage/schema.go index c816489..a1ce566 100644 --- a/internal/v4/storage/schema.go +++ b/internal/v4/storage/schema.go @@ -3,6 +3,7 @@ package storage import ( "context" "database/sql" + "errors" "fmt" "strings" ) @@ -60,7 +61,125 @@ CREATE TABLE IF NOT EXISTS schema_migrations ( return err } } - return db.reconcileCompatibility(ctx) + if err := db.reconcileCompatibility(ctx); err != nil { + return err + } + return db.ensurePaymentAdjustmentCapacity(ctx) +} + +func (db *DB) ensurePaymentAdjustmentCapacity(ctx context.Context) error { + const oldConstraint = "adjustment_paise INTEGER NOT NULL CHECK(adjustment_paise BETWEEN 1 AND 199)" + const newConstraint = "adjustment_paise INTEGER NOT NULL CHECK(adjustment_paise BETWEEN 1 AND 599)" + var createSQL string + if err := db.SQL.QueryRowContext(ctx, `SELECT sql FROM sqlite_schema WHERE type='table' AND name='payments'`).Scan(&createSQL); err != nil { + return fmt.Errorf("read payments schema: %w", err) + } + if strings.Contains(createSQL, newConstraint) { + return nil + } + if !strings.Contains(createSQL, oldConstraint) { + return fmt.Errorf("payments adjustment constraint is incompatible") + } + + conn, err := db.SQL.Conn(ctx) + if err != nil { + return fmt.Errorf("acquire payments schema connection: %w", err) + } + defer conn.Close() + if _, err := conn.ExecContext(ctx, "PRAGMA foreign_keys=OFF"); err != nil { + return fmt.Errorf("disable foreign keys for payments rebuild: %w", err) + } + foreignKeysDisabled := true + committed := false + defer func() { + if !committed { + _, _ = conn.ExecContext(context.Background(), "ROLLBACK") + } + if foreignKeysDisabled { + _, _ = conn.ExecContext(context.Background(), "PRAGMA foreign_keys=ON") + } + }() + if _, err := conn.ExecContext(ctx, "BEGIN IMMEDIATE"); err != nil { + return wrapTransactionError("begin payments constraint migration", err) + } + + const rebuild = ` +DROP TRIGGER IF EXISTS amount_reservations_payment_consistency_insert; +DROP TRIGGER IF EXISTS amount_reservations_payment_consistency_update; +DROP TRIGGER IF EXISTS payments_reservation_consistency_update; +CREATE TABLE payments_capacity_v4 ( + id TEXT PRIMARY KEY, + name TEXT NOT NULL CHECK(length(trim(name)) BETWEEN 1 AND 120), + external_id TEXT, + metadata_json TEXT NOT NULL DEFAULT '{}' CHECK(json_valid(metadata_json)), + requested_amount_paise INTEGER NOT NULL CHECK(requested_amount_paise > 0 AND requested_amount_paise % 100 = 0), + payable_amount_paise INTEGER NOT NULL CHECK(payable_amount_paise > requested_amount_paise AND payable_amount_paise % 100 BETWEEN 1 AND 99), + adjustment_paise INTEGER NOT NULL CHECK(adjustment_paise BETWEEN 1 AND 599), + currency TEXT NOT NULL DEFAULT 'INR' CHECK(currency = 'INR'), + collection_profile_id TEXT NOT NULL REFERENCES collection_profiles(id) ON UPDATE RESTRICT ON DELETE RESTRICT, + upi_id_snapshot TEXT NOT NULL, + payee_name_snapshot TEXT, + status TEXT NOT NULL CHECK(status IN ('pending','paid','expired','cancelled')), + created_at INTEGER NOT NULL, + expires_at INTEGER NOT NULL, + grace_until INTEGER NOT NULL, + reuse_after INTEGER NOT NULL, + paid_at INTEGER, + payer_name TEXT, + payer_upi_id TEXT, + internal_note TEXT, + CHECK(payable_amount_paise = requested_amount_paise + adjustment_paise), + CHECK(created_at < expires_at AND expires_at < grace_until AND grace_until < reuse_after), + CHECK((status = 'paid' AND paid_at IS NOT NULL) OR (status != 'paid' AND paid_at IS NULL)) +) STRICT; +INSERT INTO payments_capacity_v4( + id,name,external_id,metadata_json,requested_amount_paise,payable_amount_paise,adjustment_paise,currency, + collection_profile_id,upi_id_snapshot,payee_name_snapshot,status,created_at,expires_at,grace_until,reuse_after, + paid_at,payer_name,payer_upi_id,internal_note) +SELECT id,name,external_id,metadata_json,requested_amount_paise,payable_amount_paise,adjustment_paise,currency, + collection_profile_id,upi_id_snapshot,payee_name_snapshot,status,created_at,expires_at,grace_until,reuse_after, + paid_at,payer_name,payer_upi_id,internal_note FROM payments; +DROP TABLE payments; +ALTER TABLE payments_capacity_v4 RENAME TO payments; +CREATE INDEX idx_payments_external_id ON payments(external_id); +CREATE INDEX idx_payments_status_created ON payments(status, created_at DESC); +CREATE INDEX idx_payments_profile_payable ON payments(collection_profile_id, payable_amount_paise); +` + if _, err := conn.ExecContext(ctx, rebuild); err != nil { + return fmt.Errorf("rebuild payments for expanded adjustment capacity: %w", err) + } + if _, err := conn.ExecContext(ctx, relayPayloadIntegritySQL); err != nil { + return fmt.Errorf("restore payment reservation integrity triggers: %w", err) + } + if _, err := conn.ExecContext(ctx, "COMMIT"); err != nil { + return wrapTransactionError("commit payments constraint migration", err) + } + committed = true + if _, err := conn.ExecContext(ctx, "PRAGMA foreign_keys=ON"); err != nil { + return fmt.Errorf("restore foreign keys after payments rebuild: %w", err) + } + foreignKeysDisabled = false + var fkEnabled int + if err := conn.QueryRowContext(ctx, "PRAGMA foreign_keys").Scan(&fkEnabled); err != nil || fkEnabled != 1 { + return fmt.Errorf("foreign keys not restored after payments rebuild: enabled=%d err=%w", fkEnabled, err) + } + + var check string + if err := conn.QueryRowContext(ctx, "PRAGMA integrity_check").Scan(&check); err != nil { + return fmt.Errorf("check sqlite integrity after payments migration: %w", err) + } + if check != "ok" { + return fmt.Errorf("sqlite integrity check failed after payments migration: %s", check) + } + rowsFK, err := conn.QueryContext(ctx, "PRAGMA foreign_key_check") + if err != nil { + return fmt.Errorf("check foreign keys after payments migration: %w", err) + } + defer rowsFK.Close() + if rowsFK.Next() { + return errors.New("foreign key check failed after payments migration") + } + return rowsFK.Err() } func readSchemaVersions(ctx context.Context, queryer schemaQueryer) ([]int, error) { @@ -511,7 +630,7 @@ CREATE TABLE payments ( metadata_json TEXT NOT NULL DEFAULT '{}' CHECK(json_valid(metadata_json)), requested_amount_paise INTEGER NOT NULL CHECK(requested_amount_paise > 0 AND requested_amount_paise % 100 = 0), payable_amount_paise INTEGER NOT NULL CHECK(payable_amount_paise > requested_amount_paise AND payable_amount_paise % 100 BETWEEN 1 AND 99), - adjustment_paise INTEGER NOT NULL CHECK(adjustment_paise BETWEEN 1 AND 199), + adjustment_paise INTEGER NOT NULL CHECK(adjustment_paise BETWEEN 1 AND 599), currency TEXT NOT NULL DEFAULT 'INR' CHECK(currency = 'INR'), collection_profile_id TEXT NOT NULL REFERENCES collection_profiles(id) ON UPDATE RESTRICT ON DELETE RESTRICT, upi_id_snapshot TEXT NOT NULL,