diff --git a/.changeset/few-oranges-project.md b/.changeset/few-oranges-project.md new file mode 100644 index 000000000..4056f72ff --- /dev/null +++ b/.changeset/few-oranges-project.md @@ -0,0 +1,5 @@ +--- +"@effect-app/infra": minor +--- + +Add optional SQL root-level projected columns for repository-backed SQLite and PostgreSQL stores. diff --git a/packages/effect-app/src/Model/Repository/internal/internal.ts b/packages/effect-app/src/Model/Repository/internal/internal.ts index c2136b979..3ced56f33 100644 --- a/packages/effect-app/src/Model/Repository/internal/internal.ts +++ b/packages/effect-app/src/Model/Repository/internal/internal.ts @@ -19,6 +19,7 @@ import * as Context from "../../../Context.js" import * as Effect from "../../../Effect.js" import { flatMapOption } from "../../../Effect.js" import * as Option from "../../../Option.js" +import { makeRootLevelFieldColumns } from "../../../rootLevelFields.js" import * as S from "../../../Schema.js" import { type Codec, NonNegativeInt } from "../../../Schema.js" import { setupRequestContextFromCurrent } from "../../../setupRequest.ts" @@ -640,6 +641,8 @@ export function makeStore() { mapTo: (e: E, etag: string | undefined) => Encoded, idKey: IdKey ) => { + const rootLevelFieldColumns = makeRootLevelFieldColumns(schema, idKey) + function makeStore( makeInitial?: Effect.Effect, config?: Omit, "partitionValue"> & { @@ -679,6 +682,7 @@ export function makeStore() { : undefined, { ...config, + rootLevelFieldColumns, partitionValue: config?.partitionValue ?? ((_) => "primary") /*(isIntegrationEvent(r) ? r.companyId : r.id*/ } diff --git a/packages/effect-app/src/Model/Repository/makeRepo.ts b/packages/effect-app/src/Model/Repository/makeRepo.ts index 58a115ee0..cb8e5b810 100644 --- a/packages/effect-app/src/Model/Repository/makeRepo.ts +++ b/packages/effect-app/src/Model/Repository/makeRepo.ts @@ -41,6 +41,14 @@ export interface RepositoryOptions< * use the config.defaultValues instead for simple default values */ jitM?: (pm: Encoded) => Encoded + /** + * Repository storage options. + * + * `rootLevelFieldsWhenAvailable` switches SQL repositories from reading/writing + * derived root fields via `data` to reading/writing those fields via dedicated + * root columns instead. Existing table/data migrations are expected to be handled + * offline rather than as runtime fallback behavior. + */ config?: Omit, "partitionValue"> & { partitionValue?: (e?: Encoded) => string } diff --git a/packages/effect-app/src/Store.ts b/packages/effect-app/src/Store.ts index 3267aa27a..92aba0d20 100644 --- a/packages/effect-app/src/Store.ts +++ b/packages/effect-app/src/Store.ts @@ -25,8 +25,27 @@ export interface UniqueKey { readonly paths: string[] } +export type RootLevelFieldColumnKind = "string" | "number" | "boolean" | "json" + +export interface RootLevelFieldColumn { + readonly key: string + readonly columnName: string + readonly kind: RootLevelFieldColumnKind +} + export interface StoreConfig { partitionValue: (e?: E) => string + /** + * For SQL adapters, switch storage/querying for derived root-level fields from the + * document `data` column to dedicated root columns. + * + * When disabled, reads/writes/queries use `data`. + * When enabled, derived root fields are read/written/queried via dedicated columns + * instead. Schema/data migrations for existing tables are expected to happen + * offline rather than on the fly. + */ + rootLevelFieldsWhenAvailable?: boolean + rootLevelFieldColumns?: readonly RootLevelFieldColumn[] /** * Primarily used for testing, creating namespaces in the database to separate data e.g to run multiple tests in isolation within the same database. * Memory/Disk use separate store instances per namespace. CosmosDB uses namespace-prefixed partition keys. SQL uses a `_namespace` column. @@ -274,4 +293,9 @@ export interface StorageConfig { url: Redacted.Redacted prefix: string dbName: string + /** + * Adapter default for switching derived root-level fields from the `data` column + * to dedicated SQL columns. Repositories can still opt in / out per table. + */ + rootLevelFieldsWhenAvailable?: boolean } diff --git a/packages/effect-app/src/rootLevelFields.ts b/packages/effect-app/src/rootLevelFields.ts new file mode 100644 index 000000000..88ef3adec --- /dev/null +++ b/packages/effect-app/src/rootLevelFields.ts @@ -0,0 +1,63 @@ +import * as SchemaAST from "effect/SchemaAST" +import type * as S from "./Schema.js" +import type { RootLevelFieldColumn, RootLevelFieldColumnKind } from "./Store.js" + +const walkTransformation = (ast: SchemaAST.AST): SchemaAST.AST => { + if (ast._tag === "Declaration" && ast.typeParameters.length > 0) { + return walkTransformation(ast.typeParameters[0]!) + } + return ast +} + +const getColumnKind = (ast: SchemaAST.AST): RootLevelFieldColumnKind => { + const encoded = walkTransformation(SchemaAST.toEncoded(ast)) + switch (encoded._tag) { + case "String": + return "string" + case "Number": + return "number" + case "Boolean": + return "boolean" + case "Literal": + switch (typeof encoded.literal) { + case "string": + return "string" + case "number": + return "number" + case "boolean": + return "boolean" + default: + return "json" + } + case "Union": { + const scalarKinds = encoded.types.flatMap((type) => { + const kind = getColumnKind(type) + return kind === "json" ? [] : [kind] + }) + const distinctKinds = [...new Set(scalarKinds)] + return distinctKinds.length === 1 ? distinctKinds[0]! : "json" + } + default: + return "json" + } +} + +const makeColumnName = (key: string) => `__root_${key}` + +export const makeRootLevelFieldColumns = ( + schema: S.Schema, + idKey: PropertyKey +): readonly RootLevelFieldColumn[] => { + const encoded = walkTransformation(SchemaAST.toEncoded(schema.ast)) + if (!SchemaAST.isObjects(encoded)) { + return [] + } + + return encoded.propertySignatures.flatMap((property) => { + const key = String(property.name) + if (key === String(idKey) || key === "id" || key === "_etag") { + return [] + } + return [{ key, kind: getColumnKind(property.type), columnName: makeColumnName(key) }] + }) +} diff --git a/packages/infra/src/Store/SQL.ts b/packages/infra/src/Store/SQL.ts index d4484190d..c052ba5d2 100644 --- a/packages/infra/src/Store/SQL.ts +++ b/packages/infra/src/Store/SQL.ts @@ -15,7 +15,8 @@ import { SqlClient } from "effect/unstable/sql" import { OptimisticConcurrencyException } from "../errors.js" import { InfraLogger } from "../logger.js" import { annotateDb, type DbSystem } from "../otel.js" -import { buildWhereSQLQuery, logQuery, type SQLDialect, sqliteDialect } from "./SQL/query.js" +import { omitRootLevelFieldColumnsFromData, type RootLevelFieldColumn } from "./rootLevelFields.js" +import { buildWhereSQLQuery, logQuery, normalizeProjectedColumnValue, projectedColumnSqlType, quoteIdentifier, type SQLDialect, sqliteDialect } from "./SQL/query.js" import { makeETag } from "./utils.js" export type WithNsTransactionFn = (effect: Effect.Effect) => Effect.Effect @@ -26,23 +27,44 @@ export class WithNsTransaction /** @internal */ export const parseRow = ( - row: { id: string; _etag: string | null; data: string }, + row: { id: string; _etag: string | null; data: string } & Record, idKey: PropertyKey, - defaultValues: Partial + defaultValues: Partial, + rootLevelFieldColumns: readonly RootLevelFieldColumn[] = [] ): PersistenceModelType => { - const data = (typeof row.data === "string" ? JSON.parse(row.data) : row.data) as object - return { ...defaultValues, ...data, [idKey]: row.id, _etag: row._etag ?? undefined } as PersistenceModelType + const data = omitRootLevelFieldColumnsFromData( + (typeof row.data === "string" ? JSON.parse(row.data) : row.data) as Record, + rootLevelFieldColumns + ) + const projectedFields = rootLevelFieldColumns.reduce>((acc, column) => { + const value = row[column.columnName] + if (value !== null && value !== undefined) { + acc[column.key] = normalizeProjectedColumnValue(column, value) + } + return acc + }, {}) + return { + ...defaultValues, + ...data, + ...projectedFields, + [idKey]: row.id, + _etag: row._etag ?? undefined + } as PersistenceModelType } const parseSelectRow = ( row: Record, - idKey: PropertyKey + idKey: PropertyKey, + rootLevelFieldColumns: readonly RootLevelFieldColumn[] = [] ): any => { const result: Record = {} + const projectedFields = new Map(rootLevelFieldColumns.map((column) => [column.key, column] as const)) for (const [key, value] of Object.entries(row)) { if (key === "id") { result[idKey as string] = value result["id"] = value + } else if (projectedFields.has(key)) { + result[key] = normalizeProjectedColumnValue(projectedFields.get(key)!, value) } else if (typeof value === "string") { try { result[key] = JSON.parse(value) @@ -56,322 +78,377 @@ const parseSelectRow = ( return result } +const serializeProjectedValue = (system: DbSystem, column: RootLevelFieldColumn, value: unknown) => { + if (column.kind === "json") { + return value === undefined ? null : JSON.stringify(value) + } + if (column.kind === "boolean" && system === "sqlite" && value !== null && value !== undefined) { + return value === true ? 1 : 0 + } + return value ?? null +} + function makeSQLStoreInt(system: DbSystem, dialect: SQLDialect, jsonColumnType: string) { - return Effect.fnUntraced(function*({ prefix }: StorageConfig) { - const sql = yield* SqlClient.SqlClient - return { - make: Effect.fnUntraced(function*( - name: string, - idKey: IdKey, - seed?: Effect.Effect, E, R>, - config?: StoreConfig - ) { - type PM = PersistenceModelType - const tableName = `${prefix}${name}` - const defaultValues = config?.defaultValues ?? {} - - const resolveNamespace = !config?.allowNamespace - ? Effect.succeed("primary") - : storeId.pipe(Effect.map((namespace) => { - if (namespace !== "primary" && !config.allowNamespace!(namespace)) { - throw new Error(`Namespace ${namespace} not allowed!`) - } - return namespace - })) + return Effect.fnUntraced( + function*({ prefix, rootLevelFieldsWhenAvailable: rootLevelFieldsWhenAvailableDefault }: StorageConfig) { + const sql = yield* SqlClient.SqlClient + return { + make: Effect.fnUntraced( + function*( + name: string, + idKey: IdKey, + seed?: Effect.Effect, E, R>, + config?: StoreConfig + ) { + type PM = PersistenceModelType + const tableName = `${prefix}${name}` + const defaultValues = config?.defaultValues ?? {} + const rootLevelFieldColumns = config?.rootLevelFieldColumns ?? [] + const useRootLevelFields = config?.rootLevelFieldsWhenAvailable + ?? rootLevelFieldsWhenAvailableDefault + ?? false + const activeRootLevelFieldColumns = useRootLevelFields ? rootLevelFieldColumns : [] + const selectColumnsSql = activeRootLevelFieldColumns.length > 0 + ? `, ${activeRootLevelFieldColumns.map((column) => quoteIdentifier(column.columnName)).join(", ")}` + : "" + const projectedColumnsDefinitionSql = activeRootLevelFieldColumns.length > 0 + ? `, ${ + activeRootLevelFieldColumns + .map((column) => + `${quoteIdentifier(column.columnName)} ${projectedColumnSqlType(dialect, column.kind)}` + ) + .join(", ") + }` + : "" + + const resolveNamespace = !config?.allowNamespace + ? Effect.succeed("primary") + : storeId.pipe(Effect.map((namespace) => { + if (namespace !== "primary" && !config.allowNamespace!(namespace)) { + throw new Error(`Namespace ${namespace} not allowed!`) + } + return namespace + })) + + const exec = (query: string, params?: readonly unknown[]) => + sql.unsafe(query, params as any).pipe(Effect.orDie) + + const ensureTable = sql + .unsafe( + `CREATE TABLE IF NOT EXISTS "${tableName}" (id TEXT NOT NULL, _namespace TEXT NOT NULL DEFAULT 'primary', _etag TEXT, data ${jsonColumnType} NOT NULL${projectedColumnsDefinitionSql}, PRIMARY KEY (id, _namespace))` + ) + .pipe( + Effect.andThen( + sql.unsafe( + `CREATE TABLE IF NOT EXISTS "_migrations" (id TEXT NOT NULL, version TEXT NOT NULL, PRIMARY KEY (id, version))` + ) + ), + Effect.orDie, + Effect.asVoid + ) - const ensureTable = sql - .unsafe( - `CREATE TABLE IF NOT EXISTS "${tableName}" (id TEXT NOT NULL, _namespace TEXT NOT NULL DEFAULT 'primary', _etag TEXT, data ${jsonColumnType} NOT NULL, PRIMARY KEY (id, _namespace))` - ) - .pipe( - Effect.andThen( - sql.unsafe( - `CREATE TABLE IF NOT EXISTS "_migrations" (id TEXT NOT NULL, version TEXT NOT NULL, PRIMARY KEY (id, version))` + const toRow = (e: PM) => { + const newE = makeETag(e) + const id = newE[idKey] as string + const { _etag, [idKey]: _id, ...rest } = newE as any + const data = JSON.stringify(omitRootLevelFieldColumnsFromData(rest, activeRootLevelFieldColumns)) + const rootLevelFieldValues = activeRootLevelFieldColumns.map((column) => + serializeProjectedValue(system, column, rest[column.key]) ) - ), - Effect.orDie, - Effect.asVoid - ) + return { id, _etag: newE._etag!, data, item: newE, rootLevelFieldValues } + } - const toRow = (e: PM) => { - const newE = makeETag(e) - const id = newE[idKey] as string - const { _etag, [idKey]: _id, ...rest } = newE as any - const data = JSON.stringify(rest) - return { id, _etag: newE._etag!, data, item: newE } - } + const setInternal = Effect.fnUntraced(function*(e: PM, ns: string) { + const row = toRow(e) + if (e._etag) { + const projectedSetSql = activeRootLevelFieldColumns.length > 0 + ? `, ${ + activeRootLevelFieldColumns.map((column) => `${quoteIdentifier(column.columnName)} = ?`).join(", ") + }` + : "" + yield* exec( + `UPDATE "${tableName}" SET _etag = ?, data = ?${projectedSetSql} WHERE id = ? AND _etag = ? AND _namespace = ?`, + [row._etag, row.data, ...row.rootLevelFieldValues, row.id, e._etag, ns] + ) + const existing = yield* exec( + `SELECT _etag FROM "${tableName}" WHERE id = ? AND _namespace = ?`, + [row.id, ns] + ) + const current = (existing as any[])[0] + if (!current || current._etag !== row._etag) { + if (current) { + return yield* new OptimisticConcurrencyException({ + type: name, + id: row.id, + current: current._etag, + found: e._etag, + code: 412 + }) + } + return yield* new OptimisticConcurrencyException({ + type: name, + id: row.id, + current: "", + found: e._etag, + code: 404 + }) + } + } else { + const projectedColumnsSql = activeRootLevelFieldColumns.length > 0 + ? `, ${activeRootLevelFieldColumns.map((column) => quoteIdentifier(column.columnName)).join(", ")}` + : "" + const projectedPlaceholders = activeRootLevelFieldColumns.map(() => "?").join(", ") + yield* exec( + `INSERT INTO "${tableName}" (id, _namespace, _etag, data${projectedColumnsSql}) VALUES (?, ?, ?, ?${ + projectedPlaceholders.length > 0 ? `, ${projectedPlaceholders}` : "" + })`, + [row.id, ns, row._etag, row.data, ...row.rootLevelFieldValues] + ) + } + return row.item + }) - const exec = (query: string, params?: readonly unknown[]) => sql.unsafe(query, params as any).pipe(Effect.orDie) + const bulkSetInternal = (items: NonEmptyReadonlyArray, ns: string) => + sql + .withTransaction(Effect.forEach(items, (e) => setInternal(e, ns))) + .pipe( + Effect.orDie, + Effect.map((_) => _ as unknown as NonEmptyReadonlyArray) + ) - const setInternal = Effect.fnUntraced(function*(e: PM, ns: string) { - const row = toRow(e) - if (e._etag) { - yield* exec( - `UPDATE "${tableName}" SET _etag = ?, data = ? WHERE id = ? AND _etag = ? AND _namespace = ?`, - [row._etag, row.data, row.id, e._etag, ns] - ) - const existing = yield* exec( - `SELECT _etag FROM "${tableName}" WHERE id = ? AND _namespace = ?`, - [row.id, ns] - ) - const current = (existing as any[])[0] - if (!current || current._etag !== row._etag) { - if (current) { - return yield* new OptimisticConcurrencyException({ - type: name, - id: row.id, - current: current._etag, - found: e._etag, - code: 412 - }) + const ctx = yield* Effect.context() + const seedCache = new Map>() + const makeSeedEffect = Effect.fnUntraced(function*(ns: string) { + yield* ensureTable + if (!seed) return + const existing = yield* exec( + `SELECT id FROM "_migrations" WHERE id = ? AND version = ?`, + [`${tableName}::${ns}`, tableName] + ) + if ((existing as any[]).length > 0) return + yield* InfraLogger.logInfo(`Seeding data for ${name} (namespace: ${ns})`) + const items = yield* seed.pipe(Effect.provide(ctx), Effect.orDie) + const ne = toNonEmptyArray([...items]) + if (Option.isSome(ne)) yield* bulkSetInternal(ne.value, ns) + yield* exec( + `INSERT INTO "_migrations" (id, version) VALUES (?, ?)`, + [`${tableName}::${ns}`, tableName] + ) + }) + const seedNamespace = (ns: string) => { + const existingEffect = seedCache.get(ns) + if (existingEffect) { + return existingEffect } - return yield* new OptimisticConcurrencyException({ - type: name, - id: row.id, - current: "", - found: e._etag, - code: 404 - }) + const seedEffect = Effect.cached(Effect.uninterruptible(makeSeedEffect(ns))).pipe(Effect.runSync) + seedCache.set(ns, seedEffect) + return seedEffect } - } else { - yield* exec( - `INSERT INTO "${tableName}" (id, _namespace, _etag, data) VALUES (?, ?, ?, ?)`, - [row.id, ns, row._etag, row.data] - ) - } - return row.item - }) - - const bulkSetInternal = (items: NonEmptyReadonlyArray, ns: string) => - sql - .withTransaction(Effect.forEach(items, (e) => setInternal(e, ns))) - .pipe( - Effect.orDie, - Effect.map((_) => _ as unknown as NonEmptyReadonlyArray) - ) + const s: Store = { + seedNamespace: (ns) => seedNamespace(ns), - const ctx = yield* Effect.context() - const seedCache = new Map>() - const makeSeedEffect = Effect.fnUntraced(function*(ns: string) { - yield* ensureTable - if (!seed) return - const existing = yield* exec( - `SELECT id FROM "_migrations" WHERE id = ? AND version = ?`, - [`${tableName}::${ns}`, tableName] - ) - if ((existing as any[]).length > 0) return - yield* InfraLogger.logInfo(`Seeding data for ${name} (namespace: ${ns})`) - const items = yield* seed.pipe(Effect.provide(ctx), Effect.orDie) - const ne = toNonEmptyArray([...items]) - if (Option.isSome(ne)) yield* bulkSetInternal(ne.value, ns) - yield* exec( - `INSERT INTO "_migrations" (id, version) VALUES (?, ?)`, - [`${tableName}::${ns}`, tableName] - ) - }) - const seedNamespace = (ns: string) => { - let cached = seedCache.get(ns) - if (!cached) { - cached = Effect.cached(Effect.uninterruptible(makeSeedEffect(ns))).pipe(Effect.runSync) - seedCache.set(ns, cached) - } - return cached - } - const s: Store = { - seedNamespace: (ns) => seedNamespace(ns), + all: resolveNamespace.pipe( + Effect.flatMap((ns) => { + const sqlText = `SELECT id, _etag, data${selectColumnsSql} FROM "${tableName}" WHERE _namespace = ?` + return exec(sqlText, [ns]) + .pipe( + Effect.map((rows) => + (rows as any[]).map((r) => + parseRow(r, idKey, defaultValues, activeRootLevelFieldColumns) + ) + ), + annotateDb({ + operation: "all", + system, + collection: tableName, + namespace: ns, + entity: name, + query: sqlText + }) + ) + }) + ), - all: resolveNamespace.pipe( - Effect.flatMap((ns) => { - const sqlText = `SELECT id, _etag, data FROM "${tableName}" WHERE _namespace = ?` - return exec(sqlText, [ns]) - .pipe( - Effect.map((rows) => (rows as any[]).map((r) => parseRow(r, idKey, defaultValues))), - annotateDb({ - operation: "all", - system, - collection: tableName, - namespace: ns, - entity: name, - query: sqlText + find: (id) => + resolveNamespace.pipe( + Effect.flatMap((ns) => { + const sqlText = + `SELECT id, _etag, data${selectColumnsSql} FROM "${tableName}" WHERE id = ? AND _namespace = ?` + return exec(sqlText, [id, ns]) + .pipe( + Effect.map((rows) => { + const row = (rows as any[])[0] + return row + ? Option.some(parseRow(row, idKey, defaultValues, activeRootLevelFieldColumns)) + : Option.none() + }), + annotateDb({ + operation: "find", + system, + collection: tableName, + namespace: ns, + entity: name, + query: sqlText, + extra: { "app.entity.id": id } + }) + ) }) - ) - }) - ), + ), + + filter: (f: FilterArgs) => { + const filter = f + .filter + type M = U extends undefined ? Encoded + : Pick + return resolveNamespace + .pipe(Effect + .flatMap((ns) => + Effect + .sync(() => { + return buildWhereSQLQuery( + dialect, + idKey, + filter ? [{ t: "where-scope", result: filter, relation: "some" }] : [], + tableName, + defaultValues, + f + .select as + | NonEmptyReadonlyArray< + string | { + key: string + subKeys: readonly string[] + } | { + key: string + computed: ComputedProjectionIrExpression + } + > + | undefined, + f + .order, + f + .skip, + f + .limit, + ns, + activeRootLevelFieldColumns + ) + }) + .pipe( + Effect + .tap((q) => logQuery(q)), + Effect.tap((q) => Effect.annotateCurrentSpan({ "db.query.text": q.sql })), + Effect.flatMap((q) => + exec(q.sql, q.params).pipe( + Effect.map((rows) => { + if (f.select) { + return (rows as any[]).map((r) => { + const selected = parseSelectRow(r, idKey, activeRootLevelFieldColumns) + return { + ...Struct.pick( + defaultValues as any, + f.select!.filter((_) => typeof _ === "string") as never[] + ), + ...selected + } as M + }) + } + return (rows as any[]).map((r) => + parseRow(r, idKey, defaultValues, activeRootLevelFieldColumns) as any as M + ) + }) + ) + ), + annotateDb({ + operation: "filter", + system, + collection: tableName, + namespace: ns, + entity: name + }) + ) + )) + }, - find: (id) => - resolveNamespace.pipe( - Effect.flatMap((ns) => { - const sqlText = `SELECT id, _etag, data FROM "${tableName}" WHERE id = ? AND _namespace = ?` - return exec(sqlText, [id, ns]) - .pipe( - Effect.map((rows) => { - const row = (rows as any[])[0] - return row - ? Option.some(parseRow(row, idKey, defaultValues)) - : Option.none() - }), + set: (e) => + resolveNamespace.pipe(Effect.flatMap((ns) => + setInternal(e, ns).pipe( annotateDb({ - operation: "find", + operation: "set", system, collection: tableName, namespace: ns, entity: name, - query: sqlText, - extra: { "app.entity.id": id } + extra: { "app.entity.id": e[idKey] } }) ) - }) - ), - - filter: (f: FilterArgs) => { - const filter = f - .filter - type M = U extends undefined ? Encoded - : Pick - return resolveNamespace - .pipe(Effect - .flatMap((ns) => - Effect - .sync(() => { - return buildWhereSQLQuery( - dialect, - idKey, - filter ? [{ t: "where-scope", result: filter, relation: "some" }] : [], - tableName, - defaultValues, - f - .select as - | NonEmptyReadonlyArray< - string | { - key: string - subKeys: readonly string[] - } | { - key: string - computed: ComputedProjectionIrExpression - } - > - | undefined, - f - .order, - f - .skip, - f - .limit, - ns - ) + )), + + batchSet: (items) => + resolveNamespace.pipe(Effect.flatMap((ns) => + bulkSetInternal(items, ns).pipe( + annotateDb({ + operation: "batchSet", + system, + collection: tableName, + namespace: ns, + entity: name + }) + ) + )), + + bulkSet: (items) => + resolveNamespace.pipe(Effect.flatMap((ns) => + bulkSetInternal(items, ns).pipe( + annotateDb({ + operation: "bulkSet", + system, + collection: tableName, + namespace: ns, + entity: name }) + ) + )), + + batchRemove: (ids) => { + const placeholders = ids.map(() => "?").join(", ") + return resolveNamespace.pipe(Effect.flatMap((ns) => { + const sqlText = `DELETE FROM "${tableName}" WHERE id IN (${placeholders}) AND _namespace = ?` + return exec(sqlText, [...ids, ns]) .pipe( - Effect - .tap((q) => logQuery(q)), - Effect.tap((q) => Effect.annotateCurrentSpan({ "db.query.text": q.sql })), - Effect.flatMap((q) => - exec(q.sql, q.params).pipe( - Effect.map((rows) => { - if (f.select) { - return (rows as any[]).map((r) => { - const selected = parseSelectRow(r, idKey) - return { - ...Struct.pick( - defaultValues as any, - f.select!.filter((_) => typeof _ === "string") as never[] - ), - ...selected - } as M - }) - } - return (rows as any[]).map((r) => parseRow(r, idKey, defaultValues) as any as M) - }) - ) - ), + Effect.asVoid, annotateDb({ - operation: "filter", + operation: "batchRemove", system, collection: tableName, namespace: ns, - entity: name + entity: name, + query: sqlText }) ) - )) - }, - - set: (e) => - resolveNamespace.pipe(Effect.flatMap((ns) => - setInternal(e, ns).pipe( - annotateDb({ - operation: "set", - system, - collection: tableName, - namespace: ns, - entity: name, - extra: { "app.entity.id": e[idKey] } - }) - ) - )), - - batchSet: (items) => - resolveNamespace.pipe(Effect.flatMap((ns) => - bulkSetInternal(items, ns).pipe( - annotateDb({ - operation: "batchSet", - system, - collection: tableName, - namespace: ns, - entity: name - }) - ) - )), - - bulkSet: (items) => - resolveNamespace.pipe(Effect.flatMap((ns) => - bulkSetInternal(items, ns).pipe( - annotateDb({ - operation: "bulkSet", - system, - collection: tableName, - namespace: ns, - entity: name - }) - ) - )), + })) + }, - batchRemove: (ids) => { - const placeholders = ids.map(() => "?").join(", ") - return resolveNamespace.pipe(Effect.flatMap((ns) => { - const sqlText = `DELETE FROM "${tableName}" WHERE id IN (${placeholders}) AND _namespace = ?` - return exec(sqlText, [...ids, ns]) - .pipe( - Effect.asVoid, + queryRaw: (query) => + s.all.pipe( + Effect.map(query.memory), annotateDb({ - operation: "batchRemove", + operation: "queryRaw", system, collection: tableName, - namespace: ns, - entity: name, - query: sqlText + entity: name }) ) - })) - }, - - queryRaw: (query) => - s.all.pipe( - Effect.map(query.memory), - annotateDb({ - operation: "queryRaw", - system, - collection: tableName, - entity: name - }) - ) - } + } - // Eagerly seed primary namespace on initialization - yield* seedNamespace("primary") + // Eagerly seed primary namespace on initialization + yield* seedNamespace("primary") - return s - }) + return s + } + ) + } } - }) + ) } type WithNsSqlFn = ( @@ -381,7 +458,7 @@ type WithNsSqlFn = ( function makeSQLiteStorePerNs( withNsSql: WithNsSqlFn, - { prefix }: StorageConfig + { prefix, rootLevelFieldsWhenAvailable: rootLevelFieldsWhenAvailableDefault }: StorageConfig ) { return { make: Effect.fnUntraced(function*( @@ -393,6 +470,21 @@ function makeSQLiteStorePerNs( type PM = PersistenceModelType const tableName = `${prefix}${name}` const defaultValues = config?.defaultValues ?? {} + const rootLevelFieldColumns = config?.rootLevelFieldColumns ?? [] + const useRootLevelFields = config?.rootLevelFieldsWhenAvailable ?? rootLevelFieldsWhenAvailableDefault ?? false + const activeRootLevelFieldColumns = useRootLevelFields ? rootLevelFieldColumns : [] + const selectColumnsSql = activeRootLevelFieldColumns.length > 0 + ? `, ${activeRootLevelFieldColumns.map((column) => quoteIdentifier(column.columnName)).join(", ")}` + : "" + const projectedColumnsDefinitionSql = activeRootLevelFieldColumns.length > 0 + ? `, ${ + activeRootLevelFieldColumns + .map((column) => + `${quoteIdentifier(column.columnName)} ${projectedColumnSqlType(sqliteDialect, column.kind)}` + ) + .join(", ") + }` + : "" const resolveNamespace = !config?.allowNamespace ? Effect.succeed("primary") @@ -407,8 +499,11 @@ function makeSQLiteStorePerNs( const newE = makeETag(e) const id = newE[idKey] as string const { _etag, [idKey]: _id, ...rest } = newE as any - const data = JSON.stringify(rest) - return { id, _etag: newE._etag!, data, item: newE } + const data = JSON.stringify(omitRootLevelFieldColumnsFromData(rest, activeRootLevelFieldColumns)) + const rootLevelFieldValues = activeRootLevelFieldColumns.map((column) => + serializeProjectedValue("sqlite", column, rest[column.key]) + ) + return { id, _etag: newE._etag!, data, item: newE, rootLevelFieldValues } } const exec = (ns: string, query: string, params?: readonly unknown[]) => @@ -418,7 +513,7 @@ function makeSQLiteStorePerNs( withNsSql(ns, (sql) => sql .unsafe( - `CREATE TABLE IF NOT EXISTS "${tableName}" (id TEXT NOT NULL PRIMARY KEY, _etag TEXT, data JSON NOT NULL)` + `CREATE TABLE IF NOT EXISTS "${tableName}" (id TEXT NOT NULL PRIMARY KEY, _etag TEXT, data JSON NOT NULL${projectedColumnsDefinitionSql})` ) .pipe( Effect.andThen( @@ -433,10 +528,13 @@ function makeSQLiteStorePerNs( const setInternal = Effect.fnUntraced(function*(e: PM, ns: string) { const row = toRow(e) if (e._etag) { + const projectedSetSql = activeRootLevelFieldColumns.length > 0 + ? `, ${activeRootLevelFieldColumns.map((column) => `${quoteIdentifier(column.columnName)} = ?`).join(", ")}` + : "" yield* exec( ns, - `UPDATE "${tableName}" SET _etag = ?, data = ? WHERE id = ? AND _etag = ?`, - [row._etag, row.data, row.id, e._etag] + `UPDATE "${tableName}" SET _etag = ?, data = ?${projectedSetSql} WHERE id = ? AND _etag = ?`, + [row._etag, row.data, ...row.rootLevelFieldValues, row.id, e._etag] ) const existing = yield* exec( ns, @@ -463,10 +561,16 @@ function makeSQLiteStorePerNs( }) } } else { + const projectedColumnsSql = activeRootLevelFieldColumns.length > 0 + ? `, ${activeRootLevelFieldColumns.map((column) => quoteIdentifier(column.columnName)).join(", ")}` + : "" + const projectedPlaceholders = activeRootLevelFieldColumns.map(() => "?").join(", ") yield* exec( ns, - `INSERT INTO "${tableName}" (id, _etag, data) VALUES (?, ?, ?)`, - [row.id, row._etag, row.data] + `INSERT INTO "${tableName}" (id, _etag, data${projectedColumnsSql}) VALUES (?, ?, ?${ + projectedPlaceholders.length > 0 ? `, ${projectedPlaceholders}` : "" + })`, + [row.id, row._etag, row.data, ...row.rootLevelFieldValues] ) } return row.item @@ -503,22 +607,25 @@ function makeSQLiteStorePerNs( ) }) const seedNamespace = (ns: string) => { - let cached = seedCache.get(ns) - if (!cached) { - cached = Effect.cached(Effect.uninterruptible(makeSeedEffect(ns))).pipe(Effect.runSync) - seedCache.set(ns, cached) + const existingEffect = seedCache.get(ns) + if (existingEffect) { + return existingEffect } - return cached + const seedEffect = Effect.cached(Effect.uninterruptible(makeSeedEffect(ns))).pipe(Effect.runSync) + seedCache.set(ns, seedEffect) + return seedEffect } const s: Store = { seedNamespace: (ns) => seedNamespace(ns), all: resolveNamespace.pipe(Effect.flatMap((ns) => { - const sqlText = `SELECT id, _etag, data FROM "${tableName}"` + const sqlText = `SELECT id, _etag, data${selectColumnsSql} FROM "${tableName}"` return exec(ns, sqlText) .pipe( - Effect.map((rows) => (rows as any[]).map((r) => parseRow(r, idKey, defaultValues))), + Effect.map((rows) => + (rows as any[]).map((r) => parseRow(r, idKey, defaultValues, activeRootLevelFieldColumns)) + ), annotateDb({ operation: "all", system: "sqlite", @@ -533,13 +640,13 @@ function makeSQLiteStorePerNs( find: (id) => resolveNamespace.pipe( Effect.flatMap((ns) => { - const sqlText = `SELECT id, _etag, data FROM "${tableName}" WHERE id = ?` + const sqlText = `SELECT id, _etag, data${selectColumnsSql} FROM "${tableName}" WHERE id = ?` return exec(ns, sqlText, [id]) .pipe( Effect.map((rows) => { const row = (rows as any[])[0] return row - ? Option.some(parseRow(row, idKey, defaultValues)) + ? Option.some(parseRow(row, idKey, defaultValues, activeRootLevelFieldColumns)) : Option.none() }), annotateDb({ @@ -588,7 +695,9 @@ function makeSQLiteStorePerNs( f .skip, f - .limit + .limit, + undefined, + activeRootLevelFieldColumns ) ) .pipe( @@ -600,7 +709,7 @@ function makeSQLiteStorePerNs( Effect.map((rows) => { if (f.select) { return (rows as any[]).map((r) => { - const selected = parseSelectRow(r, idKey) + const selected = parseSelectRow(r, idKey, activeRootLevelFieldColumns) return { ...Struct.pick( defaultValues as any, @@ -610,7 +719,9 @@ function makeSQLiteStorePerNs( } as M }) } - return (rows as any[]).map((r) => parseRow(r, idKey, defaultValues) as any as M) + return (rows as any[]).map((r) => + parseRow(r, idKey, defaultValues, activeRootLevelFieldColumns) as any as M + ) }) ) ), diff --git a/packages/infra/src/Store/SQL/Pg.ts b/packages/infra/src/Store/SQL/Pg.ts index fe80dd4cf..25b0cf2e2 100644 --- a/packages/infra/src/Store/SQL/Pg.ts +++ b/packages/infra/src/Store/SQL/Pg.ts @@ -12,28 +12,50 @@ import { SqlClient } from "effect/unstable/sql" import { OptimisticConcurrencyException } from "../../errors.js" import { InfraLogger } from "../../logger.js" import { annotateDb } from "../../otel.js" +import { omitRootLevelFieldColumnsFromData, type RootLevelFieldColumn } from "../rootLevelFields.js" import { makeETag } from "../utils.js" -import { buildWhereSQLQuery, logQuery, pgDialect } from "./query.js" +import { buildWhereSQLQuery, logQuery, normalizeProjectedColumnValue, pgDialect, projectedColumnSqlType, quoteIdentifier } from "./query.js" const parseRow = ( - row: { id: string; _etag: string | null; data: unknown }, + row: { id: string; _etag: string | null; data: unknown } & Record, idKey: PropertyKey, - defaultValues: Partial + defaultValues: Partial, + rootLevelFieldColumns: readonly RootLevelFieldColumn[] = [] ): PersistenceModelType => { - const data = (typeof row.data === "string" ? JSON.parse(row.data) : row.data) as object - return { ...defaultValues, ...data, [idKey]: row.id, _etag: row._etag ?? undefined } as PersistenceModelType + const data = omitRootLevelFieldColumnsFromData( + (typeof row.data === "string" ? JSON.parse(row.data) : row.data) as Record, + rootLevelFieldColumns + ) + const projectedFields = rootLevelFieldColumns.reduce>((acc, column) => { + const value = row[column.columnName] + if (value !== null && value !== undefined) { + acc[column.key] = normalizeProjectedColumnValue(column, value) + } + return acc + }, {}) + return { + ...defaultValues, + ...data, + ...projectedFields, + [idKey]: row.id, + _etag: row._etag ?? undefined + } as PersistenceModelType } const parseSelectRow = ( row: Record, idKey: PropertyKey, - defaultValues: Record + defaultValues: Record, + rootLevelFieldColumns: readonly RootLevelFieldColumn[] = [] ): any => { const result: Record = { ...defaultValues } + const projectedFields = new Map(rootLevelFieldColumns.map((column) => [column.key, column] as const)) for (const [key, value] of Object.entries(row)) { if (key === "id") { result[idKey as string] = value result["id"] = value + } else if (projectedFields.has(key)) { + result[key] = normalizeProjectedColumnValue(projectedFields.get(key)!, value) } else { result[key] = value } @@ -41,321 +63,370 @@ const parseSelectRow = ( return result } -const makePgStore = Effect.fnUntraced(function*({ prefix }: StorageConfig) { - const sql = yield* SqlClient.SqlClient - return { - make: Effect.fnUntraced(function*( - name: string, - idKey: IdKey, - seed?: Effect.Effect, E, R>, - config?: StoreConfig - ) { - type PM = PersistenceModelType - const tableName = `${prefix}${name}` - const defaultValues = config?.defaultValues ?? {} - - const resolveNamespace = !config?.allowNamespace - ? Effect.succeed("primary") - : storeId.pipe(Effect.map((namespace) => { - if (namespace !== "primary" && !config.allowNamespace!(namespace)) { - throw new Error(`Namespace ${namespace} not allowed!`) - } - return namespace - })) - - const ensureTable = sql - .unsafe( - `CREATE TABLE IF NOT EXISTS "${tableName}" (id TEXT NOT NULL, _namespace TEXT NOT NULL DEFAULT 'primary', _etag TEXT, data JSONB NOT NULL, PRIMARY KEY (id, _namespace))` - ) - .pipe( - Effect.andThen( - sql.unsafe( - `CREATE TABLE IF NOT EXISTS "_migrations" (id TEXT NOT NULL, version TEXT NOT NULL, PRIMARY KEY (id, version))` - ) - ), - Effect.orDie, - Effect.asVoid - ) +const makePgStore = Effect.fnUntraced( + function*({ prefix, rootLevelFieldsWhenAvailable: rootLevelFieldsWhenAvailableDefault }: StorageConfig) { + const sql = yield* SqlClient.SqlClient + return { + make: Effect.fnUntraced(function*( + name: string, + idKey: IdKey, + seed?: Effect.Effect, E, R>, + config?: StoreConfig + ) { + type PM = PersistenceModelType + const tableName = `${prefix}${name}` + const defaultValues = config?.defaultValues ?? {} + const rootLevelFieldColumns = config?.rootLevelFieldColumns ?? [] + const useRootLevelFields = config?.rootLevelFieldsWhenAvailable ?? rootLevelFieldsWhenAvailableDefault ?? false + const activeRootLevelFieldColumns = useRootLevelFields ? rootLevelFieldColumns : [] + const selectColumnsSql = activeRootLevelFieldColumns.length > 0 + ? `, ${activeRootLevelFieldColumns.map((column) => quoteIdentifier(column.columnName)).join(", ")}` + : "" + const projectedColumnsDefinitionSql = activeRootLevelFieldColumns.length > 0 + ? `, ${ + activeRootLevelFieldColumns + .map((column) => + `${quoteIdentifier(column.columnName)} ${projectedColumnSqlType(pgDialect, column.kind)}` + ) + .join(", ") + }` + : "" - const toRow = (e: PM) => { - const newE = makeETag(e) - const id = newE[idKey] as string - const { _etag, [idKey]: _id, ...rest } = newE as any - const data = JSON.stringify(rest) - return { id, _etag: newE._etag!, data, item: newE } - } + const resolveNamespace = !config?.allowNamespace + ? Effect.succeed("primary") + : storeId.pipe(Effect.map((namespace) => { + if (namespace !== "primary" && !config.allowNamespace!(namespace)) { + throw new Error(`Namespace ${namespace} not allowed!`) + } + return namespace + })) - const exec = (query: string, params?: readonly unknown[]) => sql.unsafe(query, params as any).pipe(Effect.orDie) + const exec = (query: string, params?: readonly unknown[]) => sql.unsafe(query, params as any).pipe(Effect.orDie) - const setInternal = Effect.fnUntraced(function*(e: PM, ns: string) { - const row = toRow(e) - if (e._etag) { - yield* exec( - `UPDATE "${tableName}" SET _etag = $1, data = $2 WHERE id = $3 AND _etag = $4 AND _namespace = $5`, - [row._etag, row.data, row.id, e._etag, ns] + const ensureTable = sql + .unsafe( + `CREATE TABLE IF NOT EXISTS "${tableName}" (id TEXT NOT NULL, _namespace TEXT NOT NULL DEFAULT 'primary', _etag TEXT, data JSONB NOT NULL${projectedColumnsDefinitionSql}, PRIMARY KEY (id, _namespace))` ) - const existing = yield* exec( - `SELECT _etag FROM "${tableName}" WHERE id = $1 AND _namespace = $2`, - [row.id, ns] + .pipe( + Effect.andThen( + sql.unsafe( + `CREATE TABLE IF NOT EXISTS "_migrations" (id TEXT NOT NULL, version TEXT NOT NULL, PRIMARY KEY (id, version))` + ) + ), + Effect.orDie, + Effect.asVoid ) - const current = (existing as any[])[0] - if (!current || current._etag !== row._etag) { - if (current) { + + const toRow = (e: PM) => { + const newE = makeETag(e) + const id = newE[idKey] as string + const { _etag, [idKey]: _id, ...rest } = newE as any + const data = JSON.stringify(omitRootLevelFieldColumnsFromData(rest, activeRootLevelFieldColumns)) + const rootLevelFieldValues = activeRootLevelFieldColumns.map((column) => { + if (column.kind === "json") { + return rest[column.key] === undefined ? null : JSON.stringify(rest[column.key]) + } + return rest[column.key] ?? null + }) + return { id, _etag: newE._etag!, data, item: newE, rootLevelFieldValues } + } + + const setInternal = Effect.fnUntraced(function*(e: PM, ns: string) { + const row = toRow(e) + if (e._etag) { + const projectedSetSql = activeRootLevelFieldColumns.length > 0 + ? `, ${ + activeRootLevelFieldColumns + .map((column, index) => `${quoteIdentifier(column.columnName)} = $${index + 3}`) + .join(", ") + }` + : "" + const idParam = row.rootLevelFieldValues.length + 3 + const etagParam = row.rootLevelFieldValues.length + 4 + const namespaceParam = row.rootLevelFieldValues.length + 5 + yield* exec( + `UPDATE "${tableName}" SET _etag = $1, data = $2${projectedSetSql} WHERE id = $${idParam} AND _etag = $${etagParam} AND _namespace = $${namespaceParam}`, + [row._etag, row.data, ...row.rootLevelFieldValues, row.id, e._etag, ns] + ) + const existing = yield* exec( + `SELECT _etag FROM "${tableName}" WHERE id = $1 AND _namespace = $2`, + [row.id, ns] + ) + const current = (existing as any[])[0] + if (!current || current._etag !== row._etag) { + if (current) { + return yield* new OptimisticConcurrencyException({ + type: name, + id: row.id, + current: current._etag, + found: e._etag, + code: 412 + }) + } return yield* new OptimisticConcurrencyException({ type: name, id: row.id, - current: current._etag, + current: "", found: e._etag, - code: 412 + code: 404 }) } - return yield* new OptimisticConcurrencyException({ - type: name, - id: row.id, - current: "", - found: e._etag, - code: 404 - }) + } else { + const projectedColumnsSql = activeRootLevelFieldColumns.length > 0 + ? `, ${activeRootLevelFieldColumns.map((column) => quoteIdentifier(column.columnName)).join(", ")}` + : "" + const projectedValuesSql = activeRootLevelFieldColumns + .map((_, index) => `$${index + 5}`) + .join(", ") + yield* exec( + `INSERT INTO "${tableName}" (id, _namespace, _etag, data${projectedColumnsSql}) VALUES ($1, $2, $3, $4${ + projectedValuesSql.length > 0 ? `, ${projectedValuesSql}` : "" + })`, + [row.id, ns, row._etag, row.data, ...row.rootLevelFieldValues] + ) } - } else { - yield* exec( - `INSERT INTO "${tableName}" (id, _namespace, _etag, data) VALUES ($1, $2, $3, $4)`, - [row.id, ns, row._etag, row.data] - ) - } - return row.item - }) + return row.item + }) - const bulkSetInternal = (items: NonEmptyReadonlyArray, ns: string) => - sql - .withTransaction(Effect.forEach(items, (e) => setInternal(e, ns))) - .pipe( - Effect.orDie, - Effect.map((_) => _ as unknown as NonEmptyReadonlyArray) - ) + const bulkSetInternal = (items: NonEmptyReadonlyArray, ns: string) => + sql + .withTransaction(Effect.forEach(items, (e) => setInternal(e, ns))) + .pipe( + Effect.orDie, + Effect.map((_) => _ as unknown as NonEmptyReadonlyArray) + ) - const ctx = yield* Effect.context() - const seedCache = new Map>() - const makeSeedEffect = Effect.fnUntraced(function*(ns: string) { - yield* ensureTable - if (!seed) return - const existing = yield* exec( - `SELECT id FROM "_migrations" WHERE id = $1 AND version = $2`, - [`${tableName}::${ns}`, tableName] - ) - if ((existing as any[]).length > 0) return - yield* InfraLogger.logInfo(`Seeding data for ${name} (namespace: ${ns})`) - const items = yield* seed.pipe(Effect.provide(ctx), Effect.orDie) - const ne = toNonEmptyArray([...items]) - if (Option.isSome(ne)) yield* bulkSetInternal(ne.value, ns) - yield* exec( - `INSERT INTO "_migrations" (id, version) VALUES ($1, $2)`, - [`${tableName}::${ns}`, tableName] - ) - }) - const seedNamespace = (ns: string) => { - let cached = seedCache.get(ns) - if (!cached) { - cached = Effect.cached(Effect.uninterruptible(makeSeedEffect(ns))).pipe(Effect.runSync) - seedCache.set(ns, cached) + const ctx = yield* Effect.context() + const seedCache = new Map>() + const makeSeedEffect = Effect.fnUntraced(function*(ns: string) { + yield* ensureTable + if (!seed) return + const existing = yield* exec( + `SELECT id FROM "_migrations" WHERE id = $1 AND version = $2`, + [`${tableName}::${ns}`, tableName] + ) + if ((existing as any[]).length > 0) return + yield* InfraLogger.logInfo(`Seeding data for ${name} (namespace: ${ns})`) + const items = yield* seed.pipe(Effect.provide(ctx), Effect.orDie) + const ne = toNonEmptyArray([...items]) + if (Option.isSome(ne)) yield* bulkSetInternal(ne.value, ns) + yield* exec( + `INSERT INTO "_migrations" (id, version) VALUES ($1, $2)`, + [`${tableName}::${ns}`, tableName] + ) + }) + const seedNamespace = (ns: string) => { + let cached = seedCache.get(ns) + if (!cached) { + cached = Effect.cached(Effect.uninterruptible(makeSeedEffect(ns))).pipe(Effect.runSync) + seedCache.set(ns, cached) + } + return cached } - return cached - } - const s: Store = { - seedNamespace: (ns) => seedNamespace(ns), - - all: resolveNamespace.pipe( - Effect.flatMap((ns) => { - const sqlText = `SELECT id, _etag, data FROM "${tableName}" WHERE _namespace = $1` - return exec(sqlText, [ns]) - .pipe( - Effect.map((rows) => (rows as any[]).map((r) => parseRow(r, idKey, defaultValues))), - annotateDb({ - operation: "all", - system: "postgresql", - collection: tableName, - namespace: ns, - entity: name, - query: sqlText - }) - ) - }) - ), + const s: Store = { + seedNamespace: (ns) => seedNamespace(ns), - find: (id) => - resolveNamespace.pipe(Effect - .flatMap((ns) => { - const sqlText = `SELECT id, _etag, data FROM "${tableName}" WHERE id = $1 AND _namespace = $2` - return exec(sqlText, [id, ns]) + all: resolveNamespace.pipe( + Effect.flatMap((ns) => { + const sqlText = `SELECT id, _etag, data${selectColumnsSql} FROM "${tableName}" WHERE _namespace = $1` + return exec(sqlText, [ns]) .pipe( - Effect.map((rows) => { - const row = (rows as any[])[0] - return row - ? Option.some(parseRow(row, idKey, defaultValues)) - : Option.none() - }), + Effect.map((rows) => + (rows as any[]).map((r) => parseRow(r, idKey, defaultValues, activeRootLevelFieldColumns)) + ), annotateDb({ - operation: "find", + operation: "all", system: "postgresql", collection: tableName, namespace: ns, entity: name, - query: sqlText, - extra: { "app.entity.id": id } + query: sqlText }) ) - })), + }) + ), - filter: (f: FilterArgs) => { - const filter = f - .filter - type M = U extends undefined ? Encoded : Pick - return resolveNamespace.pipe(Effect.flatMap((ns) => - Effect - .sync(() => { - const q = buildWhereSQLQuery( - pgDialect, - idKey, - filter ? [{ t: "where-scope", result: filter, relation: "some" }] : [], - tableName, - defaultValues, - f.select as - | NonEmptyReadonlyArray< - string | { - key: string - subKeys: readonly string[] - } | { - key: string - computed: ComputedProjectionIrExpression - } - > - | undefined, - f.order, - f.skip, - f.limit - ) - const nsPlaceholder = pgDialect.placeholder(q.params.length + 1) - const hasWhere = q.sql.includes("WHERE") - const nsSql = hasWhere - ? q.sql.replace("WHERE", `WHERE _namespace = ${nsPlaceholder} AND`) - : q.sql.replace( - `FROM "${tableName}"`, - `FROM "${tableName}" WHERE _namespace = ${nsPlaceholder}` - ) - return { sql: nsSql, params: [...q.params, ns] } - }) - .pipe( - Effect.tap((q) => logQuery(q)), - Effect.tap((q) => Effect.annotateCurrentSpan({ "db.query.text": q.sql })), - Effect.flatMap((q) => - exec(q.sql, q.params).pipe( + find: (id) => + resolveNamespace.pipe(Effect + .flatMap((ns) => { + const sqlText = + `SELECT id, _etag, data${selectColumnsSql} FROM "${tableName}" WHERE id = $1 AND _namespace = $2` + return exec(sqlText, [id, ns]) + .pipe( Effect.map((rows) => { - if (f.select) { - return (rows as any[]).map((r) => { - const selected = parseSelectRow(r, idKey, {}) - return { - ...Struct.pick( - defaultValues as any, - f.select!.filter((_) => typeof _ === "string") as never[] - ), - ...selected - } as M - }) - } - return (rows as any[]).map((r) => parseRow(r, idKey, defaultValues) as any as M) + const row = (rows as any[])[0] + return row + ? Option.some(parseRow(row, idKey, defaultValues, activeRootLevelFieldColumns)) + : Option.none() + }), + annotateDb({ + operation: "find", + system: "postgresql", + collection: tableName, + namespace: ns, + entity: name, + query: sqlText, + extra: { "app.entity.id": id } }) ) - ), + })), + + filter: (f: FilterArgs) => { + const filter = f + .filter + type M = U extends undefined ? Encoded : Pick + return resolveNamespace.pipe(Effect.flatMap((ns) => + Effect + .sync(() => { + const q = buildWhereSQLQuery( + pgDialect, + idKey, + filter ? [{ t: "where-scope", result: filter, relation: "some" }] : [], + tableName, + defaultValues, + f.select as + | NonEmptyReadonlyArray< + string | { + key: string + subKeys: readonly string[] + } | { + key: string + computed: ComputedProjectionIrExpression + } + > + | undefined, + f.order, + f.skip, + f.limit, + undefined, + activeRootLevelFieldColumns + ) + const nsPlaceholder = pgDialect.placeholder(q.params.length + 1) + const hasWhere = q.sql.includes("WHERE") + const nsSql = hasWhere + ? q.sql.replace("WHERE", `WHERE _namespace = ${nsPlaceholder} AND`) + : q.sql.replace( + `FROM "${tableName}"`, + `FROM "${tableName}" WHERE _namespace = ${nsPlaceholder}` + ) + return { sql: nsSql, params: [...q.params, ns] } + }) + .pipe( + Effect.tap((q) => logQuery(q)), + Effect.tap((q) => Effect.annotateCurrentSpan({ "db.query.text": q.sql })), + Effect.flatMap((q) => + exec(q.sql, q.params).pipe( + Effect.map((rows) => { + if (f.select) { + return (rows as any[]).map((r) => { + const selected = parseSelectRow(r, idKey, {}, activeRootLevelFieldColumns) + return { + ...Struct.pick( + defaultValues as any, + f.select!.filter((_) => typeof _ === "string") as never[] + ), + ...selected + } as M + }) + } + return (rows as any[]).map((r) => + parseRow(r, idKey, defaultValues, activeRootLevelFieldColumns) as any as M + ) + }) + ) + ), + annotateDb({ + operation: "filter", + system: "postgresql", + collection: tableName, + namespace: ns, + entity: name + }) + ) + )) + }, + + set: (e) => + resolveNamespace.pipe(Effect.flatMap((ns) => + setInternal(e, ns).pipe( annotateDb({ - operation: "filter", + operation: "set", + system: "postgresql", + collection: tableName, + namespace: ns, + entity: name, + extra: { "app.entity.id": e[idKey] } + }) + ) + )), + + batchSet: (items) => + resolveNamespace.pipe(Effect.flatMap((ns) => + bulkSetInternal(items, ns).pipe( + annotateDb({ + operation: "batchSet", system: "postgresql", collection: tableName, namespace: ns, entity: name }) ) - )) - }, + )), - set: (e) => - resolveNamespace.pipe(Effect.flatMap((ns) => - setInternal(e, ns).pipe( - annotateDb({ - operation: "set", - system: "postgresql", - collection: tableName, - namespace: ns, - entity: name, - extra: { "app.entity.id": e[idKey] } - }) - ) - )), + bulkSet: (items) => + resolveNamespace.pipe(Effect.flatMap((ns) => + bulkSetInternal(items, ns).pipe( + annotateDb({ + operation: "bulkSet", + system: "postgresql", + collection: tableName, + namespace: ns, + entity: name + }) + ) + )), - batchSet: (items) => - resolveNamespace.pipe(Effect.flatMap((ns) => - bulkSetInternal(items, ns).pipe( - annotateDb({ - operation: "batchSet", - system: "postgresql", - collection: tableName, - namespace: ns, - entity: name - }) - ) - )), + batchRemove: (ids) => { + const placeholders = ids.map((_, i) => `$${i + 1}`).join(", ") + const nsPlaceholder = `$${ids.length + 1}` + return resolveNamespace.pipe(Effect.flatMap((ns) => { + const sqlText = + `DELETE FROM "${tableName}" WHERE id IN (${placeholders}) AND _namespace = ${nsPlaceholder}` + return exec(sqlText, [...ids, ns]) + .pipe( + Effect.asVoid, + annotateDb({ + operation: "batchRemove", + system: "postgresql", + collection: tableName, + namespace: ns, + entity: name, + query: sqlText + }) + ) + })) + }, - bulkSet: (items) => - resolveNamespace.pipe(Effect.flatMap((ns) => - bulkSetInternal(items, ns).pipe( + queryRaw: (query) => + s.all.pipe( + Effect.map(query.memory), annotateDb({ - operation: "bulkSet", + operation: "queryRaw", system: "postgresql", collection: tableName, - namespace: ns, entity: name }) ) - )), - - batchRemove: (ids) => { - const placeholders = ids.map((_, i) => `$${i + 1}`).join(", ") - const nsPlaceholder = `$${ids.length + 1}` - return resolveNamespace.pipe(Effect.flatMap((ns) => { - const sqlText = `DELETE FROM "${tableName}" WHERE id IN (${placeholders}) AND _namespace = ${nsPlaceholder}` - return exec(sqlText, [...ids, ns]) - .pipe( - Effect.asVoid, - annotateDb({ - operation: "batchRemove", - system: "postgresql", - collection: tableName, - namespace: ns, - entity: name, - query: sqlText - }) - ) - })) - }, - - queryRaw: (query) => - s.all.pipe( - Effect.map(query.memory), - annotateDb({ - operation: "queryRaw", - system: "postgresql", - collection: tableName, - entity: name - }) - ) - } + } - // Eagerly seed primary namespace on initialization - yield* seedNamespace("primary") + // Eagerly seed primary namespace on initialization + yield* seedNamespace("primary") - return s - }) + return s + }) + } } -}) +) export function PgStoreLayer(cfg: StorageConfig) { return StoreMaker diff --git a/packages/infra/src/Store/SQL/query.ts b/packages/infra/src/Store/SQL/query.ts index 77532f770..b294eb8e8 100644 --- a/packages/infra/src/Store/SQL/query.ts +++ b/packages/infra/src/Store/SQL/query.ts @@ -6,6 +6,7 @@ import type { AggregateIrExpression, ComputedProjectionIrExpression, ComputedPro import { assertUnreachable } from "effect-app/utils" import { InfraLogger } from "../../logger.js" import { isRelationCheck } from "../codeFilter.js" +import type { RootLevelFieldColumn, RootLevelFieldColumnKind } from "../rootLevelFields.js" export interface SQLDialect { readonly jsonExtract: (path: string) => string @@ -143,12 +144,199 @@ export function logQuery(q: { sql: string; params: unknown[] }) { })) } +export const quoteIdentifier = (value: string) => `"${value.replaceAll("\"", "\"\"")}"` + +export const projectedColumnFieldExpr = (column: RootLevelFieldColumn) => quoteIdentifier(column.columnName) + +export const projectedColumnSelectExpr = ( + dialect: SQLDialect, + column: RootLevelFieldColumn +) => + column.kind === "boolean" && dialect.jsonColumnType === "JSON" + ? `CASE ${projectedColumnFieldExpr(column)} WHEN 1 THEN 'true' WHEN 0 THEN 'false' ELSE 'null' END` + : projectedColumnFieldExpr(column) + +export const projectedColumnSqlType = ( + dialect: SQLDialect, + kind: RootLevelFieldColumnKind +) => { + switch (kind) { + case "string": + return "TEXT" + case "number": + return dialect.jsonColumnType === "JSON" ? "REAL" : "DOUBLE PRECISION" + case "boolean": + return dialect.jsonColumnType === "JSON" ? "INTEGER" : "BOOLEAN" + case "json": + return dialect.jsonColumnType + default: + return assertUnreachable(kind) + } +} + +export const normalizeProjectedColumnValue = ( + column: RootLevelFieldColumn, + value: unknown +) => { + if (column.kind === "json") { + if (typeof value === "string") { + try { + return JSON.parse(value) + } catch { + return value + } + } + return value + } + if (column.kind === "boolean") { + if (typeof value === "number") { + return value !== 0 + } + if (typeof value === "string") { + if (value === "true") return true + if (value === "false") return false + } + } + return value +} + const dottedToJsonPath = (path: string) => path .split(".") .filter((p) => p !== "-1") .join(".") +const sqliteJsonPathArg = (path: string) => `'${(path.length > 0 ? `$.${path}` : "$").replaceAll("'", "''")}'` + +const extractJsonFromSourceExpr = ( + dialect: SQLDialect, + sourceExpr: string, + path: string, + asJson: boolean +) => { + if (dialect.jsonColumnType === "JSON") { + const pathArg = sqliteJsonPathArg(path) + if (!asJson) { + return `json_extract(${sourceExpr}, ${pathArg})` + } + return `CASE WHEN json_type(${sourceExpr}, ${pathArg}) IS NULL THEN NULL WHEN json_type(${sourceExpr}, ${pathArg}) = 'true' THEN 'true' WHEN json_type(${sourceExpr}, ${pathArg}) = 'false' THEN 'false' ELSE json_quote(json_extract(${sourceExpr}, ${pathArg})) END` + } + + const parts: ReadonlyArray = path.length > 0 ? path.split(".") : [] + if (parts.length === 0) { + return asJson ? `(${sourceExpr})` : `(${sourceExpr})#>> '{}'` + } + if (!asJson) { + const last = parts[parts.length - 1]! + const parents = parts.slice(0, -1) + return `(${sourceExpr})${parents.map((part) => `->'${part}'`).join("")}->>'${last}'` + } + return `(${sourceExpr})${parts.map((part) => `->'${part}'`).join("")}` +} + +const arrayLengthFromSourceExpr = ( + dialect: SQLDialect, + sourceExpr: string, + path: string +) => + dialect.jsonColumnType === "JSON" + ? `json_array_length(${sourceExpr}, ${sqliteJsonPathArg(path)})` + : `jsonb_array_length(${extractJsonFromSourceExpr(dialect, sourceExpr, path, true)})` + +const jsonEachFromSourceExpr = ( + dialect: SQLDialect, + sourceExpr: string, + path: string, + alias: string +) => + dialect.jsonColumnType === "JSON" + ? `json_each(${sourceExpr}, ${sqliteJsonPathArg(path)}) AS ${alias}` + : `jsonb_array_elements(${extractJsonFromSourceExpr(dialect, sourceExpr, path, true)}) AS ${alias}` + +const jsonArrayContainsFromSourceExpr = ( + dialect: SQLDialect, + sourceExpr: string, + path: string, + valPlaceholder: string +) => + dialect.jsonColumnType === "JSON" + ? `EXISTS(SELECT 1 FROM json_each(${sourceExpr}, ${sqliteJsonPathArg(path)}) WHERE value = ${valPlaceholder})` + : `${extractJsonFromSourceExpr(dialect, sourceExpr, path, true)} @> ${valPlaceholder}::jsonb` + +const jsonArrayNotContainsFromSourceExpr = ( + dialect: SQLDialect, + sourceExpr: string, + path: string, + valPlaceholder: string +) => + dialect.jsonColumnType === "JSON" + ? `NOT EXISTS(SELECT 1 FROM json_each(${sourceExpr}, ${sqliteJsonPathArg(path)}) WHERE value = ${valPlaceholder})` + : `NOT (${extractJsonFromSourceExpr(dialect, sourceExpr, path, true)} @> ${valPlaceholder}::jsonb)` + +const jsonArrayContainsAnyFromSourceExpr = ( + dialect: SQLDialect, + sourceExpr: string, + path: string, + valPlaceholders: readonly string[] +) => + dialect.jsonColumnType === "JSON" + ? `EXISTS(SELECT 1 FROM json_each(${sourceExpr}, ${sqliteJsonPathArg(path)}) WHERE value IN (${ + valPlaceholders.join(", ") + }))` + : `(${ + valPlaceholders.map((v) => `${extractJsonFromSourceExpr(dialect, sourceExpr, path, true)} @> ${v}::jsonb`).join( + " OR " + ) + })` + +const jsonArrayNotContainsAnyFromSourceExpr = ( + dialect: SQLDialect, + sourceExpr: string, + path: string, + valPlaceholders: readonly string[] +) => + dialect.jsonColumnType === "JSON" + ? `NOT EXISTS(SELECT 1 FROM json_each(${sourceExpr}, ${sqliteJsonPathArg(path)}) WHERE value IN (${ + valPlaceholders.join(", ") + }))` + : `NOT (${ + valPlaceholders.map((v) => `${extractJsonFromSourceExpr(dialect, sourceExpr, path, true)} @> ${v}::jsonb`).join( + " OR " + ) + })` + +const jsonArrayContainsAllFromSourceExpr = ( + dialect: SQLDialect, + sourceExpr: string, + path: string, + valPlaceholders: readonly string[] +) => + dialect.jsonColumnType === "JSON" + ? valPlaceholders + .map((v) => `EXISTS(SELECT 1 FROM json_each(${sourceExpr}, ${sqliteJsonPathArg(path)}) WHERE value = ${v})`) + .join(" AND ") + : valPlaceholders.map((v) => `${extractJsonFromSourceExpr(dialect, sourceExpr, path, true)} @> ${v}::jsonb`).join( + " AND " + ) + +const jsonArrayNotContainsAllFromSourceExpr = ( + dialect: SQLDialect, + sourceExpr: string, + path: string, + valPlaceholders: readonly string[] +) => + dialect.jsonColumnType === "JSON" + ? `NOT (${ + valPlaceholders + .map((v) => `EXISTS(SELECT 1 FROM json_each(${sourceExpr}, ${sqliteJsonPathArg(path)}) WHERE value = ${v})`) + .join(" AND ") + })` + : `NOT (${ + valPlaceholders.map((v) => `${extractJsonFromSourceExpr(dialect, sourceExpr, path, true)} @> ${v}::jsonb`).join( + " AND " + ) + })` + const sqlStringLiteral = (value: string) => `'${value.replaceAll("'", "''")}'` export function buildWhereSQLQuery( @@ -175,16 +363,39 @@ export function buildWhereSQLQuery( order?: NonEmptyReadonlyArray<{ key: string; direction: "ASC" | "DESC" }>, skip?: number, limit?: number, - namespace?: string + namespace?: string, + rootLevelFieldColumns: readonly RootLevelFieldColumn[] = [] ) { const params: unknown[] = [] let paramIndex = 1 + const projectedColumns = new Map(rootLevelFieldColumns.map((column) => [column.key, column] as const)) + const projectedJsonColumnForPath = (path: string, relation?: string) => { + if (relation) { + return + } + const [topKey, ...subPathParts] = path.split(".") + const projected = projectedColumns.get(topKey) + if (!projected || projected.kind !== "json") { + return + } + return { + sourceExpr: quoteIdentifier(projected.columnName), + subPath: subPathParts.join(".") + } + } const addParam = (value: unknown): string => { params.push(dialect.serializeScalar(value)) return dialect.placeholder(paramIndex++) } + const addJsonParam = (value: unknown): string => { + params.push(JSON.stringify(value)) + return dialect.placeholder(paramIndex++) + } + + const isRootLevelPath = (path: string, relation?: string) => !relation && !path.includes(".") + const fieldExpr = (path: string, relation?: string): string => { if (path === idKey || path === "id") return "id" if (relation && path.includes(".-1.")) { @@ -197,8 +408,28 @@ export function buildWhereSQLQuery( } if (path.endsWith(".length")) { const arrPath = dottedToJsonPath(path.slice(0, -".length".length)) + const projectedJson = projectedJsonColumnForPath(arrPath, relation) + if (projectedJson) { + return arrayLengthFromSourceExpr(dialect, projectedJson.sourceExpr, projectedJson.subPath) + } return dialect.arrayLength(arrPath) } + if (isRootLevelPath(path, relation)) { + const projected = projectedColumns.get(path) + if (projected) { + const expr = projectedColumnFieldExpr(projected) + if (path in defaultValues) { + return `COALESCE(${expr}, ${ + projected.kind === "json" ? addJsonParam(defaultValues[path]) : addParam(defaultValues[path]) + })` + } + return expr + } + } + const projectedJson = projectedJsonColumnForPath(path, relation) + if (projectedJson) { + return extractJsonFromSourceExpr(dialect, projectedJson.sourceExpr, projectedJson.subPath, false) + } const jsonPath = dottedToJsonPath(path) const expr = dialect.jsonExtract(jsonPath) const topKey = path.split(".")[0] @@ -211,6 +442,14 @@ export function buildWhereSQLQuery( const statement = (x: FilterR, relation?: string): string => { const resolvedPath = x.path === idKey ? "id" : x.path const k = fieldExpr(resolvedPath, relation) + const projectedRoot = isRootLevelPath(resolvedPath, relation) + ? projectedColumns.get(resolvedPath) + : undefined + const isProjectedJsonRoot = projectedRoot?.kind === "json" + const jsonComparisonParam = (value: unknown) => { + const placeholder = addJsonParam(value) + return dialect.jsonColumnType === "JSON" ? placeholder : `${placeholder}::jsonb` + } switch (x.op) { case "in": { @@ -219,7 +458,7 @@ export function buildWhereSQLQuery( const nonNullVals = vals.filter((v) => v != null) const parts: string[] = [] if (nonNullVals.length > 0) { - const placeholders = nonNullVals.map((v) => addParam(v)) + const placeholders = nonNullVals.map((v) => isProjectedJsonRoot ? jsonComparisonParam(v) : addParam(v)) parts.push(`${k} IN (${placeholders.join(", ")})`) } if (hasNull) parts.push(`${k} IS NULL`) @@ -231,7 +470,7 @@ export function buildWhereSQLQuery( const nonNullVals = vals.filter((v) => v != null) const parts: string[] = [] if (nonNullVals.length > 0) { - const placeholders = nonNullVals.map((v) => addParam(v)) + const placeholders = nonNullVals.map((v) => isProjectedJsonRoot ? jsonComparisonParam(v) : addParam(v)) parts.push(`${k} NOT IN (${placeholders.join(", ")})`) } if (hasNull) parts.push(`${k} IS NOT NULL`) @@ -241,38 +480,66 @@ export function buildWhereSQLQuery( case "includes": { const arrPath = dottedToJsonPath(resolvedPath) const v = addParam(x.value) - return dialect.jsonArrayContains(arrPath, v) + const projectedJson = projectedJsonColumnForPath(arrPath, relation) + return projectedJson + ? jsonArrayContainsFromSourceExpr(dialect, projectedJson.sourceExpr, projectedJson.subPath, v) + : dialect.jsonArrayContains(arrPath, v) } case "notIncludes": { const arrPath = dottedToJsonPath(resolvedPath) const v = addParam(x.value) - return dialect.jsonArrayNotContains(arrPath, v) + const projectedJson = projectedJsonColumnForPath(arrPath, relation) + return projectedJson + ? jsonArrayNotContainsFromSourceExpr(dialect, projectedJson.sourceExpr, projectedJson.subPath, v) + : dialect.jsonArrayNotContains(arrPath, v) } case "includes-any": { const arrPath = dottedToJsonPath(resolvedPath) const vals = x.value as unknown as readonly unknown[] const placeholders = vals.map((v) => addParam(dialect.serializeJsonValue(v))) - return dialect.jsonArrayContainsAny(arrPath, placeholders) + const projectedJson = projectedJsonColumnForPath(arrPath, relation) + return projectedJson + ? jsonArrayContainsAnyFromSourceExpr(dialect, projectedJson.sourceExpr, projectedJson.subPath, placeholders) + : dialect.jsonArrayContainsAny(arrPath, placeholders) } case "notIncludes-any": { const arrPath = dottedToJsonPath(resolvedPath) const vals = x.value as unknown as readonly unknown[] const placeholders = vals.map((v) => addParam(dialect.serializeJsonValue(v))) - return dialect.jsonArrayNotContainsAny(arrPath, placeholders) + const projectedJson = projectedJsonColumnForPath(arrPath, relation) + return projectedJson + ? jsonArrayNotContainsAnyFromSourceExpr( + dialect, + projectedJson.sourceExpr, + projectedJson.subPath, + placeholders + ) + : dialect.jsonArrayNotContainsAny(arrPath, placeholders) } case "includes-all": { const arrPath = dottedToJsonPath(resolvedPath) const vals = x.value as unknown as readonly unknown[] const placeholders = vals.map((v) => addParam(dialect.serializeJsonValue(v))) - return dialect.jsonArrayContainsAll(arrPath, placeholders) + const projectedJson = projectedJsonColumnForPath(arrPath, relation) + return projectedJson + ? jsonArrayContainsAllFromSourceExpr(dialect, projectedJson.sourceExpr, projectedJson.subPath, placeholders) + : dialect.jsonArrayContainsAll(arrPath, placeholders) } case "notIncludes-all": { const arrPath = dottedToJsonPath(resolvedPath) const vals = x.value as unknown as readonly unknown[] const placeholders = vals.map((v) => addParam(dialect.serializeJsonValue(v))) - return dialect.jsonArrayNotContainsAll(arrPath, placeholders) + const projectedJson = projectedJsonColumnForPath(arrPath, relation) + return projectedJson + ? jsonArrayNotContainsAllFromSourceExpr( + dialect, + projectedJson.sourceExpr, + projectedJson.subPath, + placeholders + ) + : dialect.jsonArrayNotContainsAll(arrPath, placeholders) } case "contains": { @@ -318,13 +585,13 @@ export function buildWhereSQLQuery( } case "neq": { if (x.value === null) return `${k} IS NOT NULL` - const v = addParam(x.value) + const v = isProjectedJsonRoot ? jsonComparisonParam(x.value) : addParam(x.value) return `${k} <> ${v}` } case undefined: case "eq": { if (x.value === null) return `${k} IS NULL` - const v = addParam(x.value) + const v = isProjectedJsonRoot ? jsonComparisonParam(x.value) : addParam(x.value) return `${k} = ${v}` } default: @@ -336,7 +603,10 @@ export function buildWhereSQLQuery( // Optimize tautological/contradictory conditions if (every && inner === "1=1") return "1=1" if (!every && inner === "1=0") return "1=0" - const from = dialect.jsonEachFrom(rel, `_${rel}`) + const projectedJson = projectedJsonColumnForPath(rel) + const from = projectedJson + ? jsonEachFromSourceExpr(dialect, projectedJson.sourceExpr, projectedJson.subPath, `_${rel}`) + : dialect.jsonEachFrom(rel, `_${rel}`) // ∀x.P(x) ≡ ¬∃x.¬P(x), i.e. NOT EXISTS(... WHERE NOT P) return every ? `NOT EXISTS(SELECT 1 FROM ${from} WHERE NOT (${inner}))` @@ -400,7 +670,10 @@ export function buildWhereSQLQuery( const computedSelectExpr = (key: string, computed: ComputedProjectionIrExpression): string => { const relationPath = dottedToJsonPath(computed.path) const relationAlias = `_${computed.path}` - const relationFrom = dialect.jsonEachFrom(relationPath, relationAlias) + const projectedJson = projectedJsonColumnForPath(relationPath) + const relationFrom = projectedJson + ? jsonEachFromSourceExpr(dialect, projectedJson.sourceExpr, projectedJson.subPath, relationAlias) + : dialect.jsonEachFrom(relationPath, relationAlias) const toNumber = (expr: string) => dialect.jsonColumnType === "JSON" ? `CAST(${expr} AS REAL)` : `(${expr})::numeric` const compileExpr = (expression: ComputedProjectionMathIrExpression): string => { @@ -479,7 +752,11 @@ export function buildWhereSQLQuery( } case "relation-length": { const arrPath = dottedToJsonPath(computed.path) - return `${dialect.arrayLength(arrPath)} AS "${key}"` + const projectedJson = projectedJsonColumnForPath(arrPath) + const expr = projectedJson + ? arrayLengthFromSourceExpr(dialect, projectedJson.sourceExpr, projectedJson.subPath) + : dialect.arrayLength(arrPath) + return `${expr} AS "${key}"` } case "relation-collect-fields": { const branches = computed.fields.map((field) => { @@ -528,11 +805,20 @@ export function buildWhereSQLQuery( : false const getSelectExpr = (): string => { - if (!select) return "id, _etag, data" + if (!select) { + const projectedSelect = rootLevelFieldColumns.length > 0 + ? `, ${rootLevelFieldColumns.map((column) => quoteIdentifier(column.columnName)).join(", ")}` + : "" + return `id, _etag, data${projectedSelect}` + } const fields = select.map((s) => { if (typeof s === "string") { if (s === idKey || s === "id") return `id` if (s === "_etag") return `_etag` + const projected = projectedColumns.get(s) + if (projected) { + return `${projectedColumnSelectExpr(dialect, projected)} AS "${s}"` + } return `${dialect.jsonExtractJson(s)} AS "${s}"` } if ("computed" in s) { diff --git a/packages/infra/src/Store/rootLevelFields.ts b/packages/infra/src/Store/rootLevelFields.ts new file mode 100644 index 000000000..7d1868dc4 --- /dev/null +++ b/packages/infra/src/Store/rootLevelFields.ts @@ -0,0 +1,85 @@ +import type * as S from "effect-app/Schema" +import * as SchemaAST from "effect/SchemaAST" + +export type RootLevelFieldColumnKind = "string" | "number" | "boolean" | "json" + +export interface RootLevelFieldColumn { + readonly key: string + readonly columnName: string + readonly kind: RootLevelFieldColumnKind +} + +const walkTransformation = (ast: SchemaAST.AST): SchemaAST.AST => { + if (ast._tag === "Declaration" && ast.typeParameters.length > 0) { + return walkTransformation(ast.typeParameters[0]!) + } + return ast +} + +const getColumnKind = (ast: SchemaAST.AST): RootLevelFieldColumnKind => { + const encoded = walkTransformation(SchemaAST.toEncoded(ast)) + switch (encoded._tag) { + case "String": + return "string" + case "Number": + return "number" + case "Boolean": + return "boolean" + case "Literal": + switch (typeof encoded.literal) { + case "string": + return "string" + case "number": + return "number" + case "boolean": + return "boolean" + default: + return "json" + } + case "Union": { + const scalarKinds = encoded + .types + .flatMap((type) => { + const kind = getColumnKind(type) + return kind === "json" ? [] : [kind] + }) + const distinctKinds = [...new Set(scalarKinds)] + return distinctKinds.length === 1 ? distinctKinds[0]! : "json" + } + default: + return "json" + } +} + +const makeColumnName = (key: string) => `__root_${key}` + +export const makeRootLevelFieldColumns = ( + schema: S.Schema, + idKey: PropertyKey +): readonly RootLevelFieldColumn[] => { + const encoded = walkTransformation(SchemaAST.toEncoded(schema.ast)) + if (!SchemaAST.isObjects(encoded)) { + return [] + } + + return encoded.propertySignatures.flatMap((property) => { + const key = String(property.name) + if (key === String(idKey) || key === "id" || key === "_etag") { + return [] + } + return [{ key, kind: getColumnKind(property.type), columnName: makeColumnName(key) }] + }) +} + +export const omitRootLevelFieldColumnsFromData = >( + data: A, + rootLevelFieldColumns: readonly RootLevelFieldColumn[] +) => { + if (rootLevelFieldColumns.length === 0) { + return data + } + const rootLevelFieldKeys = new Set(rootLevelFieldColumns.map((column) => column.key)) + return Object.fromEntries( + Object.entries(data).filter(([key]) => !rootLevelFieldKeys.has(key)) + ) +} diff --git a/packages/infra/test/sql-store.test.ts b/packages/infra/test/sql-store.test.ts index 433f7373e..663f377aa 100644 --- a/packages/infra/test/sql-store.test.ts +++ b/packages/infra/test/sql-store.test.ts @@ -1,7 +1,9 @@ /* eslint-disable @typescript-eslint/no-explicit-any */ import type Sqlite from "better-sqlite3" import BetterSqlite from "better-sqlite3" +import * as S from "effect-app/Schema" import { describe, expect, it } from "vitest" +import { makeRootLevelFieldColumns, type RootLevelFieldColumn } from "../src/Store/rootLevelFields.js" import { parseRow } from "../src/Store/SQL.js" import { buildWhereSQLQuery, pgDialect, sqliteDialect } from "../src/Store/SQL/query.js" import { makeETag } from "../src/Store/utils.js" @@ -9,6 +11,42 @@ import { makeETag } from "../src/Store/utils.js" const query = (db: Sqlite.Database, sql: string, params: unknown[] = []) => db.prepare(sql).all(...params as any[]) as any[] +const projectedRootLevelFields: readonly RootLevelFieldColumn[] = [ + { key: "name", columnName: "__root_name", kind: "string" }, + { key: "age", columnName: "__root_age", kind: "number" }, + { key: "flag", columnName: "__root_flag", kind: "boolean" } +] + +const projectedJsonRootLevelFields: readonly RootLevelFieldColumn[] = [ + { key: "meta", columnName: "__root_meta", kind: "json" }, + { key: "items", columnName: "__root_items", kind: "json" } +] + +describe("root-level projected field metadata", () => { + it("derives root-level projected columns for scalar and complex encoded fields", () => { + const schema = S.Struct({ + id: S.String, + name: S.String, + age: S.Number, + active: S.Boolean, + createdAt: S.Date, + status: S.NullOr(S.String), + meta: S.Struct({ city: S.String }), + tags: S.Array(S.String) + }) + + expect(makeRootLevelFieldColumns(schema, "id")).toEqual([ + { key: "name", columnName: "__root_name", kind: "string" }, + { key: "age", columnName: "__root_age", kind: "number" }, + { key: "active", columnName: "__root_active", kind: "boolean" }, + { key: "createdAt", columnName: "__root_createdAt", kind: "string" }, + { key: "status", columnName: "__root_status", kind: "string" }, + { key: "meta", columnName: "__root_meta", kind: "json" }, + { key: "tags", columnName: "__root_tags", kind: "json" } + ]) + }) +}) + // --- Query builder unit tests --- describe("SQL query builder (SQLite dialect)", () => { @@ -48,6 +86,40 @@ describe("SQL query builder (SQLite dialect)", () => { expect(result.params).toContain(18) }) + it("uses projected root columns directly for root-level filters", () => { + const result = buildWhereSQLQuery( + sqliteDialect, + "id", + [{ t: "where", path: "age", op: "gt", value: 18 as any }], + "users", + {}, + undefined, + undefined, + undefined, + undefined, + undefined, + projectedRootLevelFields + ) + expect(result.sql).toContain(`"__root_age" > ?`) + }) + + it("uses projected JSON root columns for nested filters", () => { + const result = buildWhereSQLQuery( + sqliteDialect, + "id", + [{ t: "where", path: "meta.city", op: "eq", value: "NYC" }], + "users", + {}, + undefined, + undefined, + undefined, + undefined, + undefined, + projectedJsonRootLevelFields + ) + expect(result.sql).toContain(`json_extract("__root_meta", '$.city')`) + }) + it("where or", () => { const result = buildWhereSQLQuery( sqliteDialect, @@ -250,6 +322,31 @@ describe("SQL query builder (SQLite dialect)", () => { expect(result.params).toContain("%picked%") }) + it("computed relation count uses projected JSON root arrays directly", () => { + const result = buildWhereSQLQuery( + sqliteDialect, + "id", + [], + "users", + {}, + [{ + key: "pickedCount", + computed: { + _tag: "relation-count", + path: "items", + filter: [{ t: "where", path: "items.-1.description", op: "contains", value: "picked" }] + } + }], + undefined, + undefined, + undefined, + undefined, + projectedJsonRootLevelFields + ) + expect(result.sql).toContain(`json_each("__root_items", '$')`) + expect(result.sql).toContain(`AS _items`) + }) + it("computed relation any projection (sqlite bool encoding)", () => { const result = buildWhereSQLQuery( sqliteDialect, @@ -711,6 +808,40 @@ describe("SQL query builder (PostgreSQL dialect)", () => { ) expect(result.sql).toContain(`COALESCE(jsonb_agg(DISTINCT _items->>'articleId'), '[]'::jsonb)`) }) + + it("uses projected root columns directly for pg filters", () => { + const result = buildWhereSQLQuery( + pgDialect, + "id", + [{ t: "where", path: "flag", op: "eq", value: true as any }], + "users", + {}, + undefined, + undefined, + undefined, + undefined, + undefined, + projectedRootLevelFields + ) + expect(result.sql).toContain(`"__root_flag" = $1`) + }) + + it("uses projected JSON root columns for pg nested filters", () => { + const result = buildWhereSQLQuery( + pgDialect, + "id", + [{ t: "where", path: "meta.city", op: "eq", value: "NYC" }], + "users", + {}, + undefined, + undefined, + undefined, + undefined, + undefined, + projectedJsonRootLevelFields + ) + expect(result.sql).toContain(`("__root_meta")->>'city' = $1`) + }) }) // --- Integration tests with in-memory SQLite (direct, no Effect SQL client) --- @@ -811,6 +942,93 @@ describe("SQL Store (SQLite integration)", () => { expect(query(db, q3.sql, q3.params).length).toBe(2) })) + it("projected root columns do not fall back to legacy JSON data", () => + withDb((db) => { + db.exec( + `CREATE TABLE IF NOT EXISTS "test_projected" (id TEXT PRIMARY KEY, _etag TEXT, data JSON NOT NULL, "__root_name" TEXT, "__root_age" REAL, "__root_flag" INTEGER)` + ) + db + .prepare(`INSERT INTO "test_projected" (id, _etag, data) VALUES (?, ?, ?)`) + .run("1", "etag1", JSON.stringify({ name: "Alice", age: 30, flag: true })) + db + .prepare( + `INSERT INTO "test_projected" (id, _etag, data, "__root_name", "__root_age", "__root_flag") VALUES (?, ?, ?, ?, ?, ?)` + ) + .run("2", "etag2", JSON.stringify({ name: "Bob", age: 10, flag: false }), "Bob", 10, 0) + + const q = buildWhereSQLQuery( + sqliteDialect, + "id", + [{ t: "where", path: "age", op: "gt", value: 20 as any }], + "test_projected", + {}, + undefined, + undefined, + undefined, + undefined, + undefined, + projectedRootLevelFields + ) + const rows = query(db, q.sql, q.params) + expect(rows).toHaveLength(0) + })) + + it("projected root JSON columns can replace data storage for nested fields", () => + withDb((db) => { + db.exec( + `CREATE TABLE IF NOT EXISTS "test_projected_json" (id TEXT PRIMARY KEY, _etag TEXT, data JSON NOT NULL, "__root_meta" JSON, "__root_items" JSON)` + ) + db + .prepare( + `INSERT INTO "test_projected_json" (id, _etag, data, "__root_meta", "__root_items") VALUES (?, ?, ?, ?, ?)` + ) + .run( + "1", + "etag1", + JSON.stringify({ untouched: true }), + JSON.stringify({ city: "NYC", zip: "10001" }), + JSON.stringify([{ description: "picked" }, { description: "open" }]) + ) + + const nested = buildWhereSQLQuery( + sqliteDialect, + "id", + [{ t: "where", path: "meta.city", op: "eq", value: "NYC" }], + "test_projected_json", + {}, + undefined, + undefined, + undefined, + undefined, + undefined, + projectedJsonRootLevelFields + ) + expect(query(db, nested.sql, nested.params)).toHaveLength(1) + + const relation = buildWhereSQLQuery( + sqliteDialect, + "id", + [], + "test_projected_json", + {}, + [{ + key: "pickedCount", + computed: { + _tag: "relation-count", + path: "items", + filter: [{ t: "where", path: "items.-1.description", op: "contains", value: "picked" }] + } + }], + undefined, + undefined, + undefined, + undefined, + projectedJsonRootLevelFields + ) + const [row] = query(db, relation.sql, relation.params) + expect((row as any).pickedCount).toBe(1) + })) + it("queries work when data does not contain id", () => withDb((db) => { db.exec( @@ -1570,11 +1788,16 @@ describe("sqliteDialect.jsonExtractJson boolean round-trip", () => { describe("toRow strips _etag and id from data", () => { // Replicate the toRow logic from SQL.ts to test in isolation - const toRow = (e: any, idKey: IdKey) => { + const toRow = ( + e: any, + idKey: IdKey, + rootLevelFieldColumns: readonly RootLevelFieldColumn[] = [] + ) => { const newE = makeETag(e) const id = newE[idKey] as string const { _etag, [idKey]: _id, ...rest } = newE as any - const data = JSON.stringify(rest) + const projectedKeys = new Set(rootLevelFieldColumns.map((column) => column.key)) + const data = JSON.stringify(Object.fromEntries(Object.entries(rest).filter(([key]) => !projectedKeys.has(key)))) return { id, _etag: newE._etag!, data, item: newE } } @@ -1622,6 +1845,21 @@ describe("toRow strips _etag and id from data", () => { expect(parsed).not.toHaveProperty("id") expect(parsed).not.toHaveProperty("_etag") }) + + it("omits projected root fields from data", () => { + const row = toRow( + { id: "1", _etag: undefined, name: "Alice", meta: { city: "NYC" }, untouched: true }, + "id", + [ + { key: "name", columnName: "__root_name", kind: "string" }, + { key: "meta", columnName: "__root_meta", kind: "json" } + ] + ) + const parsed = JSON.parse(row.data) as any + expect(parsed).not.toHaveProperty("name") + expect(parsed).not.toHaveProperty("meta") + expect(parsed.untouched).toBe(true) + }) }) describe("parseRow reconstructs full object from row", () => { @@ -1708,4 +1946,69 @@ describe("parseRow reconstructs full object from row", () => { expect(reconstructed.tags).toEqual(["admin"]) expect(reconstructed._etag).toBe(newE._etag) }) + + it("prefers projected root-level columns when available", () => { + const result: any = parseRow( + { + id: "1", + _etag: "e1", + data: JSON.stringify({ name: "stale", flag: false }), + __root_name: "Alice", + __root_flag: 1 + }, + "id", + {}, + projectedRootLevelFields + ) + expect(result.name).toBe("Alice") + expect(result.flag).toBe(true) + }) + + it("does not fall back to data for projected root-level columns", () => { + const result: any = parseRow( + { + id: "1", + _etag: "e1", + data: JSON.stringify({ name: "legacy", flag: true }) + }, + "id", + {}, + projectedRootLevelFields + ) + expect(result.name).toBeUndefined() + expect(result.flag).toBeUndefined() + }) + + it("reconstructs projected JSON root-level columns when available", () => { + const result: any = parseRow( + { + id: "1", + _etag: "e1", + data: JSON.stringify({ untouched: true }), + __root_meta: JSON.stringify({ city: "NYC" }), + __root_items: JSON.stringify([{ description: "picked" }]) + }, + "id", + {}, + projectedJsonRootLevelFields + ) + expect(result.meta).toEqual({ city: "NYC" }) + expect(result.items).toEqual([{ description: "picked" }]) + expect(result.untouched).toBe(true) + }) + + it("does not fall back to data for projected JSON root-level columns", () => { + const result: any = parseRow( + { + id: "1", + _etag: "e1", + data: JSON.stringify({ meta: { city: "legacy" }, items: [{ description: "legacy" }] }) + }, + "id", + {}, + projectedJsonRootLevelFields + ) + expect(result.meta).toBeUndefined() + expect(result.items).toBeUndefined() + }) })