diff --git a/src/core/postgres-engine.ts b/src/core/postgres-engine.ts index 041eb8068..b41408bdf 100644 --- a/src/core/postgres-engine.ts +++ b/src/core/postgres-engine.ts @@ -1276,11 +1276,28 @@ export class PostgresEngine implements BrainEngine { } async updateSourceConfig(sourceId: string, patch: Record): Promise { - // v0.38: atomic JSONB merge. `||` is the Postgres concat operator — - // for jsonb, right-side keys overwrite left-side; nested object keys - // are NOT deep-merged (use jsonb_set for nested paths). The patch - // shape this autopilot wave uses is flat (`last_full_cycle_at`, - // `archive_*`, etc.) so concat is sufficient. Idempotent on re-run. + // Atomic single-statement merge. The previous read-then-write form dropped + // concurrent updates: two callers patching different keys could both read + // the same old config and the later `SET config = ...` clobbered the + // earlier patch. These keys are written by background cycle/autopilot + // paths, so the merge must happen inside the UPDATE (parity with + // pglite-engine.updateSourceConfig, which already uses JSONB `||`). + // + // The CASE normalizes historical bad shapes inline (so `config` is re-read + // against the row-locked latest version — a CTE/subquery snapshot would + // reintroduce the lost-update race under READ COMMITTED): older code paths + // could store config as a JSONB string (double-encoded) or as a JSONB array + // of patch objects. We coerce those to a flat object before the `||` merge + // so doctor and source routing keep getting flat keys. + // + // String branch guard: a JSONB string whose inner text is NOT itself valid + // JSON (one of the historical bad shapes this path repairs) would make the + // bare `::jsonb` cast raise `invalid input syntax for type json`, failing + // the whole UPDATE. Postgres has no `try_cast`, so we gate the cast with + // the SQL `IS JSON` predicate (Postgres 16+): parseable inner text is + // double-encoded config and gets parsed; unparseable text falls back to `{}`. + // The guard keeps the merge a single atomic statement (no extra round-trip, + // no lost-update race). // // MUST use sql.json(patch) inside the template tag — postgres-js's // positional executeRaw + `$1::jsonb` cast DOUBLE-ENCODES the @@ -1294,7 +1311,25 @@ export class PostgresEngine implements BrainEngine { const sql = this.sql; const result = await sql` UPDATE sources - SET config = COALESCE(config, '{}'::jsonb) || ${sql.json(patch as Parameters[0])} + SET config = + CASE + WHEN jsonb_typeof(config) = 'object' THEN config + WHEN jsonb_typeof(config) = 'string' + THEN CASE + WHEN (config #>> '{}') IS JSON + THEN COALESCE(NULLIF((config #>> '{}'), '')::jsonb, '{}'::jsonb) + ELSE '{}'::jsonb + END + WHEN jsonb_typeof(config) = 'array' + THEN COALESCE( + (SELECT jsonb_object_agg(kv.key, kv.value) + FROM jsonb_array_elements(config) elem, + jsonb_each(elem) kv), + '{}'::jsonb + ) + ELSE '{}'::jsonb + END + || ${sql.json(patch as Parameters[0])} WHERE id = ${sourceId} `; return (result.count ?? 0) > 0; diff --git a/test/list-all-sources.test.ts b/test/list-all-sources.test.ts index 4d0f9ddb5..fddfc9b0a 100644 --- a/test/list-all-sources.test.ts +++ b/test/list-all-sources.test.ts @@ -135,4 +135,51 @@ describe('engine.updateSourceConfig', () => { const all = await engine.listAllSources(); expect(all.find(s => s.id === 'delta')!.config.last_full_cycle_at).toBe('2026-05-22T11:00:00.000Z'); }); + + // IS JSON guard: the postgres-engine atomic merge gates its `::jsonb` cast + // behind the SQL `IS JSON` predicate so a historical bad row whose config is + // a JSONB string of NON-JSON text normalizes to `{}` instead of raising + // `invalid input syntax for type json` and aborting the UPDATE. + // + // We exercise the SQL CASE expression directly via executeRaw against PGLite + // (which ships Postgres 17 + IS JSON parity) rather than calling + // PostgresEngine.updateSourceConfig (which requires a live Postgres pool). + test('IS-JSON guard: non-JSON string config normalizes to {} on merge', async () => { + const patch = { merged_key: 'v' }; + + const guarded = (configExpr: string) => ` + SELECT ( + CASE + WHEN jsonb_typeof(${configExpr}) = 'object' THEN ${configExpr} + WHEN jsonb_typeof(${configExpr}) = 'string' + THEN CASE + WHEN (${configExpr} #>> '{}') IS JSON + THEN COALESCE(NULLIF((${configExpr} #>> '{}'), '')::jsonb, '{}'::jsonb) + ELSE '{}'::jsonb + END + WHEN jsonb_typeof(${configExpr}) = 'array' + THEN COALESCE( + (SELECT jsonb_object_agg(kv.key, kv.value) + FROM jsonb_array_elements(${configExpr}) elem, + jsonb_each(elem) kv), + '{}'::jsonb + ) + ELSE '{}'::jsonb + END || $1::jsonb + ) AS result`; + + // 1. JSONB string holding NON-JSON text → normalizes to {} then merges patch. + const bad = await engine.executeRaw<{ result: Record }>( + guarded(`to_jsonb('garbage text'::text)`), + [JSON.stringify(patch)], + ); + expect(bad[0].result).toEqual({ merged_key: 'v' }); + + // 2. JSONB string holding double-encoded valid JSON object → parsed + merged. + const good = await engine.executeRaw<{ result: Record }>( + guarded(`to_jsonb('{"x":1}'::text)`), + [JSON.stringify(patch)], + ); + expect(good[0].result).toEqual({ x: 1, merged_key: 'v' }); + }); });