From 35c260ac3338d8850d4b9f40df72c4f0be042994 Mon Sep 17 00:00:00 2001 From: digua Date: Mon, 22 Jun 2026 22:23:18 +0800 Subject: [PATCH] =?UTF-8?q?fix(sync):=20repair=20pull=20pagination=20curso?= =?UTF-8?q?r=EF=BC=88#237=EF=BC=89?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- apps/desktop/tsconfig.node.json | 1 + packages/sync/src/pull-engine.test.ts | 76 ++++++++++++++++++++++++++- packages/sync/src/pull-engine.ts | 69 +++++++++++++++++++++++- 3 files changed, 143 insertions(+), 3 deletions(-) diff --git a/apps/desktop/tsconfig.node.json b/apps/desktop/tsconfig.node.json index 666628ce..00e41749 100644 --- a/apps/desktop/tsconfig.node.json +++ b/apps/desktop/tsconfig.node.json @@ -17,6 +17,7 @@ "../../packages/sync/**/*", "../../packages/http-routes/**/*" ], + "exclude": ["**/node_modules/**", "**/dist/**", "../../**/*.test.ts", "../../**/*.spec.ts", "../../tests/**"], "compilerOptions": { "composite": true, "types": ["electron-vite/node"], diff --git a/packages/sync/src/pull-engine.test.ts b/packages/sync/src/pull-engine.test.ts index c5f591a3..0f7671fa 100644 --- a/packages/sync/src/pull-engine.test.ts +++ b/packages/sync/src/pull-engine.test.ts @@ -69,6 +69,7 @@ function createEngine(options: { files: string[] importResult: Awaited> dataSource: DataSource + fetchParams?: FetchParams[] sessionUpdates?: Array<{ sessionId: string; updates: Partial }> pullResults?: Array<{ status: 'success' | 'error'; detail: string }> }): PullEngine { @@ -78,8 +79,9 @@ function createEngine(options: { _baseUrl: string, _remoteSessionId: string, _token: string, - _params: FetchParams + params: FetchParams ): Promise { + options.fetchParams?.push({ ...params }) const file = files.shift() if (!file) throw new Error('Unexpected retry fetch') return file @@ -218,4 +220,76 @@ describe('PullEngine', () => { assert.equal(pullResults.at(-1)?.status, 'error') assert.equal(sessionUpdates.at(-1)?.updates.lastStatus, 'error') }) + + it('continues pagination with nextOffset when nextSince is absent', async () => { + const session = createSession() + session.lastPullAt = 0 + const dataSource = createDataSource() + dataSource.sessions = [session] + const firstPage = writeTempJson({ + chatlab: { version: '0.0.2', exportedAt: 100 }, + meta: { name: 'Test Session', platform: 'test', type: 'group' }, + members: [{ platformId: 'u1', accountName: 'Alice' }], + messages: [ + { sender: 'u1', timestamp: 100, type: 0, content: 'page 1a' }, + { sender: 'u1', timestamp: 101, type: 0, content: 'page 1b' }, + ], + sync: { hasMore: true, nextOffset: 2, watermark: 200 }, + }) + const secondPage = writeTempJson({ + chatlab: { version: '0.0.2', exportedAt: 100 }, + meta: { name: 'Test Session', platform: 'test', type: 'group' }, + members: [{ platformId: 'u1', accountName: 'Alice' }], + messages: [{ sender: 'u1', timestamp: 200, type: 0, content: 'page 2' }], + sync: { hasMore: false, watermark: 200 }, + }) + const fetchParams: FetchParams[] = [] + const engine = createEngine({ + files: [firstPage, secondPage], + dataSource, + fetchParams, + importResult: { + success: true, + newMessageCount: 1, + sessionId: session.targetSessionId, + }, + }) + + const result = await withImmediateTimers(() => engine.executePullSession(dataSource.id, dataSource, session)) + + assert.equal(result.success, true) + assert.equal(fetchParams.length, 2) + assert.equal(fetchParams[0]?.offset, undefined) + assert.equal(fetchParams[1]?.offset, 2) + }) + + it('persists the imported message cursor with overlap instead of the wall clock', async () => { + const session = createSession() + session.lastPullAt = 0 + const dataSource = createDataSource() + dataSource.sessions = [session] + const page = writeTempJson({ + chatlab: { version: '0.0.2', exportedAt: 100 }, + meta: { name: 'Test Session', platform: 'test', type: 'group' }, + members: [{ platformId: 'u1', accountName: 'Alice' }], + messages: [{ sender: 'u1', timestamp: 2000, type: 0, content: 'latest imported message' }], + sync: { hasMore: false, watermark: 999999 }, + }) + const sessionUpdates: Array<{ sessionId: string; updates: Partial }> = [] + const engine = createEngine({ + files: [page], + dataSource, + sessionUpdates, + importResult: { + success: true, + newMessageCount: 1, + sessionId: session.targetSessionId, + }, + }) + + const result = await withImmediateTimers(() => engine.executePullSession(dataSource.id, dataSource, session)) + + assert.equal(result.success, true) + assert.equal(sessionUpdates.at(-1)?.updates.lastPullAt, 1940) + }) }) diff --git a/packages/sync/src/pull-engine.ts b/packages/sync/src/pull-engine.ts index e92a7217..bd2a5f1e 100644 --- a/packages/sync/src/pull-engine.ts +++ b/packages/sync/src/pull-engine.ts @@ -102,6 +102,44 @@ function fileContainsMessages(filePath: string): boolean { } } +function getMaxMessageTimestampFromFile(filePath: string): number | null { + try { + let maxTs: number | null = null + const visitTimestamp = (value: unknown) => { + const ts = typeof value === 'string' && value.trim() !== '' ? Number(value) : value + if (typeof ts === 'number' && Number.isFinite(ts)) { + maxTs = maxTs === null ? ts : Math.max(maxTs, ts) + } + } + + if (filePath.endsWith('.jsonl')) { + const content = fs.readFileSync(filePath, 'utf-8') + for (const line of content.split('\n')) { + const trimmed = line.trim() + if (!trimmed) continue + try { + const obj = JSON.parse(trimmed) + if (obj._type === 'message') visitTimestamp(obj.timestamp) + } catch { + continue + } + } + return maxTs + } + + const raw = fs.readFileSync(filePath, 'utf-8') + const parsed = JSON.parse(raw) + if (Array.isArray(parsed.messages)) { + for (const message of parsed.messages) { + visitTimestamp(message?.timestamp) + } + } + return maxTs + } catch { + return null + } +} + function cleanupTempFile(filePath: string): void { try { if (fs.existsSync(filePath)) fs.unlinkSync(filePath) @@ -193,6 +231,8 @@ export class PullEngine { let totalNewMessages = 0 let since = sess.lastPullAt + let offset: number | undefined + let nextPullSince = sess.lastPullAt let pageCount = 0 let resyncAttempted = false @@ -203,6 +243,7 @@ export class PullEngine { pageCount++ const tempFile = await this.fetcher.fetchToTempFile(ds.baseUrl, sess.remoteSessionId, ds.token, { since, + offset, limit: ds.pullLimit, }) @@ -224,6 +265,7 @@ export class PullEngine { await new Promise((r) => setTimeout(r, retryDelays[ri])) const retryFile = await this.fetcher.fetchToTempFile(ds.baseUrl, sess.remoteSessionId, ds.token, { since, + offset, limit: ds.pullLimit, }) const retryStat = fs.statSync(retryFile) @@ -233,6 +275,7 @@ export class PullEngine { cleanupTempFile(retryFile) continue } + const retryMaxTs = getMaxMessageTimestampFromFile(retryFile) const retryResult = await this.importTempFile(ds.baseUrl, sess, retryFile) cleanupTempFile(retryFile) @@ -240,6 +283,8 @@ export class PullEngine { resyncAttempted = true this.logger.info(`[Pull] Resetting since=0 for "${sess.name}" full resync`) since = 0 + offset = undefined + nextPullSince = 0 pageCount = 0 sess.targetSessionId = '' sess.lastPullAt = 0 @@ -279,6 +324,7 @@ export class PullEngine { this.dsManager.updateSession(sourceId, sess.id, { targetSessionId: retryResult.sessionId }) } totalNewMessages += retryResult.newMessageCount + if (retryMaxTs !== null) nextPullSince = Math.max(nextPullSince, retryMaxTs) this.progressMap.set(sess.id, { sessionId: sess.id, sessionName: sess.name, @@ -288,6 +334,10 @@ export class PullEngine { }) if (retrySync?.hasMore && retrySync.nextSince !== undefined) { since = retrySync.nextSince + offset = undefined + nextPullSince = Math.max(nextPullSince, retrySync.nextSince) + } else if (retrySync?.hasMore && retrySync.nextOffset !== undefined) { + offset = retrySync.nextOffset } retryHasMore = !!retrySync?.hasMore retrySuccess = true @@ -305,6 +355,7 @@ export class PullEngine { } const sync = sync0 + const maxMessageTs = getMaxMessageTimestampFromFile(tempFile) const result = await this.importTempFile(ds.baseUrl, sess, tempFile) cleanupTempFile(tempFile) @@ -312,6 +363,8 @@ export class PullEngine { resyncAttempted = true this.logger.info(`[Pull] Resetting since=0 for "${sess.name}" full resync`) since = 0 + offset = undefined + nextPullSince = 0 pageCount = 0 sess.targetSessionId = '' sess.lastPullAt = 0 @@ -350,6 +403,7 @@ export class PullEngine { } totalNewMessages += result.newMessageCount + if (maxMessageTs !== null) nextPullSince = Math.max(nextPullSince, maxMessageTs) this.progressMap.set(sess.id, { sessionId: sess.id, sessionName: sess.name, @@ -358,9 +412,20 @@ export class PullEngine { done: false, }) + if (sync?.nextSince !== undefined) nextPullSince = Math.max(nextPullSince, sync.nextSince) + if (!sync || !sync.hasMore) break - if (sync.nextSince !== undefined) since = sync.nextSince + // 游标推进优先使用时间戳链；旧数据源只返回 nextOffset 时，继续使用 offset 续拉同一个 since 窗口。 + if (sync.nextSince !== undefined) { + since = sync.nextSince + offset = undefined + } else if (sync.nextOffset !== undefined) { + offset = sync.nextOffset + } else { + this.logger.warn(`[Pull] "${sess.name}" returned hasMore=true without nextSince or nextOffset, stopping`) + break + } } catch (importErr) { cleanupTempFile(tempFile) throw importErr @@ -372,7 +437,7 @@ export class PullEngine { } this.dsManager.updateSession(sourceId, sess.id, { - lastPullAt: Math.floor(Date.now() / 1000) - PULL_OVERLAP_SECONDS, + lastPullAt: Math.max(0, nextPullSince - PULL_OVERLAP_SECONDS), lastStatus: 'success', lastNewMessages: totalNewMessages, lastError: '',