openvibers/openvibe.live Commits
Loyalty ledgers use deterministic per-event idempotency keys (ADR-012 rule 5). Media requests: a paid request is written first (status failed, charge_state 'charging', out of the queue and history) and its charge is keyed by its id: OpenCoins and channel points live:media_req:<id> (was Date.now()+Math.random()), Vibes on Billing live:media_charge:<id> linked to the request from the start (was a random UUID); it joins the queue once paid, a refused charge removes the row, and a wallet that never answers (asked twice with the same key) leaves it 'unknown' for the new media-charge-reconcile job, which charges again with the same key (a replay if it landed) and refunds it, so the viewer ends where they started. Channel points: every debit and credit goes through db.applyChannelPoints with a key in the new channel_points_log (a repeated key moves nothing, a reused key for another event is refused, a refused debit leaves no row): watch = watch row + minute, chat = the user's minute (holds across a restart), follow = user + streamer, bonus = claim window, media = request id, redemption = redemption id (limits are checked and the row created and paid in one transaction, so there is no take-then-refund), and a streamer rejecting a redemption twice refunds once. Admin OpenCoins grants are recorded first and keyed live:admin_grant:<grant id> (was Date.now()+Math.random()); an Idempotency-Key repeat is the same grant. test/loyalty-keys.test.js shows retries never spend or credit twice; billing-authority checks the Billing key and its replay
Changes
@@@ -771,6 +771,19 @@ function initDb() {updated_at DATETIME DEFAULT CURRENT_TIMESTAMP,PRIMARY KEY (user_id, streamer_id))`);++ // Every channel-points debit and credit, keyed per event (ADR-012 rule 5): a retried award,++ // spend or refund with the same key is applied once. Keys: live:cp:<event>:<id>, and++ // live:media_req:<request id> / live:media_refund:<request id> for media requests.++ database.exec(`CREATE TABLE IF NOT EXISTS channel_points_log (++ id INTEGER PRIMARY KEY AUTOINCREMENT,++ idempotency_key TEXT NOT NULL UNIQUE,++ user_id INTEGER NOT NULL,++ streamer_id INTEGER NOT NULL,++ delta INTEGER NOT NULL,++ reason TEXT,++ created_at DATETIME DEFAULT CURRENT_TIMESTAMP++ )`);++ database.exec('CREATE INDEX IF NOT EXISTS idx_cp_log_user ON channel_points_log(user_id, streamer_id, id)');} catch (e) { console.warn('[DB] channel_points migration:', e.message); }// Kick chatroom-id cache. Kick's v2 API (which exposes the Pusher chatroom id)@@@ -929,6 +942,11 @@ function initDb() {// setting — a streamer switching from Vibes to points would otherwise refund the// wrong currency to everyone still queued.if (!mrCols.includes('currency')) database.exec("ALTER TABLE media_requests ADD COLUMN currency TEXT DEFAULT 'opencoins'");++ // A paid request is written before it is charged, so the charge can be keyed by its id++ // (live:media_req:<id>, ADR-012 rule 5). Until the charge answers it is status 'failed'++ // with charge_state 'charging' (out of the queue and the history); 'unknown' means the++ // wallet never answered and mediaQueue.reconcileCharges() settles it. NULL = settled.++ if (!mrCols.includes('charge_state')) database.exec('ALTER TABLE media_requests ADD COLUMN charge_state TEXT');const msCols = database.pragma('table_info(media_request_settings)').map(c => c.name);if (!msCols.includes('cost_mode')) database.exec("ALTER TABLE media_request_settings ADD COLUMN cost_mode TEXT DEFAULT 'flat' CHECK(cost_mode IN ('flat','per_minute'))");@@@ -4508,22 +4526,49 @@ function getChannelPoints(userId, streamerId) {const r = get('SELECT balance FROM channel_points WHERE user_id = ? AND streamer_id = ?', [userId, streamerId]);return r ? r.balance : 0;}--function addChannelPoints(userId, streamerId, amount) {-- if (!userId || !streamerId || !amount) return getChannelPoints(userId, streamerId);-- run(`INSERT INTO channel_points (user_id, streamer_id, balance, updated_at)-- VALUES (?, ?, ?, CURRENT_TIMESTAMP)-- ON CONFLICT(user_id, streamer_id) DO UPDATE SET-- balance = balance + excluded.balance, updated_at = CURRENT_TIMESTAMP`,-- [userId, streamerId, amount]);-- return getChannelPoints(userId, streamerId);++/**++ * Apply one channel-points event, keyed (ADR-012 rule 5): { applied, replayed, balance }.++ * A key seen before is a replay: nothing moves, and the same key for a different user, channel or++ * amount is refused. A debit (delta < 0) only happens when the balance covers it; a refused debit++ * leaves no log row, so the same key can be tried again later.++ */++function applyChannelPoints({ userId, streamerId, delta, key, reason = null }) {++ if (!key || typeof key !== 'string') throw new TypeError('channel points: an idempotency key is required for every debit and credit');++ if (!userId || !streamerId || !Number.isInteger(delta) || delta === 0) {++ return { applied: false, replayed: false, balance: getChannelPoints(userId, streamerId) };++ }++ return getDb().transaction(() => {++ const seen = get('SELECT user_id, streamer_id, delta FROM channel_points_log WHERE idempotency_key = ?', [key]);++ if (seen) {++ if (seen.user_id !== userId || seen.streamer_id !== streamerId || seen.delta !== delta) {++ throw new Error(`channel points: idempotency key ${key} was already used for a different event`);++ }++ return { applied: false, replayed: true, balance: getChannelPoints(userId, streamerId) };++ }++ if (delta < 0) {++ const res = run(`UPDATE channel_points SET balance = balance + ?, updated_at = CURRENT_TIMESTAMP++ WHERE user_id = ? AND streamer_id = ? AND balance >= ?`, [delta, userId, streamerId, -delta]);++ if (!res.changes) return { applied: false, replayed: false, balance: getChannelPoints(userId, streamerId) };++ } else {++ run(`INSERT INTO channel_points (user_id, streamer_id, balance, updated_at)++ VALUES (?, ?, ?, CURRENT_TIMESTAMP)++ ON CONFLICT(user_id, streamer_id) DO UPDATE SET++ balance = balance + excluded.balance, updated_at = CURRENT_TIMESTAMP`, [userId, streamerId, delta]);++ }++ run('INSERT INTO channel_points_log (idempotency_key, user_id, streamer_id, delta, reason) VALUES (?, ?, ?, ?, ?)',++ [key, userId, streamerId, delta, reason ? String(reason).slice(0, 200) : null]);++ return { applied: true, replayed: false, balance: getChannelPoints(userId, streamerId) };++ })();++}++/** Credit channel points for one event (key required). Returns the new balance. */++function addChannelPoints(userId, streamerId, amount, key, reason) {++ return applyChannelPoints({ userId, streamerId, delta: amount, key, reason }).balance;}--// Atomic spend — returns true only if the viewer had enough for this streamer.--function deductChannelPoints(userId, streamerId, amount) {++/** Atomic spend for one event (key required): true if taken now or already taken under this key. */++function deductChannelPoints(userId, streamerId, amount, key, reason) {if (!userId || !streamerId) return false;-- const res = run(`UPDATE channel_points SET balance = balance - ?, updated_at = CURRENT_TIMESTAMP-- WHERE user_id = ? AND streamer_id = ? AND balance >= ?`,-- [amount, userId, streamerId, amount]);-- return (res?.changes || 0) > 0;++ const r = applyChannelPoints({ userId, streamerId, delta: -amount, key, reason });++ return r.applied || r.replayed;}// ── Kick chatroom-id cache (survives the Cloudflare-blocked v2 API) ──@@@ -6204,12 +6249,12 @@ function upsertMediaRequestSettings(userId, fields = {}) {return getMediaRequestSettingsByUserId(userId);}--function createMediaRequest({ streamer_id, stream_id, user_id, username, input, canonical_url, embed_url, provider, title, thumbnail_url, duration_seconds, cost, queue_position, currency }) {++function createMediaRequest({ streamer_id, stream_id, user_id, username, input, canonical_url, embed_url, provider, title, thumbnail_url, duration_seconds, cost, queue_position, currency, status, charge_state }) {return run(`INSERT INTO media_requests (streamer_id, stream_id, user_id, username, input, canonical_url, embed_url,-- provider, title, thumbnail_url, duration_seconds, cost, queue_position, currency-- ) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)`,++ provider, title, thumbnail_url, duration_seconds, cost, queue_position, currency, status, charge_state++ ) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)`,[streamer_id,stream_id || null,@@@ -6225,10 +6270,17 @@ function createMediaRequest({ streamer_id, stream_id, user_id, username, input,cost,queue_position ?? 0,currency || 'opencoins',++ status || 'pending',++ charge_state || null,]);}++/** Drop a paid request whose charge was refused (nothing was taken); only while it is still 'charging'. */++function removeUnchargedMediaRequest(id) {++ return run("DELETE FROM media_requests WHERE id = ? AND charge_state = 'charging'", [id]);++}++function getMediaRequestById(id) {return get('SELECT * FROM media_requests WHERE id = ?', [id]);}@@@ -6250,7 +6302,7 @@ function getPendingMediaRequestsByStreamer(streamerId, limit = 50) {}function getRecentMediaRequestsByStreamer(streamerId, limit = 15) {-- return all(`SELECT * FROM media_requests WHERE streamer_id = ? AND status IN ('played', 'skipped', 'removed', 'failed') ORDER BY COALESCE(ended_at, requested_at) DESC, id DESC LIMIT ?`, [streamerId, limit]);++ return all(`SELECT * FROM media_requests WHERE streamer_id = ? AND status IN ('played', 'skipped', 'removed', 'failed') AND charge_state IS NOT 'charging' ORDER BY COALESCE(ended_at, requested_at) DESC, id DESC LIMIT ?`, [streamerId, limit]);}function countPendingMediaRequestsForUser(streamerId, userId) {@@@ -7560,7 +7612,7 @@ module.exports = {// ProfilesgetUserProfile, updateUserAvatar,getKickChannelCache, setKickChannelCache,-- getChannelPoints, addChannelPoints, deductChannelPoints,++ getChannelPoints, addChannelPoints, deductChannelPoints, applyChannelPoints,// FollowsfollowUser, unfollowUser, getFollowerCount, isFollowing, getFollowerIds,// Transactions (Vibes)@@@ -7575,7 +7627,7 @@ module.exports = {upsertWatchTime, getWatchTime, getTotalWatchTime,// Media RequestsgetMediaRequestSettingsByUserId, upsertMediaRequestSettings,-- createMediaRequest, getMediaRequestById, getMediaRequestByStreamerAndId,++ createMediaRequest, removeUnchargedMediaRequest, getMediaRequestById, getMediaRequestByStreamerAndId,getActiveMediaRequestByStreamer, getNextPendingMediaRequest,getPendingMediaRequestsByStreamer, getRecentMediaRequestsByStreamer,countPendingMediaRequestsForUser, getMediaRequestMaxQueuePosition,@@@ -1539,6 +1539,11 @@ async function start() {if (out.deleted) console.log(`[Analytics] pruned ${out.deleted} raw events older than ${out.cutoff}`);}, { initialDelayMs: 5 * 60 * 1000, jitterMs: 60 * 1000 });++ // 8b3. Media requests whose OpenCoins charge never got an answer: charged again with the same++ // key (a replay if it landed) and refunded, so no viewer pays for a request that failed.++ require('./utils/jobs').every('media-charge-reconcile', 5 * 60 * 1000, () => require('./media/media-queue').reconcileCharges(),++ { initialDelayMs: 2 * 60 * 1000, jitterMs: 30 * 1000 });++// 8c. Durable events (roadmap Wave 3): stream lifecycle goes to OpenVibe.Events through the// transactional outbox (server/events/stream-events.js). Off unless EVENTS_URL is set.try { require('./events/stream-events').init(); } catch (err) { console.warn('[Events] not started:', err.message); }@@@ -122,10 +122,16 @@ class MediaQueue {* Take `cost` from the viewer in the channel's currency. Throws a message intended to* be shown to the viewer verbatim. The viewer was quoted this exact figure by /quote* before submitting, so what they agreed to is what gets taken.++ *++ * Every charge is keyed by the media request it pays for (ADR-012 rule 5): OpenCoins and++ * channel points `live:media_req:<request id>`, Vibes on Billing `live:media_charge:<request id>`.++ * Charging the same request again (a retry, the reconciler) replays; it never takes twice.++ * An error with `outcomeUnknown` means the wallet did not answer: it may or may not have landed.*/-- async charge({ currency, cost, userId, streamerId, streamId, label }) {++ async charge({ currency, cost, userId, streamerId, streamId, label, requestId }) {++ if (!requestId) throw new Error('charge() needs the media request id (its idempotency key)');if (currency === 'points') {-- if (!db.deductChannelPoints(userId, streamerId, cost)) {++ if (!db.deductChannelPoints(userId, streamerId, cost, `live:media_req:${requestId}`, label)) {const have = db.getChannelPoints(userId, streamerId);throw new Error(`Not enough channel points — this costs ${cost}, you have ${have}.`);}@@@ -139,7 +145,7 @@ class MediaQueue {// streamer's payable); the returned charge is linked to the request for refunds.const billingActions = require('../monetization/billing-actions');try {-- return await billingActions.chargeMedia({ userId, streamerId, streamId, cost, label });++ return await billingActions.chargeMedia({ userId, streamerId, streamId, cost, label, requestId });} catch (e) {const live = billingActions.toLive(e, { insufficient: `Not enough Vibes — this costs ${cost}.` });throw new Error(live ? live.body.error : (e?.message || `Could not charge ${cost} Vibes.`));@@@ -157,17 +163,65 @@ class MediaQueue {return;}// opencoins — network wallet++ const wallet = require('../monetization/wallet-client');++ const key = `live:media_req:${requestId}`;++ const noAnswer = (e) => !e || !e.status || e.status >= 500;++ let debited;try {-- const wallet = require('../monetization/wallet-client');-- const debited = await wallet.debit(-- userId, cost, label,-- `live:media_req:${streamerId}:${userId}:${Date.now()}:${Math.random().toString(36).slice(2, 8)}`-- );-- if (!debited) throw new Error('OpenCoins wallet unavailable — link your OpenVibe.Network account first.');++ debited = await wallet.debit(userId, cost, label, key);} catch (e) {if (e && e.status === 409) throw new Error(`Not enough OpenCoins — this costs ${cost}.`);-- throw e;++ if (!noAnswer(e)) throw e;++ // No answer (unreachable, 5xx): whether it landed is unknown. The same key makes the++ // retry safe, so ask once more; reconcileCharges() settles what is still unknown after.++ try {++ debited = await wallet.debit(userId, cost, label, key);++ } catch (e2) {++ if (e2 && e2.status === 409) throw new Error(`Not enough OpenCoins — this costs ${cost}.`);++ if (!noAnswer(e2)) throw e2;++ const err = new Error('Could not reach the OpenCoins wallet — nothing was queued. If you were charged, it is refunded automatically.');++ err.outcomeUnknown = true;++ throw err;++ }++ }++ if (!debited) throw new Error('OpenCoins wallet unavailable — link your OpenVibe.Network account first.');++ }++++ /**++ * Settle requests whose OpenCoins charge never got an answer (charge_state 'unknown', or a++ * 'charging' row left by a crash). The viewer was told it failed, so the request stays failed++ * and the coins must end where they started: charge again with the same key (a replay if the++ * first one landed, a new debit if it did not), then refund it (`live:media_refund:<id>`, also++ * keyed). A 409 on the re-charge means nothing was ever taken. Anything still unanswered waits++ * for the next run.++ */++ async reconcileCharges({ olderThanMs = 60_000, limit = 50 } = {}) {++ const cutoff = new Date(Date.now() - olderThanMs).toISOString().replace('T', ' ').slice(0, 19);++ const rows = db.all(`SELECT * FROM media_requests WHERE charge_state IN ('unknown', 'charging') AND currency = 'opencoins'++ AND requested_at <= ? ORDER BY id LIMIT ?`, [cutoff, limit]);++ const wallet = require('../monetization/wallet-client');++ let settled = 0;++ for (const r of rows) {++ const label = `Media request: ${r.title || r.input}`;++ try {++ let charged = true;++ try {++ const out = await wallet.debit(r.user_id, r.cost, label, `live:media_req:${r.id}`);++ if (!out) charged = false; // unlinked: nothing can have been taken++ } catch (e) {++ if (e && e.status === 409) charged = false;++ else throw e;++ }++ if (charged) {++ await wallet.credit(r.user_id, r.cost, `Refund: ${r.title || 'media request'}`, `live:media_refund:${r.id}`);++ }++ db.updateMediaRequest(r.id, { charge_state: null, refunded: charged ? 1 : 0, status: 'failed', last_error: r.last_error || 'The OpenCoins charge could not be confirmed' });++ settled++;++ } catch (e) {++ console.warn(`[MediaQueue] charge of request ${r.id} still unsettled: ${e.message}`);++ }}++ return settled;}/**@@@ -214,15 +268,10 @@ class MediaQueue {if (unpriceable) throw new Error(unpriceable);// A free channel stores cost 0, so the queue does not display a price nobody paid.const cost = currency === 'free' ? 0 : this.calculateCost(settings, normalized.duration_seconds);-- let billingCharge = null;-- if (cost > 0) {-- billingCharge = await this.charge({-- currency, cost, userId, streamerId, streamId,-- label: `Media request: ${normalized.title || trimmed}`,-- });-- }-- const queuePosition = db.getMediaRequestMaxQueuePosition(streamerId) + 1;++ // A paid request is written first, out of the queue (status 'failed', charge_state++ // 'charging'), so its charge is keyed by its id; it joins the queue once paid.++ const paid = cost > 0;const result = db.createMediaRequest({streamer_id: streamerId,stream_id: streamId,@@@ -237,8 +286,34 @@ class MediaQueue {duration_seconds: normalized.duration_seconds,cost,currency,-- queue_position: queuePosition,++ queue_position: paid ? 0 : db.getMediaRequestMaxQueuePosition(streamerId) + 1,++ status: paid ? 'failed' : 'pending',++ charge_state: paid ? 'charging' : null,});++ const requestId = Number(result.lastInsertRowid);++ let billingCharge = null;++ if (paid) {++ try {++ billingCharge = await this.charge({++ currency, cost, userId, streamerId, streamId, requestId,++ label: `Media request: ${normalized.title || trimmed}`,++ });++ } catch (e) {++ if (e && e.outcomeUnknown) {++ // Kept for reconcileCharges(), which refunds it if the debit did land.++ db.updateMediaRequest(requestId, { charge_state: 'unknown', last_error: 'The OpenCoins charge could not be confirmed' });++ } else {++ // Refused (not enough, unlinked, Billing said no): nothing was taken.++ db.removeUnchargedMediaRequest(requestId);++ }++ throw e;++ }++ // Paid: join the queue at the back (computed now, after the await).++ db.updateMediaRequest(requestId, {++ status: 'pending', charge_state: null,++ queue_position: db.getMediaRequestMaxQueuePosition(streamerId) + 1,++ });++ }// The coin ledger tracks OpenCoins only — booking a Vibes or channel-points charge// here would show up as a phantom coin spend in the viewer's history.@@@ -252,7 +327,7 @@ class MediaQueue {});}-- const request = db.getMediaRequestById(result.lastInsertRowid);++ const request = db.getMediaRequestById(requestId);if (billingCharge && billingCharge.actionId) require('../monetization/billing-actions').linkMediaCharge(billingCharge.actionId, request.id);this.broadcastQueueUpdate(streamerId);@@@ -500,7 +575,7 @@ class MediaQueue {const label = `Refund: ${request.title || 'media request'}`;try {if (currency === 'points') {-- db.addChannelPoints(request.user_id, request.streamer_id, amount);++ db.addChannelPoints(request.user_id, request.streamer_id, amount, `live:media_refund:${request.id}`, label);} else if (currency === 'vibes') {const money = require('../monetization/money-authority');if (money.writeRefusal()) {@@@ -177,12 +177,17 @@ async function donate(req, { toUserId, streamId, amount, message, goalId = null};}--/** Media request paid in Vibes: a paid interaction to the streamer. Returns { actionId, transactionId }. */--async function chargeMedia({ userId, streamerId, streamId, cost, label }) {++/**++ * Media request paid in Vibes: a paid interaction to the streamer. Returns { actionId, transactionId }.++ * Keyed by the request it pays for (`live:media_charge:<request id>`, ADR-012 rule 5), and linked++ * to it from the start, so a retried charge is replayed rather than taken twice.++ */++async function chargeMedia({ userId, streamerId, streamId, cost, label, requestId }) {++ if (!requestId) throw new Error('chargeMedia needs the media request id (its idempotency key)');const from = await subjectFor(userId);const to = await subjectFor(streamerId, { role: 'recipient' });const body = { from: ref(from), to: ref(to), amount: cost, kind: 'paid_interaction', ...(streamTarget(streamId) ? { target: streamTarget(streamId) } : {}), message: String(label || 'Media request').slice(0, 500), on_behalf_of: ref(from) };-- const { out, actionId } = await perform({ action: 'media_charge', path: '/transfers', body, key: `live:media_charge:${crypto.randomUUID()}`, liveUserId: userId });++ const { out, actionId } = await perform({ action: 'media_charge', path: '/transfers', body, key: `live:media_charge:${requestId}`, liveUserId: userId, liveRef: `media_request:${requestId}` });return { actionId, transactionId: out.transaction ? out.transaction.id : null };}@@@ -287,15 +287,21 @@ router.post('/redemptions/:id', requireAuth, (req, res) => {const s = db.getStreamById(redemption.stream_id);if (s?.user_id) refundStreamerId = s.user_id;}-- db.addChannelPoints(redemption.user_id, refundStreamerId, reward.cost);-- db.createCoinTransaction({-- user_id: redemption.user_id,-- stream_id: redemption.stream_id || null,-- amount: reward.cost,-- type: 'refund',-- reward_id: redemption.reward_id,-- message: `Refunded: ${reward.title} (rejected by streamer)`,++ // Keyed by the redemption: rejecting it twice refunds once.++ const refunded = db.applyChannelPoints({++ userId: redemption.user_id, streamerId: refundStreamerId, delta: reward.cost,++ key: `live:cp:redeem_refund:${redemption.id}`, reason: `refund redemption ${redemption.id}`,});++ if (refunded.applied) {++ db.createCoinTransaction({++ user_id: redemption.user_id,++ stream_id: redemption.stream_id || null,++ amount: reward.cost,++ type: 'refund',++ reward_id: redemption.reward_id,++ message: `Refunded: ${reward.title} (rejected by streamer)`,++ });++ }}}@@@ -312,7 +318,12 @@ router.post('/admin/grant', requireOwner, async (req, res) => {const { userId, amount, reason } = req.body;if (!userId || !amount) return res.status(400).json({ error: 'userId and amount required' });-- const newBalance = await openvibeCoins.adminGrant(userId, amount, reason);++ // A client Idempotency-Key makes a repeated submit the same grant (see adminGrant).++ const clientKey = req.get('Idempotency-Key');++ const newBalance = await openvibeCoins.adminGrant(userId, amount, reason, {++ adminId: req.user.id,++ clientKey: clientKey && /^[A-Za-z0-9._-]{8,100}$/.test(clientKey) ? clientKey : null,++ });res.json({ message: `Granted ${amount} OpenCoins`, balance: newBalance });} catch (err) {res.status(400).json({ error: err.message });@@@ -12,6 +12,11 @@* (see wallet-client.js); the legacy users.openvibe_coins_balance column is* frozen for the migration script and is never written anymore.*++ * Every channel-points debit and credit carries a deterministic per-event key (ADR-012 rule 5,++ * db.applyChannelPoints): watch = the watch_time row + minute, chat = the user's minute, follow =++ * user + streamer, bonus = the claim window, redeem / redeem refund = the redemption id. A retried++ * or replayed event moves nothing twice.++ ** Channel-point earning rates:* - Watching a live stream: per the streamer's config (default 10 / 5 min)* - Sending a chat message: 5 points (max 1 per minute)@@@ -74,7 +79,9 @@ class OpenCoins {coins *= COINS.STREAK_MULTIPLIER;}-- const total = db.addChannelPoints(userId, streamerId, coins);++ const r = db.applyChannelPoints({ userId, streamerId, delta: coins, key: `live:cp:watch:${wt.id}:${wt.minutes_watched}`, reason: 'watch' });++ if (!r.applied) return null;++ const total = r.balance;_feedPowerchat(streamerId, userId, coins);db.createCoinTransaction({user_id: userId,@@@ -110,7 +117,11 @@ class OpenCoins {if (!streamerId || streamerId === userId) return null;chatCooldowns.set(userId, now);-- const total = db.addChannelPoints(userId, streamerId, COINS.CHAT_BONUS);++ // The cooldown window is the event: after a restart (empty cooldown map) the same minute++ // still earns once.++ const r = db.applyChannelPoints({ userId, streamerId, delta: COINS.CHAT_BONUS, key: `live:cp:chat:${userId}:${Math.floor(now / COINS.CHAT_COOLDOWN_MS)}`, reason: 'chat' });++ if (!r.applied) return null;++ const total = r.balance;_feedPowerchat(streamerId, userId, COINS.CHAT_BONUS);db.createCoinTransaction({user_id: userId,@@@ -139,7 +150,9 @@ class OpenCoins {if (existing) return null;if (!streamerId || streamerId === userId) return null;-- const total = db.addChannelPoints(userId, streamerId, COINS.FOLLOW_BONUS);++ const r = db.applyChannelPoints({ userId, streamerId, delta: COINS.FOLLOW_BONUS, key: `live:cp:follow:${userId}:${streamerId}`, reason: 'follow' });++ if (!r.applied) return null;++ const total = r.balance;_feedPowerchat(streamerId, userId, COINS.FOLLOW_BONUS);db.createCoinTransaction({user_id: userId,@@@ -175,46 +188,48 @@ class OpenCoins {}if (!pointsStreamerId) throw new Error('No channel context for this reward');-- // Check user has enough points for this channel-- if (!db.deductChannelPoints(userId, pointsStreamerId, reward.cost)) {-- const cpName = (db.getChannelPointsConfig(pointsStreamerId).name) || 'Channel Points';-- throw new Error(`Not enough ${cpName}`);-- }---- // Check per-user cooldown-- if (reward.cooldown_seconds > 0) {-- const lastRedemption = db.get(-- `SELECT created_at FROM coin_redemptions WHERE reward_id = ? AND user_id = ? ORDER BY created_at DESC LIMIT 1`,-- [rewardId, userId]-- );-- if (lastRedemption) {-- const elapsed = (Date.now() - new Date(lastRedemption.created_at.replace(' ', 'T') + 'Z').getTime()) / 1000;-- if (elapsed < reward.cooldown_seconds) {-- db.addChannelPoints(userId, pointsStreamerId, reward.cost); // refund-- throw new Error(`Cooldown: wait ${Math.ceil(reward.cooldown_seconds - elapsed)}s`);++ // The limits are checked, the redemption row created and the points taken in one++ // transaction, keyed by the redemption id: nothing is taken for a refused redemption++ // (there is no take-then-refund any more), and a replayed spend moves nothing twice.++ const result = db.getDb().transaction(() => {++ // Check per-user cooldown++ if (reward.cooldown_seconds > 0) {++ const lastRedemption = db.get(++ `SELECT created_at FROM coin_redemptions WHERE reward_id = ? AND user_id = ? ORDER BY created_at DESC LIMIT 1`,++ [rewardId, userId]++ );++ if (lastRedemption) {++ const elapsed = (Date.now() - new Date(lastRedemption.created_at.replace(' ', 'T') + 'Z').getTime()) / 1000;++ if (elapsed < reward.cooldown_seconds) {++ throw new Error(`Cooldown: wait ${Math.ceil(reward.cooldown_seconds - elapsed)}s`);++ }}}-- }-- // Check max per stream-- if (reward.max_per_stream > 0 && streamId) {-- const count = db.get(-- `SELECT COUNT(*) as c FROM coin_redemptions WHERE reward_id = ? AND stream_id = ?`,-- [rewardId, streamId]-- );-- if (count && count.c >= reward.max_per_stream) {-- db.addChannelPoints(userId, pointsStreamerId, reward.cost); // refund-- throw new Error('Max redemptions reached for this stream');++ // Check max per stream++ if (reward.max_per_stream > 0 && streamId) {++ const count = db.get(++ `SELECT COUNT(*) as c FROM coin_redemptions WHERE reward_id = ? AND stream_id = ?`,++ [rewardId, streamId]++ );++ if (count && count.c >= reward.max_per_stream) {++ throw new Error('Max redemptions reached for this stream');++ }}-- }-- // Create redemption-- const result = db.createCoinRedemption({-- reward_id: rewardId,-- user_id: userId,-- stream_id: streamId,-- user_input: userInput,-- });++ // Create redemption, then take the points for it (rolled back together if short)++ const created = db.createCoinRedemption({++ reward_id: rewardId,++ user_id: userId,++ stream_id: streamId,++ user_input: userInput,++ });++ if (!db.deductChannelPoints(userId, pointsStreamerId, reward.cost, `live:cp:redeem:${created.lastInsertRowid}`, `redeem reward ${rewardId}`)) {++ const cpName = (db.getChannelPointsConfig(pointsStreamerId).name) || 'Channel Points';++ throw new Error(`Not enough ${cpName}`);++ }++ return created;++ })();// Log transactiondb.createCoinTransaction({@@@ -256,7 +271,10 @@ class OpenCoins {if (now - (bonusClaims.get(key) || 0) < windowMs) return null;bonusClaims.set(key, now);const amount = Math.max(1, (cfg.watch_amount || 10) * 3);-- const total = db.addChannelPoints(userId, streamerId, amount);++ // One claim per window, also across a restart (the throttle map above is in memory).++ const r = db.applyChannelPoints({ userId, streamerId, delta: amount, key: `live:cp:bonus:${userId}:${streamerId}:${Math.floor(now / windowMs)}`, reason: 'bonus game' });++ if (!r.applied) return null;++ const total = r.balance;_feedPowerchat(streamerId, userId, amount);db.createCoinTransaction({ user_id: userId, stream_id: streamId, amount, type: 'watch', message: 'Bonus game' });return { coins: amount, total, streamerId };@@@ -301,11 +319,37 @@ class OpenCoins {/*** Admin: grant OpenCoins to a user (network wallet credit).++ *++ * The grant is recorded locally first and the wallet credit is keyed by that record++ * (`live:admin_grant:<grant id>`, ADR-012 rule 5), so retrying a grant whose answer was lost++ * never credits twice. With `clientKey` (the request's Idempotency-Key) a repeated submit is++ * the same grant: it reuses the record and its key instead of making a second one.*/-- async adminGrant(userId, amount, reason) {-- const key = `live:admin_grant:${userId}:${Date.now()}:${Math.random().toString(36).slice(2, 8)}`;-- const result = await wallet.credit(userId, amount, reason || 'Admin grant', key);++ async adminGrant(userId, amount, reason, { adminId = null, clientKey = null } = {}) {++ const d = db.getDb();++ d.exec(`CREATE TABLE IF NOT EXISTS opencoin_admin_grants (++ id INTEGER PRIMARY KEY AUTOINCREMENT,++ client_key TEXT UNIQUE,++ admin_id INTEGER,++ user_id INTEGER NOT NULL,++ amount INTEGER NOT NULL,++ reason TEXT,++ balance INTEGER,++ created_at DATETIME DEFAULT CURRENT_TIMESTAMP++ )`);++ const ck = clientKey ? `a${adminId || 0}:${clientKey}` : null;++ let grant = ck ? d.prepare('SELECT * FROM opencoin_admin_grants WHERE client_key = ?').get(ck) : null;++ if (grant && (grant.user_id !== Number(userId) || grant.amount !== Number(amount))) {++ throw new Error('That Idempotency-Key was already used for a different grant');++ }++ if (!grant) {++ const id = d.prepare('INSERT INTO opencoin_admin_grants (client_key, admin_id, user_id, amount, reason) VALUES (?, ?, ?, ?, ?)')++ .run(ck, adminId, Number(userId), Number(amount), reason || null).lastInsertRowid;++ grant = { id };++ }++ const result = await wallet.credit(userId, amount, reason || 'Admin grant', `live:admin_grant:${grant.id}`);if (!result) throw new Error('User has no linked OpenVibe.Network account (wallet unavailable)');++ d.prepare('UPDATE opencoin_admin_grants SET balance = ? WHERE id = ?').run(result.balance ?? null, grant.id);return result.balance;}@@@ -298,12 +298,18 @@ async function check(name, fn) {db.setSetting('payments_enabled', 'false');});-- await check('a Vibes media request is a paid interaction; its refund reverses that exact transfer', async () => {++ await check('a Vibes media request is a paid interaction keyed by its request; its refund reverses that exact transfer', async () => {const mediaQueue = require('../server/media/media-queue');-- const charge = await mediaQueue.charge({ currency: 'vibes', cost: 40, userId: 2, streamerId: 3, streamId: streamB, label: 'Media request: song' });-- const c = lastCall();-- assert.strictEqual(c.path, '/transfers'); assert.strictEqual(c.body.kind, 'paid_interaction'); assert.match(c.key, /^live:media_charge:/);const reqId = Number(db.createMediaRequest({ streamer_id: 3, stream_id: streamB, user_id: 2, username: 'ann', input: 'x', canonical_url: 'https://x.test/a', embed_url: null, provider: 'youtube', title: 'song', thumbnail_url: null, duration_seconds: 60, cost: 40, queue_position: 1, currency: 'vibes' }).lastInsertRowid);++ const charge = await mediaQueue.charge({ currency: 'vibes', cost: 40, userId: 2, streamerId: 3, streamId: streamB, label: 'Media request: song', requestId: reqId });++ const c = lastCall();++ assert.strictEqual(c.path, '/transfers'); assert.strictEqual(c.body.kind, 'paid_interaction');++ assert.strictEqual(c.key, `live:media_charge:${reqId}`, 'the charge is keyed by the request it pays for (ADR-012 rule 5)');++ // A retried charge for the same request is a replay: Billing is not asked to move money again.++ const before = billing.calls.length;++ const again = await mediaQueue.charge({ currency: 'vibes', cost: 40, userId: 2, streamerId: 3, streamId: streamB, label: 'Media request: song', requestId: reqId });++ assert.strictEqual(billing.calls.length, before, 'a retried charge does not reach Billing twice');++ assert.strictEqual(again.transactionId, charge.transactionId);billingActions.linkMediaCharge(charge.actionId, reqId);const refunded = await mediaQueue.refund(reqId);assert.strictEqual(refunded, 40);@@@ -312,7 +318,7 @@ async function check(name, fn) {assert.strictEqual(r.key, `live:media_refund:${reqId}`);assert.strictEqual(db.getMediaRequestById(reqId).refunded, 1);assert.strictEqual(await mediaQueue.refund(reqId), 0, 'a second refund is a no-op');-- await assert.rejects(mediaQueue.charge({ currency: 'vibes', cost: 999999, userId: 2, streamerId: 3, streamId: streamB, label: 'x' }), /Not enough Vibes — this costs 999999/);++ await assert.rejects(mediaQueue.charge({ currency: 'vibes', cost: 999999, userId: 2, streamerId: 3, streamId: streamB, label: 'x', requestId: reqId + 1000 }), /Not enough Vibes — this costs 999999/);});await check('webhooks moved: PowerChat and card webhooks on Live answer 410 with Billing\'s URL', async () => {nnew file mode 100644@@@ -0,0 +1,242 @@++/**++ * Loyalty ledgers use deterministic per-event idempotency keys (ADR-012 rule 5), so a retry never++ * spends or credits twice.++ *++ * - Channel points: every debit and credit goes through db.applyChannelPoints with a key, logged++ * in channel_points_log; a repeated key moves nothing, a reused key for another event is refused,++ * a refused debit leaves no row. Earn keys: watch row + minute, chat minute, follow, bonus window;++ * a redemption is taken under its redemption id (limits first, no take-then-refund), and a++ * streamer rejecting it twice refunds once.++ * - Media requests: the request row is written first and the charge is keyed by its id++ * (live:media_req:<id> for OpenCoins and points; Billing's key is tested in billing-authority).++ * A wallet that never answers leaves the request 'unknown'; reconcileCharges() re-charges with++ * the same key (a replay) and refunds, so the viewer ends where they started.++ * - Admin OpenCoins grants: keyed by a local grant record; an Idempotency-Key repeat is the same grant.++ *++ * node test/loyalty-keys.test.js++ */++'use strict';++const assert = require('assert');++const fs = require('fs');++const os = require('os');++const path = require('path');++const http = require('http');++++const tmp = fs.mkdtempSync(path.join(os.tmpdir(), 'ov-loyalty-'));++process.env.DB_PATH = path.join(tmp, 'live.db');++process.env.NODE_ENV = 'test';++const quiet = console.log;++console.log = () => {};++console.warn = () => {};++++const db = require('../server/db/database');++db.initDb();++const raw = db.getDb();++const auth = require('../server/auth/auth');++const signIn = (req) => {++ const id = Number(req.headers['x-test-user'] || 0);++ const u = id ? db.getUserById(id) : null;++ if (u) req.user = u;++ return u;++};++auth.requireAuth = (req, res, next) => (signIn(req) ? next() : res.status(401).json({ error: 'Authentication required' }));++++const addUser = (id, name, role = 'streamer', owner = 0) => raw.prepare(`INSERT INTO users (id, username, display_name, email, password_hash, role, is_owner) VALUES (?, ?, ?, ?, 'x', ?, ?)`).run(id, name, name, `${name}@x`, role, owner);++addUser(1, 'owner', 'admin', 1);++addUser(2, 'viewer');++addUser(3, 'streamer');++db.ensureChannel(3);++const chan = db.getChannelByUserId(3);++const streamId = Number(db.createStream({ user_id: 3, channel_id: chan.id, title: 'live', protocol: 'webrtc' }).lastInsertRowid);++const cp = () => db.getChannelPoints(2, 3);++const logRows = (key) => raw.prepare('SELECT * FROM channel_points_log WHERE idempotency_key = ?').all(key);++++// A Network wallet that behaves like the real one: one effect per idempotency key.++const wallet = require('../server/monetization/wallet-client');++const ledger = { balance: 1000, keys: new Map(), calls: [], failNext: 0, landBeforeFailing: false };++function walletOp(kind) {++ return async (userId, amount, reason, key) => {++ ledger.calls.push({ kind, key, amount });++ if (ledger.failNext > 0) {++ ledger.failNext--;++ if (ledger.landBeforeFailing && !ledger.keys.has(key)) { ledger.keys.set(key, { kind, amount }); ledger.balance += kind === 'debit' ? -amount : amount; }++ const e = new Error('Network wallet unreachable: timeout'); e.status = 0; throw e;++ }++ if (ledger.keys.has(key)) return { balance: ledger.balance, replayed: true };++ if (kind === 'debit' && ledger.balance < amount) { const e = new Error('insufficient_funds'); e.status = 409; throw e; }++ ledger.keys.set(key, { kind, amount });++ ledger.balance += kind === 'debit' ? -amount : amount;++ return { balance: ledger.balance };++ };++}++wallet.debit = walletOp('debit');++wallet.credit = walletOp('credit');++++(async () => {++ // ── 1. The keyed channel-points log ──────────────────────────────────────────────++ assert.throws(() => db.addChannelPoints(2, 3, 10), /idempotency key is required/, 'no unkeyed credit');++ assert.throws(() => db.deductChannelPoints(2, 3, 10), /idempotency key is required/, 'no unkeyed debit');++ assert.strictEqual(db.addChannelPoints(2, 3, 100, 'live:cp:test:credit:1'), 100);++ assert.strictEqual(db.addChannelPoints(2, 3, 100, 'live:cp:test:credit:1'), 100, 'a retried credit is applied once');++ assert.strictEqual(db.deductChannelPoints(2, 3, 30, 'live:cp:test:debit:1'), true);++ assert.strictEqual(db.deductChannelPoints(2, 3, 30, 'live:cp:test:debit:1'), true, 'a retried debit answers as taken…');++ assert.strictEqual(cp(), 70, '…but takes once');++ assert.throws(() => db.addChannelPoints(2, 3, 5, 'live:cp:test:credit:1'), /already used for a different event/, 'a key reused for another amount is refused');++ assert.strictEqual(db.deductChannelPoints(2, 3, 500, 'live:cp:test:debit:big'), false, 'not enough: refused');++ assert.strictEqual(logRows('live:cp:test:debit:big').length, 0, 'a refused debit leaves no log row');++ db.addChannelPoints(2, 3, 500, 'live:cp:test:topup');++ assert.strictEqual(db.deductChannelPoints(2, 3, 500, 'live:cp:test:debit:big'), true, 'so the same key can be tried again once the balance covers it');++ assert.strictEqual(cp(), 70);++++ // ── 2. Earning: follow, chat (across a "restart"), watch ───────────────────────++ let coins = require('../server/monetization/opencoins');++ assert.ok(coins.awardFollow(2, 3));++ raw.prepare("DELETE FROM coin_transactions WHERE type = 'follow_bonus'").run(); // even without the old LIKE check…++ assert.strictEqual(coins.awardFollow(2, 3), null, '…a second follow bonus is refused by its key');++ assert.strictEqual(logRows('live:cp:follow:2:3').length, 1);++ const afterFollow = cp();++ assert.ok(coins.awardChat(2, streamId));++ delete require.cache[require.resolve('../server/monetization/opencoins')];++ coins = require('../server/monetization/opencoins'); // a restart: the in-memory chat cooldown is gone++ const minute = Math.floor(Date.now() / 60_000);++ const secondChat = coins.awardChat(2, streamId);++ if (Math.floor(Date.now() / 60_000) === minute) assert.strictEqual(secondChat, null, 'the same chat minute earns once, even after a restart');++ assert.strictEqual(cp(), afterFollow + 5 + (secondChat ? 5 : 0));++ raw.prepare('INSERT INTO watch_time (user_id, stream_id, minutes_watched) VALUES (2, ?, 4)').run(streamId);++ const watched = coins.awardWatch(2, streamId); // minute 5 of the default 5-minute interval++ assert.ok(watched && watched.coins === 10);++ const wt = db.getWatchTime(2, streamId);++ assert.strictEqual(logRows(`live:cp:watch:${wt.id}:5`).length, 1, 'the watch award is keyed by the watch row and minute');++ assert.strictEqual(db.applyChannelPoints({ userId: 2, streamerId: 3, delta: 10, key: `live:cp:watch:${wt.id}:5` }).replayed, true, 'replaying that minute moves nothing');++++ // ── 3. Redemptions: limits before the spend, keyed by redemption, reject refunds once ──++ const rewardId = Number(db.createCoinReward({ streamer_id: 3, title: 'Hydrate', cost: 40, cooldown_seconds: 3600 }).lastInsertRowid);++ const beforeRedeem = cp();++ const red = coins.redeem(2, rewardId, streamId, 'drink');++ assert.strictEqual(cp(), beforeRedeem - 40);++ assert.strictEqual(logRows(`live:cp:redeem:${red.redemption.id}`).length, 1, 'the spend is keyed by the redemption id');++ const redemptions = raw.prepare('SELECT COUNT(*) AS n FROM coin_redemptions').get().n;++ assert.throws(() => coins.redeem(2, rewardId, streamId, 'again'), /Cooldown/);++ assert.strictEqual(cp(), beforeRedeem - 40, 'a refused redemption takes nothing (no take-then-refund)');++ assert.strictEqual(raw.prepare('SELECT COUNT(*) AS n FROM coin_redemptions').get().n, redemptions, 'and leaves no redemption row');++ const pricey = Number(db.createCoinReward({ streamer_id: 3, title: 'Big', cost: 1_000_000 }).lastInsertRowid);++ assert.throws(() => coins.redeem(2, pricey, streamId, ''), /Not enough/);++ assert.strictEqual(raw.prepare('SELECT COUNT(*) AS n FROM coin_redemptions').get().n, redemptions, 'not enough points: the redemption row is rolled back');++++ const express = require('express');++ const app = express();++ app.use(express.json());++ app.use((req, res, next) => { signIn(req); next(); });++ app.use('/api/coins', require('../server/monetization/coins-routes'));++ const server = http.createServer(app).listen(0);++ const call = (method, p, user, body, headers = {}) => new Promise((resolve, reject) => {++ const req = http.request({ port: server.address().port, path: p, method, headers: { 'content-type': 'application/json', 'x-test-user': String(user), ...headers } }, (res) => {++ let text = '';++ res.on('data', (c) => { text += c; });++ res.on('end', () => { let json = null; try { json = JSON.parse(text); } catch { /* */ } resolve({ status: res.statusCode, json }); });++ });++ req.on('error', reject);++ req.end(body ? JSON.stringify(body) : undefined);++ });++ const beforeReject = cp();++ assert.strictEqual((await call('POST', `/api/coins/redemptions/${red.redemption.id}`, 3, { status: 'rejected' })).status, 200);++ assert.strictEqual((await call('POST', `/api/coins/redemptions/${red.redemption.id}`, 3, { status: 'rejected' })).status, 200);++ assert.strictEqual(cp(), beforeReject + 40, 'rejecting the same redemption twice refunds once');++ assert.strictEqual(logRows(`live:cp:redeem_refund:${red.redemption.id}`).length, 1);++++ // ── 4. Media requests ───────────────────────────────────────────────────────────++ const mq = require('../server/media/media-queue');++ let n = 0;++ mq.normalizeInput = async () => { n++; return { canonical_url: `https://media.example/${n}.mp4`, embed_url: null, provider: 'video', title: `clip ${n}`, thumbnail_url: null, duration_seconds: 30, isLive: false }; };++ mq.extractStreamUrlForRequest = async () => null;++ mq.broadcastQueueUpdate = () => {};++ db.upsertMediaRequestSettings(3, { enabled: 1, request_cost: 25, currency: 'opencoins', max_per_user: 10 });++++ const start = ledger.balance;++ const req1 = await mq.addRequest({ streamerId: 3, streamId, userId: 2, username: 'viewer', input: 'https://media.example/a.mp4' });++ assert.strictEqual(req1.status, 'pending');++ assert.strictEqual(ledger.calls.at(-1).key, `live:media_req:${req1.id}`, 'the OpenCoins spend is keyed by the request id');++ assert.strictEqual(ledger.balance, start - 25);++ // A retry of that charge (same request) is a replay at the wallet: nothing more is taken.++ await mq.charge({ currency: 'opencoins', cost: 25, userId: 2, streamerId: 3, streamId, label: 'retry', requestId: req1.id });++ assert.strictEqual(ledger.balance, start - 25, 'retrying the charge does not spend twice');++ assert.ok(db.getMediaRequestById(req1.id).queue_position >= 1, 'a paid request joins the queue');++++ // The wallet times out after the debit landed, twice: the viewer is told it failed and the++ // request waits, out of the queue, for the reconciler.++ ledger.failNext = 2; ledger.landBeforeFailing = true;++ const queueBefore = db.all("SELECT id FROM media_requests WHERE status = 'pending'").length;++ await assert.rejects(mq.addRequest({ streamerId: 3, streamId, userId: 2, username: 'viewer', input: 'https://media.example/b.mp4' }), /Could not reach the OpenCoins wallet/);++ ledger.landBeforeFailing = false;++ const unknown = raw.prepare("SELECT * FROM media_requests WHERE charge_state = 'unknown'").get();++ assert.ok(unknown, 'the request is kept with its charge unknown');++ assert.strictEqual(unknown.status, 'failed');++ assert.strictEqual(db.all("SELECT id FROM media_requests WHERE status = 'pending'").length, queueBefore, 'and is not in the queue');++ assert.strictEqual(ledger.balance, start - 50, 'the debit did land');++ assert.strictEqual(await mq.reconcileCharges({ olderThanMs: -60_000 }), 1);++ assert.strictEqual(ledger.balance, start - 25, 'the reconciler re-charged with the same key (a replay) and refunded');++ assert.deepStrictEqual(ledger.calls.filter((c) => c.key === `live:media_req:${unknown.id}`).map((c) => c.kind), ['debit', 'debit', 'debit'], 'every attempt used the one key');++ const settled = db.getMediaRequestById(unknown.id);++ assert.strictEqual(settled.charge_state, null);++ assert.strictEqual(settled.refunded, 1);++ assert.strictEqual(await mq.reconcileCharges({ olderThanMs: -60_000 }), 0, 'nothing left to settle');++ assert.strictEqual(mq.refund(unknown.id), 0, 'and it is not refunded a second time');++++ // A wallet that never answered and never took anything: the re-charge takes it, the refund gives it back.++ ledger.failNext = 2;++ await assert.rejects(mq.addRequest({ streamerId: 3, streamId, userId: 2, username: 'viewer', input: 'https://media.example/c.mp4' }), /Could not reach/);++ await mq.reconcileCharges({ olderThanMs: -60_000 });++ assert.strictEqual(ledger.balance, start - 25, 'net zero either way');++++ // Not enough OpenCoins: nothing taken, no row left behind.++ const rows = raw.prepare('SELECT COUNT(*) AS n FROM media_requests').get().n;++ ledger.balance = 10;++ await assert.rejects(mq.addRequest({ streamerId: 3, streamId, userId: 2, username: 'viewer', input: 'https://media.example/d.mp4' }), /Not enough OpenCoins/);++ assert.strictEqual(raw.prepare('SELECT COUNT(*) AS n FROM media_requests').get().n, rows, 'a refused charge leaves no request row');++ assert.strictEqual(ledger.balance, 10);++++ // Channel points: spend keyed by the request, refund keyed by the request, once.++ db.upsertMediaRequestSettings(3, { currency: 'points' });++ const pts = cp();++ const req2 = await mq.addRequest({ streamerId: 3, streamId, userId: 2, username: 'viewer', input: 'https://media.example/e.mp4' });++ assert.strictEqual(cp(), pts - 25);++ assert.strictEqual(logRows(`live:media_req:${req2.id}`).length, 1);++ assert.strictEqual(db.deductChannelPoints(2, 3, 25, `live:media_req:${req2.id}`), true, 'a retried points charge is a replay');++ assert.strictEqual(cp(), pts - 25);++ assert.strictEqual(mq.refund(req2.id), 25);++ db.updateMediaRequest(req2.id, { refunded: 0 }); // even if the refunded flag were lost…++ mq.refund(req2.id);++ assert.strictEqual(cp(), pts, '…the refund key credits once');++++ // ── 5. Admin OpenCoins grants ─────────────────────────────────────────────────++ const g0 = ledger.balance;++ let r = await call('POST', '/api/coins/admin/grant', 1, { userId: 2, amount: 50, reason: 'thanks' }, { 'Idempotency-Key': 'grant-abc-123' });++ assert.strictEqual(r.status, 200, JSON.stringify(r.json));++ const grantKey = ledger.calls.at(-1).key;++ assert.match(grantKey, /^live:admin_grant:\d+$/, 'the grant is keyed by its local record');++ r = await call('POST', '/api/coins/admin/grant', 1, { userId: 2, amount: 50, reason: 'thanks' }, { 'Idempotency-Key': 'grant-abc-123' });++ assert.strictEqual(ledger.calls.at(-1).key, grantKey, 'a repeated submit reuses the grant and its key');++ assert.strictEqual(ledger.balance, g0 + 50, 'and credits once');++ r = await call('POST', '/api/coins/admin/grant', 1, { userId: 2, amount: 99 }, { 'Idempotency-Key': 'grant-abc-123' });++ assert.strictEqual(r.status, 400, 'the same Idempotency-Key for a different grant is refused');++ await call('POST', '/api/coins/admin/grant', 1, { userId: 2, amount: 50 });++ assert.notStrictEqual(ledger.calls.at(-1).key, grantKey, 'a new grant without a key is a new event');++ assert.strictEqual(ledger.balance, g0 + 100);++++ // ── No random keys left in the loyalty paths ───────────────────────────────────++ for (const f of ['server/media/media-queue.js', 'server/monetization/opencoins.js', 'server/monetization/coins-routes.js']) {++ const keyLines = fs.readFileSync(path.join(__dirname, '..', f), 'utf8').split('\n').filter((l) => /`live:(media_req|media_refund|admin_grant|cp:)/.test(l));++ assert.ok(keyLines.length, `${f} builds loyalty keys`);++ for (const l of keyLines) assert.ok(!/Date\.now\(\)|Math\.random\(\)|randomUUID/.test(l), `${f}: a loyalty key never uses the clock or randomness: ${l.trim()}`);++ }++ const ba = fs.readFileSync(path.join(__dirname, '..', 'server/monetization/billing-actions.js'), 'utf8');++ assert.ok(/key: `live:media_charge:\$\{requestId\}`/.test(ba), 'the Billing media charge is keyed by the request id');++++ server.close();++ fs.rmSync(tmp, { recursive: true, force: true });++ quiet('loyalty-keys: ok');++ process.exit(0);++})().catch((err) => {++ quiet(err);++ process.exit(1);++});@@@ -100,7 +100,7 @@ async function check(name, fn) {assert.strictEqual(r.status, 503, `${label} ${p}: ${r.text}`);assert.strictEqual(r.json.code, 'money_writes_frozen', `${label} ${p}`);}-- await assert.rejects(mediaQueue.charge({ currency: 'vibes', cost: 10, userId: 2, streamerId: 3, streamId: null, label: 'x' }), /paused/);++ await assert.rejects(mediaQueue.charge({ currency: 'vibes', cost: 10, userId: 2, streamerId: 3, streamId: null, label: 'x', requestId: 9001 }), /paused/);}await check('only the owner can freeze; every admin can see it', async () => {
@@ -771,6 +771,19 @@ function initDb() {
@@ -771,6 +771,19 @@ function initDb() {
updated_at DATETIME DEFAULT CURRENT_TIMESTAMP,
updated_at DATETIME DEFAULT CURRENT_TIMESTAMP,
PRIMARY KEY (user_id, streamer_id)
PRIMARY KEY (user_id, streamer_id)
)`);
)`);
+ // Every channel-points debit and credit, keyed per event (ADR-012 rule 5): a retried award,
+ // spend or refund with the same key is applied once. Keys: live:cp:<event>:<id>, and
+ // live:media_req:<request id> / live:media_refund:<request id> for media requests.
+ database.exec(`CREATE TABLE IF NOT EXISTS channel_points_log (
+ id INTEGER PRIMARY KEY AUTOINCREMENT,
+ idempotency_key TEXT NOT NULL UNIQUE,
+ user_id INTEGER NOT NULL,
+ streamer_id INTEGER NOT NULL,
+ delta INTEGER NOT NULL,
+ reason TEXT,
+ created_at DATETIME DEFAULT CURRENT_TIMESTAMP
+ )`);
+ database.exec('CREATE INDEX IF NOT EXISTS idx_cp_log_user ON channel_points_log(user_id, streamer_id, id)');
} catch (e) { console.warn('[DB] channel_points migration:', e.message); }
} catch (e) { console.warn('[DB] channel_points migration:', e.message); }
// Kick chatroom-id cache. Kick's v2 API (which exposes the Pusher chatroom id)
// Kick chatroom-id cache. Kick's v2 API (which exposes the Pusher chatroom id)
@@ -929,6 +942,11 @@ function initDb() {
@@ -929,6 +942,11 @@ function initDb() {
// setting — a streamer switching from Vibes to points would otherwise refund the
// setting — a streamer switching from Vibes to points would otherwise refund the
// wrong currency to everyone still queued.
// wrong currency to everyone still queued.
if (!mrCols.includes('currency')) database.exec("ALTER TABLE media_requests ADD COLUMN currency TEXT DEFAULT 'opencoins'");
if (!mrCols.includes('currency')) database.exec("ALTER TABLE media_requests ADD COLUMN currency TEXT DEFAULT 'opencoins'");
+ // A paid request is written before it is charged, so the charge can be keyed by its id
+ // (live:media_req:<id>, ADR-012 rule 5). Until the charge answers it is status 'failed'
+ // with charge_state 'charging' (out of the queue and the history); 'unknown' means the
+ // wallet never answered and mediaQueue.reconcileCharges() settles it. NULL = settled.
+ if (!mrCols.includes('charge_state')) database.exec('ALTER TABLE media_requests ADD COLUMN charge_state TEXT');
const msCols = database.pragma('table_info(media_request_settings)').map(c => c.name);
const msCols = database.pragma('table_info(media_request_settings)').map(c => c.name);
if (!msCols.includes('cost_mode')) database.exec("ALTER TABLE media_request_settings ADD COLUMN cost_mode TEXT DEFAULT 'flat' CHECK(cost_mode IN ('flat','per_minute'))");
if (!msCols.includes('cost_mode')) database.exec("ALTER TABLE media_request_settings ADD COLUMN cost_mode TEXT DEFAULT 'flat' CHECK(cost_mode IN ('flat','per_minute'))");
@@ -4508,22 +4526,49 @@ function getChannelPoints(userId, streamerId) {
@@ -4508,22 +4526,49 @@ function getChannelPoints(userId, streamerId) {
const r = get('SELECT balance FROM channel_points WHERE user_id = ? AND streamer_id = ?', [userId, streamerId]);
const r = get('SELECT balance FROM channel_points WHERE user_id = ? AND streamer_id = ?', [userId, streamerId]);
return r ? r.balance : 0;
return r ? r.balance : 0;
}
}
-function addChannelPoints(userId, streamerId, amount) {
- if (!userId || !streamerId || !amount) return getChannelPoints(userId, streamerId);
- run(`INSERT INTO channel_points (user_id, streamer_id, balance, updated_at)
- VALUES (?, ?, ?, CURRENT_TIMESTAMP)
- ON CONFLICT(user_id, streamer_id) DO UPDATE SET
- balance = balance + excluded.balance, updated_at = CURRENT_TIMESTAMP`,
- [userId, streamerId, amount]);
- return getChannelPoints(userId, streamerId);
+/**
+ * Apply one channel-points event, keyed (ADR-012 rule 5): { applied, replayed, balance }.
+ * A key seen before is a replay: nothing moves, and the same key for a different user, channel or
+ * amount is refused. A debit (delta < 0) only happens when the balance covers it; a refused debit
+ * leaves no log row, so the same key can be tried again later.
+ */
+function applyChannelPoints({ userId, streamerId, delta, key, reason = null }) {
+ if (!key || typeof key !== 'string') throw new TypeError('channel points: an idempotency key is required for every debit and credit');
+ if (!userId || !streamerId || !Number.isInteger(delta) || delta === 0) {
+ return { applied: false, replayed: false, balance: getChannelPoints(userId, streamerId) };
+ }
+ return getDb().transaction(() => {
+ const seen = get('SELECT user_id, streamer_id, delta FROM channel_points_log WHERE idempotency_key = ?', [key]);
+ if (seen) {
+ if (seen.user_id !== userId || seen.streamer_id !== streamerId || seen.delta !== delta) {
+ throw new Error(`channel points: idempotency key ${key} was already used for a different event`);
+ }
+ return { applied: false, replayed: true, balance: getChannelPoints(userId, streamerId) };
+ }
+ if (delta < 0) {
+ const res = run(`UPDATE channel_points SET balance = balance + ?, updated_at = CURRENT_TIMESTAMP
+ WHERE user_id = ? AND streamer_id = ? AND balance >= ?`, [delta, userId, streamerId, -delta]);
+ if (!res.changes) return { applied: false, replayed: false, balance: getChannelPoints(userId, streamerId) };
+ } else {
+ run(`INSERT INTO channel_points (user_id, streamer_id, balance, updated_at)
+ VALUES (?, ?, ?, CURRENT_TIMESTAMP)
+ ON CONFLICT(user_id, streamer_id) DO UPDATE SET
+ balance = balance + excluded.balance, updated_at = CURRENT_TIMESTAMP`, [userId, streamerId, delta]);
+ }
+ run('INSERT INTO channel_points_log (idempotency_key, user_id, streamer_id, delta, reason) VALUES (?, ?, ?, ?, ?)',
+ [key, userId, streamerId, delta, reason ? String(reason).slice(0, 200) : null]);
+ return { applied: true, replayed: false, balance: getChannelPoints(userId, streamerId) };
+ })();
+}
+/** Credit channel points for one event (key required). Returns the new balance. */
+function addChannelPoints(userId, streamerId, amount, key, reason) {
+ return applyChannelPoints({ userId, streamerId, delta: amount, key, reason }).balance;
}
}
-// Atomic spend — returns true only if the viewer had enough for this streamer.
-function deductChannelPoints(userId, streamerId, amount) {
+/** Atomic spend for one event (key required): true if taken now or already taken under this key. */
+function deductChannelPoints(userId, streamerId, amount, key, reason) {
if (!userId || !streamerId) return false;
if (!userId || !streamerId) return false;
- const res = run(`UPDATE channel_points SET balance = balance - ?, updated_at = CURRENT_TIMESTAMP
- WHERE user_id = ? AND streamer_id = ? AND balance >= ?`,
- [amount, userId, streamerId, amount]);
- return (res?.changes || 0) > 0;
+ const r = applyChannelPoints({ userId, streamerId, delta: -amount, key, reason });
+ return r.applied || r.replayed;
}
}
// ── Kick chatroom-id cache (survives the Cloudflare-blocked v2 API) ──
// ── Kick chatroom-id cache (survives the Cloudflare-blocked v2 API) ──
@@ -6204,12 +6249,12 @@ function upsertMediaRequestSettings(userId, fields = {}) {
@@ -6204,12 +6249,12 @@ function upsertMediaRequestSettings(userId, fields = {}) {
return getMediaRequestSettingsByUserId(userId);
return getMediaRequestSettingsByUserId(userId);
}
}
-function createMediaRequest({ streamer_id, stream_id, user_id, username, input, canonical_url, embed_url, provider, title, thumbnail_url, duration_seconds, cost, queue_position, currency }) {
+function createMediaRequest({ streamer_id, stream_id, user_id, username, input, canonical_url, embed_url, provider, title, thumbnail_url, duration_seconds, cost, queue_position, currency, status, charge_state }) {
return run(
return run(
`INSERT INTO media_requests (
`INSERT INTO media_requests (
streamer_id, stream_id, user_id, username, input, canonical_url, embed_url,
streamer_id, stream_id, user_id, username, input, canonical_url, embed_url,
- provider, title, thumbnail_url, duration_seconds, cost, queue_position, currency
- ) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)`,
+ provider, title, thumbnail_url, duration_seconds, cost, queue_position, currency, status, charge_state
+ ) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)`,
[
[
streamer_id,
streamer_id,
stream_id || null,
stream_id || null,
@@ -6225,10 +6270,17 @@ function createMediaRequest({ streamer_id, stream_id, user_id, username, input,
@@ -6225,10 +6270,17 @@ function createMediaRequest({ streamer_id, stream_id, user_id, username, input,
cost,
cost,
queue_position ?? 0,
queue_position ?? 0,
currency || 'opencoins',
currency || 'opencoins',
+ status || 'pending',
+ charge_state || null,
]
]
);
);
}
}
+/** Drop a paid request whose charge was refused (nothing was taken); only while it is still 'charging'. */
+function removeUnchargedMediaRequest(id) {
+ return run("DELETE FROM media_requests WHERE id = ? AND charge_state = 'charging'", [id]);
+}
+
function getMediaRequestById(id) {
function getMediaRequestById(id) {
return get('SELECT * FROM media_requests WHERE id = ?', [id]);
return get('SELECT * FROM media_requests WHERE id = ?', [id]);
}
}
@@ -6250,7 +6302,7 @@ function getPendingMediaRequestsByStreamer(streamerId, limit = 50) {
@@ -6250,7 +6302,7 @@ function getPendingMediaRequestsByStreamer(streamerId, limit = 50) {
}
}
function getRecentMediaRequestsByStreamer(streamerId, limit = 15) {
function getRecentMediaRequestsByStreamer(streamerId, limit = 15) {
- return all(`SELECT * FROM media_requests WHERE streamer_id = ? AND status IN ('played', 'skipped', 'removed', 'failed') ORDER BY COALESCE(ended_at, requested_at) DESC, id DESC LIMIT ?`, [streamerId, limit]);
+ return all(`SELECT * FROM media_requests WHERE streamer_id = ? AND status IN ('played', 'skipped', 'removed', 'failed') AND charge_state IS NOT 'charging' ORDER BY COALESCE(ended_at, requested_at) DESC, id DESC LIMIT ?`, [streamerId, limit]);
}
}
function countPendingMediaRequestsForUser(streamerId, userId) {
function countPendingMediaRequestsForUser(streamerId, userId) {
@@ -7560,7 +7612,7 @@ module.exports = {
@@ -7560,7 +7612,7 @@ module.exports = {
// Profiles
// Profiles
getUserProfile, updateUserAvatar,
getUserProfile, updateUserAvatar,
getKickChannelCache, setKickChannelCache,
getKickChannelCache, setKickChannelCache,
- getChannelPoints, addChannelPoints, deductChannelPoints,
+ getChannelPoints, addChannelPoints, deductChannelPoints, applyChannelPoints,
// Follows
// Follows
followUser, unfollowUser, getFollowerCount, isFollowing, getFollowerIds,
followUser, unfollowUser, getFollowerCount, isFollowing, getFollowerIds,
// Transactions (Vibes)
// Transactions (Vibes)
@@ -7575,7 +7627,7 @@ module.exports = {
@@ -7575,7 +7627,7 @@ module.exports = {
upsertWatchTime, getWatchTime, getTotalWatchTime,
upsertWatchTime, getWatchTime, getTotalWatchTime,
// Media Requests
// Media Requests
getMediaRequestSettingsByUserId, upsertMediaRequestSettings,
getMediaRequestSettingsByUserId, upsertMediaRequestSettings,
- createMediaRequest, getMediaRequestById, getMediaRequestByStreamerAndId,
+ createMediaRequest, removeUnchargedMediaRequest, getMediaRequestById, getMediaRequestByStreamerAndId,
getActiveMediaRequestByStreamer, getNextPendingMediaRequest,
getActiveMediaRequestByStreamer, getNextPendingMediaRequest,
getPendingMediaRequestsByStreamer, getRecentMediaRequestsByStreamer,
getPendingMediaRequestsByStreamer, getRecentMediaRequestsByStreamer,
countPendingMediaRequestsForUser, getMediaRequestMaxQueuePosition,
countPendingMediaRequestsForUser, getMediaRequestMaxQueuePosition,
@@ -1539,6 +1539,11 @@ async function start() {
@@ -1539,6 +1539,11 @@ async function start() {
if (out.deleted) console.log(`[Analytics] pruned ${out.deleted} raw events older than ${out.cutoff}`);
if (out.deleted) console.log(`[Analytics] pruned ${out.deleted} raw events older than ${out.cutoff}`);
}, { initialDelayMs: 5 * 60 * 1000, jitterMs: 60 * 1000 });
}, { initialDelayMs: 5 * 60 * 1000, jitterMs: 60 * 1000 });
+ // 8b3. Media requests whose OpenCoins charge never got an answer: charged again with the same
+ // key (a replay if it landed) and refunded, so no viewer pays for a request that failed.
+ require('./utils/jobs').every('media-charge-reconcile', 5 * 60 * 1000, () => require('./media/media-queue').reconcileCharges(),
+ { initialDelayMs: 2 * 60 * 1000, jitterMs: 30 * 1000 });
+
// 8c. Durable events (roadmap Wave 3): stream lifecycle goes to OpenVibe.Events through the
// 8c. Durable events (roadmap Wave 3): stream lifecycle goes to OpenVibe.Events through the
// transactional outbox (server/events/stream-events.js). Off unless EVENTS_URL is set.
// transactional outbox (server/events/stream-events.js). Off unless EVENTS_URL is set.
try { require('./events/stream-events').init(); } catch (err) { console.warn('[Events] not started:', err.message); }
try { require('./events/stream-events').init(); } catch (err) { console.warn('[Events] not started:', err.message); }
@@ -122,10 +122,16 @@ class MediaQueue {
@@ -122,10 +122,16 @@ class MediaQueue {
* Take `cost` from the viewer in the channel's currency. Throws a message intended to
* Take `cost` from the viewer in the channel's currency. Throws a message intended to
* be shown to the viewer verbatim. The viewer was quoted this exact figure by /quote
* be shown to the viewer verbatim. The viewer was quoted this exact figure by /quote
* before submitting, so what they agreed to is what gets taken.
* before submitting, so what they agreed to is what gets taken.
+ *
+ * Every charge is keyed by the media request it pays for (ADR-012 rule 5): OpenCoins and
+ * channel points `live:media_req:<request id>`, Vibes on Billing `live:media_charge:<request id>`.
+ * Charging the same request again (a retry, the reconciler) replays; it never takes twice.
+ * An error with `outcomeUnknown` means the wallet did not answer: it may or may not have landed.
*/
*/
- async charge({ currency, cost, userId, streamerId, streamId, label }) {
+ async charge({ currency, cost, userId, streamerId, streamId, label, requestId }) {
+ if (!requestId) throw new Error('charge() needs the media request id (its idempotency key)');
if (currency === 'points') {
if (currency === 'points') {
- if (!db.deductChannelPoints(userId, streamerId, cost)) {
+ if (!db.deductChannelPoints(userId, streamerId, cost, `live:media_req:${requestId}`, label)) {
const have = db.getChannelPoints(userId, streamerId);
const have = db.getChannelPoints(userId, streamerId);
throw new Error(`Not enough channel points — this costs ${cost}, you have ${have}.`);
throw new Error(`Not enough channel points — this costs ${cost}, you have ${have}.`);
}
}
@@ -139,7 +145,7 @@ class MediaQueue {
@@ -139,7 +145,7 @@ class MediaQueue {
// streamer's payable); the returned charge is linked to the request for refunds.
// streamer's payable); the returned charge is linked to the request for refunds.
const billingActions = require('../monetization/billing-actions');
const billingActions = require('../monetization/billing-actions');
try {
try {
- return await billingActions.chargeMedia({ userId, streamerId, streamId, cost, label });
+ return await billingActions.chargeMedia({ userId, streamerId, streamId, cost, label, requestId });
} catch (e) {
} catch (e) {
const live = billingActions.toLive(e, { insufficient: `Not enough Vibes — this costs ${cost}.` });
const live = billingActions.toLive(e, { insufficient: `Not enough Vibes — this costs ${cost}.` });
throw new Error(live ? live.body.error : (e?.message || `Could not charge ${cost} Vibes.`));
throw new Error(live ? live.body.error : (e?.message || `Could not charge ${cost} Vibes.`));
@@ -157,17 +163,65 @@ class MediaQueue {
@@ -157,17 +163,65 @@ class MediaQueue {
return;
return;
}
}
// opencoins — network wallet
// opencoins — network wallet
+ const wallet = require('../monetization/wallet-client');
+ const key = `live:media_req:${requestId}`;
+ const noAnswer = (e) => !e || !e.status || e.status >= 500;
+ let debited;
try {
try {
- const wallet = require('../monetization/wallet-client');
- const debited = await wallet.debit(
- userId, cost, label,
- `live:media_req:${streamerId}:${userId}:${Date.now()}:${Math.random().toString(36).slice(2, 8)}`
- );
- if (!debited) throw new Error('OpenCoins wallet unavailable — link your OpenVibe.Network account first.');
+ debited = await wallet.debit(userId, cost, label, key);
} catch (e) {
} catch (e) {
if (e && e.status === 409) throw new Error(`Not enough OpenCoins — this costs ${cost}.`);
if (e && e.status === 409) throw new Error(`Not enough OpenCoins — this costs ${cost}.`);
- throw e;
+ if (!noAnswer(e)) throw e;
+ // No answer (unreachable, 5xx): whether it landed is unknown. The same key makes the
+ // retry safe, so ask once more; reconcileCharges() settles what is still unknown after.
+ try {
+ debited = await wallet.debit(userId, cost, label, key);
+ } catch (e2) {
+ if (e2 && e2.status === 409) throw new Error(`Not enough OpenCoins — this costs ${cost}.`);
+ if (!noAnswer(e2)) throw e2;
+ const err = new Error('Could not reach the OpenCoins wallet — nothing was queued. If you were charged, it is refunded automatically.');
+ err.outcomeUnknown = true;
+ throw err;
+ }
+ }
+ if (!debited) throw new Error('OpenCoins wallet unavailable — link your OpenVibe.Network account first.');
+ }
+
+ /**
+ * Settle requests whose OpenCoins charge never got an answer (charge_state 'unknown', or a
+ * 'charging' row left by a crash). The viewer was told it failed, so the request stays failed
+ * and the coins must end where they started: charge again with the same key (a replay if the
+ * first one landed, a new debit if it did not), then refund it (`live:media_refund:<id>`, also
+ * keyed). A 409 on the re-charge means nothing was ever taken. Anything still unanswered waits
+ * for the next run.
+ */
+ async reconcileCharges({ olderThanMs = 60_000, limit = 50 } = {}) {
+ const cutoff = new Date(Date.now() - olderThanMs).toISOString().replace('T', ' ').slice(0, 19);
+ const rows = db.all(`SELECT * FROM media_requests WHERE charge_state IN ('unknown', 'charging') AND currency = 'opencoins'
+ AND requested_at <= ? ORDER BY id LIMIT ?`, [cutoff, limit]);
+ const wallet = require('../monetization/wallet-client');
+ let settled = 0;
+ for (const r of rows) {
+ const label = `Media request: ${r.title || r.input}`;
+ try {
+ let charged = true;
+ try {
+ const out = await wallet.debit(r.user_id, r.cost, label, `live:media_req:${r.id}`);
+ if (!out) charged = false; // unlinked: nothing can have been taken
+ } catch (e) {
+ if (e && e.status === 409) charged = false;
+ else throw e;
+ }
+ if (charged) {
+ await wallet.credit(r.user_id, r.cost, `Refund: ${r.title || 'media request'}`, `live:media_refund:${r.id}`);
+ }
+ db.updateMediaRequest(r.id, { charge_state: null, refunded: charged ? 1 : 0, status: 'failed', last_error: r.last_error || 'The OpenCoins charge could not be confirmed' });
+ settled++;
+ } catch (e) {
+ console.warn(`[MediaQueue] charge of request ${r.id} still unsettled: ${e.message}`);
+ }
}
}
+ return settled;
}
}
/**
/**
@@ -214,15 +268,10 @@ class MediaQueue {
@@ -214,15 +268,10 @@ class MediaQueue {
if (unpriceable) throw new Error(unpriceable);
if (unpriceable) throw new Error(unpriceable);
// A free channel stores cost 0, so the queue does not display a price nobody paid.
// A free channel stores cost 0, so the queue does not display a price nobody paid.
const cost = currency === 'free' ? 0 : this.calculateCost(settings, normalized.duration_seconds);
const cost = currency === 'free' ? 0 : this.calculateCost(settings, normalized.duration_seconds);
- let billingCharge = null;
- if (cost > 0) {
- billingCharge = await this.charge({
- currency, cost, userId, streamerId, streamId,
- label: `Media request: ${normalized.title || trimmed}`,
- });
- }
- const queuePosition = db.getMediaRequestMaxQueuePosition(streamerId) + 1;
+ // A paid request is written first, out of the queue (status 'failed', charge_state
+ // 'charging'), so its charge is keyed by its id; it joins the queue once paid.
+ const paid = cost > 0;
const result = db.createMediaRequest({
const result = db.createMediaRequest({
streamer_id: streamerId,
streamer_id: streamerId,
stream_id: streamId,
stream_id: streamId,
@@ -237,8 +286,34 @@ class MediaQueue {
@@ -237,8 +286,34 @@ class MediaQueue {
duration_seconds: normalized.duration_seconds,
duration_seconds: normalized.duration_seconds,
cost,
cost,
currency,
currency,
- queue_position: queuePosition,
+ queue_position: paid ? 0 : db.getMediaRequestMaxQueuePosition(streamerId) + 1,
+ status: paid ? 'failed' : 'pending',
+ charge_state: paid ? 'charging' : null,
});
});
+ const requestId = Number(result.lastInsertRowid);
+ let billingCharge = null;
+ if (paid) {
+ try {
+ billingCharge = await this.charge({
+ currency, cost, userId, streamerId, streamId, requestId,
+ label: `Media request: ${normalized.title || trimmed}`,
+ });
+ } catch (e) {
+ if (e && e.outcomeUnknown) {
+ // Kept for reconcileCharges(), which refunds it if the debit did land.
+ db.updateMediaRequest(requestId, { charge_state: 'unknown', last_error: 'The OpenCoins charge could not be confirmed' });
+ } else {
+ // Refused (not enough, unlinked, Billing said no): nothing was taken.
+ db.removeUnchargedMediaRequest(requestId);
+ }
+ throw e;
+ }
+ // Paid: join the queue at the back (computed now, after the await).
+ db.updateMediaRequest(requestId, {
+ status: 'pending', charge_state: null,
+ queue_position: db.getMediaRequestMaxQueuePosition(streamerId) + 1,
+ });
+ }
// The coin ledger tracks OpenCoins only — booking a Vibes or channel-points charge
// The coin ledger tracks OpenCoins only — booking a Vibes or channel-points charge
// here would show up as a phantom coin spend in the viewer's history.
// here would show up as a phantom coin spend in the viewer's history.
@@ -252,7 +327,7 @@ class MediaQueue {
@@ -252,7 +327,7 @@ class MediaQueue {
});
});
}
}
- const request = db.getMediaRequestById(result.lastInsertRowid);
+ const request = db.getMediaRequestById(requestId);
if (billingCharge && billingCharge.actionId) require('../monetization/billing-actions').linkMediaCharge(billingCharge.actionId, request.id);
if (billingCharge && billingCharge.actionId) require('../monetization/billing-actions').linkMediaCharge(billingCharge.actionId, request.id);
this.broadcastQueueUpdate(streamerId);
this.broadcastQueueUpdate(streamerId);
@@ -500,7 +575,7 @@ class MediaQueue {
@@ -500,7 +575,7 @@ class MediaQueue {
const label = `Refund: ${request.title || 'media request'}`;
const label = `Refund: ${request.title || 'media request'}`;
try {
try {
if (currency === 'points') {
if (currency === 'points') {
- db.addChannelPoints(request.user_id, request.streamer_id, amount);
+ db.addChannelPoints(request.user_id, request.streamer_id, amount, `live:media_refund:${request.id}`, label);
} else if (currency === 'vibes') {
} else if (currency === 'vibes') {
const money = require('../monetization/money-authority');
const money = require('../monetization/money-authority');
if (money.writeRefusal()) {
if (money.writeRefusal()) {
@@ -177,12 +177,17 @@ async function donate(req, { toUserId, streamId, amount, message, goalId = null
@@ -177,12 +177,17 @@ async function donate(req, { toUserId, streamId, amount, message, goalId = null
};
};
}
}
-/** Media request paid in Vibes: a paid interaction to the streamer. Returns { actionId, transactionId }. */
-async function chargeMedia({ userId, streamerId, streamId, cost, label }) {
+/**
+ * Media request paid in Vibes: a paid interaction to the streamer. Returns { actionId, transactionId }.
+ * Keyed by the request it pays for (`live:media_charge:<request id>`, ADR-012 rule 5), and linked
+ * to it from the start, so a retried charge is replayed rather than taken twice.
+ */
+async function chargeMedia({ userId, streamerId, streamId, cost, label, requestId }) {
+ if (!requestId) throw new Error('chargeMedia needs the media request id (its idempotency key)');
const from = await subjectFor(userId);
const from = await subjectFor(userId);
const to = await subjectFor(streamerId, { role: 'recipient' });
const to = await subjectFor(streamerId, { role: 'recipient' });
const body = { from: ref(from), to: ref(to), amount: cost, kind: 'paid_interaction', ...(streamTarget(streamId) ? { target: streamTarget(streamId) } : {}), message: String(label || 'Media request').slice(0, 500), on_behalf_of: ref(from) };
const body = { from: ref(from), to: ref(to), amount: cost, kind: 'paid_interaction', ...(streamTarget(streamId) ? { target: streamTarget(streamId) } : {}), message: String(label || 'Media request').slice(0, 500), on_behalf_of: ref(from) };
- const { out, actionId } = await perform({ action: 'media_charge', path: '/transfers', body, key: `live:media_charge:${crypto.randomUUID()}`, liveUserId: userId });
+ const { out, actionId } = await perform({ action: 'media_charge', path: '/transfers', body, key: `live:media_charge:${requestId}`, liveUserId: userId, liveRef: `media_request:${requestId}` });
return { actionId, transactionId: out.transaction ? out.transaction.id : null };
return { actionId, transactionId: out.transaction ? out.transaction.id : null };
}
}
@@ -287,15 +287,21 @@ router.post('/redemptions/:id', requireAuth, (req, res) => {
@@ -287,15 +287,21 @@ router.post('/redemptions/:id', requireAuth, (req, res) => {
const s = db.getStreamById(redemption.stream_id);
const s = db.getStreamById(redemption.stream_id);
if (s?.user_id) refundStreamerId = s.user_id;
if (s?.user_id) refundStreamerId = s.user_id;
}
}
- db.addChannelPoints(redemption.user_id, refundStreamerId, reward.cost);
- db.createCoinTransaction({
- user_id: redemption.user_id,
- stream_id: redemption.stream_id || null,
- amount: reward.cost,
- type: 'refund',
- reward_id: redemption.reward_id,
- message: `Refunded: ${reward.title} (rejected by streamer)`,
+ // Keyed by the redemption: rejecting it twice refunds once.
+ const refunded = db.applyChannelPoints({
+ userId: redemption.user_id, streamerId: refundStreamerId, delta: reward.cost,
+ key: `live:cp:redeem_refund:${redemption.id}`, reason: `refund redemption ${redemption.id}`,
});
});
+ if (refunded.applied) {
+ db.createCoinTransaction({
+ user_id: redemption.user_id,
+ stream_id: redemption.stream_id || null,
+ amount: reward.cost,
+ type: 'refund',
+ reward_id: redemption.reward_id,
+ message: `Refunded: ${reward.title} (rejected by streamer)`,
+ });
+ }
}
}
}
}
@@ -312,7 +318,12 @@ router.post('/admin/grant', requireOwner, async (req, res) => {
@@ -312,7 +318,12 @@ router.post('/admin/grant', requireOwner, async (req, res) => {
const { userId, amount, reason } = req.body;
const { userId, amount, reason } = req.body;
if (!userId || !amount) return res.status(400).json({ error: 'userId and amount required' });
if (!userId || !amount) return res.status(400).json({ error: 'userId and amount required' });
- const newBalance = await openvibeCoins.adminGrant(userId, amount, reason);
+ // A client Idempotency-Key makes a repeated submit the same grant (see adminGrant).
+ const clientKey = req.get('Idempotency-Key');
+ const newBalance = await openvibeCoins.adminGrant(userId, amount, reason, {
+ adminId: req.user.id,
+ clientKey: clientKey && /^[A-Za-z0-9._-]{8,100}$/.test(clientKey) ? clientKey : null,
+ });
res.json({ message: `Granted ${amount} OpenCoins`, balance: newBalance });
res.json({ message: `Granted ${amount} OpenCoins`, balance: newBalance });
} catch (err) {
} catch (err) {
res.status(400).json({ error: err.message });
res.status(400).json({ error: err.message });
@@ -12,6 +12,11 @@
@@ -12,6 +12,11 @@
* (see wallet-client.js); the legacy users.openvibe_coins_balance column is
* (see wallet-client.js); the legacy users.openvibe_coins_balance column is
* frozen for the migration script and is never written anymore.
* frozen for the migration script and is never written anymore.
*
*
+ * Every channel-points debit and credit carries a deterministic per-event key (ADR-012 rule 5,
+ * db.applyChannelPoints): watch = the watch_time row + minute, chat = the user's minute, follow =
+ * user + streamer, bonus = the claim window, redeem / redeem refund = the redemption id. A retried
+ * or replayed event moves nothing twice.
+ *
* Channel-point earning rates:
* Channel-point earning rates:
* - Watching a live stream: per the streamer's config (default 10 / 5 min)
* - Watching a live stream: per the streamer's config (default 10 / 5 min)
* - Sending a chat message: 5 points (max 1 per minute)
* - Sending a chat message: 5 points (max 1 per minute)
@@ -74,7 +79,9 @@ class OpenCoins {
@@ -74,7 +79,9 @@ class OpenCoins {
coins *= COINS.STREAK_MULTIPLIER;
coins *= COINS.STREAK_MULTIPLIER;
}
}
- const total = db.addChannelPoints(userId, streamerId, coins);
+ const r = db.applyChannelPoints({ userId, streamerId, delta: coins, key: `live:cp:watch:${wt.id}:${wt.minutes_watched}`, reason: 'watch' });
+ if (!r.applied) return null;
+ const total = r.balance;
_feedPowerchat(streamerId, userId, coins);
_feedPowerchat(streamerId, userId, coins);
db.createCoinTransaction({
db.createCoinTransaction({
user_id: userId,
user_id: userId,
@@ -110,7 +117,11 @@ class OpenCoins {
@@ -110,7 +117,11 @@ class OpenCoins {
if (!streamerId || streamerId === userId) return null;
if (!streamerId || streamerId === userId) return null;
chatCooldowns.set(userId, now);
chatCooldowns.set(userId, now);
- const total = db.addChannelPoints(userId, streamerId, COINS.CHAT_BONUS);
+ // The cooldown window is the event: after a restart (empty cooldown map) the same minute
+ // still earns once.
+ const r = db.applyChannelPoints({ userId, streamerId, delta: COINS.CHAT_BONUS, key: `live:cp:chat:${userId}:${Math.floor(now / COINS.CHAT_COOLDOWN_MS)}`, reason: 'chat' });
+ if (!r.applied) return null;
+ const total = r.balance;
_feedPowerchat(streamerId, userId, COINS.CHAT_BONUS);
_feedPowerchat(streamerId, userId, COINS.CHAT_BONUS);
db.createCoinTransaction({
db.createCoinTransaction({
user_id: userId,
user_id: userId,
@@ -139,7 +150,9 @@ class OpenCoins {
@@ -139,7 +150,9 @@ class OpenCoins {
if (existing) return null;
if (existing) return null;
if (!streamerId || streamerId === userId) return null;
if (!streamerId || streamerId === userId) return null;
- const total = db.addChannelPoints(userId, streamerId, COINS.FOLLOW_BONUS);
+ const r = db.applyChannelPoints({ userId, streamerId, delta: COINS.FOLLOW_BONUS, key: `live:cp:follow:${userId}:${streamerId}`, reason: 'follow' });
+ if (!r.applied) return null;
+ const total = r.balance;
_feedPowerchat(streamerId, userId, COINS.FOLLOW_BONUS);
_feedPowerchat(streamerId, userId, COINS.FOLLOW_BONUS);
db.createCoinTransaction({
db.createCoinTransaction({
user_id: userId,
user_id: userId,
@@ -175,46 +188,48 @@ class OpenCoins {
@@ -175,46 +188,48 @@ class OpenCoins {
}
}
if (!pointsStreamerId) throw new Error('No channel context for this reward');
if (!pointsStreamerId) throw new Error('No channel context for this reward');
- // Check user has enough points for this channel
- if (!db.deductChannelPoints(userId, pointsStreamerId, reward.cost)) {
- const cpName = (db.getChannelPointsConfig(pointsStreamerId).name) || 'Channel Points';
- throw new Error(`Not enough ${cpName}`);
- }
-
- // Check per-user cooldown
- if (reward.cooldown_seconds > 0) {
- const lastRedemption = db.get(
- `SELECT created_at FROM coin_redemptions WHERE reward_id = ? AND user_id = ? ORDER BY created_at DESC LIMIT 1`,
- [rewardId, userId]
- );
- if (lastRedemption) {
- const elapsed = (Date.now() - new Date(lastRedemption.created_at.replace(' ', 'T') + 'Z').getTime()) / 1000;
- if (elapsed < reward.cooldown_seconds) {
- db.addChannelPoints(userId, pointsStreamerId, reward.cost); // refund
- throw new Error(`Cooldown: wait ${Math.ceil(reward.cooldown_seconds - elapsed)}s`);
+ // The limits are checked, the redemption row created and the points taken in one
+ // transaction, keyed by the redemption id: nothing is taken for a refused redemption
+ // (there is no take-then-refund any more), and a replayed spend moves nothing twice.
+ const result = db.getDb().transaction(() => {
+ // Check per-user cooldown
+ if (reward.cooldown_seconds > 0) {
+ const lastRedemption = db.get(
+ `SELECT created_at FROM coin_redemptions WHERE reward_id = ? AND user_id = ? ORDER BY created_at DESC LIMIT 1`,
+ [rewardId, userId]
+ );
+ if (lastRedemption) {
+ const elapsed = (Date.now() - new Date(lastRedemption.created_at.replace(' ', 'T') + 'Z').getTime()) / 1000;
+ if (elapsed < reward.cooldown_seconds) {
+ throw new Error(`Cooldown: wait ${Math.ceil(reward.cooldown_seconds - elapsed)}s`);
+ }
}
}
}
}
- }
- // Check max per stream
- if (reward.max_per_stream > 0 && streamId) {
- const count = db.get(
- `SELECT COUNT(*) as c FROM coin_redemptions WHERE reward_id = ? AND stream_id = ?`,
- [rewardId, streamId]
- );
- if (count && count.c >= reward.max_per_stream) {
- db.addChannelPoints(userId, pointsStreamerId, reward.cost); // refund
- throw new Error('Max redemptions reached for this stream');
+ // Check max per stream
+ if (reward.max_per_stream > 0 && streamId) {
+ const count = db.get(
+ `SELECT COUNT(*) as c FROM coin_redemptions WHERE reward_id = ? AND stream_id = ?`,
+ [rewardId, streamId]
+ );
+ if (count && count.c >= reward.max_per_stream) {
+ throw new Error('Max redemptions reached for this stream');
+ }
}
}
- }
- // Create redemption
- const result = db.createCoinRedemption({
- reward_id: rewardId,
- user_id: userId,
- stream_id: streamId,
- user_input: userInput,
- });
+ // Create redemption, then take the points for it (rolled back together if short)
+ const created = db.createCoinRedemption({
+ reward_id: rewardId,
+ user_id: userId,
+ stream_id: streamId,
+ user_input: userInput,
+ });
+ if (!db.deductChannelPoints(userId, pointsStreamerId, reward.cost, `live:cp:redeem:${created.lastInsertRowid}`, `redeem reward ${rewardId}`)) {
+ const cpName = (db.getChannelPointsConfig(pointsStreamerId).name) || 'Channel Points';
+ throw new Error(`Not enough ${cpName}`);
+ }
+ return created;
+ })();
// Log transaction
// Log transaction
db.createCoinTransaction({
db.createCoinTransaction({
@@ -256,7 +271,10 @@ class OpenCoins {
@@ -256,7 +271,10 @@ class OpenCoins {
if (now - (bonusClaims.get(key) || 0) < windowMs) return null;
if (now - (bonusClaims.get(key) || 0) < windowMs) return null;
bonusClaims.set(key, now);
bonusClaims.set(key, now);
const amount = Math.max(1, (cfg.watch_amount || 10) * 3);
const amount = Math.max(1, (cfg.watch_amount || 10) * 3);
- const total = db.addChannelPoints(userId, streamerId, amount);
+ // One claim per window, also across a restart (the throttle map above is in memory).
+ const r = db.applyChannelPoints({ userId, streamerId, delta: amount, key: `live:cp:bonus:${userId}:${streamerId}:${Math.floor(now / windowMs)}`, reason: 'bonus game' });
+ if (!r.applied) return null;
+ const total = r.balance;
_feedPowerchat(streamerId, userId, amount);
_feedPowerchat(streamerId, userId, amount);
db.createCoinTransaction({ user_id: userId, stream_id: streamId, amount, type: 'watch', message: 'Bonus game' });
db.createCoinTransaction({ user_id: userId, stream_id: streamId, amount, type: 'watch', message: 'Bonus game' });
return { coins: amount, total, streamerId };
return { coins: amount, total, streamerId };
@@ -301,11 +319,37 @@ class OpenCoins {
@@ -301,11 +319,37 @@ class OpenCoins {
/**
/**
* Admin: grant OpenCoins to a user (network wallet credit).
* Admin: grant OpenCoins to a user (network wallet credit).
+ *
+ * The grant is recorded locally first and the wallet credit is keyed by that record
+ * (`live:admin_grant:<grant id>`, ADR-012 rule 5), so retrying a grant whose answer was lost
+ * never credits twice. With `clientKey` (the request's Idempotency-Key) a repeated submit is
+ * the same grant: it reuses the record and its key instead of making a second one.
*/
*/
- async adminGrant(userId, amount, reason) {
- const key = `live:admin_grant:${userId}:${Date.now()}:${Math.random().toString(36).slice(2, 8)}`;
- const result = await wallet.credit(userId, amount, reason || 'Admin grant', key);
+ async adminGrant(userId, amount, reason, { adminId = null, clientKey = null } = {}) {
+ const d = db.getDb();
+ d.exec(`CREATE TABLE IF NOT EXISTS opencoin_admin_grants (
+ id INTEGER PRIMARY KEY AUTOINCREMENT,
+ client_key TEXT UNIQUE,
+ admin_id INTEGER,
+ user_id INTEGER NOT NULL,
+ amount INTEGER NOT NULL,
+ reason TEXT,
+ balance INTEGER,
+ created_at DATETIME DEFAULT CURRENT_TIMESTAMP
+ )`);
+ const ck = clientKey ? `a${adminId || 0}:${clientKey}` : null;
+ let grant = ck ? d.prepare('SELECT * FROM opencoin_admin_grants WHERE client_key = ?').get(ck) : null;
+ if (grant && (grant.user_id !== Number(userId) || grant.amount !== Number(amount))) {
+ throw new Error('That Idempotency-Key was already used for a different grant');
+ }
+ if (!grant) {
+ const id = d.prepare('INSERT INTO opencoin_admin_grants (client_key, admin_id, user_id, amount, reason) VALUES (?, ?, ?, ?, ?)')
+ .run(ck, adminId, Number(userId), Number(amount), reason || null).lastInsertRowid;
+ grant = { id };
+ }
+ const result = await wallet.credit(userId, amount, reason || 'Admin grant', `live:admin_grant:${grant.id}`);
if (!result) throw new Error('User has no linked OpenVibe.Network account (wallet unavailable)');
if (!result) throw new Error('User has no linked OpenVibe.Network account (wallet unavailable)');
+ d.prepare('UPDATE opencoin_admin_grants SET balance = ? WHERE id = ?').run(result.balance ?? null, grant.id);
return result.balance;
return result.balance;
}
}
@@ -298,12 +298,18 @@ async function check(name, fn) {
@@ -298,12 +298,18 @@ async function check(name, fn) {
db.setSetting('payments_enabled', 'false');
db.setSetting('payments_enabled', 'false');
});
});
- await check('a Vibes media request is a paid interaction; its refund reverses that exact transfer', async () => {
+ await check('a Vibes media request is a paid interaction keyed by its request; its refund reverses that exact transfer', async () => {
const mediaQueue = require('../server/media/media-queue');
const mediaQueue = require('../server/media/media-queue');
- const charge = await mediaQueue.charge({ currency: 'vibes', cost: 40, userId: 2, streamerId: 3, streamId: streamB, label: 'Media request: song' });
- const c = lastCall();
- assert.strictEqual(c.path, '/transfers'); assert.strictEqual(c.body.kind, 'paid_interaction'); assert.match(c.key, /^live:media_charge:/);
const reqId = Number(db.createMediaRequest({ streamer_id: 3, stream_id: streamB, user_id: 2, username: 'ann', input: 'x', canonical_url: 'https://x.test/a', embed_url: null, provider: 'youtube', title: 'song', thumbnail_url: null, duration_seconds: 60, cost: 40, queue_position: 1, currency: 'vibes' }).lastInsertRowid);
const reqId = Number(db.createMediaRequest({ streamer_id: 3, stream_id: streamB, user_id: 2, username: 'ann', input: 'x', canonical_url: 'https://x.test/a', embed_url: null, provider: 'youtube', title: 'song', thumbnail_url: null, duration_seconds: 60, cost: 40, queue_position: 1, currency: 'vibes' }).lastInsertRowid);
+ const charge = await mediaQueue.charge({ currency: 'vibes', cost: 40, userId: 2, streamerId: 3, streamId: streamB, label: 'Media request: song', requestId: reqId });
+ const c = lastCall();
+ assert.strictEqual(c.path, '/transfers'); assert.strictEqual(c.body.kind, 'paid_interaction');
+ assert.strictEqual(c.key, `live:media_charge:${reqId}`, 'the charge is keyed by the request it pays for (ADR-012 rule 5)');
+ // A retried charge for the same request is a replay: Billing is not asked to move money again.
+ const before = billing.calls.length;
+ const again = await mediaQueue.charge({ currency: 'vibes', cost: 40, userId: 2, streamerId: 3, streamId: streamB, label: 'Media request: song', requestId: reqId });
+ assert.strictEqual(billing.calls.length, before, 'a retried charge does not reach Billing twice');
+ assert.strictEqual(again.transactionId, charge.transactionId);
billingActions.linkMediaCharge(charge.actionId, reqId);
billingActions.linkMediaCharge(charge.actionId, reqId);
const refunded = await mediaQueue.refund(reqId);
const refunded = await mediaQueue.refund(reqId);
assert.strictEqual(refunded, 40);
assert.strictEqual(refunded, 40);
@@ -312,7 +318,7 @@ async function check(name, fn) {
@@ -312,7 +318,7 @@ async function check(name, fn) {
assert.strictEqual(r.key, `live:media_refund:${reqId}`);
assert.strictEqual(r.key, `live:media_refund:${reqId}`);
assert.strictEqual(db.getMediaRequestById(reqId).refunded, 1);
assert.strictEqual(db.getMediaRequestById(reqId).refunded, 1);
assert.strictEqual(await mediaQueue.refund(reqId), 0, 'a second refund is a no-op');
assert.strictEqual(await mediaQueue.refund(reqId), 0, 'a second refund is a no-op');
- await assert.rejects(mediaQueue.charge({ currency: 'vibes', cost: 999999, userId: 2, streamerId: 3, streamId: streamB, label: 'x' }), /Not enough Vibes — this costs 999999/);
+ await assert.rejects(mediaQueue.charge({ currency: 'vibes', cost: 999999, userId: 2, streamerId: 3, streamId: streamB, label: 'x', requestId: reqId + 1000 }), /Not enough Vibes — this costs 999999/);
});
});
await check('webhooks moved: PowerChat and card webhooks on Live answer 410 with Billing\'s URL', async () => {
await check('webhooks moved: PowerChat and card webhooks on Live answer 410 with Billing\'s URL', async () => {
new file mode 100644
new file mode 100644
@@ -0,0 +1,242 @@
@@ -0,0 +1,242 @@
+/**
+ * Loyalty ledgers use deterministic per-event idempotency keys (ADR-012 rule 5), so a retry never
+ * spends or credits twice.
+ *
+ * - Channel points: every debit and credit goes through db.applyChannelPoints with a key, logged
+ * in channel_points_log; a repeated key moves nothing, a reused key for another event is refused,
+ * a refused debit leaves no row. Earn keys: watch row + minute, chat minute, follow, bonus window;
+ * a redemption is taken under its redemption id (limits first, no take-then-refund), and a
+ * streamer rejecting it twice refunds once.
+ * - Media requests: the request row is written first and the charge is keyed by its id
+ * (live:media_req:<id> for OpenCoins and points; Billing's key is tested in billing-authority).
+ * A wallet that never answers leaves the request 'unknown'; reconcileCharges() re-charges with
+ * the same key (a replay) and refunds, so the viewer ends where they started.
+ * - Admin OpenCoins grants: keyed by a local grant record; an Idempotency-Key repeat is the same grant.
+ *
+ * node test/loyalty-keys.test.js
+ */
+'use strict';
+const assert = require('assert');
+const fs = require('fs');
+const os = require('os');
+const path = require('path');
+const http = require('http');
+
+const tmp = fs.mkdtempSync(path.join(os.tmpdir(), 'ov-loyalty-'));
+process.env.DB_PATH = path.join(tmp, 'live.db');
+process.env.NODE_ENV = 'test';
+const quiet = console.log;
+console.log = () => {};
+console.warn = () => {};
+
+const db = require('../server/db/database');
+db.initDb();
+const raw = db.getDb();
+const auth = require('../server/auth/auth');
+const signIn = (req) => {
+ const id = Number(req.headers['x-test-user'] || 0);
+ const u = id ? db.getUserById(id) : null;
+ if (u) req.user = u;
+ return u;
+};
+auth.requireAuth = (req, res, next) => (signIn(req) ? next() : res.status(401).json({ error: 'Authentication required' }));
+
+const addUser = (id, name, role = 'streamer', owner = 0) => raw.prepare(`INSERT INTO users (id, username, display_name, email, password_hash, role, is_owner) VALUES (?, ?, ?, ?, 'x', ?, ?)`).run(id, name, name, `${name}@x`, role, owner);
+addUser(1, 'owner', 'admin', 1);
+addUser(2, 'viewer');
+addUser(3, 'streamer');
+db.ensureChannel(3);
+const chan = db.getChannelByUserId(3);
+const streamId = Number(db.createStream({ user_id: 3, channel_id: chan.id, title: 'live', protocol: 'webrtc' }).lastInsertRowid);
+const cp = () => db.getChannelPoints(2, 3);
+const logRows = (key) => raw.prepare('SELECT * FROM channel_points_log WHERE idempotency_key = ?').all(key);
+
+// A Network wallet that behaves like the real one: one effect per idempotency key.
+const wallet = require('../server/monetization/wallet-client');
+const ledger = { balance: 1000, keys: new Map(), calls: [], failNext: 0, landBeforeFailing: false };
+function walletOp(kind) {
+ return async (userId, amount, reason, key) => {
+ ledger.calls.push({ kind, key, amount });
+ if (ledger.failNext > 0) {
+ ledger.failNext--;
+ if (ledger.landBeforeFailing && !ledger.keys.has(key)) { ledger.keys.set(key, { kind, amount }); ledger.balance += kind === 'debit' ? -amount : amount; }
+ const e = new Error('Network wallet unreachable: timeout'); e.status = 0; throw e;
+ }
+ if (ledger.keys.has(key)) return { balance: ledger.balance, replayed: true };
+ if (kind === 'debit' && ledger.balance < amount) { const e = new Error('insufficient_funds'); e.status = 409; throw e; }
+ ledger.keys.set(key, { kind, amount });
+ ledger.balance += kind === 'debit' ? -amount : amount;
+ return { balance: ledger.balance };
+ };
+}
+wallet.debit = walletOp('debit');
+wallet.credit = walletOp('credit');
+
+(async () => {
+ // ── 1. The keyed channel-points log ──────────────────────────────────────────────
+ assert.throws(() => db.addChannelPoints(2, 3, 10), /idempotency key is required/, 'no unkeyed credit');
+ assert.throws(() => db.deductChannelPoints(2, 3, 10), /idempotency key is required/, 'no unkeyed debit');
+ assert.strictEqual(db.addChannelPoints(2, 3, 100, 'live:cp:test:credit:1'), 100);
+ assert.strictEqual(db.addChannelPoints(2, 3, 100, 'live:cp:test:credit:1'), 100, 'a retried credit is applied once');
+ assert.strictEqual(db.deductChannelPoints(2, 3, 30, 'live:cp:test:debit:1'), true);
+ assert.strictEqual(db.deductChannelPoints(2, 3, 30, 'live:cp:test:debit:1'), true, 'a retried debit answers as taken…');
+ assert.strictEqual(cp(), 70, '…but takes once');
+ assert.throws(() => db.addChannelPoints(2, 3, 5, 'live:cp:test:credit:1'), /already used for a different event/, 'a key reused for another amount is refused');
+ assert.strictEqual(db.deductChannelPoints(2, 3, 500, 'live:cp:test:debit:big'), false, 'not enough: refused');
+ assert.strictEqual(logRows('live:cp:test:debit:big').length, 0, 'a refused debit leaves no log row');
+ db.addChannelPoints(2, 3, 500, 'live:cp:test:topup');
+ assert.strictEqual(db.deductChannelPoints(2, 3, 500, 'live:cp:test:debit:big'), true, 'so the same key can be tried again once the balance covers it');
+ assert.strictEqual(cp(), 70);
+
+ // ── 2. Earning: follow, chat (across a "restart"), watch ───────────────────────
+ let coins = require('../server/monetization/opencoins');
+ assert.ok(coins.awardFollow(2, 3));
+ raw.prepare("DELETE FROM coin_transactions WHERE type = 'follow_bonus'").run(); // even without the old LIKE check…
+ assert.strictEqual(coins.awardFollow(2, 3), null, '…a second follow bonus is refused by its key');
+ assert.strictEqual(logRows('live:cp:follow:2:3').length, 1);
+ const afterFollow = cp();
+ assert.ok(coins.awardChat(2, streamId));
+ delete require.cache[require.resolve('../server/monetization/opencoins')];
+ coins = require('../server/monetization/opencoins'); // a restart: the in-memory chat cooldown is gone
+ const minute = Math.floor(Date.now() / 60_000);
+ const secondChat = coins.awardChat(2, streamId);
+ if (Math.floor(Date.now() / 60_000) === minute) assert.strictEqual(secondChat, null, 'the same chat minute earns once, even after a restart');
+ assert.strictEqual(cp(), afterFollow + 5 + (secondChat ? 5 : 0));
+ raw.prepare('INSERT INTO watch_time (user_id, stream_id, minutes_watched) VALUES (2, ?, 4)').run(streamId);
+ const watched = coins.awardWatch(2, streamId); // minute 5 of the default 5-minute interval
+ assert.ok(watched && watched.coins === 10);
+ const wt = db.getWatchTime(2, streamId);
+ assert.strictEqual(logRows(`live:cp:watch:${wt.id}:5`).length, 1, 'the watch award is keyed by the watch row and minute');
+ assert.strictEqual(db.applyChannelPoints({ userId: 2, streamerId: 3, delta: 10, key: `live:cp:watch:${wt.id}:5` }).replayed, true, 'replaying that minute moves nothing');
+
+ // ── 3. Redemptions: limits before the spend, keyed by redemption, reject refunds once ──
+ const rewardId = Number(db.createCoinReward({ streamer_id: 3, title: 'Hydrate', cost: 40, cooldown_seconds: 3600 }).lastInsertRowid);
+ const beforeRedeem = cp();
+ const red = coins.redeem(2, rewardId, streamId, 'drink');
+ assert.strictEqual(cp(), beforeRedeem - 40);
+ assert.strictEqual(logRows(`live:cp:redeem:${red.redemption.id}`).length, 1, 'the spend is keyed by the redemption id');
+ const redemptions = raw.prepare('SELECT COUNT(*) AS n FROM coin_redemptions').get().n;
+ assert.throws(() => coins.redeem(2, rewardId, streamId, 'again'), /Cooldown/);
+ assert.strictEqual(cp(), beforeRedeem - 40, 'a refused redemption takes nothing (no take-then-refund)');
+ assert.strictEqual(raw.prepare('SELECT COUNT(*) AS n FROM coin_redemptions').get().n, redemptions, 'and leaves no redemption row');
+ const pricey = Number(db.createCoinReward({ streamer_id: 3, title: 'Big', cost: 1_000_000 }).lastInsertRowid);
+ assert.throws(() => coins.redeem(2, pricey, streamId, ''), /Not enough/);
+ assert.strictEqual(raw.prepare('SELECT COUNT(*) AS n FROM coin_redemptions').get().n, redemptions, 'not enough points: the redemption row is rolled back');
+
+ const express = require('express');
+ const app = express();
+ app.use(express.json());
+ app.use((req, res, next) => { signIn(req); next(); });
+ app.use('/api/coins', require('../server/monetization/coins-routes'));
+ const server = http.createServer(app).listen(0);
+ const call = (method, p, user, body, headers = {}) => new Promise((resolve, reject) => {
+ const req = http.request({ port: server.address().port, path: p, method, headers: { 'content-type': 'application/json', 'x-test-user': String(user), ...headers } }, (res) => {
+ let text = '';
+ res.on('data', (c) => { text += c; });
+ res.on('end', () => { let json = null; try { json = JSON.parse(text); } catch { /* */ } resolve({ status: res.statusCode, json }); });
+ });
+ req.on('error', reject);
+ req.end(body ? JSON.stringify(body) : undefined);
+ });
+ const beforeReject = cp();
+ assert.strictEqual((await call('POST', `/api/coins/redemptions/${red.redemption.id}`, 3, { status: 'rejected' })).status, 200);
+ assert.strictEqual((await call('POST', `/api/coins/redemptions/${red.redemption.id}`, 3, { status: 'rejected' })).status, 200);
+ assert.strictEqual(cp(), beforeReject + 40, 'rejecting the same redemption twice refunds once');
+ assert.strictEqual(logRows(`live:cp:redeem_refund:${red.redemption.id}`).length, 1);
+
+ // ── 4. Media requests ───────────────────────────────────────────────────────────
+ const mq = require('../server/media/media-queue');
+ let n = 0;
+ mq.normalizeInput = async () => { n++; return { canonical_url: `https://media.example/${n}.mp4`, embed_url: null, provider: 'video', title: `clip ${n}`, thumbnail_url: null, duration_seconds: 30, isLive: false }; };
+ mq.extractStreamUrlForRequest = async () => null;
+ mq.broadcastQueueUpdate = () => {};
+ db.upsertMediaRequestSettings(3, { enabled: 1, request_cost: 25, currency: 'opencoins', max_per_user: 10 });
+
+ const start = ledger.balance;
+ const req1 = await mq.addRequest({ streamerId: 3, streamId, userId: 2, username: 'viewer', input: 'https://media.example/a.mp4' });
+ assert.strictEqual(req1.status, 'pending');
+ assert.strictEqual(ledger.calls.at(-1).key, `live:media_req:${req1.id}`, 'the OpenCoins spend is keyed by the request id');
+ assert.strictEqual(ledger.balance, start - 25);
+ // A retry of that charge (same request) is a replay at the wallet: nothing more is taken.
+ await mq.charge({ currency: 'opencoins', cost: 25, userId: 2, streamerId: 3, streamId, label: 'retry', requestId: req1.id });
+ assert.strictEqual(ledger.balance, start - 25, 'retrying the charge does not spend twice');
+ assert.ok(db.getMediaRequestById(req1.id).queue_position >= 1, 'a paid request joins the queue');
+
+ // The wallet times out after the debit landed, twice: the viewer is told it failed and the
+ // request waits, out of the queue, for the reconciler.
+ ledger.failNext = 2; ledger.landBeforeFailing = true;
+ const queueBefore = db.all("SELECT id FROM media_requests WHERE status = 'pending'").length;
+ await assert.rejects(mq.addRequest({ streamerId: 3, streamId, userId: 2, username: 'viewer', input: 'https://media.example/b.mp4' }), /Could not reach the OpenCoins wallet/);
+ ledger.landBeforeFailing = false;
+ const unknown = raw.prepare("SELECT * FROM media_requests WHERE charge_state = 'unknown'").get();
+ assert.ok(unknown, 'the request is kept with its charge unknown');
+ assert.strictEqual(unknown.status, 'failed');
+ assert.strictEqual(db.all("SELECT id FROM media_requests WHERE status = 'pending'").length, queueBefore, 'and is not in the queue');
+ assert.strictEqual(ledger.balance, start - 50, 'the debit did land');
+ assert.strictEqual(await mq.reconcileCharges({ olderThanMs: -60_000 }), 1);
+ assert.strictEqual(ledger.balance, start - 25, 'the reconciler re-charged with the same key (a replay) and refunded');
+ assert.deepStrictEqual(ledger.calls.filter((c) => c.key === `live:media_req:${unknown.id}`).map((c) => c.kind), ['debit', 'debit', 'debit'], 'every attempt used the one key');
+ const settled = db.getMediaRequestById(unknown.id);
+ assert.strictEqual(settled.charge_state, null);
+ assert.strictEqual(settled.refunded, 1);
+ assert.strictEqual(await mq.reconcileCharges({ olderThanMs: -60_000 }), 0, 'nothing left to settle');
+ assert.strictEqual(mq.refund(unknown.id), 0, 'and it is not refunded a second time');
+
+ // A wallet that never answered and never took anything: the re-charge takes it, the refund gives it back.
+ ledger.failNext = 2;
+ await assert.rejects(mq.addRequest({ streamerId: 3, streamId, userId: 2, username: 'viewer', input: 'https://media.example/c.mp4' }), /Could not reach/);
+ await mq.reconcileCharges({ olderThanMs: -60_000 });
+ assert.strictEqual(ledger.balance, start - 25, 'net zero either way');
+
+ // Not enough OpenCoins: nothing taken, no row left behind.
+ const rows = raw.prepare('SELECT COUNT(*) AS n FROM media_requests').get().n;
+ ledger.balance = 10;
+ await assert.rejects(mq.addRequest({ streamerId: 3, streamId, userId: 2, username: 'viewer', input: 'https://media.example/d.mp4' }), /Not enough OpenCoins/);
+ assert.strictEqual(raw.prepare('SELECT COUNT(*) AS n FROM media_requests').get().n, rows, 'a refused charge leaves no request row');
+ assert.strictEqual(ledger.balance, 10);
+
+ // Channel points: spend keyed by the request, refund keyed by the request, once.
+ db.upsertMediaRequestSettings(3, { currency: 'points' });
+ const pts = cp();
+ const req2 = await mq.addRequest({ streamerId: 3, streamId, userId: 2, username: 'viewer', input: 'https://media.example/e.mp4' });
+ assert.strictEqual(cp(), pts - 25);
+ assert.strictEqual(logRows(`live:media_req:${req2.id}`).length, 1);
+ assert.strictEqual(db.deductChannelPoints(2, 3, 25, `live:media_req:${req2.id}`), true, 'a retried points charge is a replay');
+ assert.strictEqual(cp(), pts - 25);
+ assert.strictEqual(mq.refund(req2.id), 25);
+ db.updateMediaRequest(req2.id, { refunded: 0 }); // even if the refunded flag were lost…
+ mq.refund(req2.id);
+ assert.strictEqual(cp(), pts, '…the refund key credits once');
+
+ // ── 5. Admin OpenCoins grants ─────────────────────────────────────────────────
+ const g0 = ledger.balance;
+ let r = await call('POST', '/api/coins/admin/grant', 1, { userId: 2, amount: 50, reason: 'thanks' }, { 'Idempotency-Key': 'grant-abc-123' });
+ assert.strictEqual(r.status, 200, JSON.stringify(r.json));
+ const grantKey = ledger.calls.at(-1).key;
+ assert.match(grantKey, /^live:admin_grant:\d+$/, 'the grant is keyed by its local record');
+ r = await call('POST', '/api/coins/admin/grant', 1, { userId: 2, amount: 50, reason: 'thanks' }, { 'Idempotency-Key': 'grant-abc-123' });
+ assert.strictEqual(ledger.calls.at(-1).key, grantKey, 'a repeated submit reuses the grant and its key');
+ assert.strictEqual(ledger.balance, g0 + 50, 'and credits once');
+ r = await call('POST', '/api/coins/admin/grant', 1, { userId: 2, amount: 99 }, { 'Idempotency-Key': 'grant-abc-123' });
+ assert.strictEqual(r.status, 400, 'the same Idempotency-Key for a different grant is refused');
+ await call('POST', '/api/coins/admin/grant', 1, { userId: 2, amount: 50 });
+ assert.notStrictEqual(ledger.calls.at(-1).key, grantKey, 'a new grant without a key is a new event');
+ assert.strictEqual(ledger.balance, g0 + 100);
+
+ // ── No random keys left in the loyalty paths ───────────────────────────────────
+ for (const f of ['server/media/media-queue.js', 'server/monetization/opencoins.js', 'server/monetization/coins-routes.js']) {
+ const keyLines = fs.readFileSync(path.join(__dirname, '..', f), 'utf8').split('\n').filter((l) => /`live:(media_req|media_refund|admin_grant|cp:)/.test(l));
+ assert.ok(keyLines.length, `${f} builds loyalty keys`);
+ for (const l of keyLines) assert.ok(!/Date\.now\(\)|Math\.random\(\)|randomUUID/.test(l), `${f}: a loyalty key never uses the clock or randomness: ${l.trim()}`);
+ }
+ const ba = fs.readFileSync(path.join(__dirname, '..', 'server/monetization/billing-actions.js'), 'utf8');
+ assert.ok(/key: `live:media_charge:\$\{requestId\}`/.test(ba), 'the Billing media charge is keyed by the request id');
+
+ server.close();
+ fs.rmSync(tmp, { recursive: true, force: true });
+ quiet('loyalty-keys: ok');
+ process.exit(0);
+})().catch((err) => {
+ quiet(err);
+ process.exit(1);
+});
@@ -100,7 +100,7 @@ async function check(name, fn) {
@@ -100,7 +100,7 @@ async function check(name, fn) {
assert.strictEqual(r.status, 503, `${label} ${p}: ${r.text}`);
assert.strictEqual(r.status, 503, `${label} ${p}: ${r.text}`);
assert.strictEqual(r.json.code, 'money_writes_frozen', `${label} ${p}`);
assert.strictEqual(r.json.code, 'money_writes_frozen', `${label} ${p}`);
}
}
- await assert.rejects(mediaQueue.charge({ currency: 'vibes', cost: 10, userId: 2, streamerId: 3, streamId: null, label: 'x' }), /paused/);
+ await assert.rejects(mediaQueue.charge({ currency: 'vibes', cost: 10, userId: 2, streamerId: 3, streamId: null, label: 'x', requestId: 9001 }), /paused/);
}
}
await check('only the owner can freeze; every admin can see it', async () => {
await check('only the owner can freeze; every admin can see it', async () => {