Deliver queued authored messages after the app closes #253

Merged
rockachopa merged 1 commits from timmy/252-background-authored-message-sync into main 2026-08-08 02:12:33 +00:00
8 changed files with 235 additions and 31 deletions

View File

@ -112,10 +112,12 @@ data** deletes the snapshot, and opting out deletes it automatically.
API responses and mutations are never cached by the service worker. New issue captures, API responses and mutations are never cached by the service worker. New issue captures,
issue comments, pull-request comments, and unread-update replies use bounded local issue comments, pull-request comments, and unread-update replies use bounded local
outboxes when connectivity or a retryable server failure prevents delivery. Issue outboxes when connectivity or a retryable server failure prevents delivery. Issue
captures are also mirrored into IndexedDB and registered with Background Sync, so a captures and authored messages are also mirrored into account-bound IndexedDB lanes
supporting installed browser can deliver them after every dashboard client has closed. and registered with Background Sync, so a supporting installed browser can deliver
The worker verifies the current Gitea login, shares an atomic delivery claim with the new issues, issue comments, pull-request comments, and unread-update replies after
foreground path, and preserves the original idempotency key. Browsers without every dashboard client has closed. The worker verifies the current Gitea login, shares
an atomic delivery claim with the foreground path, and preserves the original
idempotency key. Browsers without
IndexedDB or Background Sync keep the foreground reconnect behavior. Drafts shows IndexedDB or Background Sync keep the foreground reconnect behavior. Drafts shows
queued and needs-attention messages with explicit send/discard controls; reopening the queued and needs-attention messages with explicit send/discard controls; reopening the
dashboard reconciles worker completions and permanent failures into the visible outbox. dashboard reconciles worker completions and permanent failures into the visible outbox.

View File

