ElementsStore

SQL

Relational schemas, field classification, store.resource CRUD, RLS policies, and the fx.store session handle.

The SQL facet is how you declare relational tables and read or write them from Flows. Schemas use store.schema.table + field.*; I/O goes through fx.store(db).

For developers modeling domain data on okengine — mark columns, push schema, query with the session handle (or mount store.resource).

The one rule

Declare tables with store.schema.table. Touch them only via fx.store(db) — select / insert / update / delete / page / search. There is no fx.store(db).query.… Relational Query Builder on the handle.

Smallest Example

Define a table and bind the store

src/db/schema.decl.ts
import { store, field } from "okengine";

export const posts = store.schema.table("posts", {
  id: field.id().primaryKey(),
  title: field.text().notNull(),
  authorId: field.text().notNull(),
  publishedAt: field.timestamp(),
  createdAt: field.timestamp().notNull().now(),
});

export const db = store.sql("app", { schema: { posts } });

Query in a Flow

src/flows/posts/list.ts
import { on, flow, http } from "okengine";
import { db, posts } from "@/schema";

export const list = on(
  http.get(),
  flow({
    do: async (_, fx) => {
      return await fx.store(db).page(posts, {
        orderBy: [posts.createdAt],
        limit: 20,
      });
    },
  }),
);

Push schema and call

oke db push
curl -X GET http://localhost:6530/posts -H "accept: application/json"

Response:

{
  "data": [{ "id": "…", "title": "…", "authorId": "…", "publishedAt": null, "createdAt": "…" }],
  "error": null
}

Pathless declare

Starters often pass the whole decl module: store.sql("app", {schema}) where schema is `import

  • as schema from "@/db/schema.decl"`. Named keys and the exported table handles are equivalent.

Progressive Patterns

Explore SQL from a bare table to classified columns, CRUD factory, and RLS:

field.id() is text + auto id on insert. Use .okid() on an existing text column when you want the same default without the sugar factory:

src/db/schema.decl.ts
import { store, field } from "okengine";

export const notes = store.schema.table("notes", {
  id: field.id().primaryKey(),
  title: field.text().notNull(),
  body: field.text().notNull(),
  archivedAt: field.timestamp(),
  createdAt: field.timestamp().notNull().now(),
});

Field Reference

Factories mirror Drizzle pg-core names. Chain modifiers after the factory:

FactoryInfersNotes
field.id()stringtext + auto id on insert (≡ field.text().okid())
field.okid()stringPins OK ID on a text column
field.text() / varchar / charstringOptional { length?, enum? }
field.boolean()boolean
field.smallint() / integer()number
field.bigint({ mode? })number (default)mode: "number" · "bigint" · "string"
field.serial() / smallserial() / bigserial()numberNOT NULL by SQL physics
field.numeric() / decimal()string (default)Exact decimal; { precision?, scale?, mode? }
field.real() / doublePrecision()numberFloat4 / float8
field.json() / jsonb()genericNarrow with field.json<MyShape>()
field.uuid()string
field.time() / timestamp() / date() / interval()see typetimestamp / date default to Date; { mode: "string" } for ISO. On write, finite epoch-ms numbers (e.g. fx.clock.now()) coerce to Date for Postgres
field.point() / line()tuple{ mode: "xy" } / "abc" for objects
field.bytea()Buffer
field.inet() / cidr() / macaddr() / macaddr8()string
ChainMeaning
.primaryKey() · .notNull() · .unique()Constraints
.default(v) · .defaultFn(fn) · .now() · .okid()Defaults
.pii() · .sensitive() · .retain(duration)Classification tags
.searchable({ weight? }) · .embed({ model?, dims? })Hybrid search — see Search
.as(sqlName) · .describe(text) · .type<T>()SQL name, docs, TS override
.references(() => col, actions?)FK (onDelete / onUpdate: cascade · restrict · no action · set null · set default)

Declaring Stores

Bind tables once with store.sql(name, options). The name becomes the resource suffix (sql:app).

