festipod/backend/services/registration.service.js

const { ACTIVITY_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, persistMember, recordAttendance } = require('./attendance-ledger');
const { neverThrowingMatcher, verifiedEmit, singleUriOf, uriOf } = require('./emitted-activity');
const { participationSecretOf, memberUrn } = require('./member-urn');
const { readItems, expectItem } = require('./collection-items');

const AS = 'https://www.w3.org/ns/activitystreams#';
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';

// Routing matches the activity TYPE alone: routing is a hint, the verified handler re-reads the
// activity and its object from the pod and decides (see emitted-activity.js).
const JOIN_PATTERN = { type: ACTIVITY_TYPES.JOIN };
const LEAVE_PATTERN = { type: ACTIVITY_TYPES.LEAVE };

/** 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 a node names under the first of `keys` it carries (compacted, CURIE or full IRI). */
const linkOf = (node, ...keys) => singleUriOf(keys.map(key => node[key]).find(value => value !== undefined));

/** The first string a property holds (a string, a value node, or a list of either). */
const firstStringOf = value =>
  arrayOf(value)
    .map(v => (typeof v === 'string' ? v : v && typeof v === 'object' ? v['@value'] : undefined))
    .find(v => typeof v === 'string' && v.trim().length > 0);

/**
 * What a Join/Leave targets: the stored object must be a meeting point — an `as:Event` (checked by
 * `verifiedEmit`) anchored to an event by `as:context` — whose creator is its `dc:creator`, believed
 * only because 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). Resolves `{ ok, eventUri, creatorUri }` or a refusal.
 */
function meetingPointTarget(meetingPoint, meetingPointUri) {
  const eventUri = linkOf(meetingPoint, 'context', 'as:context', `${AS}context`);
  if (!eventUri) return { ok: false, reason: `the meeting point ${meetingPointUri} names no single as:context (its event)` };
  const creatorUri = linkOf(meetingPoint, 'dc:creator', DC_CREATOR);
  if (!creatorUri) return { ok: false, reason: `the meeting point ${meetingPointUri} names no single dc:creator` };
  if (!isInActorPod(meetingPointUri, creatorUri)) {
    return { ok: false, reason: `the meeting point ${meetingPointUri} is not in the pod of its creator ${creatorUri}` };
  }
  return { ok: true, eventUri, creatorUri };
}

/**
 * The notification template for the creator of a meeting point (`pod-notifications.send`, rendered in
 * the creator's language): always ANONYMOUS, the meeting point named through `meetingPointName`, the
 * link to its event's meeting points (`eventUri`). Every value travels as a template PARAMETER, never
 * inside the template, so no user text is compiled.
 *
 * Never names the participant: naming would need a trustworthy signal that the creator is one of the
 * participant's contacts, and the only one the app can read — the participant's `apods:contacts` list —
 * is forgeable (a stranger's received `Accept` of a contact `Offer` is added to it with no prior
 * request checked; platform defect reported upstream). Until that list can be trusted, every
 * notification is anonymous.
 */
function notificationTemplate(change) {
  const title =
    change === 'join'
      ? {
          en: 'A participant joined your meeting point "{{{meetingPointName}}}"',
          fr: 'Un participant a rejoint votre point de rencontre « {{{meetingPointName}}} »'
        }
      : {
          en: 'A participant left your meeting point "{{{meetingPointName}}}"',
          fr: 'Un participant a quitté votre point de rencontre « {{{meetingPointName}}} »'
        };
  return {
    title,
    actions: [{ caption: { en: 'View', fr: 'Voir' }, link: '/events/{{encodeUri eventUri}}/meeting-points' }]
  };
}

