forked from anomalyco/opencode
-
Notifications
You must be signed in to change notification settings - Fork 0
Expand file tree
/
Copy pathfence.ts
More file actions
60 lines (51 loc) · 1.87 KB
/
Copy pathfence.ts
File metadata and controls
60 lines (51 loc) · 1.87 KB
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
import { Database } from "@opencode-ai/core/database/database"
import { inArray } from "drizzle-orm"
import { EventSequenceTable } from "@opencode-ai/core/event/sql"
import { Workspace } from "@/control-plane/workspace"
import type { WorkspaceV2 } from "@opencode-ai/core/workspace"
import { Effect } from "effect"
export const HEADER = "x-opencode-sync"
export type State = Record<string, number>
export function load(db: Database.Interface["db"], ids?: string[]) {
return Effect.gen(function* () {
const rows = yield* (
ids?.length
? db.select().from(EventSequenceTable).where(inArray(EventSequenceTable.aggregate_id, ids)).all()
: db.select().from(EventSequenceTable).all()
).pipe(Effect.orDie)
return Object.fromEntries(rows.map((row) => [row.aggregate_id, row.seq]))
})
}
export function diff(prev: State, next: State) {
const ids = new Set([...Object.keys(prev), ...Object.keys(next)])
return Object.fromEntries(
[...ids]
.map((id) => [id, next[id] ?? -1] as const)
.filter(([id, seq]) => {
return (prev[id] ?? -1) !== seq
}),
)
}
export function parse(headers: Headers): State | undefined {
const raw = headers.get(HEADER)
if (!raw) return
let data
try {
data = JSON.parse(raw)
} catch {
return
}
if (!data || typeof data !== "object") return
return Object.fromEntries(
Object.entries(data).filter((entry): entry is [string, number] => {
return typeof entry[0] === "string" && Number.isInteger(entry[1])
}),
)
}
export function wait(workspaceID: WorkspaceV2.ID, state: State, signal?: AbortSignal) {
return Effect.gen(function* () {
yield* Effect.logInfo("waiting for state", { workspaceID, state })
yield* Workspace.Service.use((workspace) => workspace.waitForSync(workspaceID, state, signal))
yield* Effect.logInfo("state fully synced", { workspaceID, state })
})
}