Merge pull request 'Queue multiple offline Week-to-Today pulls' (#1250) from timmy/1249-queue-offline-week-pulls into main

frontend/dashboard.js

@@ -391,7 +391,7 @@
   });
   const weekPlan = createWeekPlan({
     fetchJson:fetchReviewJson,localDate:todayRollover.localDate,timeZone:todayRollover.timeZone,
-    storage:localStorage,getLogin:() => planningOwnerLogin,
+    storage:localStorage,getLogin:() => planningOwnerLogin,coordinator:outboxCoordinator,
   });
   function openWeekPlanner(trigger) { planningTomorrow=false; return weekFlow.open(trigger); }
   function renderTomorrowQueueSummary(value) {

frontend/week-plan.js

@@ -1,4 +1,4 @@
-function createWeekPlan({fetchJson,localDate,timeZone,storage,getLogin,now=()=>Date.now()}={}) {
+function createWeekPlan({fetchJson,localDate,timeZone,storage,getLogin,coordinator,now=()=>Date.now()}={}) {
   let week={revision:0,timezone:null,days:[]};
   let flushing=null;
   let lastConflict=null;
@@ -39,20 +39,33 @@ function createWeekPlan({fetchJson,localDate,timeZone,storage,getLogin,now=()=>D
     const login=String(getLogin?.()||'').trim().toLowerCase();
     return login?pullPrefix+encodeURIComponent(login):'';
   }
-  function pendingPull() {
+  function pullQueue() {
     const key=pullKey();
-    if(!key||!storage)return null;
+    if(!key||!storage)return [];
     try {
       const value=JSON.parse(storage.getItem(key)||'null');
-      return typeof value?.body?.operation_id==='string'&&Array.isArray(value?.today?.ids)&&
+      const entries=Array.isArray(value?.entries)?value.entries:(value?[value]:[]);
-        Number.isInteger(value?.week?.revision)&&Array.isArray(value?.week?.days)?value:null;
+      return entries.filter(entry=>typeof entry?.body?.operation_id==='string'&&Array.isArray(entry?.today?.ids)&&
-    } catch(_error){return null;}
+        Number.isInteger(entry?.week?.revision)&&Array.isArray(entry?.week?.days));
+    } catch(_error){return [];}
+  }
+  function writePullQueue(entries) {
+    const key=pullKey();if(!key||!storage)return false;
+    try {
+      if(entries.length)storage.setItem(key,JSON.stringify({version:2,entries}));
+      else storage.removeItem(key);
+      return true;
+    } catch(_error){return false;}
+  }
+  function pendingPull() {
+    return pullQueue()[0]||null;
   }
   function resumePull() {
-    const value=pendingPull();
+    const queue=pullQueue(),conflicted=queue.find(entry=>entry.conflict&&entry.current?.week);
+    const value=conflicted||queue[queue.length-1];
     if(!value)return null;
-    adopt(value.week);
+    adopt(conflicted?conflicted.current.week:value.week);
-    return {...value,sync_pending:true};
+    return {...value,...(conflicted?.current||{}),sync_pending:true};
   }
   function readConfirmed() {
     const key=confirmedKey();
@@ -502,8 +515,8 @@ function createWeekPlan({fetchJson,localDate,timeZone,storage,getLogin,now=()=>D
         return {...cloneDay(day),ids:(day.ids||[]).filter(id=>id!==identity),estimates};
       })};
       const record={body,today:optimisticToday,week:optimisticWeek,queued_at:now()};
-      try {storage.setItem(key,JSON.stringify(record));}
+      const queue=pullQueue();
-      catch(_error){throw new Error('Could not save this pull on this device. Nothing changed.');}
+      if(!writePullQueue([...queue,record]))throw new Error('Could not save this pull on this device. Nothing changed.');
       adopt(optimisticWeek);
       try{return await flushPull();}
       catch(error){
@@ -516,25 +529,44 @@ function createWeekPlan({fetchJson,localDate,timeZone,storage,getLogin,now=()=>D
   }
   function flushPull() {
     if(pulling)return pulling;
-    const record=pendingPull(),key=pullKey();
+    if(!pendingPull()||!pullKey())return Promise.resolve(false);
-    if(!record||!key)return Promise.resolve(false);
+    const run=async()=>{
-    pulling=(async()=>{
+      let result,lastResult=false;
-      let result;
+      while(pendingPull()){
-      try {
+        const current=pendingPull();
-        result=await fetchJson('api/v1/week/pull-item',{method:'POST',headers:{'Content-Type':'application/json'},body:JSON.stringify(record.body)});
+        try {
-      } catch(error) {
+          result=await fetchJson('api/v1/week/pull-item',{method:'POST',headers:{'Content-Type':'application/json'},body:JSON.stringify(current.body)});
-        if(error?.status!==409)throw error;
+        } catch(error) {
-        const [today,currentWeek]=await Promise.all([fetchJson('api/v1/today'),fetchJson('api/v1/week')]);
+          if(error?.status!==409)throw error;
-        const conflicted={...record,conflict:true,current:{today,week:currentWeek}};
+          const [today,currentWeek]=await Promise.all([fetchJson('api/v1/today'),fetchJson('api/v1/week')]);
-        storage.setItem(key,JSON.stringify(conflicted));
+          const queue=pullQueue(),conflicted={...current,conflict:true,current:{today,week:currentWeek}};
-        adopt(currentWeek);
+          if(!writePullQueue([conflicted,...queue.slice(1)]))throw new Error('Could not preserve queued Week Ahead work on this device.');
-        return {today,week:currentWeek,conflict:true,sync_pending:true};
+          adopt(currentWeek);
+          return {today,week:currentWeek,conflict:true,sync_pending:true};
+        }
+        const queue=pullQueue();
+        if(queue[0]?.body?.operation_id!==current.body.operation_id)continue;
+        const remaining=queue.slice(1).map(entry=>{
+          const identity=entry.body.identity;
+          const source=entry.week.days.find(day=>(day.ids||[]).includes(identity));
+          const estimate=Number(entry.today?.estimates?.[identity]??source?.estimates?.[identity]);
+          const today={...result.today,ids:(result.today.ids||[]).includes(identity)?[...result.today.ids]:[...result.today.ids,identity],
+            estimates:{...(result.today.estimates||{}),...(Number.isFinite(estimate)?{[identity]:estimate}:{})}};
+          const nextWeek={revision:result.week.revision,timezone:result.week.timezone||null,days:result.week.days.map(day=>{
+            const estimates={...(day.estimates||{})};delete estimates[identity];
+            return {...cloneDay(day),ids:(day.ids||[]).filter(id=>id!==identity),estimates};
+          })};
+          result={today,week:nextWeek};
+          return {...entry,body:{...entry.body,today_revision:today.revision,week_revision:nextWeek.revision},today,week:nextWeek};
+        });
+        if(!writePullQueue(remaining))throw new Error('Could not update queued Week Ahead work on this device.');
+        lastResult=result;
       }
-      const current=pendingPull();
+      adoptConfirmed(lastResult.week,{...confirmedItems,...pendingItems});
-      if(current?.body?.operation_id===record.body.operation_id)storage.removeItem(key);
+      return {...lastResult,sync_pending:false};
-      adoptConfirmed(result.week,{...confirmedItems,...pendingItems});
+    };
-      return {...result,sync_pending:Boolean(pendingPull())};
+    const delivery=coordinator?coordinator.runExclusive('week',run):run();
-    })().finally(()=>{pulling=null;});
+    pulling=Promise.resolve(delivery).finally(()=>{pulling=null;});
     return pulling;
   }
   async function startEarly(planDate,todayRevision) {

tests/test_week_plan_frontend.py

@@ -2,14 +2,18 @@ import json
 import subprocess
 from pathlib import Path
 
+import pytest
+
 
 FRONTEND = Path(__file__).parents[1] / "frontend"
 CONTROLLER = FRONTEND / "week-plan.js"
+COORDINATOR = FRONTEND / "outbox-coordinator.js"
 
 
 def run_controller(scenario: str) -> dict:
     harness = f"""
 const createWeekPlan = require({json.dumps(str(CONTROLLER))});
+const createOutboxCoordinator = require({json.dumps(str(COORDINATOR))});
 (async()=>{{ {scenario} }})().catch(error=>{{console.error(error);process.exit(1);}});
 """
     completed = subprocess.run(["node", "-e", harness], check=True, capture_output=True, text=True)
@@ -358,6 +362,128 @@ console.log(JSON.stringify({pending,resume,confirmed,requests,keys:[...values.ke
     assert result["state"]["revision"] == 8
 
 
+def test_week_controller_queues_two_offline_pulls_and_replays_them_fifo_after_reload():
+    result = run_controller("""
+const values=new Map(),delivered=[];
+const storage={getItem:key=>values.get(key)||null,setItem:(key,value)=>values.set(key,value),removeItem:key=>values.delete(key)};
+let offline=true;
+let serverToday={revision:4,ids:['active'],capacity_minutes:180,estimates:{active:30}};
+let serverWeek={revision:7,timezone:'UTC',days:[{plan_date:'2026-08-21',ids:['first','second','later'],capacity_minutes:150,estimates:{first:35,second:40,later:45}}]};
+const fetchJson=async(url,options={})=>{
+ if(offline)throw new Error('connection lost');
+ const body=JSON.parse(options.body);delivered.push(body);
+ if(body.today_revision!==serverToday.revision||body.week_revision!==serverWeek.revision){const error=new Error('changed');error.status=409;throw error;}
+ serverToday={...serverToday,revision:serverToday.revision+1,ids:[...serverToday.ids,body.identity],
+   estimates:{...serverToday.estimates,[body.identity]:serverWeek.days[0].estimates[body.identity]}};
+ serverWeek={...serverWeek,revision:serverWeek.revision+1,days:serverWeek.days.map(day=>({...day,
+   ids:day.ids.filter(id=>id!==body.identity),estimates:Object.fromEntries(Object.entries(day.estimates).filter(([id])=>id!==body.identity))}))};
+ return {today:serverToday,week:serverWeek};
+};
+const options={storage,getLogin:()=> 'Timmy',fetchJson,localDate:()=> '2026-08-20',timeZone:()=> 'UTC'};
+const first=createWeekPlan(options);first.adopt(serverWeek);
+const queuedFirst=await first.pullItem('first',serverToday,'pull-first');
+const queuedSecond=await first.pullItem('second',queuedFirst.today,'pull-second');
+const restored=createWeekPlan(options),resumed=restored.resumePull();
+offline=false;
+const confirmed=await restored.flushPull();
+console.log(JSON.stringify({queuedFirst,queuedSecond,resumed,confirmed,delivered,keys:[...values.keys()],state:restored.state()}));
+""")
+
+    assert result["queuedFirst"]["today"]["ids"] == ["active", "first"]
+    assert result["queuedSecond"]["today"]["ids"] == ["active", "first", "second"]
+    assert result["resumed"]["today"]["ids"] == ["active", "first", "second"]
+    assert [request["operation_id"] for request in result["delivered"]] == ["pull-first", "pull-second"]
+    assert result["delivered"][1]["today_revision"] == 5
+    assert result["delivered"][1]["week_revision"] == 8
+    assert result["confirmed"]["today"]["ids"] == ["active", "first", "second"]
+    assert result["confirmed"]["week"]["days"][0]["ids"] == ["later"]
+    assert result["confirmed"]["sync_pending"] is False
+    assert result["keys"] == ["stackchain.week-confirmed.v1.timmy"]
+    assert result["state"]["revision"] == 9
+
+
+@pytest.mark.parametrize(
+    "coordinator_setup",
+    [
+        """const locks={request:async(name,_options,work)=>{
+ if(held.has(name))return work(null);
+ held.add(name);try{return await work({name});}finally{held.delete(name);}
+}};
+const coordinatorA=createOutboxCoordinator({storage,locks,channelFactory:null,tabId:'a'});
+const coordinatorB=createOutboxCoordinator({storage,locks,channelFactory:null,tabId:'b'});""",
+        """const coordinatorA=createOutboxCoordinator({storage,locks:null,channelFactory:null,tabId:'a',now:()=>100});
+const coordinatorB=createOutboxCoordinator({storage,locks:null,channelFactory:null,tabId:'b',now:()=>100});""",
+    ],
+    ids=["navigator-lock", "storage-lease"],
+)
+def test_week_controller_allows_only_one_tab_to_replay_the_shared_pull_queue(coordinator_setup):
+    result = run_controller("""
+const values=new Map(),held=new Set();
+const storage={getItem:key=>values.get(key)||null,setItem:(key,value)=>values.set(key,value),removeItem:key=>values.delete(key)};
+COORDINATOR_SETUP
+let offline=true,posts=0,release;
+const gate=new Promise(resolve=>release=resolve);
+const fetchJson=async(_url,options={})=>{
+ if(offline)throw new Error('connection lost');
+ posts+=1;await gate;
+ const body=JSON.parse(options.body);
+ return {today:{revision:5,ids:['active',body.identity],capacity_minutes:180,estimates:{active:30,[body.identity]:35}},
+   week:{revision:8,timezone:'UTC',days:[{plan_date:'2026-08-21',ids:[],capacity_minutes:150,estimates:{}}]}};
+};
+const options={storage,getLogin:()=> 'timmy',fetchJson,localDate:()=> '2026-08-20',timeZone:()=> 'UTC'};
+const source=createWeekPlan({...options,coordinator:coordinatorA});source.adopt({revision:7,timezone:'UTC',days:[{plan_date:'2026-08-21',ids:['first'],capacity_minutes:150,estimates:{first:35}}]});
+await source.pullItem('first',{revision:4,ids:['active'],capacity_minutes:180,estimates:{active:30}},'pull-first');
+offline=false;
+const first=createWeekPlan({...options,coordinator:coordinatorA}),second=createWeekPlan({...options,coordinator:coordinatorB});
+const firstFlush=first.flushPull();
+await Promise.resolve();
+setImmediate(release);
+const skipped=await second.flushPull();
+release();
+const confirmed=await firstFlush;
+console.log(JSON.stringify({posts,skipped,confirmed,pending:first.pendingPull()}));
+""".replace("COORDINATOR_SETUP", coordinator_setup))
+
+    assert result["posts"] == 1
+    assert result["skipped"]["lease_skipped"] is True
+    assert result["confirmed"]["sync_pending"] is False
+    assert result["pending"] is None
+
+
+def test_week_controller_resumes_server_truth_when_a_fifo_head_conflicts():
+    result = run_controller("""
+const values=new Map();
+const storage={getItem:key=>values.get(key)||null,setItem:(key,value)=>values.set(key,value),removeItem:key=>values.delete(key)};
+let offline=true,posts=0;
+const remoteToday={revision:10,ids:['server-today'],capacity_minutes:180,estimates:{'server-today':30}};
+const remoteWeek={revision:12,timezone:'UTC',days:[{plan_date:'2026-08-21',ids:['second','third','server-week'],capacity_minutes:180,estimates:{second:40,third:45,'server-week':25}}]};
+const fetchJson=async(url,options={})=>{
+ if(offline)throw new Error('connection lost');
+ if(!options.method)return url.endsWith('/today')?remoteToday:remoteWeek;
+ posts+=1;
+ if(posts===1)return {today:{revision:5,ids:['active','first'],capacity_minutes:180,estimates:{active:30,first:35}},
+   week:{revision:8,timezone:'UTC',days:[{plan_date:'2026-08-21',ids:['second','third'],capacity_minutes:180,estimates:{second:40,third:45}}]}};
+ const error=new Error('changed');error.status=409;throw error;
+};
+const options={storage,getLogin:()=> 'timmy',fetchJson,localDate:()=> '2026-08-20',timeZone:()=> 'UTC'};
+const first=createWeekPlan(options);first.adopt({revision:7,timezone:'UTC',days:[{plan_date:'2026-08-21',ids:['first','second','third'],capacity_minutes:180,estimates:{first:35,second:40,third:45}}]});
+let today={revision:4,ids:['active'],capacity_minutes:180,estimates:{active:30}};
+today=(await first.pullItem('first',today,'pull-first')).today;
+today=(await first.pullItem('second',today,'pull-second')).today;
+await first.pullItem('third',today,'pull-third');
+offline=false;const conflicted=await first.flushPull();
+const restored=createWeekPlan(options),resumed=restored.resumePull();
+console.log(JSON.stringify({conflicted,resumed,state:restored.state(),pending:restored.pendingPull()}));
+""")
+
+    assert result["conflicted"]["conflict"] is True
+    assert result["resumed"]["conflict"] is True
+    assert result["resumed"]["today"]["ids"] == ["server-today"]
+    assert result["state"]["revision"] == 12
+    assert result["state"]["days"][0]["ids"] == ["second", "third", "server-week"]
+    assert result["pending"]["body"]["operation_id"] == "pull-second"
+
+
 def test_week_controller_preserves_a_conflicted_pull_intent_and_adopts_server_truth():
     result = run_controller("""
 const values=new Map(),requests=[];