410 lines
16 KiB
JavaScript
410 lines
16 KiB
JavaScript
function createIndexedDbTransaction(indexedDB, dbName = 'stackchain-background-outbox-v1') {
|
|
let databasePromise;
|
|
function database() {
|
|
if (!databasePromise) databasePromise = new Promise((resolve, reject) => {
|
|
const request = indexedDB.open(dbName, 1);
|
|
request.onupgradeneeded = () => {
|
|
if (!request.result.objectStoreNames.contains('issues')) {
|
|
request.result.createObjectStore('issues', { keyPath: 'id' });
|
|
}
|
|
};
|
|
request.onsuccess = () => {
|
|
const db = request.result;
|
|
db.onversionchange = () => {
|
|
db.close();
|
|
databasePromise = undefined;
|
|
};
|
|
resolve(db);
|
|
};
|
|
request.onerror = () => reject(request.error);
|
|
});
|
|
return databasePromise;
|
|
}
|
|
const requested = request => new Promise((resolve, reject) => {
|
|
request.onsuccess = () => resolve(request.result);
|
|
request.onerror = () => reject(request.error);
|
|
});
|
|
const transact = async work => {
|
|
const db = await database();
|
|
return new Promise((resolve, reject) => {
|
|
const transaction = db.transaction('issues', 'readwrite');
|
|
const objectStore = transaction.objectStore('issues');
|
|
let result;
|
|
let failed = false;
|
|
transaction.oncomplete = () => failed ? undefined : resolve(result);
|
|
transaction.onerror = () => reject(transaction.error);
|
|
transaction.onabort = () => reject(transaction.error || new Error('Issue outbox transaction aborted'));
|
|
Promise.resolve(work({
|
|
getAll: () => requested(objectStore.getAll()),
|
|
put: value => requested(objectStore.put(value)),
|
|
delete: id => requested(objectStore.delete(id)),
|
|
})).then(value => { result = value; }).catch(error => {
|
|
failed = true;
|
|
try { transaction.abort(); } catch (_abortError) { reject(error); }
|
|
reject(error);
|
|
});
|
|
});
|
|
};
|
|
transact.close = async () => {
|
|
if (!databasePromise) return;
|
|
const db = await databasePromise;
|
|
db.close();
|
|
databasePromise = undefined;
|
|
};
|
|
return transact;
|
|
}
|
|
|
|
function createIssueSyncStore({ transaction, indexedDB = globalThis.indexedDB, now = () => Date.now(), claimMs = 30000 } = {}) {
|
|
const transact = transaction || createIndexedDbTransaction(indexedDB);
|
|
|
|
async function reconcile(items, outboxLane = 'issue') {
|
|
return transact(async records => {
|
|
const existing = await records.getAll();
|
|
const incoming = new Map(items.map(item => [item.id, { ...item, outboxLane }]));
|
|
for (const current of existing) {
|
|
if (current.recordType === 'receipt-preference') continue;
|
|
const currentLane = current.outboxLane || 'issue';
|
|
if (currentLane !== outboxLane) continue;
|
|
const replacement = incoming.get(current.id);
|
|
if (!replacement) {
|
|
if (current.status !== 'sending' || Number(current.claimUntil) <= Number(now())) {
|
|
await records.delete(current.id);
|
|
}
|
|
continue;
|
|
}
|
|
if ((current.status === 'sending' && Number(current.claimUntil) > Number(now())) ||
|
|
(current.status === 'attention' && replacement.status === 'attention') ||
|
|
current.status === 'sent') {
|
|
incoming.set(current.id, current);
|
|
}
|
|
}
|
|
for (const item of incoming.values()) await records.put(item);
|
|
});
|
|
}
|
|
|
|
async function claimNext(ownerLogin) {
|
|
return transact(async records => {
|
|
const timestamp = Number(now());
|
|
const items = await records.getAll();
|
|
const item = items.find(candidate => candidate.ownerLogin === ownerLogin &&
|
|
(candidate.status === 'queued' || candidate.status === 'sending') &&
|
|
(candidate.status !== 'sending' || Number(candidate.claimUntil) <= timestamp));
|
|
if (!item) return null;
|
|
const claimed = { ...item, status: 'sending', claimUntil: timestamp + claimMs };
|
|
await records.put(claimed);
|
|
return claimed;
|
|
});
|
|
}
|
|
|
|
async function claimBatch(ownerLogin, limit = 70) {
|
|
return transact(async records => {
|
|
const timestamp = Number(now());
|
|
const eligible = (await records.getAll()).filter(candidate =>
|
|
candidate.recordType !== 'receipt-preference' &&
|
|
candidate.ownerLogin === ownerLogin &&
|
|
(candidate.status === 'queued' || candidate.status === 'sending') &&
|
|
(candidate.status !== 'sending' || Number(candidate.claimUntil) <= timestamp));
|
|
const lanes = {
|
|
issue: eligible.filter(item => (item.outboxLane || 'issue') !== 'authored'),
|
|
authored: eligible.filter(item => item.outboxLane === 'authored'),
|
|
};
|
|
const selected = [];
|
|
const maximum = Math.max(0, Number(limit) || 0);
|
|
while (selected.length < maximum && (lanes.issue.length || lanes.authored.length)) {
|
|
if (lanes.issue.length) selected.push(lanes.issue.shift());
|
|
if (selected.length < maximum && lanes.authored.length) selected.push(lanes.authored.shift());
|
|
}
|
|
const claimed = selected.map(item => ({
|
|
...item, status: 'sending', claimUntil: timestamp + claimMs,
|
|
}));
|
|
for (const item of claimed) await records.put(item);
|
|
return claimed;
|
|
});
|
|
}
|
|
|
|
async function claim(id, ownerLogin) {
|
|
return transact(async records => {
|
|
const timestamp = Number(now());
|
|
const item = (await records.getAll()).find(candidate => candidate.id === id);
|
|
if (!item || item.ownerLogin !== ownerLogin ||
|
|
!['queued', 'sending'].includes(item.status) ||
|
|
(item.status === 'sending' && Number(item.claimUntil) > timestamp)) return null;
|
|
const claimed = { ...item, status: 'sending', claimUntil: timestamp + claimMs };
|
|
await records.put(claimed);
|
|
return claimed;
|
|
});
|
|
}
|
|
|
|
async function upsert(item) {
|
|
return transact(async records => {
|
|
const current = (await records.getAll()).find(candidate => candidate.id === item.id);
|
|
if (current && ((current.status === 'sending' && Number(current.claimUntil) > Number(now())) ||
|
|
(current.status === 'attention' && item.status === 'attention') ||
|
|
current.status === 'sent')) return current;
|
|
await records.put({ ...item });
|
|
return item;
|
|
});
|
|
}
|
|
|
|
async function update(id, transform) {
|
|
return transact(async records => {
|
|
const item = (await records.getAll()).find(candidate => candidate.id === id);
|
|
if (item) await records.put(transform(item));
|
|
});
|
|
}
|
|
|
|
function preferenceId(ownerLogin) {
|
|
return 'receipt-preference:' + String(ownerLogin || '').trim();
|
|
}
|
|
|
|
async function setReceiptPreference(ownerLogin, enabled) {
|
|
const login = String(ownerLogin || '').trim();
|
|
if (!login) return;
|
|
return transact(async records => {
|
|
const id = preferenceId(login);
|
|
if (!enabled) return records.delete(id);
|
|
return records.put({ id, recordType: 'receipt-preference', ownerLogin: login, enabled: true });
|
|
});
|
|
}
|
|
|
|
async function getReceiptPreference(ownerLogin) {
|
|
const id = preferenceId(ownerLogin);
|
|
return transact(async records => Boolean(
|
|
(await records.getAll()).find(item => item.id === id)?.enabled
|
|
));
|
|
}
|
|
|
|
return {
|
|
reconcile,
|
|
upsert,
|
|
claim,
|
|
claimNext,
|
|
claimBatch,
|
|
complete: id => update(id, item => ({ ...item, status: 'sent', claimUntil: 0 })),
|
|
release: id => update(id, item => ({ ...item, status: 'queued', claimUntil: 0 })),
|
|
fail: (id, error, deliveryState) => update(id, item => ({
|
|
...item, status: 'attention', claimUntil: 0, error,
|
|
...(deliveryState ? { deliveryState } : {}),
|
|
})),
|
|
snapshot: () => transact(async records =>
|
|
(await records.getAll()).filter(item => item.recordType !== 'receipt-preference')),
|
|
countBlocked: ownerLogin => transact(async records =>
|
|
(await records.getAll()).filter(item => item.recordType !== 'receipt-preference' &&
|
|
item.status !== 'sent' && item.ownerLogin !== ownerLogin).length),
|
|
setReceiptPreference,
|
|
getReceiptPreference,
|
|
close: () => transact.close?.(),
|
|
};
|
|
}
|
|
|
|
function createBackgroundIssueSync({
|
|
store, fetchJson, base = '', maxConcurrency = 3, batchSize = 70,
|
|
batch = work => work(), requestTimeoutMs = 15000,
|
|
}) {
|
|
let purgeRequested = false;
|
|
let activeFlush = null;
|
|
const activeRequests = new Set();
|
|
const timeoutMs = Math.max(1, Number(requestTimeoutMs) || 15000);
|
|
|
|
async function requestJson(url, options = {}) {
|
|
const controller = new AbortController();
|
|
activeRequests.add(controller);
|
|
let timer;
|
|
const interrupted = new Promise((resolve, reject) => {
|
|
controller.signal.addEventListener('abort', () => {
|
|
reject(new Error(purgeRequested
|
|
? 'Background request canceled.'
|
|
: 'Background request timed out.'));
|
|
}, { once: true });
|
|
timer = setTimeout(() => controller.abort(), timeoutMs);
|
|
});
|
|
try {
|
|
return await Promise.race([
|
|
Promise.resolve().then(() => fetchJson(url, { ...options, signal: controller.signal })),
|
|
interrupted,
|
|
]);
|
|
} finally {
|
|
clearTimeout(timer);
|
|
activeRequests.delete(controller);
|
|
}
|
|
}
|
|
function receiptFor(item, status, delivered = {}) {
|
|
if (status === 'attention') {
|
|
return { id: item.id, status, kind: item.kind ? 'message' : 'issue', route: '#/my-work/drafts' };
|
|
}
|
|
if (item.kind === 'update-reply') {
|
|
return { id: item.id, status, kind: 'message', route: '#/my-work/update/' + encodeURIComponent(item.notificationId) };
|
|
}
|
|
const repository = String(item.repository || '').split('/').map(encodeURIComponent).join('/');
|
|
if (item.kind === 'issue-comment' || item.kind === 'pull-comment') {
|
|
const resource = item.kind === 'pull-comment' ? 'pull' : 'issue';
|
|
return { id: item.id, status, kind: 'message', route: '#/my-work/' + resource + '/' + repository + '/' + encodeURIComponent(item.number) };
|
|
}
|
|
return { id: item.id, status, kind: 'issue', route: '#/my-work/issue/' + repository + '/' + encodeURIComponent(delivered.number) };
|
|
}
|
|
|
|
function deliveryRequest(item) {
|
|
if (item.kind === 'update-reply') {
|
|
return authoredRequest('api/v1/notifications/' + encodeURIComponent(item.notificationId) + '/reply', item);
|
|
}
|
|
const repository = String(item.repository || '').split('/').map(encodeURIComponent).join('/');
|
|
if (item.kind === 'issue-comment' || item.kind === 'pull-comment') {
|
|
const resource = item.kind === 'pull-comment' ? 'pulls' : 'issues';
|
|
return authoredRequest(
|
|
'api/v1/repos/' + repository + '/' + resource + '/' + encodeURIComponent(item.number) + '/comments',
|
|
item,
|
|
);
|
|
}
|
|
return {
|
|
url: base + 'api/v1/repos/' + repository + '/issues',
|
|
options: {
|
|
method: 'POST',
|
|
headers: {
|
|
Accept: 'application/json',
|
|
'Content-Type': 'application/json',
|
|
'Idempotency-Key': item.operationId,
|
|
},
|
|
body: JSON.stringify({
|
|
title: item.title,
|
|
body: item.body,
|
|
label_ids: item.labelIds,
|
|
...(item.milestoneId ? { milestone_id: item.milestoneId } : {}),
|
|
...(item.dueDate ? { due_date: item.dueDate + 'T23:59:59Z' } : {}),
|
|
}),
|
|
},
|
|
};
|
|
}
|
|
|
|
function authoredRequest(url, item) {
|
|
return {
|
|
url: base + url,
|
|
options: {
|
|
method: 'POST',
|
|
headers: {
|
|
Accept: 'application/json',
|
|
'Content-Type': 'application/json',
|
|
'Idempotency-Key': item.operationId,
|
|
},
|
|
body: JSON.stringify({ body: item.body }),
|
|
},
|
|
};
|
|
}
|
|
|
|
async function deliver(item) {
|
|
const request = deliveryRequest(item);
|
|
try {
|
|
const delivered = await requestJson(request.url, request.options);
|
|
await store.complete(item.id);
|
|
const receipt = receiptFor(item, 'confirmed', delivered);
|
|
return item.kind ? { message: delivered, receipt } : { issue: delivered, receipt };
|
|
} catch (error) {
|
|
const status = Number(error?.status || 0);
|
|
if (status === 401) {
|
|
await store.release(item.id);
|
|
throw error;
|
|
}
|
|
if (status >= 400 && status < 500) {
|
|
await store.fail(
|
|
item.id,
|
|
String(error?.message || 'Issue needs attention').slice(0, 240),
|
|
error?.code === 'delivery_uncertain' ? 'uncertain' : undefined,
|
|
);
|
|
return { attention: true, error, receipt: receiptFor(item, 'attention') };
|
|
}
|
|
await store.release(item.id);
|
|
throw error;
|
|
}
|
|
}
|
|
|
|
async function send(item, currentLogin) {
|
|
if (purgeRequested) return { blocked: true };
|
|
if (!currentLogin || item.ownerLogin !== currentLogin) return { blocked: true };
|
|
await store.upsert(item);
|
|
const claimed = await store.claim(item.id, currentLogin);
|
|
if (!claimed) return { busy: true };
|
|
return deliver(claimed);
|
|
}
|
|
|
|
async function runFlush() {
|
|
const identity = await requestJson(base + 'api/v1/background-identity', {
|
|
headers: { Accept: 'application/json' }, cache: 'no-store',
|
|
});
|
|
const login = String(identity?.login || '').trim();
|
|
const confirmed = [];
|
|
const receipts = [];
|
|
let attention = 0;
|
|
if (!login) return { confirmed, blocked: 0, attention, login, receipts };
|
|
const collect = result => {
|
|
if (result.issue) confirmed.push(result.issue);
|
|
if (result.message) confirmed.push(result.message);
|
|
if (result.attention) attention += 1;
|
|
if (result.receipt) receipts.push(result.receipt);
|
|
};
|
|
if (store.claimBatch) {
|
|
const claimed = await store.claimBatch(login, batchSize);
|
|
let next = 0;
|
|
let authenticationError = null;
|
|
let transientError = null;
|
|
const worker = async () => {
|
|
while (!purgeRequested && !authenticationError && next < claimed.length) {
|
|
const item = claimed[next++];
|
|
try {
|
|
collect(await deliver(item));
|
|
} catch (error) {
|
|
if (Number(error?.status || 0) === 401) authenticationError = error;
|
|
else if (!transientError) transientError = error;
|
|
}
|
|
}
|
|
};
|
|
const concurrency = Math.max(1, Math.min(Number(maxConcurrency) || 1, claimed.length));
|
|
await Promise.all(Array.from({ length: concurrency }, worker));
|
|
if (purgeRequested) {
|
|
await Promise.all(claimed.slice(next).map(item => store.release(item.id)));
|
|
throw new Error('Background delivery canceled.');
|
|
}
|
|
if (authenticationError) {
|
|
await Promise.all(claimed.slice(next).map(item => store.release(item.id)));
|
|
throw authenticationError;
|
|
}
|
|
if (transientError) throw transientError;
|
|
} else {
|
|
while (!purgeRequested) {
|
|
const item = await store.claimNext(login);
|
|
if (!item) break;
|
|
collect(await deliver(item));
|
|
}
|
|
}
|
|
const blocked = store.countBlocked ? await store.countBlocked(login) : 0;
|
|
return { confirmed, blocked, attention, login, receipts };
|
|
}
|
|
|
|
function flush() {
|
|
if (purgeRequested) return Promise.resolve({ confirmed: [], blocked: 0, attention: 0, login: '', receipts: [] });
|
|
if (activeFlush) return activeFlush;
|
|
activeFlush = Promise.resolve().then(() => batch(runFlush)).finally(() => { activeFlush = null; });
|
|
return activeFlush;
|
|
}
|
|
|
|
async function purge() {
|
|
purgeRequested = true;
|
|
activeRequests.forEach(controller => controller.abort());
|
|
if (activeFlush) await activeFlush.catch(() => {});
|
|
await store.close?.();
|
|
}
|
|
|
|
return {
|
|
flush, send, purge,
|
|
reconcile: (items, outboxLane) => store.reconcile(items, outboxLane),
|
|
snapshot: () => store.snapshot(),
|
|
setReceiptPreference: (ownerLogin, enabled) => store.setReceiptPreference(ownerLogin, enabled),
|
|
getReceiptPreference: ownerLogin => store.getReceiptPreference(ownerLogin),
|
|
};
|
|
}
|
|
|
|
createBackgroundIssueSync.createIssueSyncStore = createIssueSyncStore;
|
|
if (typeof module !== 'undefined' && module.exports) module.exports = createBackgroundIssueSync;
|
|
if (typeof globalThis !== 'undefined') {
|
|
globalThis.createBackgroundIssueSync = createBackgroundIssueSync;
|
|
globalThis.createIssueSyncStore = createIssueSyncStore;
|
|
}
|