feat(payments): payment-event dispatcher + the provider-return UX
CI / changes (pull_request) Successful in 3s
CI / unit (pull_request) Successful in 10s
CI / integration (pull_request) Successful in 22s
CI / ui (pull_request) Successful in 1m9s
CI / conformance (pull_request) Successful in 10s
CI / gate (pull_request) Successful in 0s
CI / deploy (pull_request) Successful in 1m45s
CI / changes (pull_request) Successful in 3s
CI / unit (pull_request) Successful in 10s
CI / integration (pull_request) Successful in 22s
CI / ui (pull_request) Successful in 1m9s
CI / conformance (pull_request) Successful in 10s
CI / gate (pull_request) Successful in 0s
CI / deploy (pull_request) Successful in 1m45s
Deliver payment_events to connected clients as an in-app wallet-refresh push: a background dispatcher drains undispatched events and publishes a KindNotification "payment" signal, marking each delivered; the client bumps a wallet-refresh counter the open Wallet screen watches, re-fetching in place. A return-focus refetch is the fallback. The Robokassa Success/Fail return now serves a self-closing page (the payment opens in a separate window) so the customer drops back into the live app instead of a cold start. Integration test for the event drain/mark queue.
This commit is contained in:
@@ -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,
|
||||
|
||||
@@ -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 {
|
||||
|
||||
@@ -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) {
|
||||
|
||||
@@ -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
|
||||
|
||||
@@ -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())
|
||||
}
|
||||
|
||||
@@ -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
|
||||
|
||||
@@ -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(`<!doctype html>
|
||||
<html lang="ru"><head><meta charset="utf-8"><meta name="viewport" content="width=device-width, initial-scale=1"><title>Оплата</title></head>
|
||||
<body style="font-family: system-ui, sans-serif; text-align: center; padding: 48px 20px; color: #1a1c20">
|
||||
<p style="font-size: 1.1rem">` + message + `</p>
|
||||
<p style="color: #5b6472">Можно закрыть это окно и вернуться в приложение.</p>
|
||||
<p><a href="/app/">Открыть приложение</a></p>
|
||||
<script>setTimeout(function () { try { window.close(); } catch (e) {} }, 1200);</script>
|
||||
</body></html>`)
|
||||
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)
|
||||
})
|
||||
}
|
||||
|
||||
|
||||
@@ -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();
|
||||
}
|
||||
},
|
||||
|
||||
@@ -1,7 +1,7 @@
|
||||
<script lang="ts">
|
||||
import { onMount } from 'svelte';
|
||||
import Modal from '../components/Modal.svelte';
|
||||
import { handleError } from '../lib/app.svelte';
|
||||
import { app, handleError } from '../lib/app.svelte';
|
||||
import { gateway } from '../lib/gateway';
|
||||
import { t, i18n, type MessageKey } from '../lib/i18n/index.svelte';
|
||||
import { executionContext, formatAmount, needsWebSpendWarning, type SpendContext } from '../lib/wallet';
|
||||
@@ -39,7 +39,26 @@
|
||||
loading = false;
|
||||
}
|
||||
}
|
||||
onMount(load);
|
||||
onMount(() => {
|
||||
void load();
|
||||
// Fallback to the payment push: re-fetch when the tab regains focus, i.e. the user returned
|
||||
// from the provider's payment window.
|
||||
const onVisible = () => {
|
||||
if (!document.hidden) void load();
|
||||
};
|
||||
document.addEventListener('visibilitychange', onVisible);
|
||||
return () => document.removeEventListener('visibilitychange', onVisible);
|
||||
});
|
||||
|
||||
// Main path: a payment-intake push bumps app.walletRefresh; re-fetch in place when it changes
|
||||
// (a credit landed while the screen is open).
|
||||
let seenRefresh = app.walletRefresh;
|
||||
$effect(() => {
|
||||
if (app.walletRefresh !== seenRefresh) {
|
||||
seenRefresh = app.walletRefresh;
|
||||
void load();
|
||||
}
|
||||
});
|
||||
|
||||
function sourceLabel(source: string): string {
|
||||
return t(`wallet.source.${source}` as MessageKey);
|
||||
|
||||
Reference in New Issue
Block a user