openvibers/openvibe.live Commits
Tips deliveries stay idempotent across a Live restart: the in-memory done Map in server/tips/delivery-routes.js is now the tips_deliveries table, keyed by Idempotency-Key with the stored answer. A key is claimed before its effect runs; a repeat of an answered key returns the first answer and does nothing, a repeat while the first attempt is still running gets 409 (Tips retries), a claim older than two minutes (left by a crash) is taken over, and any answer but 200 gives the key back so the retry runs it. Answers are kept 7 days and pruned at most hourly. test/tips-delivery.test.js restarts the route module between deliveries and checks the same answer with no second chat line, the in-progress 409 and stale takeover, the released key on a refusal, and pruning
Changes
@@@ -6,7 +6,10 @@** POST /internal/tips/deliveries capability live.tips_delivery.write (Network service token,* audience openvibe.live; loopback-only like every /internal route)-- * Idempotency-Key: <interaction id>:<effect> a repeat answers the first result and does nothing++ * Idempotency-Key: <interaction id>:<effect> a repeat answers the first result and does nothing,++ * across restarts too: keys and their answers are kept in the++ * tips_deliveries table for 7 days (pruned hourly). A repeat++ * while the first is still running answers 409 (Tips retries).* body { delivery_id, effect, test, creator: { type, id: usr_… }, supporter: { name, subject? },* interaction: { id, kind, amount, currency, message }, text, tts?: { text, voice },* media?: { url }, highlight_seconds?, target?: { service: 'live', type: 'stream', id } }@@@ -22,9 +25,61 @@ const db = require('../db/database');const { guard } = require('../net/service-guard');const router = express.Router();--const done = new Map(); // Idempotency-Key → response (Tips retries within minutes; bounded)const SUBJECT_RE = /^usr_[0-9A-HJKMNP-TV-Z]{26}$/;++// ── Idempotency store (SQLite, so a Live restart does not forget a delivery) ──────────────++const KEEP_DAYS = 7; // Tips retries within minutes; a week is ample and stays small++const CLAIM_STALE_MS = 120000; // a claim this old was left by a crash: the next retry takes it over++const PRUNE_EVERY_MS = 60 * 60 * 1000;++let tablesReady = false;++let lastPrune = 0;++++function ensureTables() {++ if (tablesReady) return;++ db.getDb().exec(`CREATE TABLE IF NOT EXISTS tips_deliveries (++ idempotency_key TEXT PRIMARY KEY,++ effect TEXT,++ state TEXT NOT NULL DEFAULT 'pending', -- pending (running) | done (response stored)++ response_json TEXT,++ claimed_at INTEGER NOT NULL, -- ms epoch++ created_at DATETIME DEFAULT CURRENT_TIMESTAMP++ )`);++ db.getDb().exec('CREATE INDEX IF NOT EXISTS idx_tips_deliveries_created ON tips_deliveries(created_at)');++ tablesReady = true;++}++++/** Delete answers older than KEEP_DAYS. Returns how many rows went. */++function prune({ force = false } = {}) {++ ensureTables();++ if (!force && Date.now() - lastPrune < PRUNE_EVERY_MS) return 0;++ lastPrune = Date.now();++ return db.getDb().prepare(`DELETE FROM tips_deliveries WHERE created_at < datetime('now', ?)`).run(`-${KEEP_DAYS} days`).changes;++}++++/**++ * Claim a key before running its effect: { done: response } for a key already answered,++ * { busy: true } while another request runs it, { claimed: true } when this request may run it.++ */++function claim(key, effect) {++ ensureTables();++ const d = db.getDb();++ const now = Date.now();++ if (d.prepare("INSERT OR IGNORE INTO tips_deliveries (idempotency_key, effect, state, claimed_at) VALUES (?, ?, 'pending', ?)").run(key, effect || null, now).changes) {++ return { claimed: true };++ }++ const row = d.prepare('SELECT * FROM tips_deliveries WHERE idempotency_key = ?').get(key);++ if (!row) return claim(key, effect); // pruned or released in between++ if (row.state === 'done') return { done: JSON.parse(row.response_json || '{}') };++ if (now - row.claimed_at < CLAIM_STALE_MS) return { busy: true };++ const took = d.prepare("UPDATE tips_deliveries SET claimed_at = ? WHERE idempotency_key = ? AND state = 'pending' AND claimed_at = ?").run(now, key, row.claimed_at).changes;++ return took ? { claimed: true } : { busy: true };++}++++/** A failed attempt gives the key back, so Tips' retry runs it again. */++function release(key) {++ db.getDb().prepare("DELETE FROM tips_deliveries WHERE idempotency_key = ? AND state = 'pending'").run(key);++}++function channelUserId(subject) {if (!SUBJECT_RE.test(String(subject || ''))) return null;const row = db.getDb().prepare("SELECT user_id FROM linked_accounts WHERE service = 'network' AND subject_id = ? ORDER BY id DESC LIMIT 1").get(subject);@@@ -41,16 +96,20 @@ function streamIdFor(target, userId) {}function remember(key, out) {-- done.set(key, out);-- if (done.size > 5000) done.delete(done.keys().next().value);++ db.getDb().prepare("UPDATE tips_deliveries SET state = 'done', response_json = ? WHERE idempotency_key = ?").run(JSON.stringify(out), key);++ try { prune(); } catch (e) { console.warn('[Tips delivery] prune:', e.message); }return out;}router.post('/deliveries', guard('live.tips_delivery.write'), express.json({ limit: '32kb' }), async (req, res) => {const key = String(req.get('idempotency-key') || '');if (!/^[A-Za-z0-9_:.-]{8,200}$/.test(key)) return res.status(400).json({ error: 'Idempotency-Key required' });-- if (done.has(key)) return res.json(done.get(key));const b = req.body || {};++ const c = claim(key, b.effect);++ if (c.done) return res.json(c.done);++ if (c.busy) return res.status(409).json({ error: 'this delivery is already running; retry shortly' });++ // Every answer but a stored 200 gives the claim back (Tips retries 409 and 5xx).++ res.on('finish', () => { if (res.statusCode !== 200) { try { release(key); } catch { /* */ } } });const i = b.interaction || {};const userId = channelUserId(b.creator && b.creator.id);if (!userId) return res.status(404).json({ error: 'this creator has no Live channel' });@@@ -106,3 +165,4 @@ router.post('/deliveries', guard('live.tips_delivery.write'), express.json({ lim});module.exports = router;++module.exports.prune = prune;@@@ -1,8 +1,9 @@'use strict';// OpenVibe.Tips → Live (roadmap Wave 9): POST /internal/tips/deliveries announces a settled tip in the--// creator's chat — service token with live.tips_delivery.write, idempotent per delivery id, the--// creator found by Network subject, no Live balance touched.++// creator's chat — service token with live.tips_delivery.write, idempotent per delivery id (kept in++// SQLite, so a restart between deliveries does not repeat one), the creator found by Network++// subject, no Live balance touched.const assert = require('assert');const fs = require('fs');@@@ -78,6 +79,43 @@ const token = (cap, aud = 'openvibe.live') => serviceAuth.signServiceToken({ issassert.strictEqual(broadcasts.length, 1);assert.strictEqual(d.prepare("SELECT COUNT(*) AS n FROM chat_messages WHERE message_type = 'donation'").get().n, 1);++ // A Live restart between deliveries: the answered key is in SQLite, not in the old process.++ delete require.cache[require.resolve('../server/tips/delivery-routes')];++ const app2 = express();++ app2.use('/internal/tips', require('../server/tips/delivery-routes'));++ const server2 = await new Promise((r) => { const s = http.createServer(app2); s.listen(0, '127.0.0.1', () => r(s)); });++ const base2 = `http://127.0.0.1:${server2.address().port}`;++ const post2 = (body, key) => fetch(`${base2}/internal/tips/deliveries`, {++ method: 'POST', headers: { 'Content-Type': 'application/json', Authorization: `Bearer ${token(['live.tips_delivery.write'])}`, 'Idempotency-Key': key }, body: JSON.stringify(body),++ }).then(async (r) => ({ status: r.status, json: await r.json().catch(() => null) }));++ const afterRestart = await post2(job, 'tint_1:chat_line');++ assert.strictEqual(afterRestart.status, 200);++ assert.deepStrictEqual(afterRestart.json, a.json, 'after a restart the same key answers the first result');++ assert.strictEqual(broadcasts.length, 1, 'and delivers nothing again');++ assert.strictEqual(d.prepare("SELECT COUNT(*) AS n FROM chat_messages WHERE message_type = 'donation'").get().n, 1);++++ // A retry while the first attempt is still running is told to come back; a claim left by a++ // crash (older than two minutes) is taken over.++ d.prepare("INSERT INTO tips_deliveries (idempotency_key, effect, state, claimed_at) VALUES ('tint_9:chat_line', 'chat_line', 'pending', ?)").run(Date.now());++ assert.strictEqual((await post2({ ...job, interaction: { ...job.interaction, id: 'tint_9' } }, 'tint_9:chat_line')).status, 409);++ assert.strictEqual(broadcasts.length, 1);++ d.prepare("UPDATE tips_deliveries SET claimed_at = ? WHERE idempotency_key = 'tint_9:chat_line'").run(Date.now() - 5 * 60 * 1000);++ assert.strictEqual((await post2({ ...job, interaction: { ...job.interaction, id: 'tint_9' } }, 'tint_9:chat_line')).status, 200);++ assert.strictEqual(broadcasts.length, 2);++++ // A refused delivery gives its key back, so Tips' retry runs it.++ const gone = { ...job, creator: { type: 'user', id: 'usr_01J0000000000000000000000X' } };++ assert.strictEqual((await post2(gone, 'tint_8:chat_line')).status, 404);++ assert.strictEqual(d.prepare("SELECT COUNT(*) AS n FROM tips_deliveries WHERE idempotency_key = 'tint_8:chat_line'").get().n, 0);++++ // Pruning: answers older than 7 days go, recent ones stay.++ d.prepare("INSERT INTO tips_deliveries (idempotency_key, effect, state, response_json, claimed_at, created_at) VALUES ('tint_old:chat_line', 'chat_line', 'done', '{}', 0, datetime('now', '-8 days'))").run();++ const routes2 = require('../server/tips/delivery-routes');++ assert.strictEqual(routes2.prune({ force: true }), 1);++ assert.strictEqual(d.prepare("SELECT COUNT(*) AS n FROM tips_deliveries WHERE idempotency_key = 'tint_old:chat_line'").get().n, 0);++ assert.strictEqual(d.prepare("SELECT state FROM tips_deliveries WHERE idempotency_key = 'tint_1:chat_line'").get().state, 'done');++ server2.close();++// TTS on the (offline) channel room.const t = await post({ ...job, effect: 'tts', tts: { text: 'read me', voice: 'gary' } }, { key: 'tint_1:tts' });assert.strictEqual(t.status, 200);
@@ -6,7 +6,10 @@
@@ -6,7 +6,10 @@
*
*
* POST /internal/tips/deliveries capability live.tips_delivery.write (Network service token,
* POST /internal/tips/deliveries capability live.tips_delivery.write (Network service token,
* audience openvibe.live; loopback-only like every /internal route)
* audience openvibe.live; loopback-only like every /internal route)
- * Idempotency-Key: <interaction id>:<effect> a repeat answers the first result and does nothing
+ * Idempotency-Key: <interaction id>:<effect> a repeat answers the first result and does nothing,
+ * across restarts too: keys and their answers are kept in the
+ * tips_deliveries table for 7 days (pruned hourly). A repeat
+ * while the first is still running answers 409 (Tips retries).
* body { delivery_id, effect, test, creator: { type, id: usr_… }, supporter: { name, subject? },
* body { delivery_id, effect, test, creator: { type, id: usr_… }, supporter: { name, subject? },
* interaction: { id, kind, amount, currency, message }, text, tts?: { text, voice },
* interaction: { id, kind, amount, currency, message }, text, tts?: { text, voice },
* media?: { url }, highlight_seconds?, target?: { service: 'live', type: 'stream', id } }
* media?: { url }, highlight_seconds?, target?: { service: 'live', type: 'stream', id } }
@@ -22,9 +25,61 @@ const db = require('../db/database');
@@ -22,9 +25,61 @@ const db = require('../db/database');
const { guard } = require('../net/service-guard');
const { guard } = require('../net/service-guard');
const router = express.Router();
const router = express.Router();
-const done = new Map(); // Idempotency-Key → response (Tips retries within minutes; bounded)
const SUBJECT_RE = /^usr_[0-9A-HJKMNP-TV-Z]{26}$/;
const SUBJECT_RE = /^usr_[0-9A-HJKMNP-TV-Z]{26}$/;
+// ── Idempotency store (SQLite, so a Live restart does not forget a delivery) ──────────────
+const KEEP_DAYS = 7; // Tips retries within minutes; a week is ample and stays small
+const CLAIM_STALE_MS = 120000; // a claim this old was left by a crash: the next retry takes it over
+const PRUNE_EVERY_MS = 60 * 60 * 1000;
+let tablesReady = false;
+let lastPrune = 0;
+
+function ensureTables() {
+ if (tablesReady) return;
+ db.getDb().exec(`CREATE TABLE IF NOT EXISTS tips_deliveries (
+ idempotency_key TEXT PRIMARY KEY,
+ effect TEXT,
+ state TEXT NOT NULL DEFAULT 'pending', -- pending (running) | done (response stored)
+ response_json TEXT,
+ claimed_at INTEGER NOT NULL, -- ms epoch
+ created_at DATETIME DEFAULT CURRENT_TIMESTAMP
+ )`);
+ db.getDb().exec('CREATE INDEX IF NOT EXISTS idx_tips_deliveries_created ON tips_deliveries(created_at)');
+ tablesReady = true;
+}
+
+/** Delete answers older than KEEP_DAYS. Returns how many rows went. */
+function prune({ force = false } = {}) {
+ ensureTables();
+ if (!force && Date.now() - lastPrune < PRUNE_EVERY_MS) return 0;
+ lastPrune = Date.now();
+ return db.getDb().prepare(`DELETE FROM tips_deliveries WHERE created_at < datetime('now', ?)`).run(`-${KEEP_DAYS} days`).changes;
+}
+
+/**
+ * Claim a key before running its effect: { done: response } for a key already answered,
+ * { busy: true } while another request runs it, { claimed: true } when this request may run it.
+ */
+function claim(key, effect) {
+ ensureTables();
+ const d = db.getDb();
+ const now = Date.now();
+ if (d.prepare("INSERT OR IGNORE INTO tips_deliveries (idempotency_key, effect, state, claimed_at) VALUES (?, ?, 'pending', ?)").run(key, effect || null, now).changes) {
+ return { claimed: true };
+ }
+ const row = d.prepare('SELECT * FROM tips_deliveries WHERE idempotency_key = ?').get(key);
+ if (!row) return claim(key, effect); // pruned or released in between
+ if (row.state === 'done') return { done: JSON.parse(row.response_json || '{}') };
+ if (now - row.claimed_at < CLAIM_STALE_MS) return { busy: true };
+ const took = d.prepare("UPDATE tips_deliveries SET claimed_at = ? WHERE idempotency_key = ? AND state = 'pending' AND claimed_at = ?").run(now, key, row.claimed_at).changes;
+ return took ? { claimed: true } : { busy: true };
+}
+
+/** A failed attempt gives the key back, so Tips' retry runs it again. */
+function release(key) {
+ db.getDb().prepare("DELETE FROM tips_deliveries WHERE idempotency_key = ? AND state = 'pending'").run(key);
+}
+
function channelUserId(subject) {
function channelUserId(subject) {
if (!SUBJECT_RE.test(String(subject || ''))) return null;
if (!SUBJECT_RE.test(String(subject || ''))) return null;
const row = db.getDb().prepare("SELECT user_id FROM linked_accounts WHERE service = 'network' AND subject_id = ? ORDER BY id DESC LIMIT 1").get(subject);
const row = db.getDb().prepare("SELECT user_id FROM linked_accounts WHERE service = 'network' AND subject_id = ? ORDER BY id DESC LIMIT 1").get(subject);
@@ -41,16 +96,20 @@ function streamIdFor(target, userId) {
@@ -41,16 +96,20 @@ function streamIdFor(target, userId) {
}
}
function remember(key, out) {
function remember(key, out) {
- done.set(key, out);
- if (done.size > 5000) done.delete(done.keys().next().value);
+ db.getDb().prepare("UPDATE tips_deliveries SET state = 'done', response_json = ? WHERE idempotency_key = ?").run(JSON.stringify(out), key);
+ try { prune(); } catch (e) { console.warn('[Tips delivery] prune:', e.message); }
return out;
return out;
}
}
router.post('/deliveries', guard('live.tips_delivery.write'), express.json({ limit: '32kb' }), async (req, res) => {
router.post('/deliveries', guard('live.tips_delivery.write'), express.json({ limit: '32kb' }), async (req, res) => {
const key = String(req.get('idempotency-key') || '');
const key = String(req.get('idempotency-key') || '');
if (!/^[A-Za-z0-9_:.-]{8,200}$/.test(key)) return res.status(400).json({ error: 'Idempotency-Key required' });
if (!/^[A-Za-z0-9_:.-]{8,200}$/.test(key)) return res.status(400).json({ error: 'Idempotency-Key required' });
- if (done.has(key)) return res.json(done.get(key));
const b = req.body || {};
const b = req.body || {};
+ const c = claim(key, b.effect);
+ if (c.done) return res.json(c.done);
+ if (c.busy) return res.status(409).json({ error: 'this delivery is already running; retry shortly' });
+ // Every answer but a stored 200 gives the claim back (Tips retries 409 and 5xx).
+ res.on('finish', () => { if (res.statusCode !== 200) { try { release(key); } catch { /* */ } } });
const i = b.interaction || {};
const i = b.interaction || {};
const userId = channelUserId(b.creator && b.creator.id);
const userId = channelUserId(b.creator && b.creator.id);
if (!userId) return res.status(404).json({ error: 'this creator has no Live channel' });
if (!userId) return res.status(404).json({ error: 'this creator has no Live channel' });
@@ -106,3 +165,4 @@ router.post('/deliveries', guard('live.tips_delivery.write'), express.json({ lim
@@ -106,3 +165,4 @@ router.post('/deliveries', guard('live.tips_delivery.write'), express.json({ lim
});
});
module.exports = router;
module.exports = router;
+module.exports.prune = prune;
@@ -1,8 +1,9 @@
@@ -1,8 +1,9 @@
'use strict';
'use strict';
// OpenVibe.Tips → Live (roadmap Wave 9): POST /internal/tips/deliveries announces a settled tip in the
// OpenVibe.Tips → Live (roadmap Wave 9): POST /internal/tips/deliveries announces a settled tip in the
-// creator's chat — service token with live.tips_delivery.write, idempotent per delivery id, the
-// creator found by Network subject, no Live balance touched.
+// creator's chat — service token with live.tips_delivery.write, idempotent per delivery id (kept in
+// SQLite, so a restart between deliveries does not repeat one), the creator found by Network
+// subject, no Live balance touched.
const assert = require('assert');
const assert = require('assert');
const fs = require('fs');
const fs = require('fs');
@@ -78,6 +79,43 @@ const token = (cap, aud = 'openvibe.live') => serviceAuth.signServiceToken({ iss
@@ -78,6 +79,43 @@ const token = (cap, aud = 'openvibe.live') => serviceAuth.signServiceToken({ iss
assert.strictEqual(broadcasts.length, 1);
assert.strictEqual(broadcasts.length, 1);
assert.strictEqual(d.prepare("SELECT COUNT(*) AS n FROM chat_messages WHERE message_type = 'donation'").get().n, 1);
assert.strictEqual(d.prepare("SELECT COUNT(*) AS n FROM chat_messages WHERE message_type = 'donation'").get().n, 1);
+ // A Live restart between deliveries: the answered key is in SQLite, not in the old process.
+ delete require.cache[require.resolve('../server/tips/delivery-routes')];
+ const app2 = express();
+ app2.use('/internal/tips', require('../server/tips/delivery-routes'));
+ const server2 = await new Promise((r) => { const s = http.createServer(app2); s.listen(0, '127.0.0.1', () => r(s)); });
+ const base2 = `http://127.0.0.1:${server2.address().port}`;
+ const post2 = (body, key) => fetch(`${base2}/internal/tips/deliveries`, {
+ method: 'POST', headers: { 'Content-Type': 'application/json', Authorization: `Bearer ${token(['live.tips_delivery.write'])}`, 'Idempotency-Key': key }, body: JSON.stringify(body),
+ }).then(async (r) => ({ status: r.status, json: await r.json().catch(() => null) }));
+ const afterRestart = await post2(job, 'tint_1:chat_line');
+ assert.strictEqual(afterRestart.status, 200);
+ assert.deepStrictEqual(afterRestart.json, a.json, 'after a restart the same key answers the first result');
+ assert.strictEqual(broadcasts.length, 1, 'and delivers nothing again');
+ assert.strictEqual(d.prepare("SELECT COUNT(*) AS n FROM chat_messages WHERE message_type = 'donation'").get().n, 1);
+
+ // A retry while the first attempt is still running is told to come back; a claim left by a
+ // crash (older than two minutes) is taken over.
+ d.prepare("INSERT INTO tips_deliveries (idempotency_key, effect, state, claimed_at) VALUES ('tint_9:chat_line', 'chat_line', 'pending', ?)").run(Date.now());
+ assert.strictEqual((await post2({ ...job, interaction: { ...job.interaction, id: 'tint_9' } }, 'tint_9:chat_line')).status, 409);
+ assert.strictEqual(broadcasts.length, 1);
+ d.prepare("UPDATE tips_deliveries SET claimed_at = ? WHERE idempotency_key = 'tint_9:chat_line'").run(Date.now() - 5 * 60 * 1000);
+ assert.strictEqual((await post2({ ...job, interaction: { ...job.interaction, id: 'tint_9' } }, 'tint_9:chat_line')).status, 200);
+ assert.strictEqual(broadcasts.length, 2);
+
+ // A refused delivery gives its key back, so Tips' retry runs it.
+ const gone = { ...job, creator: { type: 'user', id: 'usr_01J0000000000000000000000X' } };
+ assert.strictEqual((await post2(gone, 'tint_8:chat_line')).status, 404);
+ assert.strictEqual(d.prepare("SELECT COUNT(*) AS n FROM tips_deliveries WHERE idempotency_key = 'tint_8:chat_line'").get().n, 0);
+
+ // Pruning: answers older than 7 days go, recent ones stay.
+ d.prepare("INSERT INTO tips_deliveries (idempotency_key, effect, state, response_json, claimed_at, created_at) VALUES ('tint_old:chat_line', 'chat_line', 'done', '{}', 0, datetime('now', '-8 days'))").run();
+ const routes2 = require('../server/tips/delivery-routes');
+ assert.strictEqual(routes2.prune({ force: true }), 1);
+ assert.strictEqual(d.prepare("SELECT COUNT(*) AS n FROM tips_deliveries WHERE idempotency_key = 'tint_old:chat_line'").get().n, 0);
+ assert.strictEqual(d.prepare("SELECT state FROM tips_deliveries WHERE idempotency_key = 'tint_1:chat_line'").get().state, 'done');
+ server2.close();
+
// TTS on the (offline) channel room.
// TTS on the (offline) channel room.
const t = await post({ ...job, effect: 'tts', tts: { text: 'read me', voice: 'gary' } }, { key: 'tint_1:tts' });
const t = await post({ ...job, effect: 'tts', tts: { text: 'read me', voice: 'gary' } }, { key: 'tint_1:tts' });
assert.strictEqual(t.status, 200);
assert.strictEqual(t.status, 200);