festipod/backend/services/registration.service.js

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