fix(sync): protect data source configuration from failed writes

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)
+})

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<typeof fs.readFileSync>) => {
+    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<typeof fs.writeFileSync>) => {
+        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')
+  })
+}

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)
+      }
     }
   }
 

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<typeof fs.writeFileSync>) => {
+      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()

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 }
     }
   }

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])
+})

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)