diff --git a/electron/main/merger/index.ts b/electron/main/merger/index.ts index 63ab3efb..25d22e61 100644 --- a/electron/main/merger/index.ts +++ b/electron/main/merger/index.ts @@ -6,24 +6,18 @@ import * as fs from 'fs' import * as path from 'path' -import { generateMessageKey, getCollidingPlatformIds, normalizePlatformId } from '@openchatlab/core' -import type { MergerMember, MergerMessage } from '@openchatlab/core' import { checkConflictsFromSources, buildMergedOutput, serializeChatLabToJsonl } from '@openchatlab/node-runtime' -import type { MergerDataSource, MergerSourceMeta, ChatLabOutput } from '@openchatlab/node-runtime' -import { parseFileSync, detectFormat } from '../parser' +import type { ChatLabOutput } from '@openchatlab/node-runtime' import { importData } from '../database/core' import { TempDbReader } from './tempCache' import { getPathProvider } from '../path-context' -import type { ParseResult, ChatPlatform, ChatType } from '../../../src/types/base' +import type { ChatPlatform, ChatType } from '../../../src/types/base' import type { ChatLabFormat, - ChatLabMember, ChatLabMessage, - FileParseInfo, ConflictCheckResult, MergeParams, MergeResult, - MergeSource, } from '../../../src/types/format' function getDefaultOutputDir(): string { @@ -42,27 +36,6 @@ function generateOutputFilename(name: string, format: 'json' | 'jsonl' = 'json') return `${safeName}_merged_${date}.${format}` } -/** - * Parse a file to get basic info (for preview). - * Prefer parser.parseFileInfo for more detailed info. - */ -export async function parseFileInfo(filePath: string): Promise { - const format = detectFormat(filePath) - if (!format) { - throw new Error('Cannot detect file format') - } - - const result = await parseFileSync(filePath) - - return { - name: result.meta.name, - format: format.name, - platform: result.meta.platform, - messageCount: result.messages.length, - memberCount: result.members.length, - } -} - // ==================== ChatLabOutput → ChatLabFormat conversion ==================== function chatLabOutputToFormat(data: ChatLabOutput): ChatLabFormat { @@ -90,209 +63,6 @@ function chatLabOutputToFormat(data: ChatLabOutput): ChatLabFormat { } } -// ==================== TempDbReader → MergerDataSource adapter ==================== - -function wrapTempDbReader(reader: TempDbReader): MergerDataSource { - return { - getMeta(): MergerSourceMeta | null { - const meta = reader.getMeta() - if (!meta) return null - return { - name: meta.name, - platform: meta.platform, - type: meta.type, - groupId: meta.groupId, - groupAvatar: meta.groupAvatar, - } - }, - getMembers(): MergerMember[] { - return reader.getMembers().map((m) => ({ - platformId: m.platformId, - accountName: m.accountName, - groupNickname: m.groupNickname, - avatar: m.avatar, - })) - }, - getMessageCount(): number { - return reader.getMessageCount() - }, - streamMessages(batchSize: number, callback: (messages: MergerMessage[]) => void): void { - reader.streamMessages(batchSize, (messages) => { - callback( - messages.map((msg) => ({ - senderPlatformId: msg.senderPlatformId, - senderAccountName: msg.senderAccountName, - senderGroupNickname: msg.senderGroupNickname, - timestamp: msg.timestamp, - type: msg.type, - content: msg.content, - })) - ) - }) - }, - } -} - -// ==================== Merge with cache (legacy, non-TempDB path) ==================== - -export async function mergeFilesWithCache(params: MergeParams, cache: Map): Promise { - try { - const { filePaths, outputName, outputDir, conflictResolutions: _conflictResolutions, andAnalyze } = params - - console.log('[Merger] mergeFilesWithCache: Starting merge') - - const parseResults: Array<{ result: ParseResult; source: string }> = [] - for (const filePath of filePaths) { - let result: ParseResult - if (cache.has(filePath)) { - result = cache.get(filePath)! - console.log(`[Merger] Cache hit: ${path.basename(filePath)}`) - } else { - console.log(`[Merger] Cache miss, re-parsing: ${path.basename(filePath)}`) - result = await parseFileSync(filePath) - } - parseResults.push({ result, source: path.basename(filePath) }) - } - - return executeMergeLegacy(parseResults, outputName, outputDir, andAnalyze) - } catch (err) { - return { - success: false, - error: err instanceof Error ? err.message : 'Merge failed', - } - } -} - -function executeMergeLegacy( - parseResults: Array<{ result: ParseResult; source: string }>, - outputName: string, - outputDir: string | undefined, - andAnalyze: boolean -): MergeResult { - const collidingIds = getCollidingPlatformIds( - parseResults.map(({ result }) => ({ - platform: result.meta.platform || 'unknown', - members: result.members.map((m) => ({ platformId: m.platformId })), - })) - ) - - const memberMap = new Map() - for (const { result } of parseResults) { - const sourcePlatform = result.meta.platform || 'unknown' - for (const member of result.members) { - const nid = normalizePlatformId(member.platformId, sourcePlatform, collidingIds) - const existing = memberMap.get(nid) - if (existing) { - if (member.accountName) existing.accountName = member.accountName - if (member.groupNickname) existing.groupNickname = member.groupNickname - if (member.avatar) existing.avatar = member.avatar - } else { - memberMap.set(nid, { - platformId: nid, - accountName: member.accountName, - groupNickname: member.groupNickname, - avatar: member.avatar, - }) - } - } - } - - const seenKeys = new Set() - const mergedMessages: ChatLabMessage[] = [] - for (const { result } of parseResults) { - const sourcePlatform = result.meta.platform || 'unknown' - for (const msg of result.messages) { - const nid = normalizePlatformId(msg.senderPlatformId, sourcePlatform, collidingIds) - const key = generateMessageKey(msg.timestamp, nid, msg.content ?? null) - if (seenKeys.has(key)) continue - seenKeys.add(key) - mergedMessages.push({ - sender: nid, - accountName: msg.senderAccountName, - groupNickname: msg.senderGroupNickname, - timestamp: msg.timestamp, - type: msg.type, - content: msg.content, - }) - } - } - - mergedMessages.sort((a, b) => a.timestamp - b.timestamp) - - const sources: MergeSource[] = parseResults.map(({ result, source }) => ({ - filename: source, - platform: result.meta.platform, - messageCount: result.messages.length, - })) - - const groupIds = new Set(parseResults.map(({ result }) => result.meta.groupId).filter(Boolean)) - const groupId = - groupIds.size === 1 ? parseResults.find(({ result }) => result.meta.groupId)?.result.meta.groupId : undefined - const groupAvatar = groupId - ? parseResults.filter(({ result }) => result.meta.groupId === groupId).pop()?.result.meta.groupAvatar - : undefined - - const chatLabHeader = { - version: '0.0.1', - exportedAt: Math.floor(Date.now() / 1000), - generator: 'ChatLab Merge Tool', - description: `Merged from ${parseResults.length} files`, - } - - const uniquePlatforms = [...new Set(parseResults.map(({ result }) => result.meta.platform || 'unknown'))] - const mergedPlatform = uniquePlatforms.length === 1 ? uniquePlatforms[0] : 'mixed' - - const chatLabMeta = { - name: outputName, - platform: mergedPlatform as ChatPlatform, - type: parseResults[0].result.meta.type as ChatType, - sources, - groupId, - groupAvatar, - } - - const targetDir = outputDir || getDefaultOutputDir() - ensureOutputDir(targetDir) - const outputPath = path.join(targetDir, generateOutputFilename(outputName, 'json')) - - const chatLabData: ChatLabFormat = { - chatlab: chatLabHeader, - meta: chatLabMeta, - members: Array.from(memberMap.values()), - messages: mergedMessages, - } - fs.writeFileSync(outputPath, JSON.stringify(chatLabData, null, 2), 'utf-8') - - let sessionId: string | undefined - if (andAnalyze) { - sessionId = importData({ - meta: { - name: chatLabMeta.name, - platform: chatLabMeta.platform, - type: chatLabMeta.type, - groupId: chatLabMeta.groupId, - groupAvatar: chatLabMeta.groupAvatar, - }, - members: chatLabData.members.map((member) => ({ - platformId: member.platformId, - accountName: member.accountName, - groupNickname: member.groupNickname, - avatar: member.avatar, - })), - messages: chatLabData.messages.map((msg) => ({ - senderPlatformId: msg.sender, - senderAccountName: msg.accountName, - senderGroupNickname: msg.groupNickname, - timestamp: msg.timestamp, - type: msg.type, - content: msg.content, - })), - }) - } - - return { success: true, outputPath, sessionId } -} - // ==================== TempDB-based merge (delegates to node-runtime) ==================== export async function checkConflictsWithTempDb( @@ -301,7 +71,7 @@ export async function checkConflictsWithTempDb( ): Promise { const readers: TempDbReader[] = [] try { - const dataSources: Array<{ source: MergerDataSource; filename: string }> = [] + const dataSources: Array<{ source: ReturnType; filename: string }> = [] for (const filePath of filePaths) { const tempDbPath = tempDbCache.get(filePath) if (!tempDbPath) { @@ -309,7 +79,7 @@ export async function checkConflictsWithTempDb( } const reader = new TempDbReader(tempDbPath) readers.push(reader) - dataSources.push({ source: wrapTempDbReader(reader), filename: path.basename(filePath) }) + dataSources.push({ source: reader.toDataSource(), filename: path.basename(filePath) }) } return checkConflictsFromSources(dataSources) @@ -330,7 +100,7 @@ export async function mergeFilesWithTempDb( const readers: TempDbReader[] = [] try { - const dataSources: Array<{ source: MergerDataSource; filename: string }> = [] + const dataSources: Array<{ source: ReturnType; filename: string }> = [] for (const filePath of filePaths) { const tempDbPath = tempDbCache.get(filePath) if (!tempDbPath) { @@ -338,13 +108,12 @@ export async function mergeFilesWithTempDb( } const reader = new TempDbReader(tempDbPath) readers.push(reader) - dataSources.push({ source: wrapTempDbReader(reader), filename: path.basename(filePath) }) + dataSources.push({ source: reader.toDataSource(), filename: path.basename(filePath) }) } const result = buildMergedOutput(dataSources, outputName) const chatLabData = result.chatLabData - // Write to disk const targetDir = outputDir || getDefaultOutputDir() ensureOutputDir(targetDir) const filename = generateOutputFilename(outputName, outputFormat) @@ -443,7 +212,7 @@ export async function exportSessionToTempFile(sessionId: string): Promise = new Set() - private messageCount: number = 0 - - constructor(dbPath: string) { - this.db = createTempDatabase(dbPath) - - // 准备语句 - this.insertMeta = this.db.prepare(` - INSERT INTO meta (name, platform, type, group_id, group_avatar) VALUES (?, ?, ?, ?, ?) - `) - this.insertMember = this.db.prepare(` - INSERT OR IGNORE INTO member (platform_id, account_name, group_nickname, avatar) VALUES (?, ?, ?, ?) - `) - this.insertMessage = this.db.prepare(` - INSERT INTO message (sender_platform_id, sender_account_name, sender_group_nickname, timestamp, type, content) - VALUES (?, ?, ?, ?, ?, ?) - `) - - // 开始事务 - this.db.exec('BEGIN TRANSACTION') - } - - /** - * 写入元信息 - */ - writeMeta(meta: ParsedMeta): void { - this.insertMeta.run(meta.name, meta.platform, meta.type, meta.groupId || null, meta.groupAvatar || null) - } - - /** - * 写入成员（批量） - */ - writeMembers(members: ParsedMember[]): void { - for (const m of members) { - if (!this.memberSet.has(m.platformId)) { - this.memberSet.add(m.platformId) - this.insertMember.run(m.platformId, m.accountName || null, m.groupNickname || null, m.avatar || null) - } - } - } - - /** - * 写入消息（批量） - */ - writeMessages(messages: ParsedMessage[]): void { - for (const msg of messages) { - // 确保成员存在（消息中没有头像信息，设为 null） - if (!this.memberSet.has(msg.senderPlatformId)) { - this.memberSet.add(msg.senderPlatformId) - this.insertMember.run( - msg.senderPlatformId, - msg.senderAccountName || null, - msg.senderGroupNickname || null, - null - ) - } - - this.insertMessage.run( - msg.senderPlatformId, - msg.senderAccountName || null, - msg.senderGroupNickname || null, - msg.timestamp, - msg.type, - msg.content || null - ) - this.messageCount++ - } - } - - /** - * 完成写入（提交事务） - */ - finish(): { messageCount: number; memberCount: number } { - this.db.exec('COMMIT') - const result = { - messageCount: this.messageCount, - memberCount: this.memberSet.size, - } - this.db.close() - return result - } - - /** - * 取消写入（回滚事务） - */ - abort(): void { - try { - this.db.exec('ROLLBACK') - } catch { - // 忽略回滚错误 - } - this.db.close() - } -} - -/** - * 临时数据库读取器 - * 用于流式读取合并时的数据 + * Electron TempDbReader — opens the temp DB with better-sqlite3 + * and delegates to the shared TempDbReader via BetterSqliteAdapter. */ export class TempDbReader { - private db: Database.Database + private inner: SharedTempDbReader private dbPath: string constructor(dbPath: string) { this.dbPath = dbPath - this.db = new Database(dbPath, { readonly: true }) - this.db.pragma('journal_mode = WAL') + const db = new Database(dbPath, { readonly: true }) + db.pragma('journal_mode = WAL') + this.inner = new SharedTempDbReader(new BetterSqliteAdapter(db)) } - /** - * 读取元信息 - */ - getMeta(): ParsedMeta | null { - const row = this.db.prepare('SELECT * FROM meta LIMIT 1').get() as - | { name: string; platform: string; type: string; group_id: string | null; group_avatar: string | null } - | undefined - if (!row) return null - return { - name: row.name, - platform: row.platform, - type: row.type === ChatType.PRIVATE ? ChatType.PRIVATE : ChatType.GROUP, - groupId: row.group_id || undefined, - groupAvatar: row.group_avatar || undefined, - } + getMeta() { + return this.inner.getMeta() } - /** - * 读取所有成员 - */ - getMembers(): ParsedMember[] { - const rows = this.db.prepare('SELECT * FROM member').all() as Array<{ - platform_id: string - account_name: string | null - group_nickname: string | null - avatar: string | null - }> - return rows.map((r) => ({ - platformId: r.platform_id, - accountName: r.account_name || r.platform_id, // 如果没有账号名称，使用 platformId - groupNickname: r.group_nickname || undefined, - avatar: r.avatar || undefined, - })) + getMembers() { + return this.inner.getMembers() } - /** - * 获取消息总数 - */ - getMessageCount(): number { - const row = this.db.prepare('SELECT COUNT(*) as count FROM message').get() as { count: number } - return row.count + getMessageCount() { + return this.inner.getMessageCount() } - /** - * 流式读取消息（分批） - * @param batchSize 每批消息数量 - * @param callback 处理每批消息的回调 - */ - streamMessages(batchSize: number, callback: (messages: ParsedMessage[]) => void): void { - const stmt = this.db.prepare(` - SELECT sender_platform_id, sender_account_name, sender_group_nickname, timestamp, type, content - FROM message - ORDER BY timestamp ASC - LIMIT ? OFFSET ? - `) - - let offset = 0 - while (true) { - const rows = stmt.all(batchSize, offset) as Array<{ - sender_platform_id: string - sender_account_name: string | null - sender_group_nickname: string | null - timestamp: number - type: number - content: string | null - }> - - if (rows.length === 0) break - - const messages: ParsedMessage[] = rows.map((r) => ({ - senderPlatformId: r.sender_platform_id, - senderAccountName: r.sender_account_name || r.sender_platform_id, - senderGroupNickname: r.sender_group_nickname || undefined, - timestamp: r.timestamp, - type: r.type as ParsedMessage['type'], - content: r.content, - })) - - callback(messages) - offset += batchSize - } + streamMessages(batchSize: number, callback: Parameters[1]) { + return this.inner.streamMessages(batchSize, callback) } - /** - * 获取所有消息（用于冲突检测，内存中处理） - * 注意：对于超大文件，应使用 streamMessages - */ - getAllMessages(): ParsedMessage[] { - const rows = this.db - .prepare( - ` - SELECT sender_platform_id, sender_account_name, sender_group_nickname, timestamp, type, content - FROM message - ORDER BY timestamp ASC - ` - ) - .all() as Array<{ - sender_platform_id: string - sender_account_name: string | null - sender_group_nickname: string | null - timestamp: number - type: number - content: string | null - }> - - return rows.map((r) => ({ - senderPlatformId: r.sender_platform_id, - senderAccountName: r.sender_account_name || r.sender_platform_id, - senderGroupNickname: r.sender_group_nickname || undefined, - timestamp: r.timestamp, - type: r.type as ParsedMessage['type'], - content: r.content, - })) + getAllMessages() { + return this.inner.getAllMessages() } - /** - * 关闭数据库连接 - */ - close(): void { - this.db.close() + close() { + this.inner.close() } - /** - * 获取数据库路径 - */ - getPath(): string { + getPath() { return this.dbPath } -} -/** - * 删除临时数据库文件 - */ -export function deleteTempDatabase(dbPath: string): void { - try { - const walPath = dbPath + '-wal' - const shmPath = dbPath + '-shm' - - if (fs.existsSync(dbPath)) fs.unlinkSync(dbPath) - if (fs.existsSync(walPath)) fs.unlinkSync(walPath) - if (fs.existsSync(shmPath)) fs.unlinkSync(shmPath) - - console.log(`[TempCache] Deleted temp database: ${dbPath}`) - } catch (error) { - console.error(`[TempCache] Failed to delete temp database: ${dbPath}`, error) + toDataSource() { + return this.inner.toDataSource() } } -/** - * 清理所有临时数据库（应用启动时调用） - */ +export function deleteTempDatabase(dbPath: string): void { + sharedDeleteTempDb(dbPath) +} + export function cleanupAllTempDatabases(): void { try { - const dir = getTempDir() - if (!fs.existsSync(dir)) return - - const files = fs.readdirSync(dir) - for (const file of files) { - if (file.startsWith('merge_') && file.endsWith('.db')) { - const filePath = path.join(dir, file) - deleteTempDatabase(filePath) - } - } - console.log('[TempCache] Cleaned up all temp databases') + sharedCleanupTempDbs(getTempDir()) } catch (error) { console.error('[TempCache] Failed to clean up temp databases:', error) } diff --git a/packages/server/src/http/index.ts b/packages/server/src/http/index.ts index 82702650..794f716d 100644 --- a/packages/server/src/http/index.ts +++ b/packages/server/src/http/index.ts @@ -115,7 +115,7 @@ export async function startHttpServer(options?: HttpServerOptions): Promise<{ registerSessionRoutes(server, dbManager) registerNlpRoutes(server, dbManager) registerAiRoutes(server, dbManager, convManager) - registerWebRoutes(server, dbManager) + registerWebRoutes(server, dbManager, { pathProvider, nativeBinding }) initSync(server, dbManager, pathProvider, { port, host, token }) diff --git a/packages/server/src/http/routes/web.ts b/packages/server/src/http/routes/web.ts index b483950d..7d964830 100644 --- a/packages/server/src/http/routes/web.ts +++ b/packages/server/src/http/routes/web.ts @@ -47,7 +47,14 @@ import { generateSessionSummary, type SummaryDeps, exportFilterResultToMarkdown, + streamParseFileInfo, + checkConflictsFromSources, + buildMergedOutput, + serializeChatLabToJsonl, + exportSessionToJson, + TempDbReader, } from '@openchatlab/node-runtime' +import { MergeSessionCache } from '../../merger/merge-cache' import { getChatSessionSummary, saveChatSessionSummary, @@ -144,7 +151,15 @@ function buildSummaryDeps( } } -export function registerWebRoutes(server: FastifyInstance, dbManager: DatabaseManager): void { +export function registerWebRoutes( + server: FastifyInstance, + dbManager: DatabaseManager, + options?: { pathProvider?: import('@openchatlab/core').PathProvider; nativeBinding?: string } +): void { + const mergeCache = options?.pathProvider + ? new MergeSessionCache(options.pathProvider, { nativeBinding: options.nativeBinding }) + : null + mergeCache?.cleanupOrphans() // ==================== 会话管理 ==================== server.get('/_web/sessions', async () => { @@ -986,4 +1001,208 @@ export function registerWebRoutes(server: FastifyInstance, dbManager: DatabaseMa } } }) + + // ==================== 合并功能 ==================== + + if (mergeCache) { + 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 */ + } + } + } + + // Cleanup merge cache after successful merge + 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 = dbManager.open(sid) + if (!db) return reply.code(404).send({ error: `Session not found: ${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 } + }) + } } diff --git a/packages/server/src/merger/merge-cache.ts b/packages/server/src/merger/merge-cache.ts new file mode 100644 index 00000000..a6f9dbf7 --- /dev/null +++ b/packages/server/src/merger/merge-cache.ts @@ -0,0 +1,103 @@ +/** + * 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() + 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) + } + } + } +}