Persist the catalog index across backend restarts and de-index deleted events

Each indexed event is one entry in the backend's settings dataset, default graph (an undeclared named graph is kept in memory by fuseki and lost on restart): the item JSON verbatim plus its first-index time, written by one atomic upsert, so concurrent re-adds stay one entry. The catalog now has its own delete matcher, because the platform one compares formerType as:Event to the compacted Event and never fired: an owner's deletion removes the entry, also after a restart. GET /catalog keeps its shape.

backend/services/catalog-store.js

@@ -0,0 +1,125 @@
+/**
+ * The catalog index, persisted in the backend's own triplestore so it survives a
+ * backend restart (it used to be process memory).
+ *
+ * WHERE IT LIVES. In the backend's main dataset (`settings`), in the DEFAULT graph.
+ * Not in a dedicated named graph: that dataset is assembled from TDB2 graphs, and only
+ * the graphs its assembler declares are TDB2-backed — a triple written to any other
+ * named graph is held in memory by Fuseki and lost when Fuseki restarts (verified on
+ * the local stack, 2026-10-12). The default graph is the one that persists.
+ *
+ * ONE ENTRY PER EVENT. The entry's subject IS the event URI, with exactly two
+ * Festipod predicates: the snapshot (the resource as the catalog serves it, as JSON
+ * text, so the `GET /catalog` items are returned byte-for-byte as they were indexed)
+ * and the time it was first indexed (the listing order). Re-indexing replaces the
+ * snapshot in ONE SPARQL update, which Fuseki runs as a single write transaction, so
+ * concurrent re-adds of the same event cannot leave two snapshots; the listing also
+ * groups by subject, so even a stray duplicate could never list an event twice.
+ */
+
+const SNAPSHOT = 'http://festipod.org/catalogSnapshot';
+const INDEXED_AT = 'http://festipod.org/catalogIndexedAt';
+
+/** An IRI as a SPARQL IRIREF, refusing the characters the grammar forbids in one. */
+function iriRef(uri) {
+  if (typeof uri !== 'string' || uri.length === 0 || /[\u0000- <>"{}|^`\\]/.test(uri)) {
+    throw new Error(`Not an IRI the catalog can store: ${JSON.stringify(uri)}`);
+  }
+  return `<${uri}>`;
+}
+
+const STRING_ESCAPES = { '\\': '\\\\', '"': '\\"', '\n': '\\n', '\r': '\\r', '\t': '\\t', '\b': '\\b', '\f': '\\f' };
+
+/** A string as a SPARQL double-quoted literal. */
+function stringLiteral(value) {
+  return `"${String(value).replace(/[\\"\n\r\t\b\f]/g, c => STRING_ESCAPES[c])}"`;
+}
+
+/**
+ * Index (or re-index) one event: replace its snapshot, keep the time it was first
+ * indexed (or stamp it now). One update request = one write transaction.
+ */
+function upsertEntryUpdate(uri, resource) {
+  const s = iriRef(uri);
+  return `
+    DELETE { ${s} <${SNAPSHOT}> ?old }
+    INSERT { ${s} <${SNAPSHOT}> ${stringLiteral(JSON.stringify(resource))} . ${s} <${INDEXED_AT}> ?at }
+    WHERE {
+      OPTIONAL { ${s} <${SNAPSHOT}> ?old }
+      OPTIONAL { ${s} <${INDEXED_AT}> ?first }
+      BIND(COALESCE(?first, NOW()) AS ?at)
+    }
+  `;
+}
+
+/** Remove one event's entry — only the catalog's own predicates, nothing else about that URI. */
+function removeEntryUpdate(uri) {
+  const s = iriRef(uri);
+  return `
+    DELETE { ${s} ?p ?o }
+    WHERE { ${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)
+  WHERE {
+    ?uri <${SNAPSHOT}> ?snap .
+    OPTIONAL { ?uri <${INDEXED_AT}> ?at }
+  }
+  GROUP BY ?uri
+  ORDER BY ASC(?indexedAt) ASC(STR(?uri))
+`;
+
+/** A SELECT binding's value: the triplestore service returns RDF/JS terms. */
+function termValue(term) {
+  return term && typeof term === 'object' ? term.value : term;
+}
+
+/**
+ * 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 };
+}
+
+// ── Store operations: `ctx` is a moleculer context (anything with `call`) ──
+
+async function upsertEntry(ctx, { dataset, uri, resource }) {
+  await ctx.call('triplestore.update', { query: upsertEntryUpdate(uri, resource), dataset, webId: 'system' });
+}
+
+async function removeEntry(ctx, { dataset, uri }) {
+  await ctx.call('triplestore.update', { query: removeEntryUpdate(uri), dataset, webId: 'system' });
+}
+
+async function listEntries(ctx, { dataset }) {
+  // The default `accept` (JSON) makes the service parse SELECT results into RDF/JS terms.
+  const rows = await ctx.call('triplestore.query', { query: LIST_ENTRIES_QUERY, dataset, webId: 'system' });
+  return itemsFromRows(rows);
+}
+
+module.exports = {
+  SNAPSHOT,
+  INDEXED_AT,
+  LIST_ENTRIES_QUERY,
+  iriRef,
+  stringLiteral,
+  upsertEntryUpdate,
+  removeEntryUpdate,
+  itemsFromRows,
+  upsertEntry,
+  removeEntry,
+  listEntries
+};

backend/services/catalog-store.test.js

@@ -0,0 +1,124 @@
+const { test, expect, describe } = require('bun:test');
+const {
+  SNAPSHOT,
+  INDEXED_AT,
+  LIST_ENTRIES_QUERY,
+  iriRef,
+  stringLiteral,
+  upsertEntryUpdate,
+  removeEntryUpdate,
+  itemsFromRows,
+  upsertEntry,
+  removeEntry,
+  listEntries
+} = require('./catalog-store');
+
+const EVENT = 'http://localhost:3000/alice/data/0b5e7c1e-event';
+
+/** Decode the first SPARQL double-quoted literal of a query (the grammar's ECHAR escapes). */
+function firstLiteral(query) {
+  const match = /"((?:[^"\\]|\\.)*)"/.exec(query);
+  if (!match) throw new Error('no literal in query');
+  const map = { t: '\t', b: '\b', n: '\n', r: '\r', f: '\f', '"': '"', "'": "'", '\\': '\\' };
+  return match[1].replace(/\\(.)/g, (_, c) => map[c]);
+}
+
+function fakeCtx(result) {
+  const calls = [];
+  return { calls, call: async (action, params) => (calls.push({ action, params }), result) };
+}
+
+describe('catalog store — SPARQL it sends', () => {
+  test('an IRI is wrapped as an IRIREF; anything that could break out of one is refused', () => {
+    expect(iriRef(EVENT)).toBe(`<${EVENT}>`);
+    for (const bad of ['', 'http://x/a b', 'http://x/a>', 'http://x/<a', 'http://x/"a', 'http://x/a\\b', 'http://x/{a}', undefined, 42]) {
+      expect(() => iriRef(bad)).toThrow();
+    }
+  });
+
+  test('a string literal escapes quotes, backslashes and line breaks', () => {
+    expect(stringLiteral('a"b\\c\nd\re\tf')).toBe('"a\\"b\\\\c\\nd\\re\\tf"');
+  });
+
+  test('the indexed snapshot decodes back to exactly the resource, whatever its text holds', () => {
+    const resource = {
+      '@context': ['https://www.w3.org/ns/activitystreams', 'http://localhost:3003/.well-known/context.jsonld'],
+      id: EVENT,
+      type: 'Event',
+      name: 'Concert "plein air" } ; DROP ALL ; # \\ fin',
+      content: 'ligne 1\nligne 2\ttab',
+      'apods:attendees': 'http://localhost:3000/alice/data/attendees'
+    };
+    const query = upsertEntryUpdate(EVENT, resource);
+    expect(JSON.parse(firstLiteral(query))).toEqual(resource);
+    expect(firstLiteral(query)).toBe(JSON.stringify(resource));
+  });
+
+  test('re-indexing replaces the snapshot and keeps the first-indexed time, in one update', () => {
+    const query = upsertEntryUpdate(EVENT, { id: EVENT });
+    expect(query).toContain(`DELETE { <${EVENT}> <${SNAPSHOT}> ?old }`);
+    expect(query).toContain(`<${EVENT}> <${INDEXED_AT}> ?at`);
+    expect(query).toContain('COALESCE(?first, NOW())');
+    // One operation: no `;` separator outside the literal.
+    expect(query.replace(/"(?:[^"\\]|\\.)*"/g, '""')).not.toContain(';');
+  });
+
+  test('removal touches only the catalog predicates of that event', () => {
+    const query = removeEntryUpdate(EVENT);
+    expect(query).toContain(`DELETE { <${EVENT}> ?p ?o }`);
+    expect(query).toContain(`FILTER(?p IN (<${SNAPSHOT}>, <${INDEXED_AT}>))`);
+  });
+
+  test('the listing yields one row per event, oldest first', () => {
+    expect(LIST_ENTRIES_QUERY).toContain('GROUP BY ?uri');
+    expect(LIST_ENTRIES_QUERY).toContain('ORDER BY ASC(?indexedAt)');
+  });
+});
+
+describe('catalog store — rows to items', () => {
+  test('snapshots are parsed back to the items, in row order', () => {
+    const a = { id: 'http://p/a', type: 'Event', name: 'A' };
+    const b = { id: 'http://p/b', type: 'Event', 'as:context': 'http://p/a' };
+    const rows = [
+      { uri: { value: a.id }, snapshot: { value: JSON.stringify(a) } },
+      { uri: { value: b.id }, snapshot: { value: JSON.stringify(b) } }
+    ];
+    expect(itemsFromRows(rows)).toEqual({ items: [a, b], invalid: [] });
+  });
+
+  test('an unreadable snapshot is skipped and reported, not fatal', () => {
+    const rows = [
+      { uri: { value: 'http://p/bad' }, snapshot: { value: '{not json' } },
+      { uri: { value: 'http://p/ok' }, snapshot: { value: '{"id":"http://p/ok"}' } }
+    ];
+    expect(itemsFromRows(rows)).toEqual({ items: [{ id: 'http://p/ok' }], invalid: ['http://p/bad'] });
+  });
+
+  test('no rows, no items', () => {
+    expect(itemsFromRows([])).toEqual({ items: [], invalid: [] });
+    expect(itemsFromRows(undefined)).toEqual({ items: [], invalid: [] });
+  });
+});
+
+describe('catalog store — triplestore calls', () => {
+  test('index, remove and list go to the given dataset as the system', async () => {
+    const ctx = fakeCtx([]);
+    await upsertEntry(ctx, { dataset: 'settings', uri: EVENT, resource: { id: EVENT } });
+    await removeEntry(ctx, { dataset: 'settings', uri: EVENT });
+    await listEntries(ctx, { dataset: 'settings' });
+    expect(ctx.calls.map(c => c.action)).toEqual(['triplestore.update', 'triplestore.update', 'triplestore.query']);
+    for (const { params } of ctx.calls) {
+      expect(params.dataset).toBe('settings');
+      expect(params.webId).toBe('system');
+    }
+    expect(ctx.calls[0].params.query).toBe(upsertEntryUpdate(EVENT, { id: EVENT }));
+    expect(ctx.calls[1].params.query).toBe(removeEntryUpdate(EVENT));
+    expect(ctx.calls[2].params.query).toBe(LIST_ENTRIES_QUERY);
+  });
+
+  test('a URI that is not a storable IRI is refused before anything is sent', async () => {
+    const ctx = fakeCtx([]);
+    await expect(upsertEntry(ctx, { dataset: 'settings', uri: 'http://x/a> } ; DROP ALL', resource: {} })).rejects.toThrow();
+    expect(ctx.calls).toEqual([]);
+  });
+});

