@lota-sdk/pg

Postgres platform for the Lota SDK: typed repositories, graph edges + bounded traversal, vector + lexical retrieval with fusion, claim queues, multi-schema routing, and goose migrations — on Bun, with Kysely as the query engine and Zod as the runtime contract.

• Drivers: node-postgres (postgres://) in production, PGlite (pglite://) embedded for tests/dev. One facade, identical behavior.
• Casing: snake_case in the database, camelCase in TypeScript — CamelCasePlugin bridges both directions (including raw sql results); jsonb payload keys pass through untouched.
• Migrations: SQL-first goose files. Deploys run the goose binary; processes only verifyMigrations() at boot — no runtime DDL.

Setup

import { createLotaPg, sql } from '@lota-sdk/pg'

const db = await createLotaPg({
  url: process.env.DATABASE_URL!,       // or 'pglite://memory'
  schema: 'lota',
  applicationName: 'lota-server',
  pool: { max: 15 },
})

await db.ping()

Kernel migrations (extensions, lota_pg_touch() timestamps trigger, lota_pg_uuidv7(), the lota_english text-search config) ship with the package:

bunx --package @lota-sdk/pg lota-pg install-goose        # pinned goose binary -> .tools/goose
bunx --package @lota-sdk/pg lota-pg migrate up --kernel  # once per database
bunx --package @lota-sdk/pg lota-pg migrate up --dir migrations/lota
bunx --package @lota-sdk/pg lota-pg migrate verify --dir migrations/lota

Tables & repositories

import { z } from 'zod'
import { defineTableSpec } from '@lota-sdk/pg/schema'
import { createRepo, scopedRepo } from '@lota-sdk/pg/repo'

const memoryRow = z.object({
  id: z.uuid(),
  organizationId: z.uuid(),
  content: z.string(),
  metadata: z.record(z.string(), z.unknown()),
  embedding: z.array(z.number()).nullish(),   // nullable column ⇒ .nullish()
  archivedAt: z.coerce.date().nullish(),
  createdAt: z.coerce.date(),
  updatedAt: z.coerce.date(),
})

export const memoryTable = defineTableSpec({
  name: 'memory',
  row: memoryRow,
  tenant: { scope: 'org' },
  vector: { field: 'embedding', dims: 256 },
  lexical: { sources: ['content'] },
  indexes: [{ on: ['organizationId', 'createdAt desc'] }],
})

const memories = scopedRepo(createRepo(db, memoryTable), { organizationId })
await memories.create({ content: 'hello', metadata: {} })
await memories.update(id, { archivedAt: null })          // null clears, undefined preserves
await memories.findMany({ 'metadata.topic': 'pricing' }) // jsonb dotted-path filter
const page = await memories.page({ limit: 50 })          // keyset (created_at, id)

created_at/updated_at are DB-owned (trigger): immutable + per-statement. Generate the DDL draft from specs, review, commit:

bunx --package @lota-sdk/pg lota-pg gen-migration --specs src/db/specs.ts --schema lota --out migrations/lota/00001_init.sql

Graph

import { defineEdgeSpec, createGraph } from '@lota-sdk/pg/graph'

const memoryRelation = defineEdgeSpec({
  name: 'memoryRelation',
  row: memoryRelationRow,
  from: { references: memoryTable },
  to: { references: memoryTable },
  kind: { values: ['supports', 'contradicts', 'supersedes'] },
  // dedupe: 'unique-triple' (default) — distinct supporters stay distinct,
  // same-pair re-assertions collapse via ON CONFLICT.
})

const graph = createGraph(db, memoryRelation, { scope: { organizationId: orgId } })
await graph.relate(a, b, { kind: 'supports', confidence: 0.8 })
const hood = await graph.neighbors([a], { direction: 'both', kinds: ['supports'] })
const summary = await graph.counts(ids, { collect: { nodeField: 'content' } }) // ordered far-node texts
const reach = await graph.walk(a, { maxDepth: 3 })                             // recursive CTE, cycle-safe

Retrieval

import { vectorLeg, lexicalLeg, hybrid, rrf } from '@lota-sdk/pg/retrieval'

const results = await hybrid(db, {
  legs: [
    vectorLeg(memoryTable, embedding, { k: 50, efSearch: 100, scope: { organizationId: orgId }, where: sql`archived_at is null` }),
    lexicalLeg(memoryTable, query, { k: 50, scope: { organizationId: orgId }, where: sql`archived_at is null` }),
  ],
  fuse: rrf({ k: 60, weights: [0.65, 0.35] }),
  limit: 20,
})

NULL embeddings are always excluded; GUCs are SET LOCAL-scoped; fused scores are strictly re-sorted (empty legs contribute 0.0).

Claims

import { claimOne, reclaimExpired, advanceWatermark } from '@lota-sdk/pg/claims'

const job = await claimOne(db, jobTable, {
  where: sql`status = 'queued' and due_at <= now()`,
  scope: { organizationId: orgId },
  orderBy: sql`due_at asc`,
  set: { status: 'running', leaseOwner: workerId, claimedAt: sql`now()` },
})  // FOR UPDATE SKIP LOCKED — no double claims, no blocking

Multi-schema & provisioning

import { provisionSchema } from '@lota-sdk/pg/migrate'

const repoIntel = db.forSchema('repo_intel_org_x')             // same pool
await provisionSchema(db, { schema: 'repo_intel_org_x', dir: 'migrations/repo-intel' })

Migration files reference their schema as {{schema}}; provisioning is advisory-locked and idempotent.

Testing

import { createTestDb, applyTableSpecs } from '@lota-sdk/pg/testing'

const { db, truncateAll, close } = await createTestDb()        // PGlite + kernel, in-process
await applyTableSpecs(db, [memoryTable])                        // real gen-migration output

Integration tests run against real Postgres 18: bun run db:test:up && TEST_DATABASE_URL=postgres://lota:lota@localhost:54329/lota_test bun run test:integration

Effect

import { PgDatabaseTag, layer, tryDb } from '@lota-sdk/pg/effect'   // optional peer: effect

Development

bun run check = format + lint + typecheck + knip + unit tests (same flow as the other Lota SDK packages).