@ -1,4 +1,4 @@
function createAuthoredOutbox({ storage, fetchJson, coordinator, getOwnerLogin = () => '', createOperationId, now = () => Date.now(), maxItems = 50 }) { function createAuthoredOutbox({ storage, fetchJson, coordinator, backgroundSync, getOwnerLogin = () => '', createOperationId, now = () => Date.now(), maxItems = 50 }) {
const storageKey = 'stackchain.authored-outbox.v1'; const storageKey = 'stackchain.authored-outbox.v1';
const makeId = createOperationId || (() => const makeId = createOperationId || (() =>
globalThis.crypto?.randomUUID?.() || String(Date.now()) + '-' + Math.random().toString(16).slice(2) globalThis.crypto?.randomUUID?.() || String(Date.now()) + '-' + Math.random().toString(16).slice(2)
@ -17,6 +17,11 @@ function createAuthoredOutbox({ storage, fetchJson, coordinator, getOwnerLogin =
function write(items) { function write(items) {
storage?.setItem(storageKey, JSON.stringify({ version: 2, items })); storage?.setItem(storageKey, JSON.stringify({ version: 2, items }));
coordinator?.notify('authored'); coordinator?.notify('authored');
if (backgroundSync?.reconcile) {
Promise.resolve(backgroundSync.reconcile(items, 'authored'))
.then(() => items.length ? backgroundSync.requestSync?.() : undefined)
.catch(() => { /* Foreground reconnect remains the compatibility fallback. */ });
}
} }
function enqueue(message) { function enqueue(message) {
@ -84,15 +89,27 @@ function createAuthoredOutbox({ storage, fetchJson, coordinator, getOwnerLogin =
if (pending.has(item.id)) return pending.get(item.id); if (pending.has(item.id)) return pending.get(item.id);
const request = (async () => { const request = (async () => {
try { try {
const result = await fetchJson(endpoint(item), { let result;
method: 'POST', if (backgroundSync?.send) {
headers: { const delivery = await backgroundSync.send(item, currentLogin);
Accept: 'application/json', if (delivery.attention) {
'Content-Type': 'application/json', const error = delivery.error || new Error('Message needs attention');
'Idempotency-Key': item.operationId, error.status = Number(error.status || 422);
}, throw error;
body: JSON.stringify({ body: item.body }), }
}); result = delivery.message;
} else {
result = await fetchJson(endpoint(item), {
method: 'POST',
headers: {
Accept: 'application/json',
'Content-Type': 'application/json',
'Idempotency-Key': item.operationId,
},
body: JSON.stringify({ body: item.body }),
});
}
if (!result) return { blocked: true };
discard(item.id); discard(item.id);
return { result }; return { result };
} catch (error) { } catch (error) {
@ -149,7 +166,23 @@ function createAuthoredOutbox({ storage, fetchJson, coordinator, getOwnerLogin =
return retryItem(id, currentLogin); return retryItem(id, currentLogin);
} }
return { enqueue, update, discard, flush, retry, list: () => read().map(item => ({ ...item })) }; function reconcileBackground(records) {
const statuses = new Map((records || []).map(item => [item.id, item]));
const items = read().flatMap(item => {
const background = statuses.get(item.id);
if (background?.status === 'sent') return [];
if (background?.status === 'attention') return [{
...item,
status: 'attention',
error: String(background.error || 'Message needs attention').slice(0, 240),
}];
return [item];
});
write(items);
return items;
}
return { enqueue, update, discard, flush, retry, reconcileBackground, list: () => read().map(item => ({ ...item })) };
} }
if (typeof module !== 'undefined' && module.exports) module.exports = createAuthoredOutbox; if (typeof module !== 'undefined' && module.exports) module.exports = createAuthoredOutbox;

View File

@ -43,11 +43,13 @@ function createIndexedDbTransaction(indexedDB, dbName = 'stackchain-background-o
function createIssueSyncStore({ transaction, indexedDB = globalThis.indexedDB, now = () => Date.now(), claimMs = 30000 } = {}) { function createIssueSyncStore({ transaction, indexedDB = globalThis.indexedDB, now = () => Date.now(), claimMs = 30000 } = {}) {
const transact = transaction || createIndexedDbTransaction(indexedDB); const transact = transaction || createIndexedDbTransaction(indexedDB);
async function reconcile(items) { async function reconcile(items, outboxLane = 'issue') {
return transact(async records => { return transact(async records => {
const existing = await records.getAll(); const existing = await records.getAll();
const incoming = new Map(items.map(item => [item.id, { ...item }])); const incoming = new Map(items.map(item => [item.id, { ...item, outboxLane }]));
for (const current of existing) { for (const current of existing) {
const currentLane = current.outboxLane || 'issue';
if (currentLane !== outboxLane) continue;
const replacement = incoming.get(current.id); const replacement = incoming.get(current.id);
if (!replacement) { if (!replacement) {
if (current.status !== 'sending' || Number(current.claimUntil) <= Number(now())) { if (current.status !== 'sending' || Number(current.claimUntil) <= Number(now())) {
@ -125,8 +127,18 @@ function createIssueSyncStore({ transaction, indexedDB = globalThis.indexedDB, n
} }
function createBackgroundIssueSync({ store, fetchJson, base = '' }) { function createBackgroundIssueSync({ store, fetchJson, base = '' }) {
function issueRequest(item) { 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('/'); 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 { return {
url: base + 'api/v1/repos/' + repository + '/issues', url: base + 'api/v1/repos/' + repository + '/issues',
options: { options: {
@ -147,12 +159,27 @@ function createBackgroundIssueSync({ store, fetchJson, base = '' }) {
}; };
} }
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) { async function deliver(item) {
const request = issueRequest(item); const request = deliveryRequest(item);
try { try {
const issue = await fetchJson(request.url, request.options); const delivered = await fetchJson(request.url, request.options);
await store.complete(item.id); await store.complete(item.id);
return { issue }; return item.kind ? { message: delivered } : { issue: delivered };
} catch (error) { } catch (error) {
const status = Number(error?.status || 0); const status = Number(error?.status || 0);
if (status >= 400 && status < 500) { if (status >= 400 && status < 500) {
@ -185,6 +212,7 @@ function createBackgroundIssueSync({ store, fetchJson, base = '' }) {
if (!item) break; if (!item) break;
const result = await deliver(item); const result = await deliver(item);
if (result.issue) confirmed.push(result.issue); if (result.issue) confirmed.push(result.issue);
if (result.message) confirmed.push(result.message);
if (result.attention) attention += 1; if (result.attention) attention += 1;
} }
const blocked = store.countBlocked ? await store.countBlocked(login) : 0; const blocked = store.countBlocked ? await store.countBlocked(login) : 0;
@ -193,7 +221,7 @@ function createBackgroundIssueSync({ store, fetchJson, base = '' }) {
return { return {
flush, send, flush, send,
reconcile: items => store.reconcile(items), reconcile: (items, outboxLane) => store.reconcile(items, outboxLane),
snapshot: () => store.snapshot(), snapshot: () => store.snapshot(),
}; };
} }

View File

@ -913,14 +913,17 @@ textarea { resize: vertical; min-height: 120px; }
backgroundSync: backgroundIssueSync, backgroundSync: backgroundIssueSync,
getOwnerLogin: () => confirmedOwnerLogin, getOwnerLogin: () => confirmedOwnerLogin,
}); });
if (backgroundIssueSync) {
backgroundIssueSync.snapshot().then(records => issueOutbox.reconcileBackground(records))
.catch(() => { /* The foreground localStorage outbox remains available. */ });
}
const authoredOutbox = createAuthoredOutbox({ const authoredOutbox = createAuthoredOutbox({
storage: localStorage, fetchJson: fetchReviewJson, coordinator: outboxCoordinator, storage: localStorage, fetchJson: fetchReviewJson, coordinator: outboxCoordinator,
backgroundSync: backgroundIssueSync,
getOwnerLogin: () => confirmedOwnerLogin, getOwnerLogin: () => confirmedOwnerLogin,
}); });
if (backgroundIssueSync) {
backgroundIssueSync.snapshot().then(records => {
issueOutbox.reconcileBackground(records);
authoredOutbox.reconcileBackground(records);
}).catch(() => { /* The foreground localStorage outboxes remain available. */ });
}
const shareParams = new URLSearchParams(location.search); const shareParams = new URLSearchParams(location.search);
const sharedLaunch = { const sharedLaunch = {
title: shareParams.get('title') || '', title: shareParams.get('title') || '',

View File

@ -1,6 +1,6 @@
const BASE = new URL('./', self.location.href).pathname; const BASE = new URL('./', self.location.href).pathname;
importScripts(BASE + 'static/background-issue-sync.js'); importScripts(BASE + 'static/background-issue-sync.js');
const CACHE = 'stackchain-dashboard-shell-v12'; const CACHE = 'stackchain-dashboard-shell-v13';
const OUTAGE_STATUSES = new Set([500, 502, 503, 504]); const OUTAGE_STATUSES = new Set([500, 502, 503, 504]);
const SHELL = [ const SHELL = [
BASE, BASE,

View File

@ -152,6 +152,63 @@ process.stdout.write(JSON.stringify({{first,second,items:outbox.list()}}));
assert len(output["items"]) == 1 assert len(output["items"]) == 1
def test_authored_outbox_mirrors_to_background_sync_and_reconciles_worker_results():
script = f"""
const createAuthoredOutbox = require({json.dumps(str(OUTBOX))});
const values = new Map(); const mirrors=[]; let requested=0;
const storage = {{getItem:k=>values.get(k)||null,setItem:(k,v)=>values.set(k,v)}};
const backgroundSync = {{
reconcile: async (items, lane) => mirrors.push({{items:items.map(item=>({{...item}})),lane}}),
requestSync: async () => {{requested += 1;}},
}};
const outbox=createAuthoredOutbox({{storage,getOwnerLogin:()=>'timmy',backgroundSync}});
const sent=outbox.enqueue({{kind:'issue-comment',repository:'o/r',number:1,body:'Sent',operationId:'sent'}});
const failed=outbox.enqueue({{kind:'update-reply',notificationId:2,body:'Fix me',operationId:'failed'}});
setTimeout(() => {{
const reconciled=outbox.reconcileBackground([
{{...sent,status:'sent'}},
{{...failed,status:'attention',error:'Reply rejected'}},
]);
setTimeout(() => process.stdout.write(JSON.stringify({{mirrors,requested,reconciled}})), 0);
}}, 0);
"""
output = run_node(script)
assert output["mirrors"][0]["lane"] == "authored"
assert [item["operationId"] for item in output["mirrors"][-1]["items"]] == ["failed"]
assert output["requested"] >= 1
assert [(item["operationId"], item["status"]) for item in output["reconciled"]] == [
("failed", "attention")
]
def test_authored_outbox_foreground_send_uses_atomic_background_delivery():
script = f"""
const createAuthoredOutbox = require({json.dumps(str(OUTBOX))});
const values = new Map(); let backgroundCalls=0; let directCalls=0;
const storage = {{getItem:k=>values.get(k)||null,setItem:(k,v)=>values.set(k,v)}};
const backgroundSync = {{
reconcile: async()=>{{}}, requestSync:async()=>{{}},
send: async (item, owner) => {{backgroundCalls += 1; return {{message:{{id:3,owner}}}};}},
}};
const outbox=createAuthoredOutbox({{
storage,getOwnerLogin:()=>'timmy',backgroundSync,
fetchJson:async()=>{{directCalls += 1; return {{id:4}};}},
}});
outbox.enqueue({{kind:'pull-comment',repository:'o/r',number:2,body:'Review',operationId:'once'}});
(async()=>{{
const result=await outbox.flush('timmy');
process.stdout.write(JSON.stringify({{backgroundCalls,directCalls,result,remaining:outbox.list()}}));
}})();
"""
output = run_node(script)
assert output["backgroundCalls"] == 1
assert output["directCalls"] == 0
assert output["result"]["confirmed"] == [{"id": 3, "owner": "timmy"}]
assert output["remaining"] == []
@pytest.mark.anyio @pytest.mark.anyio
async def test_mobile_dashboard_loads_and_operates_authored_message_outbox(): async def test_mobile_dashboard_loads_and_operates_authored_message_outbox():
html = await dashboard() html = await dashboard()
@ -164,3 +221,5 @@ async def test_mobile_dashboard_loads_and_operates_authored_message_outbox():
assert "authoredOutbox.retry(item.outbox_id, activeFlushLogin)" in html assert "authoredOutbox.retry(item.outbox_id, activeFlushLogin)" in html
assert "authoredOutbox.discard(item.outbox_id)" in html assert "authoredOutbox.discard(item.outbox_id)" in html
assert "if (result?.queued)" in html assert "if (result?.queued)" in html
assert "backgroundSync: backgroundIssueSync" in html
assert "authoredOutbox.reconcileBackground(records)" in html

View File

@ -64,6 +64,83 @@ const fetchJson = async (url, options = {{}}) => {{
} }
@pytest.mark.parametrize(
("item", "expected_url"),
[
(
{"kind": "issue-comment", "repository": "stackchain/api", "number": 7},
"api/v1/repos/stackchain/api/issues/7/comments",
),
(
{"kind": "pull-comment", "repository": "stackchain/web", "number": 8},
"api/v1/repos/stackchain/web/pulls/8/comments",
),
(
{"kind": "update-reply", "notificationId": 9},
"api/v1/notifications/9/reply",
),
],
)
def test_closed_app_sync_delivers_each_authored_message_kind(item, expected_url):
authored = {
"id": "message-op",
"operationId": "message-op",
"ownerLogin": "timmy",
"status": "queued",
"body": "Ship this reply",
**item,
}
script = f"""
const createBackgroundIssueSync = require({json.dumps(str(SYNC))});
let queued = {json.dumps(authored)};
const calls = [];
const store = {{
claimNext: async owner => queued?.ownerLogin === owner ? (queued = null, {json.dumps(authored)}) : null,
complete: async () => {{}}, release: async () => {{}}, fail: async () => {{}},
countBlocked: async () => 0,
}};
const fetchJson = async (url, options = {{}}) => {{
calls.push({{url, options}});
if (url === 'api/v1/background-identity') return {{login:'timmy'}};
return {{id:42}};
}};
(async () => {{
const result = await createBackgroundIssueSync({{store,fetchJson}}).flush();
process.stdout.write(JSON.stringify({{calls,result}}));
}})();
"""
output = run_node(script)
mutation = output["calls"][1]
assert mutation["url"] == expected_url
assert mutation["options"]["headers"]["Idempotency-Key"] == "message-op"
assert json.loads(mutation["options"]["body"]) == {"body": "Ship this reply"}
assert output["result"]["confirmed"] == [{"id": 42}]
def test_reconciling_one_outbox_lane_preserves_the_other_lane():
script = f"""
const createBackgroundIssueSync = require({json.dumps(str(SYNC))});
const records=new Map();let tail=Promise.resolve();
const transaction=work=>{{const run=tail.then(()=>work({{
getAll:async()=>[...records.values()].map(value=>({{...value}})),
put:async value=>records.set(value.id,{{...value}}),delete:async id=>records.delete(id),
}}));tail=run.catch(()=>{{}});return run;}};
(async()=>{{
const store=createBackgroundIssueSync.createIssueSyncStore({{transaction}});
await store.reconcile([{{id:'issue',ownerLogin:'timmy',status:'queued'}}], 'issue');
await store.reconcile([{{id:'message',kind:'issue-comment',ownerLogin:'timmy',status:'queued'}}], 'authored');
await store.reconcile([], 'issue');
process.stdout.write(JSON.stringify(await store.snapshot()));
}})();
"""
output = run_node(script)
assert output == [
{"id": "message", "kind": "issue-comment", "ownerLogin": "timmy", "status": "queued", "outboxLane": "authored"}
]
def test_closed_app_sync_leaves_another_accounts_issue_queued(): def test_closed_app_sync_leaves_another_accounts_issue_queued():
script = f""" script = f"""
const createBackgroundIssueSync = require({json.dumps(str(SYNC))}); const createBackgroundIssueSync = require({json.dumps(str(SYNC))});
@ -255,7 +332,7 @@ const transaction=work=>{{const run=tail.then(()=>work({{
assert output["replay"] is None assert output["replay"] is None
assert output["snapshot"] == [ assert output["snapshot"] == [
{"id": "done", "ownerLogin": "timmy", "status": "sent", "claimUntil": 0} {"id": "done", "ownerLogin": "timmy", "status": "sent", "outboxLane": "issue", "claimUntil": 0}
] ]
@ -311,5 +388,7 @@ async def test_dashboard_wires_indexeddb_outbox_and_background_sync_fallback():
assert "createBackgroundIssueSync({" in html assert "createBackgroundIssueSync({" in html
assert "backgroundSync: backgroundIssueSync" in html assert "backgroundSync: backgroundIssueSync" in html
assert "registration.sync.register('stackchain-issue-outbox-v1')" in html assert "registration.sync.register('stackchain-issue-outbox-v1')" in html
assert "backgroundIssueSync.snapshot().then(records => issueOutbox.reconcileBackground(records))" in html assert "backgroundIssueSync.snapshot().then(records => {" in html
assert "issueOutbox.reconcileBackground(records);" in html
assert "authoredOutbox.reconcileBackground(records);" in html
assert "if ('indexedDB' in window)" in html assert "if ('indexedDB' in window)" in html

View File

@ -70,10 +70,10 @@ async function dispatchSync(tag) {{
return json.loads(completed.stdout) return json.loads(completed.stdout)
def test_background_issue_sync_ships_in_a_new_shell_cache(): def test_background_authored_sync_ships_in_a_new_shell_cache():
source = WORKER.read_text() source = WORKER.read_text()
assert "stackchain-dashboard-shell-v12" in source assert "stackchain-dashboard-shell-v13" in source
def test_background_sync_event_flushes_closed_app_issue_outbox_only_for_its_tag(): def test_background_sync_event_flushes_closed_app_issue_outbox_only_for_its_tag():