refactor(services): unify CLI Web and Electron service layer

Extract shared business logic from monolithic web.ts into reusable
services in @openchatlab/node-runtime (session, member, index, summary,
export), using SessionRuntimeAdapter to abstract platform differences.
Split web.ts into 12 domain-specific route modules under routes/web/.

Co-authored-by: Cursor <cursoragent@cursor.com>
This commit is contained in:
diguaandCursor authored and digua committed 2026-05-22 10:20:39 +08:00
1 parent 808b9026f2
commit dd3cb10414
24 files changed
+1721 -1344

No files matched your search

File diff suppressed because it is too large. Load diff
+172
View File
@@ -0,0 +1,172 @@
import type { FastifyInstance } from 'fastify'
import type { DatabaseManager } from '@openchatlab/node-runtime'
import { createJiebaNlpProvider, type SessionRuntimeAdapter } from '@openchatlab/node-runtime'
import {
getTimeRange,
getAvailableYears,
getMemberActivity,
getHourlyActivity,
getDailyActivity,
getWeekdayActivity,
getMessageTypeStats,
getMonthlyActivity,
getYearlyActivity,
getMessageLengthDistribution,
getRelationshipStats,
getCatchphraseAnalysis,
getMentionAnalysis,
getMentionGraph,
getLaughAnalysis,
getClusterGraph,
getLanguagePreferenceAnalysis,
} from '@openchatlab/core'
import { parseTimeFilter } from './helpers'
type FilteredQuery = { startTs?: string; endTs?: string; memberId?: string }
export function registerAnalyticsRoutes(
server: FastifyInstance,
_dbManager: DatabaseManager,
adapter: SessionRuntimeAdapter
): void {
// ==================== 时间范围 ====================
server.get<{ Params: { id: string } }>('/_web/sessions/:id/years', async (request) => {
const db = adapter.ensureReadonly(request.params.id)
return getAvailableYears(db)
})
server.get<{ Params: { id: string } }>('/_web/sessions/:id/time-range', async (request) => {
const db = adapter.ensureReadonly(request.params.id)
return getTimeRange(db)
})
// ==================== 统计分析 ====================
server.get<{ Params: { id: string }; Querystring: FilteredQuery }>(
'/_web/sessions/:id/stats/member-activity',
async (request) => {
const db = adapter.ensureReadonly(request.params.id)
return getMemberActivity(db, parseTimeFilter(request.query))
}
)
server.get<{ Params: { id: string }; Querystring: FilteredQuery }>(
'/_web/sessions/:id/stats/hourly',
async (request) => {
const db = adapter.ensureReadonly(request.params.id)
return getHourlyActivity(db, parseTimeFilter(request.query))
}
)
server.get<{ Params: { id: string }; Querystring: FilteredQuery }>(
'/_web/sessions/:id/stats/daily',
async (request) => {
const db = adapter.ensureReadonly(request.params.id)
return getDailyActivity(db, parseTimeFilter(request.query))
}
)
server.get<{ Params: { id: string }; Querystring: FilteredQuery }>(
'/_web/sessions/:id/stats/weekday',
async (request) => {
const db = adapter.ensureReadonly(request.params.id)
return getWeekdayActivity(db, parseTimeFilter(request.query))
}
)
server.get<{ Params: { id: string }; Querystring: FilteredQuery }>(
'/_web/sessions/:id/stats/message-types',
async (request) => {
const db = adapter.ensureReadonly(request.params.id)
return getMessageTypeStats(db, parseTimeFilter(request.query))
}
)
// ==================== 高级分析 ====================
server.get<{ Params: { id: string }; Querystring: FilteredQuery }>(
'/_web/sessions/:id/analytics/relationship',
async (request) => {
const db = adapter.ensureReadonly(request.params.id)
return getRelationshipStats(db, parseTimeFilter(request.query))
}
)
server.get<{ Params: { id: string }; Querystring: FilteredQuery }>(
'/_web/sessions/:id/analytics/catchphrase',
async (request) => {
const db = adapter.ensureReadonly(request.params.id)
return getCatchphraseAnalysis(db, parseTimeFilter(request.query))
}
)
server.get<{ Params: { id: string }; Querystring: FilteredQuery }>(
'/_web/sessions/:id/analytics/mention',
async (request) => {
const db = adapter.ensureReadonly(request.params.id)
return getMentionAnalysis(db, parseTimeFilter(request.query))
}
)
server.get<{ Params: { id: string }; Querystring: FilteredQuery }>(
'/_web/sessions/:id/analytics/mention-graph',
async (request) => {
const db = adapter.ensureReadonly(request.params.id)
return getMentionGraph(db, parseTimeFilter(request.query))
}
)
server.get<{ Params: { id: string }; Querystring: FilteredQuery }>(
'/_web/sessions/:id/analytics/laugh',
async (request) => {
const db = adapter.ensureReadonly(request.params.id)
return getLaughAnalysis(db, parseTimeFilter(request.query))
}
)
server.get<{ Params: { id: string }; Querystring: FilteredQuery & { topEdges?: string } }>(
'/_web/sessions/:id/analytics/cluster',
async (request) => {
const db = adapter.ensureReadonly(request.params.id)
const filter = parseTimeFilter(request.query)
const topEdges = request.query.topEdges ? parseInt(request.query.topEdges, 10) : undefined
return getClusterGraph(db, filter, topEdges ? { topEdges } : undefined)
}
)
server.get<{ Params: { id: string }; Querystring: FilteredQuery & { locale?: string } }>(
'/_web/sessions/:id/analytics/language-preference',
async (request) => {
const db = adapter.ensureReadonly(request.params.id)
const filter = parseTimeFilter(request.query)
const locale = request.query.locale || 'zh-CN'
const nlpProvider = createJiebaNlpProvider()
return getLanguagePreferenceAnalysis(db, { locale, timeFilter: filter, nlpProvider })
}
)
server.get<{ Params: { id: string }; Querystring: FilteredQuery }>(
'/_web/sessions/:id/analytics/monthly-activity',
async (request) => {
const db = adapter.ensureReadonly(request.params.id)
return getMonthlyActivity(db, parseTimeFilter(request.query))
}
)
server.get<{ Params: { id: string }; Querystring: FilteredQuery }>(
'/_web/sessions/:id/analytics/yearly-activity',
async (request) => {
const db = adapter.ensureReadonly(request.params.id)
return getYearlyActivity(db, parseTimeFilter(request.query))
}
)
server.get<{ Params: { id: string }; Querystring: FilteredQuery }>(
'/_web/sessions/:id/analytics/message-length-distribution',
async (request) => {
const db = adapter.ensureReadonly(request.params.id)
return getMessageLengthDistribution(db, parseTimeFilter(request.query))
}
)
}
+50
View File
@@ -0,0 +1,50 @@
import * as fs from 'fs'
import * as path from 'path'
import type { FastifyInstance } from 'fastify'
export function registerCacheRoutes(server: FastifyInstance, downloadsDir: string): void {
server.post<{
Body: { filename: string; dataUrl: string }
}>('/_web/cache/save-to-downloads', async (request) => {
const { filename, dataUrl } = request.body
if (!filename || !dataUrl) {
return { success: false, error: 'filename and dataUrl are required' }
}
const base64Prefix = dataUrl.indexOf(',')
const base64Data = base64Prefix >= 0 ? dataUrl.slice(base64Prefix + 1) : dataUrl
const buffer = Buffer.from(base64Data, 'base64')
if (!fs.existsSync(downloadsDir)) {
fs.mkdirSync(downloadsDir, { recursive: true })
}
const filePath = path.join(downloadsDir, filename)
fs.writeFileSync(filePath, buffer)
return { success: true, filePath }
})
server.post<{
Body: { filePath: string }
}>('/_web/cache/show-in-folder', async (request) => {
const { filePath } = request.body
if (!filePath) {
return { success: false, error: 'filePath is required' }
}
const { exec } = await import('child_process')
const platform = process.platform
const dir = path.dirname(filePath)
if (platform === 'darwin') {
exec(`open -R "${filePath}"`)
} else if (platform === 'win32') {
exec(`explorer /select,"${filePath}"`)
} else {
exec(`xdg-open "${dir}"`)
}
return { success: true }
})
}
+40
View File
@@ -0,0 +1,40 @@
import type { FastifyInstance } from 'fastify'
import { exportService, type SessionRuntimeAdapter } from '@openchatlab/node-runtime'
export function registerExportRoutes(server: FastifyInstance, adapter: SessionRuntimeAdapter): void {
server.post<{
Params: { id: string }
Body: {
sessionName: string
filterMode: 'condition' | 'session'
keywords?: string[]
timeFilter?: { startTs: number; endTs: number }
senderIds?: number[]
contextSize?: number
chatSessionIds?: number[]
}
}>('/_web/sessions/:id/export/markdown', async (request, reply) => {
const { id } = request.params
const body = request.body as any
const sessionName = body?.sessionName || id
const { result, content } = exportService.exportMarkdown(adapter, {
sessionId: id,
sessionName,
filterMode: body.filterMode || 'condition',
keywords: body.keywords,
timeFilter: body.timeFilter,
senderIds: body.senderIds,
contextSize: body.contextSize,
chatSessionIds: body.chatSessionIds,
})
if (!result.success) {
return reply.code(500).send({ error: result.error })
}
reply.header('Content-Type', 'text/markdown; charset=utf-8')
reply.header('Content-Disposition', `attachment; filename="${encodeURIComponent(sessionName)}_export.md"`)
return reply.send(content)
})
}
+41
View File
@@ -0,0 +1,41 @@
/**
* Shared helpers for web route modules.
*/
import * as fs from 'fs'
import * as path from 'path'
import type { TimeFilter } from '@openchatlab/shared-types'
import type { DatabaseManager } from '@openchatlab/node-runtime'
export function parseTimeFilter(query: Record<string, string | undefined>): TimeFilter | undefined {
const { startTs, endTs, memberId } = query
if (!startTs && !endTs && !memberId) return undefined
const filter: TimeFilter = {}
if (startTs) filter.startTs = parseInt(startTs, 10)
if (endTs) filter.endTs = parseInt(endTs, 10)
if (memberId) filter.memberId = parseInt(memberId, 10)
return filter
}
export function resolveNativeBinding(): string | undefined {
if (process.versions.electron) return undefined
const nativePath = path.resolve(__dirname, '../../../../native/better_sqlite3.node')
if (fs.existsSync(nativePath)) return nativePath
return undefined
}
export function getAiDataDir(dbManager: DatabaseManager): string {
const pathProvider = (dbManager as any)['pathProvider']
if (!pathProvider) {
throw Object.assign(new Error('PathProvider not available'), { statusCode: 500 })
}
return pathProvider.getAiDataDir()
}
export function ensureDb(dbManager: DatabaseManager, sessionId: string) {
const db = dbManager.open(sessionId)
if (!db) {
throw Object.assign(new Error(`Session not found: ${sessionId}`), { statusCode: 404 })
}
return db
}
+365
View File
@@ -0,0 +1,365 @@
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 {
streamImport,
incrementalImport,
analyzeIncrementalImport,
analyzeNewImport,
detectFormat,
detectAllFormats,
getSupportedFormats,
scanMultiChatFile,
findEntryFileInDirectory,
} from '../../../import'
import { resolveNativeBinding } from './helpers'
const DEMO_BASE_URL = 'https://chatlab.fun/assets/demo'
function cleanupTemp(...paths: string[]) {
for (const p of paths) {
try {
const stat = fs.statSync(p)
if (stat.isDirectory()) {
fs.rmSync(p, { recursive: true, force: true })
} else {
fs.unlinkSync(p)
}
} catch {
/* ignore */
}
}
}
export function registerImportRoutes(server: FastifyInstance, dbManager: DatabaseManager): void {
server.get('/_web/supported-formats', async () => {
return getSupportedFormats()
})
server.post('/_web/detect-format', 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-detect-'))
const tmpPath = path.join(tmpDir, data.filename || 'upload')
try {
const chunks: Buffer[] = []
for await (const chunk of data.file) {
chunks.push(chunk)
}
fs.writeFileSync(tmpPath, Buffer.concat(chunks))
const format = detectFormat(tmpPath)
const allFormats = detectAllFormats(tmpPath)
return { format, allFormats }
} finally {
cleanupTemp(tmpPath, tmpDir)
}
})
server.post('/_web/scan-multi-chat', 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-scan-'))
const tmpPath = path.join(tmpDir, data.filename || 'upload')
try {
const chunks: Buffer[] = []
for await (const chunk of data.file) {
chunks.push(chunk)
}
fs.writeFileSync(tmpPath, Buffer.concat(chunks))
const chats = await scanMultiChatFile(tmpPath)
return { chats }
} finally {
cleanupTemp(tmpPath, tmpDir)
}
})
server.post('/_web/import', 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-import-'))
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))
const formatId = (data.fields?.formatId as any)?.value as string | undefined
const chatIndexStr = (data.fields?.chatIndex as any)?.value as string | undefined
const chatIndex = chatIndexStr !== undefined ? parseInt(chatIndexStr, 10) : undefined
reply.raw.writeHead(200, {
'Content-Type': 'text/event-stream',
'Cache-Control': 'no-cache',
Connection: 'keep-alive',
})
function sendEvent(event: string, eventData: unknown) {
reply.raw.write(`event: ${event}\ndata: ${JSON.stringify(eventData)}\n\n`)
}
try {
const nativeBinding = resolveNativeBinding()
const result = await streamImport(dbManager, tmpPath, {
formatId,
chatIndex,
nativeBinding,
onProgress: (p) => sendEvent('progress', p),
})
if (result.success) {
sendEvent('done', {
success: true,
sessionId: result.sessionId,
messageCount: result.diagnostics?.messagesWritten ?? 0,
memberCount: 0,
})
} else {
sendEvent('error', { success: false, error: result.error })
}
} catch (err) {
sendEvent('error', { success: false, error: err instanceof Error ? err.message : String(err) })
} finally {
reply.raw.end()
cleanupTemp(tmpPath, tmpDir)
}
})
// ==================== Directory Import ====================
server.post('/_web/import-directory', async (request, reply) => {
const parts = (request as any).parts()
if (!parts) return reply.code(400).send({ error: 'No files uploaded' })
const tmpDir = fs.mkdtempSync(path.join(os.tmpdir(), 'chatlab-dir-import-'))
const relativePaths: string[] = []
const fileBuffers: { data: Buffer; filename: string }[] = []
try {
for await (const part of parts) {
if (part.type === 'field' && part.fieldname === 'relativePaths') {
relativePaths.push(String(part.value))
} else if (part.type === 'file') {
const chunks: Buffer[] = []
for await (const chunk of part.file) {
chunks.push(chunk)
}
fileBuffers.push({ data: Buffer.concat(chunks), filename: part.filename || '' })
}
}
for (let i = 0; i < fileBuffers.length; i++) {
let relPath = relativePaths[i] || fileBuffers[i].filename || `file_${i}`
const segments = relPath.split('/')
if (segments.length > 1) {
relPath = segments.slice(1).join('/')
}
const targetPath = path.join(tmpDir, relPath)
fs.mkdirSync(path.dirname(targetPath), { recursive: true })
fs.writeFileSync(targetPath, fileBuffers[i].data)
}
const entryPath = findEntryFileInDirectory(tmpDir)
if (!entryPath) {
return reply.code(400).send({ error: 'No recognizable import format found in directory' })
}
reply.raw.writeHead(200, {
'Content-Type': 'text/event-stream',
'Cache-Control': 'no-cache',
Connection: 'keep-alive',
})
function sendEvent(event: string, eventData: unknown) {
reply.raw.write(`event: ${event}\ndata: ${JSON.stringify(eventData)}\n\n`)
}
const nativeBinding = resolveNativeBinding()
const result = await streamImport(dbManager, entryPath, {
nativeBinding,
onProgress: (p) => sendEvent('progress', p),
})
if (result.success) {
sendEvent('done', {
success: true,
sessionId: result.sessionId,
messageCount: result.diagnostics?.messagesWritten ?? 0,
memberCount: 0,
})
} else {
sendEvent('error', { success: false, error: result.error })
}
} catch (err) {
if (!reply.raw.headersSent) {
return reply.code(500).send({ error: err instanceof Error ? err.message : String(err) })
}
reply.raw.write(
`event: error\ndata: ${JSON.stringify({ success: false, error: err instanceof Error ? err.message : String(err) })}\n\n`
)
} finally {
reply.raw.end()
cleanupTemp(tmpDir)
}
})
// ==================== Incremental Import ====================
server.post<{ Params: { id: string } }>('/_web/sessions/:id/import/incremental', async (request, reply) => {
const sessionId = request.params.id
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-inc-'))
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))
reply.raw.writeHead(200, {
'Content-Type': 'text/event-stream',
'Cache-Control': 'no-cache',
Connection: 'keep-alive',
})
function sendEvent(event: string, eventData: unknown) {
reply.raw.write(`event: ${event}\ndata: ${JSON.stringify(eventData)}\n\n`)
}
try {
const result = await incrementalImport(dbManager, sessionId, tmpPath, {
onProgress: (p) => sendEvent('progress', p),
})
if (result.success) {
sendEvent('done', result)
} else {
sendEvent('error', { success: false, error: result.error })
}
} catch (err) {
sendEvent('error', { success: false, error: err instanceof Error ? err.message : String(err) })
} finally {
reply.raw.end()
cleanupTemp(tmpPath, tmpDir)
}
})
server.post<{ Params: { id: string } }>('/_web/sessions/:id/import/incremental/analyze', async (request, reply) => {
const sessionId = request.params.id
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-analyze-'))
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 {
return await analyzeIncrementalImport(dbManager, sessionId, tmpPath)
} catch (err) {
return reply.code(500).send({ error: err instanceof Error ? err.message : String(err) })
} finally {
cleanupTemp(tmpPath, tmpDir)
}
})
// ==================== Analyze New Import (dry-run) ====================
server.post('/_web/import/analyze', 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-analyze-'))
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 {
return await analyzeNewImport(tmpPath)
} catch (err) {
return reply.code(500).send({ error: err instanceof Error ? err.message : String(err) })
} finally {
cleanupTemp(tmpPath, tmpDir)
}
})
// ==================== Demo Import ====================
server.post<{ Body: { locale?: string } }>('/_web/demo/import', async (request, reply) => {
const locale = (request.body as any)?.locale || 'en'
const nativeBinding = resolveNativeBinding()
reply.raw.writeHead(200, {
'Content-Type': 'text/event-stream',
'Cache-Control': 'no-cache',
Connection: 'keep-alive',
})
function sendEvent(event: string, eventData: unknown) {
reply.raw.write(`event: ${event}\ndata: ${JSON.stringify(eventData)}\n\n`)
}
const tmpDir = fs.mkdtempSync(path.join(os.tmpdir(), 'chatlab-demo-'))
const groupPath = path.join(tmpDir, 'demo-group.json')
const privatePath = path.join(tmpDir, 'demo-private.json')
try {
sendEvent('progress', { stage: 'downloading', current: 1, total: 2 })
const groupResp = await fetch(`${DEMO_BASE_URL}/${locale}/demo-group.json`, {
signal: AbortSignal.timeout(60_000),
})
if (!groupResp.ok) throw new Error(`Download group demo failed: ${groupResp.status}`)
fs.writeFileSync(groupPath, Buffer.from(await groupResp.arrayBuffer()))
sendEvent('progress', { stage: 'downloading', current: 2, total: 2 })
const privateResp = await fetch(`${DEMO_BASE_URL}/${locale}/demo-private.json`, {
signal: AbortSignal.timeout(60_000),
})
if (!privateResp.ok) throw new Error(`Download private demo failed: ${privateResp.status}`)
fs.writeFileSync(privatePath, Buffer.from(await privateResp.arrayBuffer()))
sendEvent('progress', { stage: 'importing', current: 1, total: 2 })
const groupResult = await streamImport(dbManager, groupPath, { nativeBinding })
if (!groupResult.success) throw new Error(groupResult.error || 'Failed to import group demo')
sendEvent('progress', { stage: 'importing', current: 2, total: 2 })
const privateResult = await streamImport(dbManager, privatePath, { nativeBinding })
if (!privateResult.success) throw new Error(privateResult.error || 'Failed to import private demo')
sendEvent('progress', { stage: 'done', current: 2, total: 2 })
sendEvent('result', {
success: true,
groupSessionId: groupResult.sessionId,
privateSessionId: privateResult.sessionId,
})
} catch (err) {
const message = err instanceof Error ? err.message : String(err)
sendEvent('progress', { stage: 'error', current: 0, total: 2, message })
sendEvent('result', { success: false, error: message })
} finally {
reply.raw.end()
cleanupTemp(tmpDir)
}
})
}
+63
View File
@@ -0,0 +1,63 @@
/**
* ChatLab Internal Web API — /_web/ routes
*
* 供 CLI serve Web 前端使用的内部 API(无认证、UI 友好的响应格式)。
* 数据格式直接对齐 QueryAdapter 接口,避免前端二次转换。
*
* Route modules:
* sessions – Session CRUD
* members – Member management
* analytics – Stats and advanced analytics
* sql – SQL Lab and plugin query
* sessionIndex – Session index generation + FTS
* summaries – LLM summary generation
* import – File / directory / incremental import + demo
* merge – Merge parse / conflicts / execute
* export – Markdown export
* cache – Save to downloads + show in folder
*/
import * as os from 'os'
import * as path from 'path'
import type { FastifyInstance } from 'fastify'
import type { DatabaseManager } from '@openchatlab/node-runtime'
import { createDatabaseManagerAdapter } from '@openchatlab/node-runtime'
import { MergeSessionCache } from '../../../merger/merge-cache'
import { registerSessionRoutes } from './sessions'
import { registerMemberRoutes } from './members'
import { registerAnalyticsRoutes } from './analytics'
import { registerSqlRoutes } from './sql'
import { registerSessionIndexRoutes } from './session-index'
import { registerSummaryRoutes } from './summaries'
import { registerImportRoutes } from './import'
import { registerMergeRoutes } from './merge'
import { registerExportRoutes } from './export'
import { registerCacheRoutes } from './cache'
export function registerWebRoutes(
server: FastifyInstance,
dbManager: DatabaseManager,
options?: { pathProvider?: import('@openchatlab/core').PathProvider; nativeBinding?: string }
): void {
const adapter = createDatabaseManagerAdapter(dbManager)
const mergeCache = options?.pathProvider
? new MergeSessionCache(options.pathProvider, { nativeBinding: options.nativeBinding })
: null
mergeCache?.cleanupOrphans()
const downloadsDir = options?.pathProvider?.getDownloadsDir() || path.join(os.homedir(), 'Downloads')
registerSessionRoutes(server, adapter)
registerMemberRoutes(server, adapter)
registerAnalyticsRoutes(server, dbManager, adapter)
registerSqlRoutes(server, adapter)
registerSessionIndexRoutes(server, adapter)
registerSummaryRoutes(server, dbManager, adapter)
registerImportRoutes(server, dbManager)
if (mergeCache) {
registerMergeRoutes(server, dbManager, mergeCache)
}
registerExportRoutes(server, adapter)
registerCacheRoutes(server, downloadsDir)
}
+55
View File
@@ -0,0 +1,55 @@
import type { FastifyInstance } from 'fastify'
import { memberService, type SessionRuntimeAdapter } from '@openchatlab/node-runtime'
export function registerMemberRoutes(server: FastifyInstance, adapter: SessionRuntimeAdapter): void {
server.get<{ Params: { id: string } }>('/_web/sessions/:id/members', async (request) => {
return memberService.getMembers(adapter, request.params.id)
})
server.get<{
Params: { id: string }
Querystring: { page?: string; pageSize?: string; search?: string; sortOrder?: string }
}>('/_web/sessions/:id/members/paginated', async (request) => {
return memberService.getMembersPaginated(adapter, request.params.id, {
page: parseInt(request.query.page || '1', 10),
pageSize: parseInt(request.query.pageSize || '20', 10),
search: request.query.search,
sortOrder: request.query.sortOrder === 'asc' ? 'asc' : 'desc',
})
})
server.patch<{ Params: { id: string; memberId: string }; Body: { aliases: string[] } }>(
'/_web/sessions/:id/members/:memberId/aliases',
async (request) => {
const memberId = parseInt(request.params.memberId, 10)
memberService.updateMemberAliases(adapter, request.params.id, memberId, request.body.aliases)
return { success: true }
}
)
server.delete<{ Params: { id: string; memberId: string } }>(
'/_web/sessions/:id/members/:memberId',
async (request) => {
const memberId = parseInt(request.params.memberId, 10)
memberService.deleteMember(adapter, request.params.id, memberId)
return { success: true }
}
)
server.post<{ Params: { id: string }; Body: { memberId1: number; memberId2: number } }>(
'/_web/sessions/:id/members/merge',
async (request) => {
const { memberId1, memberId2 } = request.body
memberService.mergeMembers(adapter, request.params.id, memberId1, memberId2)
return { success: true }
}
)
server.get<{ Params: { id: string; memberId: string } }>(
'/_web/sessions/:id/members/:memberId/history',
async (request) => {
const memberId = parseInt(request.params.memberId, 10)
return memberService.getMemberNameHistory(adapter, request.params.id, memberId)
}
)
}
+216
View File
@@ -0,0 +1,216 @@
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 }
})
}
@@ -0,0 +1,41 @@
import type { FastifyInstance } from 'fastify'
import { sessionIndexService, type SessionRuntimeAdapter } from '@openchatlab/node-runtime'
export function registerSessionIndexRoutes(server: FastifyInstance, adapter: SessionRuntimeAdapter): void {
server.post<{
Params: { id: string }
Body: { gapThreshold?: number }
}>('/_web/sessions/:id/generate-index', async (request) => {
const gapThreshold = (request.body as any)?.gapThreshold ?? 1800
const sessionCount = sessionIndexService.generateIndex(adapter, request.params.id, gapThreshold)
return { sessionCount }
})
server.post<{ Params: { id: string } }>('/_web/sessions/:id/clear-index', async (request) => {
sessionIndexService.clearIndex(adapter, request.params.id)
return { success: true }
})
server.get<{
Params: { id: string }
Querystring: { keywords: string; limit?: string; offset?: string }
}>('/_web/sessions/:id/search/fts', async (request, reply) => {
if (!sessionIndexService.getFtsStatus(adapter, request.params.id)) {
return reply.code(400).send({ error: 'FTS index not built for this session' })
}
const keywords = request.query.keywords.split(/\s+/).filter(Boolean)
if (keywords.length === 0) return { rowids: [], total: 0 }
const limit = parseInt(request.query.limit || '100', 10)
const offset = parseInt(request.query.offset || '0', 10)
return sessionIndexService.searchFts(adapter, request.params.id, keywords, limit, offset)
})
server.get<{ Params: { id: string } }>('/_web/sessions/:id/fts/status', async (request) => {
return { hasFtsIndex: sessionIndexService.getFtsStatus(adapter, request.params.id) }
})
server.post<{ Params: { id: string } }>('/_web/sessions/:id/fts/rebuild', async (request) => {
const result = sessionIndexService.rebuildFts(adapter, request.params.id)
return { success: true, indexed: result.indexed }
})
}
+38
View File
@@ -0,0 +1,38 @@
import type { FastifyInstance } from 'fastify'
import { sessionService, type SessionRuntimeAdapter } from '@openchatlab/node-runtime'
export function registerSessionRoutes(server: FastifyInstance, adapter: SessionRuntimeAdapter): void {
server.get('/_web/sessions', async () => {
return sessionService.listAnalysisSessions(adapter)
})
server.get<{ Params: { id: string } }>('/_web/sessions/:id', async (request) => {
return sessionService.getAnalysisSession(adapter, request.params.id)
})
server.delete<{ Params: { id: string } }>('/_web/sessions/:id', async (request, reply) => {
const { id } = request.params
try {
const deleted = sessionService.deleteSession(adapter, id)
if (!deleted) {
return reply.code(404).send({ success: false, error: 'File not found' })
}
return { success: true }
} catch (err) {
return reply.code(500).send({ success: false, error: String(err) })
}
})
server.patch<{ Params: { id: string }; Body: { name: string } }>('/_web/sessions/:id/name', async (request) => {
sessionService.renameSession(adapter, request.params.id, request.body.name)
return { success: true }
})
server.patch<{ Params: { id: string }; Body: { ownerId: string | null } }>(
'/_web/sessions/:id/owner',
async (request) => {
sessionService.updateSessionOwnerId(adapter, request.params.id, request.body.ownerId ?? null)
return { success: true }
}
)
}
+47
View File
@@ -0,0 +1,47 @@
import type { FastifyInstance } from 'fastify'
import type { SessionRuntimeAdapter } from '@openchatlab/node-runtime'
import { executeSql, getSchemaDetailed } from '@openchatlab/core'
export function registerSqlRoutes(server: FastifyInstance, adapter: SessionRuntimeAdapter): void {
server.post<{ Params: { id: string }; Body: { sql: string } }>('/_web/sessions/:id/sql', async (request, reply) => {
const db = adapter.ensureReadonly(request.params.id)
const { sql } = request.body || {}
if (!sql || typeof sql !== 'string') {
return reply.code(400).send({ error: 'Missing sql parameter' })
}
try {
return executeSql(db, sql, { timing: true })
} catch (err: unknown) {
const message = err instanceof Error ? err.message : 'SQL execution error'
return reply.code(400).send({ error: message })
}
})
server.get<{ Params: { id: string } }>('/_web/sessions/:id/schema', async (request) => {
const db = adapter.ensureReadonly(request.params.id)
return getSchemaDetailed(db)
})
server.post<{
Params: { id: string }
Body: { sql: string; params?: unknown[] | Record<string, unknown> }
}>('/_web/sessions/:id/query', async (request) => {
const db = adapter.ensureReadonly(request.params.id)
const { sql, params = [] } = request.body as { sql: string; params?: unknown[] | Record<string, unknown> }
if (!sql || typeof sql !== 'string') {
throw Object.assign(new Error('Missing or invalid "sql" field'), { statusCode: 400 })
}
const stmt = db.prepare(sql.trim())
if (!stmt.readonly) {
throw Object.assign(new Error('Only READ-ONLY statements are allowed'), { statusCode: 403 })
}
if (Array.isArray(params)) {
return stmt.all(...params)
}
return stmt.all(params)
})
}
+55
View File
@@ -0,0 +1,55 @@
import type { FastifyInstance } from 'fastify'
import type { DatabaseManager, SessionRuntimeAdapter, SummaryServiceDeps } from '@openchatlab/node-runtime'
import { summaryService } from '@openchatlab/node-runtime'
import { getAiDataDir } from './helpers'
import { getDefaultAssistantConfig, buildPiModel } from '../../../ai/llm-config'
function createSummaryDeps(dbManager: DatabaseManager): SummaryServiceDeps {
const aiDataDir = getAiDataDir(dbManager)
return {
getLlmConfig() {
return getDefaultAssistantConfig(aiDataDir)
},
buildPiModel(config) {
return buildPiModel(config as ReturnType<typeof getDefaultAssistantConfig> & object)
},
}
}
export function registerSummaryRoutes(
server: FastifyInstance,
dbManager: DatabaseManager,
adapter: SessionRuntimeAdapter
): void {
const deps = createSummaryDeps(dbManager)
server.post<{
Params: { id: string }
Body: { chatSessionId: number; locale?: string; forceRegenerate?: boolean }
}>('/_web/sessions/:id/summaries/generate', async (request, reply) => {
const { chatSessionId, locale, forceRegenerate } = request.body
const result = await summaryService.generateSummary(adapter, request.params.id, chatSessionId, deps, {
locale,
forceRegenerate,
})
if ('error' in result && !result.success) {
return reply.code(400).send({ error: result.error })
}
return result
})
server.post<{
Params: { id: string }
Body: { locale?: string; forceRegenerate?: boolean }
}>('/_web/sessions/:id/summaries/generate-all', async (request, reply) => {
const { locale, forceRegenerate } = request.body
const result = await summaryService.generateAllSummaries(adapter, request.params.id, deps, {
locale,
forceRegenerate,
})
if (result.error) {
return reply.code(400).send({ error: result.error })
}
return result
})
}