Compare commits
3 Commits
feat/perf-
...
feat/offli
| Author | SHA1 | Date | |
|---|---|---|---|
|
|
ea8f1184e6 | ||
| c0dd9e70ef | |||
|
|
e5ce3fe87a |
@@ -276,6 +276,7 @@ describe('buildProjectView', () => {
|
||||
})
|
||||
|
||||
it('unbuilt views return a notImplemented marker, not fabricated data', () => {
|
||||
expect(buildProjectView('standup', undefined, snap, asOf)).toEqual({ notImplemented: 'standup' })
|
||||
expect(buildProjectView('milestone', undefined, snap, asOf)).toEqual({ notImplemented: 'milestone' })
|
||||
expect(buildProjectView('runway', undefined, snap, asOf)).toEqual({ notImplemented: 'runway' })
|
||||
})
|
||||
})
|
||||
|
||||
@@ -25,6 +25,8 @@ export type {
|
||||
|
||||
export { CACHE_SCHEMA, initCache, readIssue, upsertIssue } from './cache/cache-v0.js'
|
||||
export type { CacheDriver } from './cache/cache-v0.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,
|
||||
|
||||
67
packages/core/src/queue/write-queue-v0.test.ts
Normal file
67
packages/core/src/queue/write-queue-v0.test.ts
Normal file
@@ -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])
|
||||
})
|
||||
})
|
||||
71
packages/core/src/queue/write-queue-v0.ts
Normal file
71
packages/core/src/queue/write-queue-v0.ts
Normal file
@@ -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 }
|
||||
}
|
||||
Reference in New Issue
Block a user