roedyrustam/vibes-plug/database-orm-expert

// src/db/schema.ts
import { pgTable, uuid, varchar, bigint, timestamp, index, uniqueIndex } from 'drizzle-orm/pg-core';
import { relations, sql } from 'drizzle-orm';
export const tenants = pgTable('tenants', {
  id: uuid('id').defaultRandom().primaryKey(),
  slug: varchar('slug', { length: 64 }).notNull(),
  createdAt: timestamp('created_at', { withTimezone: true }).defaultNow().notNull(),
}, (t) => [
  uniqueIndex('idx_tenants_slug').on(t.slug),
]);
export const wallets = pgTable('wallets', {
  id: uuid('id').defaultRandom().primaryKey(),
  tenantId: uuid('tenant_id').notNull().references(() => tenants.id, { onDelete: 'restrict' }),
  currency: varchar('currency', { length: 3 }).notNull().default('IDR'),
  balanceMinor: bigint('balance_minor', { mode: 'bigint' }).notNull().default(0n),
  version: bigint('version', { mode: 'number' }).notNull().default(1),
  updatedAt: timestamp('updated_at', { withTimezone: true }).defaultNow().notNull(),
}, (t) => [
  index('idx_wallets_tenant_id').on(t.tenantId),
]);
export const paymentIdempotency = pgTable('payment_idempotency', {
  idempotencyKey: varchar('idempotency_key', { length: 128 }).primaryKey(),
  walletId: uuid('wallet_id').notNull().references(() => wallets.id, { onDelete: 'cascade' }),
  amountMinor: bigint('amount_minor', { mode: 'bigint' }).notNull(),
  status: varchar('status', { length: 24 }).notNull(),
  createdAt: timestamp('created_at', { withTimezone: true }).defaultNow().notNull(),
}, (t) => [
  index('idx_payment_idempotency_wallet').on(t.walletId),
]);
export const walletsRelations = relations(wallets, ({ one }) => ({
  tenant: one(tenants, { fields: [wallets.tenantId], references: [tenants.id] }),
}));

// src/db/mutations/debit-wallet.ts
import { eq, sql } from 'drizzle-orm';
import { db } from '../client';
import { wallets, paymentIdempotency } from '../schema';
export async function debitWalletAtomic(input: {
  walletId: string;
  amountMinor: bigint;
  idempotencyKey: string;
}) {
  if (input.amountMinor <= 0n) {
    throw new Error('Debit amount must be strictly positive');
  }
  return await db.transaction(async (tx) => {
    // 1. Check idempotency key first
    const [existing] = await tx
      .select()
      .from(paymentIdempotency)
      .where(eq(paymentIdempotency.idempotencyKey, input.idempotencyKey))
      .limit(1);
    if (existing) {
      return { status: 'ALREADY_PROCESSED' as const, record: existing };
    }
    // 2. Acquire pessimistic row-level lock (SELECT ... FOR UPDATE)
    const [wallet] = await tx
      .select()
      .from(wallets)
      .where(eq(wallets.id, input.walletId))
      .for('update');
    if (!wallet) {
      throw new Error(`Wallet ${input.walletId} not found`);
    }
    if (wallet.balanceMinor < input.amountMinor) {
      throw new Error('Insufficient wallet balance');
    }
    // 3. Deduct balance and increment optimistic version
    const [updatedWallet] = await tx
      .update(wallets)
      .set({
        balanceMinor: sql`${wallets.balanceMinor} - ${input.amountMinor}`,
        version: sql`${wallets.version} + 1`,
        updatedAt: new Date(),
      })
      .where(eq(wallets.id, input.walletId))
      .returning();
    // 4. Record idempotency lock in same transaction
    const [record] = await tx
      .insert(paymentIdempotency)
      .values({
        idempotencyKey: input.idempotencyKey,
        walletId: input.walletId,
        amountMinor: input.amountMinor,
        status: 'COMPLETED',
      })
      .returning();
    return { status: 'COMPLETED' as const, wallet: updatedWallet, record };
  }, {
    isolationLevel: 'read committed',
  });
}

import { and, desc, eq, lt, or } from 'drizzle-orm';
import { db } from './client';
import { wallets } from './schema';
export async function listWalletsCursor(params: {
  tenantId: string;
  limit?: number;
  cursor?: { updatedAt: Date; id: string };
}) {
  const limit = Math.min(params.limit ?? 25, 100);
  const cursorFilter = params.cursor
    ? or(
        lt(wallets.updatedAt, params.cursor.updatedAt),
        and(eq(wallets.updatedAt, params.cursor.updatedAt), lt(wallets.id, params.cursor.id))
      )
    : undefined;
  const rows = await db
    .select()
    .from(wallets)
    .where(and(eq(wallets.tenantId, params.tenantId), cursorFilter))
    .orderBy(desc(wallets.updatedAt), desc(wallets.id))
    .limit(limit + 1);
  const hasNextPage = rows.length > limit;
  const items = hasNextPage ? rows.slice(0, limit) : rows;
  const lastItem = items[items.length - 1];
  return {
    items,
    nextCursor: hasNextPage && lastItem
      ? { updatedAt: lastItem.updatedAt, id: lastItem.id }
      : null,
  };
}

-- Safe Zero-Lock Constraint & Concurrent Index Migration
ALTER TABLE wallets ADD COLUMN IF NOT EXISTS ledger_account_id UUID;
 
CREATE INDEX CONCURRENTLY IF NOT EXISTS idx_wallets_ledger_account_id
  ON wallets (ledger_account_id);
 
-- Add foreign key without locking existing writes (Postgres 14+)
ALTER TABLE wallets
  ADD CONSTRAINT fk_wallets_ledger_account
  FOREIGN KEY (ledger_account_id) REFERENCES accounts(id) NOT VALID;
 
-- Validate constraint using ShareUpdateExclusiveLock (does not block reads/writes)
ALTER TABLE wallets VALIDATE CONSTRAINT fk_wallets_ledger_account;

import { neon } from '@neondatabase/serverless';
import { drizzle } from 'drizzle-orm/neon-http';
import { Redis } from '@upstash/redis';
import * as schema from './schema';
const sqlClient = neon(process.env.DATABASE_URL!);
export const edgeDb = drizzle(sqlClient, { schema });
const redis = new Redis({
  url: process.env.UPSTASH_REDIS_REST_URL!,
  token: process.env.UPSTASH_REDIS_REST_TOKEN!,
});
export async function getTenantCached(slug: string) {
  const cacheKey = `tenant:v1:${slug}`;
  const cached = await redis.get<typeof schema.tenants.$inferSelect>(cacheKey);
  if (cached) return cached;
  const tenant = await edgeDb.query.tenants.findFirst({
    where: (t, { eq }) => eq(t.slug, slug),
  });
  if (tenant) {
    await redis.set(cacheKey, tenant, { ex: 900 });
  }
  return tenant ?? null;
}