diff --git a/PLAN.md b/PLAN.md index 3817dc0..854a31d 100644 --- a/PLAN.md +++ b/PLAN.md @@ -515,8 +515,11 @@ chargeback **never drives the balance negative** (D27 stands, `balances_chips_ch the offer page (`/offer/`), the order/`fund` engine (idempotent, honours an expired order), the `internal/robokassa` adapter, the `POST /wallet/order` + internal Result-callback handlers (a D36 confirmed-email gate on `direct`), the pending reaper, the `wallet.order` edge wire + the public -`/pay/*` routes, the Wallet purchase CTA, and the contour deploy env (IsTest forced). Remaining: -the `payment_events` dispatcher (delivery), then the VK and TG-Stars rails and refunds. +`/pay/*` routes, the Wallet purchase CTA, the contour deploy env (IsTest forced), and the +`payment_events` dispatcher — an in-app wallet-refresh push (KindNotification `"payment"`) with a +self-closing provider-return page (the payment opens in a separate window) and a return-focus +refetch fallback. Remaining: the VK and TG-Stars rails and refunds; and hiding the ad banner on a +no-ads purchase (a spend-path `NotifyBanner`, deferred with the owner's agreement). **Goal.** Accept real money on all three rails into the payments domain: order-flow, verified provider callbacks, idempotency, the TG bot SQLite outbox, the event dispatcher, diff --git a/backend/cmd/backend/main.go b/backend/cmd/backend/main.go index 6c3fc64..cbdbc88 100644 --- a/backend/cmd/backend/main.go +++ b/backend/cmd/backend/main.go @@ -320,6 +320,9 @@ func run(ctx context.Context, cfg config.Config, logger *zap.Logger) error { // Sweep expired pending payment orders on a cadence (cosmetic hygiene; a late valid callback // still credits). Runs until ctx is cancelled. go runOrderReaper(ctx, paymentsSvc, logger) + // Deliver pending payment_events to connected clients as an in-app wallet-refresh push (the + // credit already landed in the ledger; a return-focus poll is the client-side fallback). + go runPaymentDispatcher(ctx, paymentsSvc, hub, logger) errc := make(chan error, 2) go func() { errc <- pushSrv.Run(ctx) }() @@ -350,6 +353,32 @@ func runOrderReaper(ctx context.Context, p *payments.Service, log *zap.Logger) { } } +// runPaymentDispatcher delivers pending payment_events to connected clients as a wallet-refresh +// signal (KindNotification / "payment"), marking each delivered, until ctx is cancelled. The credit +// already committed to the ledger; this is only the in-app push so an open wallet updates in place. +func runPaymentDispatcher(ctx context.Context, p *payments.Service, pub notify.Publisher, log *zap.Logger) { + t := time.NewTicker(3 * time.Second) + defer t.Stop() + for { + select { + case <-ctx.Done(): + return + case <-t.C: + evs, err := p.UndispatchedEvents(ctx, 50) + if err != nil { + log.Warn("payment dispatcher: read failed", zap.Error(err)) + continue + } + for _, e := range evs { + pub.Publish(notify.Notification(e.AccountID, notify.NotifyPayment)) + if err := p.MarkEventDispatched(ctx, e.EventID); err != nil { + log.Warn("payment dispatcher: mark failed", zap.String("event", e.EventID.String()), zap.Error(err)) + } + } + } + } +} + // newMailer builds the confirm-code mailer: an SMTP relay when a host is // configured, otherwise the development log mailer (the code is logged, not sent). func newMailer(cfg account.SMTPConfig, logger *zap.Logger) account.Mailer { diff --git a/backend/internal/inttest/payments_intake_test.go b/backend/internal/inttest/payments_intake_test.go index 756d73c..17b4f2b 100644 --- a/backend/internal/inttest/payments_intake_test.go +++ b/backend/internal/inttest/payments_intake_test.go @@ -101,6 +101,45 @@ func TestPaymentsFundAmountMismatch(t *testing.T) { } } +// TestPaymentsEventDispatchDrain verifies the payment_events dispatcher queue: a recorded event is +// returned as undispatched until marked, then drops out (so the dispatcher delivers it once). +func TestPaymentsEventDispatchDrain(t *testing.T) { + ctx := context.Background() + svc := newPaymentsService() + acc := uuid.New() + if err := svc.RecordPaymentEvent(ctx, acc, nil, "succeeded", []byte(`{"chips":10,"source":"direct"}`)); err != nil { + t.Fatalf("record event: %v", err) + } + + // testDB is shared, so filter the queue to our account. + find := func() *payments.PaymentEvent { + evs, err := svc.UndispatchedEvents(ctx, 100) + if err != nil { + t.Fatalf("undispatched: %v", err) + } + for i := range evs { + if evs[i].AccountID == acc { + return &evs[i] + } + } + return nil + } + + mine := find() + if mine == nil { + t.Fatal("recorded event not in the undispatched queue") + } + if mine.Type != "succeeded" { + t.Errorf("event type = %s, want succeeded", mine.Type) + } + if err := svc.MarkEventDispatched(ctx, mine.EventID); err != nil { + t.Fatalf("mark dispatched: %v", err) + } + if find() != nil { + t.Error("event still undispatched after MarkEventDispatched") + } +} + // TestPaymentsExpiredOrderStillCredits verifies an expired pending order is still honoured by a // later valid callback (§9/D23: expiry is cosmetic, the money is real). func TestPaymentsExpiredOrderStillCredits(t *testing.T) { diff --git a/backend/internal/notify/notify.go b/backend/internal/notify/notify.go index e95f0f2..8cfe7c8 100644 --- a/backend/internal/notify/notify.go +++ b/backend/internal/notify/notify.go @@ -73,6 +73,10 @@ const ( // (e.g. an email was confirmed via the one-tap deeplink opened in another browser), // so it re-fetches profile.get. It carries no payload. In-app only. NotifyProfile = "profile" + // NotifyPayment tells the client that the viewer's wallet changed from a payment-intake event + // (a credit or a refund), so it re-fetches the wallet in place. It carries no payload. In-app + // only; the payment_events dispatcher emits it after the crediting transaction commits. + NotifyPayment = "payment" // NotifyUserBlocked confirms to the blocker that a per-user block took effect, // carrying the blocked account, so every one of the blocker's sessions updates the // in-game block/add-friend controls and the struck name in place. It is delivered diff --git a/backend/internal/payments/service_intake.go b/backend/internal/payments/service_intake.go index e57aac9..ccf0ffe 100644 --- a/backend/internal/payments/service_intake.go +++ b/backend/internal/payments/service_intake.go @@ -77,3 +77,14 @@ func (s *Service) ExpireOrders(ctx context.Context) (int, error) { func (s *Service) RecordPaymentEvent(ctx context.Context, accountID uuid.UUID, orderID *uuid.UUID, eventType string, payload []byte) error { return s.store.insertPaymentEvent(ctx, accountID, orderID, eventType, payload, s.clock()) } + +// UndispatchedEvents returns up to limit payment events awaiting delivery. The dispatcher drains +// them and marks each delivered via MarkEventDispatched. +func (s *Service) UndispatchedEvents(ctx context.Context, limit int) ([]PaymentEvent, error) { + return s.store.undispatchedEvents(ctx, limit) +} + +// MarkEventDispatched stamps a payment event as delivered so it is not re-sent. +func (s *Service) MarkEventDispatched(ctx context.Context, eventID uuid.UUID) error { + return s.store.markEventDispatched(ctx, eventID, s.clock()) +} diff --git a/backend/internal/payments/store_intake.go b/backend/internal/payments/store_intake.go index e9b49ae..605a6d2 100644 --- a/backend/internal/payments/store_intake.go +++ b/backend/internal/payments/store_intake.go @@ -292,6 +292,42 @@ func (s *Store) insertPaymentEvent(ctx context.Context, accountID uuid.UUID, ord return nil } +// PaymentEvent is an undispatched lifecycle event the dispatcher delivers to an account. +type PaymentEvent struct { + EventID uuid.UUID + AccountID uuid.UUID + Type string +} + +// undispatchedEvents reads up to limit payment events not yet delivered, oldest first. +func (s *Store) undispatchedEvents(ctx context.Context, limit int) ([]PaymentEvent, error) { + rows, err := s.db.QueryContext(ctx, + `SELECT event_id, account_id, type FROM payments.payment_events + WHERE dispatched_at IS NULL ORDER BY created_at LIMIT $1`, limit) + if err != nil { + return nil, fmt.Errorf("payments: read undispatched events: %w", err) + } + defer rows.Close() + var out []PaymentEvent + for rows.Next() { + var e PaymentEvent + if err := rows.Scan(&e.EventID, &e.AccountID, &e.Type); err != nil { + return nil, fmt.Errorf("payments: scan event: %w", err) + } + out = append(out, e) + } + return out, rows.Err() +} + +// markEventDispatched stamps an event as delivered so it is not re-sent. +func (s *Store) markEventDispatched(ctx context.Context, eventID uuid.UUID, now time.Time) error { + if _, err := s.db.ExecContext(ctx, + `UPDATE payments.payment_events SET dispatched_at = $2 WHERE event_id = $1`, eventID, now); err != nil { + return fmt.Errorf("payments: mark event dispatched: %w", err) + } + return nil +} + // orderTTL reads the configured pending-order lifetime in whole seconds. func (s *Store) orderTTL(ctx context.Context) (int, error) { var cfg model.Config diff --git a/gateway/internal/connectsrv/server.go b/gateway/internal/connectsrv/server.go index 2aaac25..9549732 100644 --- a/gateway/internal/connectsrv/server.go +++ b/gateway/internal/connectsrv/server.go @@ -199,8 +199,8 @@ func (s *Server) HTTPHandler() http.Handler { // Direct-rail (Robokassa) return + callback routes: the server Result callback (the single // crediting signal, proxied to the backend intake) and the browser Success/Fail redirects. mux.Handle("/pay/robokassa/result", s.robokassaResultHandler()) - mux.Handle("/pay/robokassa/success", s.robokassaRedirectHandler()) - mux.Handle("/pay/robokassa/fail", s.robokassaRedirectHandler()) + mux.Handle("/pay/robokassa/success", s.robokassaReturnHandler("Оплата принята.")) + mux.Handle("/pay/robokassa/fail", s.robokassaReturnHandler("Оплата не завершена.")) // The client posts its local-move-preview adoption telemetry here (session-gated). mux.Handle("/metrics/local-eval", s.localEvalMetricsHandler()) // The index.html boot guard beacons here when it turns a client away on the unsupported-engine @@ -522,12 +522,24 @@ func (s *Server) robokassaResultHandler() http.Handler { }) } -// robokassaRedirectHandler redirects the customer's browser back into the app after Robokassa's -// Success/Fail return. The credit rides the server Result callback, never these redirects, so this -// only navigates home; the wallet reflects the credit once the callback lands. -func (s *Server) robokassaRedirectHandler() http.Handler { +// robokassaReturnHandler serves a minimal self-closing page for Robokassa's Success/Fail browser +// return. The payment opens in a separate window, so on return this closes that window and drops the +// customer back into the live app (never a cold app start); a link is the fallback if the window +// cannot self-close. The credit rides the server Result callback, never this redirect — the wallet +// updates in place from the payment push, with a return-focus refetch as the fallback. +func (s *Server) robokassaReturnHandler(message string) http.Handler { + page := []byte(` +
` + message + `
+Можно закрыть это окно и вернуться в приложение.
+ + +`) return http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { - http.Redirect(w, r, "/app/", http.StatusSeeOther) + w.Header().Set("Content-Type", "text/html; charset=utf-8") + w.Header().Set("Cache-Control", "no-store") + _, _ = w.Write(page) }) } diff --git a/ui/src/lib/app.svelte.ts b/ui/src/lib/app.svelte.ts index 173f59d..8c4882b 100644 --- a/ui/src/lib/app.svelte.ts +++ b/ui/src/lib/app.svelte.ts @@ -137,6 +137,9 @@ export const app = $state<{ /** Whether an operator feedback reply awaits the player, for the lobby ⚙️ badge (combined * with friend requests) and the Settings → Info badge. */ feedbackReplyUnread: boolean; + /** A monotone counter bumped when a payment-intake push signals the wallet changed; an open + * Wallet screen watches it and re-fetches in place. */ + walletRefresh: number; /** Whether to show the "outdated invite link" notice: set when a Telegram deep-link friend * code is already used/expired, so the visitor lands in the lobby with a gentle pointer to * the bot instead of a scary error on the Friends screen. */ @@ -175,6 +178,7 @@ export const app = $state<{ chatUnread: {}, messageUnread: {}, feedbackReplyUnread: false, + walletRefresh: 0, staleInvite: false, welcomeRedeem: false, resync: 0, @@ -439,6 +443,12 @@ function openStream(): void { if (e.sub === 'profile') { void refreshProfile(); } + // A payment-intake event credited (or refunded) the viewer's wallet: bump the signal an + // open Wallet screen watches, so it re-fetches in place (the return-focus poll is the + // fallback when the live stream is down). + if (e.sub === 'payment') { + app.walletRefresh++; + } void refreshNotifications(); } }, diff --git a/ui/src/screens/Wallet.svelte b/ui/src/screens/Wallet.svelte index 23c5a29..8377342 100644 --- a/ui/src/screens/Wallet.svelte +++ b/ui/src/screens/Wallet.svelte @@ -1,7 +1,7 @@