diff --git a/backend/services/catalog-store.js b/backend/services/catalog-store.js new file mode 100644 index 0000000..1c23b78 --- /dev/null +++ b/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 +}; diff --git a/backend/services/catalog-store.test.js b/backend/services/catalog-store.test.js new file mode 100644 index 0000000..a2dcdc3 --- /dev/null +++ b/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/ 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([]); + }); +}); diff --git a/backend/services/catalog.service.js b/backend/services/catalog.service.js index 99afe5c..2a10b41 100644 --- a/backend/services/catalog.service.js +++ b/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}`); } }