backend/services/catalog.service.js

@@ -1,6 +1,16 @@
 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');
+
+function arrayOf(value) {
+  return value === undefined || value === null ? [] : Array.isArray(value) ? value : [value];
+}
+
+function uriOf(resource) {
+  return typeof resource === 'string' ? resource : resource && (resource.id || resource['@id']);
+}
 
 module.exports = {
   name: 'catalog',
@@ -8,33 +18,70 @@ module.exports = {
   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).
+  activities: {
+    create: PodResourcesHandlerMixin.activities.create,
+    update: PodResourcesHandlerMixin.activities.update,
+    // 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`.
+    delete: {
+      async match(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 };
+        const tombstone = dereferencedActivity.object;
+        const formerTypes = arrayOf(tombstone.formerType);
+        if (formerTypes.length === 0) return { match: false, dereferencedActivity };
+        try {
+          const [former, own] = await Promise.all([
+            this.broker.call('jsonld.parser.expandTypes', { types: formerTypes, context: tombstone['@context'] }),
+            this.broker.call('jsonld.parser.expandTypes', { types: this.settings.type })
+          ]);
+          return { match: former.some(type => own.includes(type)), dereferencedActivity };
+        } catch (e) {
+          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));
+      }
+    }
+  },
   actions: {
     discover: {
-      async handler() {
+      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(', ')}`);
         return {
           '@context': 'https://www.w3.org/ns/activitystreams',
           type: 'Collection',
           summary: 'Public Festipod events and meeting points',
-          totalItems: this.index ? Object.keys(this.index).length : 0,
-          items: this.index ? Object.values(this.index) : []
+          totalItems: items.length,
+          items
         };
       }
     }
   },
   methods: {
     async onCreate(ctx, resource) {
-      if (!this.index) this.index = {};
-      this.index[resource.id || resource['@id']] = resource;
-      this.logger.info(`Catalog indexed new event ${resource.id || resource['@id']}`);
+      const uri = uriOf(resource);
+      await upsertEntry(ctx, { dataset: CONFIG.MAIN_DATASET, uri, resource });
+      this.logger.info(`Catalog indexed new event ${uri}`);
     },
     async onUpdate(ctx, resource) {
-      if (!this.index) this.index = {};
-      this.index[resource.id || resource['@id']] = resource;
-      this.logger.info(`Catalog updated event ${resource.id || resource['@id']}`);
+      const uri = uriOf(resource);
+      await upsertEntry(ctx, { dataset: CONFIG.MAIN_DATASET, uri, resource });
+      this.logger.info(`Catalog updated event ${uri}`);
     },
     async onDelete(ctx, resourceUri) {
-      if (!this.index) return;
-      delete this.index[resourceUri];
+      await removeEntry(ctx, { dataset: CONFIG.MAIN_DATASET, uri: resourceUri });
       this.logger.info(`Catalog removed event ${resourceUri}`);
     }
   }