OptionTypeDefaultMeaning
schematable map or module(omit)Tables for push / migrate / fx.store
classifytable → column → tags{}Explicit tags; wins over schema-derived on conflict
descriptionstringstore nameConsole / docs label

Named keys — pass the tables you care about:

export const db = store.sql("app", { schema: { posts, notes } });

Module star — starters re-export everything from the decl file:

src/core.ts
import { store } from "okengine";
import * as schema from "@/db/schema.decl";

export const db = store.sql("app", { schema });

Consequence: every fx.store(db) touch stamps reads / writes as sql:app on the Flow — Manifest, Console, and least privilege follow from that.

Session Handle

fx.store(db) returns a SqlStoreHandle. Every method goes through the same masking / RLS / CDC boundary.

MethodPurpose
select() / select(cols).from(t).where?.().orderBy?.().limit?.()Fluent select
insert(t).values(row)Insert (returning() optional)
update(t).set(row).where(cond)Update — where required
findById(t, id)PK lookup
delete(t, id) / delete(t).where(cond)Delete by PK or predicate
exists(t, idOrWhere)Presence check
upsert(t, matchOn, values, { onExisting? })Insert-or-skip; "update" opts into overwrite
increment(t, id, column, by?)Atomic numeric bump (by default 1)
count(t, where?)COUNT(*)
page(t, options)Offset or keyset page
search(t, options)BM25 ± LSH — Search
raw(sql, params?)Parameterized SQL (? placeholders)

Each verb binds with await fx.store(db).<method>(…) inside do:

Fluent select, or findById for a PK:

src/flows/notes/[id]/get.ts
import { on, flow, http } from "okengine";
import { z } from "zod";
import { eq } from "drizzle-orm";
import { db, notes } from "@/schema";

export const get = on(
  http.get({
    in: z.object({ id: z.string() }),
    out: z.object({ id: z.string(), title: z.string() }),
    errors: { NotFound: z.object({ id: z.string() }) },
  }),
  flow({
    do: async ({ id }, fx) => {
      const [note] = await fx.store(db).select().from(notes).where(eq(notes.id, id));
      if (!note) return fx.fail("NotFound", { id });
      return note;
    },
  }),
);

findById(notes, id) is the same PK path without a fluent chain.

Also on the handle: exists · increment · count · raw · search. See Search for hybrid ranking.

Resources

Detailed section

Mount details, live SSE, and verb tables live on HTTP · Resources. This section is the factory contract — store.resource(db, table, options).

store.resource(db, table, options) builds five Flows. Pass notesResource.all() (or any bag with list · create · get · update · remove) to on(http.resource(path, ops)).

The factory registers no routes:

src/flows/notes/resource.ts
import { store } from "okengine";
import { z } from "zod";
import { db, notes } from "@/schema";

export const notesResource = store.resource(db, notes, {
  in: z.object({ title: z.string().min(1), body: z.string() }),
  out: z.object({
    id: z.string(),
    title: z.string(),
    body: z.string(),
  }),
  list: {
    search: [notes.title],
    filter: [notes.archivedAt],
    cursor: [notes.createdAt, notes.id],
  },
});

The URL id segment is always :id. Update is PATCH, not PUT.

OpMethodPathTypical status
listGET/notes200 + { data, error, meta }
createPOST/notes201 Created (fx.json.create)
getGET/notes/:id200, or NotFound
updatePATCH/notes/:id200, or NotFound
removeDELETE/notes/:id204 No Content (fx.json.empty)
liveGET/notes/liveSSE — only when live is on

Schema Extras & RLS

Detailed section

If you only need a gate + owner policy, jump to Progressive Patterns → RLS. Extras are the third argument array on store.schema.table(name, cols, extras).

Pass extras after the column map:

