Merge pull request 'Deliver queued authored messages after the app closes' (#253) from timmy/252-background-authored-message-sync into main

README.md

@@ -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,
 issue comments, pull-request comments, and unread-update replies use bounded local
 outboxes when connectivity or a retryable server failure prevents delivery. Issue
-captures are also mirrored into IndexedDB and registered with Background Sync, so a
-supporting installed browser can deliver them after 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
+captures and authored messages are also mirrored into account-bound IndexedDB lanes
+and registered with Background Sync, so a supporting installed browser can deliver
+new issues, issue comments, pull-request comments, and unread-update replies after
+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
 queued and needs-attention messages with explicit send/discard controls; reopening the
 dashboard reconciles worker completions and permanent failures into the visible outbox.

frontend/authored-outbox.js

@@ -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 makeId = createOperationId || (() =>
     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) {
     storage?.setItem(storageKey, JSON.stringify({ version: 2, items }));
     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) {
@@ -84,15 +89,27 @@ function createAuthoredOutbox({ storage, fetchJson, coordinator, getOwnerLogin =
     if (pending.has(item.id)) return pending.get(item.id);
     const request = (async () => {
       try {
-        const 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 }),
-        });
+        let result;
+        if (backgroundSync?.send) {
+          const delivery = await backgroundSync.send(item, currentLogin);
+          if (delivery.attention) {
+            const error = delivery.error || new Error('Message needs attention');
+            error.status = Number(error.status || 422);
+            throw error;
+          }
+          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);
         return { result };
       } catch (error) {
@@ -149,7 +166,23 @@ function createAuthoredOutbox({ storage, fetchJson, coordinator, getOwnerLogin =
     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;

frontend/background-issue-sync.js

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

frontend/index.html

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

frontend/service-worker.js

@@ -1,6 +1,6 @@
 const BASE = new URL('./', self.location.href).pathname;
 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 SHELL = [
   BASE,

tests/test_authored_outbox.py

@@ -152,6 +152,63 @@ process.stdout.write(JSON.stringify({{first,second,items:outbox.list()}}));
     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
 async def test_mobile_dashboard_loads_and_operates_authored_message_outbox():
     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.discard(item.outbox_id)" in html
     assert "if (result?.queued)" in html
+    assert "backgroundSync: backgroundIssueSync" in html
+    assert "authoredOutbox.reconcileBackground(records)" in html

tests/test_background_issue_sync.py

@@ -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():
     script = f"""
 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["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 "backgroundSync: backgroundIssueSync" 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

tests/test_service_worker.py

@@ -70,10 +70,10 @@ async function dispatchSync(tag) {{
     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()
 
-    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():