File size: 6,467 Bytes
09aec41 | 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 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 152 153 154 155 156 157 158 159 160 161 162 163 164 165 166 167 168 169 170 171 172 173 174 175 176 177 178 179 180 181 182 183 184 185 186 | /** Reconnect-safe Workspace baseline and increment producer. */
import type { Context } from '@deepseek-ai/cordis'
import { Deque } from '@deepseek-ai/dsh-deque'
import type { DomainChanged } from '@deepseek-ai/dsh-storage-domain'
import type { Workspace, WorkspaceRecord } from '@deepseek-ai/dsh-workspace'
import {
workspaceDomainState,
workspaceRecord,
WorkspaceId,
} from '@deepseek-ai/dsh-workspace'
import type {
WorkspaceBaseline,
WorkspaceFollowFrame,
WorkspaceView,
} from './types.ts'
/**
* Project one authoritative Workspace entity into its Remote value.
* @param workspace - authoritative registry entity.
* @returns detached Workspace projection for Remote consumers.
*/
export function workspaceView(workspace: Workspace): WorkspaceView {
return {
workspaceId: workspace.id,
path: workspace.path,
title: workspace.title,
sessionIds: [...workspace.sessionIds],
createdAt: workspace.createdAt,
updatedAt: workspace.updatedAt,
}
}
function changedWorkspaceView(workspaceId: string, value: unknown): WorkspaceView {
const record: WorkspaceRecord = workspaceRecord.parse(value)
return {
workspaceId: WorkspaceId(workspaceId),
path: record.path,
title: record.title,
sessionIds: [...record.sessionIds],
createdAt: record.createdAt,
updatedAt: record.updatedAt,
}
}
/** Owns Workspace domain observation and all active follow generations. */
export class WorkspaceFeed {
private readonly followers = new Set<WorkspaceFollower>()
private knownIds: Set<string>
private order: readonly string[]
private archived: readonly string[]
/** @param ctx - Host context containing the authoritative Workspace registry. */
constructor(private readonly ctx: Context) {
const baseline = ctx.workspaceRegistry.list()
this.knownIds = new Set(baseline.map(workspace => String(workspace.id)))
this.order = baseline.map(workspace => String(workspace.id))
this.archived = ctx.workspaceRegistry.archivedSessionIds.map(String)
ctx.on('domain/changed', (change: DomainChanged) => { this.changed(change) })
ctx.effect(() => () => {
for (const follower of this.followers) follower.close()
this.followers.clear()
}, 'workspace-controller.feed')
}
/**
* Read the complete current projection synchronously.
* @returns all active Workspaces and archived Session identities.
*/
baseline(): WorkspaceBaseline {
return {
items: this.ctx.workspaceRegistry.list().map(workspaceView),
archivedSessionIds: [...this.ctx.workspaceRegistry.archivedSessionIds],
}
}
/**
* Open one generation beginning with a complete baseline.
* @param signal - generation cancellation.
* @returns baseline followed by ordered Workspace increments.
*/
async *follow(signal: AbortSignal): AsyncIterable<WorkspaceFollowFrame> {
signal.throwIfAborted()
const follower = new WorkspaceFollower()
this.followers.add(follower)
try {
yield { type: 'baseline', value: this.baseline() }
yield* follower.read(signal)
} finally {
this.followers.delete(follower)
follower.close()
}
}
private changed(change: DomainChanged): void {
if (change.domain !== 'workspace') return
if (change.table === '') {
if (change.operation !== 'put') return
const state = workspaceDomainState.parse(change.value)
const nextOrder = state.workspaceIds.map(String)
const orderChanged = !sameStrings(this.order, nextOrder)
for (const id of state.workspaceIds) {
if (this.knownIds.has(id)) continue
const workspace = this.ctx.workspaceRegistry.get(id)
if (workspace === undefined) {
throw new Error(`committed Workspace registry references missing Workspace "${id}"`)
}
this.knownIds.add(id)
this.publish({ type: 'upsert', workspace: workspaceView(workspace) })
}
this.order = nextOrder
if (orderChanged) this.publish({ type: 'order', workspaceIds: [...state.workspaceIds] })
const nextArchived = state.archivedSessionIds.map(String)
if (!sameStrings(this.archived, nextArchived)) {
this.archived = nextArchived
this.publish({ type: 'archived', archivedSessionIds: [...state.archivedSessionIds] })
}
return
}
if (change.table !== 'workspaces') return
if (change.operation === 'deleted') {
if (!this.knownIds.delete(change.key)) return
this.publish({ type: 'remove', workspaceId: WorkspaceId(change.key) })
return
}
if (!this.knownIds.has(change.key)) return
this.publish({
type: 'upsert',
workspace: changedWorkspaceView(change.key, change.value),
})
}
private publish(frame: Exclude<WorkspaceFollowFrame, { readonly type: 'baseline' }>): void {
for (const follower of this.followers) follower.push(frame)
}
}
function sameStrings(left: readonly string[], right: readonly string[]): boolean {
return left.length === right.length && left.every((value, index) => value === right[index])
}
class WorkspaceFollower {
private readonly frames = new Deque<WorkspaceFollowFrame>()
private waiting: (() => void) | undefined
private closed = false
push(frame: WorkspaceFollowFrame): void {
/* v8 ignore next -- closed followers are removed before later publication can reach them. */
if (this.closed) return
this.frames.pushBack(frame)
this.waiting?.()
}
close(): void {
if (this.closed) return
this.closed = true
this.waiting?.()
}
async *read(signal: AbortSignal): AsyncIterable<WorkspaceFollowFrame> {
while (!this.closed && !signal.aborted) {
const frame = this.frames.popFront()
if (frame !== undefined) {
yield frame
continue
}
await this.wait(signal)
}
}
private wait(signal: AbortSignal): Promise<void> {
return new Promise((resolve) => {
const finish = (): void => {
signal.removeEventListener('abort', finish)
/* v8 ignore next -- one read owns the sole installed wait callback. */
if (this.waiting === finish) this.waiting = undefined
resolve()
}
this.waiting = finish
signal.addEventListener('abort', finish, { once: true })
/* v8 ignore next -- native signals and the private queue cannot change during this synchronous setup. */
if (signal.aborted || this.closed || this.frames.size > 0) finish()
})
}
}
|