diff --git a/AGENTS.md b/AGENTS.md index 3f6e777f..a5ab04cd 100644 --- a/AGENTS.md +++ b/AGENTS.md @@ -4,4 +4,5 @@ - 每次完成任务后，对产生修改的文件进行类型检查、lint检查和format格式化，指定修改的文件路径去执行，确保代码质量 - 在执行检查时，如果有其他与本次修改无关的报错，也需要一并修复 - 多语言：代码中的日志、注释、AI 工具描述、错误消息等非 UI 文本默认使用英文。当有运行时 locale 可用时（如工具返回结果、AI 看到的文本），应通过 `isChineseLocale(locale)` 等机制支持中英双语。数据清洗中与聊天平台格式匹配的标签（如 `[分享]`、`[图片]`）保持原始语言不变。UI 文案的国际化遵循 `.docs/rules/i18n.md` +- 多端复用：维护 Electron 和 CLI Web 的共享业务逻辑时，优先在 `packages/node-runtime/src/services/` 下实现，禁止在路由/IPC handler 中绕过 core 直接写 SQL。详见 `.docs/README.md` 的"多端逻辑复用"章节。 - Commit 规范：使用 Conventional Commits。scope 规则——通用改动 scope 随意（如 `ai`、`import`、`sidebar` 等模块名）；仅当改动是**平台特有**时才使用平台 scope（`electron`、`cli`、`web`）。 diff --git a/apps/cli/src/http/routes/web.ts b/apps/cli/src/http/routes/web.ts deleted file mode 100644 index 2222dcb5..00000000 --- a/apps/cli/src/http/routes/web.ts +++ /dev/null @@ -1,1344 +0,0 @@ -/** - * ChatLab Internal Web API — /_web/ routes - * - * 供 CLI serve Web 前端使用的内部 API（无认证、UI 友好的响应格式）。 - * 数据格式直接对齐 QueryAdapter 接口，避免前端二次转换。 - */ - -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 type { TimeFilter } from '@openchatlab/shared-types' -import { - getSessionInfo, - getMembersWithAliases, - getMembersPaginated, - getTimeRange, - getAvailableYears, - getMemberActivity, - getHourlyActivity, - getDailyActivity, - getWeekdayActivity, - getMessageTypeStats, - getMonthlyActivity, - getYearlyActivity, - getMessageLengthDistribution, - executeSql, - getSchemaDetailed, - getRelationshipStats, - getCatchphraseAnalysis, - getMentionAnalysis, - getMentionGraph, - getLaughAnalysis, - getClusterGraph, - getLanguagePreferenceAnalysis, - generateSessionIndex, - clearSessionIndex, -} from '@openchatlab/core' -import { - createJiebaNlpProvider, - hasFtsTable, - searchByFts, - rebuildFtsIndex, - completeSimple, - type PiTextContent, - generateSessionSummary, - type SummaryDeps, - exportFilterResultToMarkdown, - streamParseFileInfo, - checkConflictsFromSources, - buildMergedOutput, - serializeChatLabToJsonl, - exportSessionToJson, - TempDbReader, -} from '@openchatlab/node-runtime' -import { MergeSessionCache } from '../../merger/merge-cache' -import { - getChatSessionSummary, - saveChatSessionSummary, - getChatSessionList, - getSessionMessages, -} from '@openchatlab/core' -import { - streamImport, - incrementalImport, - analyzeIncrementalImport, - analyzeNewImport, - detectFormat, - detectAllFormats, - getSupportedFormats, - scanMultiChatFile, - findEntryFileInDirectory, -} from '../../import' -import { getDefaultAssistantConfig, buildPiModel } from '../../ai/llm-config' - -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 -} - -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 -} - -function ensureWritableDb(dbManager: DatabaseManager, sessionId: string) { - dbManager.close(sessionId) - const db = dbManager.open(sessionId, { readonly: false }) - if (!db) { - throw Object.assign(new Error(`Session not found: ${sessionId}`), { statusCode: 404 }) - } - return db -} - -function parseTimeFilter(query: Record): 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 -} - -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() -} - -function buildSummaryDeps( - db: ReturnType, - llmConfig: ReturnType & object, - _aiDataDir: string -): SummaryDeps { - const piModel = buildPiModel(llmConfig) - return { - loadMessages(chatSessionId, limit = 500) { - const data = getSessionMessages(db!, chatSessionId, limit) - if (!data) return null - return data.messages.map((m) => ({ senderName: m.senderName, content: m.content })) - }, - saveSummary(chatSessionId, summary) { - saveChatSessionSummary(db!, chatSessionId, summary) - }, - getSummary(chatSessionId) { - return getChatSessionSummary(db!, chatSessionId) - }, - async llmComplete(systemPrompt, userPrompt, options) { - const result = await completeSimple( - piModel, - { - systemPrompt, - messages: [{ role: 'user', content: [{ type: 'text', text: userPrompt }], timestamp: Date.now() }] as any, - }, - { apiKey: llmConfig.apiKey, maxTokens: options?.maxTokens, temperature: options?.temperature } - ) - return result.content - .filter((item): item is PiTextContent => item.type === 'text') - .map((item) => item.text) - .join('') - }, - t: (key: string) => key, - } -} - -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 () => { - const sessionIds = dbManager.listSessionIds() - return sessionIds - .map((id) => { - const db = dbManager.open(id) - if (!db) return null - const info = getSessionInfo(db) - if (!info) return null - return { - ...info, - id, - dbPath: '', - memberAvatar: null, - aiConversationCount: 0, - } - }) - .filter(Boolean) - }) - - server.get<{ Params: { id: string } }>('/_web/sessions/:id', async (request) => { - const db = ensureDb(dbManager, request.params.id) - const info = getSessionInfo(db) - if (!info) return null - return { - ...info, - id: request.params.id, - dbPath: '', - memberAvatar: null, - aiConversationCount: 0, - } - }) - - server.delete<{ Params: { id: string } }>('/_web/sessions/:id', async (request, reply) => { - const { id } = request.params - try { - dbManager.close(id) - const dbPath = dbManager.getDbPath(id) - const fs = await import('fs') - if (fs.existsSync(dbPath)) { - fs.unlinkSync(dbPath) - return { success: true } - } - return reply.code(404).send({ success: false, error: 'File not found' }) - } 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) => { - const db = ensureDb(dbManager, request.params.id) - const { name } = request.body - db.prepare('UPDATE meta SET name = ?').run(name) - return { success: true } - }) - - server.patch<{ Params: { id: string }; Body: { ownerId: string | null } }>( - '/_web/sessions/:id/owner', - async (request) => { - const db = ensureDb(dbManager, request.params.id) - const { ownerId } = request.body - db.prepare('UPDATE meta SET owner_id = ?').run(ownerId ?? null) - return { success: true } - } - ) - - // ==================== 时间范围 ==================== - - server.get<{ Params: { id: string } }>('/_web/sessions/:id/years', async (request) => { - const db = ensureDb(dbManager, request.params.id) - return getAvailableYears(db) - }) - - server.get<{ Params: { id: string } }>('/_web/sessions/:id/time-range', async (request) => { - const db = ensureDb(dbManager, request.params.id) - return getTimeRange(db) - }) - - // ==================== 统计分析 ==================== - - server.get<{ - Params: { id: string } - Querystring: { startTs?: string; endTs?: string; memberId?: string } - }>('/_web/sessions/:id/stats/member-activity', async (request) => { - const db = ensureDb(dbManager, request.params.id) - const filter = parseTimeFilter(request.query) - return getMemberActivity(db, filter) - }) - - server.get<{ - Params: { id: string } - Querystring: { startTs?: string; endTs?: string; memberId?: string } - }>('/_web/sessions/:id/stats/hourly', async (request) => { - const db = ensureDb(dbManager, request.params.id) - const filter = parseTimeFilter(request.query) - return getHourlyActivity(db, filter) - }) - - server.get<{ - Params: { id: string } - Querystring: { startTs?: string; endTs?: string; memberId?: string } - }>('/_web/sessions/:id/stats/daily', async (request) => { - const db = ensureDb(dbManager, request.params.id) - const filter = parseTimeFilter(request.query) - return getDailyActivity(db, filter) - }) - - server.get<{ - Params: { id: string } - Querystring: { startTs?: string; endTs?: string; memberId?: string } - }>('/_web/sessions/:id/stats/weekday', async (request) => { - const db = ensureDb(dbManager, request.params.id) - const filter = parseTimeFilter(request.query) - return getWeekdayActivity(db, filter) - }) - - server.get<{ - Params: { id: string } - Querystring: { startTs?: string; endTs?: string; memberId?: string } - }>('/_web/sessions/:id/stats/message-types', async (request) => { - const db = ensureDb(dbManager, request.params.id) - const filter = parseTimeFilter(request.query) - return getMessageTypeStats(db, filter) - }) - - // ==================== 成员管理 ==================== - - server.get<{ Params: { id: string } }>('/_web/sessions/:id/members', async (request) => { - const db = ensureDb(dbManager, request.params.id) - return getMembersWithAliases(db) - }) - - server.get<{ - Params: { id: string } - Querystring: { page?: string; pageSize?: string; search?: string; sortOrder?: string } - }>('/_web/sessions/:id/members/paginated', async (request) => { - const db = ensureDb(dbManager, request.params.id) - const result = getMembersPaginated(db, { - 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', - }) - return { - items: result.members, - total: result.total, - page: result.page, - pageSize: result.pageSize, - totalPages: result.totalPages, - } - }) - - server.patch<{ Params: { id: string; memberId: string }; Body: { aliases: string[] } }>( - '/_web/sessions/:id/members/:memberId/aliases', - async (request) => { - const db = ensureDb(dbManager, request.params.id) - const memberId = parseInt(request.params.memberId, 10) - const { aliases } = request.body - db.prepare('UPDATE member SET aliases = ? WHERE id = ?').run(JSON.stringify(aliases), memberId) - return { success: true } - } - ) - - server.delete<{ Params: { id: string; memberId: string } }>( - '/_web/sessions/:id/members/:memberId', - async (request) => { - const db = ensureDb(dbManager, request.params.id) - const memberId = parseInt(request.params.memberId, 10) - db.prepare('DELETE FROM message WHERE sender_id = ?').run(memberId) - db.prepare('DELETE FROM member WHERE id = ?').run(memberId) - return { success: true } - } - ) - - server.post<{ Params: { id: string }; Body: { memberId1: number; memberId2: number } }>( - '/_web/sessions/:id/members/merge', - async (request) => { - const db = ensureDb(dbManager, request.params.id) - const { memberId1, memberId2 } = request.body - db.prepare('UPDATE message SET sender_id = ? WHERE sender_id = ?').run(memberId1, memberId2) - db.prepare('DELETE FROM member WHERE id = ?').run(memberId2) - return { success: true } - } - ) - - server.get<{ Params: { id: string; memberId: string } }>( - '/_web/sessions/:id/members/:memberId/history', - async (request) => { - const db = ensureDb(dbManager, request.params.id) - const memberId = parseInt(request.params.memberId, 10) - - const rows = db - .prepare( - `SELECT - sender_account_name as accountName, - sender_group_nickname as groupNickname, - MIN(ts) as startTs, - MAX(ts) as endTs - FROM message - WHERE sender_id = ? - GROUP BY sender_account_name, sender_group_nickname - ORDER BY startTs` - ) - .all(memberId) as any[] - - const history: Array<{ nameType: string; name: string; startTs: number; endTs: number | null }> = [] - for (const row of rows) { - if (row.accountName) { - history.push({ nameType: 'account_name', name: row.accountName, startTs: row.startTs, endTs: row.endTs }) - } - if (row.groupNickname) { - history.push({ nameType: 'group_nickname', name: row.groupNickname, startTs: row.startTs, endTs: row.endTs }) - } - } - return history - } - ) - - // ==================== 高级分析（analytics） ==================== - - server.get<{ - Params: { id: string } - Querystring: { startTs?: string; endTs?: string; memberId?: string } - }>('/_web/sessions/:id/analytics/relationship', async (request) => { - const db = ensureDb(dbManager, request.params.id) - const filter = parseTimeFilter(request.query) - return getRelationshipStats(db, filter) - }) - - server.get<{ - Params: { id: string } - Querystring: { startTs?: string; endTs?: string; memberId?: string } - }>('/_web/sessions/:id/analytics/catchphrase', async (request) => { - const db = ensureDb(dbManager, request.params.id) - const filter = parseTimeFilter(request.query) - return getCatchphraseAnalysis(db, filter) - }) - - server.get<{ - Params: { id: string } - Querystring: { startTs?: string; endTs?: string; memberId?: string } - }>('/_web/sessions/:id/analytics/mention', async (request) => { - const db = ensureDb(dbManager, request.params.id) - const filter = parseTimeFilter(request.query) - return getMentionAnalysis(db, filter) - }) - - server.get<{ - Params: { id: string } - Querystring: { startTs?: string; endTs?: string; memberId?: string } - }>('/_web/sessions/:id/analytics/mention-graph', async (request) => { - const db = ensureDb(dbManager, request.params.id) - const filter = parseTimeFilter(request.query) - return getMentionGraph(db, filter) - }) - - server.get<{ - Params: { id: string } - Querystring: { startTs?: string; endTs?: string; memberId?: string } - }>('/_web/sessions/:id/analytics/laugh', async (request) => { - const db = ensureDb(dbManager, request.params.id) - const filter = parseTimeFilter(request.query) - return getLaughAnalysis(db, filter) - }) - - server.get<{ - Params: { id: string } - Querystring: { startTs?: string; endTs?: string; memberId?: string; topEdges?: string } - }>('/_web/sessions/:id/analytics/cluster', async (request) => { - const db = ensureDb(dbManager, 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: { startTs?: string; endTs?: string; memberId?: string; locale?: string } - }>('/_web/sessions/:id/analytics/language-preference', async (request) => { - const db = ensureDb(dbManager, 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: { startTs?: string; endTs?: string; memberId?: string } - }>('/_web/sessions/:id/analytics/monthly-activity', async (request) => { - const db = ensureDb(dbManager, request.params.id) - const filter = parseTimeFilter(request.query) - return getMonthlyActivity(db, filter) - }) - - server.get<{ - Params: { id: string } - Querystring: { startTs?: string; endTs?: string; memberId?: string } - }>('/_web/sessions/:id/analytics/yearly-activity', async (request) => { - const db = ensureDb(dbManager, request.params.id) - const filter = parseTimeFilter(request.query) - return getYearlyActivity(db, filter) - }) - - server.get<{ - Params: { id: string } - Querystring: { startTs?: string; endTs?: string; memberId?: string } - }>('/_web/sessions/:id/analytics/message-length-distribution', async (request) => { - const db = ensureDb(dbManager, request.params.id) - const filter = parseTimeFilter(request.query) - return getMessageLengthDistribution(db, filter) - }) - - // ==================== SQL Lab ==================== - - server.post<{ Params: { id: string }; Body: { sql: string } }>('/_web/sessions/:id/sql', async (request, reply) => { - const db = ensureDb(dbManager, 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 = ensureDb(dbManager, request.params.id) - return getSchemaDetailed(db) - }) - - // ==================== 插件查询（参数化只读 SQL） ==================== - - server.post<{ - Params: { id: string } - Body: { sql: string; params?: unknown[] | Record } - }>('/_web/sessions/:id/query', async (request) => { - const db = ensureDb(dbManager, request.params.id) - const { sql, params = [] } = request.body as { sql: string; params?: unknown[] | Record } - - 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) - }) - - // ==================== 导出 ==================== - - 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 chunks: string[] = [] - const result = exportFilterResultToMarkdown( - { - sessionId: id, - sessionName, - filterMode: body.filterMode || 'condition', - keywords: body.keywords, - timeFilter: body.timeFilter, - senderIds: body.senderIds, - contextSize: body.contextSize, - chatSessionIds: body.chatSessionIds, - }, - { - openDatabase(sessionId: string) { - return dbManager.open(sessionId) ?? null - }, - }, - { - write(chunk: string) { - chunks.push(chunk) - }, - end() { - /* collected in chunks array */ - }, - } - ) - - 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(chunks.join('')) - }) - - // ==================== 会话索引 ==================== - - server.post<{ - Params: { id: string } - Body: { gapThreshold?: number } - }>('/_web/sessions/:id/generate-index', async (request) => { - const db = ensureWritableDb(dbManager, request.params.id) - const gapThreshold = (request.body as any)?.gapThreshold ?? 1800 - const sessionCount = generateSessionIndex(db, gapThreshold) - return { sessionCount } - }) - - server.post<{ Params: { id: string } }>('/_web/sessions/:id/clear-index', async (request) => { - const db = ensureWritableDb(dbManager, request.params.id) - clearSessionIndex(db) - return { success: true } - }) - - // ==================== FTS Search & Rebuild ==================== - - server.get<{ - Params: { id: string } - Querystring: { keywords: string; limit?: string; offset?: string } - }>('/_web/sessions/:id/search/fts', async (request, reply) => { - const db = ensureDb(dbManager, request.params.id) - if (!hasFtsTable(db)) { - 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 searchByFts(db, keywords, limit, offset) - }) - - server.get<{ Params: { id: string } }>('/_web/sessions/:id/fts/status', async (request) => { - const db = ensureDb(dbManager, request.params.id) - return { hasFtsIndex: hasFtsTable(db) } - }) - - server.post<{ Params: { id: string } }>('/_web/sessions/:id/fts/rebuild', async (request) => { - const db = ensureWritableDb(dbManager, request.params.id) - const result = rebuildFtsIndex(db) - return { success: true, indexed: result.indexed } - }) - - // ==================== Session Summary (LLM-generated) ==================== - - server.post<{ - Params: { id: string } - Body: { chatSessionId: number; locale?: string; forceRegenerate?: boolean } - }>('/_web/sessions/:id/summaries/generate', async (request, reply) => { - const sessionId = request.params.id - const { chatSessionId, locale, forceRegenerate } = request.body - - const aiDataDir = getAiDataDir(dbManager) - const llmConfig = getDefaultAssistantConfig(aiDataDir) - if (!llmConfig) { - return reply.code(400).send({ error: 'No LLM configuration available' }) - } - - const db = ensureWritableDb(dbManager, sessionId) - const deps = buildSummaryDeps(db, llmConfig, aiDataDir) - - const result = await generateSessionSummary(deps, chatSessionId, { locale, forceRegenerate }) - return result - }) - - server.post<{ - Params: { id: string } - Body: { locale?: string; forceRegenerate?: boolean } - }>('/_web/sessions/:id/summaries/generate-all', async (request, reply) => { - const sessionId = request.params.id - const { locale, forceRegenerate } = request.body - - const aiDataDir = getAiDataDir(dbManager) - const llmConfig = getDefaultAssistantConfig(aiDataDir) - if (!llmConfig) { - return reply.code(400).send({ error: 'No LLM configuration available' }) - } - - const db = ensureWritableDb(dbManager, sessionId) - const chatSessions = getChatSessionList(db) - const deps = buildSummaryDeps(db, llmConfig, aiDataDir) - - let success = 0 - let failed = 0 - - for (const cs of chatSessions) { - const result = await generateSessionSummary(deps, cs.id, { locale, forceRegenerate }) - if (result.success) success++ - else failed++ - } - - return { success, failed, total: chatSessions.length } - }) - - // ==================== Demo 示例数据 ==================== - - const DEMO_BASE_URL = 'https://chatlab.fun/assets/demo' - - 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, data: unknown) { - reply.raw.write(`event: ${event}\ndata: ${JSON.stringify(data)}\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 { - try { - fs.unlinkSync(groupPath) - } catch { - /* ignore */ - } - try { - fs.unlinkSync(privatePath) - } catch { - /* ignore */ - } - try { - fs.rmdirSync(tmpDir) - } catch { - /* ignore */ - } - reply.raw.end() - } - }) - - // ==================== 导入管线 ==================== - - 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 { - try { - fs.unlinkSync(tmpPath) - } catch { - /* ignore */ - } - try { - fs.rmdirSync(tmpDir) - } catch { - /* ignore */ - } - } - }) - - 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 { - try { - fs.unlinkSync(tmpPath) - } catch { - /* ignore */ - } - try { - fs.rmdirSync(tmpDir) - } catch { - /* ignore */ - } - } - }) - - 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, data: unknown) { - reply.raw.write(`event: ${event}\ndata: ${JSON.stringify(data)}\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() - try { - fs.unlinkSync(tmpPath) - } catch { - /* ignore */ - } - try { - fs.rmdirSync(tmpDir) - } catch { - /* ignore */ - } - } - }) - - // ==================== 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-')) - - // Interleaved fields: relativePaths (field) and files (file) come in order - 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 || '' }) - } - } - - // Write files preserving directory structure - for (let i = 0; i < fileBuffers.length; i++) { - let relPath = relativePaths[i] || fileBuffers[i].filename || `file_${i}` - - // Strip top-level directory from webkitRelativePath (e.g. "export_dir/chunks/..." → "chunks/...") - 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() - fs.rmSync(tmpDir, { recursive: true, force: true }) - } - }) - - // ==================== 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() - try { - fs.unlinkSync(tmpPath) - } catch { - /* ignore */ - } - try { - fs.rmdirSync(tmpDir) - } catch { - /* ignore */ - } - } - }) - - 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 { - const result = await analyzeIncrementalImport(dbManager, sessionId, tmpPath) - return result - } catch (err) { - return reply.code(500).send({ error: err instanceof Error ? err.message : String(err) }) - } finally { - try { - fs.unlinkSync(tmpPath) - } catch { - /* ignore */ - } - try { - fs.rmdirSync(tmpDir) - } catch { - /* ignore */ - } - } - }) - - // ==================== 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 { - const result = await analyzeNewImport(tmpPath) - return result - } catch (err) { - return reply.code(500).send({ error: err instanceof Error ? err.message : String(err) }) - } finally { - try { - fs.unlinkSync(tmpPath) - } catch { - /* ignore */ - } - try { - fs.rmdirSync(tmpDir) - } catch { - /* ignore */ - } - } - }) - - // ==================== 合并功能 ==================== - - 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 } - }) - } - - // ==================== Cache / File Operations ==================== - - const downloadsDir = options?.pathProvider?.getDownloadsDir() || path.join(os.homedir(), 'Downloads') - - 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 } - }) -} diff --git a/apps/cli/src/http/routes/web/analytics.ts b/apps/cli/src/http/routes/web/analytics.ts new file mode 100644 index 00000000..d1abac3b --- /dev/null +++ b/apps/cli/src/http/routes/web/analytics.ts @@ -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)) + } + ) +} diff --git a/apps/cli/src/http/routes/web/cache.ts b/apps/cli/src/http/routes/web/cache.ts new file mode 100644 index 00000000..55a663e7 --- /dev/null +++ b/apps/cli/src/http/routes/web/cache.ts @@ -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 } + }) +} diff --git a/apps/cli/src/http/routes/web/export.ts b/apps/cli/src/http/routes/web/export.ts new file mode 100644 index 00000000..a90e95e0 --- /dev/null +++ b/apps/cli/src/http/routes/web/export.ts @@ -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) + }) +} diff --git a/apps/cli/src/http/routes/web/helpers.ts b/apps/cli/src/http/routes/web/helpers.ts new file mode 100644 index 00000000..f8cfcb68 --- /dev/null +++ b/apps/cli/src/http/routes/web/helpers.ts @@ -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): 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 +} diff --git a/apps/cli/src/http/routes/web/import.ts b/apps/cli/src/http/routes/web/import.ts new file mode 100644 index 00000000..724cc8de --- /dev/null +++ b/apps/cli/src/http/routes/web/import.ts @@ -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) + } + }) +} diff --git a/apps/cli/src/http/routes/web/index.ts b/apps/cli/src/http/routes/web/index.ts new file mode 100644 index 00000000..483ad191 --- /dev/null +++ b/apps/cli/src/http/routes/web/index.ts @@ -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) +} diff --git a/apps/cli/src/http/routes/web/members.ts b/apps/cli/src/http/routes/web/members.ts new file mode 100644 index 00000000..df5989ec --- /dev/null +++ b/apps/cli/src/http/routes/web/members.ts @@ -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) + } + ) +} diff --git a/apps/cli/src/http/routes/web/merge.ts b/apps/cli/src/http/routes/web/merge.ts new file mode 100644 index 00000000..83d47c29 --- /dev/null +++ b/apps/cli/src/http/routes/web/merge.ts @@ -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 } + }) +} diff --git a/apps/cli/src/http/routes/web/session-index.ts b/apps/cli/src/http/routes/web/session-index.ts new file mode 100644 index 00000000..e6bc6ce5 --- /dev/null +++ b/apps/cli/src/http/routes/web/session-index.ts @@ -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 } + }) +} diff --git a/apps/cli/src/http/routes/web/sessions.ts b/apps/cli/src/http/routes/web/sessions.ts new file mode 100644 index 00000000..b7f4340b --- /dev/null +++ b/apps/cli/src/http/routes/web/sessions.ts @@ -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 } + } + ) +} diff --git a/apps/cli/src/http/routes/web/sql.ts b/apps/cli/src/http/routes/web/sql.ts new file mode 100644 index 00000000..117d3202 --- /dev/null +++ b/apps/cli/src/http/routes/web/sql.ts @@ -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 } + }>('/_web/sessions/:id/query', async (request) => { + const db = adapter.ensureReadonly(request.params.id) + const { sql, params = [] } = request.body as { sql: string; params?: unknown[] | Record } + + 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) + }) +} diff --git a/apps/cli/src/http/routes/web/summaries.ts b/apps/cli/src/http/routes/web/summaries.ts new file mode 100644 index 00000000..4abb1c2a --- /dev/null +++ b/apps/cli/src/http/routes/web/summaries.ts @@ -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 & 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 + }) +} diff --git a/packages/node-runtime/src/index.ts b/packages/node-runtime/src/index.ts index 286dbedc..17fd1804 100644 --- a/packages/node-runtime/src/index.ts +++ b/packages/node-runtime/src/index.ts @@ -322,3 +322,19 @@ export type { SessionMessage, SessionInfo, } from './ai' + +// Shared application services (session / member / index / summary / export) +export * as sessionService from './services/session-service' +export * as memberService from './services/member-service' +export * as sessionIndexService from './services/session-index-service' +export * as summaryService from './services/summary-service' +export * as exportService from './services/export-service' +export { createDatabaseManagerAdapter } from './services' +export type { + SessionRuntimeAdapter, + AnalysisSessionDTO, + ListSessionsOptions, + MembersPaginatedDTO, + LlmConfig, + SummaryServiceDeps, +} from './services' diff --git a/packages/node-runtime/src/services/adapters/database-manager-adapter.ts b/packages/node-runtime/src/services/adapters/database-manager-adapter.ts new file mode 100644 index 00000000..61422c66 --- /dev/null +++ b/packages/node-runtime/src/services/adapters/database-manager-adapter.ts @@ -0,0 +1,45 @@ +/** + * DatabaseManager adapter for the shared service layer. + * + * Bridges DatabaseManager's connection lifecycle to the SessionRuntimeAdapter + * interface expected by session/member services. + */ + +import * as fs from 'fs' +import type { DatabaseAdapter } from '@openchatlab/core' +import type { DatabaseManager } from '../../database-manager' +import type { SessionRuntimeAdapter } from './types' + +export function createDatabaseManagerAdapter(dbManager: DatabaseManager): SessionRuntimeAdapter { + return { + listSessionIds: () => dbManager.listSessionIds(), + openReadonly: (sessionId) => dbManager.open(sessionId), + openWritable: (sessionId) => dbManager.openWritable(sessionId), + closeSession: (sessionId) => dbManager.close(sessionId), + getDbPath: (sessionId) => dbManager.getDbPath(sessionId), + + deleteSessionFile(sessionId: string): boolean { + dbManager.close(sessionId) + const dbPath = dbManager.getDbPath(sessionId) + if (!fs.existsSync(dbPath)) return false + fs.unlinkSync(dbPath) + return true + }, + + ensureReadonly(sessionId: string): DatabaseAdapter { + const db = dbManager.open(sessionId) + if (!db) { + throw Object.assign(new Error(`Session not found: ${sessionId}`), { statusCode: 404 }) + } + return db + }, + + ensureWritable(sessionId: string): DatabaseAdapter { + const db = dbManager.openWritable(sessionId) + if (!db) { + throw Object.assign(new Error(`Session not found: ${sessionId}`), { statusCode: 404 }) + } + return db + }, + } +} diff --git a/packages/node-runtime/src/services/adapters/index.ts b/packages/node-runtime/src/services/adapters/index.ts new file mode 100644 index 00000000..58459620 --- /dev/null +++ b/packages/node-runtime/src/services/adapters/index.ts @@ -0,0 +1,2 @@ +export type { SessionRuntimeAdapter } from './types' +export { createDatabaseManagerAdapter } from './database-manager-adapter' diff --git a/packages/node-runtime/src/services/adapters/types.ts b/packages/node-runtime/src/services/adapters/types.ts new file mode 100644 index 00000000..48a75fa9 --- /dev/null +++ b/packages/node-runtime/src/services/adapters/types.ts @@ -0,0 +1,25 @@ +/** + * Adapter interfaces for the shared service layer. + * + * Services depend on these interfaces instead of concrete DatabaseManager or + * Electron worker implementations — making them reusable across CLI Web and Electron. + */ + +import type { DatabaseAdapter } from '@openchatlab/core' + +export interface SessionRuntimeAdapter { + listSessionIds(): string[] + openReadonly(sessionId: string): DatabaseAdapter | null + openWritable(sessionId: string): DatabaseAdapter | null + closeSession(sessionId: string): void + getDbPath(sessionId: string): string + + /** Delete the session database file. Returns false if not found. */ + deleteSessionFile(sessionId: string): boolean + + /** Open readonly, throw 404 if not found. */ + ensureReadonly(sessionId: string): DatabaseAdapter + + /** Open writable (with auto-migration), throw 404 if not found. */ + ensureWritable(sessionId: string): DatabaseAdapter +} diff --git a/packages/node-runtime/src/services/export-service.ts b/packages/node-runtime/src/services/export-service.ts new file mode 100644 index 00000000..2ebdcea7 --- /dev/null +++ b/packages/node-runtime/src/services/export-service.ts @@ -0,0 +1,32 @@ +/** + * Shared export service. + * + * Wraps node-runtime's Markdown exporter with adapter-based DB opening. + */ + +import { exportFilterResultToMarkdown, type ExportFilterParams, type ExportResult } from '../export' +import type { SessionRuntimeAdapter } from './adapters' + +export function exportMarkdown( + adapter: SessionRuntimeAdapter, + params: ExportFilterParams +): { result: ExportResult; content: string } { + const chunks: string[] = [] + const result = exportFilterResultToMarkdown( + params, + { + openDatabase(sessionId: string) { + return adapter.openReadonly(sessionId) + }, + }, + { + write(chunk: string) { + chunks.push(chunk) + }, + end() { + /* collected in chunks array */ + }, + } + ) + return { result, content: chunks.join('') } +} diff --git a/packages/node-runtime/src/services/index.ts b/packages/node-runtime/src/services/index.ts new file mode 100644 index 00000000..992d04c5 --- /dev/null +++ b/packages/node-runtime/src/services/index.ts @@ -0,0 +1,34 @@ +// Adapters +export type { SessionRuntimeAdapter } from './adapters' +export { createDatabaseManagerAdapter } from './adapters' + +// Session service +export { + listAnalysisSessions, + getAnalysisSession, + renameSession, + updateSessionOwnerId, + deleteSession, +} from './session-service' +export type { AnalysisSessionDTO, ListSessionsOptions } from './session-service' + +// Member service +export { + getMembers, + getMembersPaginated, + updateMemberAliases, + mergeMembers, + deleteMember, + getMemberNameHistory, +} from './member-service' +export type { MembersPaginatedDTO } from './member-service' + +// Session index service +export { generateIndex, clearIndex, getFtsStatus, searchFts, rebuildFts } from './session-index-service' + +// Summary service +export { generateSummary, generateAllSummaries } from './summary-service' +export type { LlmConfig, SummaryServiceDeps } from './summary-service' + +// Export service +export { exportMarkdown } from './export-service' diff --git a/packages/node-runtime/src/services/member-service.ts b/packages/node-runtime/src/services/member-service.ts new file mode 100644 index 00000000..00eb913d --- /dev/null +++ b/packages/node-runtime/src/services/member-service.ts @@ -0,0 +1,105 @@ +/** + * Shared member service layer. + * + * Delegates to @openchatlab/core member-ops and query functions, + * ensuring CLI Web and Electron use identical business semantics + * (merge strategy, delete scope, name history source, etc.). + */ + +import { + getMembersWithAliases as coreGetMembersWithAliases, + getMembersPaginated as coreGetMembersPaginated, + getMemberNameHistory as coreGetMemberNameHistory, + updateMemberAliases as coreUpdateMemberAliases, + mergeMembers as coreMergeMembers, + deleteMember as coreDeleteMember, +} from '@openchatlab/core' +import type { MemberWithAliases, MembersPaginationParams, MembersPaginatedResult } from '@openchatlab/core' +import type { SessionRuntimeAdapter } from './adapters' + +export type { MemberWithAliases, MembersPaginationParams, MembersPaginatedResult } + +export interface MembersPaginatedDTO { + items: MemberWithAliases[] + total: number + page: number + pageSize: number + totalPages: number +} + +/** + * Get all members with aliases for a session. + */ +export function getMembers(adapter: SessionRuntimeAdapter, sessionId: string): MemberWithAliases[] { + const db = adapter.ensureReadonly(sessionId) + return coreGetMembersWithAliases(db) +} + +/** + * Get paginated member list with search and sort. + */ +export function getMembersPaginated( + adapter: SessionRuntimeAdapter, + sessionId: string, + params: MembersPaginationParams +): MembersPaginatedDTO { + const db = adapter.ensureReadonly(sessionId) + const result = coreGetMembersPaginated(db, params) + return { + items: result.members, + total: result.total, + page: result.page, + pageSize: result.pageSize, + totalPages: result.totalPages, + } +} + +/** + * Update member aliases. + * Uses core's updateMemberAliases which properly serializes the JSON array. + */ +export function updateMemberAliases( + adapter: SessionRuntimeAdapter, + sessionId: string, + memberId: number, + aliases: string[] +): boolean { + const db = adapter.ensureWritable(sessionId) + return coreUpdateMemberAliases(db, memberId, aliases) +} + +/** + * Merge two members using core semantics: + * - Primary determined by message count (higher wins, lower id on tie) + * - Merges aliases, avatar, account_name, group_nickname + * - Reassigns messages and member_name_history + * - Updates meta.owner_id if secondary was owner + * - Runs in a transaction + */ +export function mergeMembers( + adapter: SessionRuntimeAdapter, + sessionId: string, + memberId1: number, + memberId2: number +): boolean { + const db = adapter.ensureWritable(sessionId) + return coreMergeMembers(db, memberId1, memberId2) +} + +/** + * Delete a member and all associated data: + * - messages, member_name_history, member row (in transaction) + */ +export function deleteMember(adapter: SessionRuntimeAdapter, sessionId: string, memberId: number): boolean { + const db = adapter.ensureWritable(sessionId) + return coreDeleteMember(db, memberId) +} + +/** + * Get member name change history from the member_name_history table. + * Uses core's getMemberNameHistory (not the web.ts message-derived version). + */ +export function getMemberNameHistory(adapter: SessionRuntimeAdapter, sessionId: string, memberId: number) { + const db = adapter.ensureReadonly(sessionId) + return coreGetMemberNameHistory(db, memberId) +} diff --git a/packages/node-runtime/src/services/session-index-service.ts b/packages/node-runtime/src/services/session-index-service.ts new file mode 100644 index 00000000..96f45791 --- /dev/null +++ b/packages/node-runtime/src/services/session-index-service.ts @@ -0,0 +1,43 @@ +/** + * Shared session index service. + * + * Wraps core's session index operations with proper writable DB handling. + */ + +import { + generateSessionIndex as coreGenerateSessionIndex, + clearSessionIndex as coreClearSessionIndex, +} from '@openchatlab/core' +import { hasFtsTable, searchByFts, rebuildFtsIndex } from '../fts' +import type { SessionRuntimeAdapter } from './adapters' + +export function generateIndex(adapter: SessionRuntimeAdapter, sessionId: string, gapThreshold: number = 1800): number { + const db = adapter.ensureWritable(sessionId) + return coreGenerateSessionIndex(db, gapThreshold) +} + +export function clearIndex(adapter: SessionRuntimeAdapter, sessionId: string): void { + const db = adapter.ensureWritable(sessionId) + coreClearSessionIndex(db) +} + +export function getFtsStatus(adapter: SessionRuntimeAdapter, sessionId: string): boolean { + const db = adapter.ensureReadonly(sessionId) + return hasFtsTable(db) +} + +export function searchFts( + adapter: SessionRuntimeAdapter, + sessionId: string, + keywords: string[], + limit: number, + offset: number +) { + const db = adapter.ensureReadonly(sessionId) + return searchByFts(db, keywords, limit, offset) +} + +export function rebuildFts(adapter: SessionRuntimeAdapter, sessionId: string) { + const db = adapter.ensureWritable(sessionId) + return rebuildFtsIndex(db) +} diff --git a/packages/node-runtime/src/services/session-service.ts b/packages/node-runtime/src/services/session-service.ts new file mode 100644 index 00000000..52bc01d1 --- /dev/null +++ b/packages/node-runtime/src/services/session-service.ts @@ -0,0 +1,134 @@ +/** + * Shared session service layer. + * + * Provides unified session CRUD used by both CLI Web routes and Electron IPC/API. + * All business logic (sorting, data shape, validation) lives here — + * callers only handle protocol-specific concerns (HTTP params, IPC channels). + */ + +import { + getSessionInfo, + getSessionMeta, + getSummaryCount, + getPrivateChatMemberAvatar, + isChatSessionDb, + buildSessionInfo, + renameSession as coreRenameSession, + updateSessionOwnerId as coreUpdateSessionOwnerId, +} from '@openchatlab/core' +import type { CoreSessionInfo, DatabaseAdapter, SessionOverview } from '@openchatlab/core' +import type { SessionRuntimeAdapter } from './adapters' + +export interface AnalysisSessionDTO extends CoreSessionInfo { + id: string + dbPath: string + memberAvatar: string | null + aiConversationCount: number +} + +/** + * Optional hooks for platform-specific behavior. + * Electron can provide a cached overview resolver and post-list enrichment; + * CLI Web can omit these to use default (no-cache) behavior. + */ +export interface ListSessionsOptions { + /** + * Resolve overview stats for a session, optionally from cache. + * When provided, used instead of the default `getSessionOverview` SQL query. + * Electron worker uses this to read from JSON file cache (resolveOverview). + */ + resolveOverview?(db: DatabaseAdapter, sessionId: string): SessionOverview + + /** + * Enrich each session DTO after construction. + * Electron uses this to fill aiConversationCount, etc. + */ + enrichSession?(dto: AnalysisSessionDTO): AnalysisSessionDTO +} + +function buildSession( + db: DatabaseAdapter, + id: string, + dbPath: string, + options?: ListSessionsOptions +): AnalysisSessionDTO | null { + const meta = getSessionMeta(db) + if (!meta) return null + + const overview = options?.resolveOverview ? options.resolveOverview(db, id) : undefined + const info = overview ? buildSessionInfo(meta, overview, getSummaryCount(db)) : getSessionInfo(db) + if (!info) return null + + let memberAvatar: string | null = null + if (meta.type === 'private') { + memberAvatar = getPrivateChatMemberAvatar(db, meta.name, meta.ownerId) + } + + let dto: AnalysisSessionDTO = { ...info, id, dbPath, memberAvatar, aiConversationCount: 0 } + if (options?.enrichSession) { + dto = options.enrichSession(dto) + } + return dto +} + +/** + * List all valid sessions, sorted by importedAt descending. + * + * Accepts optional hooks so Electron can plug in cached overview resolution + * and session enrichment without duplicating the listing logic. + */ +export function listAnalysisSessions( + adapter: SessionRuntimeAdapter, + options?: ListSessionsOptions +): AnalysisSessionDTO[] { + const sessionIds = adapter.listSessionIds() + const sessions: AnalysisSessionDTO[] = [] + + for (const id of sessionIds) { + const db = adapter.openReadonly(id) + if (!db) continue + if (!isChatSessionDb(db)) continue + + const dto = buildSession(db, id, adapter.getDbPath(id), options) + if (dto) sessions.push(dto) + } + + return sessions.sort((a, b) => b.importedAt - a.importedAt) +} + +/** + * Get a single session by ID. + */ +export function getAnalysisSession( + adapter: SessionRuntimeAdapter, + sessionId: string, + options?: ListSessionsOptions +): AnalysisSessionDTO | null { + const db = adapter.openReadonly(sessionId) + if (!db) return null + return buildSession(db, sessionId, adapter.getDbPath(sessionId), options) +} + +/** + * Rename a session (updates meta.name). + */ +export function renameSession(adapter: SessionRuntimeAdapter, sessionId: string, name: string): void { + const db = adapter.ensureWritable(sessionId) + coreRenameSession(db, name) +} + +/** + * Update session owner_id. + */ +export function updateSessionOwnerId(adapter: SessionRuntimeAdapter, sessionId: string, ownerId: string | null): void { + const db = adapter.ensureWritable(sessionId) + coreUpdateSessionOwnerId(db, ownerId) +} + +/** + * Delete a session (close DB + remove file). + * Returns true if deleted, false if file not found. + */ +export function deleteSession(adapter: SessionRuntimeAdapter, sessionId: string): boolean { + return adapter.deleteSessionFile(sessionId) +} diff --git a/packages/node-runtime/src/services/summary-service.ts b/packages/node-runtime/src/services/summary-service.ts new file mode 100644 index 00000000..5f29731e --- /dev/null +++ b/packages/node-runtime/src/services/summary-service.ts @@ -0,0 +1,101 @@ +/** + * Shared summary service. + * + * Encapsulates LLM config loading, SummaryDeps construction, + * and summary generation loop — shared across CLI Web and Electron. + */ + +import { + getChatSessionSummary, + saveChatSessionSummary, + getChatSessionList, + getSessionMessages, +} from '@openchatlab/core' +import type { DatabaseAdapter } from '@openchatlab/core' +import { generateSessionSummary, type SummaryDeps, completeSimple, type PiTextContent } from '../ai' +import type { SessionRuntimeAdapter } from './adapters' + +export interface LlmConfig { + apiKey: string +} + +export interface SummaryServiceDeps { + getLlmConfig(): LlmConfig | null + buildPiModel(config: LlmConfig): ReturnType +} + +function buildSummaryDeps(db: DatabaseAdapter, llmConfig: LlmConfig, deps: SummaryServiceDeps): SummaryDeps { + const piModel = deps.buildPiModel(llmConfig) + return { + loadMessages(chatSessionId, limit = 500) { + const data = getSessionMessages(db, chatSessionId, limit) + if (!data) return null + return data.messages.map((m) => ({ senderName: m.senderName, content: m.content })) + }, + saveSummary(chatSessionId, summary) { + saveChatSessionSummary(db, chatSessionId, summary) + }, + getSummary(chatSessionId) { + return getChatSessionSummary(db, chatSessionId) + }, + async llmComplete(systemPrompt, userPrompt, options) { + const result = await completeSimple( + piModel, + { + systemPrompt, + messages: [{ role: 'user', content: [{ type: 'text', text: userPrompt }], timestamp: Date.now() }] as any, + }, + { apiKey: llmConfig.apiKey, maxTokens: options?.maxTokens, temperature: options?.temperature } + ) + return result.content + .filter((item): item is PiTextContent => item.type === 'text') + .map((item) => item.text) + .join('') + }, + t: (key: string) => key, + } +} + +export async function generateSummary( + adapter: SessionRuntimeAdapter, + sessionId: string, + chatSessionId: number, + serviceDeps: SummaryServiceDeps, + options?: { locale?: string; forceRegenerate?: boolean } +) { + const llmConfig = serviceDeps.getLlmConfig() + if (!llmConfig) { + return { success: false as const, error: 'No LLM configuration available' } + } + + const db = adapter.ensureWritable(sessionId) + const deps = buildSummaryDeps(db, llmConfig, serviceDeps) + return generateSessionSummary(deps, chatSessionId, options) +} + +export async function generateAllSummaries( + adapter: SessionRuntimeAdapter, + sessionId: string, + serviceDeps: SummaryServiceDeps, + options?: { locale?: string; forceRegenerate?: boolean } +) { + const llmConfig = serviceDeps.getLlmConfig() + if (!llmConfig) { + return { success: 0, failed: 0, total: 0, error: 'No LLM configuration available' } + } + + const db = adapter.ensureWritable(sessionId) + const chatSessions = getChatSessionList(db) + const deps = buildSummaryDeps(db, llmConfig, serviceDeps) + + let success = 0 + let failed = 0 + + for (const cs of chatSessions) { + const result = await generateSessionSummary(deps, cs.id, options) + if (result.success) success++ + else failed++ + } + + return { success, failed, total: chatSessions.length } +}