const { ACTIVITY_TYPES, OBJECT_TYPES, matchActivity } = require('@semapps/activitypub'); const { PodActivitiesHandlerMixin } = require('@activitypods/app'); const { arrayOf } = require('@semapps/ldp'); const CONFIG = require('../config/config'); const { isInActorPod } = require('./actor-pod'); const { createKeyedSerializer } = require('./keyed-serializer'); const { isNewerAct, lastAttendance, recordAttendance } = require('./attendance-ledger'); const { neverThrowingMatcher, verifiedEmit, verifiedReceive, singleUriOf, uriOf } = require('./emitted-activity'); const APODS_ATTENDEES = 'http://activitypods.org/ns/core#attendees'; const APODS_CONTACTS = 'http://activitypods.org/ns/core#contacts'; const DC_CREATOR = 'http://purl.org/dc/terms/creator'; const JOIN_PATTERN = { type: ACTIVITY_TYPES.JOIN, object: { type: OBJECT_TYPES.EVENT } }; const LEAVE_PATTERN = { type: ACTIVITY_TYPES.LEAVE, object: { type: OBJECT_TYPES.EVENT } }; const AS_PUBLISHED = 'https://www.w3.org/ns/activitystreams#published'; /** The time the pod stamped on a stored activity (`published`, under any of its spellings). */ const publishedOf = activity => { if (!activity || typeof activity !== 'object') return undefined; const value = activity.published ?? activity['as:published'] ?? activity[AS_PUBLISHED]; return value && typeof value === 'object' ? value['@value'] : value; }; /** The single URI an event names under a compacted or full-IRI key. */ const linkOf = (node, compacted, full) => singleUriOf(node[compacted] ?? node[full]); // Participation: a user joins (or leaves) a meeting point with a Join (Leave) posted to her outbox, // addressed to its organizer. Two effects, each from one side's box: // - EMITTED (the joiner's outbox): the organizer joins the joiner's contacts group, so the // organizer may read her profile; // - RECEIVED (the organizer's inbox): the joiner is added to (removed from) the meeting point's // attendees collection. // The webhook that reports either is a hint (emitted-activity.js): each handler acts only on the // activity and the meeting point as the pod stores them, and only for the people they really name: // - the joiner is the actor whose pod holds the activity, never a value of the webhook body; // - the organizer is the meeting point's `dc:creator`, accepted only when the meeting point lies in // that creator's pod — the pod stamps the creator, and nobody can store a resource in someone // else's pod; // - the collection and the group written to are the ones the STORED meeting point and the // joiner's STORED actor document name, and must lie in the pod of whoever owns them. // Joins and leaves of one meeting point are processed one at a time (keyed-serializer.js), and the // order they reach that lock does not decide: under the lock, an act is applied only when it is newer — // by the time the participant's pod stamped on it — than the last act applied for that participant and // meeting point (`attendance-ledger.js`, persisted). Whatever order the notifications run in, the // participant ends in the state of her latest act. The collection is read when acting. module.exports = { name: 'registration', mixins: [PodActivitiesHandlerMixin], created() { this.serialize = createKeyedSerializer(); }, async started() { this.logger.info('Registration watcher subscribed'); }, methods: { /** * The attendees collection of a meeting point of `actorUri`'s pod, from the STORED meeting point. * It is attached asynchronously when the meeting point is created (attendees service), so a * Join/Leave processed in that window sees none: the meeting point is read again a bounded number * of times until it names one; only then give up. The collection must lie in the same pod. */ async attendeesCollectionOf(ctx, meetingPoint, meetingPointUri, actorUri) { let collectionUri = linkOf(meetingPoint, 'apods:attendees', APODS_ATTENDEES); for (let attempt = 0; !collectionUri && attempt < 3; attempt++) { await new Promise(resolve => setTimeout(resolve, 500)); try { const { status, body } = await ctx.call('pod-resources.get', { resourceUri: meetingPointUri, actorUri }); if (status === 200 && body) collectionUri = linkOf(body, 'apods:attendees', APODS_ATTENDEES); } catch (e) { this.logger.warn(`Could not re-read ${meetingPointUri} while waiting for its attendees collection: ${e.message}`); } } if (collectionUri && !isInActorPod(collectionUri, actorUri)) { this.logger.warn(`${meetingPointUri} names an attendees collection outside its pod: ignored`); return undefined; } return collectionUri; }, /** * Applies one participant act (`join` / `leave`, the stored activity `activity`) to the meeting * point's collection — under the meeting point's lock, and only when the act is newer than the last * one applied for that participant there (see the header). A stale act changes nothing; an act * without a pod stamp cannot be placed and changes nothing. The decision is recorded once the * collection matches it. Gives up before any write once its time is out (`signal`). * * KNOWN LIMIT — a write in flight at the timeout: an add or remove ALREADY SENT when the time ran * out still lands, and is not recorded. If a newer act of the same participant was applied in the * meantime (on a read taken before the late write landed), the collection shows the older act's * effect until the participant acts again. It is deliberately NOT corrected from the ledger * afterwards: when nothing superseded the late act (a rejoin, say), the late write is right and it * is the ledger that is behind — re-applying the ledger would undo it, and the two cases cannot be * told apart here. Needs a pod slower than the 30 s limit. */ async updateAttendance(ctx, { change, memberUri, meetingPointUri, meetingPoint, activity, actorUri }) { const act = { activityUri: uriOf(activity), published: publishedOf(activity), change }; if (Number.isNaN(Date.parse(act.published))) { this.logger.warn(`${change} ${act.activityUri} of ${memberUri} carries no pod stamp (published): it cannot be ordered, unchanged`); return; } await this.serialize(meetingPointUri, async signal => { const dataset = CONFIG.MAIN_DATASET; const last = await lastAttendance(ctx, { dataset, meetingPointUri, memberUri }); if (!isNewerAct(act, last)) { this.logger.info(`${change} ${act.activityUri} of ${memberUri} on ${meetingPointUri} is not newer than the last act applied (${last.activityUri}, possibly the same act notified again): unchanged`); return; } const collectionUri = await this.attendeesCollectionOf(ctx, meetingPoint, meetingPointUri, actorUri); if (!collectionUri) { this.logger.warn(`No attendees collection on ${meetingPointUri}, skipping ${change}`); return; } if (signal.aborted) return; const attending = arrayOf(await ctx.call('pod-collections.getItems', { collectionUri, actorUri })).includes(memberUri); if (signal.aborted) return; if (change === 'join' && !attending) { await ctx.call('attendees.add', { collectionUri, itemUri: memberUri, actorUri }); } else if (change === 'leave' && attending) { await ctx.call('attendees.remove', { collectionUri, itemUri: memberUri, actorUri }); } else { this.logger.info(`${memberUri} already ${change === 'join' ? 'attends' : 'does not attend'} ${meetingPointUri}`); } if (signal.aborted) return; await recordAttendance(ctx, { dataset, meetingPointUri, memberUri, act }); }); } }, activities: { join: { match: neverThrowingMatcher(function (activity, fetcher) { return matchActivity(JOIN_PATTERN, activity, fetcher); }), // The joiner's outbox: `actorUri` is the joiner. The meeting point lies in the organizer's pod. onEmit: verifiedEmit( 'Join', async function (ctx, { object: meetingPoint, objectUri: meetingPointUri }, actorUri) { const organizerUri = linkOf(meetingPoint, 'dc:creator', DC_CREATOR); if (!organizerUri || !isInActorPod(meetingPointUri, organizerUri)) { this.logger.warn(`${meetingPointUri} is not in the pod of the creator it names: no contact added`); return; } const { body: actor } = await ctx.call('pod-resources.get', { resourceUri: actorUri, actorUri }); const contactsUri = actor && linkOf(actor, 'apods:contacts', APODS_CONTACTS); if (!contactsUri) return; if (!isInActorPod(contactsUri, actorUri)) { this.logger.warn(`${actorUri} names a contacts collection outside its pod: no contact added`); return; } const contactsGroupUri = await ctx.call('pod-wac-groups.getUriFromCollectionUri', { collectionUri: contactsUri }); await ctx.call('pod-wac-groups.addMember', { groupUri: contactsGroupUri, memberUri: organizerUri, actorUri }); }, { objectType: 'Event', objectIn: 'any' } ), // The organizer's inbox: `actorUri` is the organizer; the meeting point must lie in her pod. onReceive: verifiedReceive( 'Join', async function (ctx, { activity, emitterUri, object, objectUri }, actorUri) { await this.updateAttendance(ctx, { change: 'join', memberUri: emitterUri, meetingPointUri: objectUri, meetingPoint: object, activity, actorUri }); }, { objectType: 'Event' } ) }, leave: { match: neverThrowingMatcher(function (activity, fetcher) { return matchActivity(LEAVE_PATTERN, activity, fetcher); }), onReceive: verifiedReceive( 'Leave', async function (ctx, { activity, emitterUri, object, objectUri }, actorUri) { await this.updateAttendance(ctx, { change: 'leave', memberUri: emitterUri, meetingPointUri: objectUri, meetingPoint: object, activity, actorUri }); }, { objectType: 'Event' } ) } } };