refactor(db): move Postgres access from drizzle to effect-orm

CLAUDE.md

@@ -9,7 +9,7 @@ This is a monorepo with the following structure:
 - `apps/web/` - React frontend using Vite, TanStack Router, and TailwindCSS
 - `apps/backendv2/` - Backend API using Bun runtime and Effect-TS
 - `apps/cluster/` - Effect Cluster service for distributed workflows and background jobs
-- `packages/db/` - Shared database package using Drizzle ORM and PostgreSQL
+- `packages/db/` - Shared database package: effect-orm table definitions, the `Database` service, and migrations
 - `packages/domain/` - Shared domain types, RPC contracts, and cluster definitions
 - `.context/` - Git Subtrees for context of how to use specific libraries (In this case Effect and Effect Atom)
 
@@ -92,7 +92,9 @@ When working with Effect, Effect Atom, or TanStack DB, refer to these directorie
 
 ### Database (packages/db)
 
-- `bun run db` - Run Drizzle Kit commands for schema management
+- `bun run db:migrate` - Apply migrations to `DATABASE_URL` (adopts a database `drizzle-kit push` built)
+- `bun run db:generate --name <name>` - Diff `src/tables` against the migrations and write the next one
+- `bun run db:check` / `db:status` / `db:verify` - Validate the folder, list applied migrations, compare with a live schema
 
 ## Tech Stack
 
@@ -112,7 +114,7 @@ When working with Effect, Effect Atom, or TanStack DB, refer to these directorie
 
 - **Runtime**: Bun
 - **Framework**: Effect-TS for functional programming
-- **Database**: PostgreSQL with Drizzle ORM
+- **Database**: PostgreSQL with effect-orm (`@maple-dev/effect-orm`) over `@effect/sql-pg`
 - **Auth**: Clerk (`@clerk/backend` JWT verification + webhook sync)
 - **API**: RPC-style endpoints via Effect Http Api
 
