Index: src/routes/activitypub.js =================================================================== --- src/routes/activitypub.js (revision 58618499cb0e921606ed135944e69fa0deba1627) +++ src/routes/activitypub.js (revision 4f322bcbd9476e63f3ae5ef8d9487a47535b2f8b) @@ -268,8 +268,26 @@ // they follow) as Create(Note) items, so an app (Shaer) can build a unified // feed. Anyone else gets 403; the inbox stays write-only for the public. -router.get('/ap/users/:slug/inbox', (req, res) => { +router.get('/ap/users/:slug/inbox', async (req, res) => { const auth = OAuth.verifyBearer(req.headers.authorization); if (!auth || auth.site.slug !== req.params.slug) return res.status(403).end(); const base = baseUrl(req); + // Wachten is een UITBREIDING van deze lezing, geen tweede endpoint (shaer-n05). + // Geef `since` (de shaer:cursor van je vorige antwoord) en `wait` mee, en het + // antwoord blijft hangen tot er iets is of de tijd om is. Zonder die twee + // gedraagt de route zich exact zoals altijd. + // + // Bewust hetzelfde antwoord in plaats van een "er is nieuws"-seintje: dan + // hoeft er niets nieuws geparsed te worden, is er geen tweede beschrijving van + // de kaartvorm die uit de pas kan lopen, en scheelt het de client een tweede + // ronde. + const wachtS = Math.min(Math.max(parseInt(req.query.wait, 10) || 0, 0), 50); + if (req.query.since && wachtS > 0) { + const afbreken = new AbortController(); + res.on('close', () => afbreken.abort()); // client hing op: niet doorgaan met wachten + await AP.waitForFeedChange(auth.site.slug, { + since: String(req.query.since), waitMs: wachtS * 1000, signal: afbreken.signal, + }); + if (res.writableEnded || afbreken.signal.aborted) return undefined; + } // Gated feature (FEP-633c): may this account see EXTERNAL embeds? A ward's // world outside the fediverse is the guardians' call. The gate is applied @@ -444,7 +462,12 @@ 'shaer:externalLinks': playbackAllowed, }, + // Het merk van wat hierin zit. Geef hem terug als `since` om op het + // volgende te wachten. NA het samenstellen bepaald, zodat hij precies dekt + // wat je in handen hebt en niet iets dat er ondertussen bij kwam. + 'shaer:cursor': AP.feedCursor(auth.site.slug), totalItems: items.length, orderedItems: items, }); + return undefined; }); Index: src/services/ActivityPubService.js =================================================================== --- src/services/ActivityPubService.js (revision 58618499cb0e921606ed135944e69fa0deba1627) +++ src/services/ActivityPubService.js (revision 4f322bcbd9476e63f3ae5ef8d9487a47535b2f8b) @@ -3485,4 +3485,74 @@ } +/** + * Een merk voor "is er iets veranderd aan wat de inbox-lezing zou opleveren?" + * (shaer-n05). + * + * Alle VIER de poten die de inbox samenvoegt tellen mee -- tijdlijn, berichten, + * antwoorden op je eigen posts, en wat je zelf verstuurde. Zou er een ontbreken, + * dan blijft een wachtende client slapen terwijl er wel degelijk iets is + * bijgekomen, en dat is erger dan niet wachten: het lijkt te werken. + * + * rowid en niet een tijdstempel: rowid loopt strikt op per invoeging, terwijl + * twee dingen in dezelfde seconde kunnen aankomen en een `published` van een + * andere server niet te vertrouwen is. + * + * Ondoorzichtig voor de client. Hij krijgt hem terug en geeft hem ongewijzigd + * mee; de vorm mag veranderen zonder dat dat iets breekt. + */ +export function feedCursor(slug) { + const max = (sql, ...args) => { try { const r = db.prepare(sql).get(...args); return (r && r.n) || 0; } catch { return 0; } }; + const t = max('SELECT MAX(rowid) AS n FROM ap_timeline WHERE slug = ?', slug); + const m = max('SELECT MAX(rowid) AS n FROM ap_mentions WHERE slug = ?', slug); + const o = max('SELECT MAX(rowid) AS n FROM ap_outbox WHERE site_slug = ?', slug); + const r = max(`SELECT MAX(i.rowid) AS n FROM ap_interactions i + JOIN posts p ON p.id = i.post_id + JOIN sites s ON s.id = p.site_id + WHERE s.slug = ? AND i.kind = 'reply'`, slug); + return `${t}.${m}.${o}.${r}`; +} + +// Zoveel clients mogen er tegelijk op EEN account staan wachten. Een client met +// een kapotte herverbind-lus mag de instance niet vastzetten; de overtolligen +// krijgen gewoon meteen antwoord in plaats van een fout. +const FEED_WAIT_MAX = 4; +const _wachters = new Map(); + +/** + * Wacht tot de inbox-lezing iets anders zou opleveren dan bij `since`. + * + * Bewust met een interne tik en niet met een gebeurtenis-emitter. Een emitter + * moet op ELKE plek worden aangeroepen waar er iets bijkomt, en de plek die je + * vergeet is precies de melding die nooit aankomt. Twee tot vier MAX(rowid)- + * queries per seconde is niets, en dit kan niets missen. Prijs: hooguit een tik + * vertraging. + */ +export async function waitForFeedChange(slug, opts = {}) { + const tickMs = Math.max(50, opts.tickMs || 1000); + const waitMs = Math.max(0, opts.waitMs || 0); + const since = String(opts.since || ''); + let cursor = feedCursor(slug); + // Geen sinds, al iets veranderd, of niet willen wachten: meteen antwoorden. + if (!since || since !== cursor || !waitMs) return { cursor, changed: !!since && since !== cursor, waited: false }; + + const bezet = _wachters.get(slug) || 0; + if (bezet >= FEED_WAIT_MAX) return { cursor, changed: false, waited: false, busy: true }; + _wachters.set(slug, bezet + 1); + try { + const einde = Date.now() + waitMs; + while (Date.now() < einde) { + if (opts.signal && opts.signal.aborted) break; // client hing op + const rest = Math.min(tickMs, einde - Date.now()); + await new Promise((r) => setTimeout(r, rest)); + cursor = feedCursor(slug); + if (cursor !== since) return { cursor, changed: true, waited: true }; + } + return { cursor, changed: false, waited: true }; + } finally { + const n = (_wachters.get(slug) || 1) - 1; + if (n > 0) _wachters.set(slug, n); else _wachters.delete(slug); + } +} + export function getDirectMessages(slug, limit) { try { @@ -5254,4 +5324,5 @@ buildActor, buildNote, buildCreate, buildOutbox, buildFollowers, buildFollowing, buildFeatured, followerCount, deliver, fetchActor, verifyRequest, handleInbox, deliverCreate, deliverDelete, deliverUpdate, deliverActorUpdate, resyncFeaturedPins, + feedCursor, waitForFeedChange, getInteractions, getInteractionById, setInteractionBoosted, setInteractionLiked, buildReplyNote, getOutboxNote, getSentNotes, deliverReply, resolveRemoteNote, noteAudience, mayReadNote, listOutbox, deliverOutboxDelete, deliverOutboxUpdate, deliverDirectNote, Index: test/feed-wait.test.js =================================================================== --- test/feed-wait.test.js (revision 4f322bcbd9476e63f3ae5ef8d9487a47535b2f8b) +++ test/feed-wait.test.js (revision 4f322bcbd9476e63f3ae5ef8d9487a47535b2f8b) @@ -0,0 +1,131 @@ +// Wachten op nieuws, als uitbreiding van de inbox-lezing (shaer-n05). +// +// Geen tweede endpoint en geen seintje-formaat: dezelfde lezing, die desgevraagd +// even blijft hangen. Wat hier bewaakt wordt is de MERKSTEEN -- als die niet +// beweegt bij iets dat de inbox wel zou tonen, blijft een wachtende client +// slapen terwijl er nieuws is. Dat is erger dan niet wachten, want het lijkt te +// werken. +// +// Run: npm test + +import { test } from 'node:test'; +import assert from 'node:assert/strict'; + +process.env.DATABASE_PATH = ':memory:'; +process.env.PUBLIC_BASE_URL = 'https://klonkt.test'; + +const dbMod = await import('../src/config/database.js'); +const db = dbMod.default; +dbMod.initializeDatabase(); +const AP = await import('../src/services/ActivityPubService.js'); + +db.prepare('INSERT INTO users (id, username, email, password_hash, role) VALUES (?,?,?,?,?)') + .run('u1', 'u1', 'u1@t', 'x', 'god'); +db.prepare('INSERT INTO sites (id, slug, title, owner_id) VALUES (?,?,?,?)').run('s1', 'me', 'Me', 'u1'); +db.prepare('INSERT INTO sites (id, slug, title, owner_id) VALUES (?,?,?,?)').run('s2', 'buur', 'Buur', 'u1'); +db.prepare('INSERT INTO posts (id, site_id, slug, author_id, title, content, status) VALUES (?,?,?,?,?,?,?)') + .run('p1', 's1', 'mijn-post', 'u1', 'Post', '

