|
|
@@ -135,6 +135,16 @@ export class ChatCacheStore {
|
|
|
const sessionColumns = this.db.prepare('PRAGMA table_info(chat_sessions)').all() as Array<{ name: string }>
|
|
|
this.db.exec('BEGIN IMMEDIATE')
|
|
|
try {
|
|
|
+ if (!sessionColumns.some(column => column.name === 'sync_revision')) {
|
|
|
+ this.db.exec('ALTER TABLE chat_sessions ADD COLUMN sync_revision INTEGER NOT NULL DEFAULT 0')
|
|
|
+ }
|
|
|
+ if (!sessionColumns.some(column => column.name === 'message_total')) {
|
|
|
+ this.db.exec('ALTER TABLE chat_sessions ADD COLUMN message_total INTEGER')
|
|
|
+ }
|
|
|
+ if (!sessionColumns.some(column => column.name === 'metadata_complete')) {
|
|
|
+ this.db.exec('ALTER TABLE chat_sessions ADD COLUMN metadata_complete INTEGER NOT NULL DEFAULT 0')
|
|
|
+ this.db.exec("UPDATE chat_sessions SET metadata_complete=1 WHERE json_valid(data) AND (json_type(data,'$.message_count') IS NOT NULL OR json_type(data,'$.model') IS NOT NULL)")
|
|
|
+ }
|
|
|
if (!sessionColumns.some(column => column.name === 'authoritative_title')) {
|
|
|
this.db.exec('ALTER TABLE chat_sessions ADD COLUMN authoritative_title TEXT')
|
|
|
}
|
|
|
@@ -227,6 +237,40 @@ export class ChatCacheStore {
|
|
|
return row ? { response: responseFromRow(row), state: String(row.sync_state) as ChatCacheState } : undefined
|
|
|
}
|
|
|
|
|
|
+ localList(owner: string, profile: string, page: { offset?: number; limit?: number; archived?: string }): UpstreamResponse {
|
|
|
+ return this.mergeOwnedSessionsIntoList(owner, profile, {
|
|
|
+ status: 200, headers: new Headers({ 'content-type': 'application/json' }),
|
|
|
+ body: Buffer.from('{"sessions":[]}'),
|
|
|
+ }, page)
|
|
|
+ }
|
|
|
+
|
|
|
+ needsListMetadata(owner: string, profile: string): boolean {
|
|
|
+ return Boolean(this.db.prepare(`SELECT 1 FROM chat_sessions WHERE owner=? AND owned_at IS NOT NULL
|
|
|
+ AND (?='' OR profile=?) AND metadata_complete=0 LIMIT 1`).get(owner, profile, profile))
|
|
|
+ }
|
|
|
+
|
|
|
+ localDetail(owner: string, profile: string, sessionID: string): StoredSnapshot | undefined {
|
|
|
+ const row = this.db.prepare(`SELECT data,sync_state FROM chat_sessions
|
|
|
+ WHERE owner=? AND profile=? AND session_id=? AND owned_at IS NOT NULL`)
|
|
|
+ .get(owner, profile, sessionID) as { data: string; sync_state: ChatCacheState } | undefined
|
|
|
+ if (!row) return undefined
|
|
|
+ return { response: { status: 200, headers: new Headers({ 'content-type': 'application/json' }),
|
|
|
+ body: Buffer.from(row.data) }, state: row.sync_state }
|
|
|
+ }
|
|
|
+
|
|
|
+ revision(owner: string, profile: string, sessionID: string): number {
|
|
|
+ return Number((this.db.prepare('SELECT sync_revision FROM chat_sessions WHERE owner=? AND profile=? AND session_id=?')
|
|
|
+ .get(owner, profile, sessionID) as { sync_revision?: number } | undefined)?.sync_revision ?? 0)
|
|
|
+ }
|
|
|
+
|
|
|
+ hasMessage(owner: string, profile: string, sessionID: string, message: Record<string, unknown>, position: number): boolean {
|
|
|
+ const id = messageID(message, String(position))
|
|
|
+ const row = this.db.prepare('SELECT position FROM chat_messages WHERE owner=? AND profile=? AND session_id=? AND message_id=?')
|
|
|
+ .get(owner, profile, sessionID, id) as { position: number } | undefined
|
|
|
+ if (row && row.position !== position) throw new Error('Stored history changed; explicit refresh is required')
|
|
|
+ return Boolean(row)
|
|
|
+ }
|
|
|
+
|
|
|
putSnapshot(owner: string, key: string, kind: string, profile: string, sessionID: string | undefined,
|
|
|
response: UpstreamResponse, ownedList = false, requestStartedAt = Date.now()): boolean {
|
|
|
const payload = parseResponse(response)
|
|
|
@@ -235,7 +279,7 @@ export class ChatCacheStore {
|
|
|
const isOwnedChat = ownedList || this.payloadBelongsToOwnedChat(owner, profile, sessionID, payload)
|
|
|
if (!isOwnedChat) return false
|
|
|
const now = Date.now()
|
|
|
- this.db.exec('BEGIN IMMEDIATE')
|
|
|
+ this.db.exec('SAVEPOINT chat_snapshot')
|
|
|
try {
|
|
|
this.ingestPayload(owner, profile, sessionID, payload, now, kind, requestStartedAt)
|
|
|
this.db.prepare(`INSERT INTO chat_snapshots(owner,cache_key,kind,profile,session_id,status,headers,body,sync_state,saved_at)
|
|
|
@@ -243,26 +287,44 @@ export class ChatCacheStore {
|
|
|
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')
|
|
|
+ this.db.exec('RELEASE chat_snapshot')
|
|
|
return true
|
|
|
} catch (error) {
|
|
|
- this.db.exec('ROLLBACK')
|
|
|
+ this.db.exec('ROLLBACK TO chat_snapshot; RELEASE chat_snapshot')
|
|
|
+ throw error
|
|
|
+ }
|
|
|
+ }
|
|
|
+
|
|
|
+ applySync(owner: string, profile: string, sessionID: string, revision: number,
|
|
|
+ pages: Array<{ key: string; kind: string; response: UpstreamResponse; startedAt: number }>, force: boolean): boolean {
|
|
|
+ if (this.revision(owner, profile, sessionID) !== revision) return false
|
|
|
+ this.db.exec('SAVEPOINT history_sync')
|
|
|
+ try {
|
|
|
+ if (force) this.db.prepare('DELETE FROM chat_messages WHERE owner=? AND profile=? AND session_id=?').run(owner, profile, sessionID)
|
|
|
+ this.db.prepare("DELETE FROM chat_snapshots WHERE owner=? AND profile=? AND session_id=? AND kind='messages'").run(owner, profile, sessionID)
|
|
|
+ for (const page of pages) this.putSnapshot(owner, page.key, page.kind, profile, sessionID, page.response, false, page.startedAt)
|
|
|
+ this.db.exec('RELEASE history_sync')
|
|
|
+ return true
|
|
|
+ } catch (error) {
|
|
|
+ this.db.exec('ROLLBACK TO history_sync; RELEASE history_sync')
|
|
|
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
|
|
|
+ const session = this.db.prepare(`SELECT data,sync_state,complete,message_total FROM chat_sessions
|
|
|
WHERE owner=? AND profile=? AND session_id=?`).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)
|
|
|
+ if (!session) return undefined
|
|
|
+ const totalRow = this.db.prepare('SELECT COUNT(*) count, MAX(position)+1 extent FROM chat_messages WHERE owner=? AND profile=? AND session_id=?')
|
|
|
+ .get(owner, profile, sessionID) as { count: number; extent: number | null }
|
|
|
+ if (session.message_total == null && !bool(session.complete)) return undefined
|
|
|
+ const total = Number(session.message_total ?? totalRow.extent ?? 0)
|
|
|
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))
|
|
|
+ if (messages.length !== end - start) return undefined // Fetch only an actually missing page.
|
|
|
const body = Buffer.from(JSON.stringify({
|
|
|
session_id: sessionID,
|
|
|
session: JSON.parse(String(session.data)),
|
|
|
@@ -281,7 +343,7 @@ export class ChatCacheStore {
|
|
|
}
|
|
|
}
|
|
|
|
|
|
- recordEvent(owner: string, profile: string, sessionID: string, frame: Record<string, unknown>): void {
|
|
|
+ recordEvent(owner: string, profile: string, sessionID: string, frame: Record<string, unknown>): boolean {
|
|
|
const rawSeq = Number(frame.seq ?? frame.sequence ?? 0)
|
|
|
const type = String(frame.type ?? 'unknown')
|
|
|
const payload = JSON.stringify(frame.payload ?? {})
|
|
|
@@ -292,22 +354,38 @@ export class ChatCacheStore {
|
|
|
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)
|
|
|
+ const inserted = 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)
|
|
|
+ if (!inserted.changes) return false
|
|
|
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)
|
|
|
+ const metadataOnly = ['session.info', 'session.title', 'session.title.updated', 'session.usage', 'gateway.ready'].includes(type) && state !== 'gap'
|
|
|
+ this.db.prepare(`UPDATE chat_sessions SET sync_state=CASE WHEN ? THEN sync_state ELSE ? END,
|
|
|
+ sync_revision=sync_revision+?,last_event_seq=MAX(last_event_seq,?),last_synced_at=?
|
|
|
+ WHERE owner=? AND profile=? AND session_id=?`).run(metadataOnly ? 1 : 0, state, metadataOnly ? 0 : 1, rawSeq > 0 ? rawSeq : lastSeen, now, owner, profile, sessionID)
|
|
|
+ if (type === 'session.info') {
|
|
|
+ const payload = object(frame.payload) ?? {}
|
|
|
+ const info = object(payload.info) ?? payload
|
|
|
+ const row = this.db.prepare('SELECT data FROM chat_sessions WHERE owner=? AND profile=? AND session_id=?')
|
|
|
+ .get(owner, profile, sessionID) as { data: string } | undefined
|
|
|
+ if (row) {
|
|
|
+ const summary = object(JSON.parse(row.data)) ?? {}
|
|
|
+ for (const field of ['model', 'provider', 'fast']) if (info[field] !== undefined) summary[field] = info[field]
|
|
|
+ this.db.prepare('UPDATE chat_sessions SET data=? WHERE owner=? AND profile=? AND session_id=?')
|
|
|
+ .run(JSON.stringify(summary), owner, profile, sessionID)
|
|
|
+ }
|
|
|
+ }
|
|
|
if (type === 'session.title' || type === 'session.title.updated') {
|
|
|
const title = String(object(frame.payload)?.title ?? frame.title ?? '').trim()
|
|
|
if (title) this.updateOwnedSessionTitle(owner, profile, sessionID, title, now)
|
|
|
}
|
|
|
- this.db.prepare(`UPDATE chat_snapshots SET sync_state=? WHERE owner=? AND (session_id=? OR kind='list')`)
|
|
|
- .run(state, owner, sessionID)
|
|
|
+ this.db.prepare(`UPDATE chat_snapshots SET sync_state=? WHERE owner=? AND (session_id=? OR kind='list')
|
|
|
+ AND (?=0 OR kind<>'messages')`).run(state, owner, sessionID, metadataOnly ? 1 : 0)
|
|
|
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)
|
|
|
+ return true
|
|
|
}
|
|
|
|
|
|
recordRoute(owner: string, profile: string, sessionID: string, runtimeID: string): void {
|
|
|
@@ -330,7 +408,7 @@ export class ChatCacheStore {
|
|
|
this.db.prepare(`INSERT INTO chat_sessions(owner,profile,session_id,source,data,sync_state,complete,last_synced_at,owned_at,ownership_evidence)
|
|
|
VALUES(?,?,?,?,?,?,?,?,?,?) ON CONFLICT(owner,profile,session_id) DO UPDATE SET
|
|
|
source=CASE WHEN excluded.source='unknown' THEN chat_sessions.source ELSE excluded.source END,
|
|
|
- sync_state='stale',complete=0,last_synced_at=excluded.last_synced_at,
|
|
|
+ sync_state='stale',sync_revision=chat_sessions.sync_revision+1,last_synced_at=excluded.last_synced_at,
|
|
|
owned_at=COALESCE(chat_sessions.owned_at,excluded.owned_at),
|
|
|
ownership_evidence=COALESCE(chat_sessions.ownership_evidence,excluded.ownership_evidence)`)
|
|
|
.run(owner, profile, sessionID, source, JSON.stringify(summary), 'stale', 0, now,
|
|
|
@@ -681,8 +759,8 @@ export class ChatCacheStore {
|
|
|
const storedSession = preservesAuthoritativeTitle
|
|
|
? { ...session, title: authoritativeTitle }
|
|
|
: session
|
|
|
- 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,
|
|
|
+ this.db.prepare(`INSERT INTO chat_sessions(owner,profile,session_id,runtime_id,source,data,sync_state,complete,last_synced_at,metadata_complete)
|
|
|
+ VALUES(?,?,?,?,?,?,?,?,?,1) ON CONFLICT(owner,profile,session_id) DO UPDATE SET data=excluded.data,metadata_complete=1,
|
|
|
source=CASE WHEN excluded.source='unknown' THEN chat_sessions.source ELSE excluded.source END,
|
|
|
sync_state='current',last_synced_at=excluded.last_synced_at`)
|
|
|
.run(owner, profile, id, null, source || 'unknown', JSON.stringify(storedSession), 'current', 0, now)
|
|
|
@@ -700,7 +778,7 @@ export class ChatCacheStore {
|
|
|
}
|
|
|
}
|
|
|
const messages = Array.isArray(payload.messages) ? payload.messages.flatMap(value => object(value) ? [object(value)!] : []) : []
|
|
|
- if (!messages.length || !sessionID) return
|
|
|
+ if (!Array.isArray(payload.messages) || !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)
|
|
|
@@ -710,7 +788,8 @@ export class ChatCacheStore {
|
|
|
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`)
|
|
|
+ data=excluded.data,content_hash=excluded.content_hash
|
|
|
+ WHERE chat_messages.content_hash<>excluded.content_hash OR chat_messages.position<>excluded.position`)
|
|
|
.run(owner, profile, sessionID, messageID(message, String(start + index)), start + index, data,
|
|
|
createHash('sha256').update(data).digest('hex'))
|
|
|
for (const sourcePath of referencedPaths(message)) {
|
|
|
@@ -719,10 +798,12 @@ export class ChatCacheStore {
|
|
|
.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)
|
|
|
+ const coverage = this.db.prepare(`SELECT COUNT(*) count, COUNT(DISTINCT position) positions, MIN(position) first, MAX(position) last
|
|
|
+ FROM chat_messages WHERE owner=? AND profile=? AND session_id=?`).get(owner, profile, sessionID) as Record<string, number>
|
|
|
+ const complete = total === 0 ? coverage.count === 0
|
|
|
+ : coverage.count === total && coverage.positions === total && coverage.first === 0 && coverage.last === total - 1
|
|
|
+ this.db.prepare(`UPDATE chat_sessions SET complete=?,message_total=?,sync_state='current',last_synced_at=?
|
|
|
+ WHERE owner=? AND profile=? AND session_id=?`).run(complete ? 1 : 0, total, now, owner, profile, sessionID)
|
|
|
}
|
|
|
|
|
|
private updateOwnedSessionTitle(owner: string, profile: string, sessionID: string, title: string, now: number): void {
|
|
|
@@ -748,6 +829,7 @@ export class ChatCacheStore {
|
|
|
|
|
|
export class ChatCacheCoordinator {
|
|
|
private reconciling = new Map<string, Promise<void>>()
|
|
|
+ onSynchronized: (owner: string, profile: string, sessionID: string) => void = () => {}
|
|
|
constructor(readonly store: ChatCacheStore, readonly upstream: UpstreamServiceSession,
|
|
|
readonly mode: ChatCacheMode = 'prefer-local') {}
|
|
|
|
|
|
@@ -764,15 +846,20 @@ export class ChatCacheCoordinator {
|
|
|
state: 'current',
|
|
|
}
|
|
|
}
|
|
|
- const local = this.store.snapshot(owner, key)
|
|
|
+ if (this.mode === 'prefer-local' && kind === 'list' && ownedList && !this.store.needsListMetadata(owner, profile)) {
|
|
|
+ return { response: this.store.localList(owner, profile, listPage), source: 'local', state: 'current' }
|
|
|
+ }
|
|
|
+ const local = kind === 'list' && ownedList && this.store.needsListMetadata(owner, profile) ? undefined : kind === 'detail' && sessionID
|
|
|
+ ? this.store.localDetail(owner, profile, sessionID) ?? this.store.snapshot(owner, key)
|
|
|
+ : this.store.snapshot(owner, key)
|
|
|
const projectedLocal = local && kind === 'list' && ownedList
|
|
|
? {
|
|
|
...local,
|
|
|
response: this.store.mergeOwnedSessionsIntoList(owner, profile, local.response, listPage),
|
|
|
}
|
|
|
: local
|
|
|
- if (this.mode === 'prefer-local' && local?.state === 'current') {
|
|
|
- return { response: local.response, source: 'local', state: local.state }
|
|
|
+ if (this.mode === 'prefer-local' && local) {
|
|
|
+ return { response: projectedLocal!.response, source: 'local', state: local.state }
|
|
|
}
|
|
|
try {
|
|
|
const requestStartedAt = Date.now()
|
|
|
@@ -799,22 +886,19 @@ export class ChatCacheCoordinator {
|
|
|
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') {
|
|
|
+ // A completion may still be committing its authoritative tail. Join it for
|
|
|
+ // the latest page; older, already stored pages never wait for the network.
|
|
|
+ if (offset === 0) await this.reconciling.get(JSON.stringify([owner, profile, sessionID]))?.catch(() => {})
|
|
|
const page = this.store.messagePage(owner, profile, sessionID, offset, limit)
|
|
|
- if (page?.state === 'current') return { response: page.response, source: 'local', state: page.state }
|
|
|
+ if (page) 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.ownsSession(owner, profile, sessionID)
|
|
|
- && !this.store.messagePage(owner, profile, sessionID, 0, 1)) {
|
|
|
- void this.reconcile(owner, profile, sessionID)
|
|
|
- }
|
|
|
- return result
|
|
|
+ return this.read(owner, key, 'messages', profile, sessionID, load)
|
|
|
}
|
|
|
|
|
|
observe(owner: string, profile: string, sessionID: string, frame: Record<string, unknown>): void {
|
|
|
- this.store.recordEvent(owner, profile, sessionID, frame)
|
|
|
- if (this.mode !== 'upstream-only' && String(frame.type) === 'message.complete') {
|
|
|
- void this.reconcile(owner, profile, sessionID)
|
|
|
+ if (!this.store.recordEvent(owner, profile, sessionID, frame)) return
|
|
|
+ if (['message.complete', 'gateway.reset'].includes(String(frame.type))) {
|
|
|
+ void this.reconcile(owner, profile, sessionID).catch(() => {})
|
|
|
}
|
|
|
}
|
|
|
|
|
|
@@ -831,29 +915,60 @@ export class ChatCacheCoordinator {
|
|
|
this.store.recordRoute(owner, profile, sessionID, runtimeID)
|
|
|
}
|
|
|
|
|
|
- async reconcile(owner: string, profile: string, sessionID: string): Promise<void> {
|
|
|
- if (this.mode === 'upstream-only') return
|
|
|
+ async reconcile(owner: string, profile: string, sessionID: string, force = false): Promise<void> {
|
|
|
+ if (this.mode === 'upstream-only' || !this.store.ownsSession(owner, profile, sessionID)) return
|
|
|
const key = JSON.stringify([owner, profile, sessionID])
|
|
|
- if (this.reconciling.has(key)) return this.reconciling.get(key)
|
|
|
+ const pending = this.reconciling.get(key)
|
|
|
+ if (pending) {
|
|
|
+ await pending
|
|
|
+ if (force || this.store.localDetail(owner, profile, sessionID)?.state !== 'current') {
|
|
|
+ return this.reconcile(owner, profile, sessionID, force)
|
|
|
+ }
|
|
|
+ return
|
|
|
+ }
|
|
|
const task = (async () => {
|
|
|
+ const revision = this.store.revision(owner, profile, sessionID)
|
|
|
+ const previous = this.store.messagePage(owner, profile, sessionID, 0, 1)
|
|
|
+ const previousTotal = previous ? Number(object(parseResponse(previous.response)?.pagination)?.total ?? 0) : 0
|
|
|
const detailPath = `/api/sessions/${encodeURIComponent(sessionID)}`
|
|
|
- const detailSearch = new URLSearchParams({ profile })
|
|
|
const detailStartedAt = Date.now()
|
|
|
- const detail = await this.upstream.request(detailPath, { search: detailSearch })
|
|
|
- this.store.putSnapshot(owner, `detail:${profile}:${sessionID}`, 'detail', profile, sessionID, detail, false, detailStartedAt)
|
|
|
+ const detail = await this.upstream.request(detailPath, { search: new URLSearchParams({ profile }), cache: 'reload' })
|
|
|
+ if (detail.status !== 200) throw new Error('Unable to synchronize conversation metadata')
|
|
|
+ const pages = [{ key: `detail:${profile}:${sessionID}`, kind: 'detail', response: detail, startedAt: detailStartedAt }]
|
|
|
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 messagesStartedAt = Date.now()
|
|
|
- const response = await this.upstream.request(`${detailPath}/messages`, { search })
|
|
|
- if (!this.store.putSnapshot(owner, `messages:${profile}:${sessionID}:${offset}:500`, 'messages', profile, sessionID, response, false, messagesStartedAt)) return
|
|
|
+ let done = false
|
|
|
+ let expectedTotal: number | undefined
|
|
|
+ // Stop as soon as the authoritative tail overlaps durable messages. Only
|
|
|
+ // explicit refresh (or uncached history) walks all the way to the beginning.
|
|
|
+ for (let page = 0; page < 1_000; page++) {
|
|
|
+ const search = new URLSearchParams({ profile, offset: String(offset), limit: '100', order: 'latest', include_compacted: 'true' })
|
|
|
+ const startedAt = Date.now()
|
|
|
+ const response = await this.upstream.request(`${detailPath}/messages`, { search, cache: 'reload' })
|
|
|
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
|
|
|
+ if (response.status !== 200 || !Array.isArray(payload?.messages)) throw new Error('Unable to synchronize conversation messages')
|
|
|
+ const messages = payload.messages.flatMap(value => object(value) ? [object(value)!] : [])
|
|
|
+ const pagination = object(payload.pagination) ?? {}
|
|
|
+ const total = Number(pagination.total ?? payload.total ?? messages.length)
|
|
|
+ if (!force && total < previousTotal) throw new Error('Stored history changed; explicit refresh is required')
|
|
|
+ if (expectedTotal !== undefined && expectedTotal !== total) throw new Error('Conversation changed during synchronization')
|
|
|
+ expectedTotal = total
|
|
|
+ const anchors = !force ? messages.map((message, index) =>
|
|
|
+ this.store.hasMessage(owner, profile, sessionID, message, total - offset - messages.length + index)) : []
|
|
|
+ const overlap = anchors.some(Boolean)
|
|
|
+ pages.push({ key: `messages:${profile}:${sessionID}:${offset}:100`, kind: 'messages', response, startedAt })
|
|
|
+ if (overlap || !bool(pagination.hasMore ?? pagination.has_more ?? payload.hasMore ?? payload.has_more)) { done = true; break }
|
|
|
+ if (!messages.length) break
|
|
|
offset += messages.length
|
|
|
}
|
|
|
- })().catch(() => {}).finally(() => this.reconciling.delete(key))
|
|
|
+ if (!done) throw new Error('Conversation synchronization is incomplete')
|
|
|
+ if (!this.store.applySync(owner, profile, sessionID, revision, pages, force)) {
|
|
|
+ // Preserve the invalidation when a newer event arrived during the read.
|
|
|
+ // Its completion joins this flight and schedules the next tail read.
|
|
|
+ if (force) throw new Error('Conversation changed during refresh; retry after the reply completes')
|
|
|
+ } else {
|
|
|
+ this.onSynchronized(owner, profile, sessionID)
|
|
|
+ }
|
|
|
+ })().finally(() => { if (this.reconciling.get(key) === task) this.reconciling.delete(key) })
|
|
|
this.reconciling.set(key, task)
|
|
|
return task
|
|
|
}
|
LSamien