feat: transactions and rls rules

Implements Row-Level Security (RLS) policies and SQL transaction management.

Row-Level Security (RLS):
- Database function get_current_organization_id() reads app.current_organization_id from session
- RLS policies on tables with organization_id filter rows automatically
- Schema uses organizationRLSPolicy() helper to enable RLS per table

SQL Transactions:
- Domain Layer (@domain/shared): SqlClient interface for database operations
- Platform Layer (@platform/db-postgres): SqlClientLive with automatic RLS context
- App Layer (apps/*): Boundaries provide SqlClientLive with organization context

Usage patterns:
- Repositories use sqlClient.query() for single operations
- Use cases use sqlClient.transaction() for multi-step operations
- Routes provide SqlClientLive(client, organizationId) for RLS enforcement

Key behaviors:
- Every transaction sets app.current_organization_id session variable
- Nested transactions share connection (pass-through proxy)
- Domain errors propagate; failures trigger automatic rollback

Author: geclos

AGENTS.md

@@ -240,7 +240,109 @@ Base config: `tsconfig.base.json`
 - Weaviate adapter stack lives in `packages/platform/db-weaviate`
 - Domain models are independent from table/row shapes
 - Mapping from DB rows to domain objects belongs in platform adapters
-- **Apps use pool-based connections**: Use `createPostgresPool()` in `apps/*/clients.ts` for direct pool access
+- **Apps use SqlClient for all DB access**: Boundaries provide `SqlClientLive` layer with organization context for RLS enforcement
+
+### SqlClient and Row-Level Security (RLS)
+
+All Postgres access flows through `SqlClient`—a domain-level service that abstracts database operations and enforces organization scoping via RLS.
+
+**Architecture:**
+- **Domain Layer** (`@domain/shared`): `SqlClient` interface with `transaction()` and `query()` methods
+- **Platform Layer** (`@platform/db-postgres`): `SqlClientLive` implementation with automatic RLS context setting
+- **App Layer** (`apps/*`): Boundaries provide `SqlClientLive` with the request's organization context
+
+**Key behaviors:**
+- Every transaction automatically sets `app.current_organization_id` session variable
+- RLS policies filter all queries by this organization ID at the database level
+- Nested transactions share the same connection (pass-through proxy—no nested transaction overhead)
+- Domain errors propagate through Effect error channel; database errors become `RepositoryError`
+
+**Usage in boundaries (apps):**
+
+```typescript
+// apps/api/src/routes/projects.ts
+import { SqlClientLive } from "@platform/db-postgres"
+import { ProjectRepositoryLive } from "@platform/db-postgres"
+
+app.openapi(createProjectRoute, async (c) => {
+  const project = await Effect.runPromise(
+    createProjectUseCase(input).pipe(
+      Effect.provide(ProjectRepositoryLive),
+      Effect.provide(SqlClientLive(c.var.postgresClient, c.var.organization.id)),
+    ),
+  )
+  return c.json(toProjectResponse(project), 201)
+})
+```
+
+```typescript
+// apps/web/src/domains/projects/projects.functions.ts
+import { getPostgresClient } from "../../server/clients.ts"
+
+export const createProject = createServerFn({ method: "POST" })
+  .handler(async ({ data }) => {
+    const { organizationId } = await requireSession()
+    const client = getPostgresClient()
+    
+    const project = await Effect.runPromise(
+      createProjectUseCase({...}).pipe(
+        Effect.provide(ProjectRepositoryLive),
+        Effect.provide(SqlClientLive(client, organizationId)),
+      )
+    )
+    return toRecord(project)
+  })
+```
+
+**Usage in use-cases (multi-operation transactions):**
+
+```typescript
+// packages/domain/auth/src/use-cases/complete-auth-intent.ts
+export const completeAuthIntentUseCase = (input) =>
+  Effect.gen(function* () {
+    const sqlClient = yield* SqlClient
+    
+    // Wraps multi-step operation in single transaction with RLS
+    yield* sqlClient.transaction(handleIntentByType(intent, input.session))
+  })
+
+const handleSignup = (intent, session) =>
+  Effect.gen(function* () {
+    const users = yield* UserRepository
+    const memberships = yield* MembershipRepository
+    
+    // All operations share the same transaction + RLS context
+    const organization = yield* createOrganizationUseCase({...})
+    yield* memberships.save(createMembership({...}))
+    yield* users.setNameIfMissing({...})
+  })
+```
+
+**Usage in repositories (single operations):**
+
+```typescript
+// packages/platform/db-postgres/src/repositories/project-repository.ts
+export const ProjectRepositoryLive = Layer.effect(
+  ProjectRepository,
+  Effect.gen(function* () {
+    const sqlClient = (yield* SqlClient) as SqlClientShape<Operator>
+    
+    return {
+      findById: (id) =>
+        sqlClient
+          .query((db) => db.select().from(projects).where(eq(projects.id, id)))
+          .pipe(Effect.flatMap(...)),
+          
+      save: (project) =>
+        Effect.gen(function* () {
+          yield* sqlClient.query((db) =>
+            db.insert(projects).values(row).onConflictDoUpdate({...})
+          )
+        }),
+    }
+  })
+)
+```
 
 ### Postgres Management
 
@@ -610,17 +712,41 @@ The web app uses a **server-centric, query-driven** architecture built on the Ta
 **Server Functions** — All data fetching and mutations use `createServerFn` from `@tanstack/react-start`:
 
 ```typescript
+import { Effect } from "effect"
+import { ProjectRepository, createProjectUseCase } from "@domain/projects"
+import { ProjectRepositoryLive, SqlClientLive } from "@platform/db-postgres"
+import { getPostgresClient } from "../../server/clients.ts"
+
 // Query (GET)
 export const listProjects = createServerFn({ method: "GET" }).handler(async () => {
   const { organizationId } = await requireSession()
-  const { db } = getPostgresClient()
-  // compose domain use-cases...
+  const client = getPostgresClient()
+  
+  return await Effect.runPromise(
+    Effect.gen(function* () {
+      const repo = yield* ProjectRepository
+      return yield* repo.findAll()
+    }).pipe(
+      Effect.provide(ProjectRepositoryLive),
+      Effect.provide(SqlClientLive(client, organizationId))
+    )
+  )
 })
 
 // Mutation (POST) with Zod validation
 export const createProject = createServerFn({ method: "POST" })
-  .inputValidator(z.object({ name: z.string().min(1) }))
-  .handler(async ({ data }) => { ... })
+  .inputValidator(createProjectSchema)
+  .handler(async ({ data }) => {
+    const { userId, organizationId } = await requireSession()
+    const client = getPostgresClient()
+    
+    return await Effect.runPromise(
+      createProjectUseCase({...}).pipe(
+        Effect.provide(ProjectRepositoryLive),
+        Effect.provide(SqlClientLive(client, organizationId))
+      )
+    )
+  })
 ```
 
 Server functions live in `apps/web/src/domains/*/functions.ts`.

apps/api/src/middleware/auth.ts

@@ -1,5 +1,7 @@
+import { ApiKeyRepository } from "@domain/api-keys"
 import { OrganizationId, UnauthorizedError, UserId } from "@domain/shared"
-import { type PostgresDb, createApiKeyPostgresRepository } from "@platform/db-postgres"
+import { ApiKeyRepositoryLive, SqlClientLive } from "@platform/db-postgres"
+import type { PostgresClient } from "@platform/db-postgres"
 import { hashToken } from "@repo/utils"
 import { Effect, Option } from "effect"
 import type { Context, MiddlewareHandler, Next } from "hono"
@@ -94,51 +96,53 @@ const validateApiKey = (
   c: Context,
   token: string,
   options?: AuthMiddlewareOptions,
-): Effect.Effect<{ organizationId: string; keyId: string } | null, never> => {
+): Promise<{ organizationId: string; keyId: string } | null> => {
   const redis = c.get("redis")
-  const adminDb = options?.adminDb ?? getAdminPostgresClient().db
-  const unscopedApiKeyRepository = createApiKeyPostgresRepository(adminDb)
-  const touchBuffer = createTouchBuffer(adminDb)
+  const adminClient = options?.adminClient ?? getAdminPostgresClient()
+  const touchBuffer = createTouchBuffer(adminClient)
 
-  return Effect.gen(function* () {
-    const startTime = Date.now()
-    const tokenHash = yield* hashToken(token)
+  return Effect.runPromise(
+    Effect.gen(function* () {
+      const startTime = Date.now()
+      const tokenHash = yield* hashToken(token)
 
-    // Try cache first for consistent lookup time (keyed by hash)
-    const cached = yield* getCachedApiKey(redis, tokenHash)
+      // Try cache first for consistent lookup time (keyed by hash)
+      const cached = yield* getCachedApiKey(redis, tokenHash)
 
-    if (cached !== undefined) {
-      // Cache hit - enforce minimum time and return
-      yield* enforceMinimumTime(startTime, MIN_VALIDATION_TIME_MS)
-      return cached
-    }
+      if (cached !== undefined) {
+        // Cache hit - enforce minimum time and return
+        yield* enforceMinimumTime(startTime, MIN_VALIDATION_TIME_MS)
+        return cached
+      }
 
-    const apiKeyOption = yield* Effect.option(unscopedApiKeyRepository.findByTokenHash(tokenHash))
+      const apiKeyRepository = yield* ApiKeyRepository
+      const apiKeyOption = yield* Effect.option(apiKeyRepository.findByTokenHash(tokenHash))
 
-    if (Option.isNone(apiKeyOption)) {
-      yield* cacheApiKeyResult(redis, tokenHash, null, INVALID_KEY_TTL_SECONDS)
-      yield* enforceMinimumTime(startTime, MIN_VALIDATION_TIME_MS)
-      return null
-    }
+      if (Option.isNone(apiKeyOption)) {
+        yield* cacheApiKeyResult(redis, tokenHash, null, INVALID_KEY_TTL_SECONDS)
+        yield* enforceMinimumTime(startTime, MIN_VALIDATION_TIME_MS)
+        return null
+      }
 
-    const apiKey = apiKeyOption.value
+      const apiKey = apiKeyOption.value
 
-    const result = {
-      organizationId: apiKey.organizationId,
-      keyId: apiKey.id,
-    }
+      const result = {
+        organizationId: apiKey.organizationId,
+        keyId: apiKey.id,
+      }
 
-    // Cache successful validation for 5 minutes (keyed by hash)
-    yield* cacheApiKeyResult(redis, tokenHash, result, VALID_KEY_TTL_SECONDS)
+      // Cache successful validation for 5 minutes (keyed by hash)
+      yield* cacheApiKeyResult(redis, tokenHash, result, VALID_KEY_TTL_SECONDS)
 
-    // Use TouchBuffer for batched updates instead of fire-and-forget
-    // This reduces database writes by 90%+ by batching updates
-    touchBuffer.touch(apiKey.id)
+      // Use TouchBuffer for batched updates instead of fire-and-forget
+      // This reduces database writes by 90%+ by batching updates
+      touchBuffer.touch(apiKey.id)
 
-    // Enforce minimum time before returning
-    yield* enforceMinimumTime(startTime, MIN_VALIDATION_TIME_MS)
-    return result
-  }).pipe(Effect.orDie)
+      // Enforce minimum time before returning
+      yield* enforceMinimumTime(startTime, MIN_VALIDATION_TIME_MS)
+      return result
+    }).pipe(Effect.provide(ApiKeyRepositoryLive), Effect.provide(SqlClientLive(adminClient)), Effect.orDie),
+  )
 }
 
 /**
@@ -174,7 +178,7 @@ const authenticateWithApiKey = (
   options?: AuthMiddlewareOptions,
 ): Effect.Effect<AuthContext | null, never> => {
   return Effect.gen(function* () {
-    const result = yield* validateApiKey(c, token, options)
+    const result = yield* Effect.promise(() => validateApiKey(c, token, options))
 
     if (result) {
       const authContext: AuthContext = {
@@ -191,7 +195,7 @@ const authenticateWithApiKey = (
 }
 
 /**
- * Authenticate via API key from the Authorization: Bearer header.
+ * Main authentication effect that validates API key from Authorization header.
  */
 const authenticate = (c: Context, options?: AuthMiddlewareOptions): Effect.Effect<AuthContext, UnauthorizedError> => {
   return Effect.gen(function* () {
@@ -219,7 +223,7 @@ const authenticate = (c: Context, options?: AuthMiddlewareOptions): Effect.Effec
  * Public routes should be excluded from this middleware.
  */
 interface AuthMiddlewareOptions {
-  readonly adminDb?: PostgresDb
+  readonly adminClient?: PostgresClient
 }
 
 export const createAuthMiddleware = (options?: AuthMiddlewareOptions): MiddlewareHandler => {

apps/api/src/middleware/organization-context.ts

@@ -1,5 +1,6 @@
+import { OrganizationRepository } from "@domain/organizations"
 import { OrganizationId, PermissionError } from "@domain/shared"
-import { createOrganizationPostgresRepository } from "@platform/db-postgres"
+import { OrganizationRepositoryLive, SqlClientLive } from "@platform/db-postgres"
 import { BadRequestError } from "@repo/utils"
 import { Effect } from "effect"
 import type { Context, MiddlewareHandler, Next } from "hono"
@@ -26,8 +27,16 @@ export const createOrganizationContextMiddleware = (): MiddlewareHandler => {
       })
     }
 
-    const organizationRepository = createOrganizationPostgresRepository(c.var.db)
-    const organization = await Effect.runPromise(organizationRepository.findById(OrganizationId(organizationIdParam)))
+    const organizationId = OrganizationId(organizationIdParam)
+    const organization = await Effect.runPromise(
+      Effect.gen(function* () {
+        const repo = yield* OrganizationRepository
+        return yield* repo.findById(organizationId)
+      }).pipe(
+        Effect.provide(OrganizationRepositoryLive),
+        Effect.provide(SqlClientLive(c.var.postgresClient, organizationId)),
+      ),
+    )
 
     c.set("organization", organization)
     await next()

apps/api/src/middleware/touch-buffer.ts

@@ -1,5 +1,7 @@
+import { ApiKeyRepository } from "@domain/api-keys"
 import { ApiKeyId } from "@domain/shared"
-import { type PostgresDb, createApiKeyPostgresRepository } from "@platform/db-postgres"
+import { ApiKeyRepositoryLive, SqlClientLive } from "@platform/db-postgres"
+import type { PostgresClient } from "@platform/db-postgres"
 import { createLogger } from "@repo/observability"
 import { Effect } from "effect"
 
@@ -40,10 +42,10 @@ class TouchBuffer {
   private flushInterval: NodeJS.Timeout | null = null
   private readonly intervalMs: number
   private readonly maxBufferSize: number
-  private readonly db: PostgresDb
+  private readonly client: PostgresClient
 
-  constructor(db: PostgresDb, config: TouchBufferConfig = {}) {
-    this.db = db
+  constructor(client: PostgresClient, config: TouchBufferConfig = {}) {
+    this.client = client
     this.intervalMs = config.intervalMs ?? 30000
     this.maxBufferSize = config.maxBufferSize ?? 10000
 
@@ -97,11 +99,17 @@ class TouchBuffer {
 
     const startTime = Date.now()
 
-    // Use the admin db connection (bypasses RLS) for cross-org batch updates.
-    const repo = createApiKeyPostgresRepository(this.db)
+    // Use the live layer pattern for cross-org batch updates (bypasses RLS)
+    const sqlClientLayer = SqlClientLive(this.client)
+    const apiKeyRepoLayer = ApiKeyRepositoryLive
 
     try {
-      await Effect.runPromise(repo.touchBatch(keyIds))
+      await Effect.runPromise(
+        Effect.gen(function* () {
+          const repo = yield* ApiKeyRepository
+          return yield* repo.touchBatch(keyIds)
+        }).pipe(Effect.provide(apiKeyRepoLayer), Effect.provide(sqlClientLayer)),
+      )
 
       const duration = Date.now() - startTime
       logger.info(`Flushed ${keyIds.length} touch updates in ${duration}ms`)
@@ -169,9 +177,9 @@ class TouchBuffer {
  */
 let touchBufferInstance: TouchBuffer | null = null
 
-export const createTouchBuffer = (db: PostgresDb, config?: TouchBufferConfig): TouchBuffer => {
+export const createTouchBuffer = (client: PostgresClient, config?: TouchBufferConfig): TouchBuffer => {
   if (!touchBufferInstance) {
-    touchBufferInstance = new TouchBuffer(db, config)
+    touchBufferInstance = new TouchBuffer(client, config)
   }
   return touchBufferInstance
 }