Changeset 5a6a457 in Klonkt

Timestamp:

06/25/2026 05:15:48 AM (3 months ago)

Author:

Robin Genis <roboburr@…>

Branches:

main

Children:

d7526bd

Parents:

99234c3

Message:

feat(fediverse): federate on save+scheduler, delivery retry-queue, music as listen-link

#1 Posts now federate to followers not only on create, but also when a draft/

scheduled post becomes published (editor save) and when the Scheduler flips a
scheduled post live — previously those silently didn't reach followers.

#2 Delivery retry-queue (ap_delivery): a failed delivery (down server/timeout) is

queued and retried with backoff (1/5/15/60/180/360 min, 6 tries) by a worker,
instead of fire-and-forget. Signing key re-derived from the slug, never stored.

#3 Music posts: audio shortcodes federate as a '🎵 listen on the site' link to the

post (protected player) instead of the raw mp3 — keeps Klonkt's audio friction
intact (no downloadable file handed to Mastodon).

Co-Authored-By: Claude <noreply@…>

Location:

src

Files:

• config/database.js (modified) (1 diff)
• routes/posts.js (modified) (1 diff)
• server.js (modified) (2 diffs)
• services/ActivityPubService.js (modified) (5 diffs)
• services/Scheduler.js (modified) (2 diffs)

src/config/database.js

@@ -389 +389 @@
     );
     CREATE INDEX IF NOT EXISTS idx_ap_blocks_target ON ap_blocks(target);
