From f14cc37f4c8ccfd07d92cee929c26e0f8130593e Mon Sep 17 00:00:00 2001 From: Sylvain Duchesne Date: Mon, 12 Oct 2026 17:08:24 +0200 Subject: [PATCH] List only Festipod-public events in the catalog, and only let an owner change their entry The catalog indexed every as:Event written to any user's outbox and served private bodies (other apps' events, friend-addressed ones); since the index is persistent, they survived restarts. It now lists only objects addressed to the Public collection in to (cc-only is unlisted: readable, never listed), using the same predicate module as the public-read grant; stored entries that do not qualify are filtered at every read and swept at startup by a conditional delete. A create, update or delete is applied only when the object lies under the pod of the actor whose outbox emitted it (same origin, path prefix with a / boundary): one user could previously remove another's entry with a Delete carrying an embedded Tombstone. Catalog handlers never throw into the webhook path, whose listener does not await them. --- backend/services/actor-pod.js | 55 ++++++ backend/services/actor-pod.test.js | 56 ++++++ backend/services/catalog-store.js | 69 +++++-- backend/services/catalog-store.test.js | 51 ++++- backend/services/catalog.service.js | 136 +++++++++++-- backend/services/catalog.service.test.js | 237 +++++++++++++++++++++++ 6 files changed, 571 insertions(+), 33 deletions(-) create mode 100644 backend/services/actor-pod.js create mode 100644 backend/services/actor-pod.test.js create mode 100644 backend/services/catalog.service.test.js diff --git a/backend/services/actor-pod.js b/backend/services/actor-pod.js new file mode 100644 index 0000000..a5ae07d --- /dev/null +++ b/backend/services/actor-pod.js @@ -0,0 +1,55 @@ +// Whether a resource lies in an actor's OWN pod — the ownership test in front of any mutation the +// backend applies because of an activity an actor emitted (the catalog's index/update/de-index). +// +// THE TRUST ANCHOR is the pod provider's namespace. An ActivityPods actor's URI is its pod root +// (`http://localhost:3000/alice`) and every resource of that pod lives under it +// (`http://localhost:3000/alice/data/…`); the pod lets no one create or modify a resource under +// another actor's path (observed 2026-10-12: an outbox `Create` or `Update` carrying another pod's +// resource is refused, 403 "Remote resource(s) cannot be created/modified"). The platform's own +// "is this resource on the actor's pod" test is the same path-prefix rule (`isLocal` in its +// fetch-pod-or-proxy mixin). +// +// NOT the actor document's `pim:storage`: it does declare the storage (`…/alice/data`), but that +// document is writable by its owner — a user can add another user's storage to her own +// `pim:storage` (verified 2026-10-12: PATCH → 204, the profile then lists both) — so it proves +// nothing about who owns a resource. +// +// `actorUri` must come from the server side — the actor whose outbox the webhook fired on — never +// from the activity's own `actor` field, which the emitting client writes. + +const WEB_PROTOCOLS = new Set(['http:', 'https:']); + +/** + * The URL when `value` is an http(s) URL written in canonical form, else undefined. Requiring the + * canonical form means no `..` / `%2e%2e` segment, host-case, default-port or other variant can + * read as one path to this test and as another to whoever later resolves it. + */ +function canonicalUrl(value) { + if (typeof value !== 'string') return undefined; + let url; + try { + url = new URL(value); + } catch { + return undefined; + } + return WEB_PROTOCOLS.has(url.protocol) && url.href === value ? url : undefined; +} + +/** + * isInActorPod(uri, actorUri) → boolean. True only when `uri` is strictly under the actor's pod + * root: same origin, and a path that continues the actor's path with a `/` and at least one more + * character (so `/alice2/…` is not under `/alice`). False for anything else, including a missing or + * non-canonical argument, the actor URI itself (with or without a trailing slash), and an actor at + * the origin root (it would own the whole host). + */ +function isInActorPod(uri, actorUri) { + const resource = canonicalUrl(uri); + const actor = canonicalUrl(actorUri); + if (!resource || !actor) return false; + const root = actor.pathname.replace(/\/+$/, ''); + if (root.length === 0) return false; + const prefix = `${root}/`; + return resource.origin === actor.origin && resource.pathname.startsWith(prefix) && resource.pathname.length > prefix.length; +} + +module.exports = { isInActorPod }; diff --git a/backend/services/actor-pod.test.js b/backend/services/actor-pod.test.js new file mode 100644 index 0000000..c32b423 --- /dev/null +++ b/backend/services/actor-pod.test.js @@ -0,0 +1,56 @@ +const { test, expect } = require('bun:test'); +const { isInActorPod } = require('./actor-pod'); + +const ALICE = 'http://localhost:3000/alice'; +const BOB = 'http://localhost:3000/bob'; + +test("a resource of the actor's own pod is in it", () => { + expect(isInActorPod(`${ALICE}/data/0b5e7c1e-event`, ALICE)).toBe(true); + expect(isInActorPod(`${ALICE}/data/as/event/x`, ALICE)).toBe(true); + expect(isInActorPod(`${ALICE}/data/x#fragment`, ALICE)).toBe(true); + // An actor URI written with a trailing slash owns the same namespace. + expect(isInActorPod(`${ALICE}/data/x`, `${ALICE}/`)).toBe(true); +}); + +test("another actor's resource is not — including a name that merely starts the same", () => { + expect(isInActorPod(`${BOB}/data/x`, ALICE)).toBe(false); + expect(isInActorPod('http://localhost:3000/alice2/data/x', ALICE)).toBe(false); + expect(isInActorPod('http://localhost:3000/alicedata/x', ALICE)).toBe(false); +}); + +test('another origin is not, even with the same path', () => { + expect(isInActorPod('http://localhost:3001/alice/data/x', ALICE)).toBe(false); + expect(isInActorPod('https://localhost:3000/alice/data/x', ALICE)).toBe(false); + expect(isInActorPod('http://evil.example/alice/data/x', ALICE)).toBe(false); +}); + +test('a path that only LOOKS under the actor is refused (non-canonical forms)', () => { + for (const uri of [ + `${ALICE}/../bob/data/x`, + `${ALICE}/./data/x`, + `${ALICE}/%2e%2e/bob/data/x`, + `${ALICE}/data/../../bob/data/x`, + 'HTTP://localhost:3000/alice/data/x', + 'http://LOCALHOST:3000/alice/data/x', + 'http://localhost:3000//alice/data/x/../../../bob', + ` ${ALICE}/data/x`, + `${ALICE}/data/x ` + ]) { + expect(isInActorPod(uri, ALICE)).toBe(false); + } +}); + +test('the actor URI itself, its parent, and missing or malformed input are refused', () => { + expect(isInActorPod(ALICE, ALICE)).toBe(false); + expect(isInActorPod(`${ALICE}/`, ALICE)).toBe(false); + expect(isInActorPod('http://localhost:3000/', ALICE)).toBe(false); + expect(isInActorPod(undefined, ALICE)).toBe(false); + expect(isInActorPod(`${ALICE}/data/x`, undefined)).toBe(false); + expect(isInActorPod({ id: `${ALICE}/data/x` }, ALICE)).toBe(false); + expect(isInActorPod('not a url', ALICE)).toBe(false); + expect(isInActorPod('urn:x:alice/data', 'urn:x:alice')).toBe(false); +}); + +test('an actor at the origin root owns nothing (it would own the whole host)', () => { + expect(isInActorPod('http://localhost:3000/bob/data/x', 'http://localhost:3000/')).toBe(false); +}); diff --git a/backend/services/catalog-store.js b/backend/services/catalog-store.js index 1c23b78..4889934 100644 --- a/backend/services/catalog-store.js +++ b/backend/services/catalog-store.js @@ -61,6 +61,22 @@ function removeEntryUpdate(uri) { `; } +/** + * Remove one event's entry ONLY while its snapshot is still exactly `snapshot` — the text a caller + * read and judged. If the event was re-indexed in between, its new snapshot does not match and + * nothing is removed, so a sweep never deletes an entry it did not see. + */ +function removeEntryIfSnapshotUpdate(uri, snapshot) { + const s = iriRef(uri); + return ` + DELETE { ${s} ?p ?o } + WHERE { + ${s} <${SNAPSHOT}> ${stringLiteral(snapshot)} . + ${s} ?p ?o . FILTER(?p IN (<${SNAPSHOT}>, <${INDEXED_AT}>)) + } + `; +} + /** Every entry, one row per event, in the order events were first indexed. */ const LIST_ENTRIES_QUERY = ` SELECT ?uri (SAMPLE(?snap) AS ?snapshot) (MIN(?at) AS ?indexedAt) @@ -77,21 +93,32 @@ function termValue(term) { return term && typeof term === 'object' ? term.value : term; } +/** + * The entries from the listing rows: each event URI, its snapshot text as stored, and the item + * parsed from it. A snapshot that is not valid JSON is skipped and reported (`invalid`). + */ +function entriesFromRows(rows) { + const entries = []; + const invalid = []; + for (const row of rows || []) { + const uri = termValue(row.uri); + const snapshot = termValue(row.snapshot); + try { + entries.push({ uri, snapshot, item: JSON.parse(snapshot) }); + } catch { + invalid.push(uri); + } + } + return { entries, invalid }; +} + /** * The catalog items from the listing rows. A snapshot that is not valid JSON is * skipped and reported, rather than failing the whole public listing. */ function itemsFromRows(rows) { - const items = []; - const invalid = []; - for (const row of rows || []) { - try { - items.push(JSON.parse(termValue(row.snapshot))); - } catch { - invalid.push(termValue(row.uri)); - } - } - return { items, invalid }; + const { entries, invalid } = entriesFromRows(rows); + return { items: entries.map(entry => entry.item), invalid }; } // ── Store operations: `ctx` is a moleculer context (anything with `call`) ── @@ -110,6 +137,23 @@ async function listEntries(ctx, { dataset }) { return itemsFromRows(rows); } +/** + * Remove every entry whose item fails `keep(item)`, each one only if its snapshot is still the one + * judged (see `removeEntryIfSnapshotUpdate`). An entry whose snapshot cannot be parsed is not + * judged, so it is left in place (the listing never serves it). Resolves the removed URIs. + */ +async function sweepEntries(ctx, { dataset, keep }) { + const rows = await ctx.call('triplestore.query', { query: LIST_ENTRIES_QUERY, dataset, webId: 'system' }); + const { entries } = entriesFromRows(rows); + const removed = []; + for (const { uri, snapshot, item } of entries) { + if (keep(item)) continue; + await ctx.call('triplestore.update', { query: removeEntryIfSnapshotUpdate(uri, snapshot), dataset, webId: 'system' }); + removed.push(uri); + } + return removed; +} + module.exports = { SNAPSHOT, INDEXED_AT, @@ -118,8 +162,11 @@ module.exports = { stringLiteral, upsertEntryUpdate, removeEntryUpdate, + removeEntryIfSnapshotUpdate, + entriesFromRows, itemsFromRows, upsertEntry, removeEntry, - listEntries + listEntries, + sweepEntries }; diff --git a/backend/services/catalog-store.test.js b/backend/services/catalog-store.test.js index a2dcdc3..80b568a 100644 --- a/backend/services/catalog-store.test.js +++ b/backend/services/catalog-store.test.js @@ -7,10 +7,13 @@ const { stringLiteral, upsertEntryUpdate, removeEntryUpdate, + removeEntryIfSnapshotUpdate, + entriesFromRows, itemsFromRows, upsertEntry, removeEntry, - listEntries + listEntries, + sweepEntries } = require('./catalog-store'); const EVENT = 'http://localhost:3000/alice/data/0b5e7c1e-event'; @@ -122,3 +125,49 @@ describe('catalog store — triplestore calls', () => { expect(ctx.calls).toEqual([]); }); }); + +describe('catalog store — sweeping entries that fail a predicate', () => { + const listed = { id: 'http://p/listed', type: 'Event', to: 'as:Public' }; + const unlisted = { id: 'http://p/private', type: 'Event', name: 'secret "address" \\ 12 rue X\nfloor 2' }; + const rowOf = item => ({ uri: { value: item.id }, snapshot: { value: JSON.stringify(item) } }); + const keepListed = item => item.to === 'as:Public'; + + test('the conditional removal names the exact snapshot text it judged, and only the catalog predicates', () => { + const snapshot = JSON.stringify(unlisted); + const query = removeEntryIfSnapshotUpdate(unlisted.id, snapshot); + expect(firstLiteral(query)).toBe(snapshot); + expect(query).toContain(`<${unlisted.id}> <${SNAPSHOT}> "`); + expect(query).toContain(`DELETE { <${unlisted.id}> ?p ?o }`); + expect(query).toContain(`FILTER(?p IN (<${SNAPSHOT}>, <${INDEXED_AT}>))`); + expect(query.replace(/"(?:[^"\\]|\\.)*"/g, '""')).not.toContain(';'); + expect(() => removeEntryIfSnapshotUpdate('http://x/a> } ; DROP ALL', snapshot)).toThrow(); + }); + + test('rows become entries carrying the URI, the stored text and the parsed item', () => { + const rows = [rowOf(listed), { uri: { value: 'http://p/bad' }, snapshot: { value: '{oops' } }]; + expect(entriesFromRows(rows)).toEqual({ + entries: [{ uri: listed.id, snapshot: JSON.stringify(listed), item: listed }], + invalid: ['http://p/bad'] + }); + }); + + test('only the entries failing the predicate are removed, each conditionally on what was read', async () => { + const rows = [rowOf(listed), rowOf(unlisted), { uri: { value: 'http://p/bad' }, snapshot: { value: '{oops' } }]; + const ctx = fakeCtx(rows); + const removed = await sweepEntries(ctx, { dataset: 'settings', keep: keepListed }); + expect(removed).toEqual([unlisted.id]); + expect(ctx.calls.map(c => c.action)).toEqual(['triplestore.query', 'triplestore.update']); + expect(ctx.calls[0].params.query).toBe(LIST_ENTRIES_QUERY); + expect(ctx.calls[1].params.query).toBe(removeEntryIfSnapshotUpdate(unlisted.id, JSON.stringify(unlisted))); + for (const { params } of ctx.calls) { + expect(params.dataset).toBe('settings'); + expect(params.webId).toBe('system'); + } + }); + + test('nothing to sweep, nothing written', async () => { + const ctx = fakeCtx([rowOf(listed)]); + expect(await sweepEntries(ctx, { dataset: 'settings', keep: keepListed })).toEqual([]); + expect(ctx.calls.map(c => c.action)).toEqual(['triplestore.query']); + }); +}); diff --git a/backend/services/catalog.service.js b/backend/services/catalog.service.js index 2a10b41..bb98c1d 100644 --- a/backend/services/catalog.service.js +++ b/backend/services/catalog.service.js @@ -2,7 +2,9 @@ const urlJoin = require('url-join'); const { PodResourcesHandlerMixin } = require('@activitypods/app'); const { matchActivity, ACTIVITY_TYPES, OBJECT_TYPES } = require('@semapps/activitypub'); const CONFIG = require('../config/config'); -const { upsertEntry, removeEntry, listEntries } = require('./catalog-store'); +const { upsertEntry, removeEntry, listEntries, sweepEntries } = require('./catalog-store'); +const { hasListedPublicIntent } = require('./public-intent'); +const { isInActorPod } = require('./actor-pod'); function arrayOf(value) { return value === undefined || value === null ? [] : Array.isArray(value) ? value : [value]; @@ -12,26 +14,66 @@ function uriOf(resource) { return typeof resource === 'string' ? resource : resource && (resource.id || resource['@id']); } +/** + * What the catalog may hold and serve: an item addressed to the ActivityStreams Public collection + * in `to` — public AND listed (`hasListedPublicIntent`, built on the public-read gate's own + * predicate). An event with no addressing (another app's, one posted directly on the pod), one + * addressed to chosen recipients, or a public-but-unlisted one (Public in `cc` only) is never + * listed. Festipod's create path writes `to: as:Public` on its events and meeting points, so both + * are listed. Judged on the object alone, as the public-read gate judges it. + */ +const isListable = item => hasListedPublicIntent(item); + +/** + * A matcher that never throws: on error it logs and reports no match. The watcher runs every + * service's matchers for an activity in sequence without catching, inside a webhook call the + * platform does not await — a throw would skip the handlers after this one and end the process. + */ +function neverThrowingMatcher(match) { + return async function (activity, fetcher) { + try { + return await match.call(this, activity, fetcher); + } catch (e) { + this.logger.error(`Catalog could not match activity ${uriOf(activity)}: ${e.message}`); + return { match: false, dereferencedActivity: activity }; + } + }; +} + module.exports = { name: 'catalog', mixins: [PodResourcesHandlerMixin], settings: { shapeTreeUri: urlJoin(CONFIG.SHAPE_REPOSITORY_URL, 'shapetrees/as/Event') }, - // Moleculer does not merge a custom schema key: declaring `activities` here - // replaces the mixin's whole map. `create` and `update` are the mixin's own - // handlers, reused as they are; only `delete` is redefined (see below). + // Moleculer does not merge a custom schema key: declaring `activities` here replaces the mixin's + // whole map. `create` and `update` keep the mixin's matchers (made never-throwing), but every + // `onEmit` hands the handler the actor the WATCHER resolved — the owner of the outbox the + // webhook fired on — instead of the mixin's `activity.actor`, which the emitting client writes. activities: { - create: PodResourcesHandlerMixin.activities.create, - update: PodResourcesHandlerMixin.activities.update, + create: { + match: neverThrowingMatcher(PodResourcesHandlerMixin.activities.create.match), + async onEmit(ctx, activity, actorUri) { + await this.onCreate(ctx, activity.object, actorUri); + } + }, + update: { + match: neverThrowingMatcher(PodResourcesHandlerMixin.activities.update.match), + async onEmit(ctx, activity, actorUri) { + await this.onUpdate(ctx, activity.object, actorUri); + } + }, // The mixin's delete handler never fires for a deleted event: it matches the // Tombstone's `formerType` against the COMPACTED type (`Event`) while the // Tombstone the Pod serves carries `as:Event` (verified 2026-10-12), and on a // match it would call `onUpdate` with the Tombstone instead of `onDelete`. Same // pattern here — an emitted Delete of a Tombstone whose former type is this // service's type — with both types compared EXPANDED, routed to `onDelete`. + // The Tombstone may be EMBEDDED in the activity (the pod accepts a Delete of any URI + // with an embedded Tombstone, verified 2026-10-12): it is then used as written, so + // only `onDelete`'s ownership test stands between it and someone else's entry. delete: { - async match(activity, fetcher) { + match: neverThrowingMatcher(async function (activity, fetcher) { const pattern = { type: ACTIVITY_TYPES.DELETE, object: { type: OBJECT_TYPES.TOMBSTONE } }; const { match, dereferencedActivity } = await matchActivity(pattern, activity, fetcher); if (!match) return { match: false, dereferencedActivity }; @@ -48,41 +90,93 @@ module.exports = { this.logger.warn(`Catalog could not read the former type of ${uriOf(tombstone)}: ${e.message}`); return { match: false, dereferencedActivity }; } - }, - async onEmit(ctx, activity) { - await this.onDelete(ctx, uriOf(activity.object)); + }), + async onEmit(ctx, activity, actorUri) { + await this.onDelete(ctx, uriOf(activity.object), actorUri); } } }, + /** + * Entries the catalog would not serve are removed from its store at startup, so it does not + * keep copies of bodies that were never public (indexed before the listing rule existed). The + * listing filters them anyway (`discover`); this only stops holding them. Never fatal. + */ + async started() { + try { + const removed = await sweepEntries(this.broker, { dataset: CONFIG.MAIN_DATASET, keep: isListable }); + if (removed.length > 0) { + this.logger.info(`Catalog removed ${removed.length} entries not listed to the Public collection: ${removed.join(', ')}`); + } + } catch (e) { + this.logger.error(`Catalog could not sweep entries not listed to the Public collection: ${e.message}`); + } + }, actions: { discover: { async handler(ctx) { const { items, invalid } = await listEntries(ctx, { dataset: CONFIG.MAIN_DATASET }); if (invalid.length > 0) this.logger.warn(`Catalog skipped unreadable entries: ${invalid.join(', ')}`); + // Filtered on every read, whatever wrote the store: only listed public items are served. + const listed = items.filter(isListable); return { '@context': 'https://www.w3.org/ns/activitystreams', type: 'Collection', summary: 'Public Festipod events and meeting points', - totalItems: items.length, - items + totalItems: listed.length, + items: listed }; } } }, + // Each mutation is the same composition, in this order: (1) the object must lie in the pod of + // the actor whose outbox emitted the activity (`isInActorPod`) — no actor indexes, overwrites or + // removes another actor's entry; (2) only then is the listing rule applied; (3) only then is the + // store written. A handler never throws: it runs inside the watcher (see `neverThrowingMatcher`). methods: { - async onCreate(ctx, resource) { + async onCreate(ctx, resource, actorUri) { const uri = uriOf(resource); - await upsertEntry(ctx, { dataset: CONFIG.MAIN_DATASET, uri, resource }); - this.logger.info(`Catalog indexed new event ${uri}`); + await this.guarded(`index ${uri}`, async () => { + if (!this.isOwnedBy(uri, actorUri, 'index')) return; + if (!isListable(resource)) return this.logNotListed(uri); + await upsertEntry(ctx, { dataset: CONFIG.MAIN_DATASET, uri, resource }); + this.logger.info(`Catalog indexed new event ${uri}`); + }); }, - async onUpdate(ctx, resource) { + async onUpdate(ctx, resource, actorUri) { const uri = uriOf(resource); - await upsertEntry(ctx, { dataset: CONFIG.MAIN_DATASET, uri, resource }); - this.logger.info(`Catalog updated event ${uri}`); + await this.guarded(`update ${uri}`, async () => { + if (!this.isOwnedBy(uri, actorUri, 'update')) return; + if (!isListable(resource)) { + // No longer listed (or never was): it must not stay listed from an earlier version. + await removeEntry(ctx, { dataset: CONFIG.MAIN_DATASET, uri }); + return this.logNotListed(uri); + } + await upsertEntry(ctx, { dataset: CONFIG.MAIN_DATASET, uri, resource }); + this.logger.info(`Catalog updated event ${uri}`); + }); }, - async onDelete(ctx, resourceUri) { - await removeEntry(ctx, { dataset: CONFIG.MAIN_DATASET, uri: resourceUri }); - this.logger.info(`Catalog removed event ${resourceUri}`); + async onDelete(ctx, resourceUri, actorUri) { + await this.guarded(`remove ${resourceUri}`, async () => { + if (!this.isOwnedBy(resourceUri, actorUri, 'remove')) return; + await removeEntry(ctx, { dataset: CONFIG.MAIN_DATASET, uri: resourceUri }); + this.logger.info(`Catalog removed event ${resourceUri}`); + }); + }, + /** The ownership test of every mutation; a refusal is logged, as it may be an attempt. */ + isOwnedBy(uri, actorUri, operation) { + if (isInActorPod(uri, actorUri)) return true; + this.logger.warn(`Catalog refused to ${operation} ${uri}: not in the pod of ${actorUri}, whose outbox emitted it`); + return false; + }, + logNotListed(uri) { + this.logger.info(`Catalog does not list ${uri}: not addressed to the Public collection in \`to\``); + }, + async guarded(what, run) { + try { + await run(); + } catch (e) { + this.logger.error(`Catalog could not ${what}: ${e.message}`); + } } } }; diff --git a/backend/services/catalog.service.test.js b/backend/services/catalog.service.test.js new file mode 100644 index 0000000..2d98202 --- /dev/null +++ b/backend/services/catalog.service.test.js @@ -0,0 +1,237 @@ +const { test, expect, describe } = require('bun:test'); +const catalog = require('./catalog.service'); +const { removeEntryUpdate, upsertEntryUpdate, LIST_ENTRIES_QUERY, removeEntryIfSnapshotUpdate } = require('./catalog-store'); + +const ALICE = 'http://localhost:3000/alice'; +const BOB = 'http://localhost:3000/bob'; +const AS_EVENT = 'https://www.w3.org/ns/activitystreams#Event'; + +const event = (owner, extra = {}) => ({ id: `${owner}/data/${extra.slug || 'e1'}`, type: 'Event', name: 'Concert', ...extra }); +const listedEvent = owner => event(owner, { to: 'as:Public' }); + +/** The service as moleculer would run it: its methods bound to one object with a recording logger. */ +function service({ broker } = {}) { + const logs = { info: [], warn: [], error: [] }; + const svc = { + logger: { info: m => logs.info.push(m), warn: m => logs.warn.push(m), error: m => logs.error.push(m) }, + settings: { type: 'as:Event' }, + broker: broker || { call: async () => [] } + }; + for (const [name, fn] of Object.entries(catalog.methods)) svc[name] = fn.bind(svc); + return { svc, logs }; +} + +/** A moleculer context recording triplestore calls; `rows` answers queries; `fail` makes every call throw. */ +function store({ rows = [], fail = false } = {}) { + const calls = []; + return { + calls, + call: async (action, params) => { + calls.push({ action, params }); + if (fail) throw new Error('triplestore down'); + return action === 'triplestore.query' ? rows : undefined; + } + }; +} + +const queries = ctx => ctx.calls.map(c => c.params.query); + +describe('catalog mutations — ownership first, then the listing rule, then the store', () => { + test("the owner's listed public event is indexed (the normal case)", async () => { + const { svc, logs } = service(); + const ctx = store(); + const e = listedEvent(ALICE); + await svc.onCreate(ctx, e, ALICE); + expect(queries(ctx)).toEqual([upsertEntryUpdate(e.id, e)]); + expect(logs.info).toContain(`Catalog indexed new event ${e.id}`); + }); + + test("the owner's event without listed public intent is never indexed", async () => { + for (const e of [ + event(ALICE), // no addressing: another app's event, or one posted directly on the pod + event(ALICE, { to: BOB }), // addressed to a friend + event(ALICE, { to: BOB, cc: 'as:Public' }), // public but unlisted + event(ALICE, { bto: 'as:Public' }) + ]) { + const { svc, logs } = service(); + const ctx = store(); + await svc.onCreate(ctx, e, ALICE); + expect(ctx.calls).toEqual([]); + expect(logs.info.some(m => m.includes('does not list'))).toBe(true); + } + }); + + test("an actor cannot index an object of another actor's pod, listed or not", async () => { + const { svc, logs } = service(); + const ctx = store(); + await svc.onCreate(ctx, listedEvent(BOB), ALICE); + expect(ctx.calls).toEqual([]); + expect(logs.warn[0]).toContain(`Catalog refused to index ${BOB}/data/e1: not in the pod of ${ALICE}`); + }); + + test("an update re-indexes the owner's listed event, and de-indexes it once it is no longer listed", async () => { + const { svc } = service(); + const listed = listedEvent(ALICE); + const ctx = store(); + await svc.onUpdate(ctx, listed, ALICE); + expect(queries(ctx)).toEqual([upsertEntryUpdate(listed.id, listed)]); + + const unlisted = event(ALICE, { to: BOB, cc: 'as:Public' }); + const ctx2 = store(); + await svc.onUpdate(ctx2, unlisted, ALICE); + expect(queries(ctx2)).toEqual([removeEntryUpdate(unlisted.id)]); + }); + + test("an update emitted by another actor writes nothing — neither an upsert nor a de-index", async () => { + for (const e of [listedEvent(BOB), event(BOB)]) { + const { svc, logs } = service(); + const ctx = store(); + await svc.onUpdate(ctx, e, ALICE); + expect(ctx.calls).toEqual([]); + expect(logs.warn[0]).toContain('Catalog refused to update'); + } + }); + + test("the owner's deletion de-indexes; another actor's deletion is refused", async () => { + const own = store(); + await service().svc.onDelete(own, `${ALICE}/data/e1`, ALICE); + expect(queries(own)).toEqual([removeEntryUpdate(`${ALICE}/data/e1`)]); + + const { svc, logs } = service(); + const foreign = store(); + await svc.onDelete(foreign, `${BOB}/data/e1`, ALICE); + expect(foreign.calls).toEqual([]); + expect(logs.warn[0]).toContain(`Catalog refused to remove ${BOB}/data/e1: not in the pod of ${ALICE}`); + }); + + test("a URI that only looks like it is in the actor's pod is refused", async () => { + for (const uri of [`${ALICE}/../bob/data/e1`, `${ALICE}/%2e%2e/bob/data/e1`, 'http://localhost:3000/alice2/data/e1']) { + const ctx = store(); + await service().svc.onDelete(ctx, uri, ALICE); + await service().svc.onCreate(ctx, { id: uri, type: 'Event', to: 'as:Public' }, ALICE); + expect(ctx.calls).toEqual([]); + } + }); + + test('without a known emitting actor, nothing is written', async () => { + const ctx = store(); + await service().svc.onCreate(ctx, listedEvent(ALICE), undefined); + await service().svc.onUpdate(ctx, listedEvent(ALICE), undefined); + await service().svc.onDelete(ctx, `${ALICE}/data/e1`, undefined); + expect(ctx.calls).toEqual([]); + }); + + test('a store failure is logged, never thrown into the webhook path', async () => { + const { svc, logs } = service(); + const ctx = store({ fail: true }); + await expect(svc.onCreate(ctx, listedEvent(ALICE), ALICE)).resolves.toBeUndefined(); + await expect(svc.onUpdate(ctx, listedEvent(ALICE), ALICE)).resolves.toBeUndefined(); + await expect(svc.onUpdate(ctx, event(ALICE), ALICE)).resolves.toBeUndefined(); + await expect(svc.onDelete(ctx, `${ALICE}/data/e1`, ALICE)).resolves.toBeUndefined(); + expect(logs.error).toHaveLength(4); + expect(logs.error[0]).toContain('triplestore down'); + }); + + test('a malformed resource is refused or logged, never thrown', async () => { + const { svc } = service(); + const ctx = store(); + await expect(svc.onCreate(ctx, undefined, ALICE)).resolves.toBeUndefined(); + await expect(svc.onUpdate(ctx, { type: 'Event', to: 'as:Public' }, ALICE)).resolves.toBeUndefined(); + expect(ctx.calls).toEqual([]); + }); +}); + +describe('catalog activity handlers — the emitting actor is the one the watcher resolved', () => { + test("onEmit uses the outbox owner, not the activity's self-declared actor", async () => { + const { svc, logs } = service(); + const ctx = store(); + // Alice's outbox carries an activity claiming Bob as its actor, about Bob's event. + const bobs = listedEvent(BOB); + await catalog.activities.create.onEmit.call(svc, ctx, { type: 'Create', actor: BOB, object: bobs }, ALICE); + await catalog.activities.update.onEmit.call(svc, ctx, { type: 'Update', actor: BOB, object: bobs }, ALICE); + await catalog.activities.delete.onEmit.call(svc, ctx, { type: 'Delete', actor: BOB, object: { id: bobs.id, type: 'Tombstone' } }, ALICE); + expect(ctx.calls).toEqual([]); + expect(logs.warn).toHaveLength(3); + }); + + test("the owner's own activities go through", async () => { + const { svc } = service(); + const ctx = store(); + const e = listedEvent(ALICE); + await catalog.activities.create.onEmit.call(svc, ctx, { type: 'Create', actor: ALICE, object: e }, ALICE); + await catalog.activities.delete.onEmit.call(svc, ctx, { type: 'Delete', actor: ALICE, object: { id: e.id, type: 'Tombstone' } }, ALICE); + expect(queries(ctx)).toEqual([upsertEntryUpdate(e.id, e), removeEntryUpdate(e.id)]); + }); + + test('an embedded Tombstone of an event matches the delete handler (it is then judged by ownership)', async () => { + const expand = t => (t === 'as:Event' || t === 'Event' ? AS_EVENT : t); + const broker = { call: async (action, { types }) => [].concat(types).map(expand) }; + const { svc } = service({ broker }); + const activity = { type: 'Delete', object: { id: `${BOB}/data/e1`, type: 'Tombstone', formerType: 'as:Event' } }; + const result = await catalog.activities.delete.match.call(svc, activity, async () => false); + expect(result.match).toBe(true); + }); + + test('a matcher that fails reports no match instead of throwing', async () => { + const { svc, logs } = service(); + const failingFetcher = async () => { + throw new Error('proxy unreachable'); + }; + for (const key of ['create', 'update', 'delete']) { + const activity = { type: key[0].toUpperCase() + key.slice(1), object: `${ALICE}/data/e1` }; + const result = await catalog.activities[key].match.call(svc, activity, failingFetcher); + expect(result.match).toBe(false); + } + expect(logs.error).toHaveLength(3); + expect(logs.error[0]).toContain('proxy unreachable'); + }); +}); + +describe('catalog listing — only listed public items are served, shape unchanged', () => { + const rowOf = item => ({ uri: { value: item.id }, snapshot: { value: JSON.stringify(item) } }); + + test('entries already stored that are not listed public are filtered out at read', async () => { + const listed = listedEvent(ALICE); + const meetingPoint = { ...listedEvent(ALICE), id: `${ALICE}/data/mp`, 'as:context': listed.id }; + const stored = [ + rowOf(event(BOB, { slug: 'private', content: 'secret address 12 rue X' })), + rowOf(listed), + rowOf(event(BOB, { slug: 'friend', to: ALICE, content: 'only for my friend' })), + rowOf(event(BOB, { slug: 'unlisted', cc: 'as:Public' })), + rowOf(meetingPoint) + ]; + const { svc } = service(); + const result = await catalog.actions.discover.handler.call(svc, store({ rows: stored })); + expect(Object.keys(result).sort()).toEqual(['@context', 'items', 'summary', 'totalItems', 'type']); + expect(result['@context']).toBe('https://www.w3.org/ns/activitystreams'); + expect(result.type).toBe('Collection'); + expect(result.items).toEqual([listed, meetingPoint]); + expect(result.totalItems).toBe(2); + expect(JSON.stringify(result)).not.toContain('secret address'); + }); +}); + +describe('catalog startup — entries it would not serve are removed from its store', () => { + const rowOf = item => ({ uri: { value: item.id }, snapshot: { value: JSON.stringify(item) } }); + + test('the sweep removes exactly the unlisted entries, conditionally on what it read', async () => { + const listed = listedEvent(ALICE); + const secret = event(BOB, { slug: 'private', content: 'secret address' }); + const unlisted = event(BOB, { slug: 'unlisted', cc: 'as:Public' }); + const broker = store({ rows: [rowOf(listed), rowOf(secret), rowOf(unlisted)] }); + const { svc, logs } = service({ broker }); + await catalog.started.call(svc); + expect(queries(broker)).toEqual([ + LIST_ENTRIES_QUERY, + removeEntryIfSnapshotUpdate(secret.id, JSON.stringify(secret)), + removeEntryIfSnapshotUpdate(unlisted.id, JSON.stringify(unlisted)) + ]); + expect(logs.info[0]).toContain('Catalog removed 2 entries not listed to the Public collection'); + }); + + test('a failing sweep is logged and never prevents the start', async () => { + const { svc, logs } = service({ broker: store({ fail: true }) }); + await expect(catalog.started.call(svc)).resolves.toBeUndefined(); + expect(logs.error[0]).toContain('Catalog could not sweep'); + }); +});