wushenghua/hapi
1
0
Fork 0
mirror of https://github.com/wu736139669/hapi.git synced 2026-08-06 06:41:56 +00:00
Code Issues Packages Projects Releases Wiki Activity

refactor(sync): replace message reloads with incremental tail sync

Browse Source
This commit is contained in:
weishu
2026-07-28 12:20:53 +08:00
parent 2235b924a7
commit faf70c64dd
28 changed files with 2927 additions and 2077 deletions
Download Patch File Download Diff File Expand all files Collapse all files
hub/src/store/index.ts
+19 -1
View File
@@ -29,11 +29,12 @@ export { ScratchlistStore } from './scratchlistStore'
export { SessionStore } from './sessionStore'
export { UserStore } from './userStore'
const SCHEMA_VERSION: number = 12
const SCHEMA_VERSION: number = 13
const REQUIRED_TABLES = [
'sessions',
'machines',
'messages',
'message_epochs',
'users',
'push_subscriptions',
'fcm_devices',
@@ -138,6 +139,7 @@ export class Store {
9: () => this.migrateFromV9ToV10(),
10: () => this.migrateFromV10ToV11(),
11: () => this.migrateFromV11ToV12(),
12: () => this.migrateFromV12ToV13(),
})
if (currentVersion === 0) {
@@ -245,6 +247,12 @@ export class Store {
ON messages(scheduled_at)
WHERE scheduled_at IS NOT NULL AND invoked_at IS NULL;
CREATE TABLE IF NOT EXISTS message_epochs (
session_id TEXT PRIMARY KEY,
epoch INTEGER NOT NULL DEFAULT 0,
FOREIGN KEY (session_id) REFERENCES sessions(id) ON DELETE CASCADE
);
CREATE TABLE IF NOT EXISTS users (
id INTEGER PRIMARY KEY AUTOINCREMENT,
platform TEXT NOT NULL,
@@ -519,6 +527,16 @@ export class Store {
`)
}
private migrateFromV12ToV13(): void {
this.db.exec(`
CREATE TABLE IF NOT EXISTS message_epochs (
session_id TEXT PRIMARY KEY,
epoch INTEGER NOT NULL DEFAULT 0,
FOREIGN KEY (session_id) REFERENCES sessions(id) ON DELETE CASCADE
)
`)
}
private getSessionColumnNames(): Set<string> {
const rows = this.db.prepare('PRAGMA table_info(sessions)').all() as Array<{ name: string }>
return new Set(rows.map((row) => row.name))
hub/src/store/messageStore.ts
+28 -2
View File
@@ -10,6 +10,10 @@ import {
getFirstMessages,
getDeliverableMessagesAfter,
getMessagesByPosition,
getMessagesAfterPosition,
getNewestMessagePosition,
getMessageEpoch,
bumpMessageEpoch,
getLocalMessageStates,
getUninvokedLocalMessages,
getMatureScheduledMessages,
@@ -25,6 +29,7 @@ import {
type CancelQueuedMessageResult,
type LookupQueuedMessageResult,
type LocalMessageState,
type MessagePosition,
} from './messages'
export class MessageStore {
@@ -66,6 +71,27 @@ export class MessageStore {
return getMessagesByPosition(this.db, sessionId, limit, before)
}
getMessagesAfterPosition(
sessionId: string,
limit: number,
after: MessagePosition,
until?: MessagePosition
): StoredMessage[] {
return getMessagesAfterPosition(this.db, sessionId, limit, after, until)
}
getNewestMessagePosition(sessionId: string): MessagePosition | null {
return getNewestMessagePosition(this.db, sessionId)
}
getMessageEpoch(sessionId: string): number {
return getMessageEpoch(this.db, sessionId)
}
bumpMessageEpoch(sessionId: string): number {
return bumpMessageEpoch(this.db, sessionId)
}
getLocalMessageStates(sessionId: string, localIds: string[]): LocalMessageState[] {
return getLocalMessageStates(this.db, sessionId, localIds)
}
@@ -106,8 +132,8 @@ export class MessageStore {
return lookupQueuedMessage(this.db, sessionId, messageId)
}
deleteQueuedMessageById(sessionId: string, messageId: string): void {
deleteQueuedMessageById(this.db, sessionId, messageId)
deleteQueuedMessageById(sessionId: string, messageId: string): boolean {
return deleteQueuedMessageById(this.db, sessionId, messageId)
}
markMessagesInvoked(sessionId: string, localIds: string[], invokedAt: number): void {
hub/src/store/messages.test.ts
+71
View File
@@ -24,6 +24,7 @@ describe('cancelQueuedMessage', () => {
// Row should be gone from uninvoked list
const remaining = store.messages.getUninvokedLocalMessages(session.id)
expect(remaining).toHaveLength(0)
expect(store.messages.getMessageEpoch(session.id)).toBe(1)
})
it('already-invoked: returns status=invoked with full message row, row stays in DB', () => {
@@ -67,6 +68,7 @@ describe('cancelQueuedMessage', () => {
if (second.status === 'cancelled') {
expect(second.localId).toBeNull()
}
expect(store.messages.getMessageEpoch(session.id)).toBe(1)
})
it('non-existent messageId: returns status=cancelled with localId=null', () => {
@@ -173,6 +175,75 @@ describe('cancelQueuedMessage', () => {
})
})
describe('position pagination and structural epochs', () => {
it('returns rows strictly after a cursor and respects an inclusive snapshot head', () => {
const store = makeStore()
const session = makeSession(store, 'position-after')
const first = store.messages.addMessage(session.id, { text: 'first' })
const second = store.messages.addMessage(session.id, { text: 'second' })
const third = store.messages.addMessage(session.id, { text: 'third' })
store.messages.addMessage(session.id, { text: 'fourth' })
const rows = store.messages.getMessagesAfterPosition(
session.id,
10,
{ at: first.invokedAt ?? first.createdAt, seq: first.seq },
{ at: third.invokedAt ?? third.createdAt, seq: third.seq }
)
expect(rows.map((message) => message.id)).toEqual([second.id, third.id])
})
it('reports the newest composite position', () => {
const store = makeStore()
const session = makeSession(store, 'position-head')
const first = store.messages.addMessage(session.id, { text: 'first' })
const second = store.messages.addMessage(session.id, { text: 'second' })
expect(store.messages.getNewestMessagePosition(session.id)).toEqual({
at: second.invokedAt ?? second.createdAt,
seq: second.seq
})
expect(first.seq).toBeLessThan(second.seq)
})
it('bumps both epochs when session history is merged', () => {
const store = makeStore()
const source = makeSession(store, 'epoch-merge-source')
const target = makeSession(store, 'epoch-merge-target')
store.messages.addMessage(source.id, { text: 'source' })
store.messages.addMessage(target.id, { text: 'target' })
const result = store.messages.mergeSessionMessages(source.id, target.id)
expect(result.moved).toBe(1)
expect(store.messages.getMessageEpoch(source.id)).toBe(1)
expect(store.messages.getMessageEpoch(target.id)).toBe(1)
})
it('bumps the target epoch when a copied message lands behind the cached head', () => {
const store = makeStore()
const target = makeSession(store, 'epoch-copy-target')
const head = store.messages.addMessage(target.id, { text: 'head' })
const headPosition = {
at: head.invokedAt ?? head.createdAt,
seq: head.seq
}
const copied = store.messages.copyMessageToSession(target.id, {
content: { text: 'historical' },
createdAt: headPosition.at - 1_000,
localId: null,
invokedAt: headPosition.at - 1_000,
scheduledAt: null
})
expect(copied.seq).toBeGreaterThan(head.seq)
expect(store.messages.getMessagesAfterPosition(target.id, 10, headPosition)).toEqual([])
expect(store.messages.getMessageEpoch(target.id)).toBe(1)
})
})
describe('addMessage: scheduledAt invariants', () => {
it('rejects scheduledAt without a localId — would silently invoke immediately', () => {
const store = makeStore()
hub/src/store/messages.ts
+99 -7
View File
@@ -15,6 +15,11 @@ type DbMessageRow = {
scheduled_at: number | null
}
export type MessagePosition = {
at: number
seq: number
}
function toStoredMessage(row: DbMessageRow): StoredMessage {
return {
id: row.id,
@@ -144,6 +149,11 @@ export function copyMessageToSession(
if (!row) {
throw new Error('Failed to copy message into target session')
}
// Copies preserve the source display timestamp, so a new high-seq row can
// still land behind a Web client's cached composite tail cursor. Mark the
// target history as structurally changed so incremental readers reset.
bumpMessageEpoch(db, sessionId)
return toStoredMessage(row)
}
@@ -222,7 +232,7 @@ export function getMessagesByPosition(
db: Database,
sessionId: string,
limit: number,
before?: { at: number; seq: number }
before?: MessagePosition
): StoredMessage[] {
const safeLimit = Number.isFinite(limit) ? Math.max(1, Math.min(200, limit)) : 200
const beforeClause = before
@@ -245,6 +255,72 @@ export function getMessagesByPosition(
return rows.reverse().map(toStoredMessage)
}
/** Return messages strictly after a display-position cursor in ascending order.
* `until`, when supplied, is an inclusive fixed snapshot head so a catch-up
* loop does not chase messages appended while it is running. */
export function getMessagesAfterPosition(
db: Database,
sessionId: string,
limit: number,
after: MessagePosition,
until?: MessagePosition
): StoredMessage[] {
const safeLimit = Number.isFinite(limit) ? Math.max(1, Math.min(200, limit)) : 200
const untilClause = until
? `AND (
COALESCE(invoked_at, created_at) < @untilAt
OR (COALESCE(invoked_at, created_at) = @untilAt AND seq <= @untilSeq)
)`
: ''
const rows = db.prepare(`
SELECT *, COALESCE(invoked_at, created_at) AS position_at
FROM messages
WHERE session_id = @sessionId
AND (
COALESCE(invoked_at, created_at) > @afterAt
OR (COALESCE(invoked_at, created_at) = @afterAt AND seq > @afterSeq)
)
${untilClause}
ORDER BY position_at ASC, seq ASC
LIMIT @limit
`).all({
sessionId,
afterAt: after.at,
afterSeq: after.seq,
untilAt: until?.at ?? null,
untilSeq: until?.seq ?? null,
limit: safeLimit
}) as DbMessageRow[]
return rows.map(toStoredMessage)
}
export function getNewestMessagePosition(db: Database, sessionId: string): MessagePosition | null {
const row = db.prepare(`
SELECT COALESCE(invoked_at, created_at) AS position_at, seq
FROM messages
WHERE session_id = ?
ORDER BY position_at DESC, seq DESC
LIMIT 1
`).get(sessionId) as { position_at: number; seq: number } | undefined
return row ? { at: row.position_at, seq: row.seq } : null
}
export function getMessageEpoch(db: Database, sessionId: string): number {
const row = db.prepare(
'SELECT epoch FROM message_epochs WHERE session_id = ?'
).get(sessionId) as { epoch: number } | undefined
return row?.epoch ?? 0
}
export function bumpMessageEpoch(db: Database, sessionId: string): number {
db.prepare(`
INSERT INTO message_epochs (session_id, epoch)
VALUES (?, 1)
ON CONFLICT(session_id) DO UPDATE SET epoch = epoch + 1
`).run(sessionId)
return getMessageEpoch(db, sessionId)
}
/** Returns user messages that have a localId but no invoked_at.
* Includes future scheduled messages — used to surface all queued messages
* (including scheduled) for the Web floating bar on refresh / secondary clients. */
@@ -471,11 +547,15 @@ export function cancelQueuedMessage(
return { status: 'invoked' as const, message: toStoredMessage(row) }
}
db.prepare(`
const deleted = db.prepare(`
DELETE FROM messages
WHERE session_id = ? AND (id = ? OR local_id = ?) AND invoked_at IS NULL
`).run(sessionId, messageId, messageId)
if (deleted.changes > 0) {
bumpMessageEpoch(db, sessionId)
}
return { status: 'cancelled' as const, localId: row.local_id }
})()
}
@@ -525,11 +605,18 @@ export function deleteQueuedMessageById(
db: Database,
sessionId: string,
messageId: string
): void {
db.prepare(`
DELETE FROM messages
WHERE session_id = ? AND (id = ? OR local_id = ?) AND invoked_at IS NULL
`).run(sessionId, messageId, messageId)
): boolean {
return db.transaction(() => {
const deleted = db.prepare(`
DELETE FROM messages
WHERE session_id = ? AND (id = ? OR local_id = ?) AND invoked_at IS NULL
`).run(sessionId, messageId, messageId)
if (deleted.changes > 0) {
bumpMessageEpoch(db, sessionId)
return true
}
return false
})()
}
/** Mark messages as invoked at the given server timestamp.
@@ -602,6 +689,11 @@ export function mergeSessionMessages(
'UPDATE messages SET session_id = ? WHERE session_id = ?'
).run(toSessionId, fromSessionId)
if (result.changes > 0) {
bumpMessageEpoch(db, fromSessionId)
bumpMessageEpoch(db, toSessionId)
}
db.exec('COMMIT')
return { moved: result.changes, oldMaxSeq, newMaxSeq }
} catch (error) {
hub/src/store/migration-v12.test.ts
+3 -4
View File
@@ -32,7 +32,7 @@ describe('Store V11→V12 migration: session_scratchlist table', () => {
expect(rows).toHaveLength(1)
})
it('V11 DB migrates to V12 via Store: session_scratchlist created', () => {
it('V11 DB migrates through V13 via Store: session_scratchlist created', () => {
const dir = mkdtempSync(join(tmpdir(), 'hapi-migration-v12-test-'))
const dbPath = join(dir, 'test.db')
let store: Store | undefined
@@ -61,7 +61,7 @@ describe('Store V11→V12 migration: session_scratchlist table', () => {
}
})
it('V9 DB migrates to V12 (multi-hop service_tier + fcm_devices + scratchlist)', () => {
it('V9 DB migrates through V13 (multi-hop service_tier + fcm_devices + scratchlist)', () => {
const dir = mkdtempSync(join(tmpdir(), 'hapi-migration-v9-to-v12-'))
const dbPath = join(dir, 'test.db')
let store: Store | undefined
@@ -84,7 +84,7 @@ describe('Store V11→V12 migration: session_scratchlist table', () => {
}
})
it('V12 DB reopen is idempotent: schema unchanged', () => {
it('current DB reopen is idempotent: schema unchanged', () => {
const dir = mkdtempSync(join(tmpdir(), 'hapi-migration-v12-idempotent-'))
const dbPath = join(dir, 'test.db')
let store1: Store | undefined
@@ -334,4 +334,3 @@ function createV11Schema(db: Database): void {
CREATE INDEX IF NOT EXISTS idx_fcm_devices_token ON fcm_devices(token);
`)
}
hub/src/store/migration-v13.test.ts
+137
View File
@@ -0,0 +1,137 @@
import { describe, expect, it } from 'bun:test'
import { Database } from 'bun:sqlite'
import { mkdtempSync, rmSync } from 'node:fs'
import { join } from 'node:path'
import { tmpdir } from 'node:os'
import { Store } from './index'
describe('Store V12→V13 migration: message_epochs', () => {
it('fresh DB has message_epochs table', () => {
const store = new Store(':memory:')
expect(tableExists(store, 'message_epochs')).toBe(true)
store.close()
})
it('V12 DB migrates to V13 and preserves existing messages', () => {
const dir = mkdtempSync(join(tmpdir(), 'hapi-migration-v13-test-'))
const dbPath = join(dir, 'test.db')
let store: Store | undefined
try {
const db = new Database(dbPath, { create: true, readwrite: true, strict: true })
db.exec('PRAGMA journal_mode = WAL')
db.exec('PRAGMA foreign_keys = ON')
createV12Schema(db)
db.exec(`
INSERT INTO sessions (id, created_at, updated_at) VALUES ('session-1', 1, 1);
INSERT INTO messages (id, session_id, content, created_at, seq, invoked_at)
VALUES ('message-1', 'session-1', '{}', 1, 1, 1);
PRAGMA user_version = 12;
`)
db.close()
store = new Store(dbPath)
expect(tableExists(store, 'message_epochs')).toBe(true)
expect(store.messages.getMessageEpoch('session-1')).toBe(0)
expect(store.messages.getMessages('session-1')).toHaveLength(1)
} finally {
store?.close()
rmSync(dir, { recursive: true, force: true })
}
})
})
function tableExists(store: Store, name: string): boolean {
const db: Database = (store as unknown as { db: Database }).db
const row = db.prepare(
"SELECT name FROM sqlite_master WHERE type='table' AND name=?"
).get(name) as { name: string } | null
return row !== null
}
function createV12Schema(db: Database): void {
db.exec(`
CREATE TABLE sessions (
id TEXT PRIMARY KEY,
tag TEXT,
namespace TEXT NOT NULL DEFAULT 'default',
machine_id TEXT,
created_at INTEGER NOT NULL,
updated_at INTEGER NOT NULL,
metadata TEXT,
metadata_version INTEGER DEFAULT 1,
agent_state TEXT,
agent_state_version INTEGER DEFAULT 1,
model TEXT,
model_reasoning_effort TEXT,
effort TEXT,
service_tier TEXT,
todos TEXT,
todos_updated_at INTEGER,
team_state TEXT,
team_state_updated_at INTEGER,
active INTEGER DEFAULT 0,
active_at INTEGER,
seq INTEGER DEFAULT 0
);
CREATE TABLE machines (
id TEXT PRIMARY KEY,
namespace TEXT NOT NULL DEFAULT 'default',
created_at INTEGER NOT NULL,
updated_at INTEGER NOT NULL,
metadata TEXT,
metadata_version INTEGER DEFAULT 1,
runner_state TEXT,
runner_state_version INTEGER DEFAULT 1,
active INTEGER DEFAULT 0,
active_at INTEGER,
seq INTEGER DEFAULT 0
);
CREATE TABLE messages (
id TEXT PRIMARY KEY,
session_id TEXT NOT NULL,
content TEXT NOT NULL,
created_at INTEGER NOT NULL,
seq INTEGER NOT NULL,
local_id TEXT,
invoked_at INTEGER,
scheduled_at INTEGER,
FOREIGN KEY (session_id) REFERENCES sessions(id) ON DELETE CASCADE
);
CREATE TABLE users (
id INTEGER PRIMARY KEY AUTOINCREMENT,
platform TEXT NOT NULL,
platform_user_id TEXT NOT NULL,
namespace TEXT NOT NULL DEFAULT 'default',
created_at INTEGER NOT NULL,
UNIQUE(platform, platform_user_id)
);
CREATE TABLE push_subscriptions (
id INTEGER PRIMARY KEY AUTOINCREMENT,
namespace TEXT NOT NULL,
endpoint TEXT NOT NULL,
p256dh TEXT NOT NULL,
auth TEXT NOT NULL,
created_at INTEGER NOT NULL,
UNIQUE(namespace, endpoint)
);
CREATE TABLE fcm_devices (
id INTEGER PRIMARY KEY AUTOINCREMENT,
namespace TEXT NOT NULL,
token TEXT NOT NULL,
platform TEXT NOT NULL,
device_id TEXT NOT NULL,
created_at INTEGER NOT NULL,
updated_at INTEGER NOT NULL,
UNIQUE(namespace, device_id, platform)
);
CREATE TABLE session_scratchlist (
session_id TEXT NOT NULL,
entry_id TEXT NOT NULL,
text TEXT NOT NULL,
created_at INTEGER NOT NULL,
updated_at INTEGER NOT NULL,
PRIMARY KEY (session_id, entry_id),
FOREIGN KEY (session_id) REFERENCES sessions(id) ON DELETE CASCADE
);
`)
}
hub/src/sync/messageService.test.ts
+97
View File
@@ -243,6 +243,11 @@ describe('MessageService message pagination', () => {
expect(page.messages.map((message) => message.id)).toEqual([second.id, third.id])
expect(page.page.nextBeforeAt).toBe(2_000)
expect(page.page.nextBeforeSeq).toBe(second.seq)
expect(page.page.direction).toBe('latest')
expect(page.page.epoch).toBe(0)
expect(page.page.reset).toBe(false)
expect(page.page.snapshotHeadAt).toBe(3_000)
expect(page.page.snapshotHeadSeq).toBe(third.seq)
expect(page.page.hasMore).toBe(true)
expect(first.id).toBeDefined()
})
@@ -266,6 +271,7 @@ describe('MessageService message pagination', () => {
expect(older.messages.map((message) => message.id)).toEqual([first.id])
expect(older.page.nextBeforeAt).toBe(1_000)
expect(older.page.nextBeforeSeq).toBe(first.seq)
expect(older.page.direction).toBe('before')
expect(older.page.hasMore).toBe(false)
expect(second.id).toBeDefined()
expect(third.id).toBeDefined()
@@ -305,6 +311,97 @@ describe('MessageService message pagination', () => {
expect(page.page.nextBeforeSeq).toBe(invoked.seq)
expect(page.page.hasMore).toBe(true)
})
it('pages forward to a fixed snapshot head', () => {
const store = makeStore()
const session = makeSession(store, 'page-after')
const first = store.messages.addMessage(session.id, 'first', 'local-first')
const second = store.messages.addMessage(session.id, 'second', 'local-second')
const third = store.messages.addMessage(session.id, 'third', 'local-third')
const fourth = store.messages.addMessage(session.id, 'fourth', 'local-fourth')
store.messages.markMessagesInvoked(session.id, ['local-first'], 1_000)
store.messages.markMessagesInvoked(session.id, ['local-second'], 2_000)
store.messages.markMessagesInvoked(session.id, ['local-third'], 3_000)
store.messages.markMessagesInvoked(session.id, ['local-fourth'], 4_000)
const service = makeService(store)
const firstDelta = service.getMessagesPage(session.id, {
limit: 1,
after: { at: 1_000, seq: first.seq },
epoch: 0
})
expect(firstDelta.messages.map((message) => message.id)).toEqual([second.id])
expect(firstDelta.page).toMatchObject({
direction: 'after',
nextAfterAt: 2_000,
nextAfterSeq: second.seq,
snapshotHeadAt: 4_000,
snapshotHeadSeq: fourth.seq,
hasMore: true,
reset: false
})
const fifth = store.messages.addMessage(session.id, 'fifth', 'local-fifth')
store.messages.markMessagesInvoked(session.id, ['local-fifth'], 5_000)
const secondDelta = service.getMessagesPage(session.id, {
limit: 10,
after: { at: firstDelta.page.nextAfterAt!, seq: firstDelta.page.nextAfterSeq! },
until: { at: firstDelta.page.snapshotHeadAt!, seq: firstDelta.page.snapshotHeadSeq! },
epoch: firstDelta.page.epoch
})
expect(secondDelta.messages.map((message) => message.id)).toEqual([third.id, fourth.id])
expect(secondDelta.page.hasMore).toBe(false)
expect(secondDelta.messages.some((message) => message.id === fifth.id)).toBe(false)
})
it('completes a fixed snapshot when no raw row remains before its head', () => {
const store = makeStore()
const session = makeSession(store, 'page-after-empty-snapshot-gap')
const first = store.messages.addMessage(session.id, 'first', 'local-first')
const latest = store.messages.addMessage(session.id, 'latest', 'local-latest')
store.messages.markMessagesInvoked(session.id, ['local-first'], 1_000)
store.messages.markMessagesInvoked(session.id, ['local-latest'], 4_000)
const response = makeService(store).getMessagesPage(session.id, {
limit: 10,
after: { at: 2_000, seq: first.seq },
until: { at: 3_000, seq: latest.seq },
epoch: 0
})
expect(response.messages).toEqual([])
expect(response.page).toMatchObject({
direction: 'after',
nextAfterAt: 3_000,
nextAfterSeq: latest.seq,
snapshotHeadAt: 3_000,
snapshotHeadSeq: latest.seq,
hasMore: false
})
})
it('returns a reset latest page when the structural epoch changed', () => {
const store = makeStore()
const session = makeSession(store, 'page-after-reset')
const first = store.messages.addMessage(session.id, 'first', 'local-first')
store.messages.markMessagesInvoked(session.id, ['local-first'], 1_000)
const queued = store.messages.addMessage(session.id, 'queued', 'local-queued')
store.messages.cancelQueuedMessage(session.id, queued.id)
const response = makeService(store).getMessagesPage(session.id, {
limit: 10,
after: { at: 1_000, seq: first.seq },
epoch: 0
})
expect(response.page.direction).toBe('latest')
expect(response.page.reset).toBe(true)
expect(response.page.epoch).toBe(1)
expect(response.messages.map((message) => message.id)).toEqual([first.id])
})
})
describe('MessageService.getQueuedState', () => {
hub/src/sync/messageService.ts
+121 -14
View File
@@ -10,13 +10,25 @@ import {
unwrapRoleWrappedRecordEnvelope
} from '@hapi/protocol/messages'
import { isObject } from '@hapi/protocol'
import type { QueuedStateResponse } from '@hapi/protocol/apiTypes'
import type { MessagesResponse, QueuedStateResponse } from '@hapi/protocol/apiTypes'
import type { Server } from 'socket.io'
import { randomUUID } from 'node:crypto'
import type { Store, CancelQueuedMessageResult } from '../store'
import { EventPublisher } from './eventPublisher'
type StoredMessageForDelivery = ReturnType<Store['messages']['getMessages']>[number]
type MessagePosition = { at: number; seq: number }
function messagePosition(message: StoredMessageForDelivery): MessagePosition {
return {
at: message.invokedAt ?? message.createdAt,
seq: message.seq
}
}
function comparePosition(a: MessagePosition, b: MessagePosition): number {
return a.at !== b.at ? a.at - b.at : a.seq - b.seq
}
function isWebVisibleStoredMessage(message: StoredMessageForDelivery): boolean {
return !isRedundantGoalStatusEventContent(message.content)
@@ -139,25 +151,54 @@ export class MessageService {
getMessagesPage(
sessionId: string,
options: { limit: number; before?: { at: number; seq: number } | null }
): {
messages: DecryptedMessage[]
page: {
options: {
limit: number
nextBeforeSeq: number | null
nextBeforeAt: number | null
hasMore: boolean
before?: MessagePosition | null
after?: MessagePosition | null
until?: MessagePosition | null
epoch?: number | null
}
} {
let before = options.before ?? undefined
let pageRows = this.store.messages.getMessagesByPosition(sessionId, options.limit, before)
): MessagesResponse {
const epoch = this.store.messages.getMessageEpoch(sessionId)
if (options.after) {
if (options.epoch !== undefined && options.epoch !== null && options.epoch !== epoch) {
return this.getLatestOrBeforeMessagesPage(sessionId, options.limit, null, epoch, true)
}
return this.getAfterMessagesPage(
sessionId,
options.limit,
options.after,
options.until ?? null,
epoch
)
}
return this.getLatestOrBeforeMessagesPage(
sessionId,
options.limit,
options.before ?? null,
epoch,
false
)
}
private getLatestOrBeforeMessagesPage(
sessionId: string,
limit: number,
requestedBefore: MessagePosition | null,
epoch: number,
reset: boolean
): MessagesResponse {
const direction = requestedBefore ? 'before' as const : 'latest' as const
const snapshotHead = this.store.messages.getNewestMessagePosition(sessionId)
let before = requestedBefore ?? undefined
let pageRows = this.store.messages.getMessagesByPosition(sessionId, limit, requestedBefore ?? undefined)
// Latest-page request (no cursor): also include uninvoked local user messages
// out-of-band, so refresh / secondary clients can still see queued rows even
// when their position key (createdAt) places them outside the latest page.
// The cursor stays anchored to pageRows so out-of-band rows don't affect
// pagination of older pages.
let queuedRows = before === undefined
let queuedRows = requestedBefore === null
? this.store.messages.getUninvokedLocalMessages(sessionId)
: []
@@ -190,7 +231,7 @@ export class MessageService {
while (messages.length === 0 && hasMore && oldestSeq !== null && oldestPositionAt !== null) {
before = { at: oldestPositionAt, seq: oldestSeq }
pageRows = this.store.messages.getMessagesByPosition(sessionId, options.limit, before)
pageRows = this.store.messages.getMessagesByPosition(sessionId, limit, before)
queuedRows = []
byId = new Map<string, typeof pageRows[number]>()
@@ -219,9 +260,75 @@ export class MessageService {
return {
messages,
page: {
limit: options.limit,
direction,
limit,
epoch,
reset,
nextBeforeSeq: oldestSeq,
nextBeforeAt: oldestPositionAt,
nextAfterSeq: null,
nextAfterAt: null,
snapshotHeadSeq: snapshotHead?.seq ?? null,
snapshotHeadAt: snapshotHead?.at ?? null,
hasMore
}
}
}
private getAfterMessagesPage(
sessionId: string,
limit: number,
after: MessagePosition,
requestedUntil: MessagePosition | null,
epoch: number
): MessagesResponse {
const currentHead = this.store.messages.getNewestMessagePosition(sessionId)
const snapshotHead = currentHead && requestedUntil
? (comparePosition(requestedUntil, currentHead) <= 0 ? requestedUntil : currentHead)
: requestedUntil ?? currentHead
if (!snapshotHead || comparePosition(snapshotHead, after) <= 0) {
return {
messages: [],
page: {
direction: 'after',
limit,
epoch,
reset: false,
nextBeforeSeq: null,
nextBeforeAt: null,
nextAfterSeq: after.seq,
nextAfterAt: after.at,
snapshotHeadSeq: snapshotHead?.seq ?? null,
snapshotHeadAt: snapshotHead?.at ?? null,
hasMore: false
}
}
}
const pageRows = this.store.messages.getMessagesAfterPosition(
sessionId,
limit,
after,
snapshotHead
)
const last = pageRows[pageRows.length - 1] ?? null
const nextAfter = last ? messagePosition(last) : snapshotHead
const hasMore = last !== null && comparePosition(nextAfter, snapshotHead) < 0
return {
messages: toVisibleDecryptedMessages(pageRows),
page: {
direction: 'after',
limit,
epoch,
reset: false,
nextBeforeSeq: null,
nextBeforeAt: null,
nextAfterSeq: nextAfter.seq,
nextAfterAt: nextAfter.at,
snapshotHeadSeq: snapshotHead.seq,
snapshotHeadAt: snapshotHead.at,
hasMore
}
}
hub/src/sync/syncEngine.ts
+7 -9
View File
@@ -8,7 +8,7 @@
*/
import { isKnownFlavor, type LocalResumeTarget, type ResumableSession } from '@hapi/protocol'
import type { CursorChatStoreStatus, CursorMigrateOutcome, CursorMigrateToAcpRequest, QueuedStateResponse, SlashCommandsResponse } from '@hapi/protocol/apiTypes'
import type { CursorChatStoreStatus, CursorMigrateOutcome, CursorMigrateToAcpRequest, MessagesResponse, QueuedStateResponse, SlashCommandsResponse } from '@hapi/protocol/apiTypes'
import type { AgentFlavor, CodexCollaborationMode, DecryptedMessage, PermissionMode, Session, SyncEvent } from '@hapi/protocol/types'
import { unwrapRoleWrappedRecordEnvelope } from '@hapi/protocol/messages'
import type { Server } from 'socket.io'
@@ -323,16 +323,14 @@ export class SyncEngine {
getMessagesPage(
sessionId: string,
options: { limit: number; before?: { at: number; seq: number } | null }
): {
messages: DecryptedMessage[]
page: {
options: {
limit: number
nextBeforeSeq: number | null
nextBeforeAt: number | null
hasMore: boolean
before?: { at: number; seq: number } | null
after?: { at: number; seq: number } | null
until?: { at: number; seq: number } | null
epoch?: number | null
}
} {
): MessagesResponse {
return this.messageService.getMessagesPage(sessionId, options)
}
hub/src/web/routes/messages.test.ts
+130 -1
View File
@@ -11,6 +11,8 @@ import type { SyncEngine } from '../../sync/syncEngine'
import type { WebAppEnv } from '../middleware/auth'
import { createMessagesRoutes } from './messages'
type GetMessagesPage = SyncEngine['getMessagesPage']
// TS note: engine is cast to unknown→SyncEngine so test helpers don't need to
// satisfy the full SyncEngine shape (only the subset the route under test uses).
@@ -21,6 +23,7 @@ import { createMessagesRoutes } from './messages'
function createApp(opts: {
active?: boolean
sendMessage?: (sessionId: string, payload: unknown) => Promise<void>
getMessagesPage?: GetMessagesPage
getQueuedState?: (sessionId: string, localIds: string[]) => {
queuedLocalIds: string[]
invokedLocalMessages: Array<{ localId: string; invokedAt: number }>
@@ -40,6 +43,22 @@ function createApp(opts: {
.map((localId) => ({ localId, invokedAt: 1_000 }))
}
})
const getMessagesPage = opts.getMessagesPage ?? (() => ({
messages: [],
page: {
direction: 'latest',
limit: 50,
epoch: 0,
reset: false,
nextBeforeSeq: null,
nextBeforeAt: null,
nextAfterSeq: null,
nextAfterAt: null,
snapshotHeadSeq: null,
snapshotHeadAt: null,
hasMore: false
}
}))
const engine = {
resolveSessionAccess: () => ({
@@ -50,7 +69,7 @@ function createApp(opts: {
sendMessage,
getQueuedState,
cancelQueuedMessage: async () => ({ status: 'cancelled' }),
getMessagesPage: () => ({ messages: [], page: {} }),
getMessagesPage,
} as unknown as SyncEngine
const app = new Hono<WebAppEnv>()
@@ -63,6 +82,116 @@ function createApp(opts: {
return { app, sentMessages, queuedStateCalls }
}
describe('GET /api/sessions/:id/messages', () => {
it('uses latest mode by default and returns the full page metadata', async () => {
const calls: Array<{ sessionId: string; options: Parameters<GetMessagesPage>[1] }> = []
const { app } = createApp({
getMessagesPage: (sessionId, options) => {
calls.push({ sessionId, options })
return {
messages: [],
page: {
direction: 'latest',
limit: options.limit,
epoch: 4,
reset: false,
nextBeforeSeq: 10,
nextBeforeAt: 1_000,
nextAfterSeq: null,
nextAfterAt: null,
snapshotHeadSeq: 20,
snapshotHeadAt: 2_000,
hasMore: true
}
}
}
})
const response = await app.request('/api/sessions/session-1/messages')
expect(response.status).toBe(200)
expect(calls).toEqual([{
sessionId: 'session-1',
options: { limit: 50, before: null, after: null, until: null, epoch: null }
}])
expect(await response.json()).toEqual({
messages: [],
page: {
direction: 'latest',
limit: 50,
epoch: 4,
reset: false,
nextBeforeSeq: 10,
nextBeforeAt: 1_000,
nextAfterSeq: null,
nextAfterAt: null,
snapshotHeadSeq: 20,
snapshotHeadAt: 2_000,
hasMore: true
}
})
})
it('forwards after, snapshot-head, epoch, and limit query parameters', async () => {
const calls: Array<{ sessionId: string; options: Parameters<GetMessagesPage>[1] }> = []
const { app } = createApp({
getMessagesPage: (sessionId, options) => {
calls.push({ sessionId, options })
return {
messages: [],
page: {
direction: 'after',
limit: options.limit,
epoch: options.epoch ?? 0,
reset: false,
nextBeforeSeq: null,
nextBeforeAt: null,
nextAfterSeq: 11,
nextAfterAt: 1_100,
snapshotHeadSeq: 20,
snapshotHeadAt: 2_000,
hasMore: true
}
}
}
})
const response = await app.request(
'/api/sessions/session-1/messages?afterAt=1000&afterSeq=10&untilAt=2000&untilSeq=20&epoch=3&limit=25'
)
expect(response.status).toBe(200)
expect(calls).toEqual([{
sessionId: 'session-1',
options: {
limit: 25,
before: null,
after: { at: 1_000, seq: 10 },
until: { at: 2_000, seq: 20 },
epoch: 3
}
}])
})
it('rejects mixed directional cursors before calling the engine', async () => {
let called = false
const { app } = createApp({
getMessagesPage: () => {
called = true
throw new Error('must not be called')
}
})
const response = await app.request(
'/api/sessions/session-1/messages?beforeAt=1000&beforeSeq=10&afterAt=2000&afterSeq=20'
)
expect(response.status).toBe(400)
expect(await response.json()).toMatchObject({ error: 'Invalid query' })
expect(called).toBe(false)
})
})
// ---------------------------------------------------------------------------
// #2 server-side scheduledAt upper bound
// ---------------------------------------------------------------------------
hub/src/web/routes/messages.ts
+13 -1
View File
@@ -28,7 +28,19 @@ export function createMessagesRoutes(getSyncEngine: () => SyncEngine | null): Ho
const before = parsed.data.beforeAt !== undefined && parsed.data.beforeSeq !== undefined
? { at: parsed.data.beforeAt, seq: parsed.data.beforeSeq }
: null
return c.json(engine.getMessagesPage(sessionId, { limit, before }))
const after = parsed.data.afterAt !== undefined && parsed.data.afterSeq !== undefined
? { at: parsed.data.afterAt, seq: parsed.data.afterSeq }
: null
const until = parsed.data.untilAt !== undefined && parsed.data.untilSeq !== undefined
? { at: parsed.data.untilAt, seq: parsed.data.untilSeq }
: null
return c.json(engine.getMessagesPage(sessionId, {
limit,
before,
after,
until,
epoch: parsed.data.epoch ?? null
}))
})
app.delete('/sessions/:id/messages/:messageId', async (c) => {