+    CREATE TABLE IF NOT EXISTS ap_delivery (
+      id INTEGER PRIMARY KEY AUTOINCREMENT,
+      slug TEXT NOT NULL,          -- our site/actor that signs the delivery
+      inbox TEXT NOT NULL,         -- recipient inbox URL
+      body TEXT NOT NULL,          -- the activity JSON to POST
+      attempts INTEGER NOT NULL DEFAULT 0,
+      next_at DATETIME NOT NULL DEFAULT CURRENT_TIMESTAMP,
+      created_at DATETIME DEFAULT CURRENT_TIMESTAMP
+    );
+    CREATE INDEX IF NOT EXISTS idx_ap_delivery_due ON ap_delivery(next_at);
   `);
 }

src/routes/posts.js

@@ -353 +353 @@
     }
   } catch (e) { /* FTS issues non-fatal */ }
+
+  // ActivityPub: federate when a post BECOMES published (draft/scheduled → published
+  // via the editor). A brand-new published post is handled in the create route.
+  if (finalStatus === 'published' && post.status !== 'published' && !fanOnly) {
+    ActivityPubService.deliverCreate(site, {
+      id: post.id, slug: finalSlug, title: title || finalSlug,
+      content: cleanContent, cover_image_url: cover_image_url || null,
+      published_at: publishedAt, created_at: post.created_at,
+    }).catch(() => { /* best-effort */ });
+  }
 
   res.redirect(`${res.locals.siteUrlBase || ''}/${finalSlug}`);

src/server.js

@@ -61 +61 @@
 import ogRoutes from './routes/og.js';
 import apRoutes from './routes/activitypub.js';
-import { apWants } from './services/ActivityPubService.js';
+import { apWants, startDeliveryWorker } from './services/ActivityPubService.js';
 
 // SESSION_SECRET: use the env var if set. Otherwise auto-generate a strong one
@@ -163 +163 @@
 initializeDatabase();
 startScheduler(); // release planning: publish scheduled posts when publish_at is reached
+startDeliveryWorker(); // retry failed fediverse deliveries with backoff
 
 // Safety net: guarantee that there is always a primary site (solo/hub/circle).

src/services/ActivityPubService.js

@@ -121 +121 @@
   for (const m of body.matchAll(/<img\b[^>]*\bsrc="([^"]+)"[^>]*>/gi)) urls.push(abs(m[1]));
   body = body.replace(/<img\b[^>]*>/gi, '');
-  // Strip Klonkt audio shortcodes ([[track:…]] etc.) — they'd federate raw as
-  // ugly text (audio federation itself is a later phase).
+  // Audio shortcodes: do NOT federate the raw audio file — Klonkt deliberately
+  // gates audio (the /audio/stream URL has friction), and shipping it as an AP
+  // audio attachment would hand Mastodon a plain, downloadable mp3 URL. Instead,
+  // replace the shortcodes with a "🎵 listen on the site" link so the post invites
+  // a click-through to the protected player (discovery without leaking the file).
+  const esc = (s) => String(s == null ? '' : s).replace(/[<>&]/g, (c) => ({ '<': '&lt;', '>': '&gt;', '&': '&amp;' }[c]));
+  const audioLabels = [];
+  try {
+    for (const m of body.matchAll(/\[\[track:([A-Za-z0-9_-]+)\]\]/g)) { const r = db.prepare('SELECT title FROM audio_tracks WHERE id = ?').get(m[1]); if (r && r.title) audioLabels.push(r.title); }
+    for (const m of body.matchAll(/\[\[album:([^\]]+)\]\]/g)) audioLabels.push(m[1].trim());
+  } catch { /* non-fatal */ }
+  const hadAudio = /\[\[(track|album|playlist):/i.test(body);
   body = body.replace(/\[\[(track|album|playlist):[^\]]+\]\]/gi, '');
+  if (hadAudio) {
+    const lbl = audioLabels.length ? esc(audioLabels.slice(0, 4).join(', ')) : '';
+    body += `<p>🎵 ${lbl ? `<strong>${lbl}</strong> — ` : ''}<a href="${human}">listen on ${esc(site.title || 'the site')}</a></p>`;
+  }
   const seen = new Set();
   const attachment = urls.filter(Boolean)
@@ -328 +342 @@
     return await r.json();
   } catch { return null; }
+}
+
+// ── Delivery queue with retries ───────────────────────────────────
+// Outbound deliveries are tried immediately; on failure (down server, timeout,
+// non-2xx) they're queued and retried with backoff so a briefly-offline follower
+// doesn't silently miss the post. The signing key is NOT stored — the worker
+// re-derives it from the actor slug at send time.
+const DELIVERY_MAX_ATTEMPTS = 6;
+const DELIVERY_BACKOFF_MIN = [1, 5, 15, 60, 180, 360];
+let _insDeliv, _dueDeliv, _delDeliv, _bumpDeliv;
+function deliveryStmts() {
+  if (!_insDeliv) {
+    _insDeliv = db.prepare('INSERT INTO ap_delivery (slug, inbox, body, attempts, next_at) VALUES (?,?,?,0,CURRENT_TIMESTAMP)');
+    _dueDeliv = db.prepare("SELECT * FROM ap_delivery WHERE datetime(next_at) <= datetime('now') ORDER BY next_at LIMIT 30");
+    _delDeliv = db.prepare('DELETE FROM ap_delivery WHERE id = ?');
+    _bumpDeliv = db.prepare('UPDATE ap_delivery SET attempts = ?, next_at = ? WHERE id = ?');
+  }
+  return { ins: _insDeliv, due: _dueDeliv, del: _delDeliv, bump: _bumpDeliv };
+}
+export function enqueueDelivery(slug, inbox, activity) {
+  if (!slug || !inbox || !activity) return;
+  try { deliveryStmts().ins.run(slug, inbox, JSON.stringify(activity)); } catch { /* ignore */ }
+}
+// Deliver now; queue for retry if it fails.
+export async function deliverWithRetry(slug, inbox, activity, keyId, privPem) {
+  if (!inbox) return;
+  try { const st = await deliver(inbox, activity, keyId, privPem); if (st >= 200 && st < 300) return; } catch { /* queue below */ }
+  enqueueDelivery(slug, inbox, activity);
+}
+export async function processDeliveryQueue() {
+  let rows;
+  try { rows = deliveryStmts().due.all(); } catch { return; }
+  if (!rows || !rows.length) return;
+  const base = (process.env.PUBLIC_BASE_URL || '').replace(/\/+$/, '');
+  for (const row of rows) {
+    let ok = false;
+    try {
+      const keys = getOrCreateKeys(row.slug);
+      const st = await deliver(row.inbox, JSON.parse(row.body), `${actorId(base, row.slug)}#main-key`, keys.private_pem);
+      ok = st >= 200 && st < 300;
+    } catch { ok = false; }
+    if (ok) { deliveryStmts().del.run(row.id); continue; }
+    const attempts = row.attempts + 1;
+    if (attempts >= DELIVERY_MAX_ATTEMPTS) { deliveryStmts().del.run(row.id); console.warn('[AP] delivery gave up after', attempts, 'tries →', row.inbox); continue; }
+    const mins = DELIVERY_BACKOFF_MIN[Math.min(attempts, DELIVERY_BACKOFF_MIN.length - 1)];
+    deliveryStmts().bump.run(attempts, new Date(Date.now() + mins * 60000).toISOString(), row.id);
+  }
+}
+let _delivTimer = null;
+export function startDeliveryWorker() {
+  if (_delivTimer) return;
+  _delivTimer = setInterval(() => { processDeliveryQueue().catch(() => {}); }, 60 * 1000);
+  if (_delivTimer.unref) _delivTimer.unref();
 }
 
