feat(server): add merge HTTP routes and simplify Electron merger layer

- Server: add MergeSessionCache + 5 merge endpoints (parse, conflicts, execute, clear, export-for-merge)
- Electron tempCache.ts: delegate to shared TempDbReader/TempDbWriter via BetterSqliteAdapter
- Electron merger/index.ts: remove dead wrapTempDbReader, use reader.toDataSource() directly
This commit is contained in:
digua committed 2026-05-18 21:07:04 +08:00
1 parent a3a1d289ba
commit 09cdf165b5
5 files changed
+371 -537

No files matched your search

+7 -238
View File
@@ -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<FileParseInfo> {
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<string, ParseResult>): Promise<MergeResult> {
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<string, ChatLabMember>()
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<string>()
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<ConflictCheckResult> {
const readers: TempDbReader[] = []
try {
const dataSources: Array<{ source: MergerDataSource; filename: string }> = []
const dataSources: Array<{ source: ReturnType<TempDbReader['toDataSource']>; 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<TempDbReader['toDataSource']>; 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<string
const messages = db
.prepare(
`SELECT
`SELECT
m.platform_id as sender,
msg.sender_account_name as accountName,
msg.sender_group_nickname as groupNickname,
+40 -297
View File
@@ -1,28 +1,28 @@
/**
* 临时数据库缓存管理器
* 用于合并功能:将解析结果存入临时 SQLite 数据库,避免内存溢出
* Temp DB cache — Electron adapter layer.
*
* Keeps Electron-specific path helpers (getTempDir, generateTempDbPath).
* TempDbReader wraps the shared implementation with better-sqlite3 opening.
* TempDbWriter/cleanup delegate to shared @openchatlab/node-runtime.
*/
import Database from 'better-sqlite3'
import * as fs from 'fs'
import * as path from 'path'
import { ChatType, type ParsedMember, type ParsedMessage } from '../../../src/types/base'
import type { ParsedMeta } from '../parser'
import { BetterSqliteAdapter, TEMP_DB_SCHEMA } from '@openchatlab/node-runtime'
import {
TempDbReader as SharedTempDbReader,
deleteTempDatabase as sharedDeleteTempDb,
cleanupTempDatabases as sharedCleanupTempDbs,
} from '@openchatlab/node-runtime'
import { getPathProvider } from '../path-context'
import { ensureDir } from '../paths'
/**
* 获取临时数据库目录
*/
function getTempDir(): string {
const dir = getPathProvider().getTempDir()
ensureDir(dir)
return dir
}
/**
* 生成临时数据库文件路径
*/
export function generateTempDbPath(sourceFilePath: string): string {
const timestamp = Date.now()
const random = Math.random().toString(36).substring(2, 8)
@@ -32,329 +32,72 @@ export function generateTempDbPath(sourceFilePath: string): string {
}
/**
* 创建临时数据库并初始化表结构
* Create a temp database and return the raw better-sqlite3 handle.
* Used by streamImport.ts which wraps it in BetterSqliteAdapter.
*/
export function createTempDatabase(dbPath: string): Database.Database {
const db = new Database(dbPath)
db.pragma('journal_mode = WAL')
db.pragma('synchronous = NORMAL')
db.exec(`
CREATE TABLE IF NOT EXISTS meta (
name TEXT NOT NULL,
platform TEXT NOT NULL,
type TEXT NOT NULL,
group_id TEXT,
group_avatar TEXT
);
CREATE TABLE IF NOT EXISTS member (
platform_id TEXT PRIMARY KEY,
account_name TEXT,
group_nickname TEXT,
avatar TEXT
);
CREATE TABLE IF NOT EXISTS message (
id INTEGER PRIMARY KEY AUTOINCREMENT,
sender_platform_id TEXT NOT NULL,
sender_account_name TEXT,
sender_group_nickname TEXT,
timestamp INTEGER NOT NULL,
type INTEGER NOT NULL,
content TEXT
);
CREATE INDEX IF NOT EXISTS idx_message_ts ON message(timestamp);
CREATE INDEX IF NOT EXISTS idx_message_sender ON message(sender_platform_id);
`)
db.exec(TEMP_DB_SCHEMA)
return db
}
/**
* 临时数据库写入器
* 用于流式写入解析结果
*/
export class TempDbWriter {
private db: Database.Database
private insertMeta: Database.Statement
private insertMember: Database.Statement
private insertMessage: Database.Statement
private memberSet: Set<string> = 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<SharedTempDbReader['streamMessages']>[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)
}
+1 -1
View File
@@ -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 })
+220 -1
View File
@@ -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 }
})
}
}
+103
View File
@@ -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<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)
}
}
}
}