feat(merge): migrate merge routes from IPC to shared HTTP package

Unify the merge workflow (parse/conflicts/execute/clear/export-for-merge)
across CLI, Electron, and web-serve into a single set of shared routes in
@openchatlab/http-routes.

Key changes:
- Move MergeSessionCache from CLI to node-runtime shared package
- Add dual-mode parse route (multipart upload for web, JSON filePath for Electron)
- Integrate merge routes into Electron internal API server with @fastify/multipart
- Create installMergeShims() with filePath-to-handle mapping for frontend
- Remove Electron merge IPC handlers, merger module, and preload bridges (~790 lines deleted)
- Address CodeX review: CHAT_DB_TABLES in openDatabase, no chatApi patching,
  outputName sanitization, resp.ok checks, @fastify/multipart in desktop deps

Co-authored-by: Cursor <cursoragent@cursor.com>
This commit is contained in:
diguaandCursor committed 2026-06-02 15:41:21 +08:00
1 parent 65fb5179cd
commit ae90db63a5
22 files changed
+432 -790

No files matched your search

+12 -4
View File
@@ -34,7 +34,6 @@ import { createDatabaseManagerAdapter } from '@openchatlab/node-runtime'
import { registerSharedRoutes } from '@openchatlab/http-routes'
import { MergeSessionCache } from '../../../merger/merge-cache'
import { registerImportRoutes } from './import'
import { registerMergeRoutes } from './merge'
import { registerCacheRoutes } from './cache'
import { getVersion } from '../../../version'
@@ -95,6 +94,14 @@ export function registerWebRoutes(
const resolvedPathProvider = options?.pathProvider ?? fallbackPathProvider
const ai = options?.aiContext
const cliStreamImport = async (dm: typeof dbManager, filePath: string) => {
const { streamImport } = await import('../../../import/stream-import')
const result = await streamImport(dm, filePath)
if (!result.sessionId) throw new Error('Import succeeded but no sessionId returned')
return { sessionId: result.sessionId }
}
registerSharedRoutes(
server,
{
@@ -103,6 +110,10 @@ export function registerWebRoutes(
pathProvider: resolvedPathProvider,
getVersion,
nativeBinding: options?.nativeBinding,
...(mergeCache && {
mergeSessionCache: mergeCache,
streamImport: cliStreamImport,
}),
...(ai && {
aiDataDir: ai.aiDataDir,
conversationManager: ai.convManager,
@@ -118,9 +129,6 @@ export function registerWebRoutes(
// CLI-specific routes not yet migrated to @openchatlab/http-routes
registerImportRoutes(server, dbManager)
if (mergeCache) {
registerMergeRoutes(server, dbManager, mergeCache)
}
registerCacheRoutes(server, resolvedPathProvider)
server.get('/_web/system/check-update', async () => {
-216
View File
@@ -1,216 +0,0 @@
import * as fs from 'fs'
import * as os from 'os'
import * as path from 'path'
import type { FastifyInstance } from 'fastify'
import type { DatabaseManager } from '@openchatlab/node-runtime'
import {
streamParseFileInfo,
checkConflictsFromSources,
buildMergedOutput,
serializeChatLabToJsonl,
exportSessionToJson,
TempDbReader,
} from '@openchatlab/node-runtime'
import { MergeSessionCache } from '../../../merger/merge-cache'
import { ensureDb } from './helpers'
export function registerMergeRoutes(
server: FastifyInstance,
dbManager: DatabaseManager,
mergeCache: MergeSessionCache
): void {
server.post('/_web/merge/parse', async (request, reply) => {
const data = await (request as any).file()
if (!data) return reply.code(400).send({ error: 'No file uploaded' })
const tmpDir = fs.mkdtempSync(path.join(os.tmpdir(), 'chatlab-merge-'))
const tmpPath = path.join(tmpDir, data.filename || 'upload')
const chunks: Buffer[] = []
for await (const chunk of data.file) {
chunks.push(chunk)
}
fs.writeFileSync(tmpPath, Buffer.concat(chunks))
try {
const result = await streamParseFileInfo(tmpPath, {
createTempDatabase(sourceFilePath: string) {
return mergeCache.createTempDatabase(path.basename(sourceFilePath))
},
onProgress() {
/* no-op for HTTP merge parse */
},
})
const handle = mergeCache.store(data.filename || 'upload', result.tempDbPath)
return {
handle,
name: result.name,
format: result.format,
platform: result.platform,
messageCount: result.messageCount,
memberCount: result.memberCount,
fileSize: result.fileSize,
}
} finally {
try {
fs.unlinkSync(tmpPath)
} catch {
/* ignore */
}
try {
fs.rmdirSync(tmpDir)
} catch {
/* ignore */
}
}
})
server.post<{ Body: { handles: string[] } }>('/_web/merge/conflicts', async (request, reply) => {
const { handles } = request.body as { handles?: string[] }
if (!handles || !Array.isArray(handles) || handles.length === 0) {
return reply.code(400).send({ error: 'Missing or empty handles array' })
}
const readers: TempDbReader[] = []
try {
const dataSources: Array<{ source: import('@openchatlab/node-runtime').MergerDataSource; filename: string }> = []
for (const handle of handles) {
const entry = mergeCache.openReader(handle)
if (!entry) return reply.code(404).send({ error: `Handle not found: ${handle}` })
readers.push(entry.reader)
dataSources.push({ source: entry.reader.toDataSource(), filename: entry.filename })
}
return checkConflictsFromSources(dataSources)
} finally {
for (const r of readers) r.close()
}
})
server.post<{
Body: { handles: string[]; outputName: string; format?: 'json' | 'jsonl'; andImport?: boolean }
}>('/_web/merge/execute', async (request, reply) => {
const { handles, outputName, format = 'json', andImport } = request.body as any
if (!handles || !Array.isArray(handles) || handles.length === 0) {
return reply.code(400).send({ error: 'Missing handles' })
}
if (!outputName) {
return reply.code(400).send({ error: 'Missing outputName' })
}
const readers: TempDbReader[] = []
try {
const dataSources: Array<{ source: import('@openchatlab/node-runtime').MergerDataSource; filename: string }> = []
for (const handle of handles) {
const entry = mergeCache.openReader(handle)
if (!entry) return reply.code(404).send({ error: `Handle not found: ${handle}` })
readers.push(entry.reader)
dataSources.push({ source: entry.reader.toDataSource(), filename: entry.filename })
}
const merged = buildMergedOutput(dataSources, outputName)
let sessionId: string | undefined
if (andImport) {
const { streamImport: serverStreamImport } = await import('../../../import/stream-import')
const jsonData = JSON.stringify(merged.chatLabData)
const tmpDir = fs.mkdtempSync(path.join(os.tmpdir(), 'chatlab-merged-'))
const tmpPath = path.join(tmpDir, `${outputName}.json`)
fs.writeFileSync(tmpPath, jsonData, 'utf-8')
try {
const importResult = await serverStreamImport(dbManager, tmpPath)
sessionId = importResult.sessionId
} finally {
try {
fs.unlinkSync(tmpPath)
} catch {
/* ignore */
}
try {
fs.rmdirSync(tmpDir)
} catch {
/* ignore */
}
}
}
for (const handle of handles) {
mergeCache.delete(handle)
}
if (format === 'jsonl') {
const lines: string[] = []
for (const line of serializeChatLabToJsonl(merged.chatLabData)) {
lines.push(line)
}
return { success: true, sessionId, data: lines.join('\n') }
}
return { success: true, sessionId, data: merged.chatLabData }
} finally {
for (const r of readers) r.close()
}
})
server.post<{ Body: { handle?: string } }>('/_web/merge/clear', async (request) => {
const { handle } = (request.body as any) || {}
if (handle) {
mergeCache.delete(handle)
} else {
mergeCache.clear()
}
return { success: true }
})
server.post<{
Body: { sessionIds: string[] }
}>('/_web/sessions/export-for-merge', async (request, reply) => {
const { sessionIds } = request.body as { sessionIds?: string[] }
if (!sessionIds || !Array.isArray(sessionIds) || sessionIds.length === 0) {
return reply.code(400).send({ error: 'Missing sessionIds' })
}
const handles: Array<{ sessionId: string; handle: string }> = []
for (const sid of sessionIds) {
const db = ensureDb(dbManager, sid)
const exported = exportSessionToJson(db)
const { db: tempDb, tempDbPath } = mergeCache.createTempDatabase(exported.meta.name)
const { TempDbWriter } = await import('@openchatlab/node-runtime')
const writer = new TempDbWriter(tempDb)
writer.writeMeta({
name: exported.meta.name,
platform: exported.meta.platform,
type: exported.meta.type,
groupId: exported.meta.groupId,
groupAvatar: exported.meta.groupAvatar,
})
writer.writeMembers(
exported.members.map((m) => ({
platformId: m.platformId,
accountName: m.accountName,
groupNickname: m.groupNickname,
avatar: m.avatar,
}))
)
writer.writeMessages(
exported.messages.map((msg) => ({
senderPlatformId: msg.sender,
senderAccountName: msg.accountName,
senderGroupNickname: msg.groupNickname,
timestamp: msg.timestamp,
type: msg.type,
content: msg.content,
}))
)
writer.finish()
const handle = mergeCache.store(exported.meta.name, tempDbPath)
handles.push({ sessionId: sid, handle })
}
return { success: true, handles }
})
}
+1 -103
View File
@@ -1,103 +1 @@
/**
* Server-side merge session cache.
*
* Manages temporary databases for the multi-file merge workflow.
* Each uploaded/parsed file gets a temp DB handle that persists
* across HTTP requests until merge completes or cache is cleared.
*/
import * as fs from 'fs'
import * as path from 'path'
import * as crypto from 'crypto'
import type { PathProvider, DatabaseAdapter } from '@openchatlab/core'
import {
openBetterSqliteDatabase,
TempDbReader,
deleteTempDatabase,
cleanupTempDatabases,
} from '@openchatlab/node-runtime'
interface CacheEntry {
tempDbPath: string
filename: string
createdAt: number
}
export class MergeSessionCache {
private cache = new Map<string, CacheEntry>()
private tempDir: string
private nativeBinding?: string
constructor(pathProvider: PathProvider, options?: { nativeBinding?: string }) {
this.tempDir = pathProvider.getTempDir()
this.nativeBinding = options?.nativeBinding
if (!fs.existsSync(this.tempDir)) {
fs.mkdirSync(this.tempDir, { recursive: true })
}
}
generateTempDbPath(filename: string): string {
const ts = Date.now()
const rand = Math.random().toString(36).slice(2, 8)
const safeName = filename.replace(/[/\\?%*:|"<>]/g, '_').slice(0, 40)
return path.join(this.tempDir, `merge_${safeName}_${ts}_${rand}.db`)
}
/** Store a parsed file's temp DB and return a handle for subsequent requests. */
store(filename: string, tempDbPath: string): string {
const handle = crypto.randomUUID()
this.cache.set(handle, { tempDbPath, filename, createdAt: Date.now() })
return handle
}
/** Open a TempDbReader for a given handle. Caller must close the reader. */
openReader(handle: string): { reader: TempDbReader; filename: string } | null {
const entry = this.cache.get(handle)
if (!entry || !fs.existsSync(entry.tempDbPath)) return null
const adapter = openBetterSqliteDatabase(entry.tempDbPath, {
readonly: true,
nativeBinding: this.nativeBinding,
})
return { reader: new TempDbReader(adapter), filename: entry.filename }
}
/** Create a writable temp DB for streaming parse. */
createTempDatabase(filename: string): { db: DatabaseAdapter; tempDbPath: string } {
const tempDbPath = this.generateTempDbPath(filename)
const adapter = openBetterSqliteDatabase(tempDbPath, { nativeBinding: this.nativeBinding })
return { db: adapter, tempDbPath }
}
/** Remove a single entry and its temp DB file. */
delete(handle: string): void {
const entry = this.cache.get(handle)
if (entry) {
deleteTempDatabase(entry.tempDbPath)
this.cache.delete(handle)
}
}
/** Clear all entries and delete their temp DB files. */
clear(): void {
for (const entry of this.cache.values()) {
deleteTempDatabase(entry.tempDbPath)
}
this.cache.clear()
}
/** Cleanup orphan temp DBs from previous runs. */
cleanupOrphans(): void {
cleanupTempDatabases(this.tempDir)
}
/** Auto-cleanup entries older than maxAge (default 1 hour). */
cleanupExpired(maxAgeMs = 3600_000): void {
const now = Date.now()
for (const [handle, entry] of this.cache) {
if (now - entry.createdAt > maxAgeMs) {
deleteTempDatabase(entry.tempDbPath)
this.cache.delete(handle)
}
}
}
}
export { MergeSessionCache } from '@openchatlab/node-runtime'