238 lines
8.8 KiB
JavaScript
238 lines
8.8 KiB
JavaScript
function createLaterSync({ storage, getLogin, fetchJson, onRemoteRecords, onStatus, createOperationId, createChannel, coordinator,
|
|
setTimer = globalThis.setTimeout, clearTimer = globalThis.clearTimeout, retryBaseMs = 1000, retryMaxMs = 30000 }) {
|
|
const prefix = 'stackchain.later-sync.v1.';
|
|
const migrationPrefix = 'stackchain.later-sync-migrated.v1.';
|
|
const snapshotPrefix = 'stackchain.later-sync-snapshot.v1.';
|
|
let flushing = null;
|
|
let channel = null;
|
|
let channelKey = '';
|
|
let retryTimer = null;
|
|
let retryAttempt = 0;
|
|
const knownOperationKeys = new Set();
|
|
|
|
function cancelRetry() {
|
|
if (retryTimer !== null) clearTimer?.(retryTimer);
|
|
retryTimer = null;
|
|
}
|
|
|
|
function scheduleRetry(error, ownerKey) {
|
|
if (retryTimer !== null || !pending().length || !ownerKey) return;
|
|
const advised = Number(error?.retryAfter);
|
|
const delayMs = Number.isFinite(advised) && advised >= 0
|
|
? advised * 1000
|
|
: Math.min(retryMaxMs, retryBaseMs * (2 ** retryAttempt));
|
|
retryAttempt += 1;
|
|
onStatus?.('retrying', { delayMs });
|
|
retryTimer = setTimer?.(async () => {
|
|
retryTimer = null;
|
|
if (key() !== ownerKey) return false;
|
|
return flush();
|
|
}, delayMs);
|
|
retryTimer?.unref?.();
|
|
}
|
|
|
|
function key() {
|
|
const login = String(getLogin?.() || '').trim().toLowerCase();
|
|
return login ? prefix + encodeURIComponent(login) : '';
|
|
}
|
|
|
|
function snapshotKey() {
|
|
const storageKey = key();
|
|
return storageKey ? snapshotPrefix + storageKey.slice(prefix.length) : '';
|
|
}
|
|
|
|
function savedRevision() {
|
|
try {
|
|
const snapshot = JSON.parse(storage?.getItem(snapshotKey()) || 'null');
|
|
return Number.isInteger(snapshot?.revision) ? snapshot.revision : -1;
|
|
} catch (_error) {
|
|
return -1;
|
|
}
|
|
}
|
|
|
|
function validRecords(records) {
|
|
return records && typeof records === 'object' && !Array.isArray(records);
|
|
}
|
|
|
|
function adopt(plan, broadcast = true) {
|
|
if (!Number.isInteger(plan?.revision) || !validRecords(plan?.records)) return false;
|
|
if (plan.revision < savedRevision()) return false;
|
|
const snapshot = { revision: plan.revision, records: plan.records };
|
|
try {
|
|
storage?.setItem(snapshotKey(), JSON.stringify(snapshot));
|
|
} catch (_error) {
|
|
// Server truth remains usable in this tab when storage is unavailable.
|
|
}
|
|
onRemoteRecords?.(plan.records);
|
|
if (broadcast) channel?.postMessage(snapshot);
|
|
return true;
|
|
}
|
|
|
|
function ensureChannel() {
|
|
const storageKey = key();
|
|
if (!storageKey || channelKey === storageKey) return;
|
|
channel?.close?.();
|
|
const factory = createChannel || (globalThis.window?.BroadcastChannel
|
|
? name => new globalThis.window.BroadcastChannel(name)
|
|
: null);
|
|
channelKey = storageKey;
|
|
channel = factory?.('stackchain-later-' + storageKey.slice(prefix.length)) || null;
|
|
channel?.addEventListener?.('message', event => {
|
|
if (key() === storageKey) adopt(event.data, false);
|
|
});
|
|
}
|
|
|
|
function pending() {
|
|
const storageKey = key();
|
|
if (!storageKey || !storage) return [];
|
|
const recordPrefix = storageKey + '.operation.';
|
|
try {
|
|
const legacy = JSON.parse(storage.getItem(storageKey) || '[]');
|
|
if (Array.isArray(legacy)) {
|
|
legacy.forEach((operation, index) => {
|
|
if (!operation?.operation_id) return;
|
|
const recordKey = recordPrefix + encodeURIComponent(operation.operation_id);
|
|
storage.setItem(recordKey, JSON.stringify({ operation, queued_at: index }));
|
|
knownOperationKeys.add(recordKey);
|
|
});
|
|
if (legacy.length) storage.removeItem(storageKey);
|
|
}
|
|
const keys = new Set([...knownOperationKeys].filter(candidate => candidate.startsWith(recordPrefix)));
|
|
for (let index = 0; index < Number(storage.length || 0); index += 1) {
|
|
const candidate = storage.key?.(index);
|
|
if (candidate?.startsWith(recordPrefix)) keys.add(candidate);
|
|
}
|
|
const records = [...keys].map(recordKey => {
|
|
const record = JSON.parse(storage.getItem(recordKey) || 'null');
|
|
return record ? { ...record, recordKey } : null;
|
|
}).filter(record => record?.operation)
|
|
.sort((left, right) => Number(left.queued_at || 0) - Number(right.queued_at || 0) ||
|
|
left.operation.operation_id.localeCompare(right.operation.operation_id));
|
|
const latestByItem = new Map();
|
|
records.forEach(record => latestByItem.set(record.operation.item_id, record));
|
|
records.filter(record => latestByItem.get(record.operation.item_id) !== record).forEach(record => {
|
|
storage.removeItem(record.recordKey);
|
|
knownOperationKeys.delete(record.recordKey);
|
|
});
|
|
return records.filter(record => latestByItem.get(record.operation.item_id) === record)
|
|
.map(record => record.operation).filter(operation =>
|
|
operation && typeof operation.operation_id === 'string' &&
|
|
['defer', 'restore'].includes(operation.action) && typeof operation.item_id === 'string' &&
|
|
(operation.action === 'restore' || typeof operation.wake_at === 'string'));
|
|
} catch (_error) {
|
|
return [];
|
|
}
|
|
}
|
|
|
|
function removeOperation(operationId) {
|
|
const storageKey = key();
|
|
if (!storageKey || !storage) return false;
|
|
const recordKey = storageKey + '.operation.' + encodeURIComponent(operationId);
|
|
try {
|
|
storage.removeItem(recordKey);
|
|
knownOperationKeys.delete(recordKey);
|
|
return true;
|
|
} catch (_error) {
|
|
return false;
|
|
}
|
|
}
|
|
|
|
function operationId() {
|
|
if (createOperationId) return createOperationId();
|
|
if (globalThis.crypto?.randomUUID) return globalThis.crypto.randomUUID();
|
|
return Date.now().toString(36) + '-' + Math.random().toString(36).slice(2);
|
|
}
|
|
|
|
function enqueue(action, itemId, wakeAt = null) {
|
|
if (!['defer', 'restore'].includes(action) || !itemId ||
|
|
(action === 'defer' && typeof wakeAt !== 'string')) return false;
|
|
pending().filter(operation => operation.item_id === itemId)
|
|
.forEach(operation => removeOperation(operation.operation_id));
|
|
const operation = { operation_id: operationId(), action, item_id: itemId, wake_at: wakeAt };
|
|
const storageKey = key();
|
|
if (!storageKey || !storage) return false;
|
|
const recordKey = storageKey + '.operation.' + encodeURIComponent(operation.operation_id);
|
|
let saved = false;
|
|
try {
|
|
storage.setItem(recordKey, JSON.stringify({ operation, queued_at: Date.now() }));
|
|
knownOperationKeys.add(recordKey);
|
|
saved = true;
|
|
coordinator?.notify('later');
|
|
} catch (_error) { /* Report the persistence failure below. */ }
|
|
onStatus?.(saved ? 'pending' : 'error');
|
|
return saved;
|
|
}
|
|
|
|
function migrate(records) {
|
|
const storageKey = key();
|
|
if (!storageKey || !storage) return false;
|
|
const marker = migrationPrefix + storageKey.slice(prefix.length);
|
|
try {
|
|
if (storage.getItem(marker)) return false;
|
|
Object.entries(records || {}).forEach(([itemId, wakeAt]) =>
|
|
enqueue('defer', itemId, wakeAt)
|
|
);
|
|
storage.setItem(marker, '1');
|
|
return true;
|
|
} catch (_error) {
|
|
return false;
|
|
}
|
|
}
|
|
|
|
async function run() {
|
|
const ownerKey = key();
|
|
if (!ownerKey) return false;
|
|
ensureChannel();
|
|
try {
|
|
let plan = await fetchJson('api/v1/later');
|
|
let operations = pending();
|
|
while (operations.length) {
|
|
if (key() !== ownerKey) return false;
|
|
const operation = operations[0];
|
|
plan = await fetchJson('api/v1/later', {
|
|
method: 'PATCH',
|
|
headers: { 'Content-Type': 'application/json' },
|
|
body: JSON.stringify(operation),
|
|
});
|
|
if (pending().some(candidate => candidate.operation_id === operation.operation_id) &&
|
|
!removeOperation(operation.operation_id)) {
|
|
throw new Error('Could not persist Later delivery receipt');
|
|
}
|
|
operations = pending();
|
|
}
|
|
adopt(plan);
|
|
onStatus?.(pending().length ? 'pending' : 'saved');
|
|
retryAttempt = 0;
|
|
cancelRetry();
|
|
return true;
|
|
} catch (error) {
|
|
if (pending().length) scheduleRetry(error, ownerKey);
|
|
else onStatus?.('error');
|
|
return false;
|
|
}
|
|
}
|
|
|
|
function flush() {
|
|
if (!flushing) {
|
|
const delivery = coordinator ? coordinator.runExclusive('later', run) : run();
|
|
flushing = Promise.resolve(delivery).finally(() => { flushing = null; });
|
|
}
|
|
return flushing;
|
|
}
|
|
|
|
function startLifecycle({ window: windowObject, document: documentObject }) {
|
|
windowObject?.addEventListener?.('online', flush);
|
|
documentObject?.addEventListener?.('visibilitychange', () =>
|
|
documentObject.hidden ? false : flush()
|
|
);
|
|
}
|
|
|
|
coordinator?.subscribe(change => {
|
|
if (change.queue === 'later' && pending().length) flush();
|
|
});
|
|
|
|
return { enqueue, migrate, flush, pending, startLifecycle };
|
|
}
|
|
|
|
if (typeof module !== 'undefined' && module.exports) module.exports = createLaterSync;
|