// Participation (decision_2026-10-29_participation-privacy-mechanism): a person joins (leaves) a
// meeting point with an `as:Join` (`as:Leave`) posted to their OWN outbox, naming the meeting point,
// addressed to no one — so only its author and the app read it. The creator's inbox receives nothing.
// The backend acts on EMISSION (`onEmit`, the welcometomyplace registration pattern), and records:
//   - in the meeting point's `apods:attendees` collection (the creator's pod, acting as the creator):
//     the participant's MEMBER URN, never their WebID — `member-urn.js`, computed from the secret in
//     their profile. A Join whose author's profile carries no usable secret is refused, nothing written;
//   - in the participant's participations collection (their pod, acting as the participant): the
//     meeting point URI — the collection created at the first Join and kept readable by the
//     participant's `/contacts` WAC group only (`participations.service.js`).
// Then the creator is notified ANONYMOUSLY — always: naming would need a trustworthy "is the creator a
// contact of the participant" signal, and the only one the app can read (the participant's
// `apods:contacts` list) is forgeable (platform defect reported upstream, see `notificationTemplate`).
// The creator has no other right on the participant: they learn a count and opaque URNs.
// The webhook that reports the act is a hint (emitted-activity.js): the handler acts only on the
// activity and the meeting point as the pod stores them, and only for the people they really name:
//   - the participant is the owner of the outbox the webhook fired on, and the stored activity's actor;
//   - the creator is the meeting point's `dc:creator`, accepted only when the meeting point lies in
//     that creator's pod (`meetingPointTarget`);
//   - the collections written to are the ones the STORED meeting point and the participant's STORED
//     profile 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, which also holds the member URN that act applied). Whatever order the notifications run in, the
// participant ends in the state of their latest act. The member URN is STICKY per (meeting point,
// participant): chosen once and persisted BEFORE any attendee-list write, so every later act reuses it —
// the pair is never listed under two URNs, even after a half-way failure and a secret rewrite. An act is
// recorded only once BOTH collections, READ BACK, show it (collection-items.js — the platform's own
// collection calls answer "empty" on a failed read and resolve on a refused write); one that fails
// half-way is not, nor notified, and the participant's next act on that meeting point re-applies it
// (the two writes are not atomic).
module.exports = {
  name: 'registration',
  mixins: [PodActivitiesHandlerMixin],
  dependencies: ['participations'],
  created() {
    this.serialize = createKeyedSerializer();
  },
  async started() {
    this.logger.info('Registration watcher subscribed');
  },
  methods: {
    /**
     * Handles one verified Join/Leave (`change`: 'join' | 'leave') of `participantUri`, the outbox
     * owner: resolves the meeting point and the participant, applies the act, then notifies the
     * creator when the act changed the participant's state. A refusal is logged; nothing is written.
     */
    async onParticipation(ctx, change, { activity, objectUri: meetingPointUri, object: meetingPoint }, participantUri) {
      const refuse = reason => this.logger.warn(`${this.name} refused the ${change} ${uriOf(activity)} of ${participantUri}: ${reason}`);
      const target = meetingPointTarget(meetingPoint, meetingPointUri);
      if (!target.ok) return refuse(target.reason);
      const participant = await this.participantOf(ctx, participantUri);
      if (!participant.ok) return refuse(participant.reason);
      const outcome = await this.updateAttendance(ctx, {
        change,
        eventUri: target.eventUri,
        secret: participant.secret,
        meetingPointUri,
        meetingPoint,
        activity,
        creatorUri: target.creatorUri,
        participantUri,
        profileUri: participant.profileUri,
        contactsUri: participant.contactsUri
      });
      if (outcome && outcome.changed) {
        await this.notifyCreator(ctx, {
          change,
          meetingPointUri,
          meetingPoint,
          eventUri: target.eventUri,
          creatorUri: target.creatorUri,
          participantUri
        });
      }
    },
    /**
     * The participant as their pod describes them, read as the participant: their actor document
     * (its `url` — the `as:Profile`, which must lie in their pod — and `apods:contacts`), then the
     * profile, which must carry a usable participation secret. Resolves `{ ok, profileUri,
     * contactsUri, secret }` or a refusal. `contactsUri` is the participant's `apods:contacts`
     * collection — whose `/contacts` WAC group is what may read their participations collection.
     */
    async participantOf(ctx, participantUri) {
      const read = async uri => {
        const { status, body } = await ctx.call('pod-resources.get', { resourceUri: uri, actorUri: participantUri });
        return status === 200 && body && uriOf(body) === uri ? body : undefined;
      };
      const actor = await read(participantUri);
      if (!actor) return { ok: false, reason: `the pod does not return the actor ${participantUri}` };
      const profileUri = linkOf(actor, 'url', 'as:url', `${AS}url`);
      if (!profileUri || !isInActorPod(profileUri, participantUri)) {
        return { ok: false, reason: `${participantUri} names no single profile in their pod (${profileUri})` };
      }
      const profile = await read(profileUri);
      if (!profile) return { ok: false, reason: `the pod does not return the profile ${profileUri}` };
      const secret = participationSecretOf(profile);
      if (!secret) return { ok: false, reason: `the profile ${profileUri} carries no usable participation secret` };
      const contactsUri = linkOf(actor, 'apods:contacts', APODS_CONTACTS);
      return { ok: true, profileUri, contactsUri, secret };
    },
    /**
     * 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;
    },
    /**
     * The participant's participations collection, under the meeting point's lock: for a Join, the one
     * `participations.ensure` makes ready (created if none, contacts-only); for a Leave, the one the
     * profile links now, if any. Resolves `{ ok, collectionUri }` — `collectionUri` undefined for a
     * Leave without a collection — or `{ ok: false }` once the failure is logged: the act must then
     * write nothing and stay unrecorded, so that the next act retries.
     */
    async participationsCollectionOf(ctx, { change, participantUri, profileUri, contactsUri }) {
      try {
        const collectionUri =
          change === 'join'
            ? await ctx.call('participations.ensure', { actorUri: participantUri, profileUri, contactsUri })
            : await ctx.call('participations.collectionOf', { actorUri: participantUri, profileUri });
        return { ok: true, collectionUri };
      } catch (e) {
        this.logger.error(
          `The participations collection of ${participantUri} is not ready (${e.message}): the ${change} is NOT recorded, the next act retries`
        );
        return { ok: false };
      }
    },
    /**
     * Applies one participant act (`join` / `leave`, the stored activity `activity`) — under the
     * meeting point's lock, and only when the act is newer than the last one applied for that member
     * there (see the header). A stale act changes nothing; an act without a pod stamp cannot be placed
     * and changes nothing. The act is applied to the meeting point's attendees (as the creator), then
     * to the participant's participations collection (as the participant), each write read back
     * (`applyToAttendees`, `applyToParticipations`), and recorded once both show it; a failed read, a
     * refused write or a write that does not show leaves the act unrecorded (logged). Gives up before
     * any write once its time is out (`signal`). Resolves `{ changed }` when the act was applied —
     * `changed` when it added or removed the member, or flipped the state the ledger held — and
     * undefined when it was not.
     *
     * 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, eventUri, secret, meetingPointUri, meetingPoint, activity, creatorUri, participantUri, profileUri, contactsUri }) {
      const act = { activityUri: uriOf(activity), published: publishedOf(activity), change };
      if (Number.isNaN(Date.parse(act.published))) {
        this.logger.warn(`${change} ${act.activityUri} of ${participantUri} carries no pod stamp (published): it cannot be ordered, unchanged`);
        return undefined;
      }
      return this.serialize(meetingPointUri, async signal => {
        const dataset = CONFIG.MAIN_DATASET;
        const last = await lastAttendance(ctx, { dataset, meetingPointUri, participantUri });
        if (!isNewerAct(act, last)) {
          this.logger.info(`${change} ${act.activityUri} of ${participantUri} on ${meetingPointUri} is not newer than the last act applied (${last.activityUri}, possibly the same act notified again): unchanged`);
          return undefined;
        }
        const collectionUri = await this.attendeesCollectionOf(ctx, meetingPoint, meetingPointUri, creatorUri);
        if (!collectionUri) {
          this.logger.warn(`No attendees collection on ${meetingPointUri}, skipping ${change}`);
          return undefined;
        }
        if (signal.aborted) return undefined;
        const participations = await this.participationsCollectionOf(ctx, { change, participantUri, profileUri, contactsUri });
        if (!participations.ok || signal.aborted) return undefined;
        // The member URN the act applies is STICKY per (meeting point, participant): the one the ledger
        // already holds for the pair, else one computed from the secret on the profile now and persisted
        // write-once BEFORE any attendee-list write — the act then applies whatever member the ledger holds
        // after that write (another act's, if one landed first) — so that every later act (Join or Leave,
        // after any half-way failure, timeout or secret rewrite) uses that same URN and the pair is never
        // listed under two.
        if (signal.aborted) return undefined;
        let wrote;
        try {
          const memberUri =
            last && last.member
              ? last.member
              : await persistMember(ctx, { dataset, meetingPointUri, participantUri, member: memberUrn({ participantUri, eventUri, secret }) });
          if (signal.aborted) return undefined;
          wrote = await this.applyToAttendees(ctx, { change, collectionUri, memberUri, creatorUri, meetingPointUri, signal });
          await this.applyToParticipations(ctx, { change, collectionUri: participations.collectionUri, meetingPointUri, participantUri, signal });
        } catch (e) {
          this.logger.error(`The ${change} ${act.activityUri} of ${participantUri} on ${meetingPointUri} is NOT recorded (${e.message}): the next act retries`);
          return undefined;
        }
        if (signal.aborted) return undefined;
        await recordAttendance(ctx, { dataset, meetingPointUri, participantUri, act });
        const wasAttending = Boolean(last && last.change === 'join');
        return { changed: wrote || wasAttending !== (change === 'join') };
      });
    },
    /**
     * Applies `change` for `memberUri` to the meeting point's attendee list `collectionUri`, in the
     * creator's pod and as the creator: reads the list — a failed read throws, it never means "not
     * listed" — then adds the member on a Join when absent, removes it on a Leave when present (the
     * attendees handler's `add` / `remove`, the reference-app calls) and reads the list back: throws
     * unless the write shows. Resolves whether it wrote (false when the list already matched the act).
     * Throws instead of writing once its time is out (`signal`).
     */
    async applyToAttendees(ctx, { change, collectionUri, memberUri, creatorUri, meetingPointUri, signal }) {
      const joining = change === 'join';
      const attending = (await readItems(ctx, { collectionUri, actorUri: creatorUri })).includes(memberUri);
      if (attending === joining) {
        this.logger.info(`${memberUri} already ${joining ? 'attends' : 'does not attend'} ${meetingPointUri}`);
        return false;
      }
      if (signal.aborted) throw new Error(`out of time before the ${change} write to ${collectionUri}`);
      await ctx.call(joining ? 'attendees.add' : 'attendees.remove', { collectionUri, itemUri: memberUri, actorUri: creatorUri });
      await expectItem(ctx, { collectionUri, itemUri: memberUri, actorUri: creatorUri, present: joining });
      return true;
    },
    /**
     * Applies `change` to the participant's participations collection `collectionUri`, in their pod and
     * as them: adds the meeting point on a Join, removes it on a Leave (the participations handler's
     * `add` / `remove`, the reference-app calls — adding an item already there or removing one that is
     * not changes nothing, so no read first), then reads the collection back: throws unless it shows
     * the act. Nothing to do on a Leave when the participant has no collection (`collectionUri`
     * undefined). Throws instead of writing once its time is out (`signal`).
     */
    async applyToParticipations(ctx, { change, collectionUri, meetingPointUri, participantUri, signal }) {
      if (!collectionUri) {
        this.logger.info(`${participantUri} has no participations collection: nothing to remove ${meetingPointUri} from`);
        return;
      }
      const joining = change === 'join';
      if (signal.aborted) throw new Error(`out of time before the ${change} write to ${collectionUri}`);
      await ctx.call(joining ? 'participations.add' : 'participations.remove', { collectionUri, itemUri: meetingPointUri, actorUri: participantUri });
      await expectItem(ctx, { collectionUri, itemUri: meetingPointUri, actorUri: participantUri, present: joining });
    },
    /**
     * Tells the creator of the meeting point that a participant joined or left it — ALWAYS anonymously.
     * The participant is never named: the only membership signal the app can read (the participant's
     * `apods:contacts` list) is forgeable, so naming cannot be trusted (platform defect reported
     * upstream, see `notificationTemplate`). The notification carries no `activity`: given one,
     * `pod-notifications.send` fetches the emitter's profile. A creator acting on their own meeting
     * point is not notified. A failure is logged, never thrown: the act stands.
     */
    async notifyCreator(ctx, { change, meetingPointUri, meetingPoint, eventUri, creatorUri, participantUri }) {
      try {
        if (creatorUri === participantUri) {
          this.logger.info(`${participantUri} acted on their own meeting point ${meetingPointUri}: no notification`);
          return;
        }
        this.logger.info(`Notifying ${creatorUri} of a ${change} on ${meetingPointUri}, anonymously`);
        await ctx.call('pod-notifications.send', {
          template: notificationTemplate(change),
          recipientUri: creatorUri,
          context: meetingPointUri,
          meetingPointName: firstStringOf(meetingPoint.name ?? meetingPoint['as:name'] ?? meetingPoint[`${AS}name`]) ?? '',
          eventUri
        });
      } catch (e) {
        this.logger.error(`Could not notify ${creatorUri} of a ${change} on ${meetingPointUri}: ${e.message}`);
      }
    }
  },
  activities: {
    join: {
      match: neverThrowingMatcher(function (activity, fetcher) {
        return matchActivity(JOIN_PATTERN, activity, fetcher);
      }),
      // The participant's outbox: `actorUri` is the participant; the meeting point may lie in any pod
      // (its creator's — checked by `meetingPointTarget`).
      onEmit: verifiedEmit(
        'Join',
        async function (ctx, verdict, actorUri) {
          await this.onParticipation(ctx, 'join', verdict, actorUri);
        },
        { objectType: 'Event', objectIn: 'any' }
      )
    },
    leave: {
      match: neverThrowingMatcher(function (activity, fetcher) {
        return matchActivity(LEAVE_PATTERN, activity, fetcher);
      }),
      onEmit: verifiedEmit(
        'Leave',
        async function (ctx, verdict, actorUri) {
          await this.onParticipation(ctx, 'leave', verdict, actorUri);
        },
        { objectType: 'Event', objectIn: 'any' }
      )
    }
  }
};