admin/gbrain
1
0
Fork 0
mirror of https://github.com/garrytan/gbrain.git synced 2026-07-27 22:15:33 +00:00
Code Issues Packages Projects Releases Wiki Activity

feat(engine): opt-in Postgres RLS source-scope binding (lands #2387) (#2940)

Browse Source 
Takeover of community PR #2387 with the security review's required fixes
applied. Original work by @harrisali0101.

With GBRAIN_RLS_SCOPE_BINDING=1, source-scoped Postgres read methods wrap
their queries in a transaction that binds set_config('app.scopes', $1, true)
(federated sourceIds CSV > scalar sourceId > '*') so operator-managed RLS
policies can filter rows at the SQL layer — defense-in-depth layer 2 under
the mandatory app-layer source filters.

Review fixes on top of the original PR:
- Flag-off is now a TRUE pass-through: no new per-read transaction wrap
  (the #1794 PgBouncer pool-exhaustion class). Only the three search
  methods keep a transaction when off — exactly the sql.begin() +
  SET LOCAL statement_timeout wrap they already had on master — via the
  helper's alwaysTransaction option.
- Preserved the PR's latent setseed fix: listCorpusSample pins
  setseed() + SELECT to one connection when seeded (alwaysTransaction
  gated on opts.seed), so the deterministic path can't split across
  pooled connections.
- Updated the two postgres-engine shape tests to pin the new invariant
  (search methods route through withScopedReadTransaction with
  alwaysTransaction; helper owns the sql.begin(); flag-off path is
  callback(this.sql)).
- New behavioral tests (test/postgres-engine-rls-scope.test.ts): flag-off
  pass-through, flag-off alwaysTransaction, flag-on set_config emission,
  federated > scalar > '*' precedence, and the CSV as a bound parameter
  (never interpolated).
- Fixed the helper header comment to match the actual branching behavior.
- Operator docs in docs/ENGINES.md: env var, policy SQL, the
  ALTER ROLE ... SET app.scopes='*' default requirement, FORCE ROW LEVEL
  SECURITY for owner roles, and the honest caveat about unwrapped paths.

Co-authored-by: Sinabina <sinabina@Sinabinas-MacBook-Pro-4.local>
Co-authored-by: Harris <79081645+harrisali0101@users.noreply.github.com>
Co-authored-by: Claude Fable 5 <noreply@anthropic.com>
This commit is contained in:
Time Attakc
2026-07-17 15:15:10 -07:00
committed by GitHub
co-authored by Sinabina Harris Claude Fable 5
parent 1833d95896
commit 9fe4628d02
5 changed files with 712 additions and 305 deletions
Download Patch File Download Diff File Expand all files Collapse all files
docs/ENGINES.md
+45
View File
@@ -148,6 +148,51 @@ RRF fusion, multi-query expansion, and 4-layer dedup are engine-agnostic. They o
**Why not self-hosted for v0:** The brain should be infrastructure agents use, not something you maintain. Self-hosted Postgres with Docker is a welcome community PR, but v0 optimizes for zero ops.
### Opt-in RLS source-scope binding (`GBRAIN_RLS_SCOPE_BINDING`)
Defense-in-depth layer for Postgres deployments that want the database itself
to enforce source isolation, in addition to the mandatory app-layer filters
(`sourceScopeOpts` — layer 1, always on).
**Mechanism.** With `GBRAIN_RLS_SCOPE_BINDING=1` (or `true`), the engine's
source-scoped read methods wrap their queries in a transaction that first runs
`SELECT set_config('app.scopes', $1, true)` — the value is a bound parameter
(federated `sourceIds` CSV > scalar `sourceId` > `'*'` for unscoped internal
reads), transaction-local (equivalent to `SET LOCAL`, which itself can't take
bound params). An RLS policy can then filter rows by
`current_setting('app.scopes', true)`.
**Default off.** With the env var unset, reads call through on the shared pool
exactly as before — no per-read transaction, no pool-slot hold (the search
methods keep the transaction they always had for their `SET LOCAL
statement_timeout`). Existing operators see zero behavior change.
**Enabling it** (operator-managed SQL; gbrain ships no DDL for this):
```sql
ALTER TABLE pages ENABLE ROW LEVEL SECURITY;
CREATE POLICY pages_scope_filter ON pages
USING (current_setting('app.scopes', true) = '*'
OR source_id = ANY(string_to_array(current_setting('app.scopes', true), ',')));
-- Required: connections that don't run through the scoped read helper
-- (admin, autopilot, cycle, writes) must default to unscoped, or they
-- see zero rows once the policy exists:
ALTER ROLE <runtime-role> SET app.scopes = '*';
-- If the runtime role OWNS the table, RLS is skipped for it unless forced:
ALTER TABLE pages FORCE ROW LEVEL SECURITY;
```
Safe to enable in either order: the env var without a policy is a no-op
setting; a policy without the env var is enforced only via the role default.
**Honest caveat:** only read paths routed through the scoped helper carry a
per-request scope binding — unwrapped paths (writes, admin/maintenance reads)
run under the role default and are not backstopped per caller. This is layer 2;
the app-layer source filters remain layer 1 and stay mandatory. Behavioral pins
live in `test/postgres-engine-rls-scope.test.ts`.
## PGLiteEngine (v0.7, ships)
**Dependencies:** `@electric-sql/pglite` (v0.4.4+)
llms-full.txt
+45
View File
@@ -2095,6 +2095,51 @@ RRF fusion, multi-query expansion, and 4-layer dedup are engine-agnostic. They o
**Why not self-hosted for v0:** The brain should be infrastructure agents use, not something you maintain. Self-hosted Postgres with Docker is a welcome community PR, but v0 optimizes for zero ops.
### Opt-in RLS source-scope binding (`GBRAIN_RLS_SCOPE_BINDING`)
Defense-in-depth layer for Postgres deployments that want the database itself
to enforce source isolation, in addition to the mandatory app-layer filters
(`sourceScopeOpts` — layer 1, always on).
**Mechanism.** With `GBRAIN_RLS_SCOPE_BINDING=1` (or `true`), the engine's
source-scoped read methods wrap their queries in a transaction that first runs
`SELECT set_config('app.scopes', $1, true)` — the value is a bound parameter
(federated `sourceIds` CSV > scalar `sourceId` > `'*'` for unscoped internal
reads), transaction-local (equivalent to `SET LOCAL`, which itself can't take
bound params). An RLS policy can then filter rows by
`current_setting('app.scopes', true)`.
**Default off.** With the env var unset, reads call through on the shared pool
exactly as before — no per-read transaction, no pool-slot hold (the search
methods keep the transaction they always had for their `SET LOCAL
statement_timeout`). Existing operators see zero behavior change.
**Enabling it** (operator-managed SQL; gbrain ships no DDL for this):
```sql
ALTER TABLE pages ENABLE ROW LEVEL SECURITY;
CREATE POLICY pages_scope_filter ON pages
USING (current_setting('app.scopes', true) = '*'
OR source_id = ANY(string_to_array(current_setting('app.scopes', true), ',')));
-- Required: connections that don't run through the scoped read helper
-- (admin, autopilot, cycle, writes) must default to unscoped, or they
-- see zero rows once the policy exists:
ALTER ROLE <runtime-role> SET app.scopes = '*';
-- If the runtime role OWNS the table, RLS is skipped for it unless forced:
ALTER TABLE pages FORCE ROW LEVEL SECURITY;
```
Safe to enable in either order: the env var without a policy is a no-op
setting; a policy without the env var is enforced only via the role default.
**Honest caveat:** only read paths routed through the scoped helper carry a
per-request scope binding — unwrapped paths (writes, admin/maintenance reads)
run under the role default and are not backstopped per caller. This is layer 2;
the app-layer source filters remain layer 1 and stay mandatory. Behavioral pins
live in `test/postgres-engine-rls-scope.test.ts`.
## PGLiteEngine (v0.7, ships)
**Dependencies:** `@electric-sql/pglite` (v0.4.4+)
src/core/postgres-engine.ts
+417 -301
View File
@@ -160,6 +160,88 @@ export class PostgresEngine implements BrainEngine {
return db.getConnection();
}
// Source-scope binding for Postgres RLS — opt-in via env var.
//
// When `GBRAIN_RLS_SCOPE_BINDING` is set to `1` / `true`, source-scoped
// query methods (listPages, search*, getChunks, etc.) wrap their queries
// in a transaction that begins with
// SELECT set_config('app.scopes', '<csv-of-allowed-source-ids>', true)
// (equivalent to `SET LOCAL app.scopes = '<value>'`, but works through
// parameterised SQL — `SET LOCAL` itself doesn't accept parameters)
// so Postgres RLS policies on source-scoped tables can filter rows by
// `current_setting('app.scopes', true)`. The expected policy shape:
//
// USING (current_setting('app.scopes', true) = '*'
// OR source_id = ANY(string_to_array(
// current_setting('app.scopes', true), ',')))
//
// Recommended runtime-role default:
// ALTER ROLE <runtime-role> SET app.scopes = '*';
// so admin / autopilot / cycle queries that don't pass scope info still
// see all rows. OAuth-scoped requests override the default per
// transaction with their allowed-source CSV.
//
// Default behavior (env var unset): the helper is a TRUE pass-through —
// it calls `callback(this.sql)` with no transaction wrap and no
// set_config, byte-identical to not having this helper at all. The only
// exception is callers that pass `alwaysTransaction: true` (the search
// methods, whose `SET LOCAL statement_timeout` already required a
// transaction on master) — they keep exactly the `sql.begin()` wrap
// they had before this helper existed. No read gains a new per-read
// pool-hold when the flag is off (the #1794 PgBouncer-exhaustion class).
//
// Honest caveat: only the read paths that route through this helper are
// backstopped by RLS. This is defense-in-depth layer 2; the app-layer
// source filters (sourceScopeOpts) remain layer 1 and stay mandatory.
private get rlsScopeBindingEnabled(): boolean {
const v = process.env.GBRAIN_RLS_SCOPE_BINDING;
return v === '1' || v === 'true';
}
private async withScopedReadTransaction<T>(
sourceIds: string[] | undefined,
sourceId: string | undefined,
callback: (tx: ReturnType<typeof postgres>) => Promise<T>,
opts?: { alwaysTransaction?: boolean },
): Promise<T> {
// Flag off + no pre-existing transaction need: call through on the
// shared pool exactly as master does. No tx round-trip, no pool slot
// held for the duration of the read.
if (!this.rlsScopeBindingEnabled && !opts?.alwaysTransaction) {
return await callback(this.sql);
}
// Precedence matches sourceScopeOpts: federated array > scalar > '*'
// (unscoped — relies on the recommended `ALTER ROLE ... SET
// app.scopes = '*'` default, or on no policy being installed).
let scopesValue = '*';
if (sourceIds && sourceIds.length > 0) {
scopesValue = sourceIds.join(',');
} else if (sourceId) {
scopesValue = sourceId;
}
// Note on nesting: a postgres.js transaction handle exposes
// `.savepoint()` not `.begin()`, so callbacks must not try to open
// their own `tx.begin()` inside this wrap — they'd fail with
// `tx.begin is not a function`. Callbacks that need SET LOCAL emit it
// directly on the handle (it shares this transaction).
//
// `sql.begin<T>(...)` returns `UnwrapPromiseArray<T>` in postgres.js's typings
// — TypeScript strict-generics can't narrow that back to `T` for arbitrary
// callback return shapes (TS2322). The unwrap is a no-op when the callback
// returns a single value (not an array of promises), so the cast is safe.
return (await this.sql.begin(async (tx: any) => {
if (this.rlsScopeBindingEnabled) {
// `SET LOCAL` doesn't accept parameters in PostgreSQL — using
// `tx\`SET LOCAL ... = ${val}\`` binds val as $1 and errors with
// `syntax error at or near "$1"`. set_config() is a regular function
// and accepts a parameterised value; passing `true` as the third
// argument makes it transaction-local (same scope as SET LOCAL).
await tx`SELECT set_config('app.scopes', ${scopesValue}, true)`;
}
return await callback(tx as ReturnType<typeof postgres>);
})) as T;
}
// Lifecycle
async connect(config: EngineConfig & { poolSize?: number; parentConnectionManager?: ConnectionManager }): Promise<void> {
this._savedConfig = config;
@@ -920,30 +1002,36 @@ export class PostgresEngine implements BrainEngine {
// Pages CRUD
async getPage(slug: string, opts?: { sourceId?: string; sourceIds?: string[]; includeDeleted?: boolean }): Promise<Page | null> {
const sql = this.sql;
const includeDeleted = opts?.includeDeleted === true;
const sourceId = opts?.sourceId;
const sourceIds = opts?.sourceIds;
// v0.26.5: default hides soft-deleted rows. Compose with optional source
// filter via fragment chaining (postgres.js supports sql`` composition).
// #1393: a federated grant (sourceIds[]) takes precedence over scalar
// sourceId so the exact-match read honors allowedSources, not just one source.
const sourceCondition =
sourceIds && sourceIds.length > 0
? sql`AND source_id = ANY(${sourceIds}::text[])`
: sourceId
? sql`AND source_id = ${sourceId}`
: sql``;
const deletedCondition = includeDeleted ? sql`` : sql`AND deleted_at IS NULL`;
const rows = await sql`
SELECT id, source_id, slug, type, title, compiled_truth, timeline, frontmatter, content_hash, created_at, updated_at, deleted_at,
source_kind, source_uri, ingested_via, ingested_at
FROM pages
WHERE slug = ${slug} ${sourceCondition} ${deletedCondition}
LIMIT 1
`;
if (rows.length === 0) return null;
return rowToPage(rows[0]);
// Two layers of defense:
// 1. RLS scope binding (opt-in via GBRAIN_RLS_SCOPE_BINDING): wraps the
// query in a transaction that sets `app.scopes` so the row-level
// policy on `pages` filters at the SQL layer. Pass-through when off.
// 2. App-layer source filter (#1393): a federated grant (sourceIds[])
// takes precedence over scalar sourceId so the exact-match read
// honors allowedSources, not just one source.
return await this.withScopedReadTransaction(sourceIds, sourceId, async (tx) => {
// v0.26.5: default hides soft-deleted rows. Compose with optional source
// filter via fragment chaining (postgres.js supports sql`` composition).
const sourceCondition =
sourceIds && sourceIds.length > 0
? tx`AND source_id = ANY(${sourceIds}::text[])`
: sourceId
? tx`AND source_id = ${sourceId}`
: tx``;
const deletedCondition = includeDeleted ? tx`` : tx`AND deleted_at IS NULL`;
const rows = await tx`
SELECT id, source_id, slug, type, title, compiled_truth, timeline, frontmatter, content_hash, created_at, updated_at, deleted_at,
source_kind, source_uri, ingested_via, ingested_at
FROM pages
WHERE slug = ${slug} ${sourceCondition} ${deletedCondition}
LIMIT 1
`;
if (rows.length === 0) return null;
return rowToPage(rows[0]);
});
}
/**
@@ -954,19 +1042,21 @@ export class PostgresEngine implements BrainEngine {
sourceId: string,
opts: { hash: string; frontmatterId?: string | null },
): Promise<{ slug: string; id: number } | null> {
const sql = this.sql;
const fmId = opts.frontmatterId ?? null;
const rows = await sql`
SELECT id, slug FROM pages
WHERE source_id = ${sourceId}
AND deleted_at IS NULL
AND (content_hash = ${opts.hash} OR (frontmatter->>'id' = ${fmId} AND ${fmId}::text IS NOT NULL))
ORDER BY id
LIMIT 1
`;
if (rows.length === 0) return null;
const r = rows[0] as { id: number | string; slug: string };
return { slug: r.slug, id: Number(r.id) };
// RLS scope binding: sourceId is positional here.
return await this.withScopedReadTransaction(undefined, sourceId, async (tx) => {
const rows = await tx`
SELECT id, slug FROM pages
WHERE source_id = ${sourceId}
AND deleted_at IS NULL
AND (content_hash = ${opts.hash} OR (frontmatter->>'id' = ${fmId} AND ${fmId}::text IS NOT NULL))
ORDER BY id
LIMIT 1
`;
if (rows.length === 0) return null;
const r = rows[0] as { id: number | string; slug: string };
return { slug: r.slug, id: Number(r.id) };
});
}
async putPage(slug: string, page: PageInput, opts?: { sourceId?: string }): Promise<Page> {
@@ -1235,25 +1325,31 @@ export class PostgresEngine implements BrainEngine {
const sortKey = filters?.sort && PAGE_SORT_SQL[filters.sort] ? filters.sort : 'updated_desc';
const orderBy = sql.unsafe(PAGE_SORT_SQL[sortKey]);
const rows = await sql`
SELECT p.* FROM pages p
${tagJoin}
WHERE 1=1 ${typeCondition} ${tagCondition} ${updatedCondition} ${slugCondition} ${sourceCondition} ${deletedCondition}
ORDER BY ${orderBy} LIMIT ${limit} OFFSET ${offset}
`;
return rows.map(rowToPage);
// RLS scope binding (opt-in via GBRAIN_RLS_SCOPE_BINDING): when
// enabled, this wraps the query in a transaction that sets
// `app.scopes` from filters; when disabled, it's a pass-through.
return await this.withScopedReadTransaction(filters?.sourceIds, filters?.sourceId, async (tx) => {
const rows = await tx`
SELECT p.* FROM pages p
${tagJoin}
WHERE 1=1 ${typeCondition} ${tagCondition} ${updatedCondition} ${slugCondition} ${sourceCondition} ${deletedCondition}
ORDER BY ${orderBy} LIMIT ${limit} OFFSET ${offset}
`;
return rows.map(rowToPage);
});
}
async getAllSlugs(opts?: { sourceId?: string }): Promise<Set<string>> {
const sql = this.sql;
// v0.31.8 (D12): two-branch. See pglite-engine.ts:getAllSlugs for context.
if (opts?.sourceId) {
const rows = await sql`SELECT slug FROM pages WHERE source_id = ${opts.sourceId}`;
return new Set(rows.map((r) => r.slug as string));
}
const rows = await sql`SELECT slug FROM pages`;
return new Set(rows.map((r) => r.slug as string));
// RLS scope binding (opt-in via GBRAIN_RLS_SCOPE_BINDING).
return await this.withScopedReadTransaction(undefined, opts?.sourceId, async (tx) => {
// v0.31.8 (D12): two-branch. See pglite-engine.ts:getAllSlugs for context.
if (opts?.sourceId) {
const rows = await tx`SELECT slug FROM pages WHERE source_id = ${opts.sourceId}`;
return new Set(rows.map((r: Record<string, unknown>) => r.slug as string));
}
const rows = await tx`SELECT slug FROM pages`;
return new Set(rows.map((r: Record<string, unknown>) => r.slug as string));
});
}
async listAllPageRefs(): Promise<Array<{ slug: string; source_id: string }>> {
@@ -1368,7 +1464,6 @@ export class PostgresEngine implements BrainEngine {
// 2. connection_count DESC — structural-centrality tiebreaker (D10)
// 3. slug ASC — deterministic for tests
async listPrefixSampledPages(opts: DomainBankSampleOpts): Promise<DomainBankRow[]> {
const sql = this.sql;
if (opts.prefixes.length === 0) return [];
const exclude = opts.excludeSlugs ?? [];
const staleBias = opts.staleBias === true;
@@ -1376,7 +1471,9 @@ export class PostgresEngine implements BrainEngine {
// Source scoping (D5, codex r2 #2 — federated array wins over scalar).
const sourceIds = opts.sourceIds ?? null;
const sourceId = opts.sourceId ?? null;
const rows = await sql`
// RLS scope binding (opt-in via GBRAIN_RLS_SCOPE_BINDING).
return await this.withScopedReadTransaction(opts.sourceIds, opts.sourceId, async (tx) => {
const rows = await tx`
WITH prefix_pages AS (
SELECT
p.id AS page_id,
@@ -1443,36 +1540,42 @@ export class PostgresEngine implements BrainEngine {
FROM with_chunk
ORDER BY prefix
`;
return rows.map((r): DomainBankRow => ({
slug: r.slug as string,
source_id: r.source_id as string,
prefix: r.prefix as string | null,
page_id: Number(r.page_id),
title: r.title as string | null,
compiled_truth: (r.compiled_truth as string | null) ?? '',
connection_count: Number(r.connection_count),
last_retrieved_at: r.last_retrieved_at as Date | null,
representative_chunk_id: r.representative_chunk_id == null ? null : Number(r.representative_chunk_id),
}));
return rows.map((r: Record<string, unknown>): DomainBankRow => ({
slug: r.slug as string,
source_id: r.source_id as string,
prefix: r.prefix as string | null,
page_id: Number(r.page_id),
title: r.title as string | null,
compiled_truth: (r.compiled_truth as string | null) ?? '',
connection_count: Number(r.connection_count),
last_retrieved_at: r.last_retrieved_at as Date | null,
representative_chunk_id: r.representative_chunk_id == null ? null : Number(r.representative_chunk_id),
}));
});
}
// v0.37.0 — corpus-sampling fallback when prefix-stratified can't fill M.
// Deterministic with opts.seed (setseed before SELECT); random otherwise.
async listCorpusSample(opts: CorpusSampleOpts): Promise<DomainBankRow[]> {
const sql = this.sql;
if (opts.n <= 0) return [];
const exclude = opts.excludeSlugs ?? [];
const sourceIds = opts.sourceIds ?? null;
const sourceId = opts.sourceId ?? null;
// setseed deterministic path: use SELECT setseed($1) + RANDOM(). PGLite/Postgres
// both honor setseed for the same session/transaction. For tests this gives
// identical ordering across runs.
if (typeof opts.seed === 'number') {
// Clamp to [-1, 1] required by setseed.
const clamped = Math.max(-1, Math.min(1, opts.seed));
await sql`SELECT setseed(${clamped}::float8)`;
}
const rows = await sql`
// RLS scope binding (opt-in via GBRAIN_RLS_SCOPE_BINDING).
// alwaysTransaction when seeded: setseed() only affects RANDOM() on the
// SAME connection. On a pool, a bare `sql\`SELECT setseed(...)\`` and the
// subsequent SELECT can land on different connections, silently breaking
// the deterministic path — the transaction pins both to one connection.
return await this.withScopedReadTransaction(opts.sourceIds, opts.sourceId, async (tx) => {
// setseed deterministic path: use SELECT setseed($1) + RANDOM(). PGLite/Postgres
// both honor setseed for the same session/transaction. For tests this gives
// identical ordering across runs.
if (typeof opts.seed === 'number') {
// Clamp to [-1, 1] required by setseed.
const clamped = Math.max(-1, Math.min(1, opts.seed));
await tx`SELECT setseed(${clamped}::float8)`;
}
const rows = await tx`
WITH sampled AS (
SELECT
p.id AS page_id,
@@ -1504,17 +1607,18 @@ export class PostgresEngine implements BrainEngine {
) AS representative_chunk_id
FROM sampled s
`;
return rows.map((r): DomainBankRow => ({
slug: r.slug as string,
source_id: r.source_id as string,
prefix: r.prefix as string | null,
page_id: Number(r.page_id),
title: r.title as string | null,
compiled_truth: (r.compiled_truth as string | null) ?? '',
connection_count: Number(r.connection_count),
last_retrieved_at: r.last_retrieved_at as Date | null,
representative_chunk_id: r.representative_chunk_id == null ? null : Number(r.representative_chunk_id),
}));
return rows.map((r: Record<string, unknown>): DomainBankRow => ({
slug: r.slug as string,
source_id: r.source_id as string,
prefix: r.prefix as string | null,
page_id: Number(r.page_id),
title: r.title as string | null,
compiled_truth: (r.compiled_truth as string | null) ?? '',
connection_count: Number(r.connection_count),
last_retrieved_at: r.last_retrieved_at as Date | null,
representative_chunk_id: r.representative_chunk_id == null ? null : Number(r.representative_chunk_id),
}));
}, { alwaysTransaction: typeof opts.seed === 'number' });
}
async resolveSlugs(partial: string, opts?: { sourceId?: string; sourceIds?: string[] }): Promise<string[]> {
@@ -1555,7 +1659,6 @@ export class PostgresEngine implements BrainEngine {
// list_pages etc. see zero breaking changes. A2 two-pass (Layer 7)
// consumes searchKeywordChunks for the raw chunk-grain primitive.
async searchKeyword(query: string, opts?: SearchOpts): Promise<SearchResult[]> {
const sql = this.sql;
const limit = clampSearchLimit(opts?.limit);
const offset = opts?.offset || 0;
const type = opts?.type;
@@ -1694,12 +1797,16 @@ export class PostgresEngine implements BrainEngine {
OFFSET ${offsetParam}
`;
// Search-only timeout. SET LOCAL inside sql.begin() scopes the GUC
// to the transaction so it can never leak onto a pooled connection.
const rows = await sql.begin(async sql => {
await sql`SET LOCAL statement_timeout = '8s'`;
return await sql.unsafe(rawQuery, params as Parameters<typeof sql.unsafe>[1]);
});
// RLS scope binding (opt-in via GBRAIN_RLS_SCOPE_BINDING) + search-only
// timeout. alwaysTransaction: this method needed sql.begin() on master
// already (SET LOCAL statement_timeout must be transaction-scoped so
// the GUC can never leak onto a pooled connection). Flag off → the
// wrap is identical to master's; flag on → set_config('app.scopes')
// shares the same transaction as the timeout.
const rows = await this.withScopedReadTransaction(opts?.sourceIds, opts?.sourceId, async (tx) => {
await tx`SET LOCAL statement_timeout = '8s'`;
return await tx.unsafe(rawQuery, params as Parameters<typeof tx.unsafe>[1]);
}, { alwaysTransaction: true });
return rows.map(rowToSearchResult);
}
@@ -1713,7 +1820,6 @@ export class PostgresEngine implements BrainEngine {
* contract). This is intentionally a narrow internal knob.
*/
async searchKeywordChunks(query: string, opts?: SearchOpts): Promise<SearchResult[]> {
const sql = this.sql;
const limit = clampSearchLimit(opts?.limit);
const offset = opts?.offset || 0;
const type = opts?.type;
@@ -1820,15 +1926,17 @@ export class PostgresEngine implements BrainEngine {
OFFSET ${offsetParam}
`;
const rows = await sql.begin(async sql => {
await sql`SET LOCAL statement_timeout = '8s'`;
return await sql.unsafe(rawQuery, params as Parameters<typeof sql.unsafe>[1]);
});
// RLS scope binding + search-only timeout. alwaysTransaction: master
// already wrapped this in sql.begin() for the SET LOCAL; flag off is
// identical to that wrap, flag on adds set_config in the same tx.
const rows = await this.withScopedReadTransaction(opts?.sourceIds, opts?.sourceId, async (tx) => {
await tx`SET LOCAL statement_timeout = '8s'`;
return await tx.unsafe(rawQuery, params as Parameters<typeof tx.unsafe>[1]);
}, { alwaysTransaction: true });
return rows.map(rowToSearchResult);
}
async searchVector(embedding: Float32Array, opts?: SearchOpts): Promise<SearchResult[]> {
const sql = this.sql;
const limit = clampSearchLimit(opts?.limit);
const offset = opts?.offset || 0;
const type = opts?.type;
@@ -1992,10 +2100,13 @@ export class PostgresEngine implements BrainEngine {
OFFSET ${offsetParam}
`;
const rows = await sql.begin(async sql => {
await sql`SET LOCAL statement_timeout = '8s'`;
return await sql.unsafe(rawQuery, params as Parameters<typeof sql.unsafe>[1]);
});
// RLS scope binding + search-only timeout. alwaysTransaction: master
// already wrapped this in sql.begin() for the SET LOCAL; flag off is
// identical to that wrap, flag on adds set_config in the same tx.
const rows = await this.withScopedReadTransaction(opts?.sourceIds, opts?.sourceId, async (tx) => {
await tx`SET LOCAL statement_timeout = '8s'`;
return await tx.unsafe(rawQuery, params as Parameters<typeof tx.unsafe>[1]);
}, { alwaysTransaction: true });
return rows.map(rowToSearchResult);
}
@@ -2238,15 +2349,17 @@ export class PostgresEngine implements BrainEngine {
}
async getChunks(slug: string, opts?: { sourceId?: string }): Promise<Chunk[]> {
const sql = this.sql;
const sourceId = opts?.sourceId ?? 'default';
const rows = await sql`
SELECT cc.* FROM content_chunks cc
JOIN pages p ON p.id = cc.page_id
WHERE p.slug = ${slug} AND p.source_id = ${sourceId}
ORDER BY cc.chunk_index
`;
return rows.map((r) => rowToChunk(r as Record<string, unknown>));
// RLS scope binding (opt-in via GBRAIN_RLS_SCOPE_BINDING).
return await this.withScopedReadTransaction(undefined, sourceId, async (tx) => {
const rows = await tx`
SELECT cc.* FROM content_chunks cc
JOIN pages p ON p.id = cc.page_id
WHERE p.slug = ${slug} AND p.source_id = ${sourceId}
ORDER BY cc.chunk_index
`;
return rows.map((r: Record<string, unknown>) => rowToChunk(r));
});
}
/**
@@ -2277,14 +2390,17 @@ export class PostgresEngine implements BrainEngine {
// D7: source_id scoping. v0.41.31: optional signature widens staleness
// to embedding_signature drift (NULL grandfathered).
const { where, params } = this.buildStaleChunkWhere(opts);
const rows = await this.sql.unsafe(
`SELECT count(*)::int AS count
FROM content_chunks cc
JOIN pages p ON p.id = cc.page_id
WHERE ${where}`,
params as Parameters<typeof this.sql.unsafe>[1],
);
return Number((rows[0] as { count?: number } | undefined)?.count ?? 0);
// RLS scope binding (opt-in via GBRAIN_RLS_SCOPE_BINDING).
return await this.withScopedReadTransaction(undefined, opts?.sourceId, async (tx) => {
const rows = await tx.unsafe(
`SELECT count(*)::int AS count
FROM content_chunks cc
JOIN pages p ON p.id = cc.page_id
WHERE ${where}`,
params as Parameters<typeof tx.unsafe>[1],
);
return Number((rows[0] as { count?: number } | undefined)?.count ?? 0);
});
}
async sumStaleChunkChars(opts?: { sourceId?: string; signature?: string }): Promise<number> {
@@ -2341,41 +2457,67 @@ export class PostgresEngine implements BrainEngine {
orderBy?: 'page_id' | 'updated_desc';
afterUpdatedAt?: string | null;
}): Promise<StaleChunkRow[]> {
const sql = this.sql;
const limit = opts?.batchSize ?? 2000;
const afterPid = opts?.afterPageId ?? 0;
const afterIdx = opts?.afterChunkIndex ?? -1;
const orderBy = opts?.orderBy ?? 'page_id';
// v0.41.18.0 (A13, codex #9): --priority recent path. Composite cursor
// (updated_at DESC NULLS LAST, page_id ASC, chunk_index ASC). Backed by
// idx_pages_updated_at_desc + content_chunks_stale_idx partial.
// "Next row" semantic with DESC NULLS LAST + ASC tiebreakers is:
// (updated_at < prev) OR
// (updated_at = prev AND page_id > prev_page_id) OR
// (updated_at = prev AND page_id = prev_page_id AND chunk_index > prev_chunk_index)
// First call: afterUpdatedAt undefined → returns the highest updated_at rows.
if (orderBy === 'updated_desc') {
const afterUpdated = opts?.afterUpdatedAt ?? null;
const isFirstPage = afterUpdated === null && afterPid === 0;
if (opts?.sourceId === undefined) {
const rows = isFirstPage ? await sql`
// RLS scope binding (opt-in via GBRAIN_RLS_SCOPE_BINDING).
return await this.withScopedReadTransaction(undefined, opts?.sourceId, async (tx) => {
// v0.41.18.0 (A13, codex #9): --priority recent path. Composite cursor
// (updated_at DESC NULLS LAST, page_id ASC, chunk_index ASC). Backed by
// idx_pages_updated_at_desc + content_chunks_stale_idx partial.
if (orderBy === 'updated_desc') {
const afterUpdated = opts?.afterUpdatedAt ?? null;
const isFirstPage = afterUpdated === null && afterPid === 0;
if (opts?.sourceId === undefined) {
const rows = isFirstPage ? await tx`
SELECT p.slug, cc.chunk_index, cc.chunk_text, cc.chunk_source,
cc.model, cc.token_count, p.source_id, cc.page_id,
p.updated_at
FROM content_chunks cc
JOIN pages p ON p.id = cc.page_id
WHERE cc.embedding IS NULL
AND NOT (COALESCE(p.frontmatter, '{}'::jsonb) ? 'embed_skip')
ORDER BY p.updated_at DESC NULLS LAST, p.id ASC, cc.chunk_index ASC
LIMIT ${limit}
` : await tx`
SELECT p.slug, cc.chunk_index, cc.chunk_text, cc.chunk_source,
cc.model, cc.token_count, p.source_id, cc.page_id,
p.updated_at
FROM content_chunks cc
JOIN pages p ON p.id = cc.page_id
WHERE cc.embedding IS NULL
AND NOT (COALESCE(p.frontmatter, '{}'::jsonb) ? 'embed_skip')
AND (
p.updated_at < ${afterUpdated}::timestamptz
OR (p.updated_at = ${afterUpdated}::timestamptz AND p.id > ${afterPid})
OR (p.updated_at = ${afterUpdated}::timestamptz AND p.id = ${afterPid} AND cc.chunk_index > ${afterIdx})
)
ORDER BY p.updated_at DESC NULLS LAST, p.id ASC, cc.chunk_index ASC
LIMIT ${limit}
`;
return rows as unknown as StaleChunkRow[];
}
const rows = isFirstPage ? await tx`
SELECT p.slug, cc.chunk_index, cc.chunk_text, cc.chunk_source,
cc.model, cc.token_count, p.source_id, cc.page_id,
p.updated_at
FROM content_chunks cc
JOIN pages p ON p.id = cc.page_id
WHERE cc.embedding IS NULL
AND p.source_id = ${opts.sourceId}
AND NOT (COALESCE(p.frontmatter, '{}'::jsonb) ? 'embed_skip')
ORDER BY p.updated_at DESC NULLS LAST, p.id ASC, cc.chunk_index ASC
LIMIT ${limit}
` : await sql`
` : await tx`
SELECT p.slug, cc.chunk_index, cc.chunk_text, cc.chunk_source,
cc.model, cc.token_count, p.source_id, cc.page_id,
p.updated_at
FROM content_chunks cc
JOIN pages p ON p.id = cc.page_id
WHERE cc.embedding IS NULL
AND p.source_id = ${opts.sourceId}
AND NOT (COALESCE(p.frontmatter, '{}'::jsonb) ? 'embed_skip')
AND (
p.updated_at < ${afterUpdated}::timestamptz
@@ -2387,77 +2529,35 @@ export class PostgresEngine implements BrainEngine {
`;
return rows as unknown as StaleChunkRow[];
}
const rows = isFirstPage ? await sql`
SELECT p.slug, cc.chunk_index, cc.chunk_text, cc.chunk_source,
cc.model, cc.token_count, p.source_id, cc.page_id,
p.updated_at
FROM content_chunks cc
JOIN pages p ON p.id = cc.page_id
WHERE cc.embedding IS NULL
AND p.source_id = ${opts.sourceId}
AND NOT (COALESCE(p.frontmatter, '{}'::jsonb) ? 'embed_skip')
ORDER BY p.updated_at DESC NULLS LAST, p.id ASC, cc.chunk_index ASC
LIMIT ${limit}
` : await sql`
SELECT p.slug, cc.chunk_index, cc.chunk_text, cc.chunk_source,
cc.model, cc.token_count, p.source_id, cc.page_id,
p.updated_at
FROM content_chunks cc
JOIN pages p ON p.id = cc.page_id
WHERE cc.embedding IS NULL
AND p.source_id = ${opts.sourceId}
AND NOT (COALESCE(p.frontmatter, '{}'::jsonb) ? 'embed_skip')
AND (
p.updated_at < ${afterUpdated}::timestamptz
OR (p.updated_at = ${afterUpdated}::timestamptz AND p.id > ${afterPid})
OR (p.updated_at = ${afterUpdated}::timestamptz AND p.id = ${afterPid} AND cc.chunk_index > ${afterIdx})
)
ORDER BY p.updated_at DESC NULLS LAST, p.id ASC, cc.chunk_index ASC
LIMIT ${limit}
`;
return rows as unknown as StaleChunkRow[];
}
// orderBy === 'page_id' — legacy stable cursor (unchanged below).
// Cursor-paginated: keyset pagination on (page_id, chunk_index).
// The partial index idx_chunks_embedding_null makes the WHERE fast;
// LIMIT keeps each round-trip well within statement_timeout.
//
// D7: optional source_id filter. NULL/undefined = scan all sources
// (pre-existing behavior); a value scopes to that source so
// `gbrain embed --stale --source X` actually does what it says.
//
// v0.41 (D4+D8): NOT (frontmatter ? 'embed_skip') filter applied via
// the always-JOINed pages row. Soft-blocked pages won't surface in
// the stale list; their chunks were deleted at ingest time anyway
// (D9 transition invariant), but the filter is defense-in-depth for
// pre-fix inventory that might still have orphan chunks.
if (opts?.sourceId === undefined) {
const rows = await sql`
// orderBy === 'page_id' — legacy stable cursor.
if (opts?.sourceId === undefined) {
const rows = await tx`
SELECT p.slug, cc.chunk_index, cc.chunk_text, cc.chunk_source,
cc.model, cc.token_count, p.source_id, cc.page_id
FROM content_chunks cc
JOIN pages p ON p.id = cc.page_id
WHERE cc.embedding IS NULL
AND NOT (COALESCE(p.frontmatter, '{}'::jsonb) ? 'embed_skip')
AND (cc.page_id, cc.chunk_index) > (${afterPid}, ${afterIdx})
ORDER BY cc.page_id, cc.chunk_index
LIMIT ${limit}
`;
return rows as unknown as StaleChunkRow[];
}
const rows = await tx`
SELECT p.slug, cc.chunk_index, cc.chunk_text, cc.chunk_source,
cc.model, cc.token_count, p.source_id, cc.page_id
FROM content_chunks cc
JOIN pages p ON p.id = cc.page_id
WHERE cc.embedding IS NULL
AND p.source_id = ${opts.sourceId}
AND NOT (COALESCE(p.frontmatter, '{}'::jsonb) ? 'embed_skip')
AND (cc.page_id, cc.chunk_index) > (${afterPid}, ${afterIdx})
ORDER BY cc.page_id, cc.chunk_index
LIMIT ${limit}
`;
return rows as unknown as StaleChunkRow[];
}
const rows = await sql`
SELECT p.slug, cc.chunk_index, cc.chunk_text, cc.chunk_source,
cc.model, cc.token_count, p.source_id, cc.page_id
FROM content_chunks cc
JOIN pages p ON p.id = cc.page_id
WHERE cc.embedding IS NULL
AND p.source_id = ${opts.sourceId}
AND NOT (COALESCE(p.frontmatter, '{}'::jsonb) ? 'embed_skip')
AND (cc.page_id, cc.chunk_index) > (${afterPid}, ${afterIdx})
ORDER BY cc.page_id, cc.chunk_index
LIMIT ${limit}
`;
return rows as unknown as StaleChunkRow[];
});
}
async deleteChunks(slug: string, opts?: { sourceId?: string }): Promise<void> {
@@ -2490,11 +2590,14 @@ export class PostgresEngine implements BrainEngine {
async countStalePagesForExtraction(opts?: { sourceId?: string; versionTs?: string }): Promise<number> {
const { where, params } = this.buildStalePagesWhere(opts);
const rows = await this.sql.unsafe(
`SELECT count(*)::int AS count FROM pages WHERE ${where}`,
params as Parameters<typeof this.sql.unsafe>[1],
);
return Number((rows[0] as { count?: number } | undefined)?.count ?? 0);
// RLS scope binding (opt-in via GBRAIN_RLS_SCOPE_BINDING).
return await this.withScopedReadTransaction(undefined, opts?.sourceId, async (tx) => {
const rows = await tx.unsafe(
`SELECT count(*)::int AS count FROM pages WHERE ${where}`,
params as Parameters<typeof tx.unsafe>[1],
);
return Number((rows[0] as { count?: number } | undefined)?.count ?? 0);
});
}
async listStalePagesForExtraction(opts: {
@@ -2511,19 +2614,22 @@ export class PostgresEngine implements BrainEngine {
}
params.push(opts.batchSize);
const limitIdx = params.length;
const rows = await this.sql.unsafe(
// #1768: project a deterministic full-µs UTC string alongside updated_at.
// to_char (not ::text — DateStyle-fragile) so extractStaleFromDB can stamp
// links_extracted_at = the exact updated_at and the staleness predicate clears.
`SELECT id, slug, source_id, type, title, compiled_truth, timeline, frontmatter, updated_at,
to_char(updated_at AT TIME ZONE 'UTC', 'YYYY-MM-DD"T"HH24:MI:SS.US"Z"') AS updated_at_iso
FROM pages
WHERE ${where}${afterClause}
ORDER BY id
LIMIT $${limitIdx}`,
params as Parameters<typeof this.sql.unsafe>[1],
);
return (rows as Record<string, unknown>[]).map(rowToStalePage);
// RLS scope binding (opt-in via GBRAIN_RLS_SCOPE_BINDING).
return await this.withScopedReadTransaction(undefined, opts.sourceId, async (tx) => {
const rows = await tx.unsafe(
// #1768: project a deterministic full-µs UTC string alongside updated_at.
// to_char (not ::text — DateStyle-fragile) so extractStaleFromDB can stamp
// links_extracted_at = the exact updated_at and the staleness predicate clears.
`SELECT id, slug, source_id, type, title, compiled_truth, timeline, frontmatter, updated_at,
to_char(updated_at AT TIME ZONE 'UTC', 'YYYY-MM-DD"T"HH24:MI:SS.US"Z"') AS updated_at_iso
FROM pages
WHERE ${where}${afterClause}
ORDER BY id
LIMIT $${limitIdx}`,
params as Parameters<typeof tx.unsafe>[1],
);
return (rows as Record<string, unknown>[]).map(rowToStalePage);
});
}
async markPagesExtractedBatch(refs: Array<{ slug: string; source_id: string; extractedAt?: string }>, defaultExtractedAt: string): Promise<void> {
@@ -2671,32 +2777,47 @@ export class PostgresEngine implements BrainEngine {
}
async getLinks(slug: string, opts?: { sourceId?: string; sourceIds?: string[] }): Promise<Link[]> {
const sql = this.sql;
// #2200: federated grant scopes ALL THREE page endpoints — from, to, AND the
// origin (the page that authored the edge, surfaced as origin_slug). Scoping
// only from+to would still leak an out-of-grant origin's slug; the origin
// LEFT JOIN carries the same ANY($) filter so origin_slug nulls out of grant.
// Remote MCP clients always land here.
if (opts?.sourceIds && opts.sourceIds.length > 0) {
const ids = opts.sourceIds;
const rows = await sql`
SELECT f.slug as from_slug, t.slug as to_slug,
l.link_type, l.context, l.link_source,
o.slug as origin_slug, l.origin_field
FROM links l
JOIN pages f ON f.id = l.from_page_id
JOIN pages t ON t.id = l.to_page_id
LEFT JOIN pages o ON o.id = l.origin_page_id AND o.source_id = ANY(${ids}::text[])
WHERE f.slug = ${slug} AND f.source_id = ANY(${ids}::text[]) AND t.source_id = ANY(${ids}::text[])
`;
return rows as unknown as Link[];
}
// v0.31.8 (D16) + #2200: the federated arm above is the first branch; the
// two below preserve pre-v0.31.8 semantics. Without opts.sourceId, no source
// filter (cross-source view for internal callers). With opts.sourceId, scope
// the from-page lookup. See pglite-engine.ts:getLinks for context.
if (opts?.sourceId) {
const rows = await sql`
// Two layers of defense (see getPage for the full pattern):
// 1. RLS scope binding (opt-in via GBRAIN_RLS_SCOPE_BINDING)
// 2. App-layer source filter (#2200 federated)
return await this.withScopedReadTransaction(opts?.sourceIds, opts?.sourceId, async (tx) => {
// #2200: federated grant scopes ALL THREE page endpoints — from, to, AND
// the origin (the page that authored the edge, surfaced as origin_slug).
// Scoping only from+to would still leak an out-of-grant origin's slug; the
// origin LEFT JOIN carries the same ANY($) filter so origin_slug nulls
// out of grant. Remote MCP clients always land here.
if (opts?.sourceIds && opts.sourceIds.length > 0) {
const ids = opts.sourceIds;
const rows = await tx`
SELECT f.slug as from_slug, t.slug as to_slug,
l.link_type, l.context, l.link_source,
o.slug as origin_slug, l.origin_field
FROM links l
JOIN pages f ON f.id = l.from_page_id
JOIN pages t ON t.id = l.to_page_id
LEFT JOIN pages o ON o.id = l.origin_page_id AND o.source_id = ANY(${ids}::text[])
WHERE f.slug = ${slug} AND f.source_id = ANY(${ids}::text[]) AND t.source_id = ANY(${ids}::text[])
`;
return rows as unknown as Link[];
}
// v0.31.8 (D16) + #2200: the federated arm above is the first branch; the
// two below preserve pre-v0.31.8 semantics. Without opts.sourceId, no
// source filter (cross-source view for internal callers). With
// opts.sourceId, scope the from-page lookup.
if (opts?.sourceId) {
const rows = await tx`
SELECT f.slug as from_slug, t.slug as to_slug,
l.link_type, l.context, l.link_source,
o.slug as origin_slug, l.origin_field
FROM links l
JOIN pages f ON f.id = l.from_page_id
JOIN pages t ON t.id = l.to_page_id
LEFT JOIN pages o ON o.id = l.origin_page_id
WHERE f.slug = ${slug} AND f.source_id = ${opts.sourceId}
`;
return rows as unknown as Link[];
}
const rows = await tx`
SELECT f.slug as from_slug, t.slug as to_slug,
l.link_type, l.context, l.link_source,
o.slug as origin_slug, l.origin_field
@@ -2704,45 +2825,49 @@ export class PostgresEngine implements BrainEngine {
JOIN pages f ON f.id = l.from_page_id
JOIN pages t ON t.id = l.to_page_id
LEFT JOIN pages o ON o.id = l.origin_page_id
WHERE f.slug = ${slug} AND f.source_id = ${opts.sourceId}
WHERE f.slug = ${slug}
`;
return rows as unknown as Link[];
}
const rows = await sql`
SELECT f.slug as from_slug, t.slug as to_slug,
l.link_type, l.context, l.link_source,
o.slug as origin_slug, l.origin_field
FROM links l
JOIN pages f ON f.id = l.from_page_id
JOIN pages t ON t.id = l.to_page_id
LEFT JOIN pages o ON o.id = l.origin_page_id
WHERE f.slug = ${slug}
`;
return rows as unknown as Link[];
});
}
async getBacklinks(slug: string, opts?: { sourceId?: string; sourceIds?: string[] }): Promise<Link[]> {
const sql = this.sql;
// #2200: federated grant scopes all three endpoints (mirrors getLinks) — the
// referrer (from), the queried page (to), AND the origin — so neither a
// foreign referrer nor a foreign origin slug is disclosed to the caller.
if (opts?.sourceIds && opts.sourceIds.length > 0) {
const ids = opts.sourceIds;
const rows = await sql`
SELECT f.slug as from_slug, t.slug as to_slug,
l.link_type, l.context, l.link_source,
o.slug as origin_slug, l.origin_field
FROM links l
JOIN pages f ON f.id = l.from_page_id
JOIN pages t ON t.id = l.to_page_id
LEFT JOIN pages o ON o.id = l.origin_page_id AND o.source_id = ANY(${ids}::text[])
WHERE t.slug = ${slug} AND t.source_id = ANY(${ids}::text[]) AND f.source_id = ANY(${ids}::text[])
`;
return rows as unknown as Link[];
}
// v0.31.8 (D16) + #2200: federated arm above is first; two below mirror getLinks.
if (opts?.sourceId) {
const rows = await sql`
// Two layers of defense (see getPage for the full pattern):
// 1. RLS scope binding (opt-in via GBRAIN_RLS_SCOPE_BINDING)
// 2. App-layer source filter (#2200 federated)
return await this.withScopedReadTransaction(opts?.sourceIds, opts?.sourceId, async (tx) => {
// #2200: federated grant scopes all three endpoints (mirrors getLinks) —
// the referrer (from), the queried page (to), AND the origin — so neither
// a foreign referrer nor a foreign origin slug is disclosed to the caller.
if (opts?.sourceIds && opts.sourceIds.length > 0) {
const ids = opts.sourceIds;
const rows = await tx`
SELECT f.slug as from_slug, t.slug as to_slug,
l.link_type, l.context, l.link_source,
o.slug as origin_slug, l.origin_field
FROM links l
JOIN pages f ON f.id = l.from_page_id
JOIN pages t ON t.id = l.to_page_id
LEFT JOIN pages o ON o.id = l.origin_page_id AND o.source_id = ANY(${ids}::text[])
WHERE t.slug = ${slug} AND t.source_id = ANY(${ids}::text[]) AND f.source_id = ANY(${ids}::text[])
`;
return rows as unknown as Link[];
}
// v0.31.8 (D16) + #2200: federated arm above is first; two below mirror getLinks.
if (opts?.sourceId) {
const rows = await tx`
SELECT f.slug as from_slug, t.slug as to_slug,
l.link_type, l.context, l.link_source,
o.slug as origin_slug, l.origin_field
FROM links l
JOIN pages f ON f.id = l.from_page_id
JOIN pages t ON t.id = l.to_page_id
LEFT JOIN pages o ON o.id = l.origin_page_id
WHERE t.slug = ${slug} AND t.source_id = ${opts.sourceId}
`;
return rows as unknown as Link[];
}
const rows = await tx`
SELECT f.slug as from_slug, t.slug as to_slug,
l.link_type, l.context, l.link_source,
o.slug as origin_slug, l.origin_field
@@ -2750,45 +2875,36 @@ export class PostgresEngine implements BrainEngine {
JOIN pages f ON f.id = l.from_page_id
JOIN pages t ON t.id = l.to_page_id
LEFT JOIN pages o ON o.id = l.origin_page_id
WHERE t.slug = ${slug} AND t.source_id = ${opts.sourceId}
WHERE t.slug = ${slug}
`;
return rows as unknown as Link[];
}
const rows = await sql`
SELECT f.slug as from_slug, t.slug as to_slug,
l.link_type, l.context, l.link_source,
o.slug as origin_slug, l.origin_field
FROM links l
JOIN pages f ON f.id = l.from_page_id
JOIN pages t ON t.id = l.to_page_id
LEFT JOIN pages o ON o.id = l.origin_page_id
WHERE t.slug = ${slug}
`;
return rows as unknown as Link[];
});
}
async listLinkSources(
opts?: { sourceId?: string; sourceIds?: string[] },
): Promise<{ link_source: string | null; count: number }[]> {
const sql = this.sql;
// v114 (#1941): distinct provenances + counts for `gbrain link-sources`.
// Scope by the FROM page's source (consistent with getLinks). Federated
// {sourceIds} takes precedence over scalar {sourceId}; neither = unscoped.
const sourceCondition =
opts?.sourceIds && opts.sourceIds.length > 0
? sql`WHERE f.source_id = ANY(${opts.sourceIds}::text[])`
: opts?.sourceId
? sql`WHERE f.source_id = ${opts.sourceId}`
: sql``;
const rows = await sql`
SELECT l.link_source, COUNT(*)::int AS count
FROM links l
JOIN pages f ON f.id = l.from_page_id
${sourceCondition}
GROUP BY l.link_source
ORDER BY count DESC, l.link_source ASC NULLS LAST
`;
return rows as unknown as { link_source: string | null; count: number }[];
// RLS scope binding (opt-in via GBRAIN_RLS_SCOPE_BINDING).
return await this.withScopedReadTransaction(opts?.sourceIds, opts?.sourceId, async (tx) => {
// v114 (#1941): distinct provenances + counts for `gbrain link-sources`.
// Scope by the FROM page's source (consistent with getLinks). Federated
// {sourceIds} takes precedence over scalar {sourceId}; neither = unscoped.
const sourceCondition =
opts?.sourceIds && opts.sourceIds.length > 0
? tx`WHERE f.source_id = ANY(${opts.sourceIds}::text[])`
: opts?.sourceId
? tx`WHERE f.source_id = ${opts.sourceId}`
: tx``;
const rows = await tx`
SELECT l.link_source, COUNT(*)::int AS count
FROM links l
JOIN pages f ON f.id = l.from_page_id
${sourceCondition}
GROUP BY l.link_source
ORDER BY count DESC, l.link_source ASC NULLS LAST
`;
return rows as unknown as { link_source: string | null; count: number }[];
});
}
async findByTitleFuzzy(
test/postgres-engine-rls-scope.test.ts
+181
View File
@@ -0,0 +1,181 @@
/**
* withScopedReadTransaction — opt-in Postgres RLS source-scope binding
* (GBRAIN_RLS_SCOPE_BINDING, lands community PR #2387).
*
* Behavioral pins, no real DB (fake postgres.js sql handle):
* - flag OFF (default): TRUE pass-through — callback receives the shared
* pool handle directly, no sql.begin(), no set_config. This is the
* #1794-class guard: reads must not gain a per-read pool hold.
* - flag OFF + alwaysTransaction (the search methods' SET LOCAL path):
* sql.begin() opens, still no set_config — identical to master's wrap.
* - flag ON: sql.begin() + SELECT set_config('app.scopes', $1, true)
* with federated-array > scalar > '*' precedence, and the CSV value
* carried as a BOUND PARAMETER, never interpolated into the SQL text.
*/
import { describe, test, expect } from 'bun:test';
import { PostgresEngine } from '../src/core/postgres-engine.ts';
import { withEnv } from './helpers/with-env.ts';
type Recorded = { text: string; params: unknown[] };
function makeFakeSql() {
const queries: Recorded[] = [];
let beginCalls = 0;
const record = (strings: TemplateStringsArray, ...params: unknown[]) => {
// Join the literal segments with a placeholder marker so the test can
// assert the exact SQL text shape around each bound parameter.
queries.push({ text: strings.join('${}'), params });
return Promise.resolve([]);
};
const sql = ((strings: TemplateStringsArray, ...params: unknown[]) =>
record(strings, ...params)) as unknown as Record<string, unknown> & {
(strings: TemplateStringsArray, ...params: unknown[]): Promise<unknown[]>;
begin: (cb: (tx: unknown) => Promise<unknown>) => Promise<unknown>;
};
const tx = ((strings: TemplateStringsArray, ...params: unknown[]) =>
record(strings, ...params)) as unknown as Record<string, unknown>;
sql.begin = async (cb: (t: unknown) => Promise<unknown>) => {
beginCalls++;
return await cb(tx);
};
return { sql, tx, queries, beginCalls: () => beginCalls };
}
function makeEngine(fake: ReturnType<typeof makeFakeSql>) {
const e = new PostgresEngine();
(e as unknown as { _sql: unknown })._sql = fake.sql;
(e as unknown as { _connectionStyle: string })._connectionStyle = 'instance';
// private method, invoked directly for the pin
return e as unknown as {
withScopedReadTransaction<T>(
sourceIds: string[] | undefined,
sourceId: string | undefined,
cb: (tx: unknown) => Promise<T>,
opts?: { alwaysTransaction?: boolean },
): Promise<T>;
};
}
function setConfigQueries(queries: Recorded[]): Recorded[] {
return queries.filter((q) => q.text.includes('set_config'));
}
describe('withScopedReadTransaction / flag off (default)', () => {
test('true pass-through: callback gets the shared pool handle, no begin, no set_config', async () => {
await withEnv({ GBRAIN_RLS_SCOPE_BINDING: undefined }, async () => {
const fake = makeFakeSql();
const engine = makeEngine(fake);
let received: unknown;
const result = await engine.withScopedReadTransaction(undefined, 'src-a', async (tx) => {
received = tx;
return 42;
});
expect(result).toBe(42);
expect(received).toBe(fake.sql); // the pool handle itself, not a tx
expect(fake.beginCalls()).toBe(0);
expect(setConfigQueries(fake.queries)).toHaveLength(0);
});
});
test('explicit "0" is off too', async () => {
await withEnv({ GBRAIN_RLS_SCOPE_BINDING: '0' }, async () => {
const fake = makeFakeSql();
const engine = makeEngine(fake);
await engine.withScopedReadTransaction(['a', 'b'], undefined, async () => null);
expect(fake.beginCalls()).toBe(0);
expect(setConfigQueries(fake.queries)).toHaveLength(0);
});
});
test('alwaysTransaction keeps master\'s sql.begin() wrap, still no set_config', async () => {
await withEnv({ GBRAIN_RLS_SCOPE_BINDING: undefined }, async () => {
const fake = makeFakeSql();
const engine = makeEngine(fake);
let received: unknown;
await engine.withScopedReadTransaction(
undefined,
'src-a',
async (tx) => {
received = tx;
return null;
},
{ alwaysTransaction: true },
);
expect(fake.beginCalls()).toBe(1);
expect(received).toBe(fake.tx); // a transaction handle this time
expect(setConfigQueries(fake.queries)).toHaveLength(0);
});
});
});
describe('withScopedReadTransaction / flag on', () => {
test('emits set_config(\'app.scopes\', ...) inside a transaction, before the callback', async () => {
await withEnv({ GBRAIN_RLS_SCOPE_BINDING: '1' }, async () => {
const fake = makeFakeSql();
const engine = makeEngine(fake);
let queriesAtCallback = -1;
await engine.withScopedReadTransaction(undefined, 'src-a', async () => {
queriesAtCallback = fake.queries.length;
return null;
});
expect(fake.beginCalls()).toBe(1);
const sc = setConfigQueries(fake.queries);
expect(sc).toHaveLength(1);
expect(sc[0].params).toEqual(['src-a']);
// set_config was emitted before the callback ran
expect(queriesAtCallback).toBe(1);
expect(fake.queries[0]).toBe(sc[0]);
});
});
test('"true" also enables', async () => {
await withEnv({ GBRAIN_RLS_SCOPE_BINDING: 'true' }, async () => {
const fake = makeFakeSql();
const engine = makeEngine(fake);
await engine.withScopedReadTransaction(undefined, 'src-a', async () => null);
expect(setConfigQueries(fake.queries)).toHaveLength(1);
});
});
test('federated array wins over scalar: CSV of sourceIds', async () => {
await withEnv({ GBRAIN_RLS_SCOPE_BINDING: '1' }, async () => {
const fake = makeFakeSql();
const engine = makeEngine(fake);
await engine.withScopedReadTransaction(['a', 'b', 'c'], 'ignored-scalar', async () => null);
expect(setConfigQueries(fake.queries)[0].params).toEqual(['a,b,c']);
});
});
test('empty federated array falls back to scalar', async () => {
await withEnv({ GBRAIN_RLS_SCOPE_BINDING: '1' }, async () => {
const fake = makeFakeSql();
const engine = makeEngine(fake);
await engine.withScopedReadTransaction([], 'src-b', async () => null);
expect(setConfigQueries(fake.queries)[0].params).toEqual(['src-b']);
});
});
test("unscoped (no sourceIds, no sourceId) binds '*'", async () => {
await withEnv({ GBRAIN_RLS_SCOPE_BINDING: '1' }, async () => {
const fake = makeFakeSql();
const engine = makeEngine(fake);
await engine.withScopedReadTransaction(undefined, undefined, async () => null);
expect(setConfigQueries(fake.queries)[0].params).toEqual(['*']);
});
});
test('the scopes CSV is a BOUND PARAMETER, never interpolated into SQL text', async () => {
await withEnv({ GBRAIN_RLS_SCOPE_BINDING: '1' }, async () => {
const fake = makeFakeSql();
const engine = makeEngine(fake);
const hostile = "x','y'); DROP TABLE pages; --";
await engine.withScopedReadTransaction(undefined, hostile, async () => null);
const sc = setConfigQueries(fake.queries)[0];
// Exact literal-segment shape: the value slot is the tagged-template hole.
expect(sc.text).toBe("SELECT set_config('app.scopes', ${}, true)");
expect(sc.params).toEqual([hostile]);
expect(sc.text).not.toContain(hostile);
});
});
});
test/postgres-engine.test.ts
+24 -4
View File
@@ -48,14 +48,34 @@ describe('postgres-engine / search path timeout isolation', () => {
expect(bare).toBeNull();
});
test('searchKeyword wraps its query in sql.begin()', () => {
test('searchKeyword wraps its query in a transaction (via withScopedReadTransaction alwaysTransaction)', () => {
// Post-RLS-scope-binding invariant: the search methods route through
// withScopedReadTransaction with alwaysTransaction: true, which
// guarantees a sql.begin() wrap in BOTH modes — flag off (identical to
// master's pre-helper wrap) and flag on (scoped transaction with
// set_config). See the helper tests in
// test/postgres-engine-rls-scope.test.ts for the behavioral pins.
const fn = extractMethod(SRC, 'searchKeyword');
expect(fn).toMatch(/sql\.begin\s*\(\s*async\s+sql\s*=>/);
expect(fn).toMatch(/withScopedReadTransaction\s*\(/);
expect(fn).toMatch(/alwaysTransaction:\s*true/);
});
test('searchVector wraps its query in sql.begin()', () => {
test('searchVector wraps its query in a transaction (via withScopedReadTransaction alwaysTransaction)', () => {
const fn = extractMethod(SRC, 'searchVector');
expect(fn).toMatch(/sql\.begin\s*\(\s*async\s+sql\s*=>/);
expect(fn).toMatch(/withScopedReadTransaction\s*\(/);
expect(fn).toMatch(/alwaysTransaction:\s*true/);
});
test('withScopedReadTransaction owns the sql.begin() wrap (and only opens it when needed)', () => {
// (extractMethod can't grab this one: `private async ...<T>(`.)
const stripped = stripComments(SRC);
// The transaction lives in the helper...
expect(stripped).toMatch(/this\.sql\.begin\s*\(/);
// ...and the flag-off / non-alwaysTransaction path is a true
// pass-through on the shared pool — no per-read transaction hold.
expect(stripped).toMatch(
/if\s*\(!this\.rlsScopeBindingEnabled\s*&&\s*!opts\?\.alwaysTransaction\)\s*\{\s*return\s+await\s+callback\(this\.sql\);/,
);
});
test('both search methods use SET LOCAL for the timeout', () => {