@@ -479 +546 @@
   const keyId = `${actorId(base, site.slug)}#main-key`;
   const create = buildCreate(base, site, post);
-  for (const inbox of inboxes) deliver(inbox, create, keyId, keys.private_pem).catch(() => { /* best-effort */ });
+  for (const inbox of inboxes) deliverWithRetry(site.slug, inbox, create, keyId, keys.private_pem);
 }
 
@@ -500 +567 @@
     object: { id: nid, type: 'Tombstone' },
   };
-  for (const inbox of inboxes) deliver(inbox, del, `${me}#main-key`, keys.private_pem).catch(() => { /* best-effort */ });
+  for (const inbox of inboxes) deliverWithRetry(site.slug, inbox, del, `${me}#main-key`, keys.private_pem);
 }
 
@@ -838 +905 @@
   webfingerResolve, followActor, unfollowActor, listFollowing, getTimeline, sendInteraction,
   getNotifications, listBlocks, isBlockedAny, blockTarget, unblock,
+  deliverWithRetry, enqueueDelivery, processDeliveryQueue, startDeliveryWorker,
 };

src/services/Scheduler.js

@@ -10 +10 @@
 import db from '../config/database.js';
 import HtmlSanitizerService from './HtmlSanitizerService.js';
+import ActivityPubService from './ActivityPubService.js';
 
 export function flipScheduledPosts() {
   try {
     const due = db.prepare(`
-      SELECT p.id, p.title, p.content, u.username
+      SELECT p.id, p.site_id, p.slug, p.title, p.content, p.cover_image_url, p.fan_only,
+             p.published_at, p.publish_at, p.created_at, u.username
       FROM posts p JOIN users u ON u.id = p.author_id
       WHERE p.status = 'scheduled' AND p.publish_at IS NOT NULL AND datetime(p.publish_at) <= datetime('now')
@@ -23 +25 @@
     );
     const fts = db.prepare('INSERT INTO posts_fts(content, title, author, post_id) VALUES (?, ?, ?, ?)');
+    const siteStmt = db.prepare('SELECT * FROM sites WHERE id = ?');
     for (const p of due) {
       upd.run(p.id);
       try { fts.run(HtmlSanitizerService.toPlainText(p.content || ''), p.title || '', p.username || '', p.id); } catch { /* FTS failure is non-fatal */ }
+      // ActivityPub: federate the now-published post to followers.
+      if (!p.fan_only) {
+        try {
+          const site = siteStmt.get(p.site_id);
+          if (site) {
+            ActivityPubService.deliverCreate(site, {
+              id: p.id, slug: p.slug, title: p.title || p.slug,
+              content: p.content, cover_image_url: p.cover_image_url || null,
+              published_at: p.published_at || p.publish_at, created_at: p.created_at,
+            }).catch(() => { /* best-effort */ });
+          }
+        } catch { /* non-fatal */ }
+      }
     }
     return due.length;