ExtraMeaning
store.schema.rls()Enable RLS on the table (no policies yet)
store.schema.policy(name, opts?)Named policy (for, as, using, withCheck)
store.schema.policy.gate(name)Default for: "select" — oke.gate() = '…'
store.schema.policy.owner(col)Default for: "all" — col = oke.user()
store.schema.policy.scope(scope)Default for: "insert" — oke.has_scope('…')
store.schema.policy.tenant(col)Tenant predicate when tenancy is on
store.schema.unscoped()Opt out of tenant requirement
store.schema.live(false)Opt this table out of project live default
store.schema.relations(…)Relation metadata for Drizzle emit — not a query API

Seeding

Seed blocks by env

oke db seed — standard CLI path
dev
test
prod
  • essential

    runs

    Every env, always.

    welcome note
  • dev

    runs

    ConfigEnv dev (Compose laptop).

    sample rows
  • prod

    skipped

    prod only — real deploy targets.

    webhook register

upsert — existence, not correction

matchOn → upserted
notes id="welcome"
upsertedalready-existedchanged
first run — insert

Seed never runs at boot without confirmation — only via oke db seed, or the first-boot confirm when a seed module exists and .oke/state.json has not recorded that identity yet. Categories:

BlockRuns when
essentialEvery env
devdev ConfigEnv only
prodprod only
(none extra)test — essential only
src/db/seed/index.ts
import { defineSeed, type Fx } from "okengine";
import { db } from "@/core";
import { notes } from "@/db/schema.decl";

async function welcomeNote(fx: Fx) {
  await fx.store(db).upsert(
    notes,
    { id: "welcome" },
    {
      id: "welcome",
      title: "Welcome",
      body: "Your Notes API is ready.",
      archivedAt: null,
      createdAt: new Date("2026-01-15T10:00:00.000Z"),
    },
  );
}

export default defineSeed({
  name: "notes",
  description: "Welcome + sample notes",
  essential: welcomeNote,
  // dev: sampleNotes,
  // prod: async (fx) => { /* prod-only rows */ },
});

upsert returns { status: "upserted" | "changed" | "already-existed" }. Default is insert-once; pass { onExisting: "update" } to overwrite.

CDC

Detailed section

Full consumer patterns live on Consumers · CDC. Here is the SQL-side trigger: db.table(handle).changed(column?).

db.table(orders).changed(column?) builds a CDC trigger for on(…). Input is { before, after } plus table / action / id. changed("status") stamps a column name on the Manifest — it is not an op filter.

src/flows/orders/on-status.ts
import { on, flow } from "okengine";
import { db, orders } from "@/schema";

export const onStatus = on(
  db.table(orders).changed("status"),
  flow("orders.statusChanged", {
    do: async ({ before, after }, fx) => {
      if (before?.status === after?.status) return;
      /* committed write */
    },
  }),
);

Consequence: writes must go through fx.store. A raw SQL client bypasses the CDC sink, so no consumer runs.

Drivers

DriverRuns asBest for
postgresDocker / managed PostgresDev + prod default; RLS + live
pgliteIn-process WASMTest default; RLS-capable
memoryProcess mapTiny ephemeral tests

Standard starters inherit DRIVER_DEFAULTS (postgres / pglite / postgres). Pin only when you diverge; vendor choice lives under images, not driver ids:

oke.config.ts
import { defineConfig } from "okengine/config";

export default defineConfig({
  drivers: {
    // omit store.sql to use defaults — pin only overrides
  },
  images: {
    store: {
      sql: "postgres:18-alpine",
    },
  },
});

Troubleshooting

Learn more

Next

KV

Namespaced get/set with duration TTL.

Search

BM25 ± LSH on SQL columns.

HTTP · Resources

Mount store.resource as five CRUD verbs + live.

Overview

Data at rest — SQL, KV, files, and search indexes through one typed fx.store handle.

KV

Namespaced key-value storage with duration TTL — sessions, caches, and short-lived locks via fx.store.

On this page

Smallest ExampleDefine a table and bind the storeQuery in a FlowPush schema and callProgressive PatternsField ReferenceDeclaring StoresSession HandleResourcesSchema Extras & RLSSeedingCDCDriversTroubleshootingLearn moreNext