|
@@ -531,13 +531,6 @@ export class ChatCacheStore {
|
|
|
this.markListsStale(owner)
|
|
this.markListsStale(owner)
|
|
|
}
|
|
}
|
|
|
|
|
|
|
|
- validationCandidates(owner: string, profile: string): Array<{ profile: string; session_id: string; sync_revision: number }> {
|
|
|
|
|
- // Newly created runtime sessions may not have been persisted by Hermes yet.
|
|
|
|
|
- return this.db.prepare(`SELECT profile,session_id,sync_revision FROM chat_sessions
|
|
|
|
|
- WHERE owner=? AND owned_at IS NOT NULL AND owned_at<? AND (?='' OR profile=?)`)
|
|
|
|
|
- .all(owner, Date.now() - 60_000, profile, profile) as Array<{ profile: string; session_id: string; sync_revision: number }>
|
|
|
|
|
- }
|
|
|
|
|
-
|
|
|
|
|
deleteSession(owner: string, profile: string, sessionID: string): void {
|
|
deleteSession(owner: string, profile: string, sessionID: string): void {
|
|
|
const assets = this.db.prepare(`SELECT DISTINCT local_path FROM chat_attachments
|
|
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`)
|
|
WHERE owner=? AND profile=? AND session_id=? AND local_path IS NOT NULL`)
|
|
@@ -944,66 +937,15 @@ export class ChatCacheStore {
|
|
|
|
|
|
|
|
export class ChatCacheCoordinator {
|
|
export class ChatCacheCoordinator {
|
|
|
private reconciling = new Map<string, Promise<void>>()
|
|
private reconciling = new Map<string, Promise<void>>()
|
|
|
- private validations = new Map<string, number>()
|
|
|
|
|
- private validating = new Set<string>()
|
|
|
|
|
- private stopped = false
|
|
|
|
|
onSynchronized: (owner: string, profile: string, sessionID: string) => void = () => {}
|
|
onSynchronized: (owner: string, profile: string, sessionID: string) => void = () => {}
|
|
|
constructor(readonly store: ChatCacheStore, readonly upstream: UpstreamServiceSession,
|
|
constructor(readonly store: ChatCacheStore, readonly upstream: UpstreamServiceSession,
|
|
|
readonly mode: ChatCacheMode = 'prefer-local') {}
|
|
readonly mode: ChatCacheMode = 'prefer-local') {}
|
|
|
|
|
|
|
|
- close(): void { this.stopped = true }
|
|
|
|
|
-
|
|
|
|
|
- private validateList(owner: string, profile: string, load: () => Promise<UpstreamResponse>): void {
|
|
|
|
|
- const key = JSON.stringify([owner, profile])
|
|
|
|
|
- if (this.stopped || this.validating.has(key) || Date.now() - (this.validations.get(key) ?? 0) < 30_000) return
|
|
|
|
|
- const candidates = this.store.validationCandidates(owner, profile)
|
|
|
|
|
- if (!candidates.length) return
|
|
|
|
|
- this.validations.set(key, Date.now())
|
|
|
|
|
- this.validating.add(key)
|
|
|
|
|
- void (async () => {
|
|
|
|
|
- const response = await load()
|
|
|
|
|
- if (this.stopped) return
|
|
|
|
|
- const payload = parseResponse(response)
|
|
|
|
|
- const items = payload?.sessions ?? payload?.items
|
|
|
|
|
- if (!Array.isArray(items)) return
|
|
|
|
|
- const existing = new Set(items.flatMap(value => {
|
|
|
|
|
- const session = object(value)
|
|
|
|
|
- return session ? [`${profileOf(session, profile) || 'default'}\u0000${sessionIDOf(session)}`] : []
|
|
|
|
|
- }))
|
|
|
|
|
- // A filtered/paginated list cannot prove deletion. Confirm each absent
|
|
|
|
|
- // candidate against its own profile; outages must retain local history.
|
|
|
|
|
- for (const candidate of candidates) {
|
|
|
|
|
- if (this.stopped) return
|
|
|
|
|
- if (existing.has(`${candidate.profile}\u0000${candidate.session_id}`)) continue
|
|
|
|
|
- try {
|
|
|
|
|
- const detail = await this.upstream.request(`/api/sessions/${encodeURIComponent(candidate.session_id)}`, {
|
|
|
|
|
- search: new URLSearchParams({ profile: candidate.profile }), cache: 'reload',
|
|
|
|
|
- })
|
|
|
|
|
- if (this.stopped) return
|
|
|
|
|
- if (detail.status === 404 && this.store.ownsSession(owner, candidate.profile, candidate.session_id)
|
|
|
|
|
- && this.store.revision(owner, candidate.profile, candidate.session_id) === candidate.sync_revision) {
|
|
|
|
|
- this.store.deleteSession(owner, candidate.profile, candidate.session_id)
|
|
|
|
|
- this.onSynchronized(owner, candidate.profile, candidate.session_id)
|
|
|
|
|
- }
|
|
|
|
|
- } catch { /* Retain the session and retry on a later list refresh. */ }
|
|
|
|
|
- }
|
|
|
|
|
- })().catch(() => {}).finally(() => this.validating.delete(key))
|
|
|
|
|
- }
|
|
|
|
|
-
|
|
|
|
|
async read(owner: string, key: string, kind: string, profile: string, sessionID: string | undefined,
|
|
async read(owner: string, key: string, kind: string, profile: string, sessionID: string | undefined,
|
|
|
load: () => Promise<UpstreamResponse>, ownedList = false,
|
|
load: () => Promise<UpstreamResponse>, ownedList = false,
|
|
|
listPage: { offset?: number; limit?: number; archived?: string } = {}): Promise<{ response: UpstreamResponse; source: ChatCacheSource; state: ChatCacheState }> {
|
|
listPage: { offset?: number; limit?: number; archived?: string } = {}): Promise<{ response: UpstreamResponse; source: ChatCacheSource; state: ChatCacheState }> {
|
|
|
- if (kind === 'list' && ownedList) this.validateList(owner, profile, load)
|
|
|
|
|
- const revision = sessionID && this.store.ownsSession(owner, profile, sessionID)
|
|
|
|
|
- ? this.store.revision(owner, profile, sessionID) : undefined
|
|
|
|
|
if (this.mode === 'upstream-only') {
|
|
if (this.mode === 'upstream-only') {
|
|
|
const upstreamResponse = await load()
|
|
const upstreamResponse = await load()
|
|
|
- if (sessionID && revision !== undefined && upstreamResponse.status === 404
|
|
|
|
|
- && this.store.ownsSession(owner, profile, sessionID)
|
|
|
|
|
- && this.store.revision(owner, profile, sessionID) === revision) {
|
|
|
|
|
- this.store.deleteSession(owner, profile, sessionID)
|
|
|
|
|
- this.onSynchronized(owner, profile, sessionID)
|
|
|
|
|
- }
|
|
|
|
|
return {
|
|
return {
|
|
|
response: kind === 'list' && ownedList
|
|
response: kind === 'list' && ownedList
|
|
|
? this.store.mergeOwnedSessionsIntoList(owner, profile, upstreamResponse, listPage)
|
|
? this.store.mergeOwnedSessionsIntoList(owner, profile, upstreamResponse, listPage)
|
|
@@ -1030,12 +972,6 @@ export class ChatCacheCoordinator {
|
|
|
try {
|
|
try {
|
|
|
const requestStartedAt = Date.now()
|
|
const requestStartedAt = Date.now()
|
|
|
const upstreamResponse = await load()
|
|
const upstreamResponse = await load()
|
|
|
- if (sessionID && revision !== undefined && upstreamResponse.status === 404
|
|
|
|
|
- && this.store.ownsSession(owner, profile, sessionID)
|
|
|
|
|
- && this.store.revision(owner, profile, sessionID) === revision) {
|
|
|
|
|
- this.store.deleteSession(owner, profile, sessionID)
|
|
|
|
|
- this.onSynchronized(owner, profile, sessionID)
|
|
|
|
|
- }
|
|
|
|
|
if (this.mode === 'prefer-local' && local && upstreamResponse.status >= 500) {
|
|
if (this.mode === 'prefer-local' && local && upstreamResponse.status >= 500) {
|
|
|
return { response: projectedLocal!.response, source: 'local', state: 'stale' }
|
|
return { response: projectedLocal!.response, source: 'local', state: 'stale' }
|
|
|
}
|
|
}
|
|
@@ -1105,11 +1041,6 @@ export class ChatCacheCoordinator {
|
|
|
const detailPath = `/api/sessions/${encodeURIComponent(sessionID)}`
|
|
const detailPath = `/api/sessions/${encodeURIComponent(sessionID)}`
|
|
|
const detailStartedAt = Date.now()
|
|
const detailStartedAt = Date.now()
|
|
|
const detail = await this.upstream.request(detailPath, { search: new URLSearchParams({ profile }), cache: 'reload' })
|
|
const detail = await this.upstream.request(detailPath, { search: new URLSearchParams({ profile }), cache: 'reload' })
|
|
|
- if (this.stopped) return
|
|
|
|
|
- if (detail.status === 404 && this.store.revision(owner, profile, sessionID) === revision) {
|
|
|
|
|
- this.store.deleteSession(owner, profile, sessionID)
|
|
|
|
|
- this.onSynchronized(owner, profile, sessionID)
|
|
|
|
|
- }
|
|
|
|
|
if (detail.status !== 200) throw new Error('Unable to synchronize conversation metadata')
|
|
if (detail.status !== 200) throw new Error('Unable to synchronize conversation metadata')
|
|
|
const detailPayload = parseResponse(detail) ?? {}
|
|
const detailPayload = parseResponse(detail) ?? {}
|
|
|
const detailSummary = object(detailPayload.session) ?? detailPayload
|
|
const detailSummary = object(detailPayload.session) ?? detailPayload
|
|
@@ -1125,7 +1056,6 @@ export class ChatCacheCoordinator {
|
|
|
const search = new URLSearchParams({ profile, offset: String(offset), limit: '100', order: 'latest', include_compacted: 'true' })
|
|
const search = new URLSearchParams({ profile, offset: String(offset), limit: '100', order: 'latest', include_compacted: 'true' })
|
|
|
const startedAt = Date.now()
|
|
const startedAt = Date.now()
|
|
|
const response = await this.upstream.request(`${detailPath}/messages`, { search, cache: 'reload' })
|
|
const response = await this.upstream.request(`${detailPath}/messages`, { search, cache: 'reload' })
|
|
|
- if (this.stopped) return
|
|
|
|
|
const payload = parseResponse(response)
|
|
const payload = parseResponse(response)
|
|
|
if (response.status !== 200 || !Array.isArray(payload?.messages)) throw new Error('Unable to synchronize conversation messages')
|
|
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 messages = payload.messages.flatMap(value => object(value) ? [object(value)!] : [])
|