fix(db): restringe privilegios e torna migrations atomicas

Co-authored-by: Cursor <cursoragent@cursor.com>

migrations/0003_rls.sql

@@ -93,12 +93,15 @@ returns boolean
 language sql
 stable
 security definer
-set search_path = public
+set search_path = pg_catalog
 as $$
   select exists (
-    select 1 from public.spreadsheet_jobs j
+    select 1
+    from public.spreadsheet_jobs j
+    join public.spreadsheet_models m on m.id = j.model_id
     where j.model_id = model
       and j.status in ('validating', 'ready', 'queued', 'processing')
+      and public.is_member(m.workspace_id)
   )
 $$;
 
@@ -208,14 +211,17 @@ create policy models_update on public.spreadsheet_models
   with check (public.can_manage_adapter(workspace_id));
 
 drop policy if exists jobs_all on public.spreadsheet_jobs;
+drop policy if exists jobs_select on public.spreadsheet_jobs;
 create policy jobs_select on public.spreadsheet_jobs
   for select to app_user
   using (user_id = public.auth_uid() and public.is_member(workspace_id) and public.module_enabled(workspace_id));
 
+drop policy if exists jobs_insert on public.spreadsheet_jobs;
 create policy jobs_insert on public.spreadsheet_jobs
   for insert to app_user
   with check (user_id = public.auth_uid() and public.is_member(workspace_id) and public.module_enabled(workspace_id));
 
+drop policy if exists jobs_update on public.spreadsheet_jobs;
 create policy jobs_update on public.spreadsheet_jobs
   for update to app_user
   using (user_id = public.auth_uid() and public.is_member(workspace_id) and public.module_enabled(workspace_id))
@@ -229,6 +235,7 @@ create policy sources_select on public.spreadsheet_job_sources
     where j.id = job_id and j.user_id = public.auth_uid() and public.module_enabled(j.workspace_id)
   ));
 
+drop policy if exists sources_insert on public.spreadsheet_job_sources;
 create policy sources_insert on public.spreadsheet_job_sources
   for insert to app_user
   with check (exists (
@@ -236,6 +243,7 @@ create policy sources_insert on public.spreadsheet_job_sources
     where j.id = job_id and j.user_id = public.auth_uid() and public.module_enabled(j.workspace_id)
   ));
 
+drop policy if exists sources_update on public.spreadsheet_job_sources;
 create policy sources_update on public.spreadsheet_job_sources
   for update to app_user
   using (exists (
@@ -247,6 +255,7 @@ create policy sources_update on public.spreadsheet_job_sources
     where j.id = job_id and j.user_id = public.auth_uid() and public.module_enabled(j.workspace_id)
   ));
 
+drop policy if exists sources_delete on public.spreadsheet_job_sources;
 create policy sources_delete on public.spreadsheet_job_sources
   for delete to app_user
   using (exists (
@@ -263,6 +272,7 @@ create policy mappings_select on public.spreadsheet_job_mappings
     where s.id = job_source_id and j.user_id = public.auth_uid()
   ));
 
+drop policy if exists mappings_insert on public.spreadsheet_job_mappings;
 create policy mappings_insert on public.spreadsheet_job_mappings
   for insert to app_user
   with check (exists (
@@ -271,6 +281,7 @@ create policy mappings_insert on public.spreadsheet_job_mappings
     where s.id = job_source_id and j.user_id = public.auth_uid()
   ));
 
