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

@@ -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 };

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);
+});

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
 };

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']);
+  });
+});

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}`);
+      }
     }
   }
 };

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');
+  });
+});