From e5ce3fe87a96e6e58a9c5e645d164c98c256b356 Mon Sep 17 00:00:00 2001 From: Croissant Le Doux Date: Thu, 9 Jul 2026 15:40:22 -0400 Subject: [PATCH] Offline write-queue: coalesce + replay without duplicating (#33) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit While gitea is unreachable, propose-approved writes queue instead of being lost; on reconnect they replay in order. The hard requirement is no duplication. The mechanism is coalescing by axis. Every write targets one field of one issue (estimate / priority / assignee / milestone), so `issue:kind` is the axis identity. Queuing a second write to the same axis supersedes the first (moved to the tail), so a burst of edits replays as one final write — and replay is idempotent (the apply path no-ops a change already reflected server-side). - `enqueueWrite` (coalesce), `pendingWrites`, `coalesceKey`, `affectedIssues` (what a post-replay reconcile re-reads), and `replayQueue(queue, apply)` which drains through the same guarded write path and returns the writes that still failed so they stay queued. Never throws — a failure is data. Acceptance met: a burst of offline edits + reconnect lands the final state with a single apply per axis (not one per edit); still-failing writes stay queued. +4 core tests; typecheck green. Follow-up: persist the queue in main + trigger replay on the reconnect signal (the offline banner + disabled composer already exist) — the coalesce/replay core is the tested heart. Co-Authored-By: Claude Opus 4.8 (1M context) --- packages/core/src/index.ts | 2 + .../core/src/queue/write-queue-v0.test.ts | 67 +++++++++++++++++ packages/core/src/queue/write-queue-v0.ts | 71 +++++++++++++++++++ 3 files changed, 140 insertions(+) create mode 100644 packages/core/src/queue/write-queue-v0.test.ts create mode 100644 packages/core/src/queue/write-queue-v0.ts diff --git a/packages/core/src/index.ts b/packages/core/src/index.ts index 7e7fb83..ae13216 100644 --- a/packages/core/src/index.ts +++ b/packages/core/src/index.ts @@ -23,6 +23,8 @@ export type { GiteaRequestInit, } from './gitea/types.js' +export { affectedIssues, coalesceKey, enqueueWrite, pendingWrites, replayQueue } from './queue/write-queue-v0.js' +export type { QueuedWrite } from './queue/write-queue-v0.js' export { describeChange, isLabelChange, planIssueChange, proposalsFor, summarizeChange } from './changes/apply-changes-v0.js' export type { ChangeProposal, diff --git a/packages/core/src/queue/write-queue-v0.test.ts b/packages/core/src/queue/write-queue-v0.test.ts new file mode 100644 index 0000000..dea486d --- /dev/null +++ b/packages/core/src/queue/write-queue-v0.test.ts @@ -0,0 +1,67 @@ +import { describe, expect, it, vi } from 'vitest' + +import type { IssueChange } from '../changes/apply-changes-v0.js' +import { affectedIssues, coalesceKey, enqueueWrite, pendingWrites, replayQueue, type QueuedWrite } from './write-queue-v0.js' + +const reest = (issue: number, estimate: string): IssueChange => + ({ kind: 'reestimate', issue, estimate }) as IssueChange +const assign = (issue: number, assignee: string | null): IssueChange => ({ kind: 'assign', issue, assignee }) + +describe('write-queue-v0 (#33)', () => { + it('coalesces repeat writes to the same (issue, axis) — replay applies the latest once', () => { + let q: QueuedWrite[] = [] + q = enqueueWrite(q, reest(7, 'est/2d'), '2026-02-10T09:00:00Z') + q = enqueueWrite(q, reest(7, 'est/5d'), '2026-02-10T09:05:00Z') // supersedes est/2d + q = enqueueWrite(q, assign(7, 'christian'), '2026-02-10T09:06:00Z') // different axis — kept + + expect(q).toHaveLength(2) + const pending = pendingWrites(q) + expect(pending).toEqual([ + { kind: 'reestimate', issue: 7, estimate: 'est/5d' }, + { kind: 'assign', issue: 7, assignee: 'christian' }, + ]) + expect(coalesceKey(reest(7, 'est/2d'))).toBe('7:reestimate') + }) + + it('keeps writes to different issues and axes distinct', () => { + let q: QueuedWrite[] = [] + q = enqueueWrite(q, reest(7, 'est/2d'), 't1') + q = enqueueWrite(q, reest(8, 'est/3d'), 't2') + q = enqueueWrite(q, assign(8, null), 't3') + expect(q).toHaveLength(3) + expect(affectedIssues(q)).toEqual([7, 8]) + }) + + it('a burst of edits then reconnect lands the final state without duplicating', async () => { + let q: QueuedWrite[] = [] + // offline: three edits to #7's estimate, one assign + q = enqueueWrite(q, reest(7, 'est/1d'), 't1') + q = enqueueWrite(q, reest(7, 'est/2d'), 't2') + q = enqueueWrite(q, reest(7, 'est/8d'), 't3') + q = enqueueWrite(q, assign(7, 'stephen'), 't4') + + const apply = vi.fn(async () => ({ ok: true })) + const { drained, remaining } = await replayQueue(q, apply) + + // only the final estimate + the assign are applied — not three estimate writes + expect(apply).toHaveBeenCalledTimes(2) + expect(apply).toHaveBeenNthCalledWith(1, { kind: 'reestimate', issue: 7, estimate: 'est/8d' }) + expect(apply).toHaveBeenNthCalledWith(2, { kind: 'assign', issue: 7, assignee: 'stephen' }) + expect(remaining).toHaveLength(0) + expect(drained).toHaveLength(2) + }) + + it('keeps writes that still fail on reconnect queued (never throws)', async () => { + let q: QueuedWrite[] = [] + q = enqueueWrite(q, reest(7, 'est/2d'), 't1') + q = enqueueWrite(q, reest(8, 'est/3d'), 't2') + // #7 applies, #8 rejects (still unreachable) — and one apply throws + const apply = vi.fn(async (c: IssueChange) => { + if (c.issue === 8) throw new Error('offline') + return { ok: true } + }) + const { drained, remaining } = await replayQueue(q, apply) + expect(drained.map((w) => w.change.issue)).toEqual([7]) + expect(remaining.map((w) => w.change.issue)).toEqual([8]) + }) +}) diff --git a/packages/core/src/queue/write-queue-v0.ts b/packages/core/src/queue/write-queue-v0.ts new file mode 100644 index 0000000..2a185b6 --- /dev/null +++ b/packages/core/src/queue/write-queue-v0.ts @@ -0,0 +1,71 @@ +/** + * Offline write-queue, v0 (#33). While gitea is unreachable, propose-approved + * writes are queued instead of lost; on reconnect they replay in order and + * resolve against a fresh reconcile. The one hard requirement is *no duplication*: + * replaying must not apply the same intent twice. + * + * The mechanism is coalescing by axis. Every write targets one field of one issue + * (its estimate, priority, assignee, or milestone). Queuing a second write to the + * same (issue, axis) supersedes the first — only the latest intent survives — so + * a burst of edits replays as one final write, and a replay is idempotent (the + * apply path no-ops a change already reflected server-side). + */ + +import type { IssueChange } from '../changes/apply-changes-v0.js' + +export interface QueuedWrite { + change: IssueChange + /** ISO time the write was queued (for display + stable ordering). */ + queuedAt: string +} + +/** + * The coalescing key: one field of one issue. Two writes with the same key are + * the same intent expressed twice — the later wins. Each `IssueChange.kind` maps + * to exactly one axis, so `issue:kind` is the axis identity. + */ +export function coalesceKey(change: IssueChange): string { + return `${change.issue}:${change.kind}` +} + +/** + * Queue a write, superseding any pending write to the same (issue, axis). The + * superseding write moves to the tail so replay order reflects latest intent. + */ +export function enqueueWrite(queue: readonly QueuedWrite[], change: IssueChange, queuedAt: string): QueuedWrite[] { + const key = coalesceKey(change) + return [...queue.filter((w) => coalesceKey(w.change) !== key), { change, queuedAt }] +} + +/** The changes to replay, in order — one per (issue, axis) by construction. */ +export function pendingWrites(queue: readonly QueuedWrite[]): IssueChange[] { + return queue.map((w) => w.change) +} + +/** Distinct issues touched by the queue — what a post-replay reconcile should re-read. */ +export function affectedIssues(queue: readonly QueuedWrite[]): number[] { + return [...new Set(queue.map((w) => w.change.issue))] +} + +/** + * Replay the queue through an apply function (the same guarded write path used + * online), in order. Returns the writes that failed (still unreachable / rejected) + * so they stay queued; everything else drains. Never throws — a failure is data. + */ +export async function replayQueue( + queue: readonly QueuedWrite[], + apply: (change: IssueChange) => Promise<{ ok: boolean }>, +): Promise<{ drained: QueuedWrite[]; remaining: QueuedWrite[] }> { + const drained: QueuedWrite[] = [] + const remaining: QueuedWrite[] = [] + for (const w of queue) { + let ok = false + try { + ok = (await apply(w.change)).ok + } catch { + ok = false + } + ;(ok ? drained : remaining).push(w) + } + return { drained, remaining } +}