| 123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354355356357358359360361362363364365366367368369370371372373374375376377378379380381382383384385386387388389390391392393394395396397398399400401402403404405406407408409410411412413414415416417418419420421422423424425426427428429430431432433434435436437438439440441442443444445446447448449450451452453454455456457458459 |
- import { createHash } from 'node:crypto'
- import { chmodSync, createReadStream, existsSync, mkdirSync, renameSync, statSync, unlinkSync, writeFileSync } from 'node:fs'
- import { join } from 'node:path'
- import { DatabaseSync } from 'node:sqlite'
- import type { UpstreamResponse } from './upstream.js'
- import type { UpstreamServiceSession } from './localAuth.js'
- export type ChatCacheMode = 'upstream-only' | 'shadow' | 'prefer-local'
- export type ChatCacheState = 'current' | 'stale' | 'syncing' | 'gap'
- export type ChatCacheSource = 'local' | 'upstream'
- interface StoredSnapshot {
- response: UpstreamResponse
- state: ChatCacheState
- }
- interface MessagePage {
- response: UpstreamResponse
- state: ChatCacheState
- }
- function object(value: unknown): Record<string, unknown> | undefined {
- return value && typeof value === 'object' && !Array.isArray(value)
- ? value as Record<string, unknown> : undefined
- }
- function parseResponse(response: UpstreamResponse): Record<string, unknown> | undefined {
- if (response.status < 200 || response.status >= 300) return undefined
- try { return object(JSON.parse(response.body.toString('utf8'))) } catch { return undefined }
- }
- function headerRecord(headers: Headers): string {
- return JSON.stringify(Object.fromEntries([...headers.entries()].filter(([name]) => name.toLowerCase() !== 'set-cookie')))
- }
- function responseFromRow(row: Record<string, unknown>): UpstreamResponse {
- return {
- status: Number(row.status),
- headers: new Headers(JSON.parse(String(row.headers)) as Record<string, string>),
- body: Buffer.from(String(row.body), 'base64'),
- }
- }
- function profileOf(value: Record<string, unknown>, fallback = ''): string {
- return String(value.profile ?? value.profile_name ?? fallback).trim()
- }
- function sessionIDOf(value: Record<string, unknown>): string {
- return String(value.id ?? value.session_id ?? value.sessionId ?? '').trim()
- }
- function sessionSource(value: Record<string, unknown>): string {
- return String(value.source ?? '').trim()
- }
- function messageID(value: Record<string, unknown>, fallback: string): string {
- const id = String(value.id ?? value.row_id ?? value.message_id ?? value.messageId ?? '').trim()
- return id || createHash('sha256').update(JSON.stringify(value) + fallback).digest('hex')
- }
- function bool(value: unknown): boolean {
- return value === true || value === 1 || value === '1' || value === 'true'
- }
- const PATH_PATTERN = /(?:file:\/\/)?(\/(?:Users|private|var|tmp)\/[^\s<>'"`)\]}]+\.[A-Za-z0-9]{1,12})/g
- function referencedPaths(value: unknown, output = new Set<string>()): Set<string> {
- if (typeof value === 'string') {
- for (const match of value.matchAll(PATH_PATTERN)) output.add(match[1]!)
- } else if (Array.isArray(value)) {
- for (const child of value) referencedPaths(child, output)
- } else {
- const record = object(value)
- if (record) for (const child of Object.values(record)) referencedPaths(child, output)
- }
- return output
- }
- export class ChatCacheStore {
- readonly db: DatabaseSync
- readonly assetsRoot: string
- constructor(home: string) {
- mkdirSync(home, { recursive: true, mode: 0o700 })
- const path = join(home, 'chat-cache.sqlite3')
- this.db = new DatabaseSync(path)
- chmodSync(path, 0o600)
- this.assetsRoot = join(home, 'chat-cache-assets')
- mkdirSync(this.assetsRoot, { recursive: true, mode: 0o700 })
- this.db.exec(`PRAGMA journal_mode=WAL; PRAGMA foreign_keys=ON; PRAGMA busy_timeout=5000;
- CREATE TABLE IF NOT EXISTS chat_sessions(
- owner TEXT NOT NULL, profile TEXT NOT NULL, session_id TEXT NOT NULL,
- runtime_id TEXT, source TEXT NOT NULL, data TEXT NOT NULL,
- sync_state TEXT NOT NULL DEFAULT 'stale', complete INTEGER NOT NULL DEFAULT 0,
- last_event_seq INTEGER NOT NULL DEFAULT 0, last_synced_at INTEGER NOT NULL,
- PRIMARY KEY(owner,profile,session_id));
- CREATE TABLE IF NOT EXISTS chat_messages(
- owner TEXT NOT NULL, profile TEXT NOT NULL, session_id TEXT NOT NULL,
- message_id TEXT NOT NULL, position INTEGER NOT NULL, data TEXT NOT NULL,
- content_hash TEXT NOT NULL, PRIMARY KEY(owner,profile,session_id,message_id));
- CREATE INDEX IF NOT EXISTS chat_messages_position ON chat_messages(owner,profile,session_id,position);
- CREATE TABLE IF NOT EXISTS chat_events(
- owner TEXT NOT NULL, profile TEXT NOT NULL, session_id TEXT NOT NULL,
- event_seq INTEGER NOT NULL, event_type TEXT NOT NULL, payload TEXT NOT NULL,
- received_at INTEGER NOT NULL, PRIMARY KEY(owner,profile,session_id,event_seq,event_type));
- CREATE TABLE IF NOT EXISTS chat_attachments(
- owner TEXT NOT NULL, profile TEXT NOT NULL, session_id TEXT NOT NULL,
- source_path TEXT NOT NULL, sha256 TEXT, mime_type TEXT, size INTEGER,
- local_path TEXT, state TEXT NOT NULL DEFAULT 'pending', updated_at INTEGER NOT NULL,
- PRIMARY KEY(owner,profile,session_id,source_path));
- CREATE INDEX IF NOT EXISTS chat_attachments_source ON chat_attachments(owner,source_path);
- CREATE TABLE IF NOT EXISTS chat_sync_cursors(
- owner TEXT NOT NULL, profile TEXT NOT NULL, generation TEXT,
- event_cursor INTEGER NOT NULL DEFAULT 0, sync_state TEXT NOT NULL DEFAULT 'stale',
- last_synced_at INTEGER NOT NULL, PRIMARY KEY(owner,profile));
- CREATE TABLE IF NOT EXISTS chat_snapshots(
- owner TEXT NOT NULL, cache_key TEXT NOT NULL, kind TEXT NOT NULL,
- profile TEXT NOT NULL, session_id TEXT, status INTEGER NOT NULL,
- headers TEXT NOT NULL, body TEXT NOT NULL, sync_state TEXT NOT NULL,
- saved_at INTEGER NOT NULL, PRIMARY KEY(owner,cache_key));`)
- }
- close(): void { this.db.close() }
- stats(): Record<string, number> {
- const count = (table: string) => Number((this.db.prepare(`SELECT COUNT(*) count FROM ${table}`).get() as { count: number }).count)
- const bytes = Number((this.db.prepare(`SELECT COALESCE(SUM(size),0) bytes FROM chat_attachments WHERE state='cached'`).get() as { bytes: number }).bytes)
- return {
- sessions: count('chat_sessions'),
- messages: count('chat_messages'),
- events: count('chat_events'),
- attachments: count('chat_attachments'),
- attachmentBytes: bytes,
- }
- }
- snapshot(owner: string, key: string): StoredSnapshot | undefined {
- const row = this.db.prepare('SELECT status,headers,body,sync_state FROM chat_snapshots WHERE owner=? AND cache_key=?')
- .get(owner, key) as Record<string, unknown> | undefined
- return row ? { response: responseFromRow(row), state: String(row.sync_state) as ChatCacheState } : undefined
- }
- putSnapshot(owner: string, key: string, kind: string, profile: string, sessionID: string | undefined,
- response: UpstreamResponse, knownWeb = false): boolean {
- const payload = parseResponse(response)
- if (!payload) return false
- if (kind === 'list' && !knownWeb) return false
- const isWeb = knownWeb || this.payloadIsWeb(owner, profile, sessionID, payload)
- if (!isWeb) return false
- const now = Date.now()
- this.db.exec('BEGIN IMMEDIATE')
- try {
- this.ingestPayload(owner, profile, sessionID, payload, now)
- this.db.prepare(`INSERT INTO chat_snapshots(owner,cache_key,kind,profile,session_id,status,headers,body,sync_state,saved_at)
- VALUES(?,?,?,?,?,?,?,?,?,?) ON CONFLICT(owner,cache_key) DO UPDATE SET status=excluded.status,headers=excluded.headers,
- body=excluded.body,sync_state='current',saved_at=excluded.saved_at`)
- .run(owner, key, kind, profile, sessionID ?? null, response.status, headerRecord(response.headers),
- response.body.toString('base64'), 'current', now)
- this.db.exec('COMMIT')
- return true
- } catch (error) {
- this.db.exec('ROLLBACK')
- throw error
- }
- }
- messagePage(owner: string, profile: string, sessionID: string, offset: number, limit: number): MessagePage | undefined {
- const session = this.db.prepare(`SELECT data,sync_state,complete FROM chat_sessions
- WHERE owner=? AND profile=? AND session_id=? AND source='web'`).get(owner, profile, sessionID) as Record<string, unknown> | undefined
- if (!session || !bool(session.complete)) return undefined
- const totalRow = this.db.prepare('SELECT COUNT(*) count FROM chat_messages WHERE owner=? AND profile=? AND session_id=?')
- .get(owner, profile, sessionID) as { count: number }
- const total = Number(totalRow.count)
- const end = Math.max(0, total - offset)
- const start = Math.max(0, end - limit)
- const rows = this.db.prepare(`SELECT data FROM chat_messages WHERE owner=? AND profile=? AND session_id=?
- AND position>=? AND position<? ORDER BY position`).all(owner, profile, sessionID, start, end) as Array<{ data: string }>
- const messages = rows.map(row => JSON.parse(row.data))
- const body = Buffer.from(JSON.stringify({
- session_id: sessionID,
- session: JSON.parse(String(session.data)),
- messages,
- pagination: {
- total,
- returned: messages.length,
- offset,
- limit,
- has_more: start > 0,
- },
- }))
- return {
- response: { status: 200, headers: new Headers({ 'content-type': 'application/json; charset=utf-8' }), body },
- state: String(session.sync_state) as ChatCacheState,
- }
- }
- recordEvent(owner: string, profile: string, sessionID: string, frame: Record<string, unknown>): void {
- const rawSeq = Number(frame.seq ?? frame.sequence ?? 0)
- const type = String(frame.type ?? 'unknown')
- const payload = JSON.stringify(frame.payload ?? {})
- const seq = Number.isSafeInteger(rawSeq) && rawSeq > 0
- ? rawSeq
- : Number.parseInt(createHash('sha256').update(`${type}:${payload}`).digest('hex').slice(0, 12), 16)
- const now = Date.now()
- const previous = this.db.prepare(`SELECT last_event_seq FROM chat_sessions
- WHERE owner=? AND profile=? AND session_id=?`).get(owner, profile, sessionID) as { last_event_seq?: number } | undefined
- const lastSeen = Number(previous?.last_event_seq ?? 0)
- this.db.prepare(`INSERT OR IGNORE INTO chat_events(owner,profile,session_id,event_seq,event_type,payload,received_at)
- VALUES(?,?,?,?,?,?,?)`).run(owner, profile, sessionID, Number.isSafeInteger(seq) ? seq : 0, type, payload, now)
- const state: ChatCacheState = type === 'gateway.reset' || type.includes('reset')
- || (rawSeq > 0 && lastSeen > 0 && rawSeq > lastSeen + 1) ? 'gap' : 'stale'
- this.db.prepare(`UPDATE chat_sessions SET sync_state=?,last_event_seq=MAX(last_event_seq,?),last_synced_at=?
- WHERE owner=? AND profile=? AND session_id=?`).run(state, rawSeq > 0 ? rawSeq : lastSeen, now, owner, profile, sessionID)
- this.db.prepare(`UPDATE chat_snapshots SET sync_state=? WHERE owner=? AND (session_id=? OR kind='list')`)
- .run(state, owner, sessionID)
- this.db.prepare(`INSERT INTO chat_sync_cursors(owner,profile,event_cursor,sync_state,last_synced_at)
- VALUES(?,?,?,?,?) ON CONFLICT(owner,profile) DO UPDATE SET event_cursor=MAX(event_cursor,excluded.event_cursor),
- sync_state=excluded.sync_state,last_synced_at=excluded.last_synced_at`)
- .run(owner, profile, rawSeq > 0 ? rawSeq : lastSeen, state, now)
- }
- recordRoute(owner: string, profile: string, sessionID: string, runtimeID: string): void {
- const now = Date.now()
- const summary = { id: sessionID, profile, source: 'web', title: '新对话', started_at: now, last_active: now }
- this.db.prepare(`INSERT INTO chat_sessions(owner,profile,session_id,runtime_id,source,data,sync_state,complete,last_synced_at)
- VALUES(?,?,?,?,?,?,?,?,?) ON CONFLICT(owner,profile,session_id) DO UPDATE SET runtime_id=excluded.runtime_id,
- source='web',last_synced_at=excluded.last_synced_at`)
- .run(owner, profile, sessionID, runtimeID, 'web', JSON.stringify(summary), 'stale', 0, now)
- }
- recordCommand(owner: string, profile: string, sessionID: string, method: string,
- params: Record<string, unknown>): void {
- const now = Date.now()
- const summary = { id: sessionID, profile, source: 'web', title: '新对话', started_at: now, last_active: now }
- this.db.prepare(`INSERT INTO chat_sessions(owner,profile,session_id,source,data,sync_state,complete,last_synced_at)
- VALUES(?,?,?,?,?,?,?,?) ON CONFLICT(owner,profile,session_id) DO UPDATE SET sync_state='stale',complete=0`)
- .run(owner, profile, sessionID, 'web', JSON.stringify(summary), 'stale', 0, now)
- const payload = JSON.stringify(params)
- const seq = Number.parseInt(createHash('sha256').update(`${method}:${payload}`).digest('hex').slice(0, 12), 16)
- this.db.prepare(`INSERT OR IGNORE INTO chat_events(owner,profile,session_id,event_seq,event_type,payload,received_at)
- VALUES(?,?,?,?,?,?,?)`).run(owner, profile, sessionID, seq, `command:${method}`, payload, now)
- this.db.prepare("UPDATE chat_snapshots SET sync_state='stale' WHERE owner=? AND (session_id=? OR kind='list')")
- .run(owner, sessionID)
- }
- markListsStale(owner: string): void {
- this.db.prepare("UPDATE chat_snapshots SET sync_state='stale' WHERE owner=? AND kind='list'").run(owner)
- }
- deleteSession(owner: string, profile: string, sessionID: string): void {
- const assets = this.db.prepare(`SELECT DISTINCT local_path FROM chat_attachments
- WHERE owner=? AND profile=? AND session_id=? AND local_path IS NOT NULL`)
- .all(owner, profile, sessionID) as Array<{ local_path: string }>
- this.db.exec('BEGIN IMMEDIATE')
- try {
- for (const table of ['chat_messages', 'chat_events', 'chat_attachments', 'chat_sessions']) {
- this.db.prepare(`DELETE FROM ${table} WHERE owner=? AND profile=? AND session_id=?`).run(owner, profile, sessionID)
- }
- this.db.prepare("DELETE FROM chat_snapshots WHERE owner=? AND (session_id=? OR kind='list')").run(owner, sessionID)
- this.db.exec('COMMIT')
- for (const asset of assets) {
- const referenced = this.db.prepare('SELECT 1 ok FROM chat_attachments WHERE local_path=? LIMIT 1').get(asset.local_path)
- if (!referenced) try { unlinkSync(asset.local_path) } catch { /* Already absent. */ }
- }
- } catch (error) { this.db.exec('ROLLBACK'); throw error }
- }
- attachment(owner: string, sourcePath: string): { localPath: string; mimeType?: string } | undefined {
- const row = this.db.prepare(`SELECT local_path,mime_type FROM chat_attachments
- WHERE owner=? AND source_path=? AND state='cached' ORDER BY updated_at DESC LIMIT 1`).get(owner, sourcePath) as Record<string, unknown> | undefined
- const localPath = row?.local_path ? String(row.local_path) : ''
- return localPath && existsSync(localPath) ? { localPath, ...(row?.mime_type ? { mimeType: String(row.mime_type) } : {}) } : undefined
- }
- knowsAttachment(owner: string, sourcePath: string): boolean {
- return Boolean(this.db.prepare('SELECT 1 ok FROM chat_attachments WHERE owner=? AND source_path=? LIMIT 1').get(owner, sourcePath))
- }
- isWebSession(owner: string, profile: string, sessionID: string): boolean {
- return Boolean(this.db.prepare("SELECT 1 ok FROM chat_sessions WHERE owner=? AND profile=? AND session_id=? AND source='web'")
- .get(owner, profile, sessionID))
- }
- storeAttachment(owner: string, sourcePath: string, response: UpstreamResponse): string | undefined {
- if (response.status !== 200 || !response.body.length) return undefined
- const sha256 = createHash('sha256').update(response.body).digest('hex')
- const directory = join(this.assetsRoot, sha256.slice(0, 2))
- mkdirSync(directory, { recursive: true, mode: 0o700 })
- const target = join(directory, sha256)
- if (!existsSync(target)) {
- const temporary = `${target}.${process.pid}.tmp`
- writeFileSync(temporary, response.body, { mode: 0o600 })
- renameSync(temporary, target)
- }
- const mimeType = response.headers.get('content-type')?.split(';', 1)[0]
- this.db.prepare(`UPDATE chat_attachments SET sha256=?,mime_type=?,size=?,local_path=?,state='cached',updated_at=?
- WHERE owner=? AND source_path=?`).run(sha256, mimeType ?? null, response.body.length, target, Date.now(), owner, sourcePath)
- return target
- }
- assetStream(path: string): { stream: ReturnType<typeof createReadStream>; size: number } {
- return { stream: createReadStream(path), size: statSync(path).size }
- }
- private payloadIsWeb(owner: string, profile: string, sessionID: string | undefined, payload: Record<string, unknown>): boolean {
- const session = object(payload.session) ?? (sessionID ? payload : undefined)
- if (session && sessionSource(session) === 'web') return true
- const sessions = Array.isArray(payload.items) ? payload.items : Array.isArray(payload.sessions) ? payload.sessions : []
- if (sessions.some(value => object(value) && sessionSource(object(value)!) === 'web')) return true
- if (!sessionID) return false
- return Boolean(this.db.prepare("SELECT 1 ok FROM chat_sessions WHERE owner=? AND profile=? AND session_id=? AND source='web'")
- .get(owner, profile, sessionID))
- }
- private ingestPayload(owner: string, fallbackProfile: string, sessionID: string | undefined,
- payload: Record<string, unknown>, now: number): void {
- const candidates: Record<string, unknown>[] = []
- const direct = object(payload.session)
- if (direct) candidates.push(direct)
- const listed = Array.isArray(payload.items) ? payload.items : Array.isArray(payload.sessions) ? payload.sessions : []
- for (const value of listed) if (object(value)) candidates.push(object(value)!)
- if (sessionID && !direct && sessionSource(payload)) candidates.push(payload)
- for (const session of candidates) {
- if (sessionSource(session) !== 'web') continue
- const id = sessionIDOf(session)
- if (!id) continue
- const profile = profileOf(session, fallbackProfile) || fallbackProfile || 'default'
- this.db.prepare(`INSERT INTO chat_sessions(owner,profile,session_id,runtime_id,source,data,sync_state,complete,last_synced_at)
- VALUES(?,?,?,?,?,?,?,?,?) ON CONFLICT(owner,profile,session_id) DO UPDATE SET data=excluded.data,source='web',
- sync_state='current',last_synced_at=excluded.last_synced_at`)
- .run(owner, profile, id, null, 'web', JSON.stringify(session), 'current', 0, now)
- }
- const messages = Array.isArray(payload.messages) ? payload.messages.flatMap(value => object(value) ? [object(value)!] : []) : []
- if (!messages.length || !sessionID) return
- const profile = fallbackProfile || profileOf(direct ?? {}, 'default') || 'default'
- const pagination = object(payload.pagination) ?? {}
- const total = Math.max(messages.length, Number(pagination.total ?? payload.total ?? messages.length) || messages.length)
- const offset = Math.max(0, Number(pagination.offset ?? payload.offset ?? 0) || 0)
- const start = Math.max(0, total - offset - messages.length)
- messages.forEach((message, index) => {
- const data = JSON.stringify(message)
- this.db.prepare(`INSERT INTO chat_messages(owner,profile,session_id,message_id,position,data,content_hash)
- VALUES(?,?,?,?,?,?,?) ON CONFLICT(owner,profile,session_id,message_id) DO UPDATE SET position=excluded.position,
- data=excluded.data,content_hash=excluded.content_hash`)
- .run(owner, profile, sessionID, messageID(message, String(start + index)), start + index, data,
- createHash('sha256').update(data).digest('hex'))
- for (const sourcePath of referencedPaths(message)) {
- this.db.prepare(`INSERT INTO chat_attachments(owner,profile,session_id,source_path,state,updated_at)
- VALUES(?,?,?,?,?,?) ON CONFLICT(owner,profile,session_id,source_path) DO UPDATE SET updated_at=excluded.updated_at`)
- .run(owner, profile, sessionID, sourcePath, 'pending', now)
- }
- })
- const hasMore = bool(pagination.hasMore ?? pagination.has_more ?? payload.hasMore ?? payload.has_more)
- const complete = !hasMore && offset + messages.length >= total
- this.db.prepare(`UPDATE chat_sessions SET complete=MAX(complete,?),sync_state='current',last_synced_at=?
- WHERE owner=? AND profile=? AND session_id=?`).run(complete ? 1 : 0, now, owner, profile, sessionID)
- }
- }
- export class ChatCacheCoordinator {
- private reconciling = new Map<string, Promise<void>>()
- constructor(readonly store: ChatCacheStore, readonly upstream: UpstreamServiceSession,
- readonly mode: ChatCacheMode = 'prefer-local') {}
- async read(owner: string, key: string, kind: string, profile: string, sessionID: string | undefined,
- load: () => Promise<UpstreamResponse>, knownWeb = false): Promise<{ response: UpstreamResponse; source: ChatCacheSource; state: ChatCacheState }> {
- if (this.mode === 'upstream-only') return { response: await load(), source: 'upstream', state: 'current' }
- const local = this.store.snapshot(owner, key)
- if (this.mode === 'prefer-local' && local?.state === 'current') {
- return { response: local.response, source: 'local', state: local.state }
- }
- try {
- const response = await load()
- if (this.mode === 'prefer-local' && local && response.status >= 500) {
- return { response: local.response, source: 'local', state: 'stale' }
- }
- this.store.putSnapshot(owner, key, kind, profile, sessionID, response, knownWeb)
- return { response, source: 'upstream', state: 'current' }
- } catch (error) {
- if (this.mode === 'prefer-local' && local) return { response: local.response, source: 'local', state: 'stale' }
- throw error
- }
- }
- async messages(owner: string, key: string, profile: string, sessionID: string, offset: number, limit: number,
- load: () => Promise<UpstreamResponse>): Promise<{ response: UpstreamResponse; source: ChatCacheSource; state: ChatCacheState }> {
- if (this.mode === 'prefer-local') {
- const page = this.store.messagePage(owner, profile, sessionID, offset, limit)
- if (page?.state === 'current') return { response: page.response, source: 'local', state: page.state }
- }
- const result = await this.read(owner, key, 'messages', profile, sessionID, load)
- if (result.source === 'upstream'
- && this.store.isWebSession(owner, profile, sessionID)
- && !this.store.messagePage(owner, profile, sessionID, 0, 1)) {
- void this.reconcile(owner, profile, sessionID)
- }
- return result
- }
- observe(owner: string, profile: string, sessionID: string, frame: Record<string, unknown>): void {
- if (this.mode === 'upstream-only') return
- this.store.recordEvent(owner, profile, sessionID, frame)
- if (String(frame.type) === 'message.complete') void this.reconcile(owner, profile, sessionID)
- }
- observeGlobal(owner: string, type: string): void {
- if (this.mode !== 'upstream-only' && type === 'sessions.changed') this.store.markListsStale(owner)
- }
- command(owner: string, profile: string, sessionID: string, method: string,
- params: Record<string, unknown>): void {
- if (this.mode !== 'upstream-only') this.store.recordCommand(owner, profile, sessionID, method, params)
- }
- route(owner: string, profile: string, sessionID: string, runtimeID: string): void {
- if (this.mode !== 'upstream-only') this.store.recordRoute(owner, profile, sessionID, runtimeID)
- }
- async reconcile(owner: string, profile: string, sessionID: string): Promise<void> {
- if (this.mode === 'upstream-only') return
- const key = JSON.stringify([owner, profile, sessionID])
- if (this.reconciling.has(key)) return this.reconciling.get(key)
- const task = (async () => {
- const detailPath = `/api/sessions/${encodeURIComponent(sessionID)}`
- const detailSearch = new URLSearchParams({ profile })
- const detail = await this.upstream.request(detailPath, { search: detailSearch })
- this.store.putSnapshot(owner, `detail:${profile}:${sessionID}`, 'detail', profile, sessionID, detail)
- let offset = 0
- for (let page = 0; page < 200; page++) {
- const search = new URLSearchParams({ profile, offset: String(offset), limit: '500', order: 'latest', include_compacted: 'true' })
- const response = await this.upstream.request(`${detailPath}/messages`, { search })
- if (!this.store.putSnapshot(owner, `messages:${profile}:${sessionID}:${offset}:500`, 'messages', profile, sessionID, response)) return
- const payload = parseResponse(response)
- const messages = Array.isArray(payload?.messages) ? payload.messages : []
- const pagination = object(payload?.pagination) ?? {}
- if (!bool(pagination.hasMore ?? pagination.has_more ?? payload?.hasMore ?? payload?.has_more) || !messages.length) return
- offset += messages.length
- }
- })().catch(() => {}).finally(() => this.reconciling.delete(key))
- this.reconciling.set(key, task)
- return task
- }
- async cacheAttachment(owner: string, sourcePath: string): Promise<void> {
- if (this.mode === 'upstream-only' || !this.store.knowsAttachment(owner, sourcePath) || this.store.attachment(owner, sourcePath)) return
- try {
- const response = await this.upstream.request('/api/files/download', {
- search: new URLSearchParams({ path: sourcePath }), maxResponseBytes: 100 * 1_024 * 1_024,
- })
- this.store.storeAttachment(owner, sourcePath, response)
- } catch { /* The next authorized request can retry. */ }
- }
- }
- export function chatCacheKey(kind: string, profile: string, sessionID: string | undefined, search: URLSearchParams): string {
- const normalized = new URLSearchParams(search)
- normalized.sort()
- return [kind, profile, sessionID ?? '', normalized.toString()].join(':')
- }
|