openvibers/openvibe.tips Commits
Tips' outbox is openvibe-sdk/events createServiceOutbox (plan T1): tips_event_outbox, the service as actor, internal visibility and important priority on every envelope as before; callers use emitIn(t, envelope) inside the change's transaction. Pins openvibe-sdk v0.25.0, openvibe-contracts v0.79.0, openvibe-shared v2.0.0. Both suites green (verified).
Changes
@@@ -46,10 +46,11 @@ from *interaction delivered*.(openvibe-sdk transactional outbox)- **OpenVibe.Network** — SSO for people, service tokens, JWKS, identity resolve- **OpenVibe.Live** — chat delivery through `/internal/tips/deliveries` (deployed in Live since `f11f809`; [docs/live-patch.diff](docs/live-patch.diff) is the original patch) until OpenVibe.Chat exposes paid messages; overlay consumer--- **OpenVibe.Shared** v1.25.0 (app icon, SSR footer, noscript nav, release manifest, legal pages) and-- the Network's `navbar.js`; **openvibe-contracts** v0.76.0 and **openvibe-sdk** v0.18.0 (service tokens,-- the async data layer `openvibe-sdk/db`, the PostgreSQL outbox and inbox, Valkey, pub/sub and the shared-- limit store), pinned by release tarball; `pg` and `iovalkey` (drivers), PGlite for tests++- **OpenVibe.Shared** v2.0.0 (app icon, SSR footer, noscript nav, release manifest, legal pages) and++ the Network's `navbar.js`; **openvibe-contracts** v0.79.0 and **openvibe-sdk** v0.25.0 (service tokens,++ the async data layer `openvibe-sdk/db`, the PostgreSQL outbox and inbox — the events outbox is the SDK's++ `createServiceOutbox`, plan T1 — Valkey, pub/sub and the shared limit store), pinned by release tarball;++ `pg` and `iovalkey` (drivers), PGlite for tests## Capabilities@@@ -412,7 +413,7 @@ exist here (plan §12.12):3. server-rendered public routes useful without JavaScript — ✔;4. real persistence and end-to-end workflows — ✔ against stubs; not yet against the real Billing andEvents (Billing is in shadow and has sent no events; production has no creator profile yet);--5. capability and event registration against OpenVibe.Contracts — ✔ v0.15.0, payload schemas in v0.30.2 and v0.32.0 (v0.76.0 pinned);++5. capability and event registration against OpenVibe.Contracts — ✔ v0.15.0, payload schemas in v0.30.2 and v0.32.0 (v0.79.0 pinned);6. a migration/seed strategy ✔, a written threat review ✔ ([docs/threat-review.md](docs/threat-review.md),internal; an independent review is still due), and sitemap/robots ✔ (no feed);7. acceptance tests proving the advertised functionality — ✔ (table above).@@@ -8,10 +8,10 @@"authoritative": false,"deployed": true,"deployment": "Deployed internally, not launched: deployed with `sudo ovhost deploy tips`, /opt/openvibe.tips, unit openvibe-tips, 127.0.0.1:4610 loopback only; https://openvibe.tips still serves the OpenVibe.Sites placeholder. Principal tips; Events subscriptions billing.transaction.* and billing.receipt.* -> /internal/events (secret rotated 2026-09-24); TIPS_CHAT_ADAPTER is empty (no chat side effects).",-- "database": "PostgreSQL 18 through PgBouncer (transaction mode) as the system of record, Valkey 9 for shared state (per-actor limits, overlay fan-out, overlay stream slots), ADR-035: database ov_tips from OpenVibe.Host's data role, schema in migrations/ (0001_initial expand, 0002_drop_legacy_import contract), async on openvibe-sdk/db v0.18.0, money paths serializable with row locks, PostgreSQL outbox and inbox.",-- "contracts": "openvibe-contracts v0.76.0",++ "database": "PostgreSQL 18 through PgBouncer (transaction mode) as the system of record, Valkey 9 for shared state (per-actor limits, overlay fan-out, overlay stream slots), ADR-035: database ov_tips from OpenVibe.Host's data role, schema in migrations/ (0001_initial expand, 0002_drop_legacy_import contract), async on openvibe-sdk/db v0.25.0, money paths serializable with row locks, PostgreSQL outbox and inbox.",++ "contracts": "openvibe-contracts v0.79.0","notes": "Runtime, API, SSR pages, overlays, the Billing settlement consumer and the events outbox are built, tested and deployed, together with /metrics (loopback only), truthful /api/ready (Billing and Events checks), supporter privacy (anonymous, hidden amount, private message), goal and supporter-page settings, export and erasure (tips.interaction.erased), moderation (word filter, hide/show, invited moderators, tips.interaction.moderated) and per-actor limits. Capabilities (18 tips.*) and payload schemas are in openvibe-contracts. No billing.* event had arrived: Billing runs in shadow. A written internal threat review (docs/threat-review.md) with its fixes and test/security.test.js. Not done: an independent security review; the openvibe.tips launch. Live stays the place tips happen until the Billing cutover (blocked on the owner re-pointing the PowerChat webhook) and the launch rule in README.md holds.",-- "updated": "2026-09-28",++ "updated": "2026-09-29","features": ["tips, paid messages, TTS and media requests settled by OpenVibe.Billing (never Tips' own money)","goals counted from settled payments only",@@@ -14,9 +14,9 @@"express": "^4.21.2","iovalkey": "^0.4.0","jsonwebtoken": "^9.0.2",-- "openvibe-contracts": "https://codeload.github.com/OpenVibers/OpenVibe.Contracts/tar.gz/refs/tags/v0.76.0",-- "openvibe-sdk": "https://codeload.github.com/OpenVibers/OpenVibe.SDK/tar.gz/refs/tags/v0.18.0",-- "openvibe-shared": "https://codeload.github.com/OpenVibers/OpenVibe.Shared/tar.gz/refs/tags/v1.27.0",++ "openvibe-contracts": "https://codeload.github.com/OpenVibers/OpenVibe.Contracts/tar.gz/refs/tags/v0.79.0",++ "openvibe-sdk": "https://codeload.github.com/OpenVibers/OpenVibe.SDK/tar.gz/refs/tags/v0.25.0",++ "openvibe-shared": "https://codeload.github.com/OpenVibers/OpenVibe.Shared/tar.gz/refs/tags/v2.0.0","pg": "^8.23.0"},"devDependencies": {@@@ -1117,9 +1117,9 @@}},"node_modules/openvibe-contracts": {-- "version": "0.76.0",-- "resolved": "https://codeload.github.com/OpenVibers/OpenVibe.Contracts/tar.gz/refs/tags/v0.76.0",-- "integrity": "sha512-zTdx776KWrcJKm1JTnf+0+BnsvswdvXy1ldPyse1hSqhnW/SSr98sX696eRetFUhTl8qVM1xSRCykqIUbqItPw==",++ "version": "0.79.0",++ "resolved": "https://codeload.github.com/OpenVibers/OpenVibe.Contracts/tar.gz/refs/tags/v0.79.0",++ "integrity": "sha512-FY5GLbiOq0kDw5lUPPzX6h3ATbzvsr/mX5DUCF4GRfn6vGMP0p89rZZDhgtVDaS+yCBCYMZ8zFaZxuHEkEVwVg==","license": "AGPL-3.0-only","dependencies": {"ajv": "^8.17.1",@@@ -1133,9 +1133,9 @@}},"node_modules/openvibe-sdk": {-- "version": "0.18.0",-- "resolved": "https://codeload.github.com/OpenVibers/OpenVibe.SDK/tar.gz/refs/tags/v0.18.0",-- "integrity": "sha512-Jh+lMx35cWM1byXUqt0pb3mYdOjkHneYIPOzKlHdYFhzmlUSGLlLF1WJviJxxbZ0XVy/0jTcZ38OCPGxOz7Aeg==",++ "version": "0.25.0",++ "resolved": "https://codeload.github.com/OpenVibers/OpenVibe.SDK/tar.gz/refs/tags/v0.25.0",++ "integrity": "sha512-8eurZLyFUtB8XN16hLUuYkah+qwS3tkvdzul5qdZkJTi0aZ4q8dq2QGWSq6pN5mtVPdJyuJ8Mo53r36lixmP1g==","license": "MIT","engines": {"node": ">=20"@@@ -1162,9 +1162,9 @@}},"node_modules/openvibe-shared": {-- "version": "1.27.0",-- "resolved": "https://codeload.github.com/OpenVibers/OpenVibe.Shared/tar.gz/refs/tags/v1.27.0",-- "integrity": "sha512-/H7dyFw+cURcdjyfbpyY7mNc6F+Pcb9TwkVwZHR+9d3M62q70KDhZLo/gZi71eaYxUjD8lO8tmbXu70EAa9paQ==",++ "version": "2.0.0",++ "resolved": "https://codeload.github.com/OpenVibers/OpenVibe.Shared/tar.gz/refs/tags/v2.0.0",++ "integrity": "sha512-n5CHC0hfxOo0jo/3kS4O3KiwhirhZDlx5zZTIuPyIf3Dq0yI/R6R64btx7NYwZe5dI0NjvqzEtnv19m8unwLMg==","license": "MIT","dependencies": {"jsonwebtoken": "^9.0.0"@@@ -26,9 +26,9 @@"express": "^4.21.2","iovalkey": "^0.4.0","jsonwebtoken": "^9.0.2",-- "openvibe-contracts": "https://codeload.github.com/OpenVibers/OpenVibe.Contracts/tar.gz/refs/tags/v0.76.0",-- "openvibe-sdk": "https://codeload.github.com/OpenVibers/OpenVibe.SDK/tar.gz/refs/tags/v0.18.0",-- "openvibe-shared": "https://codeload.github.com/OpenVibers/OpenVibe.Shared/tar.gz/refs/tags/v1.27.0",++ "openvibe-contracts": "https://codeload.github.com/OpenVibers/OpenVibe.Contracts/tar.gz/refs/tags/v0.79.0",++ "openvibe-sdk": "https://codeload.github.com/OpenVibers/OpenVibe.SDK/tar.gz/refs/tags/v0.25.0",++ "openvibe-shared": "https://codeload.github.com/OpenVibers/OpenVibe.Shared/tar.gz/refs/tags/v2.0.0","pg": "^8.23.0"},"engines": {@@@ -20,6 +20,7 @@ const { sql } = require('openvibe-sdk/db');const { fail, iso, prefixedId, text, positiveInt } = require('../util');const { safeUrl } = require('./profiles');const { publicName, shownName, publicView } = require('./privacy');++const { ENVELOPE } = require('../events/outbox');function createGoals(ctx) {const { db } = ctx;@@@ -102,9 +103,12 @@ function createGoals(ctx) {async function changed(t, goalId, reason, { interactionId = null, by = null } = {}) {const g = await t.one(sql`UPDATE tip_goals SET revision = revision + 1, updated_at = ${iso(ctx.now())} WHERE id = ${goalId} RETURNING *`);const [view] = await presentMany(t, [g]);-- await ctx.outbox.emit(t, 'tips.goal.updated', { type: 'goal', id: g.id, revision: g.revision }, {-- goal_id: g.id, creator: view.creator, title: g.title, target_amount: g.target_amount, current_amount: view.current_amount,-- currency: g.currency, status: g.status, reached: view.reached, reason, interaction_id: interactionId,++ await ctx.outbox.emitIn(t, {++ ...ENVELOPE, event_type: 'tips.goal.updated', subject: { type: 'goal', id: g.id, revision: g.revision },++ payload: {++ goal_id: g.id, creator: view.creator, title: g.title, target_amount: g.target_amount, current_amount: view.current_amount,++ currency: g.currency, status: g.status, reached: view.reached, reason, interaction_id: interactionId,++ },});await ctx.overlays.addGoalDelivery(t, g.creator_subject, view, { reason, interactionId, by, dedupe: `goal:${g.id}:r${g.revision}` });return view;@@@ -19,7 +19,7 @@* ctx.lockCreator(t, s) lock a creator's profile row: settings changes and per-creator limits* (20 active goals, 25 tokens, 50 moderators, 10 invitations) take it, so two* requests at once count and write one after the other-- * ctx.outbox.emit(t, …) a durable event, inside the transaction (openvibe-sdk outbox)++ * ctx.outbox.emitIn(t, …) a durable event, inside the transaction (openvibe-sdk outbox)** Lock order (so two transactions never wait on each other in a cycle): a creator's profile row, then* tip_interactions rows (SELECT … FOR UPDATE, several by id), then tip_goals rows by id, then the@@@ -55,7 +55,8 @@ function createDomain({ db, config, outbox, billing, adapters, valkey = null, noctx.within = (q, fn, opts) => (inTx(q) ? fn(q) : ctx.tx(fn, opts));ctx.afterCommit = (q, hook) => { if (inTx(q)) q.after(hook); else runHook(hook); };ctx.lockCreator = (t, subject) => t.maybe(sql`SELECT creator_subject FROM creator_tip_profiles WHERE creator_subject = ${subject} FOR UPDATE`);-- ctx.outboxKick = () => { try { outbox.kick(); } catch { /* relay off */ } };++ // The SDK's kick is async: a rejected relay-off throw would be an unhandled rejection on a floating call.++ ctx.outboxKick = () => { try { Promise.resolve(outbox.kick()).catch(() => {}); } catch { /* relay off */ } };ctx.profiles = createProfiles(ctx);ctx.goals = createGoals(ctx);@@@ -47,6 +47,7 @@ const { fail, iso, prefixedId, positiveInt, text, storable, userSubject, entityRconst { VOICES } = require('./profiles');const { parsePrivacy, publicName, shownName, privacyOf, isAnonymous, publicView, ANONYMOUS } = require('./privacy');const { BillingCallError } = require('../billing-client');++const { ENVELOPE } = require('../events/outbox');const KINDS = ['tip', 'paid_message', 'tts', 'media_request'];// Origins whose chat effects Tips delivers: its own requests, and EXTERNAL tips Billing announced@@@ -311,7 +312,7 @@ function createInteractions(ctx) {if (i.origin !== 'import' && i.moderation === 'visible') await release(t, i, filter);i = await recomputeDelivery(t, i.id);if (!test && i.origin !== 'import') {-- await ctx.outbox.emit(t, 'tips.interaction.ready', { type: 'interaction', id: i.id }, summary(i));++ await ctx.outbox.emitIn(t, { ...ENVELOPE, event_type: 'tips.interaction.ready', subject: { type: 'interaction', id: i.id }, payload: summary(i) });}if (verdict.hit) await ctx.moderation.recordScreened(t, i, verdict);ctx.afterCommit(t, () => { ctx.effects.kick(); ctx.outboxKick(); });@@@ -346,7 +347,7 @@ function createInteractions(ctx) {if (!i || i.payment_state !== 'pending') return i;const row = await t.one(sql`UPDATE tip_interactions SET payment_state = 'failed', delivery_state = 'cancelled', transfer_due = false,failure = ${`${code}${detail ? `: ${String(detail).slice(0, 240)}` : ''}`}, updated_at = ${iso(ctx.now())} WHERE id = ${i.id} RETURNING *`);-- if (!i.test) await ctx.outbox.emit(t, 'tips.interaction.cancelled', { type: 'interaction', id: i.id }, { ...summary(row), reason: 'payment_failed', code });++ if (!i.test) await ctx.outbox.emitIn(t, { ...ENVELOPE, event_type: 'tips.interaction.cancelled', subject: { type: 'interaction', id: i.id }, payload: { ...summary(row), reason: 'payment_failed', code } });return row;}@@@ -425,7 +426,7 @@ function createInteractions(ctx) {const row = await t.one(sql`UPDATE tip_interactions SET delivery_state = 'cancelled', updated_at = ${at} WHERE id = ${i.id} RETURNING *`);await t.exec(sql`UPDATE paid_messages SET status = 'cancelled', updated_at = ${at} WHERE interaction_id = ${i.id} AND status = 'queued'`);await t.exec(sql`UPDATE paid_media_requests SET status = 'cancelled', updated_at = ${at} WHERE interaction_id = ${i.id} AND status = 'queued'`);-- if (!now.test) await ctx.outbox.emit(t, 'tips.interaction.cancelled', { type: 'interaction', id: i.id }, { ...summary(row), reason: 'payment_reversed' });++ if (!now.test) await ctx.outbox.emitIn(t, { ...ENVELOPE, event_type: 'tips.interaction.cancelled', subject: { type: 'interaction', id: i.id }, payload: { ...summary(row), reason: 'payment_reversed' } });cancelled = true;}return cancelled ? 'reversed_cancelled' : 'reversed';@@@ -439,7 +440,7 @@ function createInteractions(ctx) {await t.exec(sql`UPDATE paid_media_requests SET status = 'cancelled', updated_at = ${at} WHERE interaction_id = ${i.id} AND status = 'queued'`);if (!n) return false;const row = await t.one(sql`UPDATE tip_interactions SET delivery_state = 'cancelled', updated_at = ${at} WHERE id = ${i.id} RETURNING *`);-- if (!i.test) await ctx.outbox.emit(t, 'tips.interaction.cancelled', { type: 'interaction', id: i.id }, { ...summary(row), reason });++ if (!i.test) await ctx.outbox.emitIn(t, { ...ENVELOPE, event_type: 'tips.interaction.cancelled', subject: { type: 'interaction', id: i.id }, payload: { ...summary(row), reason } });return true;}@@@ -459,7 +460,7 @@ function createInteractions(ctx) {delivered_at = CASE WHEN ${next === 'delivered'} THEN ${at}::timestamptz ELSE delivered_at END, updated_at = ${at} WHERE id = ${id} RETURNING *`);if (next === 'failed' && !i.test) {const failed = effects.filter((e) => e.state === 'failed').map(({ effect, adapter, last_error: lastError }) => ({ effect, adapter, last_error: lastError }));-- await ctx.outbox.emit(t, 'tips.interaction.failed', { type: 'interaction', id }, { ...summary(row), failed_effects: failed });++ await ctx.outbox.emitIn(t, { ...ENVELOPE, event_type: 'tips.interaction.failed', subject: { type: 'interaction', id }, payload: { ...summary(row), failed_effects: failed } });}return row;}@@@ -714,9 +715,12 @@ function createInteractions(ctx) {WHERE envelope->'subject'->>'type' = 'interaction' AND envelope->'subject'->>'id' = ANY(${ids}) AND envelope->'payload' ? 'supporter'`);for (const i of done) {if (i.test) continue;-- await ctx.outbox.emit(t, 'tips.interaction.erased', { type: 'interaction', id: i.id }, {-- interaction_id: i.id, creator: { type: 'user', id: i.creator_subject }, erased_at: at,-- redacts: { subject_type: 'interaction', subject_ids: [i.id] },++ await ctx.outbox.emitIn(t, {++ ...ENVELOPE, event_type: 'tips.interaction.erased', subject: { type: 'interaction', id: i.id },++ payload: {++ interaction_id: i.id, creator: { type: 'user', id: i.creator_subject }, erased_at: at,++ redacts: { subject_type: 'interaction', subject_ids: [i.id] },++ },});}t.after(ctx.outboxKick);@@@ -28,6 +28,7 @@ const { sql } = require('openvibe-sdk/db');const { fail, iso, prefixedId, sha256, userSubject, displayName, text, readCursor, cursorOf } = require('../util');const { publicView } = require('./privacy');const filter = require('./filter');++const { ENVELOPE } = require('../events/outbox');const INVITE_RE = /^tmin_[A-Za-z0-9_-]{43}$/;const INVITE_TTL_MS = 7 * 24 * 3600 * 1000;@@@ -130,9 +131,12 @@ function createModeration(ctx) {await t.exec(sql`INSERT INTO tip_moderation_log (creator_subject, interaction_id, action, by_role, actor, reason, created_at)VALUES (${i.creator_subject}, ${i.id}, ${action}, ${role}, ${actor}, ${reason}, ${iso(ctx.now())})`);if (!i.test) {-- await ctx.outbox.emit(t, 'tips.interaction.moderated', { type: 'interaction', id: i.id }, {-- interaction_id: i.id, creator: { type: 'user', id: i.creator_subject }, action, by: role,-- moderation_state: i.moderation, cancelled_effects: cancelled,++ await ctx.outbox.emitIn(t, {++ ...ENVELOPE, event_type: 'tips.interaction.moderated', subject: { type: 'interaction', id: i.id },++ payload: {++ interaction_id: i.id, creator: { type: 'user', id: i.creator_subject }, action, by: role,++ moderation_state: i.moderation, cancelled_effects: cancelled,++ },});ctx.afterCommit(t, () => ctx.outboxKick());}@@@ -28,6 +28,7 @@ const { fail, iso, prefixedId, sha256, text } = require('../util');const { safeUrl } = require('./profiles');const { isHidden, publicView } = require('./privacy');const { createOverlayHub } = require('./overlay-hub');++const { ENVELOPE } = require('../events/outbox');const SCOPES = ['alerts', 'goals'];const STREAM_LOCK = 4610; // the advisory-lock namespace of a creator's overlay stream (Tips' port)@@@ -247,8 +248,11 @@ function createOverlays(ctx) {WHERE seq = ANY(${pending}) AND status = 'pending' RETURNING id, creator_subject, kind, interaction_id, goal_id, test`);for (const row of done) {if (row.test) continue;-- await ctx.outbox.emit(t, 'tips.overlay.delivered', { type: 'overlay_delivery', id: row.id }, {-- delivery_id: row.id, creator: { type: 'user', id: row.creator_subject }, kind: row.kind, interaction_id: row.interaction_id, goal_id: row.goal_id,++ await ctx.outbox.emitIn(t, {++ ...ENVELOPE, event_type: 'tips.overlay.delivered', subject: { type: 'overlay_delivery', id: row.id },++ payload: {++ delivery_id: row.id, creator: { type: 'user', id: row.creator_subject }, kind: row.kind, interaction_id: row.interaction_id, goal_id: row.goal_id,++ },});}if (done.some((r) => !r.test)) t.after(ctx.outboxKick);@@@ -269,9 +273,12 @@ function createOverlays(ctx) {rows.sort((a, b) => a.seq - b.seq);for (const row of rows) {if (row.test) continue;-- await ctx.outbox.emit(t, 'tips.overlay.failed', { type: 'overlay_delivery', id: row.id }, {-- delivery_id: row.id, creator: { type: 'user', id: row.creator_subject }, kind: row.kind, interaction_id: row.interaction_id,-- goal_id: row.goal_id, reason: 'no overlay showed it within the delivery window',++ await ctx.outbox.emitIn(t, {++ ...ENVELOPE, event_type: 'tips.overlay.failed', subject: { type: 'overlay_delivery', id: row.id },++ payload: {++ delivery_id: row.id, creator: { type: 'user', id: row.creator_subject }, kind: row.kind, interaction_id: row.interaction_id,++ goal_id: row.goal_id, reason: 'no overlay showed it within the delivery window',++ },});}return rows.length;@@@ -6,63 +6,43 @@* tips.interaction.ready settled: the interaction is active and its effects are queued* tips.interaction.failed an effect gave up after its retries (payment state untouched)* tips.interaction.cancelled undelivered effects cancelled (payment reversed/failed, or the creator)++ * tips.interaction.moderated a creator (or a moderator) acted on an interaction++ * tips.interaction.erased an erasure redacted a supporter's interaction* tips.goal.updated a goal's settled total, target or status changed* tips.overlay.delivered an overlay delivery reached at least one overlay (first time only)* tips.overlay.failed no overlay received it within the delivery window*-- * emit(t, …) runs inside the transaction that makes the change (its handle `t`), so an event exists if-- * and only if its change committed. Simulations (test) never produce events. The relay publishes with-- * Tips' service token (events.event.publish, audience openvibe.events) only when EVENTS_URL and++ * emitIn(t, …) runs inside the transaction that makes the change (its handle `t`), so an event exists++ * if and only if its change committed. Simulations (test) never produce events. The relay publishes++ * with Tips' service token (events.event.publish, audience openvibe.events) only when EVENTS_URL and* OV_OAUTH_CLIENT_SECRET are set; otherwise rows wait in tips_event_outbox. Any number of processes* relay the one table: the SDK claims due rows with a lease (FOR UPDATE SKIP LOCKED).++ *++ * Every envelope Tips writes uses ACTOR: the service is the actor (a person's act rides in the payload++ * as actor_subject), visibility 'internal' and priority 'important' as before.*/--const { createClient } = require('openvibe-sdk/core');--const { createServiceTokenClient } = require('openvibe-sdk/auth');--const { createEventsClient, createPgOutbox } = require('openvibe-sdk/events');++const { createServiceOutbox } = require('openvibe-sdk/events');--const ACTOR = { type: 'service', id: 'tips' };const TABLE = 'tips_event_outbox'; // migrations/0001_initial.sql (outboxSchema)++const ACTOR = { type: 'service', id: 'tips' };++/** The defaults every Tips envelope carries (the SDK's createServiceOutbox sets none). */++const ENVELOPE = { actor: ACTOR, visibility: 'internal', priority: 'important' };++/** Tips' wiring of the SDK's shared outbox (plan T1). */function createTipsOutbox({ db, config, fetchImpl, now, log = console }) {-- const enabled = !!(config.events.url && config.oauth.clientSecret);-- const clientOpts = { baseUrls: { events: config.events.url || 'http://127.0.0.1:4300' }, retries: 0 };-- if (fetchImpl) clientOpts.fetch = fetchImpl;-- if (enabled) {-- clientOpts.tokenProvider = createServiceTokenClient({-- tokenUrl: `${config.network.internalUrl}/oauth/token`, clientId: config.oauth.clientId, clientSecret: config.oauth.clientSecret,-- scope: { 'openvibe.events': 'events.event.publish' }, ...(fetchImpl ? { fetch: fetchImpl } : {}),-- });-- } else {-- clientOpts.getToken = async () => { throw new Error('events relay disabled (EVENTS_URL / OV_OAUTH_CLIENT_SECRET unset)'); };-- }-- const events = createEventsClient(createClient(clientOpts), { source: 'tips' });-- let lastError = null;-- const outbox = createPgOutbox(db, {-- events,++ return createServiceOutbox({++ db,++ source: 'tips',table: TABLE,++ eventsUrl: config.events.url,++ networkInternalUrl: config.network.internalUrl,++ clientId: config.oauth.clientId,++ clientSecret: config.oauth.clientSecret,intervalMs: config.events.intervalMs,-- now,-- onError: (err) => {-- const msg = err && err.message;-- if (msg !== lastError) log.warn('[Tips] event publish failed (will retry):', msg);-- lastError = msg;-- },++ log,++ ...(fetchImpl ? { fetch: fetchImpl } : {}),++ ...(now ? { now } : {}),});---- /** Inside the caller's transaction `t`. */-- function emit(t, eventType, subject, payload, { visibility = 'internal', priority = 'important', traceparent } = {}) {-- return outbox.enqueue(t, { event_type: eventType, actor: ACTOR, subject, payload, visibility, priority }, { traceparent });-- }---- return {-- emit,-- outbox,-- enabled,-- start() { if (enabled) outbox.start(); },-- stop: () => outbox.stop(),-- kick() { if (enabled) outbox.kick(); },-- status: async () => ({ enabled, pending: await outbox.pending(), rejected: await outbox.rejected(), last_error: lastError }),-- };}--module.exports = { createTipsOutbox, TABLE };++module.exports = { createTipsOutbox, TABLE, ACTOR, ENVELOPE };
@@ -46,10 +46,11 @@ from *interaction delivered*.
@@ -46,10 +46,11 @@ from *interaction delivered*.
(openvibe-sdk transactional outbox)
(openvibe-sdk transactional outbox)
- **OpenVibe.Network** — SSO for people, service tokens, JWKS, identity resolve
- **OpenVibe.Network** — SSO for people, service tokens, JWKS, identity resolve
- **OpenVibe.Live** — chat delivery through `/internal/tips/deliveries` (deployed in Live since `f11f809`; [docs/live-patch.diff](docs/live-patch.diff) is the original patch) until OpenVibe.Chat exposes paid messages; overlay consumer
- **OpenVibe.Live** — chat delivery through `/internal/tips/deliveries` (deployed in Live since `f11f809`; [docs/live-patch.diff](docs/live-patch.diff) is the original patch) until OpenVibe.Chat exposes paid messages; overlay consumer
-- **OpenVibe.Shared** v1.25.0 (app icon, SSR footer, noscript nav, release manifest, legal pages) and
- the Network's `navbar.js`; **openvibe-contracts** v0.76.0 and **openvibe-sdk** v0.18.0 (service tokens,
- the async data layer `openvibe-sdk/db`, the PostgreSQL outbox and inbox, Valkey, pub/sub and the shared
- limit store), pinned by release tarball; `pg` and `iovalkey` (drivers), PGlite for tests
+- **OpenVibe.Shared** v2.0.0 (app icon, SSR footer, noscript nav, release manifest, legal pages) and
+ the Network's `navbar.js`; **openvibe-contracts** v0.79.0 and **openvibe-sdk** v0.25.0 (service tokens,
+ the async data layer `openvibe-sdk/db`, the PostgreSQL outbox and inbox — the events outbox is the SDK's
+ `createServiceOutbox`, plan T1 — Valkey, pub/sub and the shared limit store), pinned by release tarball;
+ `pg` and `iovalkey` (drivers), PGlite for tests
## Capabilities
## Capabilities
@@ -412,7 +413,7 @@ exist here (plan §12.12):
@@ -412,7 +413,7 @@ exist here (plan §12.12):
3. server-rendered public routes useful without JavaScript — ✔;
3. server-rendered public routes useful without JavaScript — ✔;
4. real persistence and end-to-end workflows — ✔ against stubs; not yet against the real Billing and
4. real persistence and end-to-end workflows — ✔ against stubs; not yet against the real Billing and
Events (Billing is in shadow and has sent no events; production has no creator profile yet);
Events (Billing is in shadow and has sent no events; production has no creator profile yet);
-5. capability and event registration against OpenVibe.Contracts — ✔ v0.15.0, payload schemas in v0.30.2 and v0.32.0 (v0.76.0 pinned);
+5. capability and event registration against OpenVibe.Contracts — ✔ v0.15.0, payload schemas in v0.30.2 and v0.32.0 (v0.79.0 pinned);
6. a migration/seed strategy ✔, a written threat review ✔ ([docs/threat-review.md](docs/threat-review.md),
6. a migration/seed strategy ✔, a written threat review ✔ ([docs/threat-review.md](docs/threat-review.md),
internal; an independent review is still due), and sitemap/robots ✔ (no feed);
internal; an independent review is still due), and sitemap/robots ✔ (no feed);
7. acceptance tests proving the advertised functionality — ✔ (table above).
7. acceptance tests proving the advertised functionality — ✔ (table above).
@@ -8,10 +8,10 @@
@@ -8,10 +8,10 @@
"authoritative": false,
"authoritative": false,
"deployed": true,
"deployed": true,
"deployment": "Deployed internally, not launched: deployed with `sudo ovhost deploy tips`, /opt/openvibe.tips, unit openvibe-tips, 127.0.0.1:4610 loopback only; https://openvibe.tips still serves the OpenVibe.Sites placeholder. Principal tips; Events subscriptions billing.transaction.* and billing.receipt.* -> /internal/events (secret rotated 2026-09-24); TIPS_CHAT_ADAPTER is empty (no chat side effects).",
"deployment": "Deployed internally, not launched: deployed with `sudo ovhost deploy tips`, /opt/openvibe.tips, unit openvibe-tips, 127.0.0.1:4610 loopback only; https://openvibe.tips still serves the OpenVibe.Sites placeholder. Principal tips; Events subscriptions billing.transaction.* and billing.receipt.* -> /internal/events (secret rotated 2026-09-24); TIPS_CHAT_ADAPTER is empty (no chat side effects).",
- "database": "PostgreSQL 18 through PgBouncer (transaction mode) as the system of record, Valkey 9 for shared state (per-actor limits, overlay fan-out, overlay stream slots), ADR-035: database ov_tips from OpenVibe.Host's data role, schema in migrations/ (0001_initial expand, 0002_drop_legacy_import contract), async on openvibe-sdk/db v0.18.0, money paths serializable with row locks, PostgreSQL outbox and inbox.",
- "contracts": "openvibe-contracts v0.76.0",
+ "database": "PostgreSQL 18 through PgBouncer (transaction mode) as the system of record, Valkey 9 for shared state (per-actor limits, overlay fan-out, overlay stream slots), ADR-035: database ov_tips from OpenVibe.Host's data role, schema in migrations/ (0001_initial expand, 0002_drop_legacy_import contract), async on openvibe-sdk/db v0.25.0, money paths serializable with row locks, PostgreSQL outbox and inbox.",
+ "contracts": "openvibe-contracts v0.79.0",
"notes": "Runtime, API, SSR pages, overlays, the Billing settlement consumer and the events outbox are built, tested and deployed, together with /metrics (loopback only), truthful /api/ready (Billing and Events checks), supporter privacy (anonymous, hidden amount, private message), goal and supporter-page settings, export and erasure (tips.interaction.erased), moderation (word filter, hide/show, invited moderators, tips.interaction.moderated) and per-actor limits. Capabilities (18 tips.*) and payload schemas are in openvibe-contracts. No billing.* event had arrived: Billing runs in shadow. A written internal threat review (docs/threat-review.md) with its fixes and test/security.test.js. Not done: an independent security review; the openvibe.tips launch. Live stays the place tips happen until the Billing cutover (blocked on the owner re-pointing the PowerChat webhook) and the launch rule in README.md holds.",
"notes": "Runtime, API, SSR pages, overlays, the Billing settlement consumer and the events outbox are built, tested and deployed, together with /metrics (loopback only), truthful /api/ready (Billing and Events checks), supporter privacy (anonymous, hidden amount, private message), goal and supporter-page settings, export and erasure (tips.interaction.erased), moderation (word filter, hide/show, invited moderators, tips.interaction.moderated) and per-actor limits. Capabilities (18 tips.*) and payload schemas are in openvibe-contracts. No billing.* event had arrived: Billing runs in shadow. A written internal threat review (docs/threat-review.md) with its fixes and test/security.test.js. Not done: an independent security review; the openvibe.tips launch. Live stays the place tips happen until the Billing cutover (blocked on the owner re-pointing the PowerChat webhook) and the launch rule in README.md holds.",
- "updated": "2026-09-28",
+ "updated": "2026-09-29",
"features": [
"features": [
"tips, paid messages, TTS and media requests settled by OpenVibe.Billing (never Tips' own money)",
"tips, paid messages, TTS and media requests settled by OpenVibe.Billing (never Tips' own money)",
"goals counted from settled payments only",
"goals counted from settled payments only",
@@ -14,9 +14,9 @@
@@ -14,9 +14,9 @@
"express": "^4.21.2",
"express": "^4.21.2",
"iovalkey": "^0.4.0",
"iovalkey": "^0.4.0",
"jsonwebtoken": "^9.0.2",
"jsonwebtoken": "^9.0.2",
- "openvibe-contracts": "https://codeload.github.com/OpenVibers/OpenVibe.Contracts/tar.gz/refs/tags/v0.76.0",
- "openvibe-sdk": "https://codeload.github.com/OpenVibers/OpenVibe.SDK/tar.gz/refs/tags/v0.18.0",
- "openvibe-shared": "https://codeload.github.com/OpenVibers/OpenVibe.Shared/tar.gz/refs/tags/v1.27.0",
+ "openvibe-contracts": "https://codeload.github.com/OpenVibers/OpenVibe.Contracts/tar.gz/refs/tags/v0.79.0",
+ "openvibe-sdk": "https://codeload.github.com/OpenVibers/OpenVibe.SDK/tar.gz/refs/tags/v0.25.0",
+ "openvibe-shared": "https://codeload.github.com/OpenVibers/OpenVibe.Shared/tar.gz/refs/tags/v2.0.0",
"pg": "^8.23.0"
"pg": "^8.23.0"
},
},
"devDependencies": {
"devDependencies": {
@@ -1117,9 +1117,9 @@
@@ -1117,9 +1117,9 @@
}
}
},
},
"node_modules/openvibe-contracts": {
"node_modules/openvibe-contracts": {
- "version": "0.76.0",
- "resolved": "https://codeload.github.com/OpenVibers/OpenVibe.Contracts/tar.gz/refs/tags/v0.76.0",
- "integrity": "sha512-zTdx776KWrcJKm1JTnf+0+BnsvswdvXy1ldPyse1hSqhnW/SSr98sX696eRetFUhTl8qVM1xSRCykqIUbqItPw==",
+ "version": "0.79.0",
+ "resolved": "https://codeload.github.com/OpenVibers/OpenVibe.Contracts/tar.gz/refs/tags/v0.79.0",
+ "integrity": "sha512-FY5GLbiOq0kDw5lUPPzX6h3ATbzvsr/mX5DUCF4GRfn6vGMP0p89rZZDhgtVDaS+yCBCYMZ8zFaZxuHEkEVwVg==",
"license": "AGPL-3.0-only",
"license": "AGPL-3.0-only",
"dependencies": {
"dependencies": {
"ajv": "^8.17.1",
"ajv": "^8.17.1",
@@ -1133,9 +1133,9 @@
@@ -1133,9 +1133,9 @@
}
}
},
},
"node_modules/openvibe-sdk": {
"node_modules/openvibe-sdk": {
- "version": "0.18.0",
- "resolved": "https://codeload.github.com/OpenVibers/OpenVibe.SDK/tar.gz/refs/tags/v0.18.0",
- "integrity": "sha512-Jh+lMx35cWM1byXUqt0pb3mYdOjkHneYIPOzKlHdYFhzmlUSGLlLF1WJviJxxbZ0XVy/0jTcZ38OCPGxOz7Aeg==",
+ "version": "0.25.0",
+ "resolved": "https://codeload.github.com/OpenVibers/OpenVibe.SDK/tar.gz/refs/tags/v0.25.0",
+ "integrity": "sha512-8eurZLyFUtB8XN16hLUuYkah+qwS3tkvdzul5qdZkJTi0aZ4q8dq2QGWSq6pN5mtVPdJyuJ8Mo53r36lixmP1g==",
"license": "MIT",
"license": "MIT",
"engines": {
"engines": {
"node": ">=20"
"node": ">=20"
@@ -1162,9 +1162,9 @@
@@ -1162,9 +1162,9 @@
}
}
},
},
"node_modules/openvibe-shared": {
"node_modules/openvibe-shared": {
- "version": "1.27.0",
- "resolved": "https://codeload.github.com/OpenVibers/OpenVibe.Shared/tar.gz/refs/tags/v1.27.0",
- "integrity": "sha512-/H7dyFw+cURcdjyfbpyY7mNc6F+Pcb9TwkVwZHR+9d3M62q70KDhZLo/gZi71eaYxUjD8lO8tmbXu70EAa9paQ==",
+ "version": "2.0.0",
+ "resolved": "https://codeload.github.com/OpenVibers/OpenVibe.Shared/tar.gz/refs/tags/v2.0.0",
+ "integrity": "sha512-n5CHC0hfxOo0jo/3kS4O3KiwhirhZDlx5zZTIuPyIf3Dq0yI/R6R64btx7NYwZe5dI0NjvqzEtnv19m8unwLMg==",
"license": "MIT",
"license": "MIT",
"dependencies": {
"dependencies": {
"jsonwebtoken": "^9.0.0"
"jsonwebtoken": "^9.0.0"
@@ -26,9 +26,9 @@
@@ -26,9 +26,9 @@
"express": "^4.21.2",
"express": "^4.21.2",
"iovalkey": "^0.4.0",
"iovalkey": "^0.4.0",
"jsonwebtoken": "^9.0.2",
"jsonwebtoken": "^9.0.2",
- "openvibe-contracts": "https://codeload.github.com/OpenVibers/OpenVibe.Contracts/tar.gz/refs/tags/v0.76.0",
- "openvibe-sdk": "https://codeload.github.com/OpenVibers/OpenVibe.SDK/tar.gz/refs/tags/v0.18.0",
- "openvibe-shared": "https://codeload.github.com/OpenVibers/OpenVibe.Shared/tar.gz/refs/tags/v1.27.0",
+ "openvibe-contracts": "https://codeload.github.com/OpenVibers/OpenVibe.Contracts/tar.gz/refs/tags/v0.79.0",
+ "openvibe-sdk": "https://codeload.github.com/OpenVibers/OpenVibe.SDK/tar.gz/refs/tags/v0.25.0",
+ "openvibe-shared": "https://codeload.github.com/OpenVibers/OpenVibe.Shared/tar.gz/refs/tags/v2.0.0",
"pg": "^8.23.0"
"pg": "^8.23.0"
},
},
"engines": {
"engines": {
@@ -20,6 +20,7 @@ const { sql } = require('openvibe-sdk/db');
@@ -20,6 +20,7 @@ const { sql } = require('openvibe-sdk/db');
const { fail, iso, prefixedId, text, positiveInt } = require('../util');
const { fail, iso, prefixedId, text, positiveInt } = require('../util');
const { safeUrl } = require('./profiles');
const { safeUrl } = require('./profiles');
const { publicName, shownName, publicView } = require('./privacy');
const { publicName, shownName, publicView } = require('./privacy');
+const { ENVELOPE } = require('../events/outbox');
function createGoals(ctx) {
function createGoals(ctx) {
const { db } = ctx;
const { db } = ctx;
@@ -102,9 +103,12 @@ function createGoals(ctx) {
@@ -102,9 +103,12 @@ function createGoals(ctx) {
async function changed(t, goalId, reason, { interactionId = null, by = null } = {}) {
async function changed(t, goalId, reason, { interactionId = null, by = null } = {}) {
const g = await t.one(sql`UPDATE tip_goals SET revision = revision + 1, updated_at = ${iso(ctx.now())} WHERE id = ${goalId} RETURNING *`);
const g = await t.one(sql`UPDATE tip_goals SET revision = revision + 1, updated_at = ${iso(ctx.now())} WHERE id = ${goalId} RETURNING *`);
const [view] = await presentMany(t, [g]);
const [view] = await presentMany(t, [g]);
- await ctx.outbox.emit(t, 'tips.goal.updated', { type: 'goal', id: g.id, revision: g.revision }, {
- goal_id: g.id, creator: view.creator, title: g.title, target_amount: g.target_amount, current_amount: view.current_amount,
- currency: g.currency, status: g.status, reached: view.reached, reason, interaction_id: interactionId,
+ await ctx.outbox.emitIn(t, {
+ ...ENVELOPE, event_type: 'tips.goal.updated', subject: { type: 'goal', id: g.id, revision: g.revision },
+ payload: {
+ goal_id: g.id, creator: view.creator, title: g.title, target_amount: g.target_amount, current_amount: view.current_amount,
+ currency: g.currency, status: g.status, reached: view.reached, reason, interaction_id: interactionId,
+ },
});
});
await ctx.overlays.addGoalDelivery(t, g.creator_subject, view, { reason, interactionId, by, dedupe: `goal:${g.id}:r${g.revision}` });
await ctx.overlays.addGoalDelivery(t, g.creator_subject, view, { reason, interactionId, by, dedupe: `goal:${g.id}:r${g.revision}` });
return view;
return view;
@@ -19,7 +19,7 @@
@@ -19,7 +19,7 @@
* ctx.lockCreator(t, s) lock a creator's profile row: settings changes and per-creator limits
* ctx.lockCreator(t, s) lock a creator's profile row: settings changes and per-creator limits
* (20 active goals, 25 tokens, 50 moderators, 10 invitations) take it, so two
* (20 active goals, 25 tokens, 50 moderators, 10 invitations) take it, so two
* requests at once count and write one after the other
* requests at once count and write one after the other
- * ctx.outbox.emit(t, …) a durable event, inside the transaction (openvibe-sdk outbox)
+ * ctx.outbox.emitIn(t, …) a durable event, inside the transaction (openvibe-sdk outbox)
*
*
* Lock order (so two transactions never wait on each other in a cycle): a creator's profile row, then
* Lock order (so two transactions never wait on each other in a cycle): a creator's profile row, then
* tip_interactions rows (SELECT … FOR UPDATE, several by id), then tip_goals rows by id, then the
* tip_interactions rows (SELECT … FOR UPDATE, several by id), then tip_goals rows by id, then the
@@ -55,7 +55,8 @@ function createDomain({ db, config, outbox, billing, adapters, valkey = null, no
@@ -55,7 +55,8 @@ function createDomain({ db, config, outbox, billing, adapters, valkey = null, no
ctx.within = (q, fn, opts) => (inTx(q) ? fn(q) : ctx.tx(fn, opts));
ctx.within = (q, fn, opts) => (inTx(q) ? fn(q) : ctx.tx(fn, opts));
ctx.afterCommit = (q, hook) => { if (inTx(q)) q.after(hook); else runHook(hook); };
ctx.afterCommit = (q, hook) => { if (inTx(q)) q.after(hook); else runHook(hook); };
ctx.lockCreator = (t, subject) => t.maybe(sql`SELECT creator_subject FROM creator_tip_profiles WHERE creator_subject = ${subject} FOR UPDATE`);
ctx.lockCreator = (t, subject) => t.maybe(sql`SELECT creator_subject FROM creator_tip_profiles WHERE creator_subject = ${subject} FOR UPDATE`);
- ctx.outboxKick = () => { try { outbox.kick(); } catch { /* relay off */ } };
+ // The SDK's kick is async: a rejected relay-off throw would be an unhandled rejection on a floating call.
+ ctx.outboxKick = () => { try { Promise.resolve(outbox.kick()).catch(() => {}); } catch { /* relay off */ } };
ctx.profiles = createProfiles(ctx);
ctx.profiles = createProfiles(ctx);
ctx.goals = createGoals(ctx);
ctx.goals = createGoals(ctx);
@@ -47,6 +47,7 @@ const { fail, iso, prefixedId, positiveInt, text, storable, userSubject, entityR
@@ -47,6 +47,7 @@ const { fail, iso, prefixedId, positiveInt, text, storable, userSubject, entityR
const { VOICES } = require('./profiles');
const { VOICES } = require('./profiles');
const { parsePrivacy, publicName, shownName, privacyOf, isAnonymous, publicView, ANONYMOUS } = require('./privacy');
const { parsePrivacy, publicName, shownName, privacyOf, isAnonymous, publicView, ANONYMOUS } = require('./privacy');
const { BillingCallError } = require('../billing-client');
const { BillingCallError } = require('../billing-client');
+const { ENVELOPE } = require('../events/outbox');
const KINDS = ['tip', 'paid_message', 'tts', 'media_request'];
const KINDS = ['tip', 'paid_message', 'tts', 'media_request'];
// Origins whose chat effects Tips delivers: its own requests, and EXTERNAL tips Billing announced
// Origins whose chat effects Tips delivers: its own requests, and EXTERNAL tips Billing announced
@@ -311,7 +312,7 @@ function createInteractions(ctx) {
@@ -311,7 +312,7 @@ function createInteractions(ctx) {
if (i.origin !== 'import' && i.moderation === 'visible') await release(t, i, filter);
if (i.origin !== 'import' && i.moderation === 'visible') await release(t, i, filter);
i = await recomputeDelivery(t, i.id);
i = await recomputeDelivery(t, i.id);
if (!test && i.origin !== 'import') {
if (!test && i.origin !== 'import') {
- await ctx.outbox.emit(t, 'tips.interaction.ready', { type: 'interaction', id: i.id }, summary(i));
+ await ctx.outbox.emitIn(t, { ...ENVELOPE, event_type: 'tips.interaction.ready', subject: { type: 'interaction', id: i.id }, payload: summary(i) });
}
}
if (verdict.hit) await ctx.moderation.recordScreened(t, i, verdict);
if (verdict.hit) await ctx.moderation.recordScreened(t, i, verdict);
ctx.afterCommit(t, () => { ctx.effects.kick(); ctx.outboxKick(); });
ctx.afterCommit(t, () => { ctx.effects.kick(); ctx.outboxKick(); });
@@ -346,7 +347,7 @@ function createInteractions(ctx) {
@@ -346,7 +347,7 @@ function createInteractions(ctx) {
if (!i || i.payment_state !== 'pending') return i;
if (!i || i.payment_state !== 'pending') return i;
const row = await t.one(sql`UPDATE tip_interactions SET payment_state = 'failed', delivery_state = 'cancelled', transfer_due = false,
const row = await t.one(sql`UPDATE tip_interactions SET payment_state = 'failed', delivery_state = 'cancelled', transfer_due = false,
failure = ${`${code}${detail ? `: ${String(detail).slice(0, 240)}` : ''}`}, updated_at = ${iso(ctx.now())} WHERE id = ${i.id} RETURNING *`);
failure = ${`${code}${detail ? `: ${String(detail).slice(0, 240)}` : ''}`}, updated_at = ${iso(ctx.now())} WHERE id = ${i.id} RETURNING *`);
- if (!i.test) await ctx.outbox.emit(t, 'tips.interaction.cancelled', { type: 'interaction', id: i.id }, { ...summary(row), reason: 'payment_failed', code });
+ if (!i.test) await ctx.outbox.emitIn(t, { ...ENVELOPE, event_type: 'tips.interaction.cancelled', subject: { type: 'interaction', id: i.id }, payload: { ...summary(row), reason: 'payment_failed', code } });
return row;
return row;
}
}
@@ -425,7 +426,7 @@ function createInteractions(ctx) {
@@ -425,7 +426,7 @@ function createInteractions(ctx) {
const row = await t.one(sql`UPDATE tip_interactions SET delivery_state = 'cancelled', updated_at = ${at} WHERE id = ${i.id} RETURNING *`);
const row = await t.one(sql`UPDATE tip_interactions SET delivery_state = 'cancelled', updated_at = ${at} WHERE id = ${i.id} RETURNING *`);
await t.exec(sql`UPDATE paid_messages SET status = 'cancelled', updated_at = ${at} WHERE interaction_id = ${i.id} AND status = 'queued'`);
await t.exec(sql`UPDATE paid_messages SET status = 'cancelled', updated_at = ${at} WHERE interaction_id = ${i.id} AND status = 'queued'`);
await t.exec(sql`UPDATE paid_media_requests SET status = 'cancelled', updated_at = ${at} WHERE interaction_id = ${i.id} AND status = 'queued'`);
await t.exec(sql`UPDATE paid_media_requests SET status = 'cancelled', updated_at = ${at} WHERE interaction_id = ${i.id} AND status = 'queued'`);
- if (!now.test) await ctx.outbox.emit(t, 'tips.interaction.cancelled', { type: 'interaction', id: i.id }, { ...summary(row), reason: 'payment_reversed' });
+ if (!now.test) await ctx.outbox.emitIn(t, { ...ENVELOPE, event_type: 'tips.interaction.cancelled', subject: { type: 'interaction', id: i.id }, payload: { ...summary(row), reason: 'payment_reversed' } });
cancelled = true;
cancelled = true;
}
}
return cancelled ? 'reversed_cancelled' : 'reversed';
return cancelled ? 'reversed_cancelled' : 'reversed';
@@ -439,7 +440,7 @@ function createInteractions(ctx) {
@@ -439,7 +440,7 @@ function createInteractions(ctx) {
await t.exec(sql`UPDATE paid_media_requests SET status = 'cancelled', updated_at = ${at} WHERE interaction_id = ${i.id} AND status = 'queued'`);
await t.exec(sql`UPDATE paid_media_requests SET status = 'cancelled', updated_at = ${at} WHERE interaction_id = ${i.id} AND status = 'queued'`);
if (!n) return false;
if (!n) return false;
const row = await t.one(sql`UPDATE tip_interactions SET delivery_state = 'cancelled', updated_at = ${at} WHERE id = ${i.id} RETURNING *`);
const row = await t.one(sql`UPDATE tip_interactions SET delivery_state = 'cancelled', updated_at = ${at} WHERE id = ${i.id} RETURNING *`);
- if (!i.test) await ctx.outbox.emit(t, 'tips.interaction.cancelled', { type: 'interaction', id: i.id }, { ...summary(row), reason });
+ if (!i.test) await ctx.outbox.emitIn(t, { ...ENVELOPE, event_type: 'tips.interaction.cancelled', subject: { type: 'interaction', id: i.id }, payload: { ...summary(row), reason } });
return true;
return true;
}
}
@@ -459,7 +460,7 @@ function createInteractions(ctx) {
@@ -459,7 +460,7 @@ function createInteractions(ctx) {
delivered_at = CASE WHEN ${next === 'delivered'} THEN ${at}::timestamptz ELSE delivered_at END, updated_at = ${at} WHERE id = ${id} RETURNING *`);
delivered_at = CASE WHEN ${next === 'delivered'} THEN ${at}::timestamptz ELSE delivered_at END, updated_at = ${at} WHERE id = ${id} RETURNING *`);
if (next === 'failed' && !i.test) {
if (next === 'failed' && !i.test) {
const failed = effects.filter((e) => e.state === 'failed').map(({ effect, adapter, last_error: lastError }) => ({ effect, adapter, last_error: lastError }));
const failed = effects.filter((e) => e.state === 'failed').map(({ effect, adapter, last_error: lastError }) => ({ effect, adapter, last_error: lastError }));
- await ctx.outbox.emit(t, 'tips.interaction.failed', { type: 'interaction', id }, { ...summary(row), failed_effects: failed });
+ await ctx.outbox.emitIn(t, { ...ENVELOPE, event_type: 'tips.interaction.failed', subject: { type: 'interaction', id }, payload: { ...summary(row), failed_effects: failed } });
}
}
return row;
return row;
}
}
@@ -714,9 +715,12 @@ function createInteractions(ctx) {
@@ -714,9 +715,12 @@ function createInteractions(ctx) {
WHERE envelope->'subject'->>'type' = 'interaction' AND envelope->'subject'->>'id' = ANY(${ids}) AND envelope->'payload' ? 'supporter'`);
WHERE envelope->'subject'->>'type' = 'interaction' AND envelope->'subject'->>'id' = ANY(${ids}) AND envelope->'payload' ? 'supporter'`);
for (const i of done) {
for (const i of done) {
if (i.test) continue;
if (i.test) continue;
- await ctx.outbox.emit(t, 'tips.interaction.erased', { type: 'interaction', id: i.id }, {
- interaction_id: i.id, creator: { type: 'user', id: i.creator_subject }, erased_at: at,
- redacts: { subject_type: 'interaction', subject_ids: [i.id] },
+ await ctx.outbox.emitIn(t, {
+ ...ENVELOPE, event_type: 'tips.interaction.erased', subject: { type: 'interaction', id: i.id },
+ payload: {
+ interaction_id: i.id, creator: { type: 'user', id: i.creator_subject }, erased_at: at,
+ redacts: { subject_type: 'interaction', subject_ids: [i.id] },
+ },
});
});
}
}
t.after(ctx.outboxKick);
t.after(ctx.outboxKick);
@@ -28,6 +28,7 @@ const { sql } = require('openvibe-sdk/db');
@@ -28,6 +28,7 @@ const { sql } = require('openvibe-sdk/db');
const { fail, iso, prefixedId, sha256, userSubject, displayName, text, readCursor, cursorOf } = require('../util');
const { fail, iso, prefixedId, sha256, userSubject, displayName, text, readCursor, cursorOf } = require('../util');
const { publicView } = require('./privacy');
const { publicView } = require('./privacy');
const filter = require('./filter');
const filter = require('./filter');
+const { ENVELOPE } = require('../events/outbox');
const INVITE_RE = /^tmin_[A-Za-z0-9_-]{43}$/;
const INVITE_RE = /^tmin_[A-Za-z0-9_-]{43}$/;
const INVITE_TTL_MS = 7 * 24 * 3600 * 1000;
const INVITE_TTL_MS = 7 * 24 * 3600 * 1000;
@@ -130,9 +131,12 @@ function createModeration(ctx) {
@@ -130,9 +131,12 @@ function createModeration(ctx) {
await t.exec(sql`INSERT INTO tip_moderation_log (creator_subject, interaction_id, action, by_role, actor, reason, created_at)
await t.exec(sql`INSERT INTO tip_moderation_log (creator_subject, interaction_id, action, by_role, actor, reason, created_at)
VALUES (${i.creator_subject}, ${i.id}, ${action}, ${role}, ${actor}, ${reason}, ${iso(ctx.now())})`);
VALUES (${i.creator_subject}, ${i.id}, ${action}, ${role}, ${actor}, ${reason}, ${iso(ctx.now())})`);
if (!i.test) {
if (!i.test) {
- await ctx.outbox.emit(t, 'tips.interaction.moderated', { type: 'interaction', id: i.id }, {
- interaction_id: i.id, creator: { type: 'user', id: i.creator_subject }, action, by: role,
- moderation_state: i.moderation, cancelled_effects: cancelled,
+ await ctx.outbox.emitIn(t, {
+ ...ENVELOPE, event_type: 'tips.interaction.moderated', subject: { type: 'interaction', id: i.id },
+ payload: {
+ interaction_id: i.id, creator: { type: 'user', id: i.creator_subject }, action, by: role,
+ moderation_state: i.moderation, cancelled_effects: cancelled,
+ },
});
});
ctx.afterCommit(t, () => ctx.outboxKick());
ctx.afterCommit(t, () => ctx.outboxKick());
}
}
@@ -28,6 +28,7 @@ const { fail, iso, prefixedId, sha256, text } = require('../util');
@@ -28,6 +28,7 @@ const { fail, iso, prefixedId, sha256, text } = require('../util');
const { safeUrl } = require('./profiles');
const { safeUrl } = require('./profiles');
const { isHidden, publicView } = require('./privacy');
const { isHidden, publicView } = require('./privacy');
const { createOverlayHub } = require('./overlay-hub');
const { createOverlayHub } = require('./overlay-hub');
+const { ENVELOPE } = require('../events/outbox');
const SCOPES = ['alerts', 'goals'];
const SCOPES = ['alerts', 'goals'];
const STREAM_LOCK = 4610; // the advisory-lock namespace of a creator's overlay stream (Tips' port)
const STREAM_LOCK = 4610; // the advisory-lock namespace of a creator's overlay stream (Tips' port)
@@ -247,8 +248,11 @@ function createOverlays(ctx) {
@@ -247,8 +248,11 @@ function createOverlays(ctx) {
WHERE seq = ANY(${pending}) AND status = 'pending' RETURNING id, creator_subject, kind, interaction_id, goal_id, test`);
WHERE seq = ANY(${pending}) AND status = 'pending' RETURNING id, creator_subject, kind, interaction_id, goal_id, test`);
for (const row of done) {
for (const row of done) {
if (row.test) continue;
if (row.test) continue;
- await ctx.outbox.emit(t, 'tips.overlay.delivered', { type: 'overlay_delivery', id: row.id }, {
- delivery_id: row.id, creator: { type: 'user', id: row.creator_subject }, kind: row.kind, interaction_id: row.interaction_id, goal_id: row.goal_id,
+ await ctx.outbox.emitIn(t, {
+ ...ENVELOPE, event_type: 'tips.overlay.delivered', subject: { type: 'overlay_delivery', id: row.id },
+ payload: {
+ delivery_id: row.id, creator: { type: 'user', id: row.creator_subject }, kind: row.kind, interaction_id: row.interaction_id, goal_id: row.goal_id,
+ },
});
});
}
}
if (done.some((r) => !r.test)) t.after(ctx.outboxKick);
if (done.some((r) => !r.test)) t.after(ctx.outboxKick);
@@ -269,9 +273,12 @@ function createOverlays(ctx) {
@@ -269,9 +273,12 @@ function createOverlays(ctx) {
rows.sort((a, b) => a.seq - b.seq);
rows.sort((a, b) => a.seq - b.seq);
for (const row of rows) {
for (const row of rows) {
if (row.test) continue;
if (row.test) continue;
- await ctx.outbox.emit(t, 'tips.overlay.failed', { type: 'overlay_delivery', id: row.id }, {
- delivery_id: row.id, creator: { type: 'user', id: row.creator_subject }, kind: row.kind, interaction_id: row.interaction_id,
- goal_id: row.goal_id, reason: 'no overlay showed it within the delivery window',
+ await ctx.outbox.emitIn(t, {
+ ...ENVELOPE, event_type: 'tips.overlay.failed', subject: { type: 'overlay_delivery', id: row.id },
+ payload: {
+ delivery_id: row.id, creator: { type: 'user', id: row.creator_subject }, kind: row.kind, interaction_id: row.interaction_id,
+ goal_id: row.goal_id, reason: 'no overlay showed it within the delivery window',
+ },
});
});
}
}
return rows.length;
return rows.length;
@@ -6,63 +6,43 @@
@@ -6,63 +6,43 @@
* tips.interaction.ready settled: the interaction is active and its effects are queued
* tips.interaction.ready settled: the interaction is active and its effects are queued
* tips.interaction.failed an effect gave up after its retries (payment state untouched)
* tips.interaction.failed an effect gave up after its retries (payment state untouched)
* tips.interaction.cancelled undelivered effects cancelled (payment reversed/failed, or the creator)
* tips.interaction.cancelled undelivered effects cancelled (payment reversed/failed, or the creator)
+ * tips.interaction.moderated a creator (or a moderator) acted on an interaction
+ * tips.interaction.erased an erasure redacted a supporter's interaction
* tips.goal.updated a goal's settled total, target or status changed
* tips.goal.updated a goal's settled total, target or status changed
* tips.overlay.delivered an overlay delivery reached at least one overlay (first time only)
* tips.overlay.delivered an overlay delivery reached at least one overlay (first time only)
* tips.overlay.failed no overlay received it within the delivery window
* tips.overlay.failed no overlay received it within the delivery window
*
*
- * emit(t, …) runs inside the transaction that makes the change (its handle `t`), so an event exists if
- * and only if its change committed. Simulations (test) never produce events. The relay publishes with
- * Tips' service token (events.event.publish, audience openvibe.events) only when EVENTS_URL and
+ * emitIn(t, …) runs inside the transaction that makes the change (its handle `t`), so an event exists
+ * if and only if its change committed. Simulations (test) never produce events. The relay publishes
+ * with Tips' service token (events.event.publish, audience openvibe.events) only when EVENTS_URL and
* OV_OAUTH_CLIENT_SECRET are set; otherwise rows wait in tips_event_outbox. Any number of processes
* OV_OAUTH_CLIENT_SECRET are set; otherwise rows wait in tips_event_outbox. Any number of processes
* relay the one table: the SDK claims due rows with a lease (FOR UPDATE SKIP LOCKED).
* relay the one table: the SDK claims due rows with a lease (FOR UPDATE SKIP LOCKED).
+ *
+ * Every envelope Tips writes uses ACTOR: the service is the actor (a person's act rides in the payload
+ * as actor_subject), visibility 'internal' and priority 'important' as before.
*/
*/
-const { createClient } = require('openvibe-sdk/core');
-const { createServiceTokenClient } = require('openvibe-sdk/auth');
-const { createEventsClient, createPgOutbox } = require('openvibe-sdk/events');
+const { createServiceOutbox } = require('openvibe-sdk/events');
-const ACTOR = { type: 'service', id: 'tips' };
const TABLE = 'tips_event_outbox'; // migrations/0001_initial.sql (outboxSchema)
const TABLE = 'tips_event_outbox'; // migrations/0001_initial.sql (outboxSchema)
+const ACTOR = { type: 'service', id: 'tips' };
+/** The defaults every Tips envelope carries (the SDK's createServiceOutbox sets none). */
+const ENVELOPE = { actor: ACTOR, visibility: 'internal', priority: 'important' };
+/** Tips' wiring of the SDK's shared outbox (plan T1). */
function createTipsOutbox({ db, config, fetchImpl, now, log = console }) {
function createTipsOutbox({ db, config, fetchImpl, now, log = console }) {
- const enabled = !!(config.events.url && config.oauth.clientSecret);
- const clientOpts = { baseUrls: { events: config.events.url || 'http://127.0.0.1:4300' }, retries: 0 };
- if (fetchImpl) clientOpts.fetch = fetchImpl;
- if (enabled) {
- clientOpts.tokenProvider = createServiceTokenClient({
- tokenUrl: `${config.network.internalUrl}/oauth/token`, clientId: config.oauth.clientId, clientSecret: config.oauth.clientSecret,
- scope: { 'openvibe.events': 'events.event.publish' }, ...(fetchImpl ? { fetch: fetchImpl } : {}),
- });
- } else {
- clientOpts.getToken = async () => { throw new Error('events relay disabled (EVENTS_URL / OV_OAUTH_CLIENT_SECRET unset)'); };
- }
- const events = createEventsClient(createClient(clientOpts), { source: 'tips' });
- let lastError = null;
- const outbox = createPgOutbox(db, {
- events,
+ return createServiceOutbox({
+ db,
+ source: 'tips',
table: TABLE,
table: TABLE,
+ eventsUrl: config.events.url,
+ networkInternalUrl: config.network.internalUrl,
+ clientId: config.oauth.clientId,
+ clientSecret: config.oauth.clientSecret,
intervalMs: config.events.intervalMs,
intervalMs: config.events.intervalMs,
- now,
- onError: (err) => {
- const msg = err && err.message;
- if (msg !== lastError) log.warn('[Tips] event publish failed (will retry):', msg);
- lastError = msg;
- },
+ log,
+ ...(fetchImpl ? { fetch: fetchImpl } : {}),
+ ...(now ? { now } : {}),
});
});
-
- /** Inside the caller's transaction `t`. */
- function emit(t, eventType, subject, payload, { visibility = 'internal', priority = 'important', traceparent } = {}) {
- return outbox.enqueue(t, { event_type: eventType, actor: ACTOR, subject, payload, visibility, priority }, { traceparent });
- }
-
- return {
- emit,
- outbox,
- enabled,
- start() { if (enabled) outbox.start(); },
- stop: () => outbox.stop(),
- kick() { if (enabled) outbox.kick(); },
- status: async () => ({ enabled, pending: await outbox.pending(), rejected: await outbox.rejected(), last_error: lastError }),
- };
}
}
-module.exports = { createTipsOutbox, TABLE };
+module.exports = { createTipsOutbox, TABLE, ACTOR, ENVELOPE };