x

', 'published'); + +let n = 0; +const nieuwePost = (slug = 'me') => db.prepare( + "INSERT INTO ap_timeline (id, slug, author_uri, content) VALUES (?,?,'https://r.test/u/a','

x

')", +).run(`https://r.test/n/${++n}`, slug); + +test('een verse tijdlijnpost beweegt de merksteen', () => { + const voor = AP.feedCursor('me'); + nieuwePost(); + assert.notEqual(AP.feedCursor('me'), voor); +}); + +test('en zo ook de andere drie poten van de inbox', () => { + // De inbox voegt vier bronnen samen. Ontbreekt er een in de merksteen, dan + // wordt een client niet wakker van precies dat soort nieuws. + const bericht = () => db.prepare( + "INSERT INTO ap_mentions (slug, object_uri, actor_uri, content) VALUES ('me',?, 'https://r.test/u/a','

hoi

')", + ).run(`https://r.test/m/${++n}`); + const antwoord = () => db.prepare( + `INSERT INTO ap_interactions (kind, post_id, object_uri, actor_uri, content) + VALUES ('reply','p1',?, 'https://r.test/u/a','

hoi

')`, + ).run(`https://r.test/r/${++n}`); + const verstuurd = () => db.prepare( + "INSERT INTO ap_outbox (id, site_slug, post_id, content) VALUES (?,'me','p1','