+drop policy if exists mappings_delete on public.spreadsheet_job_mappings;
 create policy mappings_delete on public.spreadsheet_job_mappings
   for delete to app_user
   using (exists (
@@ -282,3 +293,37 @@ create policy mappings_delete on public.spreadsheet_job_mappings
 grant usage on schema public to app_user;
 grant select, insert, update, delete on all tables in schema public to app_user;
 grant usage, select on all sequences in schema public to app_user;
+
+-- schema_migrations não tem RLS. A escrita fica só com o dono que aplica as migrations.
+revoke insert, update, delete, truncate, references, trigger on table public.schema_migrations from public, app_user;
+
+do $priv$
+declare
+  versao int := current_setting('server_version_num')::int;
+  escrita text := 'insert, update, delete, truncate, references, trigger';
+  papel text;
+begin
+  if versao >= 170000 then
+    escrita := escrita || ', maintain';
+    execute 'revoke maintain on table public.schema_migrations from public, app_user';
+  end if;
+  foreach papel in array array['anon', 'authenticated', 'service_role', 'adaptador_web', 'adaptador_worker']
+  loop
+    if exists (select 1 from pg_catalog.pg_roles where rolname = papel) then
+      execute format(
+        'revoke %s on table public.profiles, public.workspaces, public.workspace_members, public.workspace_module_flags, public.audit_log, public.spreadsheet_adapter_settings, public.spreadsheet_models, public.spreadsheet_jobs, public.spreadsheet_job_sources, public.spreadsheet_job_mappings, public.schema_migrations from %I',
+        escrita,
+        papel
+      );
+      if papel in ('anon', 'authenticated', 'service_role') then
+        execute format('revoke select on table public.schema_migrations from %I', papel);
+        execute format(
+          'alter default privileges in schema public revoke %s on tables from %I',
+          escrita,
+          papel
+        );
+      end if;
+    end if;
+  end loop;
+end
+$priv$;

src/infra/db/migrate.ts

@@ -16,6 +16,8 @@ import type { Db } from "./types";
  */
 
 const MIGRATIONS_FOLDER = "migrations";
+/** Trava só durante a transação do migrate. A tabela de controle permanece id e applied_at. */
+const MIGRATION_LOCK = 872341001;
 
 export type MigrateOptions = {
   /** Aplica o shim de identidade. Padrão: `ADAPTER_APPLY_IDENTITY_SHIM`. */
@@ -51,27 +53,35 @@ export function assertManifestMatches(files: string[]): void {
 
 export async function migrate(db: Db, options: MigrateOptions = {}): Promise<string[]> {
   const applyShim = options.applyIdentityShim ?? identityShimEnabled();
-
-  await db.exec(`
-    create table if not exists public.schema_migrations (
-      id text primary key,
-      applied_at timestamptz not null default now()
-    );
-  `);
   const files = await listMigrationFiles();
   assertManifestMatches(files);
-  const appliedRows = await db.withService((query) => query.query<{ id: string }>("select id from public.schema_migrations"));
+  const sqlByFile = new Map<string, string>();
-  const applied = new Set(appliedRows.map((row) => row.id));
-  const ran: string[] = [];
   for (const file of files) {
-    if (applied.has(file)) continue;
+    sqlByFile.set(file, await readFile(path.join(process.cwd(), MIGRATIONS_FOLDER, file), "utf8"));
-    if (isShimMigration(file) && !applyShim) continue;
-    const sql = await readFile(path.join(process.cwd(), MIGRATIONS_FOLDER, file), "utf8");
-    await db.exec(sql);
-    await db.withService((query) => query.query("insert into public.schema_migrations (id) values ($1)", [file]));
-    ran.push(file);
   }
-  return ran;
+
+  return db.withTransaction(async (tx) => {
+    await tx.exec(`select pg_catalog.pg_advisory_xact_lock(${MIGRATION_LOCK})`);
+    await tx.exec(`
+      create table if not exists public.schema_migrations (
+        id text primary key,
+        applied_at timestamptz not null default now()
+      );
+    `);
+    const appliedRows = await tx.query<{ id: string }>("select id from public.schema_migrations");
+    const applied = new Set(appliedRows.map((row) => row.id));
+    const ran: string[] = [];
+    for (const file of files) {
+      if (applied.has(file)) continue;
+      if (isShimMigration(file) && !applyShim) continue;
+      const sql = sqlByFile.get(file);
+      if (!sql) throw new Error(`migration ausente: ${file}`);
+      await tx.exec(sql);
+      await tx.query("insert into public.schema_migrations (id) values ($1)", [file]);
+      ran.push(file);
+    }
+    return ran;
+  });
 }
 
 export { EXPECTED_MIGRATIONS, SHIM_MIGRATIONS, isShimMigration, pendingMigrations } from "./migration-manifest";

src/infra/db/pglite.ts

@@ -1,5 +1,5 @@
 import { PGlite } from "@electric-sql/pglite";
-import { createMutex, type Db, type Query, type Row, assertUuid } from "./types";
+import { createMutex, type Db, type Query, type Row, type Tx, assertUuid } from "./types";
 
 type PgClient = {
   exec(sql: string): Promise<unknown>;
@@ -44,6 +44,28 @@ export function wrapPglite(client: PgClient): Db {
     close() {
       return client.close();
     },
+    withTransaction(fn) {
+      return exclusive(async () => {
+        await client.exec("begin");
+        const tx: Tx = {
+          exec(sql) {
+            return client.exec(sql).then(() => undefined);
+          },
+          async query<T = Row>(sql: string, params: unknown[] = []) {
+            const result = await client.query(sql, params);
+            return result.rows as T[];
+          },
+        };
+        try {
+          const value = await fn(tx);
+          await client.exec("commit");
+          return value;
+        } catch (error) {
+          await client.exec("rollback");
+          throw error;
+        }
+      });
+    },
   };
 }
 

src/infra/db/postgres.ts

@@ -1,6 +1,6 @@
 import postgres from "postgres";
 import type { PostgresClientOptions } from "./connection-policy";
-import { assertUuid, type Db, type Query, type Row } from "./types";
+import { assertUuid, type Db, type Query, type Row, type Tx } from "./types";
 
 export function createPostgresDb(connectionString: string, options?: PostgresClientOptions): Db {
   const sql = postgres(connectionString, options ?? { max: 8, prepare: false });
@@ -33,5 +33,19 @@ export function createPostgresDb(connectionString: string, options?: PostgresCli
     async close() {
       await sql.end();
     },
+    async withTransaction<T>(fn: (tx: Tx) => Promise<T>): Promise<T> {
+      const result = await sql.begin(async (tx) => {
+        const adapter: Tx = {
+          async exec(text: string) {
+            await tx.unsafe(text);
+          },
+          async query<R = Row>(text: string, params: unknown[] = []) {
+            return tx.unsafe(text, params as never[]) as unknown as R[];
+          },
+        };
+        return fn(adapter);
+      });
+      return result as T;
+    },
   };
 }

src/infra/db/types.ts

@@ -4,10 +4,17 @@ export interface Query {
   query<T = Row>(sql: string, params?: unknown[]): Promise<T[]>;
 }
 
+export interface Tx {
+  exec(sql: string): Promise<void>;
+  query<T = Row>(sql: string, params?: unknown[]): Promise<T[]>;
+}
+
 export interface Db {
   exec(sql: string): Promise<void>;
   withUser<T>(userId: string, fn: (query: Query) => Promise<T>): Promise<T>;
   withService<T>(fn: (query: Query) => Promise<T>): Promise<T>;
+  /** Uma conexão, uma transação. O callback inteiro confirma ou reverte junto. */
+  withTransaction<T>(fn: (tx: Tx) => Promise<T>): Promise<T>;
   close(): Promise<void>;
 }