diff --git a/apps/cli/src/sync/routes.test.ts b/apps/cli/src/sync/routes.test.ts index 86bc3539..51334303 100644 --- a/apps/cli/src/sync/routes.test.ts +++ b/apps/cli/src/sync/routes.test.ts @@ -2,17 +2,17 @@ import assert from 'node:assert/strict' import fs from 'node:fs' import os from 'node:os' import path from 'node:path' -import test from 'node:test' -import Fastify from 'fastify' +import test, { type TestContext } from 'node:test' import { DataSourceManager } from '@openchatlab/sync' import { registerAutomationRoutes, type AutomationRouteContext } from '@openchatlab/http-routes' +import { createApiServer } from '@openchatlab/http-routes/server' function makeTempDir(): string { const baseDir = process.env.CHATLAB_TEST_TMPDIR ?? (fs.existsSync('/private/tmp') ? '/private/tmp' : os.tmpdir()) return fs.mkdtempSync(path.join(baseDir, 'chatlab-sync-routes-')) } -test('DELETE import session deletes imported local session when deleteData=true', async () => { +async function createFixture(t: TestContext) { const settingsDir = makeTempDir() const dsManager = new DataSourceManager(settingsDir) const ds = dsManager.add({ @@ -24,7 +24,11 @@ test('DELETE import session deletes imported local session when deleteData=true' dsManager.updateSession(ds.id, session.id, { targetSessionId: 'local-1' }) const deletedSessionIds: string[] = [] - const app = Fastify() + const app = createApiServer() + t.after(async () => { + await app.close() + fs.rmSync(settingsDir, { recursive: true, force: true }) + }) const pullEngine: AutomationRouteContext['pullEngine'] = { triggerPull: async () => ({ success: true, newMessageCount: 0 }), triggerPullAll: async () => ({ success: true, newMessageCount: 0 }), @@ -41,15 +45,58 @@ test('DELETE import session deletes imported local session when deleteData=true' }, }) await app.ready() + return { app, dsManager, ds, session, deletedSessionIds, file: path.join(settingsDir, 'data-sources.json') } +} +test('DELETE import session deletes imported local session when deleteData=true', async (t) => { + const { app, dsManager, ds, session, deletedSessionIds } = await createFixture(t) const resp = await app.inject({ method: 'DELETE', url: `/_web/automation/data-sources/${ds.id}/sessions/${session.id}?deleteData=true`, }) - await app.close() - assert.equal(resp.statusCode, 200) assert.deepEqual(deletedSessionIds, ['local-1']) assert.equal(dsManager.get(ds.id)?.sessions.length, 0) }) + +test('corrupt data source config returns an error instead of an empty list or a successful replacement', async (t) => { + const { app, file } = await createFixture(t) + const broken = '[{"token":"private-test-token"} INVALID]' + fs.writeFileSync(file, broken) + const list = await app.inject({ method: 'GET', url: '/_web/automation/data-sources' }) + assert.equal(list.statusCode, 500) + assert.equal(list.json().success, false) + assert.ok(!list.body.includes('private-test-token')) + const add = await app.inject({ + method: 'POST', + url: '/_web/automation/data-sources', + payload: { baseUrl: 'https://another.test', token: '', intervalMinutes: 60 }, + }) + assert.equal(add.statusCode, 500) + assert.equal(fs.readFileSync(file, 'utf-8'), broken) +}) + +test('failed config persistence cannot report success or delete an imported local chat', async (t) => { + const { app, file, ds, session, deletedSessionIds } = await createFixture(t) + const original = fs.readFileSync(file, 'utf-8') + const renameSync = fs.renameSync + t.mock.method(fs, 'renameSync', (from: fs.PathLike, to: fs.PathLike) => { + if (to === file) throw Object.assign(new Error('Replacement denied'), { code: 'EPERM' }) + return renameSync(from, to) + }) + const update = await app.inject({ + method: 'PATCH', + url: `/_web/automation/data-sources/${ds.id}`, + payload: { name: 'Must not be saved' }, + }) + assert.equal(update.statusCode, 500) + assert.equal(update.json().success, false) + const remove = await app.inject({ + method: 'DELETE', + url: `/_web/automation/data-sources/${ds.id}/sessions/${session.id}?deleteData=true`, + }) + assert.equal(remove.statusCode, 500) + assert.deepEqual(deletedSessionIds, []) + assert.equal(fs.readFileSync(file, 'utf-8'), original) +}) diff --git a/packages/sync/src/data-source-manager.test.ts b/packages/sync/src/data-source-manager.test.ts new file mode 100644 index 00000000..138769ff --- /dev/null +++ b/packages/sync/src/data-source-manager.test.ts @@ -0,0 +1,106 @@ +import assert from 'node:assert/strict' +import fs from 'node:fs' +import os from 'node:os' +import path from 'node:path' +import test, { type TestContext } from 'node:test' +import { DataSourceManager } from './data-source-manager' + +function createFixture(t: TestContext) { + const root = fs.mkdtempSync(path.join(os.tmpdir(), 'chatlab-data-source-config-')) + t.after(() => fs.rmSync(root, { recursive: true, force: true })) + return { root, file: path.join(root, 'data-sources.json'), manager: new DataSourceManager(root) } +} + +const sourceInput = { name: 'Test source', baseUrl: 'https://example.test', token: 'test-token', intervalMinutes: 60 } + +test('missing config can be created and existing subscriptions survive updates and reloads', (t) => { + const { root, file, manager } = createFixture(t) + assert.deepEqual(manager.loadAll(), []) + const source = manager.add(sourceInput) + const [session] = manager.addSessions(source.id, [{ name: 'Chat', remoteSessionId: 'remote-1' }]) + manager.updateSession(source.id, session.id, { targetSessionId: 'local-1', lastPullAt: 123 }) + manager.update(source.id, { name: 'Renamed source' }) + + const reloaded = new DataSourceManager(root).get(source.id) + assert.equal(reloaded?.name, 'Renamed source') + assert.equal(reloaded?.token, sourceInput.token) + assert.equal(reloaded?.sessions[0].targetSessionId, 'local-1') + assert.equal(reloaded?.sessions[0].lastPullAt, 123) + assert.deepEqual(fs.readdirSync(root), [path.basename(file)]) + + // Older configurations without pullLimit still use the established default. + const legacy = JSON.parse(fs.readFileSync(file, 'utf-8')) + delete legacy[0].pullLimit + fs.writeFileSync(file, JSON.stringify(legacy)) + assert.equal(manager.get(source.id)?.pullLimit, 1000) +}) + +test('corrupt or incompatible config is reported and cannot be replaced by CRUD operations', (t) => { + const { file, manager } = createFixture(t) + for (const contents of ['', '[', '{}', '[{"sessions":null}]']) { + fs.writeFileSync(file, contents) + const operations = [ + () => manager.loadAll(), + () => manager.add(sourceInput), + () => manager.update('source', { name: 'Changed' }), + () => manager.delete('source'), + () => manager.addSessions('source', []), + () => manager.removeSession('source', 'session'), + () => manager.updateSession('source', 'session', { lastPullAt: 456 }), + ] + for (const operation of operations) { + assert.throws(operation) + assert.equal(fs.readFileSync(file, 'utf-8'), contents) + } + } +}) + +test('read failures are not treated as missing config', (t) => { + const { file, manager } = createFixture(t) + const source = manager.add(sourceInput) + const original = fs.readFileSync(file, 'utf-8') + const readFileSync = fs.readFileSync + const readMock = t.mock.method(fs, 'readFileSync', (...args: Parameters) => { + if (args[0] === file) throw Object.assign(new Error('Permission denied'), { code: 'EACCES' }) + return readFileSync(...args) + }) + assert.throws(() => manager.loadAll()) + assert.throws(() => manager.add(sourceInput)) + readMock.mock.restore() + assert.equal(fs.readFileSync(file, 'utf-8'), original) + assert.equal(manager.get(source.id)?.token, sourceInput.token) +}) + +for (const failure of ['partial-write', 'rename'] as const) { + test(`${failure} failure preserves config, reports failure and cleans the temporary file`, (t) => { + const { root, file, manager } = createFixture(t) + const source = manager.add(sourceInput) + const original = fs.readFileSync(file, 'utf-8') + let restore: () => void + if (failure === 'partial-write') { + const writeFileSync = fs.writeFileSync + const fault = t.mock.method(fs, 'writeFileSync', (...args: Parameters) => { + if (typeof args[0] === 'string' && path.dirname(args[0]) === root) { + writeFileSync(args[0], '[{"id":') + throw Object.assign(new Error('No space left on device'), { code: 'ENOSPC' }) + } + return writeFileSync(...args) + }) + restore = () => fault.mock.restore() + } else { + const renameSync = fs.renameSync + const fault = t.mock.method(fs, 'renameSync', (from: fs.PathLike, to: fs.PathLike) => { + if (to === file) throw Object.assign(new Error('Replacement denied'), { code: 'EPERM' }) + return renameSync(from, to) + }) + restore = () => fault.mock.restore() + } + + assert.throws(() => manager.update(source.id, { name: 'Must not be saved' })) + assert.equal(fs.readFileSync(file, 'utf-8'), original) + assert.deepEqual(fs.readdirSync(root), [path.basename(file)]) + restore() + manager.update(source.id, { name: 'Recovered' }) + assert.equal(new DataSourceManager(root).get(source.id)?.name, 'Recovered') + }) +} diff --git a/packages/sync/src/data-source-manager.ts b/packages/sync/src/data-source-manager.ts index 915d808a..21a15f37 100644 --- a/packages/sync/src/data-source-manager.ts +++ b/packages/sync/src/data-source-manager.ts @@ -7,7 +7,7 @@ * DataSource (remote server) → ImportSession[] (subscribed conversations) */ -import * as fs from 'fs' +import fs from 'node:fs' import * as path from 'path' import * as crypto from 'crypto' import { NOOP_LOGGER } from './types' @@ -15,6 +15,10 @@ import type { DataSource, DataSourceUpdatable, ImportSession, SyncLogger } from const CONFIG_FILE = 'data-sources.json' +export class DataSourceConfigError extends Error { + override name = 'DataSourceConfigError' +} + function generateId(prefix: string = 'ds'): string { return `${prefix}_${crypto.randomBytes(6).toString('hex')}` } @@ -55,34 +59,48 @@ export class DataSourceManager { // ==================== Load / Save ==================== loadAll(): DataSource[] { + const filePath = this.getConfigPath() try { - const filePath = this.getConfigPath() - if (fs.existsSync(filePath)) { - const raw = fs.readFileSync(filePath, 'utf-8') - const parsed = JSON.parse(raw) - - if (!isValidDataSourceArray(parsed)) { - this.logger.warn('[DataSource] Incompatible config format detected, returning empty.') - return [] - } - - for (const ds of parsed) { - if (!ds.pullLimit) ds.pullLimit = 1000 - } - return parsed + const raw = fs.readFileSync(filePath, 'utf-8') + let parsed: unknown + try { + parsed = JSON.parse(raw) + } catch { + // Native JSON errors can include configuration fragments containing access tokens. + throw new Error('Invalid JSON. Restore or repair the file before making changes.') } + if (!isValidDataSourceArray(parsed)) { + throw new Error('Invalid data source config format. Restore or repair the file before making changes.') + } + for (const ds of parsed) { + if (!ds.pullLimit) ds.pullLimit = 1000 + } + return parsed } catch (err) { - this.logger.error('[DataSource] Failed to load config', err) + if ((err as NodeJS.ErrnoException).code === 'ENOENT') return [] + const error = new DataSourceConfigError(`Failed to load data source config: ${filePath}`, { cause: err }) + this.logger.error('[DataSource] Failed to load config', error) + throw error } - return [] } private saveAll(sources: DataSource[]): void { + const filePath = this.getConfigPath() + const tempPath = `${filePath}.tmp-${process.pid}-${crypto.randomUUID()}` try { this.ensureDir() - fs.writeFileSync(this.getConfigPath(), JSON.stringify(sources, null, 2), 'utf-8') + fs.writeFileSync(tempPath, JSON.stringify(sources, null, 2), { encoding: 'utf-8', mode: 0o600 }) + fs.renameSync(tempPath, filePath) } catch (err) { - this.logger.error('[DataSource] Failed to save config', err) + const error = new DataSourceConfigError(`Failed to save data source config: ${filePath}`, { cause: err }) + this.logger.error('[DataSource] Failed to save config', error) + throw error + } finally { + try { + fs.rmSync(tempPath, { force: true }) + } catch (err) { + this.logger.error('[DataSource] Failed to remove temporary config', err) + } } } diff --git a/packages/sync/src/pull-engine.test.ts b/packages/sync/src/pull-engine.test.ts index 5cdab321..43563f2b 100644 --- a/packages/sync/src/pull-engine.test.ts +++ b/packages/sync/src/pull-engine.test.ts @@ -1,9 +1,10 @@ import { afterEach, describe, it } from 'node:test' import assert from 'node:assert/strict' -import * as fs from 'fs' +import fs from 'node:fs' import * as os from 'os' import * as path from 'path' import { PullEngine } from './pull-engine' +import { DataSourceManager, DataSourceConfigError } from './data-source-manager' import type { DataSource, DataImporter, @@ -148,6 +149,45 @@ function createEngine(options: { } describe('PullEngine', () => { + it('config write failure reports failure without advancing the cursor or leaving progress running', async (t) => { + const root = fs.mkdtempSync(path.join(os.tmpdir(), 'chatlab-pull-config-')) + t.after(() => fs.rmSync(root, { recursive: true, force: true })) + const manager = new DataSourceManager(root) + const source = manager.add({ baseUrl: 'https://example.test', token: '', intervalMinutes: 1 }) + const [session] = manager.addSessions(source.id, [{ name: 'Chat', remoteSessionId: 'remote-1' }]) + manager.updateSession(source.id, session.id, { lastPullAt: 100, targetSessionId: 'local-1' }) + const file = path.join(root, 'data-sources.json') + const original = fs.readFileSync(file, 'utf-8') + const emptyPage = writeTempJson({}) + const results: string[] = [] + const engine = new PullEngine({ + dsManager: manager, + fetcher: { fetchToTempFile: async () => emptyPage }, + importer: { sessionExists: () => true, importFile: async () => ({ success: true, newMessageCount: 0 }) }, + notifier: { + onSessionListChanged: () => {}, + onPullResult: (_sourceId, _sessionId, status) => results.push(status), + }, + }) + const writeFileSync = fs.writeFileSync + let failNextWrite = true + t.mock.method(fs, 'writeFileSync', (...args: Parameters) => { + if (failNextWrite && typeof args[0] === 'string' && path.dirname(args[0]) === root) { + failNextWrite = false + throw Object.assign(new Error('No space left on device'), { code: 'ENOSPC' }) + } + return writeFileSync(...args) + }) + + await withImmediateTimers(async () => { + await assert.rejects(engine.triggerPull(source.id, session.id), DataSourceConfigError) + assert.deepEqual(engine.getProgress(), []) + }) + assert.deepEqual(results, ['error']) + assert.equal(fs.readFileSync(file, 'utf-8'), original) + assert.equal(manager.get(source.id)?.sessions[0].lastPullAt, 100) + }) + it('imports a small final page instead of treating it as empty', async () => { const session = createSession() const dataSource = createDataSource() diff --git a/packages/sync/src/pull-engine.ts b/packages/sync/src/pull-engine.ts index ad646312..cfeabbe4 100644 --- a/packages/sync/src/pull-engine.ts +++ b/packages/sync/src/pull-engine.ts @@ -21,7 +21,7 @@ import type { PullSessionResult, PullProgress, } from './types' -import type { DataSourceManager } from './data-source-manager' +import { DataSourceConfigError, type DataSourceManager } from './data-source-manager' const MAX_PAGES_PER_PULL = 5000 const PULL_OVERLAP_SECONDS = 60 @@ -498,13 +498,20 @@ export class PullEngine { if (!sourceUnavailable) { this.logger.error(`[Pull] Pull failed for "${sess.name}": ${errMsg}`) } - this.dsManager.updateSession(sourceId, sess.id, { - lastPullAt: Math.floor(Date.now() / 1000), - lastStatus: 'error', - lastError: errMsg, - }) - this.notifier.onPullResult(sourceId, sess.id, 'error', errMsg) this.markProgressDone(sess.id) + try { + // A failed config commit must not be retried as an error-status write advancing the cursor. + if (error instanceof DataSourceConfigError) throw error + this.dsManager.updateSession(sourceId, sess.id, { + lastPullAt: Math.floor(Date.now() / 1000), + lastStatus: 'error', + lastError: errMsg, + }) + } catch (configError) { + this.notifier.onPullResult(sourceId, sess.id, 'error', getErrorMessage(configError)) + throw configError + } + this.notifier.onPullResult(sourceId, sess.id, 'error', errMsg) return { success: false, newMessageCount: 0, error: errMsg, sourceUnavailable: !!sourceUnavailable } } } diff --git a/packages/sync/src/scheduler.test.ts b/packages/sync/src/scheduler.test.ts new file mode 100644 index 00000000..f1d07586 --- /dev/null +++ b/packages/sync/src/scheduler.test.ts @@ -0,0 +1,58 @@ +import assert from 'node:assert/strict' +import fs from 'node:fs' +import os from 'node:os' +import path from 'node:path' +import test, { type TestContext } from 'node:test' +import { DataSourceManager } from './data-source-manager' +import { initScheduler, reloadTimer, stopAllTimers } from './scheduler' +import type { PullEngine } from './pull-engine' +import type { DataSource } from './types' + +function createFixture(t: TestContext) { + const root = fs.mkdtempSync(path.join(os.tmpdir(), 'chatlab-sync-scheduler-')) + t.after(() => { + stopAllTimers() + fs.rmSync(root, { recursive: true, force: true }) + }) + const dsManager = new DataSourceManager(root) + const source = dsManager.add({ baseUrl: 'https://example.test', token: '', intervalMinutes: 1 }) + dsManager.addSessions(source.id, [{ name: 'Test chat', remoteSessionId: 'remote-1' }]) + const pulls: string[] = [] + const pullEngine = { + async pullAllSessions(ds: DataSource) { + pulls.push(ds.id) + }, + } as PullEngine + return { file: path.join(root, 'data-sources.json'), dsManager, source, pulls, pullEngine } +} + +test('corrupt config pauses scheduler initialization without blocking the host and can be retried after repair', (t) => { + const { file, dsManager, source, pulls, pullEngine } = createFixture(t) + const original = fs.readFileSync(file, 'utf-8') + fs.writeFileSync(file, '[') + assert.doesNotThrow(() => initScheduler({ dsManager, pullEngine })) + assert.deepEqual(pulls, []) + assert.equal(fs.readFileSync(file, 'utf-8'), '[') + + fs.writeFileSync(file, original) + initScheduler({ dsManager, pullEngine }) + assert.deepEqual(pulls, [source.id]) +}) + +test('a scheduled read failure stops that timer without an uncaught exception or further pulls', (t) => { + const { file, dsManager, source, pulls, pullEngine } = createFixture(t) + t.mock.timers.enable({ apis: ['setInterval'] }) + initScheduler({ dsManager, pullEngine }) + assert.deepEqual(pulls, [source.id]) + const original = fs.readFileSync(file, 'utf-8') + fs.writeFileSync(file, '[') + assert.doesNotThrow(() => t.mock.timers.tick(60_000)) + assert.deepEqual(pulls, [source.id]) + + fs.writeFileSync(file, original) + t.mock.timers.tick(60_000) + assert.deepEqual(pulls, [source.id]) + reloadTimer(source.id, true) + t.mock.timers.tick(60_000) + assert.deepEqual(pulls, [source.id, source.id]) +}) diff --git a/packages/sync/src/scheduler.ts b/packages/sync/src/scheduler.ts index 978d79de..bcad99f7 100644 --- a/packages/sync/src/scheduler.ts +++ b/packages/sync/src/scheduler.ts @@ -36,14 +36,19 @@ function startTimer(ds: DataSource, skipInitialPull = false): void { } const timer = setInterval(() => { - const current = _dsManager.loadAll().find((s) => s.id === ds.id) - if (!current || !current.enabled || current.sessions.length === 0) { + try { + const current = _dsManager.loadAll().find((s) => s.id === ds.id) + if (!current || !current.enabled || current.sessions.length === 0) { + stopTimer(ds.id) + return + } + _pullEngine.pullAllSessions(current).catch((err) => { + _logger.error('[Pull] Scheduled pull failed', err) + }) + } catch (err) { stopTimer(ds.id) - return + _logger.error('[Pull] Scheduled pull stopped because data source config could not be loaded', err) } - _pullEngine.pullAllSessions(current).catch((err) => { - _logger.error('[Pull] Scheduled pull failed', err) - }) }, intervalMs) timers.set(ds.id, timer) @@ -62,13 +67,19 @@ export function stopTimer(id: string): void { export function initScheduler(options: SchedulerOptions): void { if (initialized) return - initialized = true _dsManager = options.dsManager _pullEngine = options.pullEngine _logger = options.logger ?? NOOP_LOGGER - const sources = _dsManager.loadAll() + let sources: DataSource[] + try { + sources = _dsManager.loadAll() + } catch (err) { + _logger.error('[Pull] Scheduler not started because data source config could not be loaded', err) + return + } + initialized = true for (const ds of sources) { if (ds.enabled && ds.sessions.length > 0) { startTimer(ds)