@@ -182,7 +184,20 @@ Available branded types in `@hazel/schema`: `OrganizationId`, `ChannelId`, `User
 
 ## Database
 
-Uses Drizzle ORM with PostgreSQL. Database schema is defined in `packages/db/src/schema/`. Use `bun run db` commands for migrations and schema management.
+Uses effect-orm (`@maple-dev/effect-orm`, source in `~/Documents/GitHub/effect-clickhouse`) over `@effect/sql-pg`. Tables are `PG.table` definitions in `packages/db/src/tables/`; migrations live in `packages/db/migrations/`.
+
+```typescript
+import { Database, PG, schema } from "@hazel/db"
+
+const db = yield* Database.Database
+const channel = yield* db.first(
+	PG.from(schema.channelsTable).select().where(($) => [$.id.eq(channelId), $.deletedAt.isNull()]),
+)
+yield* db.transaction(Effect.gen(function* () { /* every statement in here joins the transaction */ }))
+```
+
+- `db.transaction(effect)` pins every statement inside `effect` (repositories included) to one connection through fiber context; there is no transaction handle to pass around.
+- Schema changes: edit `src/tables`, then `bun run db:generate --name <name>`. effect-orm does not model enums or unique constraints: write those in a `--custom` migration.
 
 ## Architecture Notes
 

README.md

@@ -40,7 +40,7 @@ Hazel Chat is a full-stack chat application featuring:
 - **Bun** runtime
 - **Effect-TS** functional programming framework
 - **Effect RPC** for type-safe APIs
-- **Drizzle ORM** with PostgreSQL
+- **effect-orm** (`@maple-dev/effect-orm`) with PostgreSQL
 - **Clerk** for authentication
 
 ### Cluster Service
@@ -65,7 +65,7 @@ Hazel Chat is a full-stack chat application featuring:
 │   ├── electric-proxy/      # Electric SQL proxy worker
 │   └── link-preview-worker/ # Link preview generation
 ├── packages/
-│   ├── db/                  # Drizzle ORM schemas and migration artifacts (`drizzle/`)
+│   ├── db/                  # effect-orm table definitions and migrations (`migrations/`)
 │   ├── domain/              # Shared types, RPC contracts, cluster definitions
 │   ├── backend-core/        # Reusable backend services and repositories
 │   ├── schema/              # Shared schema definitions

apps/backend/package.json

@@ -41,7 +41,6 @@
 		"alchemy": "catalog:alchemy",
 		"aws4fetch": "^1.0.20",
 		"dfx": "^1.1.0",
-		"drizzle-orm": "^0.45.1",
 		"effect": "catalog:effect",
 		"jose": "^6.1.3",
 		"pg": "^8.16.3",
@@ -51,7 +50,6 @@
 		"@cloudflare/workers-types": "catalog:alchemy",
 		"@testcontainers/postgresql": "^10.18.0",
 		"@types/bun": "1.3.9",
-		"drizzle-kit": "^0.31.8",
 		"typescript": "^5.9.3"
 	}
 }

apps/backend/scripts/create-bot.ts

@@ -13,7 +13,7 @@
  * 4. Print the bot token (save it - you can't retrieve it later!)
  */
 
-import { Database, schema } from "@hazel/db"
+import { Database, PG, schema } from "@hazel/db"
 import type { BotId, BotInstallationId, OrganizationId, OrganizationMemberId, UserId } from "@hazel/schema"
 import { Effect, References } from "effect"
 import { randomUUID } from "crypto"
@@ -67,8 +67,8 @@ const createBotScript = Effect.gen(function* () {
 
 	// 1. Create a user for the bot (userType = 'machine')
 	console.log("  → Creating bot user...")
-	yield* db.execute((client) =>
-		client.insert(schema.usersTable).values({
+	yield* db.run(
+		PG.insertInto(schema.usersTable).values({
 			id: botUserId,
 			externalId,
 			email: botEmail,
@@ -82,8 +82,8 @@ const createBotScript = Effect.gen(function* () {
 
 	// 2. Create the bot
 	console.log("  → Creating bot record...")
-	yield* db.execute((client) =>
-		client.insert(schema.botsTable).values({
+	yield* db.run(
+		PG.insertInto(schema.botsTable).values({
 			id: botId,
 			userId: botUserId,
 			createdBy: botUserId,
@@ -97,8 +97,8 @@ const createBotScript = Effect.gen(function* () {
 
 	// 3. Install the bot in the organization
 	console.log("  → Installing bot in organization...")
-	yield* db.execute((client) =>
-		client.insert(schema.botInstallationsTable).values({
+	yield* db.run(
+		PG.insertInto(schema.botInstallationsTable).values({
 			id: installationId,
 			botId,
 			organizationId: orgId,
@@ -109,8 +109,8 @@ const createBotScript = Effect.gen(function* () {
 
 	// 4. Add bot user to the organization as a member
 	console.log("  → Adding bot user to organization...")
-	yield* db.execute((client) =>
-		client.insert(schema.organizationMembersTable).values({
+	yield* db.run(
+		PG.insertInto(schema.organizationMembersTable).values({
 			id: membershipId,
 			organizationId: orgId,
 			userId: botUserId,

apps/backend/scripts/rebuild-channel-access.ts

@@ -1,6 +1,6 @@
 #!/usr/bin/env bun
 
-import { and, Database, eq, isNull, ne, schema } from "@hazel/db"
+import { Database, PG, schema } from "@hazel/db"
 import { Effect, Layer, References } from "effect"
 import { ChannelAccessSyncService } from "../src/services/channel-access-sync"
 import { DatabaseLive } from "../src/services/database"
@@ -10,13 +10,12 @@ const rebuildChannelAccess = Effect.gen(function* () {
 
 	yield* Effect.logInfo("Rebuilding channel_access table")
 
-	yield* db.execute((client) => client.delete(schema.channelAccessTable))
+	yield* db.run(PG.deleteFrom(schema.channelAccessTable).allRows())
 
-	const activeNonThreadChannels = yield* db.execute((client) =>
-		client
-			.select({ id: schema.channelsTable.id })
-			.from(schema.channelsTable)
-			.where(and(isNull(schema.channelsTable.deletedAt), ne(schema.channelsTable.type, "thread"))),
+	const activeNonThreadChannels = yield* db.run(
+		PG.from(schema.channelsTable)
+			.select("id")
+			.where(($) => [$.deletedAt.isNull(), $.type.neq("thread")]),
 	)
 
 	yield* Effect.forEach(
@@ -25,11 +24,10 @@ const rebuildChannelAccess = Effect.gen(function* () {
 		{ concurrency: 20 },
 	)
 
-	const threadChannels = yield* db.execute((client) =>
-		client
-			.select({ id: schema.channelsTable.id })
-			.from(schema.channelsTable)
-			.where(and(isNull(schema.channelsTable.deletedAt), eq(schema.channelsTable.type, "thread"))),
+	const threadChannels = yield* db.run(
+		PG.from(schema.channelsTable)
+			.select("id")
+			.where(($) => [$.deletedAt.isNull(), $.type.eq("thread")]),
 	)
 
 	yield* Effect.forEach(
@@ -40,10 +38,10 @@ const rebuildChannelAccess = Effect.gen(function* () {
 		},
 	)
 
-	const countResult = yield* db.execute(
-		(client) => client.$client`SELECT COUNT(*)::int AS count FROM channel_access`,
+	const countResult = yield* db.run(
+		PG.from(schema.channelAccessTable).select(() => ({ count: PG.count() })),
 	)
-	const count = (countResult[0] as { count: number } | undefined)?.count ?? 0
+	const count = countResult[0]?.count ?? 0
 
 	console.log(
 		`channel_access rebuilt: ${activeNonThreadChannels.length} non-thread channels, ${threadChannels.length} threads, ${count} rows`,

apps/backend/scripts/reset-all.ts

@@ -1,7 +1,7 @@
 #!/usr/bin/env bun
 
 import { Database } from "@hazel/db"
-import { Effect, References } from "effect"
+import { Effect, References, Schema } from "effect"
 import { DatabaseLive } from "../src/services/database"
 
 // Parse command line arguments
@@ -54,9 +54,12 @@ const clearDatabase = Effect.gen(function* () {
 
 	for (const table of allTables) {
 		const countResult = yield* db
-			.execute((client) => client.$client`SELECT COUNT(*)::int as count FROM ${client.$client(table)}`)
+			.query(
+				Database.sql`SELECT COUNT(*)::int as count FROM ${Database.sql.identifier(table)}`,
+				Schema.Struct({ count: Schema.Number }),
+			)
 			.pipe(Effect.orDie)
-		const count = (countResult[0] as { count: number } | undefined)?.count ?? 0
+		const count = countResult[0]?.count ?? 0
 		tableCounts[table] = count
 
 		if (count === 0) {
@@ -68,7 +71,7 @@ const clearDatabase = Effect.gen(function* () {
 			log("yellow", `  [DRY RUN] Would delete ${count} rows from ${table}`)
 		} else {
 			yield* db
-				.execute((client) => client.$client`TRUNCATE TABLE ${client.$client(table)} CASCADE`)
+				.execute(Database.sql`TRUNCATE TABLE ${Database.sql.identifier(table)} CASCADE`)
 				.pipe(Effect.orDie)
 			log("green", `  ✓ Cleared ${count} rows from ${table}`)
 		}

apps/backend/scripts/seed-internal-bots.ts

@@ -11,10 +11,9 @@
  * If a bot already exists, it will be skipped and no new token will be generated.
  */
 
-import { Database, schema } from "@hazel/db"
+import { Database, PG, schema } from "@hazel/db"
 import type { BotId, UserId } from "@hazel/schema"
 import { createHash, randomUUID } from "crypto"
-import { eq } from "drizzle-orm"
 import { Effect, References } from "effect"
 import { DatabaseLive } from "../src/services/database"
 
@@ -96,25 +95,27 @@ const seedBot = Effect.fn("seedInternalBots.seedBot")(function* (provider: Inter
 	console.log("  -> Checking if bot exists...")
 
 	// Check if bot user already exists
-	const existingUser = yield* db.execute((client) =>
-		client.select().from(schema.usersTable).where(eq(schema.usersTable.externalId, externalId)).limit(1),
+	const existingUser = yield* db.run(
+		PG.from(schema.usersTable)
+			.select()
+			.where(($) => [$.externalId.eq(externalId)])
+			.limit(1),
 	)
 
 	if (existingUser.length > 0) {
 		console.log("  -> Bot already exists, skipping...")
-		console.log(`     Bot User ID: ${existingUser[0].id}`)
+		console.log(`     Bot User ID: ${existingUser[0]!.id}`)
 
 		// Get the bot record to show bot ID
-		const existingBot = yield* db.execute((client) =>
-			client
+		const existingBot = yield* db.run(
+			PG.from(schema.botsTable)
 				.select()
-				.from(schema.botsTable)
-				.where(eq(schema.botsTable.userId, existingUser[0].id as UserId))
+				.where(($) => [$.userId.eq(existingUser[0]!.id)])
 				.limit(1),
 		)
 
 		if (existingBot.length > 0) {
-			console.log(`     Bot ID: ${existingBot[0].id}`)
+			console.log(`     Bot ID: ${existingBot[0]!.id}`)
 		}
 
 		return { created: false, provider }
@@ -128,8 +129,8 @@ const seedBot = Effect.fn("seedInternalBots.seedBot")(function* (provider: Inter
 
 	// Create bot user
 	console.log("  -> Creating bot user...")
-	yield* db.execute((client) =>
-		client.insert(schema.usersTable).values({
+	yield* db.run(
+		PG.insertInto(schema.usersTable).values({
 			id: botUserId,
 			externalId,
 			email: botEmail,
@@ -143,8 +144,8 @@ const seedBot = Effect.fn("seedInternalBots.seedBot")(function* (provider: Inter
 
 	// Create bot record
 	console.log("  -> Creating bot record...")
-	yield* db.execute((client) =>
-		client.insert(schema.botsTable).values({
+	yield* db.run(
+		PG.insertInto(schema.botsTable).values({
 			id: botId,
 			userId: botUserId,
 			createdBy: botUserId,

apps/backend/src/lib/create-transactionId.ts

@@ -1,45 +1,23 @@
 import { Database } from "@hazel/db"
 import { TransactionIdFromString } from "@hazel/schema"
-import { Effect, Option, Predicate, Schema } from "effect"
+import { Effect, Schema } from "effect"
 
-export const generateTransactionId = Effect.fn("generateTransactionId")(function* (
-	tx?: <T>(
-		fn: (client: Database.TransactionClient) => Promise<T>,
-	) => Effect.Effect<T, Database.DatabaseError, never>,
-) {
-	// 1. Check explicit tx parameter (highest priority)
-	let txExecutor: Database.TxFn | undefined = tx
+const TxidRow = Schema.Struct({ txid: Schema.String })
 
-	// 2. Check TransactionContext from Effect Context (auto-propagated)
-	if (!txExecutor) {
-		const maybeCtx = yield* Effect.serviceOption(Database.TransactionContext)
-		if (Option.isSome(maybeCtx)) {
-			txExecutor = maybeCtx.value.execute
-		}
-	}
-
-	// 3. Fail if no transaction context found
-	if (!txExecutor) {
-		return yield* Effect.die("generateTransactionId must be called within a transaction")
-	}
-
-	const result = yield* txExecutor((client) =>
-		client.execute(`SELECT pg_current_xact_id()::xid::text as txid`),
-	).pipe(
-		Effect.map((rows) => rows[0]?.txid as string),
-		Effect.tap((rawTxid) =>
-			Effect.log(`[txid-debug] Raw PostgreSQL txid string: "${rawTxid}", type: ${typeof rawTxid}`),
-		),
-		Effect.flatMap((txid) => Schema.decodeEffect(TransactionIdFromString)(txid)),
-		Effect.tap((decodedTxid) =>
-			Effect.log(`[txid-debug] Decoded transactionId: ${decodedTxid}, type: ${typeof decodedTxid}`),
-		),
-		Effect.catchIf(
-			(e): e is Database.DatabaseError => Predicate.isTagged(e, "DatabaseError"),
-			(err) => Effect.die(`Database error generating transaction ID: ${err}`),
-		),
+/**
+ * The id of the open transaction, for Electric to confirm the client's optimistic write against.
+ * Must run inside `db.transaction`, which pins this statement to the transaction's connection.
+ */
+export const generateTransactionId = Effect.fn("generateTransactionId")(function* () {
+	const db = yield* Database.Database
+	const rows = yield* db
+		.query(Database.sql`SELECT pg_current_xact_id()::xid::text AS txid`, TxidRow)
+		.pipe(
+			Effect.catchTag("DatabaseError", (err) =>
+				Effect.die(`Database error generating transaction ID: ${err}`),
+			),
+		)
+	return yield* Schema.decodeEffect(TransactionIdFromString)(rows[0]?.txid ?? "").pipe(
 		Effect.catchTag("SchemaError", (err) => Effect.die(`Failed to parse transaction ID: ${err}`)),
 	)
-
-	return result
-})
+}, Database.requireTransaction)