Wire the offline write-queue into the desktop write path
The offline write-queue (packages/core/src/queue/write-queue-v0.ts, #33) was a tested pure module that nothing imported. Approving a change while gitea was unreachable made gitea:applyChange call the client directly and throw, losing the write. Now it's wired end to end: - queue-store.ts persists the queue next to the snapshot store (same degrade-to-empty-on-corruption discipline; injectable path for tests). - applyChange extracts the guarded write into applyChangeLive and, on unreachability (any error that is NOT a GiteaApiError rejection), enqueues the intent — coalesced by (issue, axis) — instead of throwing, returning { ok, queued, pending }. A genuine GiteaApiError still surfaces (a doomed write must not replay forever). - reconcile drains the queue once a successful read proves gitea is reachable, re-reading so the board reflects the replays; replays are idempotent (label plan.noop, assignee/milestone re-set). boot + stale reconcile report the pending count so the badge shows immediately offline. - Renderer: use-backlog threads `pending`; the OfflineBanner shows "N queued"; the chat approve message distinguishes a queued (offline) approval from an applied one. Also wires vitest into the desktop workspace (was missing, so the main-process suite couldn't run via `yarn test`) and fixes a stale Capture copy assertion left by the earlier posh-copy pass. Tests: queue-store.test.ts (persist/reload/coalesce/replay-drain, real core fns); 9 main-process + 169 core green; 12 demo e2e green. A one-off GITEA_LIVE smoke verified an online write lands+reverts and an offline approve queues+drains against the real repo (not committed, per the repo's no-mutating-test convention). Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
This commit is contained in:
Croissant Le Doux
@@ -15,8 +15,10 @@ import {
|
||||
applySchemaLabels,
|
||||
createGiteaClient,
|
||||
discoverRepos,
|
||||
enqueueWrite,
|
||||
ensurePmStateRepo,
|
||||
GiteaApiError,
|
||||
replayQueue,
|
||||
type DirectiveEntry,
|
||||
type GiteaClient,
|
||||
type GiteaConfig,
|
||||
@@ -46,6 +48,7 @@ import {
|
||||
const isDemo = (): boolean => process.env.COMMITEA_E2E === '1'
|
||||
|
||||
import { type AppConfig, clearConfig, loadConfig, publicConfig, saveConfig } from './config-store.js'
|
||||
import { loadQueue, saveQueue } from './queue-store.js'
|
||||
import { loadSnapshot, saveSnapshot } from './snapshot-store.js'
|
||||
|
||||
/** Walk up from cwd looking for a .env.local with a GITEA_TOKEN (dev convenience). */
|
||||
@@ -241,6 +244,59 @@ async function resolveLabelIds(client: GiteaClient, names: string[]): Promise<nu
|
||||
return names.map((n) => byName.get(n)).filter((id): id is number => id != null)
|
||||
}
|
||||
|
||||
/** The result of a live write — a label swap returns its plan, a field write the fresh issue. */
|
||||
type LiveApplyResult = { ok: true; plan?: ReturnType<typeof planIssueChange>; issue: Awaited<ReturnType<GiteaClient['getIssue']>> }
|
||||
|
||||
/**
|
||||
* Apply one approved change against gitea (the guarded write path, used both
|
||||
* online and when draining the offline queue). Label swaps (est/p) read → plan →
|
||||
* set; field writes (assign, milestone) write directly and return the fresh
|
||||
* issue. A replay is idempotent — a change already reflected server-side no-ops
|
||||
* (label plan.noop; assignee/milestone re-set to the same value). Throws on a
|
||||
* network failure (queued upstream) or a GiteaApiError (surfaced, never queued).
|
||||
*/
|
||||
async function applyChangeLive(client: GiteaClient, change: IssueChange): Promise<LiveApplyResult> {
|
||||
if (isLabelChange(change)) {
|
||||
const current = await client.getIssue(change.issue)
|
||||
const plan = planIssueChange(current.labels, change)
|
||||
if (plan.noop) return { ok: true, plan, issue: current }
|
||||
const ids = await resolveLabelIds(client, plan.labels)
|
||||
await client.setIssueLabels(change.issue, ids)
|
||||
const issue = await client.getIssue(change.issue)
|
||||
return { ok: true, plan, issue }
|
||||
}
|
||||
const issue =
|
||||
change.kind === 'assign'
|
||||
? await client.setIssueAssignees(change.issue, change.assignee ? [change.assignee] : [])
|
||||
: await client.setIssueMilestone(change.issue, change.milestone)
|
||||
return { ok: true, issue }
|
||||
}
|
||||
|
||||
/** Queue a write made while offline, coalescing by (issue, axis). Returns the new pending count. */
|
||||
function queueOffline(change: IssueChange): number {
|
||||
const next = enqueueWrite(loadQueue(), change, new Date().toISOString())
|
||||
saveQueue(next)
|
||||
return next.length
|
||||
}
|
||||
|
||||
/** How many writes are waiting to replay (shown as a queued badge). */
|
||||
function pendingCount(): number {
|
||||
return loadQueue().length
|
||||
}
|
||||
|
||||
/**
|
||||
* Replay the offline queue against gitea once we know it's reachable. Drained
|
||||
* writes are removed; any that still fail stay queued. Returns how many drained
|
||||
* (so the caller can invalidate + re-read the snapshot) and what remains.
|
||||
*/
|
||||
async function drainQueue(client: GiteaClient): Promise<{ drainedCount: number; remainingCount: number }> {
|
||||
const queue = loadQueue()
|
||||
if (queue.length === 0) return { drainedCount: 0, remainingCount: 0 }
|
||||
const { drained, remaining } = await replayQueue(queue, (change) => applyChangeLive(client, change))
|
||||
saveQueue(remaining)
|
||||
return { drainedCount: drained.length, remainingCount: remaining.length }
|
||||
}
|
||||
|
||||
export function registerGiteaIpc(): void {
|
||||
ipcMain.handle('gitea:status', () => {
|
||||
const cfg = resolveConfig()
|
||||
@@ -253,24 +309,33 @@ export function registerGiteaIpc(): void {
|
||||
|
||||
// Instant boot: the last persisted snapshot, shown before the fresh reconcile lands.
|
||||
ipcMain.handle('gitea:boot', () => {
|
||||
if (isDemo()) return { configured: true, cached: true, ...DEMO_SNAPSHOT, savedAt: DEMO_TODAY }
|
||||
if (isDemo()) return { configured: true, cached: true, pending: 0, ...DEMO_SNAPSHOT, savedAt: DEMO_TODAY }
|
||||
if (!getGiteaClient()) return { configured: false }
|
||||
const persisted = bootSnapshot()
|
||||
return persisted ? { configured: true, cached: true, ...persisted } : { configured: true, cached: false }
|
||||
return persisted
|
||||
? { configured: true, cached: true, pending: pendingCount(), ...persisted }
|
||||
: { configured: true, cached: false }
|
||||
})
|
||||
|
||||
ipcMain.handle('gitea:reconcile', async () => {
|
||||
if (isDemo()) return { configured: true, stale: false, ...DEMO_SNAPSHOT }
|
||||
if (isDemo()) return { configured: true, stale: false, pending: 0, ...DEMO_SNAPSHOT }
|
||||
const client = getGiteaClient()
|
||||
if (!client) return { configured: false, issues: [], milestones: [], deps: [], timelines: {} }
|
||||
try {
|
||||
// explicit UI sync — force fresh, and warm the cache for agent tool calls
|
||||
const snap = await getSnapshot(client, { maxAgeMs: 0 })
|
||||
return { configured: true, stale: false, ...snap }
|
||||
let snap = await getSnapshot(client, { maxAgeMs: 0 })
|
||||
// A successful reconcile proves gitea is reachable: drain any writes queued
|
||||
// while offline. If any drained, re-read so the board reflects the replays.
|
||||
const { drainedCount, remainingCount } = await drainQueue(client)
|
||||
if (drainedCount > 0) {
|
||||
invalidateSnapshot()
|
||||
snap = await getSnapshot(client, { maxAgeMs: 0 })
|
||||
}
|
||||
return { configured: true, stale: false, pending: remainingCount, ...snap }
|
||||
} catch (e) {
|
||||
// offline / gitea down — serve the last persisted snapshot rather than error out
|
||||
const persisted = bootSnapshot()
|
||||
if (persisted) return { configured: true, stale: true, ...persisted }
|
||||
if (persisted) return { configured: true, stale: true, pending: pendingCount(), ...persisted }
|
||||
throw e
|
||||
}
|
||||
})
|
||||
@@ -288,25 +353,18 @@ export function registerGiteaIpc(): void {
|
||||
ipcMain.handle('gitea:applyChange', async (_event, change: IssueChange) => {
|
||||
const client = getGiteaClient()
|
||||
if (!client) return { ok: false as const, reason: 'unconfigured' as const }
|
||||
|
||||
if (isLabelChange(change)) {
|
||||
const current = await client.getIssue(change.issue)
|
||||
const plan = planIssueChange(current.labels, change)
|
||||
if (plan.noop) return { ok: true as const, plan, issue: current }
|
||||
const ids = await resolveLabelIds(client, plan.labels)
|
||||
await client.setIssueLabels(change.issue, ids)
|
||||
const issue = await client.getIssue(change.issue)
|
||||
invalidateSnapshot()
|
||||
return { ok: true as const, plan, issue }
|
||||
try {
|
||||
const result = await applyChangeLive(client, change)
|
||||
invalidateSnapshot() // the board + forecast must reflect the write
|
||||
return result
|
||||
} catch (e) {
|
||||
// A GiteaApiError is a genuine rejection (bad label, gone issue) — surface
|
||||
// it; queueing would only replay a doomed write forever. Anything else is
|
||||
// unreachability: queue the intent so it replays on reconnect (#33).
|
||||
if (e instanceof GiteaApiError) throw e
|
||||
const pending = queueOffline(change)
|
||||
return { ok: true as const, queued: true as const, pending }
|
||||
}
|
||||
|
||||
// Field writes — the client returns the updated issue.
|
||||
const issue =
|
||||
change.kind === 'assign'
|
||||
? await client.setIssueAssignees(change.issue, change.assignee ? [change.assignee] : [])
|
||||
: await client.setIssueMilestone(change.issue, change.milestone)
|
||||
invalidateSnapshot()
|
||||
return { ok: true as const, issue }
|
||||
})
|
||||
|
||||
// capture_work filing: open each approved issue with its est/* + p/* labels.
|
||||
|
||||
Reference in New Issue
Block a user