x

')", + ).run(`o${++n}`); + + for (const [naam, doen] of [['bericht', bericht], ['antwoord', antwoord], ['verstuurd', verstuurd]]) { + const voor = AP.feedCursor('me'); + doen(); + assert.notEqual(AP.feedCursor('me'), voor, `${naam} hoort de merksteen te bewegen`); + } +}); + +test('nieuws van een ANDER account laat je merksteen staan', () => { + const voor = AP.feedCursor('me'); + nieuwePost('buur'); + assert.equal(AP.feedCursor('me'), voor, 'anders wordt iedereen wakker van andermans post'); +}); + +test('is er al iets veranderd, dan wordt er niet gewacht', async () => { + const oud = AP.feedCursor('me'); + nieuwePost(); + const begin = Date.now(); + const uit = await AP.waitForFeedChange('me', { since: oud, waitMs: 5000, tickMs: 20 }); + assert.equal(uit.changed, true); + assert.equal(uit.waited, false, 'meteen antwoorden, niet eerst de tijd volmaken'); + assert.ok(Date.now() - begin < 500); +}); + +test('zonder sinds is het gewoon de huidige stand', async () => { + const uit = await AP.waitForFeedChange('me', { waitMs: 5000, tickMs: 20 }); + assert.equal(uit.changed, false); + assert.equal(uit.waited, false); + assert.equal(uit.cursor, AP.feedCursor('me')); +}); + +test('wachten wordt afgebroken zodra er iets binnenkomt', async () => { + const oud = AP.feedCursor('me'); + const begin = Date.now(); + setTimeout(() => nieuwePost(), 60); + const uit = await AP.waitForFeedChange('me', { since: oud, waitMs: 4000, tickMs: 20 }); + assert.equal(uit.changed, true); + assert.equal(uit.waited, true); + assert.ok(Date.now() - begin < 2000, 'hij hoort wakker te worden, niet de tijd vol te maken'); + assert.notEqual(uit.cursor, oud, 'en de nieuwe merksteen komt mee'); +}); + +test('gebeurt er niets, dan komt hij leeg terug binnen de tijd', async () => { + const oud = AP.feedCursor('me'); + const begin = Date.now(); + const uit = await AP.waitForFeedChange('me', { since: oud, waitMs: 150, tickMs: 20 }); + assert.equal(uit.changed, false); + assert.equal(uit.waited, true); + assert.equal(uit.cursor, oud); + assert.ok(Date.now() - begin >= 100, 'hij hoort wel echt gewacht te hebben'); + assert.ok(Date.now() - begin < 3000, 'en niet langer dan gevraagd'); +}); + +test('ophangen breekt het wachten af', async () => { + // Zonder dit blijft er een timer draaien voor een client die er niet meer is. + const oud = AP.feedCursor('me'); + const ac = new AbortController(); + const begin = Date.now(); + setTimeout(() => ac.abort(), 60); + const uit = await AP.waitForFeedChange('me', { since: oud, waitMs: 4000, tickMs: 20, signal: ac.signal }); + assert.equal(uit.changed, false); + assert.ok(Date.now() - begin < 2000); +}); + +test('meer dan vier wachters tegelijk: de rest krijgt meteen antwoord', async () => { + // Een client met een kapotte herverbind-lus mag de instance niet vastzetten. + // De overtolligen krijgen geen fout maar gewoon de huidige stand. + const oud = AP.feedCursor('me'); + const lopend = []; + for (let i = 0; i < 4; i++) lopend.push(AP.waitForFeedChange('me', { since: oud, waitMs: 300, tickMs: 20 })); + await new Promise((r) => setTimeout(r, 30)); + const vijfde = await AP.waitForFeedChange('me', { since: oud, waitMs: 4000, tickMs: 20 }); + assert.equal(vijfde.busy, true); + assert.equal(vijfde.waited, false); + await Promise.all(lopend); + // En daarna is er weer plek: de teller moet netjes teruglopen. + const daarna = await AP.waitForFeedChange('me', { since: oud, waitMs: 60, tickMs: 20 }); + assert.notEqual(daarna.busy, true); +});