mirror of
https://github.com/wekan/wekan.git
synced 2026-11-07 11:02:33 -05:00
Lauri Ojansivu
2df3e9a8b7
Durable delivery identified every activity by a card and a list, so a rule's createSwimlane activity, which names neither, and a cross-board move's moveCardBoard activity, which names no list, were written once but delivered the ordinary way: their notifications and webhooks were not replayed after a crash. Durable delivery now takes activities without a list, and without a card. Recipients and webhook senders are checked against the board, plus the card when one is named. A card or list the activity does not name is refused. An assigned-only member never receives a board-level event, since its scope cannot be checked against their assignments. The rule stage has nothing to run for these types, because no trigger exists for them, and it refuses a list-less activity whose type has one. Thanks to xet7 !
48 lines
2.4 KiB
JavaScript
48 lines
2.4 KiB
JavaScript
'use strict';
|
|
const { EJSON } = require('bson');
|
|
const { validateSyncEffectPolicy, assertSyncEffectPolicy } = require('./syncEffectPolicy');
|
|
const { reuseWithinEvaluation } = require('./syncGuardWindow');
|
|
const { notificationActivityIdentity, planId: notificationId } = require('./syncNotificationPlan');
|
|
const { planId: webhookId } = require('./syncWebhookPlan');
|
|
const copy = value => EJSON.parse(EJSON.stringify(value), { relaxed: true });
|
|
// These adapters must reconcile their own durable receipts on replay. This
|
|
// coordinator never treats activity insertion or a resolved Promise alone as
|
|
// acknowledgement. Rules are mandatory even when notifications are disabled.
|
|
// `trigger` (manual or scheduled) reaches every stage, which checks the Sync
|
|
// activation switches itself (server/lib/syncActivation.js).
|
|
async function deliverSyncActivity({ effectId, activity, policy, assertCurrent, readPolicy,
|
|
rules, notifications, webhooks, trigger }) {
|
|
if (typeof effectId !== 'string' || !/^[a-f0-9]{64}$/.test(effectId) ||
|
|
typeof assertCurrent !== 'function' || typeof readPolicy !== 'function') throw new Error('sync-activity-delivery-invalid');
|
|
const saved = copy(activity), captured = validateSyncEffectPolicy(policy);
|
|
notificationActivityIdentity(saved);
|
|
// A board-level activity names no list (2026-10-03); one that does names it
|
|
// properly.
|
|
if (saved.listId !== undefined && (typeof saved.listId !== 'string' || !saved.listId || saved.listId.length > 1024)) {
|
|
throw new Error('sync-activity-delivery-invalid');
|
|
}
|
|
if (!captured.activities || typeof rules !== 'function' ||
|
|
(captured.notifications && (typeof notifications !== 'function' || typeof webhooks !== 'function'))) {
|
|
throw new Error('sync-activity-delivery-adapter-required');
|
|
}
|
|
const guard = reuseWithinEvaluation(async () => {
|
|
await assertCurrent();
|
|
await assertSyncEffectPolicy(captured, readPolicy);
|
|
await assertCurrent();
|
|
});
|
|
const run = async (adapter, expected) => {
|
|
await guard();
|
|
const receipt = await adapter({ effectId, activity: copy(saved), policy: { ...captured }, assertCurrent: guard, trigger });
|
|
if (receipt !== expected) throw new Error('sync-activity-delivery-unconfirmed');
|
|
await guard();
|
|
};
|
|
await run(rules, effectId);
|
|
if (captured.notifications) {
|
|
await run(notifications, notificationId(saved._id));
|
|
await run(webhooks, webhookId(saved._id));
|
|
}
|
|
await guard();
|
|
return effectId;
|
|
}
|
|
module.exports = { deliverSyncActivity };
|