Index: src/config/database.js =================================================================== --- src/config/database.js (revision 99234c3d33adbc87dfe293138b8048497840d9a1) +++ src/config/database.js (revision 5a6a457ca2bf625de84ec03d20f7757480337fde) @@ -389,4 +389,14 @@ ); 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); `); } Index: src/routes/posts.js =================================================================== --- src/routes/posts.js (revision 99234c3d33adbc87dfe293138b8048497840d9a1) +++ src/routes/posts.js (revision 5a6a457ca2bf625de84ec03d20f7757480337fde) @@ -353,4 +353,14 @@ } } 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}`); Index: src/server.js =================================================================== --- src/server.js (revision 99234c3d33adbc87dfe293138b8048497840d9a1) +++ src/server.js (revision 5a6a457ca2bf625de84ec03d20f7757480337fde) @@ -61,5 +61,5 @@ 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,4 +163,5 @@ 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). Index: src/services/ActivityPubService.js =================================================================== --- src/services/ActivityPubService.js (revision 99234c3d33adbc87dfe293138b8048497840d9a1) +++ src/services/ActivityPubService.js (revision 5a6a457ca2bf625de84ec03d20f7757480337fde) @@ -121,7 +121,21 @@ for (const m of body.matchAll(/]*\bsrc="([^"]+)"[^>]*>/gi)) urls.push(abs(m[1])); body = body.replace(/]*>/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) => ({ '<': '<', '>': '>', '&': '&' }[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 += `

🎵 ${lbl ? `${lbl} — ` : ''}listen on ${esc(site.title || 'the site')}

`; + } const seen = new Set(); const attachment = urls.filter(Boolean) @@ -328,4 +342,57 @@ 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,5 +546,5 @@ 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,5 +567,5 @@ 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,3 +905,4 @@ webfingerResolve, followActor, unfollowActor, listFollowing, getTimeline, sendInteraction, getNotifications, listBlocks, isBlockedAny, blockTarget, unblock, + deliverWithRetry, enqueueDelivery, processDeliveryQueue, startDeliveryWorker, }; Index: src/services/Scheduler.js =================================================================== --- src/services/Scheduler.js (revision 99234c3d33adbc87dfe293138b8048497840d9a1) +++ src/services/Scheduler.js (revision 5a6a457ca2bf625de84ec03d20f7757480337fde) @@ -10,9 +10,11 @@ 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,7 +25,21 @@ ); 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;