refactor(sync): share automation routes across runtimes

This commit is contained in:
digua committed 2026-07-02 10:35:02 +08:00
1 parent 70ada5031c
commit b346b51eee
13 files changed
+379 -502

No files matched your search

+13 -6
View File
@@ -8,16 +8,14 @@ import type { FastifyInstance } from 'fastify'
import type { DatabaseManager } from '@openchatlab/node-runtime'
import { PreferencesManager, createDatabaseManagerAdapter, ownerProfileService } from '@openchatlab/node-runtime'
import type { PathProvider } from '@openchatlab/core'
import { DataSourceManager, PullEngine, initScheduler, stopAllTimers } from '@openchatlab/sync'
import { DataSourceManager, PullEngine, initScheduler, stopAllTimers, reloadTimer, stopTimer } from '@openchatlab/sync'
import type { SyncLogger } from '@openchatlab/sync'
import { NodeFetcher, DirectImporter, NoopNotifier } from './adapters'
import { registerAutomationRoutes } from './routes'
import type { AutomationRouteContext } from '@openchatlab/http-routes'
export interface SyncRouteContext {
dsManager: DataSourceManager
pullEngine: PullEngine
dbManager: DatabaseManager
serverInfo: { port: number; host: string; token: string }
automation: AutomationRouteContext
}
const syncLogger: SyncLogger = {
@@ -61,7 +59,16 @@ export function initSync(
},
})
registerAutomationRoutes(server, { dsManager, pullEngine, dbManager, serverInfo })
registerAutomationRoutes(server, {
automation: {
dsManager,
pullEngine,
serverInfo,
deleteSessionData: (sessionId) => dbManager.deleteSessionDatabaseFiles(sessionId),
reloadTimer,
stopTimer,
},
})
initScheduler({
dsManager,
+12 -10
View File
@@ -6,6 +6,7 @@ import test from 'node:test'
import Fastify from 'fastify'
import { DataSourceManager } from '@openchatlab/sync'
import { registerAutomationRoutes } from './routes'
import type { AutomationRouteContext } from '@openchatlab/http-routes'
function makeTempDir(): string {
const baseDir = fs.existsSync('/private/tmp') ? '/private/tmp' : os.tmpdir()
@@ -25,19 +26,20 @@ test('DELETE import session deletes imported local session when deleteData=true'
const deletedSessionIds: string[] = []
const app = Fastify()
const pullEngine: AutomationRouteContext['pullEngine'] = {
triggerPull: async () => ({ success: true, newMessageCount: 0 }),
triggerPullAll: async () => ({ success: true, newMessageCount: 0 }),
getProgress: () => [],
}
registerAutomationRoutes(app, {
dsManager,
pullEngine: {
triggerPull: async () => ({ success: true, newMessageCount: 0 }),
triggerPullAll: async () => ({ success: true, newMessageCount: 0 }),
getProgress: () => [],
} as any,
dbManager: {
deleteSessionDatabaseFiles: (sessionId: string) => {
automation: {
dsManager,
pullEngine,
serverInfo: { port: 5200, host: '127.0.0.1', token: 'api-token' },
deleteSessionData: (sessionId: string) => {
deletedSessionIds.push(sessionId)
},
} as any,
serverInfo: { port: 5200, host: '127.0.0.1', token: 'api-token' },
},
})
await app.ready()
+1 -145
View File
@@ -1,145 +1 @@
/**
* ChatLab Internal Web API — /_web/automation/ routes
*
* Exposes sync engine functionality to the Web UI via HTTP endpoints.
*/
import type { FastifyInstance } from 'fastify'
import {
buildRemoteSessionsUrl,
parseRemoteSessionsResponse,
normalizeBaseUrl,
reloadTimer,
stopTimer,
} from '@openchatlab/sync'
import type { SyncRouteContext } from './index'
export function registerAutomationRoutes(server: FastifyInstance, ctx: SyncRouteContext): void {
const { dsManager, pullEngine, dbManager, serverInfo } = ctx
// ==================== Config (read-only in CLI mode) ====================
server.get('/_web/automation/config', async () => {
return {
enabled: true,
port: serverInfo.port,
token: serverInfo.token,
host: serverInfo.host,
}
})
// ==================== Data Sources ====================
server.get('/_web/automation/data-sources', async () => {
return dsManager.loadAll()
})
server.post<{
Body: { name?: string; baseUrl: string; token: string; intervalMinutes: number; pullLimit?: number }
}>('/_web/automation/data-sources', async (request) => {
const ds = dsManager.add(request.body)
reloadTimer(ds.id)
return ds
})
server.patch<{
Params: { id: string }
Body: {
name?: string
baseUrl?: string
token?: string
intervalMinutes?: number
pullLimit?: number
enabled?: boolean
}
}>('/_web/automation/data-sources/:id', async (request, reply) => {
const ds = dsManager.update(request.params.id, request.body)
if (!ds) return reply.code(404).send({ error: 'Data source not found' })
reloadTimer(ds.id)
return ds
})
server.delete<{ Params: { id: string } }>('/_web/automation/data-sources/:id', async (request, reply) => {
stopTimer(request.params.id)
const ok = dsManager.delete(request.params.id)
if (!ok) return reply.code(404).send({ error: 'Data source not found' })
return { success: true }
})
// ==================== Import Sessions ====================
server.post<{
Params: { id: string }
Body: { sessions: Array<{ name: string; remoteSessionId: string }> }
}>('/_web/automation/data-sources/:id/sessions', async (request, reply) => {
const added = dsManager.addSessions(request.params.id, request.body.sessions)
if (added.length === 0 && !dsManager.get(request.params.id)) {
return reply.code(404).send({ error: 'Data source not found' })
}
reloadTimer(request.params.id, true)
for (const sess of added) {
pullEngine.triggerPull(request.params.id, sess.id).catch(() => {})
}
return added
})
server.delete<{
Params: { id: string; sessId: string }
Querystring: { deleteData?: string }
}>('/_web/automation/data-sources/:id/sessions/:sessId', async (request, reply) => {
const removed = dsManager.removeSession(request.params.id, request.params.sessId)
if (!removed) return reply.code(404).send({ error: 'Session not found' })
reloadTimer(request.params.id)
if (request.query.deleteData === 'true' && removed.targetSessionId) {
dbManager.deleteSessionDatabaseFiles(removed.targetSessionId)
}
return { success: true }
})
// ==================== Pull / Sync ====================
server.post<{ Params: { id: string }; Body: { sessionId?: string } }>(
'/_web/automation/data-sources/:id/pull',
async (request) => {
return pullEngine.triggerPull(request.params.id, request.body?.sessionId)
}
)
server.post<{ Params: { id: string } }>('/_web/automation/data-sources/:id/pull-all', async (request) => {
return pullEngine.triggerPullAll(request.params.id)
})
// ==================== Sync Progress ====================
server.get('/_web/automation/sync-progress', async () => {
return pullEngine.getProgress()
})
// ==================== Remote Session Discovery ====================
server.get<{
Querystring: { baseUrl: string; token?: string; keyword?: string; limit?: string; cursor?: string }
}>('/_web/automation/remote-sessions', async (request, reply) => {
const { baseUrl, token, keyword, limit, cursor } = request.query
if (!baseUrl) return reply.code(400).send({ error: 'baseUrl is required' })
const normalizedUrl = normalizeBaseUrl(baseUrl)
const url = buildRemoteSessionsUrl(normalizedUrl, {
keyword,
limit: limit ? parseInt(limit, 10) : undefined,
cursor,
})
const headers: Record<string, string> = { Accept: 'application/json' }
if (token) headers['Authorization'] = `Bearer ${token}`
try {
const resp = await fetch(url, { headers, signal: AbortSignal.timeout(30_000) })
if (!resp.ok) throw new Error(`Remote server returned HTTP ${resp.status}`)
const body = await resp.text()
return parseRemoteSessionsResponse(body)
} catch (err: any) {
return reply.code(502).send({ error: err.message || 'Failed to fetch remote sessions' })
}
})
}
export { registerAutomationRoutes } from '@openchatlab/http-routes'
+3 -9
View File
@@ -3,7 +3,7 @@
*
* ElectronFetcher: uses electron.net.request
* WorkerImporter: uses worker thread IPC (streamImport / incrementalImport)
* BrowserWindowNotifier: uses BrowserWindow.webContents.send
* BrowserWindowNotifier: notifies renderer when imported sessions change
*/
import * as fs from 'fs'
@@ -165,13 +165,7 @@ export class BrowserWindowNotifier implements SyncNotifier {
}
}
onPullResult(sourceId: string, sessionId: string | undefined, status: 'success' | 'error', detail: string): void {
try {
for (const win of BrowserWindow.getAllWindows()) {
win.webContents.send('api:pullResult', { sourceId, sessionId, status, detail })
}
} catch {
/* ignore */
}
onPullResult(): void {
// Pull-result updates are polled through the shared HTTP route.
}
}
+13 -1
View File
@@ -32,6 +32,7 @@ import multipart from '@fastify/multipart'
import type { ConfigStorage } from '@openchatlab/node-runtime'
import {
registerSharedRoutes,
registerAutomationRoutes,
ApiError,
ApiErrorCode,
apiErrorFromUnknown,
@@ -39,6 +40,7 @@ import {
serverError,
} from '@openchatlab/http-routes'
import type { HttpRouteContext } from '@openchatlab/http-routes'
import { reloadTimer, stopTimer } from '@openchatlab/sync'
import { resolveApiKey, writeAuthProfile, deleteAuthProfile } from '@openchatlab/config'
import { getManager as getAIChatManager } from './ai/chats'
import { getManager as getAssistantManager } from './ai/assistant/manager'
@@ -157,6 +159,8 @@ export async function startInternalServer(pathProvider: PathProvider): Promise<I
const { shell } = await import('electron')
const { getDefaultUserDataDir, getUserDataDir, getDownloadsDir } = await import('./paths')
const { getDataSourceManager, getPullEngine } = await import('./ipc/api')
const routeDbManager = newDbManager
const ctx: HttpRouteContext = {
dbManager: newDbManager,
@@ -174,6 +178,13 @@ export async function startInternalServer(pathProvider: PathProvider): Promise<I
customProviderStore: new CustomProviderStore(configStorage),
customModelStore: new CustomModelStore(configStorage),
getCurrentAiLogPath: () => aiLogger.getExistingLogPath(),
automation: {
dsManager: getDataSourceManager(),
pullEngine: getPullEngine(),
deleteSessionData: (sessionId) => routeDbManager.deleteSessionDatabaseFiles(sessionId),
reloadTimer,
stopTimer,
},
semanticIndexService: newSemanticIndexService ?? undefined,
openDirectory: (dirPath) => shell.openPath(dirPath).then(() => {}),
showInFolder: (filePath) => {
@@ -242,7 +253,7 @@ export async function startInternalServer(pathProvider: PathProvider): Promise<I
reply.code(413).send(errorResponse(bodyErr))
return
}
const statusCode = (error as any).statusCode
const statusCode = (error as { statusCode?: number }).statusCode
if (statusCode && statusCode >= 400 && statusCode < 600) {
reply.code(statusCode).send({ success: false, error: { code: 'CLIENT_ERROR', message: error.message } })
return
@@ -253,6 +264,7 @@ export async function startInternalServer(pathProvider: PathProvider): Promise<I
})
registerSharedRoutes(newServer, ctx, { requireAi: true })
registerAutomationRoutes(newServer, ctx)
await newServer.listen({ port: 0, host: '127.0.0.1' })
+10 -134
View File
@@ -4,30 +4,17 @@
* Migrated to @openchatlab/sync shared package.
*/
import { ipcMain, net } from 'electron'
import { ipcMain } from 'electron'
import type { IpcContext } from './types'
import * as apiServer from '../api'
import { setConfigManager } from '../api'
import { getSettingsDir, getSystemDataDir } from '../paths'
import { apiLogger } from '../api/logger'
import { getImportingStatus } from '../api/routes/import'
import { deleteSession as deleteSessionFile } from '../database/core'
import { getInternalDbManager } from '../internal-api'
import { PreferencesManager, createDatabaseManagerAdapter, ownerProfileService } from '@openchatlab/node-runtime'
import { ElectronFetcher, WorkerImporter, BrowserWindowNotifier } from '../api/adapters'
import {
ConfigManager,
DataSourceManager,
PullEngine,
initScheduler,
stopAllTimers,
stopTimer,
reloadTimer,
buildRemoteSessionsUrl,
parseRemoteSessionsResponse,
normalizeBaseUrl,
} from '@openchatlab/sync'
import type { DataSourceUpdatable, RemoteSessionDiscoveryQuery, RemoteSessionDiscoveryResult } from '@openchatlab/sync'
import { ConfigManager, DataSourceManager, PullEngine, initScheduler, stopAllTimers } from '@openchatlab/sync'
const syncLogger = {
info: (msg: string) => apiLogger.info(msg),
@@ -76,51 +63,14 @@ export function getConfigManager(): ConfigManager {
return configManager
}
function fetchRemoteSessions(
baseUrl: string,
token?: string,
query: RemoteSessionDiscoveryQuery = {}
): Promise<RemoteSessionDiscoveryResult> {
return new Promise<RemoteSessionDiscoveryResult>((resolve, reject) => {
const url = buildRemoteSessionsUrl(normalizeBaseUrl(baseUrl), query)
export function getDataSourceManager(): DataSourceManager {
ensureInstances()
return dsManager
}
const request = net.request(url)
if (token) {
request.setHeader('Authorization', `Bearer ${token}`)
}
request.setHeader('Accept', 'application/json')
let body = ''
request.on('response', (response) => {
if (response.statusCode !== 200) {
reject(new Error(`Remote server returned HTTP ${response.statusCode}`))
return
}
response.on('data', (chunk: Buffer) => {
body += chunk.toString('utf-8')
})
response.on('end', () => {
try {
resolve(parseRemoteSessionsResponse(body))
} catch {
reject(new Error('Failed to parse remote sessions response'))
}
})
response.on('error', (err: Error) => {
reject(err)
})
})
request.on('error', (err: Error) => {
reject(err)
})
request.end()
})
export function getPullEngine(): PullEngine {
ensureInstances()
return pullEngine
}
export function registerApiHandlers(_ctx: IpcContext): void {
@@ -158,81 +108,7 @@ export function registerApiHandlers(_ctx: IpcContext): void {
return configManager.update(partial as any)
})
// ==================== Data Source Management ====================
ipcMain.handle('api:getDataSources', () => {
return dsManager.loadAll()
})
ipcMain.handle(
'api:addDataSource',
(
_event,
partial: { name?: string; baseUrl: string; token: string; intervalMinutes: number; pullLimit?: number }
) => {
return dsManager.add(partial)
}
)
ipcMain.handle('api:updateDataSource', (_event, id: string, updates: DataSourceUpdatable) => {
const ds = dsManager.update(id, updates)
if (ds) {
reloadTimer(ds.id)
}
return ds
})
ipcMain.handle('api:deleteDataSource', (_event, id: string) => {
stopTimer(id)
return dsManager.delete(id)
})
// ==================== Import Session Management ====================
ipcMain.handle(
'api:addImportSessions',
(_event, sourceId: string, sessions: Array<{ name: string; remoteSessionId: string }>) => {
const added = dsManager.addSessions(sourceId, sessions)
reloadTimer(sourceId, true)
for (const sess of added) {
pullEngine.triggerPull(sourceId, sess.id).catch(() => {})
}
return added
}
)
ipcMain.handle('api:removeImportSession', (_event, sourceId: string, sessionId: string, deleteData?: boolean) => {
const removed = dsManager.removeSession(sourceId, sessionId)
if (!removed) return false
reloadTimer(sourceId)
if (deleteData && removed.targetSessionId) {
deleteSessionFile(removed.targetSessionId)
}
return true
})
// ==================== Sync ====================
ipcMain.handle('api:triggerPull', async (_event, sourceId: string, sessionId?: string) => {
return pullEngine.triggerPull(sourceId, sessionId)
})
ipcMain.handle('api:triggerPullAll', async (_event, sourceId: string) => {
return pullEngine.triggerPullAll(sourceId)
})
// ==================== Remote Discovery ====================
ipcMain.handle(
'api:fetchRemoteSessions',
async (_event, baseUrl: string, token: string, query?: { keyword?: string; limit?: number; cursor?: string }) => {
try {
return await fetchRemoteSessions(baseUrl, token || undefined, query)
} catch (err: any) {
throw new Error(err.message || 'Failed to fetch remote sessions')
}
}
)
// Data-source sync routes have moved to the shared internal HTTP server.
}
/**
+2 -112
View File
@@ -2,7 +2,7 @@
* ChatLab API 服务 Preload API (hierarchical data source model)
*/
import { ipcRenderer } from 'electron'
import { ipcRenderer, type IpcRendererEvent } from 'electron'
export interface ApiServerConfig {
enabled: boolean
@@ -18,49 +18,6 @@ export interface ApiServerStatus {
error: string | null
}
export interface ImportSession {
id: string
name: string
remoteSessionId: string
targetSessionId: string
lastPullAt: number
lastStatus: 'idle' | 'success' | 'error'
lastError: string
lastNewMessages: number
}
export interface DataSource {
id: string
name: string
baseUrl: string
token: string
intervalMinutes: number
pullLimit: number
enabled: boolean
createdAt: number
sessions: ImportSession[]
}
export interface RemoteSession {
id: string
name: string
platform: string
type: string
messageCount?: number
memberCount?: number
lastMessageAt?: number
}
export interface RemoteSessionDiscoveryPage {
hasMore: boolean
nextCursor?: string
}
export interface RemoteSessionDiscoveryResult {
sessions: RemoteSession[]
page?: RemoteSessionDiscoveryPage
}
export const apiServerApi = {
// ==================== API 服务管理 ====================
@@ -85,78 +42,11 @@ export const apiServerApi = {
},
onStartupError: (callback: (data: { error: string }) => void): (() => void) => {
const handler = (_event: any, data: { error: string }) => callback(data)
const handler = (_event: IpcRendererEvent, data: { error: string }) => callback(data)
ipcRenderer.on('api:startupError', handler)
return () => ipcRenderer.removeListener('api:startupError', handler)
},
// ==================== 数据源管理 ====================
getDataSources: (): Promise<DataSource[]> => {
return ipcRenderer.invoke('api:getDataSources')
},
addDataSource: (partial: {
name?: string
baseUrl: string
token: string
intervalMinutes: number
pullLimit?: number
}): Promise<DataSource> => {
return ipcRenderer.invoke('api:addDataSource', partial)
},
updateDataSource: (
id: string,
updates: Partial<Pick<DataSource, 'name' | 'baseUrl' | 'token' | 'intervalMinutes' | 'pullLimit' | 'enabled'>>
): Promise<DataSource | null> => {
return ipcRenderer.invoke('api:updateDataSource', id, updates)
},
deleteDataSource: (id: string): Promise<boolean> => {
return ipcRenderer.invoke('api:deleteDataSource', id)
},
// ==================== 导入会话管理 ====================
addImportSessions: (
sourceId: string,
sessions: Array<{ name: string; remoteSessionId: string }>
): Promise<ImportSession[]> => {
return ipcRenderer.invoke('api:addImportSessions', sourceId, sessions)
},
removeImportSession: (sourceId: string, sessionId: string, deleteData?: boolean): Promise<boolean> => {
return ipcRenderer.invoke('api:removeImportSession', sourceId, sessionId, deleteData)
},
// ==================== 同步 ====================
triggerPull: (sourceId: string, sessionId?: string): Promise<{ success: boolean; error?: string }> => {
return ipcRenderer.invoke('api:triggerPull', sourceId, sessionId)
},
triggerPullAll: (sourceId: string): Promise<{ success: boolean; error?: string }> => {
return ipcRenderer.invoke('api:triggerPullAll', sourceId)
},
fetchRemoteSessions: (
baseUrl: string,
token?: string,
query?: { keyword?: string; limit?: number; cursor?: string }
): Promise<RemoteSessionDiscoveryResult> => {
return ipcRenderer.invoke('api:fetchRemoteSessions', baseUrl, token || '', query)
},
onPullResult: (
callback: (data: { sourceId: string; sessionId?: string; status: string; detail: string }) => void
): (() => void) => {
const handler = (_event: any, data: { sourceId: string; sessionId?: string; status: string; detail: string }) =>
callback(data)
ipcRenderer.on('api:pullResult', handler)
return () => ipcRenderer.removeListener('api:pullResult', handler)
},
onImportCompleted: (callback: () => void): (() => void) => {
const handler = () => callback()
ipcRenderer.on('api:importCompleted', handler)
-73
View File
@@ -269,49 +269,6 @@ interface ApiServerStatus {
error: string | null
}
interface ImportSession {
id: string
name: string
remoteSessionId: string
targetSessionId: string
lastPullAt: number
lastStatus: 'idle' | 'success' | 'error'
lastError: string
lastNewMessages: number
}
interface DataSource {
id: string
name: string
baseUrl: string
token: string
intervalMinutes: number
pullLimit: number
enabled: boolean
createdAt: number
sessions: ImportSession[]
}
interface RemoteSession {
id: string
name: string
platform: string
type: string
messageCount?: number
memberCount?: number
lastMessageAt?: number
}
interface RemoteSessionDiscoveryPage {
hasMore: boolean
nextCursor?: string
}
interface RemoteSessionDiscoveryResult {
sessions: RemoteSession[]
page?: RemoteSessionDiscoveryPage
}
interface ApiServerApi {
getConfig: () => Promise<ApiServerConfig>
getStatus: () => Promise<ApiServerStatus>
@@ -319,34 +276,6 @@ interface ApiServerApi {
setPort: (port: number) => Promise<ApiServerStatus>
regenerateToken: () => Promise<ApiServerConfig>
onStartupError: (callback: (data: { error: string }) => void) => () => void
getDataSources: () => Promise<DataSource[]>
addDataSource: (partial: {
name?: string
baseUrl: string
token: string
intervalMinutes: number
pullLimit?: number
}) => Promise<DataSource>
updateDataSource: (
id: string,
updates: Partial<Pick<DataSource, 'name' | 'baseUrl' | 'token' | 'intervalMinutes' | 'pullLimit' | 'enabled'>>
) => Promise<DataSource | null>
deleteDataSource: (id: string) => Promise<boolean>
addImportSessions: (
sourceId: string,
sessions: Array<{ name: string; remoteSessionId: string }>
) => Promise<ImportSession[]>
removeImportSession: (sourceId: string, sessionId: string, deleteData?: boolean) => Promise<boolean>
triggerPull: (sourceId: string, sessionId?: string) => Promise<{ success: boolean; error?: string }>
triggerPullAll: (sourceId: string) => Promise<{ success: boolean; error?: string }>
fetchRemoteSessions: (
baseUrl: string,
token?: string,
query?: { keyword?: string; limit?: number; cursor?: string }
) => Promise<RemoteSessionDiscoveryResult>
onPullResult: (
callback: (data: { sourceId: string; sessionId?: string; status: string; detail: string }) => void
) => () => void
onImportCompleted: (callback: () => void) => () => void
}
@@ -395,8 +324,6 @@ export {
ApiServerApi,
ApiServerConfig,
ApiServerStatus,
DataSource,
ImportSession,
InternalApi,
InternalEndpoint,
}
+45
View File
@@ -96,6 +96,51 @@ export interface HttpRouteContext {
/** AI tool debug execution - platform-specific so Electron can keep DB work in its worker. */
executeAiTool?: (params: AiToolExecuteRequest) => Promise<AiToolExecuteResult>
/** Data-source sync automation — optional because some route hosts do not enable sync. */
automation?: AutomationRouteContext
}
export interface AutomationDataSourceLike {
id: string
sessions: Array<{ id: string; targetSessionId?: string | null }>
}
export interface AutomationRouteContext {
dsManager: {
loadAll: () => unknown[]
get: (id: string) => AutomationDataSourceLike | null | undefined
add: (source: {
name?: string
baseUrl: string
token: string
intervalMinutes: number
pullLimit?: number
}) => unknown
update: (
id: string,
updates: {
name?: string
baseUrl?: string
token?: string
intervalMinutes?: number
pullLimit?: number
enabled?: boolean
}
) => unknown | null
delete: (id: string) => boolean
addSessions: (sourceId: string, sessions: Array<{ name: string; remoteSessionId: string }>) => unknown[]
removeSession: (sourceId: string, sessionId: string) => { targetSessionId?: string | null } | null | undefined
}
pullEngine: {
triggerPull: (sourceId: string, sessionId?: string) => Promise<unknown>
triggerPullAll: (sourceId: string) => Promise<unknown>
getProgress: () => unknown[]
}
serverInfo?: { port: number; host: string; token: string }
deleteSessionData?: (sessionId: string) => void
reloadTimer?: (sourceId: string, immediate?: boolean) => void
stopTimer?: (sourceId: string) => void
}
export interface AiToolExecuteRequest {
+9 -1
View File
@@ -1,4 +1,11 @@
export type { HttpRouteContext, AgentStreamRequest, AiToolExecuteRequest, AiToolExecuteResult } from './context'
export type {
HttpRouteContext,
AutomationRouteContext,
AutomationDataSourceLike,
AgentStreamRequest,
AiToolExecuteRequest,
AiToolExecuteResult,
} from './context'
export { registerSharedRoutes } from './register'
export type { SharedRouteOptions } from './register'
export { setAuthToken, setRequireAuth, authHook } from './auth'
@@ -40,3 +47,4 @@ export { registerAiLlmStreamRoutes } from './routes/web/ai-llm-stream'
export { registerAiAgentStreamRoutes } from './routes/web/ai-agent-stream'
export { registerAiToolRoutes } from './routes/web/ai-tools'
export { registerTelemetryRoutes } from './routes/web/telemetry'
export { registerAutomationRoutes } from './routes/web/automation'
@@ -17,6 +17,7 @@ import { PreferencesManager, type DatabaseManager, type SessionRuntimeAdapter }
import type { HttpRouteContext } from '../context'
import { registerSharedRoutes } from '../register'
import { registerRestSessionRoutes } from './sessions'
import { registerAutomationRoutes } from './web/automation'
import { registerSessionRoutes } from './web/sessions'
const nativeBinding = path.resolve('apps/cli/native/better_sqlite3.node')
@@ -346,6 +347,56 @@ describe('registerSharedRoutes smoke tests', () => {
}
})
it('DELETE /_web/automation/data-sources/:id/sessions/:sessId deletes imported local session when deleteData=true', async () => {
const ds = {
id: 'source-1',
sessions: [{ id: 'session-1', name: 'First chat', remoteSessionId: 'remote-1', targetSessionId: 'local-1' }],
}
const deletedSessionIds: string[] = []
const ctx = createTestContext()
ctx.automation = {
serverInfo: { port: 5200, host: '127.0.0.1', token: 'api-token' },
dsManager: {
loadAll: () => [ds],
get: (id: string) => (id === ds.id ? ds : null),
add: (source: Record<string, unknown>) => ({ id: 'new-source', sessions: [], ...source }),
update: () => null,
delete: () => false,
addSessions: () => [],
removeSession: (sourceId: string, sessionId: string) => {
if (sourceId !== ds.id || sessionId !== ds.sessions[0].id) return null
return ds.sessions.shift() ?? null
},
},
pullEngine: {
triggerPull: async () => ({ success: true, newMessageCount: 0 }),
triggerPullAll: async () => ({ success: true, newMessageCount: 0 }),
getProgress: () => [],
},
deleteSessionData: (sessionId: string) => {
deletedSessionIds.push(sessionId)
},
}
const routeApp = Fastify()
registerAutomationRoutes(routeApp, ctx)
await routeApp.ready()
try {
const resp = await routeApp.inject({
method: 'DELETE',
url: '/_web/automation/data-sources/source-1/sessions/session-1?deleteData=true',
})
assert.equal(resp.statusCode, 200)
assert.deepEqual(resp.json(), { success: true })
assert.deepEqual(deletedSessionIds, ['local-1'])
assert.deepEqual(ds.sessions, [])
} finally {
await routeApp.close()
}
})
it('PATCH /_web/sessions/:id/name updates the shared session metadata', async () => {
const db = createSessionDb()
const routeApp = Fastify()
@@ -0,0 +1,197 @@
import type { FastifyInstance } from 'fastify'
import type { HttpRouteContext } from '../../context'
type AutomationRouteHost = Pick<HttpRouteContext, 'automation'>
function requireAutomation(ctx: AutomationRouteHost) {
return ctx.automation ?? null
}
function normalizeBaseUrl(input: string): string {
let url = input.trim().replace(/\/+$/, '')
if (url && !/^https?:\/\//i.test(url)) url = `http://${url}`
if (url && !url.endsWith('/api/v1')) {
url = url.replace(/\/api\/v1$/, '') + '/api/v1'
}
return url
}
function buildRemoteSessionsUrl(
baseUrl: string,
query: { keyword?: string; limit?: string | number; cursor?: string } = {}
): string {
const searchParams = new URLSearchParams()
searchParams.set('format', 'chatlab')
if (query.keyword?.trim()) searchParams.set('keyword', query.keyword.trim())
if (query.limit && Number(query.limit) > 0) searchParams.set('limit', String(query.limit))
if (query.cursor) searchParams.set('cursor', query.cursor)
return `${baseUrl}/sessions?${searchParams.toString()}`
}
function asRecord(value: unknown): Record<string, unknown> | undefined {
return value && typeof value === 'object' && !Array.isArray(value) ? (value as Record<string, unknown>) : undefined
}
function parseRemoteSessionsResponse(body: string): {
sessions: unknown[]
page?: { hasMore: boolean; nextCursor?: string }
} {
const parsed = JSON.parse(body)
let sessions: unknown[]
let pageSource: Record<string, unknown> | undefined
if (Array.isArray(parsed)) {
sessions = parsed
} else {
const record = asRecord(parsed)
const data = asRecord(record?.data)
const sessionsSource = record?.sessions ?? data?.sessions ?? record?.data
sessions = Array.isArray(sessionsSource) ? sessionsSource : []
pageSource = asRecord(record?.page) ?? asRecord(data?.page)
}
return {
sessions,
page:
pageSource && typeof pageSource === 'object'
? {
hasMore: Boolean(pageSource.hasMore),
nextCursor: typeof pageSource.nextCursor === 'string' ? pageSource.nextCursor : undefined,
}
: undefined,
}
}
export function registerAutomationRoutes(server: FastifyInstance, ctx: AutomationRouteHost): void {
server.get('/_web/automation/config', async (_request, reply) => {
const automation = requireAutomation(ctx)
if (!automation?.serverInfo) return reply.code(404).send({ error: 'Automation is not enabled' })
return {
enabled: true,
port: automation.serverInfo.port,
token: automation.serverInfo.token,
host: automation.serverInfo.host,
}
})
server.get('/_web/automation/data-sources', async (_request, reply) => {
const automation = requireAutomation(ctx)
if (!automation) return reply.code(404).send({ error: 'Automation is not enabled' })
return automation.dsManager.loadAll()
})
server.post<{
Body: { name?: string; baseUrl: string; token: string; intervalMinutes: number; pullLimit?: number }
}>('/_web/automation/data-sources', async (request, reply) => {
const automation = requireAutomation(ctx)
if (!automation) return reply.code(404).send({ error: 'Automation is not enabled' })
const ds = automation.dsManager.add(request.body)
automation.reloadTimer?.((ds as { id: string }).id)
return ds
})
server.patch<{
Params: { id: string }
Body: {
name?: string
baseUrl?: string
token?: string
intervalMinutes?: number
pullLimit?: number
enabled?: boolean
}
}>('/_web/automation/data-sources/:id', async (request, reply) => {
const automation = requireAutomation(ctx)
if (!automation) return reply.code(404).send({ error: 'Automation is not enabled' })
const ds = automation.dsManager.update(request.params.id, request.body)
if (!ds) return reply.code(404).send({ error: 'Data source not found' })
automation.reloadTimer?.((ds as { id: string }).id)
return ds
})
server.delete<{ Params: { id: string } }>('/_web/automation/data-sources/:id', async (request, reply) => {
const automation = requireAutomation(ctx)
if (!automation) return reply.code(404).send({ error: 'Automation is not enabled' })
automation.stopTimer?.(request.params.id)
const ok = automation.dsManager.delete(request.params.id)
if (!ok) return reply.code(404).send({ error: 'Data source not found' })
return { success: true }
})
server.post<{
Params: { id: string }
Body: { sessions: Array<{ name: string; remoteSessionId: string }> }
}>('/_web/automation/data-sources/:id/sessions', async (request, reply) => {
const automation = requireAutomation(ctx)
if (!automation) return reply.code(404).send({ error: 'Automation is not enabled' })
const added = automation.dsManager.addSessions(request.params.id, request.body.sessions)
if (added.length === 0 && !automation.dsManager.get(request.params.id)) {
return reply.code(404).send({ error: 'Data source not found' })
}
automation.reloadTimer?.(request.params.id, true)
for (const sess of added as Array<{ id: string }>) {
void automation.pullEngine.triggerPull(request.params.id, sess.id).catch(() => undefined)
}
return added
})
server.delete<{
Params: { id: string; sessId: string }
Querystring: { deleteData?: string }
}>('/_web/automation/data-sources/:id/sessions/:sessId', async (request, reply) => {
const automation = requireAutomation(ctx)
if (!automation) return reply.code(404).send({ error: 'Automation is not enabled' })
const removed = automation.dsManager.removeSession(request.params.id, request.params.sessId)
if (!removed) return reply.code(404).send({ error: 'Session not found' })
automation.reloadTimer?.(request.params.id)
if (request.query.deleteData === 'true' && removed.targetSessionId) {
automation.deleteSessionData?.(removed.targetSessionId)
}
return { success: true }
})
server.post<{ Params: { id: string }; Body: { sessionId?: string } }>(
'/_web/automation/data-sources/:id/pull',
async (request, reply) => {
const automation = requireAutomation(ctx)
if (!automation) return reply.code(404).send({ error: 'Automation is not enabled' })
return automation.pullEngine.triggerPull(request.params.id, request.body?.sessionId)
}
)
server.post<{ Params: { id: string } }>('/_web/automation/data-sources/:id/pull-all', async (request, reply) => {
const automation = requireAutomation(ctx)
if (!automation) return reply.code(404).send({ error: 'Automation is not enabled' })
return automation.pullEngine.triggerPullAll(request.params.id)
})
server.get('/_web/automation/sync-progress', async (_request, reply) => {
const automation = requireAutomation(ctx)
if (!automation) return reply.code(404).send({ error: 'Automation is not enabled' })
return automation.pullEngine.getProgress()
})
server.get<{
Querystring: { baseUrl: string; token?: string; keyword?: string; limit?: string; cursor?: string }
}>('/_web/automation/remote-sessions', async (request, reply) => {
const automation = requireAutomation(ctx)
if (!automation) return reply.code(404).send({ error: 'Automation is not enabled' })
const { baseUrl, token, keyword, limit, cursor } = request.query
if (!baseUrl) return reply.code(400).send({ error: 'baseUrl is required' })
const url = buildRemoteSessionsUrl(normalizeBaseUrl(baseUrl), { keyword, limit, cursor })
const headers: Record<string, string> = { Accept: 'application/json' }
if (token) headers.Authorization = `Bearer ${token}`
try {
const resp = await fetch(url, { headers, signal: AbortSignal.timeout(30_000) })
if (!resp.ok) throw new Error(`Remote server returned HTTP ${resp.status}`)
const body = await resp.text()
return parseRemoteSessionsResponse(body)
} catch (err) {
return reply.code(502).send({ error: err instanceof Error ? err.message : 'Failed to fetch remote sessions' })
}
})
}
+23 -11
View File
@@ -111,6 +111,7 @@ interface ApiTransport {
function createElectronTransport(): ApiTransport {
const api = window.apiServerApi
const http = createWebTransport()
return {
getConfig: () => api.getConfig(),
getStatus: () => api.getStatus(),
@@ -118,16 +119,17 @@ function createElectronTransport(): ApiTransport {
setPort: (port) => api.setPort(port),
regenerateToken: () => api.regenerateToken(),
onStartupError: (cb) => api.onStartupError(cb),
getDataSources: () => api.getDataSources(),
addDataSource: (partial) => api.addDataSource(partial),
updateDataSource: (id, updates) => api.updateDataSource(id, updates),
deleteDataSource: (id) => api.deleteDataSource(id),
addImportSessions: (sourceId, sessions) => api.addImportSessions(sourceId, sessions),
removeImportSession: (sourceId, sessionId, deleteData?) => api.removeImportSession(sourceId, sessionId, deleteData),
triggerPull: (sourceId, sessionId?) => api.triggerPull(sourceId, sessionId),
triggerPullAll: (sourceId) => api.triggerPullAll(sourceId),
onPullResult: (cb) => api.onPullResult(cb),
fetchRemoteSessions: (baseUrl, token?, query?) => api.fetchRemoteSessions(baseUrl, token, query),
getDataSources: () => http.getDataSources(),
addDataSource: (partial) => http.addDataSource(partial),
updateDataSource: (id, updates) => http.updateDataSource(id, updates),
deleteDataSource: (id) => http.deleteDataSource(id),
addImportSessions: (sourceId, sessions) => http.addImportSessions(sourceId, sessions),
removeImportSession: (sourceId, sessionId, deleteData?) =>
http.removeImportSession(sourceId, sessionId, deleteData),
triggerPull: (sourceId, sessionId?) => http.triggerPull(sourceId, sessionId),
triggerPullAll: (sourceId) => http.triggerPullAll(sourceId),
onPullResult: (cb) => http.onPullResult(cb),
fetchRemoteSessions: (baseUrl, token?, query?) => http.fetchRemoteSessions(baseUrl, token, query),
}
}
@@ -138,7 +140,17 @@ function createWebTransport(): ApiTransport {
const resp = await fetchWithAuth(url, options)
if (!resp.ok) {
const body = await resp.json().catch(() => ({}))
throw new Error((body as any)?.error || `HTTP ${resp.status}`)
const errorBody = body as { error?: unknown }
const errorMessage =
typeof errorBody.error === 'string'
? errorBody.error
: typeof errorBody.error === 'object' &&
errorBody.error &&
'message' in errorBody.error &&
typeof errorBody.error.message === 'string'
? errorBody.error.message
: `HTTP ${resp.status}`
throw new Error(errorMessage)
}
return resp.json()
}