festipod/backend/services/attendance-ledger.js

// The last participation act applied for each (meeting point, participant) — what makes attendance
// independent of the order notifications arrive in.
//
// WHY. A participant's Join and Leave reach the backend as separate notifications, handled
// concurrently; the per-meeting-point lock (`keyed-serializer.js`) serializes them, but in the order
// their handlers reach it — after verification reads of unequal duration — not in the order the
// participant acted. "Join then Leave" could be applied as Leave (nothing to remove) then Join, and leave
// her listed. A Join or a Leave says what to do, not what the state should be, so the decision must
// compare it with what was already applied.
//
// THE RULE (`isNewerAct`). Each act is placed by the time the participant's pod stamped on it — its
// `published` (the pod overwrites any client value when the activity is posted to the outbox) — then by
// its URI, to break a tie. Under the lock, an act is applied only when it is newer than the last act
// applied for that pair; an older one (a late notification, a replay) changes nothing. Whatever the
// arrival order, the pair ends in the state of the participant's latest act: "join then leave" ends not
// attending, "leave then join" ends attending. All acts of one participant are stamped by her own pod,
// so they share one clock.
//
// WHERE IT LIVES. Like the catalog (`catalog-store.js`): in the backend's main dataset, DEFAULT graph
// (persisted by Fuseki), so the rule holds across a backend restart. One entry per pair, under a minted
// `urn:festipod:attendance:` subject, with three Festipod predicates: the act's URI, its stamp, its kind.

const { iriRef, stringLiteral } = require('./catalog-store');

const ACT = 'http://festipod.org/attendanceAct';
const PUBLISHED = 'http://festipod.org/attendancePublished';
const CHANGE = 'http://festipod.org/attendanceChange';

/** The subject of the (meeting point, participant) entry: both URIs percent-encoded into one URN. */
function attendanceSubject(meetingPointUri, memberUri) {
  return `urn:festipod:attendance:${encodeURIComponent(meetingPointUri)}:${encodeURIComponent(memberUri)}`;
}

/** The last act applied for the pair: one row (act, published, change) or none. */
function lastAttendanceQuery(meetingPointUri, memberUri) {
  const s = iriRef(attendanceSubject(meetingPointUri, memberUri));
  return `SELECT ?act ?published ?change WHERE { ${s} <${ACT}> ?act ; <${PUBLISHED}> ?published ; <${CHANGE}> ?change } LIMIT 1`;
}

/** Replace the pair's entry with `{ activityUri, published, change }`. One update = one transaction. */
function recordAttendanceUpdate(meetingPointUri, memberUri, { activityUri, published, change }) {
  const s = iriRef(attendanceSubject(meetingPointUri, memberUri));
  return `
    DELETE { ${s} ?p ?o }
    INSERT { ${s} <${ACT}> ${iriRef(activityUri)} ; <${PUBLISHED}> ${stringLiteral(published)} ; <${CHANGE}> ${stringLiteral(change)} }
    WHERE { OPTIONAL { ${s} ?p ?o . FILTER(?p IN (<${ACT}>, <${PUBLISHED}>, <${CHANGE}>)) } }
  `;
}

/**
 * isNewerAct(candidate, last) → boolean. `candidate` and `last` are `{ activityUri, published }`
 * (`last` undefined when nothing was applied yet). Newer = a later `published` instant, or the same
 * instant and a greater URI. A candidate without a readable stamp is never newer (it cannot be placed).
 */
function isNewerAct(candidate, last) {
  const at = Date.parse(candidate && candidate.published);
  if (Number.isNaN(at)) return false;
  if (!last) return true;
  const lastAt = Date.parse(last.published);
  if (Number.isNaN(lastAt) || at > lastAt) return true;
  if (at < lastAt) return false;
  return String(candidate.activityUri) > String(last.activityUri);
}

const termValue = term => (term && typeof term === 'object' ? term.value : term);

// ── Store operations: `ctx` is a moleculer context (anything with `call`) ──

/** The last act applied for the pair, `{ activityUri, published, change }`, or undefined. */
async function lastAttendance(ctx, { dataset, meetingPointUri, memberUri }) {
  const rows = await ctx.call('triplestore.query', { query: lastAttendanceQuery(meetingPointUri, memberUri), dataset, webId: 'system' });
  const row = Array.isArray(rows) ? rows[0] : undefined;
  if (!row) return undefined;
  return { activityUri: termValue(row.act), published: termValue(row.published), change: termValue(row.change) };
}

async function recordAttendance(ctx, { dataset, meetingPointUri, memberUri, act }) {
  await ctx.call('triplestore.update', { query: recordAttendanceUpdate(meetingPointUri, memberUri, act), dataset, webId: 'system' });
}

module.exports = {
  attendanceSubject,
  lastAttendanceQuery,
  recordAttendanceUpdate,
  isNewerAct,
  lastAttendance,
  recordAttendance
};