openvibers/openvibe.live Commits
PowerChat: reconcile missed checkout webhooks from /paid-messages
Changes
@@@ -5080,6 +5080,16 @@ function getPaymentOrderByRef(provider, ref) {return get('SELECT * FROM payment_orders WHERE provider = ? AND provider_ref = ? ORDER BY id DESC LIMIT 1', [provider, ref]);}++// PowerChat checkouts still waiting for their donation.completed webhook — the++// reconciliation sweep's work list (recent only: an intent lives an hour, and an++// order nobody paid within days is an abandoned cart, not a missed webhook).++function getPendingPowerchatOrders(days = 3) {++ return all(`SELECT * FROM payment_orders++ WHERE provider = 'powerchat' AND status = 'pending'++ AND created_at >= datetime('now', ?)++ ORDER BY id ASC`, [`-${Math.max(1, Math.round(days))} days`]);++}++function updatePaymentOrder(id, fields) {const allowed = new Set(['provider_ref', 'status', 'amount_cents', 'bucks', 'currency', 'streamer_id']);const entries = Object.entries(fields || {}).filter(([k]) => allowed.has(k));@@@ -7964,7 +7974,7 @@ module.exports = {getPowerchatConnection, getPowerchatConnectionByUsername, getPowerchatConnectionByPcUserId,upsertPowerchatConnection, updatePowerchatTokens, setPowerchatConnectionError,deletePowerchatConnection, powerchatDeliveryIsNew, cleanupPowerchatDeliveries,-- createPaymentOrder, getPaymentOrderById, getPaymentOrderByRef, updatePaymentOrder,++ createPaymentOrder, getPaymentOrderById, getPaymentOrderByRef, updatePaymentOrder, getPendingPowerchatOrders,upsertSubscription, getSubscriptionByProviderRef, getActiveSubscription, isActiveSubscriber,getSubscriptionsByStreamer, getSubscriptionsBySubscriber, getActiveSubscriberCount, setSubscriptionStatus,getSubscriptionsDueRenewal,@@@ -1018,6 +1018,8 @@ async function start() {try { require('./integrations/powerchat-platform').startCurrencyEarnFlusher(); } catch (e) { console.warn('[PowerChat] earn flusher not started:', e.message); }// Read /chat/history back so a 202-accepted relay that moderation dropped is visible.try { require('./integrations/powerchat-platform').startChatVerifier(); } catch (e) { console.warn('[PowerChat] chat verifier not started:', e.message); }++ // Backfill checkout orders whose donation.completed webhook never arrived (/paid-messages).++ try { require('./integrations/powerchat-reconcile').startReconciler(); } catch (e) { console.warn('[PowerChat] reconciler not started:', e.message); }// Auto-renew lapsed channel subs (Vibes-balance renewal; Stripe renews natively).try { require('./monetization/payments').startRenewalSweeper(); } catch (e) { console.warn('[Payments] renewal sweeper not started:', e.message); }// 5-minute live-viewer samples → the home hero's 24h sparkline.nnew file mode 100644@@@ -0,0 +1,175 @@++/**++ * powerchat-reconcile.js — backfill missed donation.completed webhooks from++ * GET /streamers/:username/paid-messages (scope paid_messages:read).++ *++ * The signed webhook is the ONLY authoritative confirmation and stays the primary++ * credit path. But webhooks can be missed for good: the receiver was down past++ * PowerChat's ~24h retry tail, ~20 consecutive failures tripped its circuit++ * breaker, or a 410 disabled the endpoint. Until now a viewer who paid during++ * such a window simply never got their Vibes / subscription.++ *++ * /paid-messages is the same confirmed money as a cursor-paginated feed, and every++ * row echoes `appExternalRef` — the pcorder:/pcsub: ref we minted onto the checkout++ * link — so paid-but-uncredited orders are found by exact match, never guessed.++ *++ * Safe to run repeatedly and concurrently with live webhooks: everything is credited++ * through the same handleAttributedDonation() the webhook uses, which is idempotent++ * on the order's status. Work-list driven: no pending order → no API call at all.++ *++ * Envelope: written against { data: { rows, nextCursor } } (the corrected upstream++ * shape) and defensive on older deployments that left nextCursor outside `data`.++ */++'use strict';++++const db = require('../db/database');++const oauth = require('./powerchat-oauth');++const checkout = require('./powerchat-checkout');++++const PAGE_LIMIT = 100;++const MAX_PAGES_PER_HOST = 5; // 500 newest confirmed tips per host per sweep++const LOOKBACK_DAYS = 3; // intents live 1h; older pending orders are abandoned carts++const GRACE_MS = 2 * 60 * 60 * 1000; // rows may be a little older than the order (clock skew, slow checkout)++const SWEEP_MS = 15 * 60 * 1000;++const FIRST_SWEEP_MS = 2 * 60 * 1000;++++function _sqliteTs(s) {++ if (!s) return 0;++ const t = Date.parse(String(s).includes('T') ? String(s) : String(s).replace(' ', 'T') + 'Z');++ return Number.isFinite(t) ? t : 0;++}++function _testAllowed() {++ try { const v = db.getSetting('powerchat_allow_test_fulfillment'); return v === true || v === 'true' || v === 1 || v === '1'; }++ catch { return false; }++}++function _hasScope(conn, scope) {++ return !!(conn && conn.access_token && conn.refresh_token && (!conn.scope || String(conn.scope).split(/\s+/).includes(scope)));++}++++// Which PowerChat account hosted this order's checkout (where the money landed), and++// which local user "receives" it for the fulfillment pipeline.++// pcsub direct → the streamer's own PowerChat; site → the site tips account.++function _hostFor(order) {++ const mode = String(order.provider_ref || 'site').split(':')[0];++ if (order.kind === 'subscription' && mode === 'direct' && order.streamer_id) {++ const conn = db.getPowerchatConnection(order.streamer_id);++ return conn && conn.powerchat_username ? { username: conn.powerchat_username, conn, receivingUserId: order.streamer_id } : null;++ }++ const site = checkout.getSiteAccount();++ if (!site) return null;++ const conn = db.getPowerchatConnectionByUsername(site.username);++ return { username: site.username, conn: conn || null, receivingUserId: conn ? conn.user_id : null };++}++function _refFor(order) {++ return order.kind === 'subscription' ? `pcsub:${order.id}` : `pcorder:${order.id}`;++}++function _page(json) {++ const d = json && json.data !== undefined ? json.data : json;++ const rows = Array.isArray(d) ? d : (d && Array.isArray(d.rows) ? d.rows : []);++ // Corrected shape nests nextCursor in data; older deployments put it beside it.++ const nextCursor = (d && !Array.isArray(d) && d.nextCursor != null) ? d.nextCursor++ : (json && json.nextCursor != null ? json.nextCursor : null);++ return { rows, nextCursor: nextCursor || null };++}++++const _hostWarned = new Map();++function _warnHost(username, msg) {++ const now = Date.now();++ if ((_hostWarned.get(username) || 0) > now - 3600000) return;++ _hostWarned.set(username, now);++ console.warn(`[PowerChat] reconcile: ${msg} (@${username})`);++}++++/** One sweep. Returns a summary; never throws. */++async function reconcileOnce() {++ const summary = { orders: 0, hosts: 0, pages: 0, rows: 0, credited: [], underpaid: [], skippedTest: 0, skippedHosts: 0, errors: 0 };++ let pending;++ try { pending = db.getPendingPowerchatOrders(LOOKBACK_DAYS) || []; } catch (e) { summary.errors++; return summary; }++ summary.orders = pending.length;++ if (!pending.length) return summary;++++ // Group the work list by host account: one paged read per host, not per order.++ const hosts = new Map(); // username → { host, orders: Map(ref → order), oldestAt }++ for (const order of pending) {++ const host = _hostFor(order);++ if (!host) continue;++ const key = host.username.toLowerCase();++ let h = hosts.get(key);++ if (!h) { h = { host, orders: new Map(), oldestAt: Infinity }; hosts.set(key, h); }++ h.orders.set(_refFor(order), order);++ h.oldestAt = Math.min(h.oldestAt, _sqliteTs(order.created_at) || Date.now());++ }++ const allowTest = _testAllowed();++++ for (const [, h] of hosts) {++ const { host, orders } = h;++ if (!_hasScope(host.conn, 'paid_messages:read')) {++ summary.skippedHosts++;++ _warnHost(host.username, `${orders.size} pending checkout(s) can't be reconciled — no connected account with paid_messages:read for the host`);++ continue;++ }++ summary.hosts++;++ let cursor = null;++ for (let page = 0; page < MAX_PAGES_PER_HOST && orders.size; page++) {++ let pg;++ try {++ const json = await oauth.apiRequest(host.conn.user_id, {++ method: 'GET', path: '/paid-messages', username: host.username,++ query: { limit: PAGE_LIMIT, ...(cursor ? { cursor } : {}) },++ });++ pg = _page(json);++ } catch (e) {++ summary.errors++;++ _warnHost(host.username, `paid-messages read failed: ${e.status || ''} ${e.message}`.trim());++ break;++ }++ summary.pages++;++ if (!pg.rows.length) break;++ let pastOldest = false;++ for (const row of pg.rows) {++ summary.rows++;++ if (_sqliteTs(row.occurredAt) && _sqliteTs(row.occurredAt) < h.oldestAt - GRACE_MS) pastOldest = true;++ const ref = row.appExternalRef ? String(row.appExternalRef) : '';++ if (!ref) continue; // someone else's tip — not ours to credit++ const order = orders.get(ref);++ if (!order) continue; // already credited, or not on this sweep's list++ if (row.source === 'developer_app') continue; // our own forwarded tip echo, never money in++ if (row.isTest && !allowTest) { summary.skippedTest++; continue; }++ // Same shape the webhook hands to the checkout module (money in integer cents).++ const data = {++ eventId: row.eventId, occurredAt: row.occurredAt, source: row.source,++ donorName: row.donorName, isAnonymous: !!row.isAnonymous, message: row.message,++ amountCents: row.amountCents, currency: row.currency,++ amountUsdCents: row.amountUsdCents != null ? row.amountUsdCents : row.amountCents,++ appExternalRef: ref, isTest: !!row.isTest, reconciled: true,++ };++ let consumed = false;++ try { consumed = checkout.handleAttributedDonation(host.receivingUserId, data); }++ catch (e) { summary.errors++; console.warn(`[PowerChat] reconcile: fulfillment for ${ref} failed: ${e.message}`); }++ orders.delete(ref);++ let after = null;++ try { after = db.getPaymentOrderById(order.id); } catch { /* */ }++ if (after && after.status === 'credited') {++ summary.credited.push({ ref, eventId: row.eventId, amountUsdCents: data.amountUsdCents, userId: order.user_id });++ console.log(`[PowerChat] reconcile: credited ${ref} from paid-messages (event ${row.eventId || '?'}, $${(Number(data.amountUsdCents || 0) / 100).toFixed(2)}) — its webhook never arrived`);++ } else if (!consumed || (after && after.status !== 'credited')) {++ summary.underpaid.push({ ref, eventId: row.eventId, amountUsdCents: data.amountUsdCents });++ }++ }++ if (!pg.nextCursor || pastOldest) break; // null cursor = last page; older rows can't match++ cursor = pg.nextCursor;++ }++ }++ return summary;++}++++let _timer = null;++function startReconciler() {++ if (_timer) return;++ const run = () => { reconcileOnce().catch((e) => console.warn('[PowerChat] reconcile sweep failed:', e.message)); };++ setTimeout(run, FIRST_SWEEP_MS).unref();++ _timer = setInterval(run, SWEEP_MS);++ if (_timer.unref) _timer.unref();++ console.log('[PowerChat] paid-messages reconciler started (backfills missed checkout webhooks every 15 min)');++}++++module.exports = { reconcileOnce, startReconciler, _test: { page: _page } };@@@ -16,7 +16,7 @@ const crypto = require('crypto');const router = express.Router();const db = require('../db/database');const config = require('../config');--const { requireAuth, optionalAuth } = require('../auth/auth');++const { requireAuth, optionalAuth, requireAdmin } = require('../auth/auth');const oauth = require('./powerchat-oauth');const webhook = require('./powerchat-webhook');@@@ -307,6 +307,18 @@ router.get('/donate-link', optionalAuth, (req, res) => {}});++// ── POST /reconcile — admin: backfill missed checkout webhooks from /paid-messages ─++// The 15-minute sweeper runs this on its own; the route exists so an operator can++// run it on demand after an outage (or to see the summary) without waiting.++router.post('/reconcile', requireAdmin, async (req, res) => {++ try {++ const summary = await require('./powerchat-reconcile').reconcileOnce();++ res.json({ ok: true, ...summary });++ } catch (err) {++ res.status(500).json({ error: 'Reconcile failed' });++ }++});++// ── POST /test-tip — simulate a tip through the LOCAL pipeline (no PowerChat call) ─// The most useful test: confirms the streamer's alert sound + chat celebration render.router.post('/test-tip', requireAuth, async (req, res) => {nnew file mode 100644@@@ -0,0 +1,119 @@++// /paid-messages reconciliation: finds paid-but-uncredited checkouts by appExternalRef,++// credits through the same idempotent fulfillment the webhook uses, pages on++// nextCursor (both envelope shapes), skips echoes / test rows / hosts without the++// scope, and makes no API call when nothing is pending.++const assert = require('assert');++const path = require('path');++function stub(modPath, exportsObj) { const full = require.resolve(modPath); require.cache[full] = { id: full, filename: full, loaded: true, exports: exportsObj }; }++++const settings = { powerchat_site_tip_username: 'alex', powerchat_allow_test_fulfillment: 'false' };++const orders = new Map();++const seed = () => {++ orders.clear();++ orders.set(41, { id: 41, user_id: 7, provider: 'powerchat', kind: 'bucks', amount_cents: 500, bucks: 500, status: 'pending', created_at: '2026-08-27 09:00:00' });++ orders.set(42, { id: 42, user_id: 8, provider: 'powerchat', kind: 'subscription', amount_cents: 500, streamer_id: 1, provider_ref: 'direct:renew', status: 'pending', created_at: '2026-08-27 09:05:00' });++ orders.set(43, { id: 43, user_id: 9, provider: 'powerchat', kind: 'subscription', amount_cents: 500, streamer_id: 3, provider_ref: 'direct', status: 'pending', created_at: '2026-08-27 09:06:00' });++};++const conns = {++ 1: { user_id: 1, powerchat_username: 'alex', access_token: 'a', refresh_token: 'r', scope: 'checkout:attribute paid_messages:read' },++ 3: { user_id: 3, powerchat_username: 'other', access_token: 'a', refresh_token: 'r', scope: 'checkout:attribute' }, // no paid_messages:read++};++const notes = [];++stub(path.join(__dirname, '../server/db/database'), {++ getSetting: (k) => settings[k] ?? null,++ getPendingPowerchatOrders: () => [...orders.values()].filter((o) => o.status === 'pending'),++ getPowerchatConnection: (id) => conns[id] || null,++ getPowerchatConnectionByUsername: (u) => Object.values(conns).find((c) => c.powerchat_username === String(u).toLowerCase()) || null,++ getPaymentOrderById: (id) => orders.get(id) || null,++ updatePaymentOrder: (id, f) => Object.assign(orders.get(id), f),++ getUserById: (id) => ({ id, username: 'u' + id, display_name: 'U' + id }),++});++stub(path.join(__dirname, '../server/config'), { baseUrl: 'https://openvibe.live' });++stub(path.join(__dirname, '../server/monetization/payments'), {++ bucksForUsd: (usd) => Math.round(usd * 100),++ fulfillBucksOrder: (o) => { orders.get(o.id).status = 'credited'; },++ fulfillSubscriptionOrder: (o, opts) => { orders.get(o.id).status = 'credited'; orders.get(o.id)._opts = opts; },++});++stub(path.join(__dirname, '../server/utils/notify'), { pushNotification: (n) => notes.push(n) });++const api = { calls: [], pages: [] };++stub(path.join(__dirname, '../server/integrations/powerchat-oauth'), {++ getConfig: () => ({ enabled: true, clientId: 'pca', baseUrl: 'https://pc.example', apiBase: 'https://pc.example/api/dev/v1' }),++ redirectUri: () => 'https://openvibe.live/cb',++ apiRequest: async (userId, req) => { api.calls.push({ userId, ...req }); const p = api.pages.shift(); if (!p) throw new Error('unscripted'); if (p.throw) throw Object.assign(new Error('x'), { status: p.throw }); return p; },++});++const reconcile = require('../server/integrations/powerchat-reconcile');++++(async () => {++ // ── nothing pending → nothing read ──++ let sum = await reconcile.reconcileOnce();++ assert.strictEqual(sum.orders, 0); assert.strictEqual(api.calls.length, 0);++++ // ── envelope parsing (corrected + legacy shapes) ──++ assert.deepStrictEqual(reconcile._test.page({ data: { rows: [{ a: 1 }], nextCursor: 'c1' } }), { rows: [{ a: 1 }], nextCursor: 'c1' });++ assert.deepStrictEqual(reconcile._test.page({ data: { rows: [] }, nextCursor: 'c2' }), { rows: [], nextCursor: 'c2' }, 'older deployments left nextCursor outside data');++ assert.deepStrictEqual(reconcile._test.page({ data: [{ b: 2 }] }), { rows: [{ b: 2 }], nextCursor: null });++ assert.deepStrictEqual(reconcile._test.page(null), { rows: [], nextCursor: null });++++ // ── the main sweep: 3 pending orders, two hosts ──++ seed();++ api.calls.length = 0;++ // Host alex: page 1 (legacy cursor placement) → page 2 (corrected).++ api.pages.push({ data: { rows: [++ { eventId: 'e-other', occurredAt: '2026-08-27T09:30:00Z', source: 'powerchat', amountCents: 300, amountUsdCents: 300, appExternalRef: null },++ { eventId: 'e-echo', occurredAt: '2026-08-27T09:20:00Z', source: 'developer_app', amountCents: 500, amountUsdCents: 500, appExternalRef: 'pcorder:41' },++ { eventId: 'e-test', occurredAt: '2026-08-27T09:19:00Z', source: 'powerchat', isTest: true, amountCents: 500, amountUsdCents: 500, appExternalRef: 'pcorder:41' },++ { eventId: 'e-41', occurredAt: '2026-08-27T09:10:00Z', source: 'powerchat', donorName: 'Pat', amountCents: 500, currency: 'USD', amountUsdCents: 500, appExternalRef: 'pcorder:41' },++ ] }, nextCursor: 'cur-2' });++ api.pages.push({ data: { rows: [++ { eventId: 'e-42', occurredAt: '2026-08-27T09:08:00Z', source: 'powerchat', donorName: 'Sam', amountCents: 500, amountUsdCents: 500, appExternalRef: 'pcsub:42' },++ ], nextCursor: 'cur-3' } });++ sum = await reconcile.reconcileOnce();++ assert.strictEqual(sum.orders, 3);++ assert.strictEqual(sum.hosts, 1, 'alex hosts both 41 (site) and 42 (direct, streamer 1 = alex)');++ assert.strictEqual(sum.skippedHosts, 1, "'other' has no paid_messages:read → skipped, no call");++ assert.strictEqual(sum.pages, 2, 'stopped after page 2: no orders left for that host despite a cursor');++ assert.strictEqual(sum.skippedTest, 1);++ assert.deepStrictEqual(sum.credited.map((c) => c.ref).sort(), ['pcorder:41', 'pcsub:42']);++ assert.deepStrictEqual(sum.underpaid, []);++ assert.strictEqual(orders.get(41).status, 'credited');++ assert.strictEqual(orders.get(42).status, 'credited');++ assert.strictEqual(orders.get(42)._opts.autoRenew, 1, 'renew flag read off provider_ref as the webhook path does');++ assert.strictEqual(orders.get(42)._opts.creditShare, false, 'direct: streamer already holds the cash');++ assert.strictEqual(orders.get(43).status, 'pending');++ const reads = api.calls.filter((c) => c.path === '/paid-messages');++ assert.strictEqual(reads.length, 2);++ assert.strictEqual(reads[0].userId, 1); assert.strictEqual(reads[0].username, 'alex');++ assert.deepStrictEqual(reads[0].query, { limit: 100 });++ assert.deepStrictEqual(reads[1].query, { limit: 100, cursor: 'cur-2' }, 'legacy top-level nextCursor was honoured');++ assert.strictEqual(notes.length, 2, 'buyers are notified exactly like a webhook credit');++++ // ── idempotent: run again → credited orders are no longer pending → only 43 remains, host skipped ──++ api.calls.length = 0;++ sum = await reconcile.reconcileOnce();++ assert.strictEqual(sum.orders, 1); assert.strictEqual(api.calls.length, 0); assert.strictEqual(sum.credited.length, 0);++++ // ── underpaid sub (direct route): not credited, reported, order stays pending ──++ // (a Vibes pack credits whatever was actually paid, so only subs have this guard)++ seed(); orders.delete(41); orders.delete(43);++ api.calls.length = 0; api.pages.length = 0;++ api.pages.push({ data: { rows: [{ eventId: 'e-low', occurredAt: '2026-08-27T09:10:00Z', source: 'powerchat', amountCents: 100, amountUsdCents: 100, appExternalRef: 'pcsub:42' }], nextCursor: null } });++ sum = await reconcile.reconcileOnce();++ assert.strictEqual(sum.credited.length, 0);++ assert.deepStrictEqual(sum.underpaid.map((u) => u.ref), ['pcsub:42']);++ assert.strictEqual(orders.get(42).status, 'pending');++++ // ── a paging stop: rows older than the oldest pending order (minus grace) end the walk ──++ seed(); orders.delete(42); orders.delete(43);++ api.calls.length = 0; api.pages.length = 0;++ api.pages.push({ data: { rows: [{ eventId: 'e-ancient', occurredAt: '2026-08-20T00:00:00Z', source: 'powerchat', amountCents: 1, amountUsdCents: 1, appExternalRef: null }], nextCursor: 'more' } });++ sum = await reconcile.reconcileOnce();++ assert.strictEqual(sum.pages, 1, 'did not follow the cursor into rows that predate every pending order');++++ // ── read failure is contained: summary.errors, nothing credited, no throw ──++ seed(); orders.delete(42); orders.delete(43);++ api.pages.length = 0; api.pages.push({ throw: 429 });++ sum = await reconcile.reconcileOnce();++ assert.strictEqual(sum.errors, 1); assert.strictEqual(orders.get(41).status, 'pending');++++ console.log('✅ powerchat reconcile tests passed');++})().catch((e) => { console.error('❌', e); process.exit(1); });
@@ -5080,6 +5080,16 @@ function getPaymentOrderByRef(provider, ref) {
@@ -5080,6 +5080,16 @@ function getPaymentOrderByRef(provider, ref) {
return get('SELECT * FROM payment_orders WHERE provider = ? AND provider_ref = ? ORDER BY id DESC LIMIT 1', [provider, ref]);
return get('SELECT * FROM payment_orders WHERE provider = ? AND provider_ref = ? ORDER BY id DESC LIMIT 1', [provider, ref]);
}
}
+// PowerChat checkouts still waiting for their donation.completed webhook — the
+// reconciliation sweep's work list (recent only: an intent lives an hour, and an
+// order nobody paid within days is an abandoned cart, not a missed webhook).
+function getPendingPowerchatOrders(days = 3) {
+ return all(`SELECT * FROM payment_orders
+ WHERE provider = 'powerchat' AND status = 'pending'
+ AND created_at >= datetime('now', ?)
+ ORDER BY id ASC`, [`-${Math.max(1, Math.round(days))} days`]);
+}
+
function updatePaymentOrder(id, fields) {
function updatePaymentOrder(id, fields) {
const allowed = new Set(['provider_ref', 'status', 'amount_cents', 'bucks', 'currency', 'streamer_id']);
const allowed = new Set(['provider_ref', 'status', 'amount_cents', 'bucks', 'currency', 'streamer_id']);
const entries = Object.entries(fields || {}).filter(([k]) => allowed.has(k));
const entries = Object.entries(fields || {}).filter(([k]) => allowed.has(k));
@@ -7964,7 +7974,7 @@ module.exports = {
@@ -7964,7 +7974,7 @@ module.exports = {
getPowerchatConnection, getPowerchatConnectionByUsername, getPowerchatConnectionByPcUserId,
getPowerchatConnection, getPowerchatConnectionByUsername, getPowerchatConnectionByPcUserId,
upsertPowerchatConnection, updatePowerchatTokens, setPowerchatConnectionError,
upsertPowerchatConnection, updatePowerchatTokens, setPowerchatConnectionError,
deletePowerchatConnection, powerchatDeliveryIsNew, cleanupPowerchatDeliveries,
deletePowerchatConnection, powerchatDeliveryIsNew, cleanupPowerchatDeliveries,
- createPaymentOrder, getPaymentOrderById, getPaymentOrderByRef, updatePaymentOrder,
+ createPaymentOrder, getPaymentOrderById, getPaymentOrderByRef, updatePaymentOrder, getPendingPowerchatOrders,
upsertSubscription, getSubscriptionByProviderRef, getActiveSubscription, isActiveSubscriber,
upsertSubscription, getSubscriptionByProviderRef, getActiveSubscription, isActiveSubscriber,
getSubscriptionsByStreamer, getSubscriptionsBySubscriber, getActiveSubscriberCount, setSubscriptionStatus,
getSubscriptionsByStreamer, getSubscriptionsBySubscriber, getActiveSubscriberCount, setSubscriptionStatus,
getSubscriptionsDueRenewal,
getSubscriptionsDueRenewal,
@@ -1018,6 +1018,8 @@ async function start() {
@@ -1018,6 +1018,8 @@ async function start() {
try { require('./integrations/powerchat-platform').startCurrencyEarnFlusher(); } catch (e) { console.warn('[PowerChat] earn flusher not started:', e.message); }
try { require('./integrations/powerchat-platform').startCurrencyEarnFlusher(); } catch (e) { console.warn('[PowerChat] earn flusher not started:', e.message); }
// Read /chat/history back so a 202-accepted relay that moderation dropped is visible.
// Read /chat/history back so a 202-accepted relay that moderation dropped is visible.
try { require('./integrations/powerchat-platform').startChatVerifier(); } catch (e) { console.warn('[PowerChat] chat verifier not started:', e.message); }
try { require('./integrations/powerchat-platform').startChatVerifier(); } catch (e) { console.warn('[PowerChat] chat verifier not started:', e.message); }
+ // Backfill checkout orders whose donation.completed webhook never arrived (/paid-messages).
+ try { require('./integrations/powerchat-reconcile').startReconciler(); } catch (e) { console.warn('[PowerChat] reconciler not started:', e.message); }
// Auto-renew lapsed channel subs (Vibes-balance renewal; Stripe renews natively).
// Auto-renew lapsed channel subs (Vibes-balance renewal; Stripe renews natively).
try { require('./monetization/payments').startRenewalSweeper(); } catch (e) { console.warn('[Payments] renewal sweeper not started:', e.message); }
try { require('./monetization/payments').startRenewalSweeper(); } catch (e) { console.warn('[Payments] renewal sweeper not started:', e.message); }
// 5-minute live-viewer samples → the home hero's 24h sparkline.
// 5-minute live-viewer samples → the home hero's 24h sparkline.
new file mode 100644
new file mode 100644
@@ -0,0 +1,175 @@
@@ -0,0 +1,175 @@
+/**
+ * powerchat-reconcile.js — backfill missed donation.completed webhooks from
+ * GET /streamers/:username/paid-messages (scope paid_messages:read).
+ *
+ * The signed webhook is the ONLY authoritative confirmation and stays the primary
+ * credit path. But webhooks can be missed for good: the receiver was down past
+ * PowerChat's ~24h retry tail, ~20 consecutive failures tripped its circuit
+ * breaker, or a 410 disabled the endpoint. Until now a viewer who paid during
+ * such a window simply never got their Vibes / subscription.
+ *
+ * /paid-messages is the same confirmed money as a cursor-paginated feed, and every
+ * row echoes `appExternalRef` — the pcorder:/pcsub: ref we minted onto the checkout
+ * link — so paid-but-uncredited orders are found by exact match, never guessed.
+ *
+ * Safe to run repeatedly and concurrently with live webhooks: everything is credited
+ * through the same handleAttributedDonation() the webhook uses, which is idempotent
+ * on the order's status. Work-list driven: no pending order → no API call at all.
+ *
+ * Envelope: written against { data: { rows, nextCursor } } (the corrected upstream
+ * shape) and defensive on older deployments that left nextCursor outside `data`.
+ */
+'use strict';
+
+const db = require('../db/database');
+const oauth = require('./powerchat-oauth');
+const checkout = require('./powerchat-checkout');
+
+const PAGE_LIMIT = 100;
+const MAX_PAGES_PER_HOST = 5; // 500 newest confirmed tips per host per sweep
+const LOOKBACK_DAYS = 3; // intents live 1h; older pending orders are abandoned carts
+const GRACE_MS = 2 * 60 * 60 * 1000; // rows may be a little older than the order (clock skew, slow checkout)
+const SWEEP_MS = 15 * 60 * 1000;
+const FIRST_SWEEP_MS = 2 * 60 * 1000;
+
+function _sqliteTs(s) {
+ if (!s) return 0;
+ const t = Date.parse(String(s).includes('T') ? String(s) : String(s).replace(' ', 'T') + 'Z');
+ return Number.isFinite(t) ? t : 0;
+}
+function _testAllowed() {
+ try { const v = db.getSetting('powerchat_allow_test_fulfillment'); return v === true || v === 'true' || v === 1 || v === '1'; }
+ catch { return false; }
+}
+function _hasScope(conn, scope) {
+ return !!(conn && conn.access_token && conn.refresh_token && (!conn.scope || String(conn.scope).split(/\s+/).includes(scope)));
+}
+
+// Which PowerChat account hosted this order's checkout (where the money landed), and
+// which local user "receives" it for the fulfillment pipeline.
+// pcsub direct → the streamer's own PowerChat; site → the site tips account.
+function _hostFor(order) {
+ const mode = String(order.provider_ref || 'site').split(':')[0];
+ if (order.kind === 'subscription' && mode === 'direct' && order.streamer_id) {
+ const conn = db.getPowerchatConnection(order.streamer_id);
+ return conn && conn.powerchat_username ? { username: conn.powerchat_username, conn, receivingUserId: order.streamer_id } : null;
+ }
+ const site = checkout.getSiteAccount();
+ if (!site) return null;
+ const conn = db.getPowerchatConnectionByUsername(site.username);
+ return { username: site.username, conn: conn || null, receivingUserId: conn ? conn.user_id : null };
+}
+function _refFor(order) {
+ return order.kind === 'subscription' ? `pcsub:${order.id}` : `pcorder:${order.id}`;
+}
+function _page(json) {
+ const d = json && json.data !== undefined ? json.data : json;
+ const rows = Array.isArray(d) ? d : (d && Array.isArray(d.rows) ? d.rows : []);
+ // Corrected shape nests nextCursor in data; older deployments put it beside it.
+ const nextCursor = (d && !Array.isArray(d) && d.nextCursor != null) ? d.nextCursor
+ : (json && json.nextCursor != null ? json.nextCursor : null);
+ return { rows, nextCursor: nextCursor || null };
+}
+
+const _hostWarned = new Map();
+function _warnHost(username, msg) {
+ const now = Date.now();
+ if ((_hostWarned.get(username) || 0) > now - 3600000) return;
+ _hostWarned.set(username, now);
+ console.warn(`[PowerChat] reconcile: ${msg} (@${username})`);
+}
+
+/** One sweep. Returns a summary; never throws. */
+async function reconcileOnce() {
+ const summary = { orders: 0, hosts: 0, pages: 0, rows: 0, credited: [], underpaid: [], skippedTest: 0, skippedHosts: 0, errors: 0 };
+ let pending;
+ try { pending = db.getPendingPowerchatOrders(LOOKBACK_DAYS) || []; } catch (e) { summary.errors++; return summary; }
+ summary.orders = pending.length;
+ if (!pending.length) return summary;
+
+ // Group the work list by host account: one paged read per host, not per order.
+ const hosts = new Map(); // username → { host, orders: Map(ref → order), oldestAt }
+ for (const order of pending) {
+ const host = _hostFor(order);
+ if (!host) continue;
+ const key = host.username.toLowerCase();
+ let h = hosts.get(key);
+ if (!h) { h = { host, orders: new Map(), oldestAt: Infinity }; hosts.set(key, h); }
+ h.orders.set(_refFor(order), order);
+ h.oldestAt = Math.min(h.oldestAt, _sqliteTs(order.created_at) || Date.now());
+ }
+ const allowTest = _testAllowed();
+
+ for (const [, h] of hosts) {
+ const { host, orders } = h;
+ if (!_hasScope(host.conn, 'paid_messages:read')) {
+ summary.skippedHosts++;
+ _warnHost(host.username, `${orders.size} pending checkout(s) can't be reconciled — no connected account with paid_messages:read for the host`);
+ continue;
+ }
+ summary.hosts++;
+ let cursor = null;
+ for (let page = 0; page < MAX_PAGES_PER_HOST && orders.size; page++) {
+ let pg;
+ try {
+ const json = await oauth.apiRequest(host.conn.user_id, {
+ method: 'GET', path: '/paid-messages', username: host.username,
+ query: { limit: PAGE_LIMIT, ...(cursor ? { cursor } : {}) },
+ });
+ pg = _page(json);
+ } catch (e) {
+ summary.errors++;
+ _warnHost(host.username, `paid-messages read failed: ${e.status || ''} ${e.message}`.trim());
+ break;
+ }
+ summary.pages++;
+ if (!pg.rows.length) break;
+ let pastOldest = false;
+ for (const row of pg.rows) {
+ summary.rows++;
+ if (_sqliteTs(row.occurredAt) && _sqliteTs(row.occurredAt) < h.oldestAt - GRACE_MS) pastOldest = true;
+ const ref = row.appExternalRef ? String(row.appExternalRef) : '';
+ if (!ref) continue; // someone else's tip — not ours to credit
+ const order = orders.get(ref);
+ if (!order) continue; // already credited, or not on this sweep's list
+ if (row.source === 'developer_app') continue; // our own forwarded tip echo, never money in
+ if (row.isTest && !allowTest) { summary.skippedTest++; continue; }
+ // Same shape the webhook hands to the checkout module (money in integer cents).
+ const data = {
+ eventId: row.eventId, occurredAt: row.occurredAt, source: row.source,
+ donorName: row.donorName, isAnonymous: !!row.isAnonymous, message: row.message,
+ amountCents: row.amountCents, currency: row.currency,
+ amountUsdCents: row.amountUsdCents != null ? row.amountUsdCents : row.amountCents,
+ appExternalRef: ref, isTest: !!row.isTest, reconciled: true,
+ };
+ let consumed = false;
+ try { consumed = checkout.handleAttributedDonation(host.receivingUserId, data); }
+ catch (e) { summary.errors++; console.warn(`[PowerChat] reconcile: fulfillment for ${ref} failed: ${e.message}`); }
+ orders.delete(ref);
+ let after = null;
+ try { after = db.getPaymentOrderById(order.id); } catch { /* */ }
+ if (after && after.status === 'credited') {
+ summary.credited.push({ ref, eventId: row.eventId, amountUsdCents: data.amountUsdCents, userId: order.user_id });
+ console.log(`[PowerChat] reconcile: credited ${ref} from paid-messages (event ${row.eventId || '?'}, $${(Number(data.amountUsdCents || 0) / 100).toFixed(2)}) — its webhook never arrived`);
+ } else if (!consumed || (after && after.status !== 'credited')) {
+ summary.underpaid.push({ ref, eventId: row.eventId, amountUsdCents: data.amountUsdCents });
+ }
+ }
+ if (!pg.nextCursor || pastOldest) break; // null cursor = last page; older rows can't match
+ cursor = pg.nextCursor;
+ }
+ }
+ return summary;
+}
+
+let _timer = null;
+function startReconciler() {
+ if (_timer) return;
+ const run = () => { reconcileOnce().catch((e) => console.warn('[PowerChat] reconcile sweep failed:', e.message)); };
+ setTimeout(run, FIRST_SWEEP_MS).unref();
+ _timer = setInterval(run, SWEEP_MS);
+ if (_timer.unref) _timer.unref();
+ console.log('[PowerChat] paid-messages reconciler started (backfills missed checkout webhooks every 15 min)');
+}
+
+module.exports = { reconcileOnce, startReconciler, _test: { page: _page } };
@@ -16,7 +16,7 @@ const crypto = require('crypto');
@@ -16,7 +16,7 @@ const crypto = require('crypto');
const router = express.Router();
const router = express.Router();
const db = require('../db/database');
const db = require('../db/database');
const config = require('../config');
const config = require('../config');
-const { requireAuth, optionalAuth } = require('../auth/auth');
+const { requireAuth, optionalAuth, requireAdmin } = require('../auth/auth');
const oauth = require('./powerchat-oauth');
const oauth = require('./powerchat-oauth');
const webhook = require('./powerchat-webhook');
const webhook = require('./powerchat-webhook');
@@ -307,6 +307,18 @@ router.get('/donate-link', optionalAuth, (req, res) => {
@@ -307,6 +307,18 @@ router.get('/donate-link', optionalAuth, (req, res) => {
}
}
});
});
+// ── POST /reconcile — admin: backfill missed checkout webhooks from /paid-messages ─
+// The 15-minute sweeper runs this on its own; the route exists so an operator can
+// run it on demand after an outage (or to see the summary) without waiting.
+router.post('/reconcile', requireAdmin, async (req, res) => {
+ try {
+ const summary = await require('./powerchat-reconcile').reconcileOnce();
+ res.json({ ok: true, ...summary });
+ } catch (err) {
+ res.status(500).json({ error: 'Reconcile failed' });
+ }
+});
+
// ── POST /test-tip — simulate a tip through the LOCAL pipeline (no PowerChat call) ─
// ── POST /test-tip — simulate a tip through the LOCAL pipeline (no PowerChat call) ─
// The most useful test: confirms the streamer's alert sound + chat celebration render.
// The most useful test: confirms the streamer's alert sound + chat celebration render.
router.post('/test-tip', requireAuth, async (req, res) => {
router.post('/test-tip', requireAuth, async (req, res) => {
new file mode 100644
new file mode 100644
@@ -0,0 +1,119 @@
@@ -0,0 +1,119 @@
+// /paid-messages reconciliation: finds paid-but-uncredited checkouts by appExternalRef,
+// credits through the same idempotent fulfillment the webhook uses, pages on
+// nextCursor (both envelope shapes), skips echoes / test rows / hosts without the
+// scope, and makes no API call when nothing is pending.
+const assert = require('assert');
+const path = require('path');
+function stub(modPath, exportsObj) { const full = require.resolve(modPath); require.cache[full] = { id: full, filename: full, loaded: true, exports: exportsObj }; }
+
+const settings = { powerchat_site_tip_username: 'alex', powerchat_allow_test_fulfillment: 'false' };
+const orders = new Map();
+const seed = () => {
+ orders.clear();
+ orders.set(41, { id: 41, user_id: 7, provider: 'powerchat', kind: 'bucks', amount_cents: 500, bucks: 500, status: 'pending', created_at: '2026-08-27 09:00:00' });
+ orders.set(42, { id: 42, user_id: 8, provider: 'powerchat', kind: 'subscription', amount_cents: 500, streamer_id: 1, provider_ref: 'direct:renew', status: 'pending', created_at: '2026-08-27 09:05:00' });
+ orders.set(43, { id: 43, user_id: 9, provider: 'powerchat', kind: 'subscription', amount_cents: 500, streamer_id: 3, provider_ref: 'direct', status: 'pending', created_at: '2026-08-27 09:06:00' });
+};
+const conns = {
+ 1: { user_id: 1, powerchat_username: 'alex', access_token: 'a', refresh_token: 'r', scope: 'checkout:attribute paid_messages:read' },
+ 3: { user_id: 3, powerchat_username: 'other', access_token: 'a', refresh_token: 'r', scope: 'checkout:attribute' }, // no paid_messages:read
+};
+const notes = [];
+stub(path.join(__dirname, '../server/db/database'), {
+ getSetting: (k) => settings[k] ?? null,
+ getPendingPowerchatOrders: () => [...orders.values()].filter((o) => o.status === 'pending'),
+ getPowerchatConnection: (id) => conns[id] || null,
+ getPowerchatConnectionByUsername: (u) => Object.values(conns).find((c) => c.powerchat_username === String(u).toLowerCase()) || null,
+ getPaymentOrderById: (id) => orders.get(id) || null,
+ updatePaymentOrder: (id, f) => Object.assign(orders.get(id), f),
+ getUserById: (id) => ({ id, username: 'u' + id, display_name: 'U' + id }),
+});
+stub(path.join(__dirname, '../server/config'), { baseUrl: 'https://openvibe.live' });
+stub(path.join(__dirname, '../server/monetization/payments'), {
+ bucksForUsd: (usd) => Math.round(usd * 100),
+ fulfillBucksOrder: (o) => { orders.get(o.id).status = 'credited'; },
+ fulfillSubscriptionOrder: (o, opts) => { orders.get(o.id).status = 'credited'; orders.get(o.id)._opts = opts; },
+});
+stub(path.join(__dirname, '../server/utils/notify'), { pushNotification: (n) => notes.push(n) });
+const api = { calls: [], pages: [] };
+stub(path.join(__dirname, '../server/integrations/powerchat-oauth'), {
+ getConfig: () => ({ enabled: true, clientId: 'pca', baseUrl: 'https://pc.example', apiBase: 'https://pc.example/api/dev/v1' }),
+ redirectUri: () => 'https://openvibe.live/cb',
+ apiRequest: async (userId, req) => { api.calls.push({ userId, ...req }); const p = api.pages.shift(); if (!p) throw new Error('unscripted'); if (p.throw) throw Object.assign(new Error('x'), { status: p.throw }); return p; },
+});
+const reconcile = require('../server/integrations/powerchat-reconcile');
+
+(async () => {
+ // ── nothing pending → nothing read ──
+ let sum = await reconcile.reconcileOnce();
+ assert.strictEqual(sum.orders, 0); assert.strictEqual(api.calls.length, 0);
+
+ // ── envelope parsing (corrected + legacy shapes) ──
+ assert.deepStrictEqual(reconcile._test.page({ data: { rows: [{ a: 1 }], nextCursor: 'c1' } }), { rows: [{ a: 1 }], nextCursor: 'c1' });
+ assert.deepStrictEqual(reconcile._test.page({ data: { rows: [] }, nextCursor: 'c2' }), { rows: [], nextCursor: 'c2' }, 'older deployments left nextCursor outside data');
+ assert.deepStrictEqual(reconcile._test.page({ data: [{ b: 2 }] }), { rows: [{ b: 2 }], nextCursor: null });
+ assert.deepStrictEqual(reconcile._test.page(null), { rows: [], nextCursor: null });
+
+ // ── the main sweep: 3 pending orders, two hosts ──
+ seed();
+ api.calls.length = 0;
+ // Host alex: page 1 (legacy cursor placement) → page 2 (corrected).
+ api.pages.push({ data: { rows: [
+ { eventId: 'e-other', occurredAt: '2026-08-27T09:30:00Z', source: 'powerchat', amountCents: 300, amountUsdCents: 300, appExternalRef: null },
+ { eventId: 'e-echo', occurredAt: '2026-08-27T09:20:00Z', source: 'developer_app', amountCents: 500, amountUsdCents: 500, appExternalRef: 'pcorder:41' },
+ { eventId: 'e-test', occurredAt: '2026-08-27T09:19:00Z', source: 'powerchat', isTest: true, amountCents: 500, amountUsdCents: 500, appExternalRef: 'pcorder:41' },
+ { eventId: 'e-41', occurredAt: '2026-08-27T09:10:00Z', source: 'powerchat', donorName: 'Pat', amountCents: 500, currency: 'USD', amountUsdCents: 500, appExternalRef: 'pcorder:41' },
+ ] }, nextCursor: 'cur-2' });
+ api.pages.push({ data: { rows: [
+ { eventId: 'e-42', occurredAt: '2026-08-27T09:08:00Z', source: 'powerchat', donorName: 'Sam', amountCents: 500, amountUsdCents: 500, appExternalRef: 'pcsub:42' },
+ ], nextCursor: 'cur-3' } });
+ sum = await reconcile.reconcileOnce();
+ assert.strictEqual(sum.orders, 3);
+ assert.strictEqual(sum.hosts, 1, 'alex hosts both 41 (site) and 42 (direct, streamer 1 = alex)');
+ assert.strictEqual(sum.skippedHosts, 1, "'other' has no paid_messages:read → skipped, no call");
+ assert.strictEqual(sum.pages, 2, 'stopped after page 2: no orders left for that host despite a cursor');
+ assert.strictEqual(sum.skippedTest, 1);
+ assert.deepStrictEqual(sum.credited.map((c) => c.ref).sort(), ['pcorder:41', 'pcsub:42']);
+ assert.deepStrictEqual(sum.underpaid, []);
+ assert.strictEqual(orders.get(41).status, 'credited');
+ assert.strictEqual(orders.get(42).status, 'credited');
+ assert.strictEqual(orders.get(42)._opts.autoRenew, 1, 'renew flag read off provider_ref as the webhook path does');
+ assert.strictEqual(orders.get(42)._opts.creditShare, false, 'direct: streamer already holds the cash');
+ assert.strictEqual(orders.get(43).status, 'pending');
+ const reads = api.calls.filter((c) => c.path === '/paid-messages');
+ assert.strictEqual(reads.length, 2);
+ assert.strictEqual(reads[0].userId, 1); assert.strictEqual(reads[0].username, 'alex');
+ assert.deepStrictEqual(reads[0].query, { limit: 100 });
+ assert.deepStrictEqual(reads[1].query, { limit: 100, cursor: 'cur-2' }, 'legacy top-level nextCursor was honoured');
+ assert.strictEqual(notes.length, 2, 'buyers are notified exactly like a webhook credit');
+
+ // ── idempotent: run again → credited orders are no longer pending → only 43 remains, host skipped ──
+ api.calls.length = 0;
+ sum = await reconcile.reconcileOnce();
+ assert.strictEqual(sum.orders, 1); assert.strictEqual(api.calls.length, 0); assert.strictEqual(sum.credited.length, 0);
+
+ // ── underpaid sub (direct route): not credited, reported, order stays pending ──
+ // (a Vibes pack credits whatever was actually paid, so only subs have this guard)
+ seed(); orders.delete(41); orders.delete(43);
+ api.calls.length = 0; api.pages.length = 0;
+ api.pages.push({ data: { rows: [{ eventId: 'e-low', occurredAt: '2026-08-27T09:10:00Z', source: 'powerchat', amountCents: 100, amountUsdCents: 100, appExternalRef: 'pcsub:42' }], nextCursor: null } });
+ sum = await reconcile.reconcileOnce();
+ assert.strictEqual(sum.credited.length, 0);
+ assert.deepStrictEqual(sum.underpaid.map((u) => u.ref), ['pcsub:42']);
+ assert.strictEqual(orders.get(42).status, 'pending');
+
+ // ── a paging stop: rows older than the oldest pending order (minus grace) end the walk ──
+ seed(); orders.delete(42); orders.delete(43);
+ api.calls.length = 0; api.pages.length = 0;
+ api.pages.push({ data: { rows: [{ eventId: 'e-ancient', occurredAt: '2026-08-20T00:00:00Z', source: 'powerchat', amountCents: 1, amountUsdCents: 1, appExternalRef: null }], nextCursor: 'more' } });
+ sum = await reconcile.reconcileOnce();
+ assert.strictEqual(sum.pages, 1, 'did not follow the cursor into rows that predate every pending order');
+
+ // ── read failure is contained: summary.errors, nothing credited, no throw ──
+ seed(); orders.delete(42); orders.delete(43);
+ api.pages.length = 0; api.pages.push({ throw: 429 });
+ sum = await reconcile.reconcileOnce();
+ assert.strictEqual(sum.errors, 1); assert.strictEqual(orders.get(41).status, 'pending');
+
+ console.log('✅ powerchat reconcile tests passed');
+})().catch((e) => { console.error('❌', e); process.exit(1); });