From 400eec2da4811423791e64177e8359dd2d71194e Mon Sep 17 00:00:00 2001 From: Phloraxx Date: Fri, 28 Aug 2026 07:20:20 +0000 Subject: [PATCH 01/36] Gate relay readiness on power health --- internal/androidrelay/status.go | 145 +++++++++++++----- internal/androidrelay/status_test.go | 78 ++++++++++ .../20260828010000_relay_power_health.go | 42 +++++ migrations/migration_test.go | 9 ++ web/src/pages/Dashboard.tsx | 2 +- web/src/pages/Settings.tsx | 2 + web/src/types.ts | 8 + 7 files changed, 247 insertions(+), 39 deletions(-) create mode 100644 migrations/20260828010000_relay_power_health.go diff --git a/internal/androidrelay/status.go b/internal/androidrelay/status.go index 3d7e591..855d259 100644 --- a/internal/androidrelay/status.go +++ b/internal/androidrelay/status.go @@ -1,6 +1,7 @@ package androidrelay import ( + "fmt" "strings" "time" @@ -19,6 +20,10 @@ type HeartbeatInput struct { DeviceModel string `json:"deviceModel"` NotificationAccess bool `json:"notificationAccess"` ListenerConnected bool `json:"listenerConnected"` + BatteryOptimizationExempt *bool `json:"batteryOptimizationExempt"` + PowerSaveMode *bool `json:"powerSaveMode"` + BackgroundRestricted *bool `json:"backgroundRestricted"` + ForegroundService *bool `json:"foregroundService"` PendingCount int `json:"pendingCount"` FailedCount int `json:"failedCount"` LastSuccessfulDeliveryAtMs int64 `json:"lastSuccessfulDeliveryAtMs"` @@ -32,42 +37,49 @@ type HeartbeatResult struct { } type DeviceStatus struct { - ID string `json:"id"` - DeviceID string `json:"deviceId"` - Name string `json:"name"` - Enabled bool `json:"enabled"` - AppVersion string `json:"appVersion"` - AndroidVersion string `json:"androidVersion"` - DeviceModel string `json:"deviceModel"` - LastSeenAt any `json:"lastSeenAt"` - LastHeartbeatAt any `json:"lastHeartbeatAt"` - HeartbeatGraceUntil any `json:"heartbeatGraceUntil"` - NotificationAccess bool `json:"notificationAccess"` - ListenerConnected bool `json:"listenerConnected"` - PendingCount int `json:"pendingCount"` - FailedCount int `json:"failedCount"` - LastClientError string `json:"lastClientError,omitempty"` - LastDeliveryAt any `json:"lastDeliveryAt"` - LastEventAt any `json:"lastEventAt"` - LastMatchedAt any `json:"lastMatchedAt"` - LastMatchedPaymentID string `json:"lastMatchedPaymentId,omitempty"` - RecentErrorCount int64 `json:"recentErrorCount"` - Active bool `json:"active"` + ID string `json:"id"` + DeviceID string `json:"deviceId"` + Name string `json:"name"` + Enabled bool `json:"enabled"` + AppVersion string `json:"appVersion"` + AndroidVersion string `json:"androidVersion"` + DeviceModel string `json:"deviceModel"` + LastSeenAt any `json:"lastSeenAt"` + LastHeartbeatAt any `json:"lastHeartbeatAt"` + HeartbeatGraceUntil any `json:"heartbeatGraceUntil"` + NotificationAccess bool `json:"notificationAccess"` + ListenerConnected bool `json:"listenerConnected"` + PowerHealthReported bool `json:"powerHealthReported"` + BatteryOptimizationExempt bool `json:"batteryOptimizationExempt"` + PowerSaveMode bool `json:"powerSaveMode"` + BackgroundRestricted bool `json:"backgroundRestricted"` + ForegroundService bool `json:"foregroundService"` + PowerHealthy bool `json:"powerHealthy"` + PendingCount int `json:"pendingCount"` + FailedCount int `json:"failedCount"` + LastClientError string `json:"lastClientError,omitempty"` + LastDeliveryAt any `json:"lastDeliveryAt"` + LastEventAt any `json:"lastEventAt"` + LastMatchedAt any `json:"lastMatchedAt"` + LastMatchedPaymentID string `json:"lastMatchedPaymentId,omitempty"` + RecentErrorCount int64 `json:"recentErrorCount"` + Active bool `json:"active"` } type Status struct { - Ready bool `json:"ready"` - EnabledDevices int `json:"enabledDevices"` - ActiveDevices int `json:"activeDevices"` - LegacyGraceDevices int `json:"legacyGraceDevices"` - StaleAfterSeconds int64 `json:"staleAfterSeconds"` - LastSeenAt any `json:"lastSeenAt"` - LastHeartbeatAt any `json:"lastHeartbeatAt"` - LastEventAt any `json:"lastEventAt"` - LastMatchedAt any `json:"lastMatchedAt"` - RecentErrorCount int64 `json:"recentErrorCount"` - PendingQueueCount int `json:"pendingQueueCount"` - FailedQueueCount int `json:"failedQueueCount"` + Ready bool `json:"ready"` + EnabledDevices int `json:"enabledDevices"` + ActiveDevices int `json:"activeDevices"` + LegacyGraceDevices int `json:"legacyGraceDevices"` + StaleAfterSeconds int64 `json:"staleAfterSeconds"` + LastSeenAt any `json:"lastSeenAt"` + LastHeartbeatAt any `json:"lastHeartbeatAt"` + LastEventAt any `json:"lastEventAt"` + LastMatchedAt any `json:"lastMatchedAt"` + RecentErrorCount int64 `json:"recentErrorCount"` + PendingQueueCount int `json:"pendingQueueCount"` + FailedQueueCount int `json:"failedQueueCount"` + PowerUnhealthyDevices int `json:"powerUnhealthyDevices"` } func (s *Service) Heartbeat(device *core.Record, in HeartbeatInput) (HeartbeatResult, error) { @@ -96,6 +108,19 @@ func (s *Service) Heartbeat(device *core.Record, in HeartbeatInput) (HeartbeatRe device.Set("device_model", trimMax(in.DeviceModel, 255)) device.Set("notification_access", in.NotificationAccess) device.Set("listener_connected", in.ListenerConnected) + if in.BatteryOptimizationExempt != nil { + device.Set("battery_optimization_exempt", *in.BatteryOptimizationExempt) + } + if in.PowerSaveMode != nil { + device.Set("power_save_mode", *in.PowerSaveMode) + } + if in.BackgroundRestricted != nil { + device.Set("background_restricted", *in.BackgroundRestricted) + } + if in.ForegroundService != nil { + device.Set("foreground_service_active", *in.ForegroundService) + } + device.Set("power_health_reported", in.BatteryOptimizationExempt != nil && in.PowerSaveMode != nil && in.BackgroundRestricted != nil && in.ForegroundService != nil) device.Set("pending_count", in.PendingCount) device.Set("failed_count", in.FailedCount) device.Set("last_client_error", trimMax(in.LastClientError, 1024)) @@ -128,8 +153,7 @@ func (s *Service) ReadyInApp(app core.App, staleAfter time.Duration) (bool, erro } continue } - seen := device.GetDateTime("last_seen_at").Time() - if !seen.IsZero() && !seen.Before(cutoff) && device.GetBool("notification_access") && device.GetBool("listener_connected") { + if relayDeviceCurrentReady(device, cutoff) { return true, nil } } @@ -164,9 +188,12 @@ func (s *Service) Status(staleAfter time.Duration) (Status, error) { status.ActiveDevices++ status.LegacyGraceDevices++ } - } else if !seen.IsZero() && !seen.Before(cutoff) && device.GetBool("notification_access") && device.GetBool("listener_connected") { + } else if relayDeviceCurrentReady(device, cutoff) { status.ActiveDevices++ } + if !relayDevicePowerReady(device) { + status.PowerUnhealthyDevices++ + } status.PendingQueueCount += device.GetInt("pending_count") status.FailedQueueCount += device.GetInt("failed_count") } @@ -204,7 +231,7 @@ func (s *Service) Devices(staleAfter time.Duration) ([]DeviceStatus, error) { heartbeat := record.GetDateTime("last_heartbeat_at").Time() graceUntil := record.GetDateTime("heartbeat_grace_until").Time() legacyGraceActive := heartbeat.IsZero() && !graceUntil.IsZero() && s.now().Before(graceUntil) - listenerOK := !heartbeat.IsZero() && record.GetBool("notification_access") && record.GetBool("listener_connected") + powerHealthy := relayDevicePowerReady(record) lastEventAt, lastMatchedAt, lastMatchedPaymentID, recentErrorCount, statusErr := s.deviceEventStatus(record.Id, s.now()) if statusErr != nil { return nil, statusErr @@ -213,10 +240,11 @@ func (s *Service) Devices(staleAfter time.Duration) ([]DeviceStatus, error) { ID: record.Id, DeviceID: record.GetString("device_id"), Name: record.GetString("name"), Enabled: record.GetBool("enabled"), AppVersion: record.GetString("app_version"), AndroidVersion: record.GetString("android_version"), DeviceModel: record.GetString("device_model"), LastSeenAt: timeValue(seen), LastHeartbeatAt: timeValue(heartbeat), HeartbeatGraceUntil: timeValue(graceUntil), NotificationAccess: record.GetBool("notification_access"), ListenerConnected: record.GetBool("listener_connected"), + PowerHealthReported: record.GetBool("power_health_reported"), BatteryOptimizationExempt: record.GetBool("battery_optimization_exempt"), PowerSaveMode: record.GetBool("power_save_mode"), BackgroundRestricted: record.GetBool("background_restricted"), ForegroundService: record.GetBool("foreground_service_active"), PowerHealthy: powerHealthy, PendingCount: record.GetInt("pending_count"), FailedCount: record.GetInt("failed_count"), LastClientError: record.GetString("last_client_error"), LastDeliveryAt: timeValue(record.GetDateTime("last_client_delivery_at").Time()), LastEventAt: lastEventAt, LastMatchedAt: lastMatchedAt, LastMatchedPaymentID: lastMatchedPaymentID, RecentErrorCount: recentErrorCount, - Active: record.GetBool("enabled") && (legacyGraceActive || (!seen.IsZero() && !seen.Before(cutoff) && listenerOK)), + Active: record.GetBool("enabled") && (legacyGraceActive || relayDeviceCurrentReady(record, cutoff)), }) } return result, nil @@ -273,6 +301,47 @@ func (s *Service) SetEnabledInApp(app core.App, recordID string, enabled bool) ( return record, nil } +func relayDeviceCurrentReady(record *core.Record, cutoff time.Time) bool { + if record == nil || record.GetDateTime("last_heartbeat_at").Time().IsZero() { + return false + } + seen := record.GetDateTime("last_seen_at").Time() + return !seen.IsZero() && !seen.Before(cutoff) && + record.GetBool("notification_access") && record.GetBool("listener_connected") && + relayDevicePowerReady(record) +} + +func relayDevicePowerReady(record *core.Record) bool { + if record == nil || !requiresPowerHealth(record.GetString("app_version")) { + return true + } + return record.GetBool("power_health_reported") && + record.GetBool("battery_optimization_exempt") && + !record.GetBool("background_restricted") && + record.GetBool("foreground_service_active") +} + +func requiresPowerHealth(version string) bool { + version = strings.TrimSpace(strings.TrimPrefix(strings.ToLower(version), "v")) + parts := strings.SplitN(version, "-", 2) + version = parts[0] + numbers := strings.Split(version, ".") + if len(numbers) < 3 { + return false + } + major, minor, patch := 0, 0, 0 + if _, err := fmt.Sscanf(numbers[0]+"."+numbers[1]+"."+numbers[2], "%d.%d.%d", &major, &minor, &patch); err != nil { + return false + } + if major != 0 { + return major > 0 + } + if minor != 3 { + return minor > 3 + } + return patch >= 1 +} + func normalizeStaleAfter(value time.Duration) time.Duration { if value <= 0 { return defaultStaleAfter diff --git a/internal/androidrelay/status_test.go b/internal/androidrelay/status_test.go index 019f4fa..523c265 100644 --- a/internal/androidrelay/status_test.go +++ b/internal/androidrelay/status_test.go @@ -242,3 +242,81 @@ func TestRelayStatusMarksStaleDeviceInactive(t *testing.T) { t.Fatalf("stale device status = %+v", status) } } + +func TestV031PowerHealthGatesReadinessButAllowsPowerSaver(t *testing.T) { + app, err := tests.NewTestApp() + if err != nil { + t.Fatal(err) + } + defer app.Cleanup() + now := time.Date(2026, 8, 28, 7, 30, 0, 0, time.UTC) + service := NewService(app, nil) + service.Now = func() time.Time { return now } + collection, _ := app.FindCollectionByNameOrId("relay_devices") + device := core.NewRecord(collection) + device.Set("device_id", "abababababababababababababababababababababababababababababababab") + device.Set("name", "Always-on phone") + device.Set("public_key_pem", "test-key") + device.Set("enabled", true) + if err := app.Save(device); err != nil { + t.Fatal(err) + } + + heartbeat := func(exempt, saver, restricted, foreground bool) { + t.Helper() + if _, err := service.Heartbeat(device, HeartbeatInput{ + SchemaVersion: 1, + AppVersion: "0.3.1", + NotificationAccess: true, + ListenerConnected: true, + BatteryOptimizationExempt: boolPointer(exempt), + PowerSaveMode: boolPointer(saver), + BackgroundRestricted: boolPointer(restricted), + ForegroundService: boolPointer(foreground), + }); err != nil { + t.Fatal(err) + } + } + + heartbeat(true, true, false, true) + status, err := service.Status(time.Hour) + if err != nil { + t.Fatal(err) + } + if !status.Ready || status.ActiveDevices != 1 || status.PowerUnhealthyDevices != 0 { + t.Fatalf("power saver should remain ready when v0.3.1 is exempt and foreground: %+v", status) + } + devices, err := service.Devices(time.Hour) + if err != nil { + t.Fatal(err) + } + if len(devices) != 1 || !devices[0].PowerHealthReported || !devices[0].PowerHealthy || !devices[0].BatteryOptimizationExempt || !devices[0].PowerSaveMode || !devices[0].ForegroundService { + t.Fatalf("unexpected power health: %+v", devices) + } + + heartbeat(false, true, false, true) + status, _ = service.Status(time.Hour) + if status.Ready || status.PowerUnhealthyDevices != 1 { + t.Fatalf("battery-optimized v0.3.1 must fail closed: %+v", status) + } + + heartbeat(true, true, true, true) + status, _ = service.Status(time.Hour) + if status.Ready { + t.Fatalf("background-restricted v0.3.1 must fail closed: %+v", status) + } + + heartbeat(true, true, false, false) + status, _ = service.Status(time.Hour) + if status.Ready { + t.Fatalf("v0.3.1 without foreground runtime must fail closed: %+v", status) + } + + heartbeat(true, true, false, true) + status, _ = service.Status(time.Hour) + if !status.Ready { + t.Fatalf("healthy always-on state should recover readiness: %+v", status) + } +} + +func boolPointer(value bool) *bool { return &value } diff --git a/migrations/20260828010000_relay_power_health.go b/migrations/20260828010000_relay_power_health.go new file mode 100644 index 0000000..a6ef675 --- /dev/null +++ b/migrations/20260828010000_relay_power_health.go @@ -0,0 +1,42 @@ +package migrations + +import ( + "github.com/pocketbase/pocketbase/core" + pbmigrations "github.com/pocketbase/pocketbase/migrations" +) + +func init() { + pbmigrations.Register(func(app core.App) error { + devices, err := app.FindCollectionByNameOrId("relay_devices") + if err != nil { + return err + } + for _, name := range []string{ + "power_health_reported", + "battery_optimization_exempt", + "power_save_mode", + "background_restricted", + "foreground_service_active", + } { + if devices.Fields.GetByName(name) == nil { + devices.Fields.Add(&core.BoolField{Name: name}) + } + } + return app.Save(devices) + }, func(app core.App) error { + devices, err := app.FindCollectionByNameOrId("relay_devices") + if err != nil { + return nil + } + for _, name := range []string{ + "power_health_reported", + "battery_optimization_exempt", + "power_save_mode", + "background_restricted", + "foreground_service_active", + } { + devices.Fields.RemoveByName(name) + } + return app.Save(devices) + }) +} diff --git a/migrations/migration_test.go b/migrations/migration_test.go index 1da3bfe..3ed3f0d 100644 --- a/migrations/migration_test.go +++ b/migrations/migration_test.go @@ -58,4 +58,13 @@ func TestDomainCollectionsOnlyExposeReadsToOperatorUsers(t *testing.T) { if reviews.Fields.GetByName("email_event") == nil { t.Fatal("review_cases.email_event migration field is missing") } + relayDevices, err := app.FindCollectionByNameOrId("relay_devices") + if err != nil { + t.Fatal(err) + } + for _, name := range []string{"power_health_reported", "battery_optimization_exempt", "power_save_mode", "background_restricted", "foreground_service_active"} { + if relayDevices.Fields.GetByName(name) == nil { + t.Fatalf("relay_devices.%s migration field is missing", name) + } + } } diff --git a/web/src/pages/Dashboard.tsx b/web/src/pages/Dashboard.tsx index ecb79f4..a6570b3 100644 --- a/web/src/pages/Dashboard.tsx +++ b/web/src/pages/Dashboard.tsx @@ -58,7 +58,7 @@ export function Dashboard() {

ANDROID RELAY

{relay?.ready ? "ready" : "unavailable"}

-

{relay ? `${relay.activeDevices}/${relay.enabledDevices} active · last heartbeat ${formatDate(relay.lastHeartbeatAt ?? undefined)} · queue ${relay.pendingQueueCount} pending / ${relay.failedQueueCount} failed · ${relay.recentErrorCount} server errors/24h` : "Relay status unavailable"}

+

{relay ? `${relay.activeDevices}/${relay.enabledDevices} active · ${relay.powerUnhealthyDevices} power-unhealthy · last heartbeat ${formatDate(relay.lastHeartbeatAt ?? undefined)} · queue ${relay.pendingQueueCount} pending / ${relay.failedQueueCount} failed · ${relay.recentErrorCount} server errors/24h` : "Relay status unavailable"}

diff --git a/web/src/pages/Settings.tsx b/web/src/pages/Settings.tsx index 678c861..9c3f9e4 100644 --- a/web/src/pages/Settings.tsx +++ b/web/src/pages/Settings.tsx @@ -317,6 +317,7 @@ export function Settings({ notify }: { notify: (value: string) => void }) {

Paytm QR checkouts fail closed when no recently active relay device is available.

Active devices
{relay ? `${relay.activeDevices} / ${relay.enabledDevices}` : "—"}
+
Power unhealthy
{relay?.powerUnhealthyDevices ?? 0}
Last heartbeat
{formatDate(relay?.lastHeartbeatAt ?? undefined)}
Last relay event
{formatDate(relay?.lastEventAt ?? undefined)}
Last matched payment
{formatDate(relay?.lastMatchedAt ?? undefined)}
@@ -331,6 +332,7 @@ export function Settings({ notify }: { notify: (value: string) => void }) { {device.deviceModel || "Android"} · app {device.appVersion || "unknown"} · fingerprint {device.deviceId ? `${device.deviceId.slice(0, 12)}…` : "unknown"} Last seen {formatDate(device.lastSeenAt ?? undefined)} · phone delivered {formatDate(device.lastDeliveryAt ?? undefined)} · last event {formatDate(device.lastEventAt ?? undefined)} · last match {formatDate(device.lastMatchedAt ?? undefined)}{!device.lastHeartbeatAt && device.heartbeatGraceUntil ? ` · legacy heartbeat grace until ${formatDate(device.heartbeatGraceUntil)}` : ""} Notifications {device.notificationAccess ? "allowed" : device.lastHeartbeatAt ? "blocked" : "not reported"} · listener {device.listenerConnected ? "connected" : device.lastHeartbeatAt ? "disconnected" : "not reported"} · queue {device.pendingCount} pending / {device.failedCount} failed · {device.recentErrorCount} server errors/24h{device.lastClientError ? ` · ${device.lastClientError}` : ""} + Power {device.powerHealthReported ? (device.powerHealthy ? "ready" : "NOT ready") : "not required/reported"} · battery {device.batteryOptimizationExempt ? "unrestricted" : device.powerHealthReported ? "optimized" : "unknown"} · foreground {device.foregroundService ? "active" : device.powerHealthReported ? "inactive" : "unknown"} · saver {device.powerSaveMode ? "on" : "off"} · background {device.backgroundRestricted ? "RESTRICTED" : "allowed"} diff --git a/web/src/types.ts b/web/src/types.ts index f1ed2b4..515e878 100644 --- a/web/src/types.ts +++ b/web/src/types.ts @@ -37,6 +37,8 @@ export type RelayStatus = { ready: boolean; enabledDevices: number; activeDevices: number; + legacyGraceDevices: number; + powerUnhealthyDevices: number; staleAfterSeconds: number; lastSeenAt?: string | null; lastHeartbeatAt?: string | null; @@ -60,6 +62,12 @@ export type RelayDevice = { heartbeatGraceUntil?: string | null; notificationAccess: boolean; listenerConnected: boolean; + powerHealthReported: boolean; + batteryOptimizationExempt: boolean; + powerSaveMode: boolean; + backgroundRestricted: boolean; + foregroundService: boolean; + powerHealthy: boolean; pendingCount: number; failedCount: number; lastClientError?: string; From 87a4a53eccf36e22bff3a65ffe597c14a555c5aa Mon Sep 17 00:00:00 2001 From: Phloraxx Date: Fri, 28 Aug 2026 12:02:38 +0000 Subject: [PATCH 02/36] Document PayGate v2 architecture and rollout --- docs/v2/00_MASTER_PLAN.md | 122 ++++++++++++++++++ docs/v2/01_TARGET_ARCHITECTURE.md | 187 ++++++++++++++++++++++++++++ docs/v2/02_MIGRATION_AND_ROLLOUT.md | 111 +++++++++++++++++ 3 files changed, 420 insertions(+) create mode 100644 docs/v2/00_MASTER_PLAN.md create mode 100644 docs/v2/01_TARGET_ARCHITECTURE.md create mode 100644 docs/v2/02_MIGRATION_AND_ROLLOUT.md diff --git a/docs/v2/00_MASTER_PLAN.md b/docs/v2/00_MASTER_PLAN.md new file mode 100644 index 0000000..5ff7d2b --- /dev/null +++ b/docs/v2/00_MASTER_PLAN.md @@ -0,0 +1,122 @@ +# PayGate v2 — Master Rewrite Plan + +Status: implementation branch, not production +Canonical branches: +- API: `rewrite/paygate-v2-foundation` +- Checkout web: `rewrite/paygate-v2-ui` +- Android: `rewrite/paygate-v2-mobile` + +## Goal + +Rewrite PayGate into a smaller, typed, modular payment system without weakening any existing financial invariant. The rewrite must reduce framework coupling, duplicated evidence logic, duplicated Razorpay code, stringly-typed status handling, oversized transactions, UI complexity, and operator friction. + +PayGate v2 remains a modular monolith. It does **not** introduce microservices, Kafka, Redis, distributed transactions, or a second primary database simply for architectural fashion. + +## Non-negotiable payment invariants + +1. Money is stored and compared as integer paise only. +2. Direct-UPI matching is exact amount + payment account + occurrence-time constrained. +3. Evidence identity is globally unique where the source provides a stable bank/reference identity. +4. Notification evidence must have a unique non-empty signed evidence reference. +5. Two or more plausible payment candidates is always an ambiguous/fail-closed result. +6. Evidence predating payment creation beyond the accepted tolerance can never pay that payment. +7. Amount fingerprints remain quarantined after expiry, cancellation, payment, or late payment. +8. Idempotency replay is resolved before new-payment readiness checks. +9. Payment mutation and durable outbound event creation are atomic. +10. External network calls never happen inside payment database transactions. +## Product surfaces + +PayGate v2 is one product with three coordinated surfaces: + +### 1. Checkout web +A public, mobile-first payment experience whose only job is to create and complete a payment safely. It should expose the minimum number of decisions, use progressive disclosure, and make the exact amount and payment state unmistakable. + +### 2. Operator web +A typed administrative console for payments, reviews, refunds, reconciliation, alerts, relay health, capacity, backups, and settings. It must stop querying PocketBase collections directly. + +### 3. PayGate Android app +The existing relay becomes the PayGate mobile application. The proven notification listener, durable local queue, WorkManager recovery, boot receiver, foreground runtime service, wake-lock-bounded delivery, and signed relay protocol remain. The UI becomes a modern operator app for health, payments, reviews and alerts. + +## Architectural direction + +The target dependency direction is: + +`HTTP / source adapters -> application services -> typed domain -> repositories/UoW -> SQLite adapter` + +Business code must not depend on PocketBase `core.Record`, record string keys, or `core.App` transaction objects. PocketBase may remain temporarily behind repository adapters during migration. + +All payment evidence sources normalize into one domain `Evidence` model and one matching engine. SMS, authenticated email, Paytm notification, reconciliation and future sources are adapters, not parallel payment engines. +## Rewrite workstreams + +### A. Domain and persistence +- Introduce typed `Money`, `Payment`, `Evidence`, `EvidenceRef`, `MatchOutcome`, `Refund`, `RelayHealth`, and typed status/ID values. +- Add repository interfaces and an explicit unit-of-work abstraction. +- Move PocketBase record mapping behind storage adapters. +- Remove application-layer `...InApp` variants as repository/UoW adoption reaches each module. +- Keep SQLite and WAL unless measured load demonstrates a real need to change. + +### B. Evidence engine +- Normalize Kotak SMS, Slice email and Paytm notifications into one `Evidence` structure. +- Use one candidate selection algorithm for point-in-time and time-window evidence. +- Make manual review use the same matcher invariants with an operator-selected candidate. +- Preserve raw source events separately from normalized evidence and redact them by retention policy. +- Shadow-test Google Messages Android notification evidence against the server-side libgm connector before any connector removal. + +### C. Durable delivery and alerts +- Keep the payment outbox concept. +- Consolidate webhook and operator notification retry/claim/lease mechanics into one durable delivery engine. +- Separate discrete alert events from persistent alert conditions. +- Replace periodic `Open()` calls for conditions with `EnsureCondition()` semantics that do not increment occurrence counts every scan. +- Persist health-condition timing instead of relying on process-local maturation maps. +### D. Payment lifecycle and jobs +- Stop running global expiry mutations as a side effect of ordinary reads and unrelated writes. +- Expire due payments in small bounded background batches. +- Let matching determine on-time versus late from persisted timestamps, not from whether an expiry cron happened first. +- Preserve `reuse_after` as the fingerprint allocation guard. +- Batch reconciliation persistence/classification instead of holding one write transaction across a large statement. +- Reject ambiguous statement dates unless a bank/import profile establishes the date convention. + +### E. Razorpay +- Replace duplicated test/live implementations with one shared rail engine. +- Retain separate constructors, credentials, storage namespaces/tables and explicit mode checks to preserve blast-radius isolation. +- Express temporary pilot restrictions, such as fixed ₹1 live checkout, as policy/configuration rather than forked code. + +### F. API and operator console +- Split API registration into public checkout, trusted integration, evidence ingest, relay, operator and provider modules. +- Introduce typed operator query endpoints and stop exposing database collection schemas to the browser. +- Use polling or a small SSE invalidation stream for live operator updates. +- Build stable response DTOs that deliberately exclude raw evidence unless a privileged detail endpoint requires it. + +### G. Customer checkout +- Reduce the checkout to one primary action per state. +- Remove duplicated explanations, secondary controls and technical terminology from the main path. +- Keep payment account selection automatic by default; only show a chooser when multiple rails are ready and there is a user-relevant reason to choose. +- Design the pending screen around amount, QR/action, countdown and authoritative verification state. +### H. Android / PayGate mobile +- Keep the relay runtime engine but replace the raw Java settings/debug screen. +- Move the user-facing application to Kotlin + Jetpack Compose while allowing the Java relay engine to coexist during migration. +- Add authenticated operator sessions with device-bound secure token storage. +- Add home dashboard, payment search/list/detail, review queue, alert inbox, relay health and settings. +- Make dangerous actions require explicit confirmation and server-side authorization/audit. +- Keep relay pairing credentials distinct from operator login credentials. + +## Explicit non-goals + +- No direct bank-account automation beyond the trusted evidence sources already supported. +- No automatic ambiguous-payment resolution. +- No automatic refund execution unless a provider API and explicit policy are later approved. +- No replay of old exhausted webhooks merely because transport has recovered. +- No secret material in mobile diagnostics, operator DTOs, logs or public API responses. +- No breaking production migration without dual-read/dual-write or a tested backfill path. + +## Success criteria + +PayGate v2 is considered ready to replace v1 only when: +- all current payment invariant and edge-case tests pass against the v2 engine; +- shadow comparison shows equivalent outcomes for existing evidence sources; +- production can roll back to the v1 path without database restore; +- customer checkout has fewer primary decisions and fewer controls per state; +- operator web and Android both use the same typed operator API; +- no UI depends on PocketBase collection names or realtime record payloads; +- v2 operational alerts do not re-open/increment continuously for unchanged conditions; +- relay health, backups, outbox and reconciliation have deterministic recovery tests. \ No newline at end of file diff --git a/docs/v2/01_TARGET_ARCHITECTURE.md b/docs/v2/01_TARGET_ARCHITECTURE.md new file mode 100644 index 0000000..66c4e0c --- /dev/null +++ b/docs/v2/01_TARGET_ARCHITECTURE.md @@ -0,0 +1,187 @@ +# PayGate v2 — Target Architecture + +## Architecture style + +PayGate remains a single deployable Go application backed by SQLite, with separate static/customer and operator/mobile clients. Internally it is a modular monolith with strict dependency direction and explicit transaction boundaries. + +```text +HTTP / relay / provider adapters + | + v + Application services + | + v + Typed domain + | + v + Unit of Work / repos + | + v + SQLite/PocketBase adapter +``` + +Only the bottom adapter layer may know about PocketBase `core.App`, `core.Record`, collection names, or database field strings. + +## Domain modules + +### Payments +Owns allocation, idempotency, state transitions, fingerprint quarantine and payment queries. It does not parse SMS/email/notifications and does not make network calls. + +### Evidence +Owns normalized evidence identity, validation and matching. Source adapters turn source-specific messages into `Evidence`; the matcher never receives raw SMS/email/notification payloads. + +### Reviews +Owns ambiguous/unmatched/manual-resolution workflow. Manual linking cannot bypass amount, account, evidence uniqueness, creation-time or quarantine invariants. +### Refunds +Owns refund reservation, idempotency, provider/reference uniqueness and explicit state transitions. Execution remains manual/provider-specific until a provider adapter is explicitly approved. + +### Reconciliation +Owns statement import, normalized statement entries, classification and discrepancy review. Parsing happens outside write transactions; persistence and classification happen in bounded batches. + +### Relay health +Stores observed device facts. A pure policy evaluates those facts into `healthy`, `degraded`, `blocked` and reason codes. Readiness code consumes policy output rather than reproducing boolean logic. + +### Operations +Owns alerts, durable delivery, backup status, retention and operational health. Alert events and alert conditions are separate concepts. + +## Core value types + +```go +type Money struct { Paise int64 } +type PaymentID string +type EvidenceID string +type AccountID string +type PaymentStatus uint8 +type MatchOutcome uint8 +``` + +String serialization belongs at HTTP/storage boundaries. The domain should prefer typed enums/value objects so invalid statuses and field-name typos cannot silently propagate. + +## Payment aggregate + +A payment contains requested/payable money, account, lifecycle timestamps, idempotency identity, external identity, and an optional applied evidence reference. Payer/evidence details may be materialized for operational convenience, but evidence remains a first-class record rather than an ad-hoc set of payment columns. +## Normalized evidence + +```go +type Evidence struct { + ID EvidenceID + Source EvidenceSource + Account AccountID + Amount Money + OccurredFrom time.Time + OccurredTo time.Time + Reference string + ReferenceKind ReferenceKind + PayerName string + UPIID string +} +``` + +Point-in-time evidence uses the same value for `OccurredFrom` and `OccurredTo`. Paytm minute-precision/notification evidence can use an interval. One matcher therefore handles all sources without a second notification-specific payment engine. + +## Match algorithm + +1. Validate source/account/amount/reference requirements. +2. Resolve duplicate evidence/reference assignment before candidate selection. +3. Clamp/validate occurrence interval against ingestion time according to source policy. +4. Query at most two candidate payments by account + exact payable amount + creation/time/quarantine constraints. +5. `0 candidates` -> unmatched/review depending on source policy. +6. `1 candidate` -> derive paid vs late from occurrence time and lifecycle timestamps. +7. `2 candidates` -> ambiguous; never choose by recency or heuristic. +8. Persist normalized evidence, payment mutation, audit entry and outbox event in one transaction. + +Manual review supplies a selected payment ID to the same invariant evaluator. It does not use a separate permissive matching implementation. +## Unit of Work + +Application services receive a `UnitOfWork` abstraction rather than `core.App`: + +```go +type UnitOfWork interface { + Payments() PaymentRepository + Evidence() EvidenceRepository + Reviews() ReviewRepository + Outbox() OutboxRepository + Audit() AuditRepository +} + +type Transactor interface { + Within(ctx context.Context, fn func(UnitOfWork) error) error +} +``` + +The initial implementation may wrap PocketBase transactions. A later plain-SQLite implementation can replace it without changing application/domain code. + +## Durable outbox/delivery + +Payment/refund events and operator notifications share claim, lease, retry and exhaustion mechanics. Destination-specific senders remain separate. + +```text +outbox item -> available -> claimed -> sending + -> delivered + -> retry scheduled + -> exhausted +``` + +The payment transaction inserts an outbox item only. Network delivery is always asynchronous and outside that transaction. +## Alert semantics + +Two APIs are required: + +```go +RecordEvent(kind, severity, dedupe, details) +EnsureCondition(kind, severity, dedupe, active, details) +``` + +`RecordEvent` increments occurrence count because a new event occurred. `EnsureCondition` updates last-seen/details while a condition remains active but does not increment each scan. A resolved condition may increment/reopen once when it becomes active again. + +## API surfaces + +### Public checkout +Minimal endpoints for account availability, creating a checkout payment and reading public payment status. No raw evidence or operator data. Strong body/rate limits. + +### Trusted integration +Server-to-server payment/refund endpoints authenticated independently from the public checkout surface. + +### Evidence ingest +Signed/secret source-specific routes that normalize and pass evidence to the evidence application service. + +### Relay +Device enrollment, signed events and heartbeat. Relay device credentials authorize relay operations only. + +### Operator +Authenticated typed queries/actions for dashboard, payments, reviews, refunds, reconciliation, alerts, relay, delivery, backups and settings. This is the only administrative API used by web/mobile clients. + +### Provider +Razorpay test/live webhooks and order operations implemented by a shared provider engine with mode-specific policy/credentials. +## Storage target + +Preferred long-term tables/entities: +- `payments` +- `evidence` and source/raw evidence records +- `review_cases` +- `refunds` +- `reconciliation_runs` / `reconciliation_entries` +- `relay_devices` / `relay_events` +- `outbox` +- `alerts` / optional alert history +- `audit_events` +- Razorpay orders/events with test/live isolation +- operator identities/sessions + +Do not collapse tables merely to reduce table count; consolidate duplicated behavior, not useful isolation. + +## Background work + +Jobs are bounded and idempotent: +- expire due payments in batches; +- claim/deliver outbox items; +- evaluate operational conditions; +- retention/redaction; +- backup verification and restore drill; +- reconciliation batches if an import is still processing. + +No ordinary `GET` should expire unrelated payments or create unrelated webhook rows as a side effect. + +## Google Messages exit strategy + +The libgm connector remains available during migration, but Android Google Messages notification observation is shadow-compared against it. If measured miss rate, latency and parsed evidence quality satisfy the acceptance threshold, remove server-side pairing/reauth/QR/session machinery and delete unsupported QR fallback paths. The removal requires measured parity, not assumption. \ No newline at end of file diff --git a/docs/v2/02_MIGRATION_AND_ROLLOUT.md b/docs/v2/02_MIGRATION_AND_ROLLOUT.md new file mode 100644 index 0000000..859e9f6 --- /dev/null +++ b/docs/v2/02_MIGRATION_AND_ROLLOUT.md @@ -0,0 +1,111 @@ +# PayGate v2 — Migration and Rollout Plan + +## Principle + +This is a strangler rewrite, not a big-bang replacement. Production remains on the current v1 path until each v2 slice has characterization tests, shadow comparison where possible, reversible schema changes, and a rollback path that does not require restoring the database. + +## Phase 0 — Stabilize v1 before extraction + +- Merge/fix relay power-health compatibility. +- Replace webhook exhaustion alert scanning with aggregate condition semantics. +- Remove/disable unsupported Google Messages QR fallback from operator-facing paths. +- Add characterization tests for payment allocation, cancellation, late evidence, stale evidence, duplicate RRN/reference, idempotency, refunds and reconciliation. +- Capture production-like anonymized fixture cases for every evidence source. + +Exit: v1 behavior is frozen by tests and known operational bugs are not being copied into v2. + +## Phase 1 — Typed domain foundation + +- Add typed money/payment/evidence/outcome types without changing persistence. +- Add repository/UoW interfaces and PocketBase-backed adapters. +- Move one read-only operator query through the repository layer first. +- Move payment record serialization/mapping into one adapter location. +- Prevent new direct `core.Record` access outside adapters with package/lint conventions. + +Exit: application services can execute through repositories while the live schema remains unchanged. +## Phase 2 — Evidence normalization in shadow mode + +- Add normalized `Evidence` and source adapter interfaces. +- Run Kotak SMS, Slice email and Paytm notification fixtures through both v1 and v2 match decision logic. +- Persist v2 normalized evidence in shadow-only storage or test fixtures first; do not mutate production payment state from v2 yet. +- Compare candidate IDs, outcome, paid/late classification and rejection reason. +- Require exact parity on all invariant cases before enabling v2 writes. + +Exit: v2 matcher produces the same safe decision as v1 for all known/fixture cases and property tests. + +## Phase 3 — Transactional v2 write path + +- Enable v2 payment/evidence mutation behind an environment/feature flag. +- Keep current schema initially and write through the repository adapter. +- Preserve existing outbox/audit rows so downstream integrations remain unchanged. +- Start with one evidence source, then expand source by source. +- Record comparison telemetry without raw sensitive evidence. + +Rollback: disable v2 matcher flag; v1 reads the same persisted payments/evidence columns. + +## Phase 4 — Operator API decoupling + +- Add typed `/api/operator/v2/*` query/action endpoints. +- Port dashboard, payments, reviews, refunds, reconciliation, alerts, relay, delivery, backups and settings. +- Web operator console and Android app consume these endpoints only. +- Replace PocketBase realtime record subscriptions with polling/SSE invalidations. +- Once no client reads PocketBase collections directly, collection schema can evolve independently. + +Exit: browser/mobile clients have no collection-name or PocketBase realtime dependency. +## Phase 5 — Customer checkout v2 + +- Ship the redesigned checkout against the existing compatible payment contract first. +- Measure create errors, abandonment between create/payment, completion latency and status-refresh errors. +- Add direct public checkout API only after equivalent limits/abuse controls are implemented in Go. +- If the Node/Hono BFF is removed, keep the old frontend deployment available for immediate DNS/route rollback. + +## Phase 6 — PayGate mobile + +- Release the modern app UI while preserving the existing relay package/application ID and signing certificate. +- Maintain existing relay pairing/device identity across upgrade. +- Introduce operator authentication independently from relay identity. +- Start mobile management read-only; add reviewed actions one by one with audit and confirmation. +- Do not allow mobile UI/background service lifecycle changes to stop the relay when the operator logs out. + +## Phase 7 — Consolidation + +- Shared Razorpay engine replaces duplicated test/live business code while retaining credential/storage isolation. +- Shared durable delivery engine replaces duplicated webhook/operator notification retry mechanics. +- Batched expiry/reconciliation replace oversized/global transaction behavior. +- Configuration is decomposed into validated sub-configs. +- Legacy source aliases/routes are removed after usage counters reach zero. + +## Phase 8 — Optional PocketBase removal + +Only consider this after operator clients and application services are fully decoupled. Implement typed SQLite repositories (prefer explicit SQL/sqlc-style generated access) and migration tooling. Keep schema compatibility or provide a proven one-way migration plus backup/restore rehearsal. +## Required release gates for every phase + +1. `gofmt`, unit/integration tests, `go vet`, staticcheck, govulncheck and relevant race tests pass. +2. Frontend/mobile contract tests cover old and new server compatibility where a rolling upgrade is possible. +3. Database migrations are additive until all old readers are removed. +4. A fresh verified backup exists before any production schema/data remediation. +5. Migration scripts have a test that runs from a realistic pre-migration database. +6. Payment/evidence duplicate constraints are checked before and after rollout. +7. No production data reset, automatic historical replay, or destructive backfill. +8. Health/readiness must degrade safely if a required verification source is unhealthy. +9. Metrics/logs used for shadow comparison must not contain raw payer evidence or secrets. +10. Every operator mutation creates an audit entry containing actor, action and target identity. + +## Cutover strategy + +Prefer percentage/source-scoped feature flags over host-level blue/green for the payment matcher because both implementations need to observe the same SQLite state. Public web UI can be blue/green independently. + +Suggested matcher progression: fixtures -> test DB -> shadow production decisions -> one source/write path -> all direct UPI sources -> operator/manual flows -> old path disabled -> old path deleted after soak. + +## Rollback triggers + +Immediate rollback for any of: +- candidate/outcome mismatch against established invariant fixtures; +- duplicate evidence/reference or duplicate active fingerprint allocation; +- unexpected increase in ambiguous/unmatched evidence; +- payment creation success but missing durable outbox/audit mutation; +- elevated SQLite busy/write-lock time attributable to v2; +- mobile/operator action bypassing authorization or audit; +- checkout regression that obscures exact amount, expiry or authoritative verification status. + +Deletion of old code occurs only after a separate stable soak period; rollback must remain possible before deletion. \ No newline at end of file From 9d73226a25fac18ef411bed6526825e673e47952 Mon Sep 17 00:00:00 2001 From: Phloraxx Date: Fri, 28 Aug 2026 12:02:38 +0000 Subject: [PATCH 03/36] Unify automatic payment evidence matching --- internal/domain/evidence.go | 64 ++++++++++ internal/payments/matcher.go | 188 +++++++++++++++++++++++++++ internal/payments/service.go | 241 +++++------------------------------ 3 files changed, 287 insertions(+), 206 deletions(-) create mode 100644 internal/domain/evidence.go create mode 100644 internal/payments/matcher.go diff --git a/internal/domain/evidence.go b/internal/domain/evidence.go new file mode 100644 index 0000000..92b9d42 --- /dev/null +++ b/internal/domain/evidence.go @@ -0,0 +1,64 @@ +package domain + +import "time" + +type EvidenceSource string +type EvidenceReferenceKind string +type MatchOutcome string + +const ( + EvidenceSourceBankSMS EvidenceSource = "bank_sms" + EvidenceSourceBankEmail EvidenceSource = "bank_email" + EvidenceSourcePaytmNotification EvidenceSource = "paytm_notification" + EvidenceSourceReconciliation EvidenceSource = "reconciliation" + EvidenceSourceManual EvidenceSource = "manual" +) + +const ( + EvidenceReferenceRRN EvidenceReferenceKind = "rrn" + EvidenceReferenceRelay EvidenceReferenceKind = "relay_reference" +) +const ( + MatchMarkedPaid MatchOutcome = "marked_paid" + MatchMarkedLate MatchOutcome = "marked_late" + MatchDuplicateRRN MatchOutcome = "duplicate_rrn" + MatchDuplicateEvidence MatchOutcome = "duplicate_evidence" + MatchRRNAccountMismatch MatchOutcome = "rrn_account_mismatch" + MatchRRNAmountMismatch MatchOutcome = "rrn_amount_mismatch" + MatchEvidenceAccountMismatch MatchOutcome = "evidence_account_mismatch" + MatchEvidenceAmountMismatch MatchOutcome = "evidence_amount_mismatch" + MatchAmbiguous MatchOutcome = "ambiguous" + MatchUnmatched MatchOutcome = "unmatched" + MatchNotMatchable MatchOutcome = "not_matchable" + MatchError MatchOutcome = "error" +) + +type Evidence struct { + Account PaymentAccount + AmountPaise int64 + OccurredFrom time.Time + OccurredUntil time.Time + + Reference string + ReferenceKind EvidenceReferenceKind + Source EvidenceSource + PayerName string + UPIID string +} + +func (e Evidence) NormalizeWindow(now time.Time) Evidence { + now = now.UTC() + e.OccurredFrom = e.OccurredFrom.UTC() + e.OccurredUntil = e.OccurredUntil.UTC() + if e.OccurredFrom.IsZero() || e.OccurredFrom.After(now) { + e.OccurredFrom = now + e.OccurredUntil = now + } + if e.OccurredUntil.IsZero() || e.OccurredUntil.Before(e.OccurredFrom) { + e.OccurredUntil = e.OccurredFrom + } + if e.OccurredUntil.After(now) { + e.OccurredUntil = now + } + return e +} diff --git a/internal/payments/matcher.go b/internal/payments/matcher.go new file mode 100644 index 0000000..d618fbc --- /dev/null +++ b/internal/payments/matcher.go @@ -0,0 +1,188 @@ +package payments + +import ( + "database/sql" + "errors" + "net/http" + "strings" + "time" + + "github.com/Phloraxx/payment-api/internal/domain" + "github.com/pocketbase/dbx" + "github.com/pocketbase/pocketbase/core" +) + +// MatchEvidenceInApp is the single automatic payment matcher used by every +// trusted evidence adapter. Source-specific parsing and authentication happen +// before this boundary; payment invariants are enforced only here. +func (s *Service) MatchEvidenceInApp(tx core.App, evidence domain.Evidence, now time.Time) (*core.Record, domain.MatchOutcome, bool, error) { + now = now.UTC() + account, _, err := s.paymentAccount(string(evidence.Account)) + if err != nil { + return nil, domain.MatchNotMatchable, false, err + } + evidence = evidence.NormalizeWindow(now) + evidence.Reference = strings.TrimSpace(evidence.Reference) + if evidence.AmountPaise <= 0 || evidence.Reference == "" { + return nil, domain.MatchNotMatchable, false, domain.New( + "EVIDENCE_NOT_MATCHABLE", + "payment evidence requires an exact amount and unique reference", + http.StatusUnprocessableEntity, + ) + } + + duplicateField, outcomes, codes, err := evidenceIdentity(evidence.ReferenceKind) + if err != nil { + return nil, domain.MatchNotMatchable, false, err + } + existing, err := tx.FindFirstRecordByData("payments", duplicateField, evidence.Reference) + if err == nil { + if existing.GetString("payment_account") != account { + return nil, outcomes.accountMismatch, false, domain.New(codes.accountMismatch, codes.accountMessage, http.StatusConflict) + } + if int64(existing.GetInt("payable_amount")) != evidence.AmountPaise { + return nil, outcomes.amountMismatch, false, domain.New(codes.amountMismatch, codes.amountMessage, http.StatusConflict) + } + return existing, outcomes.duplicate, false, nil + } + if !errors.Is(err, sql.ErrNoRows) { + return nil, domain.MatchError, false, err + } + createdBefore := evidence.OccurredUntil.Add(EvidenceTimestampTolerance) + onTime, err := tx.FindRecordsByFilter( + "payments", + "payment_account = {:account} && payable_amount = {:amount} && created_at <= {:createdBefore} && ((status = 'pending' && expires_at >= {:evidenceAt}) || (status = 'expired' && expires_at >= {:evidenceAt} && reuse_after > {:now}) || (status = 'cancelled' && resolved_at != '' && resolved_at >= {:evidenceAt} && reuse_after > {:now}))", + "created", + 2, + 0, + dbx.Params{ + "account": account, + "amount": evidence.AmountPaise, + "now": filterDate(now), + "evidenceAt": filterDate(evidence.OccurredFrom), + "createdBefore": filterDate(createdBefore), + }, + ) + if err != nil { + return nil, domain.MatchError, false, err + } + if len(onTime) > 1 { + return nil, domain.MatchAmbiguous, false, domain.AmbiguousMatch() + } + if len(onTime) == 1 { + record := onTime[0] + applyNormalizedEvidence(record, evidence, domain.StatusPaid, now, s.Config.AmountQuarantine) + if err := tx.Save(record); err != nil { + return nil, domain.MatchError, false, err + } + if err := s.schedule(tx, "payment.paid", record, now); err != nil { + return nil, domain.MatchError, false, err + } + return record, domain.MatchMarkedPaid, true, nil + } + + expired, err := s.ExpireDueInApp(tx, now) + if err != nil { + return nil, domain.MatchError, false, err + } + queued := expired > 0 + + late, err := tx.FindRecordsByFilter( + "payments", + "payment_account = {:account} && payable_amount = {:amount} && (status = 'expired' || status = 'cancelled') && reuse_after > {:now} && created_at <= {:createdBefore}", + "-created", + 2, + 0, + dbx.Params{ + "account": account, + "amount": evidence.AmountPaise, + "now": filterDate(now), + "createdBefore": filterDate(createdBefore), + }, + ) + if err != nil { + return nil, domain.MatchError, queued, err + } + if len(late) > 1 { + return nil, domain.MatchAmbiguous, queued, domain.AmbiguousMatch() + } + if len(late) == 1 { + record := late[0] + applyNormalizedEvidence(record, evidence, domain.StatusLate, now, s.Config.AmountQuarantine) + if err := tx.Save(record); err != nil { + return nil, domain.MatchError, queued, err + } + if err := s.schedule(tx, "payment.late", record, now); err != nil { + return nil, domain.MatchError, queued, err + } + return record, domain.MatchMarkedLate, true, nil + } + return nil, domain.MatchUnmatched, queued, nil +} + +type evidenceOutcomes struct { + duplicate domain.MatchOutcome + accountMismatch domain.MatchOutcome + amountMismatch domain.MatchOutcome +} + +type evidenceCodes struct { + accountMismatch string + accountMessage string + amountMismatch string + amountMessage string +} + +func evidenceIdentity(kind domain.EvidenceReferenceKind) (string, evidenceOutcomes, evidenceCodes, error) { + switch kind { + case domain.EvidenceReferenceRRN: + return "rrn", evidenceOutcomes{ + duplicate: domain.MatchDuplicateRRN, + accountMismatch: domain.MatchRRNAccountMismatch, + amountMismatch: domain.MatchRRNAmountMismatch, + }, evidenceCodes{ + accountMismatch: "RRN_ACCOUNT_MISMATCH", + accountMessage: "the UPI reference was already recorded for a different payment account", + amountMismatch: "RRN_AMOUNT_MISMATCH", + amountMessage: "the UPI reference was already recorded with a different amount", + }, nil + case domain.EvidenceReferenceRelay: + return "evidence_reference", evidenceOutcomes{ + duplicate: domain.MatchDuplicateEvidence, + accountMismatch: domain.MatchEvidenceAccountMismatch, + amountMismatch: domain.MatchEvidenceAmountMismatch, + }, evidenceCodes{ + accountMismatch: "EVIDENCE_ACCOUNT_MISMATCH", + accountMessage: "the notification evidence was already recorded for a different payment account", + amountMismatch: "EVIDENCE_AMOUNT_MISMATCH", + amountMessage: "the notification evidence was already recorded with a different amount", + }, nil + default: + return "", evidenceOutcomes{}, evidenceCodes{}, domain.New( + "EVIDENCE_REFERENCE_INVALID", + "payment evidence has an unsupported reference kind", + http.StatusUnprocessableEntity, + ) + } +} +func applyNormalizedEvidence(record *core.Record, evidence domain.Evidence, status domain.PaymentStatus, now time.Time, quarantine time.Duration) { + paidAt := evidence.OccurredFrom.UTC() + if paidAt.IsZero() || paidAt.After(now) { + paidAt = now.UTC() + } + record.Set("status", string(status)) + record.Set("payer_name", strings.TrimSpace(evidence.PayerName)) + if evidence.UPIID != "" { + record.Set("upi_id", strings.TrimSpace(evidence.UPIID)) + } + switch evidence.ReferenceKind { + case domain.EvidenceReferenceRRN: + record.Set("rrn", strings.TrimSpace(evidence.Reference)) + case domain.EvidenceReferenceRelay: + record.Set("evidence_source", string(evidence.Source)) + record.Set("evidence_reference", strings.TrimSpace(evidence.Reference)) + } + record.Set("paid_at", paidAt) + record.Set("resolved_at", now.UTC()) + extendReuseAfter(record, now.UTC().Add(quarantine)) +} diff --git a/internal/payments/service.go b/internal/payments/service.go index a3e6b15..d384939 100644 --- a/internal/payments/service.go +++ b/internal/payments/service.go @@ -269,11 +269,11 @@ func (s *Service) Match(parsed domain.ParsedSMS) (*MatchResult, error) { } now := s.now() var record *core.Record - var action string + var outcome domain.MatchOutcome var queued bool err := s.App.RunInTransaction(func(tx core.App) error { var err error - record, action, queued, err = s.MatchInApp(tx, parsed, now) + record, outcome, queued, err = s.matchBankEvidenceInApp(tx, parsed, now) return err }) if err != nil { @@ -282,105 +282,30 @@ func (s *Service) Match(parsed domain.ParsedSMS) (*MatchResult, error) { if queued { s.WakeWebhooks() } - return &MatchResult{Payment: FromRecord(record), Action: action}, nil + return &MatchResult{Payment: FromRecord(record), Action: string(outcome)}, nil } -// MatchInApp applies exact-amount matching inside the caller's transaction. -// It returns whether outgoing webhook work was queued so the caller can wake the -// delivery loop only after the transaction commits. +// MatchInApp is the backward-compatible bank-evidence adapter. New sources +// should normalize into domain.Evidence and call MatchEvidenceInApp instead. func (s *Service) MatchInApp(tx core.App, parsed domain.ParsedSMS, now time.Time) (*core.Record, string, bool, error) { - now = now.UTC() - account, _, err := s.paymentAccount(string(parsed.Account)) - if err != nil { - return nil, "not_matchable", false, err - } - rrn := strings.TrimSpace(parsed.RRN) - evidenceAt := parsed.OccurredAt.UTC() - if evidenceAt.IsZero() || evidenceAt.After(now) { - evidenceAt = now - } - if rrn == "" || parsed.AmountPaise <= 0 { - return nil, "not_matchable", false, domain.New("SMS_NOT_MATCHABLE", "bank SMS requires an exact amount and RRN", http.StatusUnprocessableEntity) - } - - existing, err := tx.FindFirstRecordByData("payments", "rrn", rrn) - if err == nil { - if existing.GetString("payment_account") != account { - return nil, "rrn_account_mismatch", false, domain.New("RRN_ACCOUNT_MISMATCH", "the UPI reference was already recorded for a different payment account", http.StatusConflict) - } - if int64(existing.GetInt("payable_amount")) != parsed.AmountPaise { - return nil, "rrn_amount_mismatch", false, domain.New("RRN_AMOUNT_MISMATCH", "the UPI reference was already recorded with a different amount", http.StatusConflict) - } - return existing, "duplicate_rrn", false, nil - } - if !errors.Is(err, sql.ErrNoRows) { - return nil, "error", false, err - } - - createdBefore := evidenceAt.Add(EvidenceTimestampTolerance) - - // Match by when the bank says the credit occurred, not merely when the SMS - // happened to reach PayGate. This prevents an on-time payment from becoming - // "late" solely because the bank/phone delivered the SMS after expiry. - onTime, err := tx.FindRecordsByFilter( - "payments", - "payment_account = {:account} && payable_amount = {:amount} && created_at <= {:createdBefore} && ((status = 'pending' && expires_at >= {:evidenceAt}) || (status = 'expired' && expires_at >= {:evidenceAt} && reuse_after > {:now}) || (status = 'cancelled' && resolved_at != '' && resolved_at >= {:evidenceAt} && reuse_after > {:now}))", - "created", - 2, - 0, - dbx.Params{"account": account, "amount": parsed.AmountPaise, "now": filterDate(now), "evidenceAt": filterDate(evidenceAt), "createdBefore": filterDate(createdBefore)}, - ) - if err != nil { - return nil, "error", false, err - } - if len(onTime) > 1 { - return nil, "ambiguous", false, domain.AmbiguousMatch() - } - if len(onTime) == 1 { - record := onTime[0] - applyEvidence(record, parsed, domain.StatusPaid, now, s.Config.AmountQuarantine) - if err := tx.Save(record); err != nil { - return nil, "error", false, err - } - if err := s.schedule(tx, "payment.paid", record, now); err != nil { - return nil, "error", false, err - } - return record, "marked_paid", true, nil - } - - expired, err := s.ExpireDueInApp(tx, now) - if err != nil { - return nil, "error", false, err - } - queued := expired > 0 + record, outcome, queued, err := s.matchBankEvidenceInApp(tx, parsed, now) + return record, string(outcome), queued, err +} - late, err := tx.FindRecordsByFilter( - "payments", - "payment_account = {:account} && payable_amount = {:amount} && (status = 'expired' || status = 'cancelled') && reuse_after > {:now} && created_at <= {:createdBefore}", - "-created", - 2, - 0, - dbx.Params{"account": account, "amount": parsed.AmountPaise, "now": filterDate(now), "evidenceAt": filterDate(evidenceAt), "createdBefore": filterDate(createdBefore)}, - ) - if err != nil { - return nil, "error", queued, err +func (s *Service) matchBankEvidenceInApp(tx core.App, parsed domain.ParsedSMS, now time.Time) (*core.Record, domain.MatchOutcome, bool, error) { + if parsed.AmountPaise <= 0 || strings.TrimSpace(parsed.RRN) == "" { + return nil, domain.MatchNotMatchable, false, domain.New("SMS_NOT_MATCHABLE", "bank SMS requires an exact amount and RRN", http.StatusUnprocessableEntity) } - if len(late) > 1 { - return nil, "ambiguous", queued, domain.AmbiguousMatch() + source := domain.EvidenceSourceBankSMS + if parsed.Account == domain.PaymentAccountSlice { + source = domain.EvidenceSourceBankEmail } - if len(late) == 1 { - record := late[0] - applyEvidence(record, parsed, domain.StatusLate, now, s.Config.AmountQuarantine) - if err := tx.Save(record); err != nil { - return nil, "error", queued, err - } - if err := s.schedule(tx, "payment.late", record, now); err != nil { - return nil, "error", queued, err - } - return record, "marked_late", true, nil - } - - return nil, "unmatched", queued, nil + return s.MatchEvidenceInApp(tx, domain.Evidence{ + Account: parsed.Account, AmountPaise: parsed.AmountPaise, + OccurredFrom: parsed.OccurredAt, OccurredUntil: parsed.OccurredAt, + Reference: strings.TrimSpace(parsed.RRN), ReferenceKind: domain.EvidenceReferenceRRN, + Source: source, PayerName: parsed.PayerName, UPIID: parsed.UPIId, + }, now) } type NotificationEvidence struct { @@ -392,118 +317,22 @@ type NotificationEvidence struct { Reference string } -// MatchNotificationInApp matches a trusted Paytm for Business notification by -// account, exact DDM amount, and notification occurrence time. Unlike bank SMS -// evidence, Paytm push notifications do not necessarily expose a UPI RRN, so a -// unique relay evidence reference is used for idempotency instead. +// MatchNotificationInApp remains the Paytm compatibility adapter while all +// automatic matching is enforced by MatchEvidenceInApp. func (s *Service) MatchNotificationInApp(tx core.App, evidence NotificationEvidence, now time.Time) (*core.Record, string, bool, error) { - now = now.UTC() - account, _, err := s.paymentAccount(string(evidence.Account)) - if err != nil { - return nil, "not_matchable", false, err - } - if account != string(domain.PaymentAccountPaytm) { - return nil, "not_matchable", false, domain.New("NOTIFICATION_ACCOUNT_INVALID", "notification evidence is only valid for the Paytm account", http.StatusBadRequest) - } - reference := strings.TrimSpace(evidence.Reference) - if evidence.AmountPaise <= 0 || reference == "" { - return nil, "not_matchable", false, domain.New("NOTIFICATION_NOT_MATCHABLE", "Paytm notification requires an exact amount and evidence reference", http.StatusUnprocessableEntity) - } - evidenceAt := evidence.OccurredAt.UTC() - evidenceUntil := evidence.OccurredUntil.UTC() - if evidenceAt.IsZero() || evidenceAt.After(now) { - evidenceAt = now - evidenceUntil = now - } - if evidenceUntil.IsZero() || evidenceUntil.Before(evidenceAt) { - evidenceUntil = evidenceAt - } - if evidenceUntil.After(now) { - evidenceUntil = now - } - - existing, err := tx.FindFirstRecordByData("payments", "evidence_reference", reference) - if err == nil { - if existing.GetString("payment_account") != account { - return nil, "evidence_account_mismatch", false, domain.New("EVIDENCE_ACCOUNT_MISMATCH", "the notification evidence was already recorded for a different payment account", http.StatusConflict) - } - if int64(existing.GetInt("payable_amount")) != evidence.AmountPaise { - return nil, "evidence_amount_mismatch", false, domain.New("EVIDENCE_AMOUNT_MISMATCH", "the notification evidence was already recorded with a different amount", http.StatusConflict) - } - return existing, "duplicate_evidence", false, nil - } - if !errors.Is(err, sql.ErrNoRows) { - return nil, "error", false, err - } - - createdBefore := evidenceUntil.Add(EvidenceTimestampTolerance) - onTime, err := tx.FindRecordsByFilter( - "payments", - "payment_account = {:account} && payable_amount = {:amount} && created_at <= {:createdBefore} && ((status = 'pending' && expires_at >= {:evidenceAt}) || (status = 'expired' && expires_at >= {:evidenceAt} && reuse_after > {:now}) || (status = 'cancelled' && resolved_at != '' && resolved_at >= {:evidenceAt} && reuse_after > {:now}))", - "created", 2, 0, - dbx.Params{"account": account, "amount": evidence.AmountPaise, "now": filterDate(now), "evidenceAt": filterDate(evidenceAt), "createdBefore": filterDate(createdBefore)}, - ) - if err != nil { - return nil, "error", false, err - } - if len(onTime) > 1 { - return nil, "ambiguous", false, domain.AmbiguousMatch() - } - if len(onTime) == 1 { - record := onTime[0] - applyNotificationEvidence(record, evidence, domain.StatusPaid, now, s.Config.AmountQuarantine) - if err := tx.Save(record); err != nil { - return nil, "error", false, err - } - if err := s.schedule(tx, "payment.paid", record, now); err != nil { - return nil, "error", false, err - } - return record, "marked_paid", true, nil - } - - expired, err := s.ExpireDueInApp(tx, now) - if err != nil { - return nil, "error", false, err - } - queued := expired > 0 - late, err := tx.FindRecordsByFilter( - "payments", - "payment_account = {:account} && payable_amount = {:amount} && (status = 'expired' || status = 'cancelled') && reuse_after > {:now} && created_at <= {:createdBefore}", - "-created", 2, 0, - dbx.Params{"account": account, "amount": evidence.AmountPaise, "now": filterDate(now), "createdBefore": filterDate(createdBefore)}, - ) - if err != nil { - return nil, "error", queued, err - } - if len(late) > 1 { - return nil, "ambiguous", queued, domain.AmbiguousMatch() - } - if len(late) == 1 { - record := late[0] - applyNotificationEvidence(record, evidence, domain.StatusLate, now, s.Config.AmountQuarantine) - if err := tx.Save(record); err != nil { - return nil, "error", queued, err - } - if err := s.schedule(tx, "payment.late", record, now); err != nil { - return nil, "error", queued, err - } - return record, "marked_late", true, nil - } - return nil, "unmatched", queued, nil -} - -func applyNotificationEvidence(record *core.Record, evidence NotificationEvidence, status domain.PaymentStatus, now time.Time, quarantine time.Duration) { - paidAt := evidence.OccurredAt.UTC() - if paidAt.IsZero() || paidAt.After(now) { - paidAt = now.UTC() - } - record.Set("status", string(status)) - record.Set("payer_name", strings.TrimSpace(evidence.PayerName)) - record.Set("evidence_source", "paytm_notification") - record.Set("evidence_reference", strings.TrimSpace(evidence.Reference)) - record.Set("paid_at", paidAt) - record.Set("resolved_at", now.UTC()) - extendReuseAfter(record, now.UTC().Add(quarantine)) + if evidence.Account != domain.PaymentAccountPaytm { + return nil, string(domain.MatchNotMatchable), false, domain.New("NOTIFICATION_ACCOUNT_INVALID", "notification evidence is only valid for the Paytm account", http.StatusBadRequest) + } + if evidence.AmountPaise <= 0 || strings.TrimSpace(evidence.Reference) == "" { + return nil, string(domain.MatchNotMatchable), false, domain.New("NOTIFICATION_NOT_MATCHABLE", "Paytm notification requires an exact amount and evidence reference", http.StatusUnprocessableEntity) + } + record, outcome, queued, err := s.MatchEvidenceInApp(tx, domain.Evidence{ + Account: evidence.Account, AmountPaise: evidence.AmountPaise, + OccurredFrom: evidence.OccurredAt, OccurredUntil: evidence.OccurredUntil, + Reference: strings.TrimSpace(evidence.Reference), ReferenceKind: domain.EvidenceReferenceRelay, + Source: domain.EvidenceSourcePaytmNotification, PayerName: evidence.PayerName, + }, now) + return record, string(outcome), queued, err } // ManualMatchInApp explicitly links reviewed bank evidence to one payment. It From 7048a7062f34f9c06dd38cf33d63efd0dfc3515b Mon Sep 17 00:00:00 2001 From: Phloraxx Date: Fri, 28 Aug 2026 12:02:38 +0000 Subject: [PATCH 04/36] Model operational alerts as persistent conditions --- cmd/payment-api/main.go | 2 +- internal/alerts/service.go | 102 +++++++++++--- internal/alerts/service_test.go | 124 ++++++++++++++++++ .../20260828020000_alert_condition_kinds.go | 51 +++++++ 4 files changed, 260 insertions(+), 19 deletions(-) create mode 100644 migrations/20260828020000_alert_condition_kinds.go diff --git a/cmd/payment-api/main.go b/cmd/payment-api/main.go index 2ca2096..f6d6140 100644 --- a/cmd/payment-api/main.go +++ b/cmd/payment-api/main.go @@ -168,7 +168,7 @@ func main() { if relayErr != nil { stdLogger.Error("relay health check failed", "error", relayErr) } else if !relayStatus.Ready { - _, _, _ = alertService.Open(alerts.Input{ + _, _, _ = alertService.EnsureOpen(alerts.Input{ Kind: "relay_unavailable", Severity: "warning", DedupeKey: "relay:paytm", Message: "Paytm verification relay is unavailable; new Paytm checkouts are blocked.", Details: relayStatus, }) diff --git a/internal/alerts/service.go b/internal/alerts/service.go index 9dc7acd..a8af5c7 100644 --- a/internal/alerts/service.go +++ b/internal/alerts/service.go @@ -65,6 +65,18 @@ func (s *Service) NotificationsEnabled() bool { } func (s *Service) Open(input Input) (string, bool, error) { + return s.upsert(input, true) +} + +// EnsureOpen records a persistent operational condition without counting every +// periodic health scan as a new occurrence. A resolved condition increments +// once when it becomes active again; an already-open condition only refreshes +// its details/last-seen state. +func (s *Service) EnsureOpen(input Input) (string, bool, error) { + return s.upsert(input, false) +} + +func (s *Service) upsert(input Input, countRepeated bool) (string, bool, error) { input.Kind = strings.TrimSpace(input.Kind) input.Severity = strings.TrimSpace(input.Severity) input.DedupeKey = strings.TrimSpace(input.DedupeKey) @@ -88,7 +100,11 @@ func (s *Service) Open(input Input) (string, bool, error) { record.Set("details", input.Details) record.Set("last_seen_at", now) record.Set("resolved_at", "") - record.Set("occurrence_count", record.GetInt("occurrence_count")+1) + count := record.GetInt("occurrence_count") + if countRepeated || wasResolved { + count++ + } + record.Set("occurrence_count", count) if s.NotificationsEnabled() { if wasResolved || severityEscalated || record.GetString("notification_status") == "disabled" { s.queueNotification(record, now) @@ -395,7 +411,7 @@ func (s *Service) CheckConnector(status gmessages.Status) error { return s.resolveProblems("connector:reauth", "connector:disconnected", "connector:unresponsive") } if status.State == "reauth_required" { - if _, _, err := s.Open(Input{Kind: "connector_reauth", Severity: "critical", DedupeKey: "connector:reauth", Message: "Google Messages requires browser reauthentication", Details: map[string]any{"state": status.State, "lastError": status.LastError}}); err != nil { + if _, _, err := s.EnsureOpen(Input{Kind: "connector_reauth", Severity: "critical", DedupeKey: "connector:reauth", Message: "Google Messages requires browser reauthentication", Details: map[string]any{"state": status.State, "lastError": status.LastError}}); err != nil { return err } } else if err := s.Resolve("connector:reauth"); err != nil { @@ -407,7 +423,7 @@ func (s *Service) CheckConnector(status gmessages.Status) error { return err } } else if status.Paired && s.problemMature("connector:disconnected", 5*time.Minute) { - if _, _, err := s.Open(Input{Kind: "connector_disconnected", Severity: "critical", DedupeKey: "connector:disconnected", Message: "Google Messages has remained disconnected for more than five minutes", Details: map[string]any{"state": status.State, "lastError": status.LastError}}); err != nil { + if _, _, err := s.EnsureOpen(Input{Kind: "connector_disconnected", Severity: "critical", DedupeKey: "connector:disconnected", Message: "Google Messages has remained disconnected for more than five minutes", Details: map[string]any{"state": status.State, "lastError": status.LastError}}); err != nil { return err } } @@ -417,7 +433,7 @@ func (s *Service) CheckConnector(status gmessages.Status) error { return err } } else if s.problemMature("connector:unresponsive", 15*time.Minute) { - if _, _, err := s.Open(Input{Kind: "connector_unresponsive", Severity: "warning", DedupeKey: "connector:unresponsive", Message: "Paired phone has not been responsive for more than fifteen minutes", Details: map[string]any{"state": status.State}}); err != nil { + if _, _, err := s.EnsureOpen(Input{Kind: "connector_unresponsive", Severity: "warning", DedupeKey: "connector:unresponsive", Message: "Paired phone has not been responsive for more than fifteen minutes", Details: map[string]any{"state": status.State}}); err != nil { return err } } @@ -437,7 +453,7 @@ func (s *Service) CheckCapacity(snapshot payments.CapacitySnapshot) error { severity = "critical" } message := fmt.Sprintf("₹%s fingerprint pool is %.0f%% utilized (%d of 99 blocked)", pool.RequestedAmount, pool.UtilizationPercent, pool.Blocked) - if _, _, err := s.Open(Input{Kind: "capacity_high", Severity: severity, DedupeKey: key, Message: message, Details: pool}); err != nil { + if _, _, err := s.EnsureOpen(Input{Kind: "capacity_high", Severity: severity, DedupeKey: key, Message: message, Details: pool}); err != nil { return err } } @@ -456,30 +472,80 @@ func (s *Service) CheckCapacity(snapshot payments.CapacitySnapshot) error { } func (s *Service) CheckWebhookExhaustion() error { - records, err := s.App.FindRecordsByFilter("webhook_deliveries", "status = 'exhausted'", "-updated", 100, 0) + count, err := s.App.CountRecords("webhook_deliveries", dbx.NewExp("status = 'exhausted'")) if err != nil { return err } - active := map[string]struct{}{} - for _, record := range records { - key := "webhook:" + record.Id - active[key] = struct{}{} - if _, _, err := s.Open(Input{Kind: "webhook_exhausted", Severity: "critical", DedupeKey: key, Message: "Outgoing webhook exhausted its retry limit", Details: map[string]any{"deliveryId": record.Id, "eventId": record.GetString("event_id"), "event": record.GetString("event"), "lastError": record.GetString("last_error")}}); err != nil { - return err + if err := s.resolveLegacyWebhookAlerts(); err != nil { + return err + } + const dedupeKey = "webhook:exhausted" + if count == 0 { + return s.Resolve(dedupeKey) + } + + exhausted, err := s.App.FindRecordsByFilter("webhook_deliveries", "status = 'exhausted'", "-updated", 1, 0) + if err != nil { + return err + } + details := map[string]any{"exhaustedCount": count} + message := fmt.Sprintf("%d outgoing webhook deliveries have exhausted their retry limit", count) + if len(exhausted) == 1 { + latest := exhausted[0] + details["latestExhaustedAt"] = latest.GetDateTime("updated").String() + details["latestEvent"] = latest.GetString("event") + latestExhaustedAt := latest.GetDateTime("updated").Time() + delivered, deliveredErr := s.App.FindRecordsByFilter("webhook_deliveries", "status = 'delivered' && delivered_at != ''", "-delivered_at", 1, 0) + if deliveredErr != nil { + return deliveredErr + } + if len(delivered) == 1 { + latestDeliveredAt := delivered[0].GetDateTime("delivered_at").Time() + details["latestDeliveredAt"] = delivered[0].GetDateTime("delivered_at").String() + if !latestDeliveredAt.IsZero() && latestDeliveredAt.After(latestExhaustedAt) { + details["transportRecovered"] = true + message += "; newer webhook deliveries have succeeded" + } } } - open, err := s.App.FindRecordsByFilter("alerts", "kind = 'webhook_exhausted' && status = 'open'", "created", 0, 0) + _, _, err = s.EnsureOpen(Input{ + Kind: "webhook_exhausted", Severity: "critical", DedupeKey: dedupeKey, + Message: message, Details: details, + }) + return err +} + +// resolveLegacyWebhookAlerts silently closes the per-delivery alerts generated +// by the pre-v2 scanner. They represent the same aggregate condition and must +// not emit hundreds of "resolved" notifications during remediation. +func (s *Service) resolveLegacyWebhookAlerts() error { + records, err := s.App.FindRecordsByFilter( + "alerts", + "kind = 'webhook_exhausted' && dedupe_key != 'webhook:exhausted' && status = 'open'", + "created", 0, 0, + ) if err != nil { return err } - for _, record := range open { - if _, ok := active[record.GetString("dedupe_key")]; !ok { - if err := s.Resolve(record.GetString("dedupe_key")); err != nil { + if len(records) == 0 { + return nil + } + now := s.now() + return s.App.RunInTransaction(func(tx core.App) error { + for _, item := range records { + record, err := tx.FindRecordById("alerts", item.Id) + if err != nil { + return err + } + record.Set("status", "resolved") + record.Set("resolved_at", now) + record.Set("notification_status", "disabled") + if err := tx.Save(record); err != nil { return err } } - } - return nil + return nil + }) } func (s *Service) problemMature(key string, delay time.Duration) bool { diff --git a/internal/alerts/service_test.go b/internal/alerts/service_test.go index 9abeb3f..4f42a60 100644 --- a/internal/alerts/service_test.go +++ b/internal/alerts/service_test.go @@ -2,16 +2,19 @@ package alerts import ( "context" + "fmt" "io" "net/http" "net/http/httptest" "testing" "time" + "github.com/Phloraxx/payment-api/internal/config" "github.com/Phloraxx/payment-api/internal/gmessages" "github.com/Phloraxx/payment-api/internal/payments" "github.com/Phloraxx/payment-api/internal/webhooks" _ "github.com/Phloraxx/payment-api/migrations" + "github.com/pocketbase/pocketbase/core" "github.com/pocketbase/pocketbase/tests" ) @@ -227,3 +230,124 @@ func TestOperatorAlertClientDoesNotFollowRedirects(t *testing.T) { t.Fatalf("notification status=%s", record.GetString("notification_status")) } } + +func TestEnsureOpenDoesNotInflatePersistentCondition(t *testing.T) { + app, err := tests.NewTestApp() + if err != nil { + t.Fatal(err) + } + defer app.Cleanup() + service := NewService(app) + now := time.Date(2026, 8, 1, 8, 0, 0, 0, time.UTC) + service.Now = func() time.Time { return now } + + id, created, err := service.EnsureOpen(Input{Kind: "relay_unavailable", Severity: "warning", DedupeKey: "relay:paytm", Message: "relay unavailable", Details: map[string]any{"ready": false}}) + if err != nil || !created { + t.Fatalf("id=%s created=%v err=%v", id, created, err) + } + now = now.Add(time.Minute) + if second, created, err := service.EnsureOpen(Input{Kind: "relay_unavailable", Severity: "warning", DedupeKey: "relay:paytm", Message: "still unavailable", Details: map[string]any{"ready": false}}); err != nil || created || second != id { + t.Fatalf("second=%s created=%v err=%v", second, created, err) + } + record, _ := app.FindRecordById("alerts", id) + if got := record.GetInt("occurrence_count"); got != 1 { + t.Fatalf("occurrences while continuously open=%d", got) + } + if err := service.Resolve("relay:paytm"); err != nil { + t.Fatal(err) + } + now = now.Add(time.Minute) + if _, _, err := service.EnsureOpen(Input{Kind: "relay_unavailable", Severity: "warning", DedupeKey: "relay:paytm", Message: "unavailable again"}); err != nil { + t.Fatal(err) + } + record, _ = app.FindRecordById("alerts", id) + if got := record.GetInt("occurrence_count"); got != 2 { + t.Fatalf("occurrences after real reactivation=%d", got) + } +} + +func TestWebhookExhaustionIsOneAggregateCondition(t *testing.T) { + app, err := tests.NewTestApp() + if err != nil { + t.Fatal(err) + } + defer app.Cleanup() + service := NewService(app) + now := time.Date(2026, 8, 1, 8, 0, 0, 0, time.UTC) + service.Now = func() time.Time { return now } + + paymentService := payments.NewService(app, config.Config{PaymentTTL: 5 * time.Minute, AmountQuarantine: 24 * time.Hour, UPIID: "operator@bank", UPIPayeeName: "PayGate"}, nil) + paymentService.Now = func() time.Time { return now } + paymentService.SuffixStart = func() (int64, error) { return 1, nil } + payment, _, err := paymentService.Create(payments.CreateInput{AmountRupees: 100, PaymentAccount: "kotak"}) + if err != nil { + t.Fatal(err) + } + + collection, err := app.FindCollectionByNameOrId("webhook_deliveries") + if err != nil { + t.Fatal(err) + } + var deliveries []*core.Record + for i := 0; i < 2; i++ { + record := core.NewRecord(collection) + record.Set("event_id", fmt.Sprintf("evt_exhausted_%d", i)) + record.Set("event", "payment.paid") + record.Set("payment", payment.ID) + record.Set("url", "https://example.invalid/webhook") + record.Set("body", `{}`) + record.Set("attempts", 8) + record.Set("status", "exhausted") + record.Set("next_attempt_at", now.Add(24*time.Hour)) + record.Set("last_error", "historical failure") + if err := app.Save(record); err != nil { + t.Fatal(err) + } + deliveries = append(deliveries, record) + if _, _, err := service.Open(Input{Kind: "webhook_exhausted", Severity: "critical", DedupeKey: "webhook:" + record.Id, Message: "legacy per-delivery alert"}); err != nil { + t.Fatal(err) + } + } + + if err := service.CheckWebhookExhaustion(); err != nil { + t.Fatal(err) + } + aggregate, err := app.FindFirstRecordByData("alerts", "dedupe_key", "webhook:exhausted") + if err != nil { + t.Fatal(err) + } + if got := aggregate.GetInt("occurrence_count"); got != 1 { + t.Fatalf("aggregate occurrences=%d", got) + } + for _, delivery := range deliveries { + legacy, err := app.FindFirstRecordByData("alerts", "dedupe_key", "webhook:"+delivery.Id) + if err != nil { + t.Fatal(err) + } + if legacy.GetString("status") != "resolved" || legacy.GetString("notification_status") != "disabled" { + t.Fatalf("legacy alert status=%s notification=%s", legacy.GetString("status"), legacy.GetString("notification_status")) + } + } + if err := service.CheckWebhookExhaustion(); err != nil { + t.Fatal(err) + } + aggregate, _ = app.FindRecordById("alerts", aggregate.Id) + if got := aggregate.GetInt("occurrence_count"); got != 1 { + t.Fatalf("aggregate occurrences after repeat scan=%d", got) + } + + for _, delivery := range deliveries { + delivery.Set("status", "delivered") + delivery.Set("delivered_at", now.Add(time.Hour)) + if err := app.Save(delivery); err != nil { + t.Fatal(err) + } + } + if err := service.CheckWebhookExhaustion(); err != nil { + t.Fatal(err) + } + aggregate, _ = app.FindRecordById("alerts", aggregate.Id) + if aggregate.GetString("status") != "resolved" { + t.Fatalf("aggregate status after recovery=%s", aggregate.GetString("status")) + } +} diff --git a/migrations/20260828020000_alert_condition_kinds.go b/migrations/20260828020000_alert_condition_kinds.go new file mode 100644 index 0000000..9b49c60 --- /dev/null +++ b/migrations/20260828020000_alert_condition_kinds.go @@ -0,0 +1,51 @@ +package migrations + +import ( + "fmt" + + "github.com/pocketbase/pocketbase/core" + pbmigrations "github.com/pocketbase/pocketbase/migrations" +) + +func init() { + pbmigrations.Register(func(app core.App) error { + alerts, err := app.FindCollectionByNameOrId("alerts") + if err != nil { + return err + } + kind, ok := alerts.Fields.GetByName("kind").(*core.SelectField) + if !ok || kind == nil { + return fmt.Errorf("alerts.kind is not a select field") + } + if !containsSelectValue(kind.Values, "relay_unavailable") { + kind.Values = append(kind.Values, "relay_unavailable") + } + return app.Save(alerts) + }, func(app core.App) error { + alerts, err := app.FindCollectionByNameOrId("alerts") + if err != nil { + return nil + } + kind, ok := alerts.Fields.GetByName("kind").(*core.SelectField) + if !ok || kind == nil { + return nil + } + values := kind.Values[:0] + for _, value := range kind.Values { + if value != "relay_unavailable" { + values = append(values, value) + } + } + kind.Values = values + return app.Save(alerts) + }) +} + +func containsSelectValue(values []string, target string) bool { + for _, value := range values { + if value == target { + return true + } + } + return false +} From bdc7fc6695f532a8c4a520f8e4e96680ed627f2c Mon Sep 17 00:00:00 2001 From: Phloraxx Date: Fri, 28 Aug 2026 12:02:55 +0000 Subject: [PATCH 05/36] Decouple operator clients from PocketBase records --- internal/api/api.go | 15 + internal/api/operator_v2.go | 262 ++++++++++++++ internal/api/operator_v2_test.go | 205 +++++++++++ internal/operatorview/service.go | 485 ++++++++++++++++++++++++++ internal/operatorview/service_test.go | 60 ++++ package-lock.json | 7 - package.json | 1 - web/src/App.tsx | 14 +- web/src/api.ts | 54 +++ web/src/pages/Dashboard.tsx | 67 +--- web/src/pages/Login.tsx | 4 +- web/src/pages/Operations.tsx | 160 +++------ web/src/pages/Payments.tsx | 87 ++--- web/src/pages/RazorpayTest.tsx | 30 +- web/src/pages/Records.tsx | 69 +--- web/src/pages/Settings.tsx | 2 +- web/src/pb.ts | 27 -- web/src/types.ts | 151 ++++---- 18 files changed, 1293 insertions(+), 407 deletions(-) create mode 100644 internal/api/operator_v2.go create mode 100644 internal/api/operator_v2_test.go create mode 100644 internal/operatorview/service.go create mode 100644 internal/operatorview/service_test.go create mode 100644 web/src/api.ts delete mode 100644 web/src/pb.ts diff --git a/internal/api/api.go b/internal/api/api.go index d3dbaff..3cc096b 100644 --- a/internal/api/api.go +++ b/internal/api/api.go @@ -101,6 +101,21 @@ func (a *API) Register(app core.App) { e.Router.GET("/api/paygate/health", a.health) e.Router.GET("/api/config", a.getConfig) e.Router.GET("/api/dashboard", a.dashboard) + e.Router.GET("/api/operator/v2/overview", a.operatorV2Overview) + e.Router.GET("/api/operator/v2/payments", a.operatorV2Payments) + e.Router.GET("/api/operator/v2/payments/{id}", a.operatorV2Payment) + e.Router.GET("/api/operator/v2/reviews", a.operatorV2Reviews) + e.Router.GET("/api/operator/v2/reviews/{id}", a.operatorV2Review) + e.Router.POST("/api/operator/v2/reviews/{id}/resolve", a.operatorV2ResolveReview).Bind(apis.BodyLimit(maxReviewRequestBytes)) + e.Router.GET("/api/operator/v2/alerts", a.operatorV2Alerts) + e.Router.GET("/api/operator/v2/relay", a.operatorV2Relay) + e.Router.GET("/api/operator/v2/reconciliation", a.operatorV2ReconciliationRuns) + e.Router.GET("/api/operator/v2/reconciliation/{id}/entries", a.operatorV2ReconciliationEntries) + e.Router.GET("/api/operator/v2/refunds", a.operatorV2Refunds) + e.Router.GET("/api/operator/v2/records/{kind}", a.operatorV2OperationalRecords) + e.Router.GET("/api/operator/v2/razorpay/{mode}/orders", a.operatorV2RazorpayOrders) + e.Router.POST("/api/operator/v2/payments/{id}/cancel", a.operatorV2CancelPayment).Bind(apis.BodyLimit(maxReviewRequestBytes)) + e.Router.POST("/api/operator/v2/reviews/{id}/dismiss", a.operatorV2DismissReview).Bind(apis.BodyLimit(maxReviewRequestBytes)) e.Router.GET("/api/capacity", a.capacity) e.Router.POST("/api/review-cases/{id}/resolve", a.resolveReview).Bind(apis.BodyLimit(maxReviewRequestBytes)) e.Router.POST("/api/reconciliation/import", a.importReconciliation).Bind(apis.BodyLimit(maxStatementRequestBytes)) diff --git a/internal/api/operator_v2.go b/internal/api/operator_v2.go new file mode 100644 index 0000000..48acd0f --- /dev/null +++ b/internal/api/operator_v2.go @@ -0,0 +1,262 @@ +package api + +import ( + "database/sql" + "errors" + "net/http" + "strconv" + "strings" + + "github.com/Phloraxx/payment-api/internal/operatorview" + "github.com/Phloraxx/payment-api/internal/reviews" + "github.com/pocketbase/pocketbase/core" +) + +func (a *API) operatorV2Overview(e *core.RequestEvent) error { + if !a.dashboardAuth(e) { + return e.UnauthorizedError("operator authentication is required", nil) + } + view, err := operatorview.New(e.App).Overview(queryLimit(e, 8)) + if err != nil { + return e.InternalServerError("failed to load operator overview", err) + } + payload := map[string]any{"overview": view, "connector": a.connectorStatus()} + if a.Payments != nil { + capacity, capacityErr := a.Payments.Capacity() + if capacityErr != nil { + return e.InternalServerError("failed to load payment capacity", capacityErr) + } + payload["capacity"] = capacity + } + if a.Backups != nil { + backup, backupErr := a.Backups.GetStatus(e.Request.Context(), false) + if backupErr != nil { + payload["backup"] = map[string]any{"enabled": a.Config.BackupCron != "", "error": backupErr.Error()} + } else { + payload["backup"] = backup + } + } + if a.AndroidRelay != nil { + relay, relayErr := a.AndroidRelay.Status(a.Config.AndroidRelayStaleAfter) + if relayErr != nil { + return e.InternalServerError("failed to load relay status", relayErr) + } + payload["relay"] = relay + } + return e.JSON(http.StatusOK, payload) +} +func (a *API) operatorV2Payments(e *core.RequestEvent) error { + if !a.dashboardAuth(e) { + return e.UnauthorizedError("operator authentication is required", nil) + } + items, err := operatorview.New(e.App).ListPayments(e.Request.URL.Query().Get("status"), queryLimit(e, 50)) + if err != nil { + if strings.Contains(err.Error(), "invalid payment status") { + return e.BadRequestError("invalid payment status", nil) + } + return e.InternalServerError("failed to list payments", err) + } + return e.JSON(http.StatusOK, map[string]any{"payments": items}) +} + +func (a *API) operatorV2Payment(e *core.RequestEvent) error { + if !a.dashboardAuth(e) { + return e.UnauthorizedError("operator authentication is required", nil) + } + item, err := operatorview.New(e.App).GetPayment(e.Request.PathValue("id")) + if errors.Is(err, sql.ErrNoRows) { + return e.NotFoundError("payment not found", nil) + } + if err != nil { + return e.InternalServerError("failed to load payment", err) + } + return e.JSON(http.StatusOK, item) +} +func (a *API) operatorV2Reviews(e *core.RequestEvent) error { + if !a.dashboardAuth(e) { + return e.UnauthorizedError("operator authentication is required", nil) + } + items, err := operatorview.New(e.App).ListReviews(e.Request.URL.Query().Get("status"), queryLimit(e, 50)) + if err != nil { + if strings.Contains(err.Error(), "invalid review status") { + return e.BadRequestError("invalid review status", nil) + } + return e.InternalServerError("failed to list reviews", err) + } + return e.JSON(http.StatusOK, map[string]any{"reviews": items}) +} + +func (a *API) operatorV2Alerts(e *core.RequestEvent) error { + if !a.dashboardAuth(e) { + return e.UnauthorizedError("operator authentication is required", nil) + } + items, err := operatorview.New(e.App).ListAlerts(e.Request.URL.Query().Get("status"), queryLimit(e, 50)) + if err != nil { + if strings.Contains(err.Error(), "invalid alert status") { + return e.BadRequestError("invalid alert status", nil) + } + return e.InternalServerError("failed to list alerts", err) + } + return e.JSON(http.StatusOK, map[string]any{"alerts": items}) +} +func (a *API) operatorV2Relay(e *core.RequestEvent) error { + if !a.dashboardAuth(e) { + return e.UnauthorizedError("operator authentication is required", nil) + } + if a.AndroidRelay == nil { + return e.JSON(http.StatusOK, map[string]any{"enabled": false, "ready": false}) + } + status, err := a.AndroidRelay.Status(a.Config.AndroidRelayStaleAfter) + if err != nil { + return e.InternalServerError("failed to load relay status", err) + } + return e.JSON(http.StatusOK, status) +} + +func queryLimit(e *core.RequestEvent, fallback int) int { + value := strings.TrimSpace(e.Request.URL.Query().Get("limit")) + if value == "" { + return fallback + } + parsed, err := strconv.Atoi(value) + if err != nil || parsed <= 0 { + return fallback + } + if parsed > 100 { + return 100 + } + return parsed +} + +func (a *API) operatorV2CancelPayment(e *core.RequestEvent) error { + if !a.dashboardAuth(e) { + return e.UnauthorizedError("operator authentication is required", nil) + } + payment, err := a.Payments.Cancel(e.Request.PathValue("id")) + if err != nil { + return writeDomainError(e, err) + } + item, err := operatorview.New(e.App).GetPayment(payment.ID) + if err != nil { + return e.InternalServerError("payment cancelled but operator view could not be loaded", err) + } + return e.JSON(http.StatusOK, item) +} + +type operatorDismissReviewBody struct { + Note string `json:"note"` +} + +func (a *API) operatorV2DismissReview(e *core.RequestEvent) error { + if !a.dashboardAuth(e) { + return e.UnauthorizedError("operator authentication is required", nil) + } + if a.Reviews == nil { + return e.NotFoundError("review service is unavailable", nil) + } + var body operatorDismissReviewBody + if err := decodeJSON(e, &body); err != nil { + return e.BadRequestError("invalid JSON body", err) + } + result, err := a.Reviews.Resolve(reviews.ResolveInput{ + CaseID: e.Request.PathValue("id"), Action: "dismissed", Note: body.Note, Actor: a.actor(e), + }) + if err != nil { + return writeDomainError(e, err) + } + return e.JSON(http.StatusOK, result) +} + +func (a *API) operatorV2Review(e *core.RequestEvent) error { + if !a.dashboardAuth(e) { + return e.UnauthorizedError("operator authentication is required", nil) + } + item, err := operatorview.New(e.App).GetReview(e.Request.PathValue("id")) + if errors.Is(err, sql.ErrNoRows) { + return e.NotFoundError("review case not found", nil) + } + if err != nil { + return e.InternalServerError("failed to load review case", err) + } + return e.JSON(http.StatusOK, item) +} + +func (a *API) operatorV2ResolveReview(e *core.RequestEvent) error { + if !a.dashboardAuth(e) { + return e.UnauthorizedError("operator authentication is required", nil) + } + if a.Reviews == nil { + return e.NotFoundError("review service is unavailable", nil) + } + var body reviewResolutionBody + if err := decodeJSON(e, &body); err != nil { + return e.BadRequestError("invalid JSON body", err) + } + result, err := a.Reviews.Resolve(reviews.ResolveInput{ + CaseID: e.Request.PathValue("id"), Action: body.Action, PaymentID: body.PaymentID, + BankReference: body.BankReference, Note: body.Note, Actor: a.actor(e), + }) + if err != nil { + return writeDomainError(e, err) + } + return e.JSON(http.StatusOK, result) +} + +func (a *API) operatorV2ReconciliationRuns(e *core.RequestEvent) error { + if !a.dashboardAuth(e) { + return e.UnauthorizedError("operator authentication is required", nil) + } + items, err := operatorview.New(e.App).ListReconciliationRuns(queryLimit(e, 50)) + if err != nil { + return e.InternalServerError("failed to list reconciliation runs", err) + } + return e.JSON(http.StatusOK, map[string]any{"runs": items}) +} +func (a *API) operatorV2ReconciliationEntries(e *core.RequestEvent) error { + if !a.dashboardAuth(e) { + return e.UnauthorizedError("operator authentication is required", nil) + } + items, err := operatorview.New(e.App).ListReconciliationEntries(e.Request.PathValue("id"), queryLimit(e, 250)) + if err != nil { + return e.InternalServerError("failed to list reconciliation entries", err) + } + return e.JSON(http.StatusOK, map[string]any{"entries": items}) +} +func (a *API) operatorV2Refunds(e *core.RequestEvent) error { + if !a.dashboardAuth(e) { + return e.UnauthorizedError("operator authentication is required", nil) + } + items, err := operatorview.New(e.App).ListRefunds(queryLimit(e, 50)) + if err != nil { + return e.InternalServerError("failed to list refunds", err) + } + return e.JSON(http.StatusOK, map[string]any{"refunds": items}) +} + +func (a *API) operatorV2OperationalRecords(e *core.RequestEvent) error { + if !a.dashboardAuth(e) { + return e.UnauthorizedError("operator authentication is required", nil) + } + items, err := operatorview.New(e.App).ListOperationalRecords(e.Request.PathValue("kind"), queryLimit(e, 50)) + if err != nil { + if strings.Contains(err.Error(), "invalid operational record kind") { + return e.NotFoundError("record view not found", nil) + } + return e.InternalServerError("failed to list operational records", err) + } + return e.JSON(http.StatusOK, map[string]any{"records": items}) +} + +func (a *API) operatorV2RazorpayOrders(e *core.RequestEvent) error { + if !a.dashboardAuth(e) { + return e.UnauthorizedError("operator authentication is required", nil) + } + items, err := operatorview.New(e.App).ListRazorpayOrders(e.Request.PathValue("mode"), queryLimit(e, 50)) + if err != nil { + if strings.Contains(err.Error(), "invalid razorpay mode") { + return e.NotFoundError("razorpay mode not found", nil) + } + return e.InternalServerError("failed to list razorpay orders", err) + } + return e.JSON(http.StatusOK, map[string]any{"orders": items}) +} diff --git a/internal/api/operator_v2_test.go b/internal/api/operator_v2_test.go new file mode 100644 index 0000000..0d2775d --- /dev/null +++ b/internal/api/operator_v2_test.go @@ -0,0 +1,205 @@ +package api + +import ( + "io" + "net/http" + "net/http/httptest" + "strings" + "testing" + "time" + + "github.com/Phloraxx/payment-api/internal/payments" + "github.com/pocketbase/pocketbase/apis" + "github.com/pocketbase/pocketbase/core" + "github.com/pocketbase/pocketbase/tests" +) + +func TestOperatorV2RequiresOperatorAuthAndReturnsTypedViews(t *testing.T) { + var paymentID string + app := apiTestFactory(t, func(app *tests.TestApp, paymentService *payments.Service) { + payment, _, err := paymentService.Create(payments.CreateInput{AmountRupees: 250, PaymentAccount: "kotak"}) + if err != nil { + t.Fatal(err) + } + paymentID = payment.ID + record, err := app.FindRecordById("payments", payment.ID) + if err != nil { + t.Fatal(err) + } + record.Set("payer_name", "Sensitive Payer") + record.Set("rrn", "123456789012") + if err := app.Save(record); err != nil { + t.Fatal(err) + } + }) + defer app.Cleanup() + users, err := app.FindCollectionByNameOrId("users") + if err != nil { + t.Fatal(err) + } + operator := core.NewRecord(users) + operator.SetEmail("operator@example.com") + operator.SetPassword("operator-password-123") + operator.SetVerified(true) + if err := app.Save(operator); err != nil { + t.Fatal(err) + } + token, err := operator.NewAuthToken() + if err != nil { + t.Fatal(err) + } + + router, err := apis.NewRouter(app) + if err != nil { + t.Fatal(err) + } + serveEvent := &core.ServeEvent{App: app, Router: router} + if err := app.OnServe().Trigger(serveEvent, func(e *core.ServeEvent) error { return nil }); err != nil { + t.Fatal(err) + } + mux, err := serveEvent.Router.BuildMux() + if err != nil { + t.Fatal(err) + } + server := httptest.NewServer(mux) + defer server.Close() + + unauth, err := server.Client().Get(server.URL + "/api/operator/v2/overview") + if err != nil { + t.Fatal(err) + } + _ = unauth.Body.Close() + if unauth.StatusCode != http.StatusUnauthorized { + t.Fatalf("unauth status=%d", unauth.StatusCode) + } + overview := operatorGet(t, server, token, "/api/operator/v2/overview") + for _, want := range []string{`"paymentCounts"`, `"recentPayments"`, paymentID} { + if !strings.Contains(overview, want) { + t.Fatalf("overview missing %q: %s", want, overview) + } + } + if strings.Contains(overview, "Sensitive Payer") || strings.Contains(overview, "123456789012") { + t.Fatalf("overview leaked payment evidence: %s", overview) + } + + list := operatorGet(t, server, token, "/api/operator/v2/payments?status=pending") + if !strings.Contains(list, paymentID) || strings.Contains(list, "Sensitive Payer") || strings.Contains(list, "123456789012") { + t.Fatalf("payment list contract=%s", list) + } + + detail := operatorGet(t, server, token, "/api/operator/v2/payments/"+paymentID) + if !strings.Contains(detail, "Sensitive Payer") || !strings.Contains(detail, "123456789012") { + t.Fatalf("operator detail missing evidence: %s", detail) + } + + smsCollection, err := app.FindCollectionByNameOrId("sms_events") + if err != nil { + t.Fatal(err) + } + smsRecord := core.NewRecord(smsCollection) + smsRecord.Set("source", "android_webhook") + smsRecord.Set("payment_account", "kotak") + smsRecord.Set("body", "RAW SECRET SMS BODY MUST NOT LEAK") + smsRecord.Set("message_time", time.Now().UTC()) + smsRecord.Set("amount", 25001) + smsRecord.Set("rrn", "998877665544") + smsRecord.Set("payer_name", "Review Payer") + smsRecord.Set("processing_status", "unmatched") + if err := app.Save(smsRecord); err != nil { + t.Fatal(err) + } + reviewCollection, err := app.FindCollectionByNameOrId("review_cases") + if err != nil { + t.Fatal(err) + } + reviewRecord := core.NewRecord(reviewCollection) + reviewRecord.Set("kind", "unmatched") + reviewRecord.Set("status", "open") + reviewRecord.Set("severity", "warning") + reviewRecord.Set("sms_event", smsRecord.Id) + reviewRecord.Set("reason", "Evidence requires operator review") + reviewRecord.Set("opened_at", time.Now().UTC()) + if err := app.Save(reviewRecord); err != nil { + t.Fatal(err) + } + reviewDetail := operatorGet(t, server, token, "/api/operator/v2/reviews/"+reviewRecord.Id) + for _, want := range []string{`"kind":"sms"`, `"reference":"998877665544"`, `"payerName":"Review Payer"`} { + if !strings.Contains(reviewDetail, want) { + t.Fatalf("review detail missing %q: %s", want, reviewDetail) + } + } + if strings.Contains(reviewDetail, "RAW SECRET SMS BODY MUST NOT LEAK") { + t.Fatalf("review detail leaked raw body: %s", reviewDetail) + } + + badKindReq, _ := http.NewRequest(http.MethodGet, server.URL+"/api/operator/v2/records/not-real", nil) + badKindReq.Header.Set("Authorization", "Bearer "+token) + badKindRes, err := server.Client().Do(badKindReq) + if err != nil { + t.Fatal(err) + } + _ = badKindRes.Body.Close() + if badKindRes.StatusCode != http.StatusNotFound { + t.Fatalf("invalid operational kind status=%d", badKindRes.StatusCode) + } + + unauthCancel, _ := http.NewRequest(http.MethodPost, server.URL+"/api/operator/v2/payments/"+paymentID+"/cancel", strings.NewReader(`{}`)) + unauthCancel.Header.Set("Content-Type", "application/json") + unauthCancelRes, err := server.Client().Do(unauthCancel) + if err != nil { + t.Fatal(err) + } + _ = unauthCancelRes.Body.Close() + if unauthCancelRes.StatusCode != http.StatusUnauthorized { + t.Fatalf("unauth cancel status=%d", unauthCancelRes.StatusCode) + } + cancelled := operatorPost(t, server, token, "/api/operator/v2/payments/"+paymentID+"/cancel", `{}`) + if !strings.Contains(cancelled, `"status":"cancelled"`) { + t.Fatalf("cancel contract=%s", cancelled) + } +} + +func operatorGet(t *testing.T, server *httptest.Server, token, path string) string { + t.Helper() + req, err := http.NewRequest(http.MethodGet, server.URL+path, nil) + if err != nil { + t.Fatal(err) + } + req.Header.Set("Authorization", "Bearer "+token) + res, err := server.Client().Do(req) + if err != nil { + t.Fatal(err) + } + defer res.Body.Close() + body, err := io.ReadAll(res.Body) + if err != nil { + t.Fatal(err) + } + if res.StatusCode != http.StatusOK { + t.Fatalf("GET %s status=%d body=%s", path, res.StatusCode, body) + } + return string(body) +} + +func operatorPost(t *testing.T, server *httptest.Server, token, path, body string) string { + t.Helper() + req, err := http.NewRequest(http.MethodPost, server.URL+path, strings.NewReader(body)) + if err != nil { + t.Fatal(err) + } + req.Header.Set("Authorization", "Bearer "+token) + req.Header.Set("Content-Type", "application/json") + res, err := server.Client().Do(req) + if err != nil { + t.Fatal(err) + } + defer res.Body.Close() + payload, err := io.ReadAll(res.Body) + if err != nil { + t.Fatal(err) + } + if res.StatusCode != http.StatusOK { + t.Fatalf("POST %s status=%d body=%s", path, res.StatusCode, payload) + } + return string(payload) +} diff --git a/internal/operatorview/service.go b/internal/operatorview/service.go new file mode 100644 index 0000000..425594a --- /dev/null +++ b/internal/operatorview/service.go @@ -0,0 +1,485 @@ +package operatorview + +import ( + "database/sql" + "errors" + "fmt" + "strings" + "time" + + "github.com/pocketbase/dbx" + "github.com/pocketbase/pocketbase/core" +) + +type Service struct{ App core.App } + +func New(app core.App) *Service { return &Service{App: app} } + +type PaymentSummary struct { + ID string `json:"id"` + PaymentAccount string `json:"paymentAccount"` + RequestedAmountPaise int64 `json:"requestedAmountPaise"` + PayableAmountPaise int64 `json:"payableAmountPaise"` + Status string `json:"status"` + CreatedAt string `json:"createdAt"` + ExpiresAt string `json:"expiresAt"` + PaidAt string `json:"paidAt,omitempty"` +} +type PaymentDetail struct { + PaymentSummary + ExternalID string `json:"externalId,omitempty"` + PayerName string `json:"payerName,omitempty"` + UPIID string `json:"upiId,omitempty"` + RRN string `json:"rrn,omitempty"` + EvidenceSource string `json:"evidenceSource,omitempty"` + EvidenceReference string `json:"evidenceReference,omitempty"` + ResolvedAt string `json:"resolvedAt,omitempty"` +} + +type ReviewSummary struct { + ID string `json:"id"` + Kind string `json:"kind"` + Status string `json:"status"` + Severity string `json:"severity"` + PaymentID string `json:"paymentId,omitempty"` + CandidatePaymentIDs []string `json:"candidatePaymentIds,omitempty"` + Reason string `json:"reason"` + OpenedAt string `json:"openedAt"` +} + +type EvidenceDetail struct { + Kind string `json:"kind"` + ID string `json:"id"` + Source string `json:"source,omitempty"` + Sender string `json:"sender,omitempty"` + Subject string `json:"subject,omitempty"` + Amount int64 `json:"amountPaise,omitempty"` + Reference string `json:"reference,omitempty"` + UPIID string `json:"upiId,omitempty"` + PayerName string `json:"payerName,omitempty"` + OccurredAt string `json:"occurredAt,omitempty"` + Description string `json:"description,omitempty"` + Status string `json:"status,omitempty"` + Notes string `json:"notes,omitempty"` +} + +type ReviewDetail struct { + ReviewSummary + Resolution string `json:"resolution,omitempty"` + ResolutionNote string `json:"resolutionNote,omitempty"` + ResolvedAt string `json:"resolvedAt,omitempty"` + Evidence *EvidenceDetail `json:"evidence,omitempty"` +} + +type AlertSummary struct { + ID string `json:"id"` + Kind string `json:"kind"` + Status string `json:"status"` + Severity string `json:"severity"` + Message string `json:"message"` + OccurrenceCount int `json:"occurrenceCount"` + FirstSeenAt string `json:"firstSeenAt"` + LastSeenAt string `json:"lastSeenAt"` + NotificationStatus string `json:"notificationStatus,omitempty"` + NotificationAttempts int `json:"notificationAttempts,omitempty"` + NotificationLastError string `json:"notificationLastError,omitempty"` + NotificationDeliveredAt string `json:"notificationDeliveredAt,omitempty"` +} + +type Overview struct { + PaymentCounts map[string]int64 `json:"paymentCounts"` + OpenReviews int64 `json:"openReviews"` + OpenAlerts int64 `json:"openAlerts"` + Recent []PaymentSummary `json:"recentPayments"` +} + +func (s *Service) Overview(limit int) (Overview, error) { + limit = clampLimit(limit, 8, 20) + counts := map[string]int64{"total": 0, "pending": 0, "paid": 0, "late": 0, "expired": 0, "cancelled": 0} + var err error + counts["total"], err = s.App.CountRecords("payments") + if err != nil { + return Overview{}, err + } + for _, status := range []string{"pending", "paid", "late", "expired", "cancelled"} { + counts[status], err = s.App.CountRecords("payments", dbx.NewExp("status = {:status}", dbx.Params{"status": status})) + if err != nil { + return Overview{}, err + } + } + openReviews, err := s.App.CountRecords("review_cases", dbx.NewExp("status = 'open'")) + if err != nil { + return Overview{}, err + } + openAlerts, err := s.App.CountRecords("alerts", dbx.NewExp("status = 'open'")) + if err != nil { + return Overview{}, err + } + recent, err := s.ListPayments("", limit) + if err != nil { + return Overview{}, err + } + return Overview{ + PaymentCounts: counts, + OpenReviews: openReviews, + OpenAlerts: openAlerts, + Recent: recent, + }, nil +} + +func (s *Service) ListPayments(status string, limit int) ([]PaymentSummary, error) { + limit = clampLimit(limit, 50, 100) + status = strings.TrimSpace(strings.ToLower(status)) + filter := "id != ''" + params := dbx.Params{} + if status != "" { + if !validPaymentStatus(status) { + return nil, fmt.Errorf("invalid payment status") + } + filter += " && status = {:status}" + params["status"] = status + } + records, err := s.App.FindRecordsByFilter("payments", filter, "-created_at", limit, 0, params) + if err != nil { + return nil, err + } + out := make([]PaymentSummary, 0, len(records)) + for _, record := range records { + out = append(out, paymentSummary(record)) + } + return out, nil +} + +func (s *Service) GetPayment(id string) (PaymentDetail, error) { + record, err := s.App.FindRecordById("payments", strings.TrimSpace(id)) + if err != nil { + if errors.Is(err, sql.ErrNoRows) { + return PaymentDetail{}, sql.ErrNoRows + } + return PaymentDetail{}, err + } + return PaymentDetail{ + PaymentSummary: paymentSummary(record), + ExternalID: record.GetString("external_id"), + PayerName: record.GetString("payer_name"), + UPIID: record.GetString("upi_id"), + RRN: record.GetString("rrn"), + EvidenceSource: record.GetString("evidence_source"), + EvidenceReference: record.GetString("evidence_reference"), + ResolvedAt: dateString(record, "resolved_at"), + }, nil +} +func (s *Service) ListReviews(status string, limit int) ([]ReviewSummary, error) { + limit = clampLimit(limit, 50, 100) + status = strings.TrimSpace(strings.ToLower(status)) + filter := "id != ''" + params := dbx.Params{} + if status != "" { + if status != "open" && status != "resolved" && status != "dismissed" { + return nil, fmt.Errorf("invalid review status") + } + filter += " && status = {:status}" + params["status"] = status + } + records, err := s.App.FindRecordsByFilter("review_cases", filter, "-opened_at", limit, 0, params) + if err != nil { + return nil, err + } + out := make([]ReviewSummary, 0, len(records)) + for _, record := range records { + out = append(out, ReviewSummary{ + ID: record.Id, Kind: record.GetString("kind"), Status: record.GetString("status"), Severity: record.GetString("severity"), + PaymentID: record.GetString("payment"), CandidatePaymentIDs: stringSlice(record.Get("candidate_payment_ids")), + Reason: record.GetString("reason"), OpenedAt: dateString(record, "opened_at"), + }) + } + return out, nil +} +func (s *Service) GetReview(id string) (ReviewDetail, error) { + record, err := s.App.FindRecordById("review_cases", strings.TrimSpace(id)) + if err != nil { + if errors.Is(err, sql.ErrNoRows) { + return ReviewDetail{}, sql.ErrNoRows + } + return ReviewDetail{}, err + } + summary := ReviewSummary{ + ID: record.Id, Kind: record.GetString("kind"), Status: record.GetString("status"), Severity: record.GetString("severity"), + PaymentID: record.GetString("payment"), CandidatePaymentIDs: stringSlice(record.Get("candidate_payment_ids")), + Reason: record.GetString("reason"), OpenedAt: dateString(record, "opened_at"), + } + detail := ReviewDetail{ReviewSummary: summary, Resolution: record.GetString("resolution"), ResolutionNote: record.GetString("resolution_note"), ResolvedAt: dateString(record, "resolved_at")} + for _, relation := range []struct{ field, collection, kind string }{ + {"sms_event", "sms_events", "sms"}, {"email_event", "email_events", "email"}, {"reconciliation_entry", "reconciliation_entries", "reconciliation"}, + } { + id := record.GetString(relation.field) + if id == "" { + continue + } + evidence, findErr := s.App.FindRecordById(relation.collection, id) + if findErr != nil { + return ReviewDetail{}, findErr + } + detail.Evidence = evidenceDetail(evidence, relation.kind) + break + } + return detail, nil +} + +func evidenceDetail(record *core.Record, kind string) *EvidenceDetail { + occurred := dateString(record, "message_time") + if kind == "reconciliation" { + occurred = dateString(record, "transaction_time") + } + status := record.GetString("processing_status") + if kind == "reconciliation" { + status = record.GetString("status") + } + return &EvidenceDetail{ + Kind: kind, ID: record.Id, Source: record.GetString("source"), Sender: record.GetString("sender"), Subject: record.GetString("subject"), + Amount: int64(record.GetInt("amount")), Reference: record.GetString("rrn"), UPIID: record.GetString("upi_id"), PayerName: record.GetString("payer_name"), + OccurredAt: occurred, Description: record.GetString("description"), Status: status, Notes: record.GetString("notes"), + } +} + +func (s *Service) ListAlerts(status string, limit int) ([]AlertSummary, error) { + limit = clampLimit(limit, 50, 100) + status = strings.TrimSpace(strings.ToLower(status)) + filter := "id != ''" + params := dbx.Params{} + if status != "" { + if status != "open" && status != "resolved" { + return nil, fmt.Errorf("invalid alert status") + } + filter += " && status = {:status}" + params["status"] = status + } + records, err := s.App.FindRecordsByFilter("alerts", filter, "-last_seen_at", limit, 0, params) + if err != nil { + return nil, err + } + out := make([]AlertSummary, 0, len(records)) + for _, record := range records { + out = append(out, AlertSummary{ + ID: record.Id, Kind: record.GetString("kind"), Status: record.GetString("status"), Severity: record.GetString("severity"), + Message: record.GetString("message"), OccurrenceCount: record.GetInt("occurrence_count"), + FirstSeenAt: dateString(record, "first_seen_at"), LastSeenAt: dateString(record, "last_seen_at"), + NotificationStatus: record.GetString("notification_status"), NotificationAttempts: record.GetInt("notification_attempts"), + NotificationLastError: record.GetString("notification_last_error"), NotificationDeliveredAt: dateString(record, "notification_delivered_at"), + }) + } + return out, nil +} +func paymentSummary(record *core.Record) PaymentSummary { + return PaymentSummary{ + ID: record.Id, + PaymentAccount: record.GetString("payment_account"), + RequestedAmountPaise: int64(record.GetInt("requested_amount")), + PayableAmountPaise: int64(record.GetInt("payable_amount")), + Status: record.GetString("status"), + CreatedAt: dateString(record, "created_at"), + ExpiresAt: dateString(record, "expires_at"), + PaidAt: dateString(record, "paid_at"), + } +} + +func dateString(record *core.Record, field string) string { + value := record.GetDateTime(field).Time() + if value.IsZero() { + return "" + } + return value.UTC().Format(time.RFC3339Nano) +} + +func validPaymentStatus(status string) bool { + switch status { + case "pending", "paid", "late", "expired", "cancelled": + return true + default: + return false + } +} +func stringSlice(value any) []string { + switch items := value.(type) { + case []string: + return append([]string(nil), items...) + case []any: + out := make([]string, 0, len(items)) + for _, item := range items { + if text, ok := item.(string); ok && strings.TrimSpace(text) != "" { + out = append(out, text) + } + } + return out + default: + return nil + } +} + +func clampLimit(value, fallback, max int) int { + if value <= 0 { + return fallback + } + if value > max { + return max + } + return value +} + +type ReconciliationRunSummary struct { + ID string `json:"id"` + Filename string `json:"filename"` + Status string `json:"status"` + TotalRows int `json:"totalRows"` + MatchedRows int `json:"matchedRows"` + UnmatchedRows int `json:"unmatchedRows"` + DuplicateRows int `json:"duplicateRows"` + ConflictRows int `json:"conflictRows"` + InvalidRows int `json:"invalidRows"` + Error string `json:"error,omitempty"` + StartedAt string `json:"startedAt"` + CompletedAt string `json:"completedAt,omitempty"` +} +type ReconciliationEntrySummary struct { + ID string `json:"id"` + RowNumber int `json:"rowNumber"` + TransactionTime string `json:"transactionTime,omitempty"` + AmountPaise int64 `json:"amountPaise,omitempty"` + Reference string `json:"reference,omitempty"` + Description string `json:"description,omitempty"` + Status string `json:"status"` + PaymentID string `json:"paymentId,omitempty"` + Notes string `json:"notes,omitempty"` +} +type RefundSummary struct { + ID string `json:"id"` + PaymentID string `json:"paymentId"` + AmountPaise int64 `json:"amountPaise"` + Status string `json:"status"` + Reason string `json:"reason,omitempty"` + Reference string `json:"reference,omitempty"` + ExternalID string `json:"externalId,omitempty"` + RequestedAt string `json:"requestedAt"` + CompletedAt string `json:"completedAt,omitempty"` +} + +func (s *Service) ListReconciliationRuns(limit int) ([]ReconciliationRunSummary, error) { + records, err := s.App.FindRecordsByFilter("reconciliation_runs", "id != ''", "-started_at", clampLimit(limit, 50, 100), 0) + if err != nil { + return nil, err + } + out := make([]ReconciliationRunSummary, 0, len(records)) + for _, r := range records { + out = append(out, ReconciliationRunSummary{ID: r.Id, Filename: r.GetString("filename"), Status: r.GetString("status"), TotalRows: r.GetInt("total_rows"), MatchedRows: r.GetInt("matched_rows"), UnmatchedRows: r.GetInt("unmatched_rows"), DuplicateRows: r.GetInt("duplicate_rows"), ConflictRows: r.GetInt("conflict_rows"), InvalidRows: r.GetInt("invalid_rows"), Error: r.GetString("error"), StartedAt: dateString(r, "started_at"), CompletedAt: dateString(r, "completed_at")}) + } + return out, nil +} +func (s *Service) ListReconciliationEntries(runID string, limit int) ([]ReconciliationEntrySummary, error) { + runID = strings.TrimSpace(runID) + if runID == "" { + return nil, fmt.Errorf("run id is required") + } + records, err := s.App.FindRecordsByFilter("reconciliation_entries", "run = {:run}", "row_number", clampLimit(limit, 250, 500), 0, dbx.Params{"run": runID}) + if err != nil { + return nil, err + } + out := make([]ReconciliationEntrySummary, 0, len(records)) + for _, r := range records { + out = append(out, ReconciliationEntrySummary{ID: r.Id, RowNumber: r.GetInt("row_number"), TransactionTime: dateString(r, "transaction_time"), AmountPaise: int64(r.GetInt("amount")), Reference: r.GetString("rrn"), Description: r.GetString("description"), Status: r.GetString("status"), PaymentID: r.GetString("payment"), Notes: r.GetString("notes")}) + } + return out, nil +} +func (s *Service) ListRefunds(limit int) ([]RefundSummary, error) { + records, err := s.App.FindRecordsByFilter("refunds", "id != ''", "-requested_at", clampLimit(limit, 50, 100), 0) + if err != nil { + return nil, err + } + out := make([]RefundSummary, 0, len(records)) + for _, r := range records { + out = append(out, RefundSummary{ID: r.Id, PaymentID: r.GetString("payment"), AmountPaise: int64(r.GetInt("amount")), Status: r.GetString("status"), Reason: r.GetString("reason"), Reference: r.GetString("reference"), ExternalID: r.GetString("external_id"), RequestedAt: dateString(r, "requested_at"), CompletedAt: dateString(r, "completed_at")}) + } + return out, nil +} + +type OperationalRecord struct { + ID string `json:"id"` + CreatedAt string `json:"createdAt,omitempty"` + Fields map[string]any `json:"fields"` +} +type operationalSpec struct { + collection, sort string + fields, dates []string +} + +func operationalRecordSpec(kind string) (operationalSpec, bool) { + switch strings.TrimSpace(strings.ToLower(kind)) { + case "sms": + return operationalSpec{"sms_events", "-created", []string{"payment_account", "source", "source_event_id", "sender", "body", "amount", "rrn", "upi_id", "payer_name", "processing_status", "matched_payment", "error"}, []string{"message_time"}}, true + case "email": + return operationalSpec{"email_events", "-created", []string{"payment_account", "source", "source_event_id", "sender", "recipient", "subject", "body", "amount", "rrn", "upi_id", "payer_name", "processing_status", "matched_payment", "error"}, []string{"message_time", "received_at"}}, true + case "audit": + return operationalSpec{"audit_events", "-occurred_at", []string{"action", "actor_email", "entity_type", "entity_id", "summary", "details"}, []string{"occurred_at"}}, true + case "webhooks": + return operationalSpec{"webhook_deliveries", "-created", []string{"event_id", "event", "payment", "status", "attempts", "response_code", "last_error"}, []string{"next_attempt_at", "last_attempt_at", "delivered_at"}}, true + default: + return operationalSpec{}, false + } +} +func (s *Service) ListOperationalRecords(kind string, limit int) ([]OperationalRecord, error) { + spec, ok := operationalRecordSpec(kind) + if !ok { + return nil, fmt.Errorf("invalid operational record kind") + } + records, err := s.App.FindRecordsByFilter(spec.collection, "id != ''", spec.sort, clampLimit(limit, 50, 100), 0) + if err != nil { + return nil, err + } + out := make([]OperationalRecord, 0, len(records)) + for _, record := range records { + fields := make(map[string]any, len(spec.fields)+len(spec.dates)) + for _, field := range spec.fields { + if value := record.Get(field); value != nil && value != "" { + fields[field] = value + } + } + for _, field := range spec.dates { + if value := dateString(record, field); value != "" { + fields[field] = value + } + } + out = append(out, OperationalRecord{ID: record.Id, CreatedAt: dateString(record, "created"), Fields: fields}) + } + return out, nil +} + +type RazorpayOrderSummary struct { + ID string `json:"id"` + AmountPaise int64 `json:"amountPaise"` + Currency string `json:"currency"` + Status string `json:"status"` + ExternalID string `json:"externalId,omitempty"` + RazorpayOrderID string `json:"razorpayOrderId,omitempty"` + RazorpayPaymentID string `json:"razorpayPaymentId,omitempty"` + ProviderStatus string `json:"providerStatus,omitempty"` + PaymentMethod string `json:"paymentMethod,omitempty"` + AmountRefunded int64 `json:"amountRefunded,omitempty"` + Error string `json:"error,omitempty"` + CreatedAt string `json:"createdAt"` + CapturedAt string `json:"capturedAt,omitempty"` +} + +func (s *Service) ListRazorpayOrders(mode string, limit int) ([]RazorpayOrderSummary, error) { + collection := "razorpay_test_orders" + if strings.TrimSpace(strings.ToLower(mode)) != "test" { + return nil, fmt.Errorf("invalid razorpay mode") + } + records, err := s.App.FindRecordsByFilter(collection, "id != ''", "-created_at", clampLimit(limit, 50, 100), 0) + if err != nil { + return nil, err + } + out := make([]RazorpayOrderSummary, 0, len(records)) + for _, r := range records { + out = append(out, RazorpayOrderSummary{ID: r.Id, AmountPaise: int64(r.GetInt("amount")), Currency: r.GetString("currency"), Status: r.GetString("status"), ExternalID: r.GetString("external_id"), RazorpayOrderID: r.GetString("razorpay_order_id"), RazorpayPaymentID: r.GetString("razorpay_payment_id"), ProviderStatus: r.GetString("provider_status"), PaymentMethod: r.GetString("payment_method"), AmountRefunded: int64(r.GetInt("amount_refunded")), Error: r.GetString("error"), CreatedAt: dateString(r, "created_at"), CapturedAt: dateString(r, "captured_at")}) + } + return out, nil +} diff --git a/internal/operatorview/service_test.go b/internal/operatorview/service_test.go new file mode 100644 index 0000000..fadd647 --- /dev/null +++ b/internal/operatorview/service_test.go @@ -0,0 +1,60 @@ +package operatorview_test + +import ( + "testing" + "time" + + "github.com/Phloraxx/payment-api/internal/config" + "github.com/Phloraxx/payment-api/internal/operatorview" + "github.com/Phloraxx/payment-api/internal/payments" + _ "github.com/Phloraxx/payment-api/migrations" + "github.com/pocketbase/pocketbase/tests" +) + +func TestOverviewAndPaymentViewsUseTypedContract(t *testing.T) { + app, err := tests.NewTestApp() + if err != nil { + t.Fatal(err) + } + defer app.Cleanup() + + cfg := config.Config{ + TestMode: true, + PaymentTTL: 5 * time.Minute, + AmountQuarantine: 24 * time.Hour, + } + paymentService := payments.NewService(app, cfg, nil) + paymentService.SuffixStart = func() (int64, error) { return 1, nil } + payment, _, err := paymentService.Create(payments.CreateInput{AmountRupees: 125, PaymentAccount: "kotak"}) + if err != nil { + t.Fatal(err) + } + + view := operatorview.New(app) + overview, err := view.Overview(5) + if err != nil { + t.Fatal(err) + } + if overview.PaymentCounts["total"] != 1 || overview.PaymentCounts["pending"] != 1 { + t.Fatalf("counts=%+v", overview.PaymentCounts) + } + if len(overview.Recent) != 1 || overview.Recent[0].ID != payment.ID { + t.Fatalf("recent=%+v", overview.Recent) + } + + pending, err := view.ListPayments("pending", 10) + if err != nil { + t.Fatal(err) + } + if len(pending) != 1 || pending[0].PayableAmountPaise != payment.PayablePaise { + t.Fatalf("pending=%+v", pending) + } + + detail, err := view.GetPayment(payment.ID) + if err != nil { + t.Fatal(err) + } + if detail.ID != payment.ID || detail.PaymentAccount != "kotak" || detail.RRN != "" { + t.Fatalf("detail=%+v", detail) + } +} diff --git a/package-lock.json b/package-lock.json index 341c6e2..0385a52 100644 --- a/package-lock.json +++ b/package-lock.json @@ -8,7 +8,6 @@ "name": "paygate-ui", "version": "1.0.0", "dependencies": { - "pocketbase": "0.27.0", "qrcode": "1.5.4", "react": "19.2.8", "react-dom": "19.2.8" @@ -1291,12 +1290,6 @@ "node": ">=10.13.0" } }, - "node_modules/pocketbase": { - "version": "0.27.0", - "resolved": "https://registry.npmjs.org/pocketbase/-/pocketbase-0.27.0.tgz", - "integrity": "sha512-K5N6d93UP/BNMbMnlZ6BUfy9VPCIvLyqhJFOsNI8OsZwzvKWEAfyD36boi5K4ECIOl5HMlo0TzuaeGdKpMwizQ==", - "license": "MIT" - }, "node_modules/postcss": { "version": "8.5.23", "resolved": "https://registry.npmjs.org/postcss/-/postcss-8.5.23.tgz", diff --git a/package.json b/package.json index c664dc1..0b0af72 100644 --- a/package.json +++ b/package.json @@ -10,7 +10,6 @@ "test": "npm run typecheck && node --test connectors/cloudflare-email-worker/worker.test.mjs && npm run build" }, "dependencies": { - "pocketbase": "0.27.0", "qrcode": "1.5.4", "react": "19.2.8", "react-dom": "19.2.8" diff --git a/web/src/App.tsx b/web/src/App.tsx index df7d121..8368d99 100644 --- a/web/src/App.tsx +++ b/web/src/App.tsx @@ -1,5 +1,5 @@ import { useEffect, useMemo, useState } from "react"; -import { pb } from "./pb"; +import { auth, refreshAuth } from "./api"; import type { Page } from "./types"; import { Dashboard } from "./pages/Dashboard"; import { Login } from "./pages/Login"; @@ -17,15 +17,15 @@ function pageFromHash(): Page { } export function App() { - const [loggedIn, setLoggedIn] = useState(pb.authStore.isValid); + const [loggedIn, setLoggedIn] = useState(auth.isValid); const [page, setPage] = useState(pageFromHash()); const [notice, setNotice] = useState(""); - useEffect(() => pb.authStore.onChange(() => setLoggedIn(pb.authStore.isValid)), []); + useEffect(() => auth.subscribe(() => setLoggedIn(auth.isValid)), []); useEffect(() => { - if (!pb.authStore.token) return; + if (!auth.token) return; const refreshAuth = async () => { - try { await pb.collection("users").authRefresh(); } catch { pb.authStore.clear(); } + try { await refreshAuth(); } catch { auth.clear(); } }; void refreshAuth(); const timer = window.setInterval(() => void refreshAuth(), 10 * 60_000); @@ -56,8 +56,8 @@ export function App() {

Operator console

- {pb.authStore.record?.email} - + {auth.email} +
diff --git a/web/src/api.ts b/web/src/api.ts new file mode 100644 index 0000000..eb54e57 --- /dev/null +++ b/web/src/api.ts @@ -0,0 +1,54 @@ +type OperatorRecord = { email?: string }; +type AuthResponse = { token?: string; record?: OperatorRecord }; +type ErrorEnvelope = { message?: string; error?: { code?: string; message?: string }; data?: Record }; + +const TOKEN_KEY = "paygate_operator_token"; +const EMAIL_KEY = "paygate_operator_email"; +const listeners = new Set<() => void>(); +let token = sessionStorage.getItem(TOKEN_KEY) ?? ""; +let email = sessionStorage.getItem(EMAIL_KEY) ?? ""; + +function emit() { listeners.forEach((listener) => listener()); } +function save(nextToken: string, nextEmail: string) { + token = nextToken.trim(); email = nextEmail.trim(); + if (token) sessionStorage.setItem(TOKEN_KEY, token); else sessionStorage.removeItem(TOKEN_KEY); + if (email) sessionStorage.setItem(EMAIL_KEY, email); else sessionStorage.removeItem(EMAIL_KEY); + emit(); +} +export const auth = { + get token() { return token; }, + get email() { return email; }, + get isValid() { return token.length > 0; }, + clear() { save("", ""); }, + subscribe(listener: () => void) { listeners.add(listener); return () => { listeners.delete(listener); }; }, +}; + +async function parse(response: Response): Promise { + const body = (await response.json().catch(() => ({}))) as ErrorEnvelope & T; + if (!response.ok) { + const fieldError = body.data ? Object.values(body.data).find((value) => value?.message)?.message : undefined; + throw new Error(body.error?.message ?? fieldError ?? body.message ?? `Request failed with HTTP ${response.status}`); + } + return body as T; +} +export async function login(emailValue: string, password: string) { + const response = await fetch("/api/collections/users/auth-with-password", { method: "POST", headers: { "Content-Type": "application/json", Accept: "application/json" }, body: JSON.stringify({ identity: emailValue.trim(), password }) }); + const body = await parse(response); + if (!body.token) throw new Error("PayGate returned no operator token"); + save(body.token, body.record?.email ?? emailValue.trim()); +} +export async function refreshAuth() { + if (!token) return; + const response = await fetch("/api/collections/users/auth-refresh", { method: "POST", headers: { Authorization: `Bearer ${token}`, Accept: "application/json" } }); + const body = await parse(response); + if (!body.token) throw new Error("PayGate returned no refreshed token"); + save(body.token, body.record?.email ?? email); +} +export async function api(path: string, init: RequestInit = {}): Promise { + const headers = new Headers(init.headers); + if (init.body && !(init.body instanceof FormData) && !headers.has("Content-Type")) headers.set("Content-Type", "application/json"); + if (token) headers.set("Authorization", `Bearer ${token}`); + const response = await fetch(path, { ...init, headers }); + try { return await parse(response); } + catch (error) { if (response.status === 401 && token) auth.clear(); throw error; } +} diff --git a/web/src/pages/Dashboard.tsx b/web/src/pages/Dashboard.tsx index a6570b3..8b7ede4 100644 --- a/web/src/pages/Dashboard.tsx +++ b/web/src/pages/Dashboard.tsx @@ -1,36 +1,27 @@ import { useCallback, useEffect, useState } from "react"; import { Badge, formatDate, Stat } from "../components/common"; -import { api, pb } from "../pb"; -import type { DashboardData } from "../types"; +import { api } from "../api"; +import type { OperatorOverviewResponse } from "../types"; import { PaymentTable } from "./Payments"; export function Dashboard() { - const [data, setData] = useState(null); + const [data, setData] = useState(null); const [error, setError] = useState(""); const load = useCallback(async () => { try { - setData(await api("/api/dashboard")); + setData(await api("/api/operator/v2/overview?limit=8")); setError(""); - } catch (err) { - setError(err instanceof Error ? err.message : "Could not load dashboard"); - } + } catch (err) { setError(err instanceof Error ? err.message : "Could not load dashboard"); } }, []); useEffect(() => { void load(); - let disposed = false; - const unsubscribers: Array<() => void> = []; - for (const collection of ["payments", "review_cases", "alerts", "refunds"]) { - void pb.collection(collection).subscribe("*", () => void load()).then((fn) => { - if (disposed) void fn(); else unsubscribers.push(fn); - }); - } - const timer = window.setInterval(() => void load(), 30_000); - return () => { disposed = true; unsubscribers.forEach((fn) => fn()); window.clearInterval(timer); }; + const timer = window.setInterval(() => { if (document.visibilityState === "visible") void load(); }, 10_000); + return () => window.clearInterval(timer); }, [load]); - const stats = data?.stats ?? {}; + const stats = data?.overview.paymentCounts ?? {}; const connector = data?.connector; const capacityPools = data?.capacity?.pools?.slice(0, 8) ?? []; const backup = data?.backup; @@ -42,46 +33,18 @@ export function Dashboard() { - 0 ? "warn" : ""} /> - 0 ? "warn" : ""} /> + 0 ? "warn" : ""} /> + 0 ? "warn" : ""} />
-
-
-

GOOGLE MESSAGES

-

{connector?.enabled ? connector.state.replaceAll("_", " ") : "disabled"}

-

{connector?.lastError || (connector?.phoneResponsive ? `Phone responding · last bank SMS ${formatDate(connector.lastMessageAt)}` : "Waiting for phone response")}

-
- -
-
-
-

ANDROID RELAY

-

{relay?.ready ? "ready" : "unavailable"}

-

{relay ? `${relay.activeDevices}/${relay.enabledDevices} active · ${relay.powerUnhealthyDevices} power-unhealthy · last heartbeat ${formatDate(relay.lastHeartbeatAt ?? undefined)} · queue ${relay.pendingQueueCount} pending / ${relay.failedQueueCount} failed · ${relay.recentErrorCount} server errors/24h` : "Relay status unavailable"}

-
- -
-
-
-

BACKUPS

-

{backup?.enabled ? `${backup.backupCount} available` : "disabled"}

-

{backup?.error || (backup?.latest ? `Latest ${backup.latest.name} · ${formatDate(backup.latest.modTime)}` : backup?.enabled ? "No backup created yet" : "Configure a backup cron")}

-
- -
+

GOOGLE MESSAGES

{connector?.enabled ? connector.state.replaceAll("_", " ") : "disabled"}

{connector?.lastError || (connector?.phoneResponsive ? `Phone responding · last bank SMS ${formatDate(connector.lastMessageAt)}` : "Waiting for phone response")}

+

ANDROID RELAY

{relay?.ready ? "ready" : "unavailable"}

{relay ? `${relay.activeDevices}/${relay.enabledDevices} active · ${relay.powerUnhealthyDevices} power-unhealthy · heartbeat ${formatDate(relay.lastHeartbeatAt ?? undefined)} · queue ${relay.pendingQueueCount}/${relay.failedQueueCount}` : "Relay status unavailable"}

+

BACKUPS

{backup?.enabled ? `${backup.backupCount} available` : "disabled"}

{backup?.error || (backup?.latest ? `Latest ${backup.latest.name} · ${formatDate(backup.latest.modTime)}` : backup?.enabled ? "No backup created yet" : "Configure a backup cron")}

99-SUFFIX POOLS

Fingerprint capacity

70% warning · 95% critical
- {!capacityPools.length ?

No active or quarantined fingerprint pools.

:
{capacityPools.map((pool) =>
-
₹{pool.requestedAmount}{pool.pending} pending · {pool.quarantined} quarantined · {pool.available} available
- - -
)}
} -
-
-

Recent payments

Realtime updates
- + {!capacityPools.length ?

No active or quarantined fingerprint pools.

:
{capacityPools.map((pool) =>
₹{pool.requestedAmount}{pool.pending} pending · {pool.quarantined} quarantined · {pool.available} available
)}
}
+

Recent payments

Typed API · 5s refresh
; } diff --git a/web/src/pages/Login.tsx b/web/src/pages/Login.tsx index 80ebdcb..035ef4a 100644 --- a/web/src/pages/Login.tsx +++ b/web/src/pages/Login.tsx @@ -1,5 +1,5 @@ import { useState, type FormEvent } from "react"; -import { pb } from "../pb"; +import { login } from "../api"; export function Login() { const [email, setEmail] = useState(""); @@ -12,7 +12,7 @@ export function Login() { setBusy(true); setError(""); try { - await pb.collection("users").authWithPassword(email.trim(), password); + await login(email, password); } catch { setError("Login failed. Check the operator credentials."); } finally { diff --git a/web/src/pages/Operations.tsx b/web/src/pages/Operations.tsx index 4c2bfa0..2baaf5f 100644 --- a/web/src/pages/Operations.tsx +++ b/web/src/pages/Operations.tsx @@ -1,123 +1,85 @@ import { useCallback, useEffect, useMemo, useState, type FormEvent, type ReactNode } from "react"; -import type { RecordModel } from "pocketbase"; import { Badge, formatDate, Modal } from "../components/common"; -import { api, pb } from "../pb"; -import type { AlertRecord, ReconciliationRun, RefundRecord, ReviewCase } from "../types"; +import { api } from "../api"; +import type { OperatorAlertSummary, OperatorEvidenceDetail, OperatorReconciliationEntry, OperatorReconciliationRun, OperatorRefund, OperatorReviewDetail, OperatorReviewSummary } from "../types"; export function ReviewsPage({ notify }: { notify: (value: string) => void }) { - const [records, setRecords] = useState([]); - const [selected, setSelected] = useState(null); + const [records, setRecords] = useState([]); + const [selected, setSelected] = useState(null); const [error, setError] = useState(""); const [showResolved, setShowResolved] = useState(false); const load = useCallback(async () => { try { - const filter = showResolved ? "" : 'status = "open"'; - const result = await pb.collection("review_cases").getList(1, 100, { - sort: "-opened_at", filter, - expand: "sms_event,email_event,reconciliation_entry,payment,resolved_by", - }); - setRecords(result.items); - if (selected) setSelected(result.items.find((item) => item.id === selected.id) ?? null); + const status = showResolved ? "" : "&status=open"; + const result = await api<{ reviews: OperatorReviewSummary[] }>(`/api/operator/v2/reviews?limit=100${status}`); + setRecords(result.reviews); + if (selected) setSelected(await api(`/api/operator/v2/reviews/${encodeURIComponent(selected.id)}`)); setError(""); - } catch (err) { - setError(err instanceof Error ? err.message : "Could not load review cases"); - } + } catch (err) { setError(err instanceof Error ? err.message : "Could not load review cases"); } }, [showResolved, selected?.id]); - useRealtimeCollection("review_cases", load); - useEffect(() => { void load(); }, [load]); + useEffect(() => { + void load(); + const timer = window.setInterval(() => { if (document.visibilityState === "visible") void load(); }, 8_000); + return () => window.clearInterval(timer); + }, [load]); + + async function open(review: OperatorReviewSummary) { + try { setSelected(await api(`/api/operator/v2/reviews/${encodeURIComponent(review.id)}`)); } + catch (err) { notify(err instanceof Error ? err.message : "Could not load review details."); } + } return
-
-

FAIL-CLOSED EVIDENCE

Attention required

Unmatched or incomplete bank evidence is never assigned automatically.

- -
- {error &&

{error}

} - {!error && !records.length &&

No review cases need attention.

} -
- {records.map((record) => setSelected(record)}> - - - - )}
OpenedTypeSeverityEvidenceStatus
{formatDate(record.opened_at || record.created)}{record.kind}{record.payment || record.sms_event || record.email_event || record.reconciliation_entry || "—"}{record.reason}
+

FAIL-CLOSED EVIDENCE

Attention required

Unmatched or incomplete evidence is never assigned automatically.

+ {error &&

{error}

}{!error && !records.length &&

No review cases need attention.

} +
{records.map((record) => void open(record)}>)}
OpenedTypeSeverityPaymentStatus
{formatDate(record.openedAt)}{record.kind}{record.paymentId || "—"}{record.reason}
{selected && setSelected(null)} onResolved={async (message) => { notify(message); setSelected(null); await load(); }} />}
; } -function ReviewModal({ review, notify, onClose, onResolved }: { review: ReviewCase; notify: (value: string) => void; onClose: () => void; onResolved: (message: string) => Promise }) { - const candidates = Array.isArray(review.candidate_payment_ids) ? review.candidate_payment_ids : []; +function ReviewModal({ review, notify, onClose, onResolved }: { review: OperatorReviewDetail; notify: (value: string) => void; onClose: () => void; onResolved: (message: string) => Promise }) { + const candidates = review.candidatePaymentIds ?? []; const [action, setAction] = useState(review.status === "open" ? "manual_match" : review.resolution || "dismissed"); - const [paymentId, setPaymentId] = useState(review.payment || candidates[0] || ""); + const [paymentId, setPaymentId] = useState(review.paymentId || candidates[0] || ""); const [bankReference, setBankReference] = useState(""); const [note, setNote] = useState(""); const [saving, setSaving] = useState(false); - const evidence = review.expand?.sms_event ?? review.expand?.email_event ?? review.expand?.reconciliation_entry; - async function resolve(event: FormEvent) { - event.preventDefault(); - setSaving(true); + event.preventDefault(); setSaving(true); try { - await api(`/api/review-cases/${review.id}/resolve`, { - method: "POST", - body: JSON.stringify({ action, paymentId: paymentId || undefined, bankReference: bankReference || undefined, note }), - }); + await api(`/api/operator/v2/reviews/${review.id}/resolve`, { method: "POST", body: JSON.stringify({ action, paymentId: paymentId || undefined, bankReference: bankReference || undefined, note }) }); await onResolved(action === "manual_match" ? "Evidence matched and audited." : "Review case resolved and audited."); - } catch (err) { - notify(err instanceof Error ? err.message : "Review resolution failed."); - } finally { - setSaving(false); - } + } catch (err) { notify(err instanceof Error ? err.message : "Review resolution failed."); } + finally { setSaving(false); } } - return -
- {review.kind} - {review.reason}{formatDate(review.opened_at)} - {candidates.length ? candidates.join(", ") : "—"} -
- {evidence && } - {review.status === "open" ?
- - {action === "manual_match" && <> - - - } -