From 4eec82d9e1ddc1a26acc063bf76de8812b90a040 Mon Sep 17 00:00:00 2001 From: Antony Rizzitelli Date: Tue, 15 Sep 2026 16:40:27 +0000 Subject: [PATCH 1/4] feat(query): add standalone staged-query entry --- package.json | 8 + scripts/test-staged-query.cjs | 125 +++++ src/antelope.test.ts | 2 +- src/atomic.ts | 204 ++------- src/common.ts | 136 +----- src/datum.ts | 103 +---- src/query.ts | 76 ++-- src/schema.ts | 169 +------ src/selection.ts | 191 +------- src/staged-query/atomic.ts | 162 +++++++ src/staged-query/common.ts | 139 ++++++ src/staged-query/datum.ts | 102 +++++ src/staged-query/index.ts | 18 + src/staged-query/query.ts | 3 + src/staged-query/schema.ts | 163 +++++++ src/staged-query/selection.ts | 190 ++++++++ src/staged-query/stream.ts | 324 +++++++++++++ src/staged-query/valueproxy.ts | 804 ++++++++++++++++++++++++++++++++ src/stream.ts | 325 +------------ src/valueproxy.ts | 805 +-------------------------------- 20 files changed, 2135 insertions(+), 1914 deletions(-) create mode 100644 scripts/test-staged-query.cjs create mode 100644 src/staged-query/atomic.ts create mode 100644 src/staged-query/common.ts create mode 100644 src/staged-query/datum.ts create mode 100644 src/staged-query/index.ts create mode 100644 src/staged-query/query.ts create mode 100644 src/staged-query/schema.ts create mode 100644 src/staged-query/selection.ts create mode 100644 src/staged-query/stream.ts create mode 100644 src/staged-query/valueproxy.ts diff --git a/package.json b/package.json index 00660ec..dab4865 100644 --- a/package.json +++ b/package.json @@ -42,6 +42,9 @@ "stream": [ "dist/stream.d.ts" ], + "staged-query": [ + "dist/staged-query/index.d.ts" + ], "valueproxy": [ "dist/valueproxy.d.ts" ] @@ -80,6 +83,10 @@ "types": "./dist/stream.d.ts", "default": "./dist/stream.js" }, + "./staged-query": { + "types": "./dist/staged-query/index.d.ts", + "default": "./dist/staged-query/index.js" + }, "./valueproxy": { "types": "./dist/valueproxy.d.ts", "default": "./dist/valueproxy.js" @@ -98,6 +105,7 @@ "prepack": "pnpm run build", "release": "pnpm run lint && pnpm run prepack && release-it", "test": "pnpm run build && ajs module test .", + "test:staged-query": "pnpm run build && node scripts/test-staged-query.cjs", "format:check": "oxfmt --check .", "knip": "knip" }, diff --git a/scripts/test-staged-query.cjs b/scripts/test-staged-query.cjs new file mode 100644 index 0000000..d5a99d2 --- /dev/null +++ b/scripts/test-staged-query.cjs @@ -0,0 +1,125 @@ +const assert = require("node:assert/strict"); +const fs = require("node:fs"); +const Module = require("node:module"); +const path = require("node:path"); + +const forbiddenImports = ["@antelopejs/interface-core", "node:async_hooks"]; +const originalLoad = Module._load; + +Module._load = function (request, parent, isMain) { + assert.equal( + forbiddenImports.includes(request), + false, + `staged query imported ${request}`, + ); + return originalLoad.call(this, request, parent, isMain); +}; + +const staged = require("../dist/staged-query"); +Module._load = originalLoad; + +const schema = new staged.Schema("parity", { + posts: { fields: {}, indexes: {} }, +}); +const query = schema + .instance("default") + .table("posts") + .filter((post) => post.key("author").eq("Alice")) + .orderBy("views", "desc") + .slice(0, 10); + +assert.deepEqual(query.build(), [ + { stage: "schema", options: { id: "parity" }, args: [] }, + { stage: "instance", options: { id: "default" }, args: [] }, + { stage: "table", options: { id: "posts" }, args: [] }, + { + stage: "filter", + options: undefined, + args: [ + { + stage: "func", + args: [ + [0], + new staged.ValueProxy({ + stage: "arg", + options: undefined, + args: [0], + }) + .key("author") + .eq("Alice"), + ], + }, + ], + }, + { + stage: "orderBy", + options: { index: "views", direction: "desc" }, + args: [], + }, + { stage: "slice", options: undefined, args: [0, 10] }, +]); + +for (const method of ["run", "then", "cursor"]) { + assert.equal(method in query, false, `staged query exposes ${method}`); +} +assert.equal(Symbol.asyncIterator in query, false); + +const atomicQuery = schema + .instance("default") + .table("posts") + .atomicMutation("post-000", { + type: "deleteIfEqual", + field: "author", + expectedValue: "Nobody", + }); +assert.equal(atomicQuery instanceof staged.AtomicMutationQuery, true); +assert.equal("run" in atomicQuery, false); + +const distDirectory = path.join(__dirname, "../dist/staged-query"); +for (const file of fs.readdirSync(distDirectory)) { + if (!file.endsWith(".js")) continue; + const source = fs.readFileSync(path.join(distDirectory, file), "utf8"); + for (const forbidden of forbiddenImports) { + assert.equal( + source.includes(forbidden), + false, + `${file} contains ${forbidden}`, + ); + } +} + +const executable = require("../dist"); +const executableTable = new executable.Schema("runtime", { + records: { fields: {}, indexes: {} }, +}) + .instance() + .table("records"); +const executableInsert = executableTable.insert({}); + +assert.equal(executableTable instanceof executable.Table, true); +assert.equal(executableTable instanceof executable.Selection, true); +assert.equal(executableTable instanceof executable.Stream, true); +assert.equal(executableTable instanceof executable.Query, true); +assert.equal(typeof executableTable.run, "function"); +assert.equal(typeof executableTable.atomicMutation, "function"); +assert.equal(executableInsert instanceof executable.Query, true); +assert.equal(typeof executableInsert.run, "function"); + +const originalRun = executable.Query.prototype.run; +executable.Query.prototype.run = async () => "not-applied"; +const executableAtomic = executableTable.atomicMutation("record-1", { + type: "deleteIfEqual", + field: "status", + expectedValue: "draft", +}); +void executableAtomic + .run() + .then((result) => assert.equal(result, "not-applied")) + .finally(() => { + executable.Query.prototype.run = originalRun; + }) + .then(() => { + console.log( + "staged-query purity, AQL stages, and root compatibility: PASS", + ); + }); diff --git a/src/antelope.test.ts b/src/antelope.test.ts index 2f5de2c..2c531b3 100644 --- a/src/antelope.test.ts +++ b/src/antelope.test.ts @@ -11,7 +11,7 @@ export default defineConfig({ source: { type: "package", package: "@antelopejs/mongodb", - version: "1.2.7", + version: "1.3.0", }, }, }, diff --git a/src/atomic.ts b/src/atomic.ts index 6b81796..96b8aac 100644 --- a/src/atomic.ts +++ b/src/atomic.ts @@ -1,46 +1,12 @@ import { Query } from "./query"; -import type { QueryStage } from "./common"; - -/** Matches an absent revision field on an existing record, not stored null. */ -export interface MissingRevision { - kind: "missing"; -} - -export interface AtomicCondition { - revisionField: keyof T & string; - expectedRevision: string | MissingRevision; -} - -/** Replaces supplied top-level fields and atomically installs a fresh revision. */ -export interface AtomicUpdate extends AtomicCondition { - type: "update"; - nextRevision: string; - patch: Partial; -} - -export interface AtomicDelete extends AtomicCondition { - type: "delete"; -} - -export type AtomicEqualityValue = string | number | boolean | Date; - -/** Deletes by one observed scalar value; does not provide revision or ABA protection. */ -export interface AtomicDeleteIfEqual { - type: "deleteIfEqual"; - field: keyof T & string; - expectedValue: AtomicEqualityValue; -} - -export type AtomicMutation = - | AtomicUpdate - | AtomicDelete - | AtomicDeleteIfEqual; - -export type AtomicMutationOutcome = "applied" | "not-applied" | "unknown"; +import { + AtomicMutationQuery as StagedAtomicMutationQuery, + ValidateAtomicMutation, + ValidateAtomicMutationTable, + type AtomicMutationOutcome, +} from "./staged-query/atomic"; const outcomes: readonly unknown[] = ["applied", "not-applied", "unknown"]; -const tableStages = ["schema", "instance", "table"]; -const identityFields = ["id", "_id"]; /** Signals an adapter that does not implement the atomic mutation contract. */ export class AtomicMutationUnsupportedError extends Error { @@ -50,149 +16,35 @@ export class AtomicMutationUnsupportedError extends Error { } } -function isRecord(value: unknown): value is Record { - return ( - value !== null && - typeof value === "object" && - [Object.prototype, null].includes(Object.getPrototypeOf(value)) - ); -} - -function isToken(value: unknown): value is string { - return typeof value === "string" && value.length > 0; -} - -function validateField(field: string) { - if (!isToken(field) || field.includes(".") || field.startsWith("$")) { - throw new TypeError("Atomic mutation fields must be literal field names"); - } -} - -function validateConstant(value: unknown, ancestors = new Set()) { - if (value === null || ["string", "boolean"].includes(typeof value)) return; - if (typeof value === "number" && Number.isFinite(value)) return; - if (value instanceof Date && Number.isFinite(value.getTime())) return; - if ((!Array.isArray(value) && !isRecord(value)) || ancestors.has(value)) { - throw new TypeError("Atomic mutation patches must contain constant data"); - } - if (Array.isArray(value) && Object.keys(value).length !== value.length) { - throw new TypeError("Atomic mutation arrays must not contain holes"); - } - ancestors.add(value); - for (const [field, child] of Object.entries(value)) { - validateField(field); - validateConstant(child, ancestors); - } - ancestors.delete(value); -} - -/** Validates literal input; adapters must additionally supply scope fields. */ -export function ValidateAtomicMutation( - key: string, - request: AtomicMutation, - protectedFields: readonly string[] = [], -): void { - if (!isToken(key) || !isRecord(request)) { - throw new TypeError("Atomic mutation requires one literal record identity"); - } - const forbidden = [...identityFields, ...protectedFields]; - if (request.type === "deleteIfEqual") { - validateEquality(request, forbidden); - return; - } - validateField(request.revisionField); - if (forbidden.includes(request.revisionField)) { - throw new TypeError("Atomic mutation revision cannot be an identity field"); - } - const expected = request.expectedRevision; - const isMissing = - isRecord(expected) && - expected.kind === "missing" && - Object.keys(expected).length === 1; - if (!isToken(expected) && !isMissing) { - throw new TypeError("Expected revision must be a token or missing tag"); - } - if (request.type === "delete") return; - if (request.type !== "update") { - throw new TypeError("Unknown atomic mutation operation"); - } - validateUpdate(request, forbidden); -} - -function validateEquality( - request: AtomicDeleteIfEqual, - forbidden: string[], -) { - validateField(request.field); - if (forbidden.includes(request.field)) { - throw new TypeError("Atomic equality field cannot be an identity field"); - } - const value = request.expectedValue; +async function run( + this: StagedAtomicMutationQuery, +): Promise { + const stages = this.build(); + const terminal = stages.at(-1)!; + ValidateAtomicMutationTable(stages.slice(0, -1)); if ( - !["string", "number", "boolean"].includes(typeof value) && - !(value instanceof Date) + terminal.stage !== "atomicMutation" || + terminal.options !== undefined || + terminal.args.length !== 2 ) { - throw new TypeError("Atomic equality requires a scalar or Date"); - } - validateConstant(value); -} - -function validateUpdate(request: AtomicUpdate, forbidden: string[]) { - if ( - !isToken(request.nextRevision) || - request.nextRevision === request.expectedRevision - ) { - throw new TypeError("Atomic mutation requires a changed revision token"); - } - if (!isRecord(request.patch)) { - throw new TypeError("Atomic mutation patch must be an object"); - } - forbidden.push(request.revisionField); - if (Object.keys(request.patch).some((field) => forbidden.includes(field))) { throw new TypeError( - "Atomic mutation patch cannot change identity or revision", + "Atomic mutation requires its canonical terminal stage", ); } - validateConstant(request.patch); + ValidateAtomicMutation(terminal.args[0], terminal.args[1]); + const result = await Query.prototype.run.call(this); + if (!outcomes.includes(result)) throw new AtomicMutationUnsupportedError(); + return result; } -/** Rejects selections and cross-instance operations before adapter dispatch. */ -export function ValidateAtomicMutationTable(stages: QueryStage[]): void { - if ( - stages.length !== tableStages.length || - stages.some((stage, index) => stage.stage !== tableStages[index]) || - (stages[1].options?.id !== undefined && - typeof stages[1].options.id !== "string") - ) { - throw new TypeError("Atomic mutation requires one instance-scoped table"); - } +function cursor(): AsyncGenerator { + throw new TypeError("Atomic mutations cannot be executed as cursors"); } -/** Executes only the explicit atomic stage; invalid adapter results fail closed. */ -export class AtomicMutationQuery extends Query { - public override async run(): Promise { - const terminal = this.stages.at(-1)!; - ValidateAtomicMutationTable(this.stages.slice(0, -1)); - if ( - terminal.stage !== "atomicMutation" || - terminal.options !== undefined || - terminal.args.length !== 2 - ) { - throw new TypeError( - "Atomic mutation requires its canonical terminal stage", - ); - } - ValidateAtomicMutation(terminal.args[0], terminal.args[1]); - const result = await super.run(); - if (!outcomes.includes(result)) throw new AtomicMutationUnsupportedError(); - return result; - } +Object.defineProperties(StagedAtomicMutationQuery.prototype, { + run: { configurable: true, value: run, writable: true }, + cursor: { configurable: true, value: cursor, writable: true }, +}); - public override cursor(): AsyncGenerator< - AtomicMutationOutcome, - void, - unknown - > { - throw new TypeError("Atomic mutations cannot be executed as cursors"); - } -} +export * from "./staged-query/atomic"; +export { StagedAtomicMutationQuery as AtomicMutationQuery }; diff --git a/src/common.ts b/src/common.ts index b1bf270..212ff6d 100644 --- a/src/common.ts +++ b/src/common.ts @@ -1,135 +1 @@ -import type { Class } from "@antelopejs/interface-core/decorators"; - -import type { Datum } from "./datum"; -import type { Query } from "./query"; -import type { ValueProxy, ValueProxyOrValue } from "./valueproxy"; - -/** - * Recursive Partial generic type - */ -export type DeepPartial = { - [K in keyof T]?: T[K] extends Array - ? Array> - : T[K] extends ReadonlyArray - ? ReadonlyArray> - : DeepPartial; -}; - -/** - * Change event - */ -export interface Changes { - /** - * The type of change that occured - * - * Possible values: added, removed, modified - */ - changeType: "added" | "removed" | "modified"; - - /** - * Value prior to the change - */ - oldValue?: T; - - /** - * New value after the change - */ - newValue?: T; -} - -export interface InsertOptions { - conflict?: "update" | "replace"; -} - -export interface QueryStage { - stage: string; - options?: any; - args: any[]; -} - -export function QueryStage(stage: string, options?: any, ...args: any[]) { - return { - stage, - options, - args, - }; -} - -export class StagedObject { - protected readonly stages: QueryStage[]; - - public constructor(newStage: QueryStage, previous?: StagedObject) { - this.stages = previous ? [...previous.stages, newStage] : [newStage]; - } - - //@internal - public build() { - return this.stages; - } - - protected stage( - type: undefined, - stage: string, - options?: any, - ...args: any[] - ): this; - protected stage( - type: Class, - stage: string, - options?: any, - ...args: any[] - ): T; - protected stage( - type: Class | undefined, - stage: string, - options?: any, - ...args: any[] - ) { - return new (type ?? (this.constructor as Class))( - { - stage, - options, - args, - }, - this, - ); - } - - private static nextargid = 0; - protected callfunc( - func: (...args: T) => any, - ...args: (typeof StagedObject)[] - ): QueryStage { - const argNumbers: number[] = []; - const argValues: StagedObject[] = []; - for (let i = 0; i < args.length; ++i) { - const id = StagedObject.nextargid++; - argNumbers[i] = id; - argValues[i] = new args[i](QueryStage("arg", undefined, id)); - } - return { - stage: "func", // TODO: using the query stage structure here doesnt make any sense - args: [argNumbers, func(...(argValues as T))], - }; - } -} - -// TODO: This adds a lot of complexity to implementations, investigate if we should remove it. -export type Value = Datum | ValueProxyOrValue; - -type UnknownObject = Record; -type ExtractTypeObject = T extends infer O - ? { - [K in keyof O]: ExtractType; - } - : never; -export type ExtractType = - T extends ValueProxy - ? A - : T extends Query - ? A - : T extends UnknownObject - ? ExtractTypeObject - : T extends Array - ? Array> - : T; +export * from "./staged-query/common"; diff --git a/src/datum.ts b/src/datum.ts index d01b10d..71a6dc1 100644 --- a/src/datum.ts +++ b/src/datum.ts @@ -1,102 +1 @@ -import { Query } from "./query"; -import { ValueProxy } from "./valueproxy"; -import type { Selection } from "./selection"; -import type { ExtractType, Value } from "./common"; - -export class Datum extends Query { - /** - * Changes the type of this datum. - * This does not actually perform any conversion, it only changes the typescript type. - * - * @returns Same datum with a different type - */ - public cast() { - return this as unknown as Datum; - } - - /** - * Indexes the datum. - * - * TODO: Better name? - * TODO: typing for compound keys (a.b.c) - * - * @param key Field name - * @param def Default value - * @returns New datum with the value - */ - public key, U = undefined>(key: K, def?: U) { - return this.stage( - Datum< - U extends undefined - ? NonNullable[K] - : NonNullable[K]> | U - >, - "key", - undefined, - key, - def, - ); - } - - /** - * Defaults the datum to a given value if it is null. - * - * @param value Default value - * @returns Current datum or given value - */ - public default(val: Value) { - return this.stage( - Datum | U>, - "default", - undefined, - val, - ); - } - - /** - * Run a mapping function on the datum. - * - * @param mapper Mapping function - * @returns New datum with the result of the mapper - */ - public do(mapper: (val: ValueProxy) => U) { - return this.stage( - Datum>, - "map", - undefined, - this.callfunc(mapper, ValueProxy), - ); - } - - /** - * Perform a foreign key lookup - * - * @param other Other table - * @param localKey Key in local object - * @param otherKey Key in other table - */ - public lookup( - other: Selection, - localKey: TK, - otherKey: keyof U, - ) { - return this.stage( - Datum & Record>, - "lookup", - { localKey, otherKey }, - other, - ); - } - - /** - * Plucks fields from the documents. - * - * TODO: Better typing - * - * @param fields Fields to keep - * @returns New datum - */ - public pluck(...fields: string[]) { - return this.stage(Datum>, "pluck", undefined, fields); - } -} +export { Datum } from "./staged-query/datum"; diff --git a/src/query.ts b/src/query.ts index 397b54e..7e086db 100644 --- a/src/query.ts +++ b/src/query.ts @@ -1,6 +1,7 @@ import { InterfaceFunction } from "@antelopejs/interface-core"; -import { StagedObject } from "./common"; +import { Query as StagedQuery } from "./staged-query/query"; +import { StagedObject } from "./staged-query/common"; //@internal export const RunQuery = @@ -62,35 +63,54 @@ class IterableCursor implements AsyncGenerator { } } -export class Query extends StagedObject implements PromiseLike { - /** - * Execute the query - * - * @returns Query result - */ - public run(): Promise { - return RunQuery(this.stages); - } +export interface QueryExecution extends PromiseLike { + run(): Promise; + cursor(): AsyncGenerator ? U : T, void, unknown>; + [Symbol.asyncIterator](): AsyncGenerator< + T extends Array ? U : T, + void, + unknown + >; +} - // oxlint-disable-next-line unicorn/no-thenable -- Query is deliberately PromiseLike so `await query` runs it; the contract requires this method. - public then( - onfulfilled?: ((value: T) => TResult1 | PromiseLike) | null, - onrejected?: ((reason: any) => TResult2 | PromiseLike) | null, - ): PromiseLike { - return this.run().then(onfulfilled, onrejected); - } +declare module "./staged-query/query" { + interface Query extends QueryExecution {} +} - //TODO: core interface function for async generators +function run(this: StagedQuery): Promise { + return RunQuery(this.build()); +} - public cursor(): AsyncGenerator< - T extends Array ? T1 : T, - void, - unknown - > { - return new IterableCursor(this.stages); - } +// oxlint-disable-next-line unicorn/no-thenable -- Query is deliberately PromiseLike so `await query` runs it; the contract requires this method. +function then( + this: StagedQuery, + onfulfilled?: ((value: T) => TResult1 | PromiseLike) | null, + onrejected?: ((reason: any) => TResult2 | PromiseLike) | null, +): PromiseLike { + return this.run().then(onfulfilled, onrejected); +} - [Symbol.asyncIterator]() { - return this.cursor(); - } +function cursor( + this: StagedQuery, +): AsyncGenerator ? U : T, void, unknown> { + return new IterableCursor(this.build()); } + +function iterate( + this: StagedQuery, +): AsyncGenerator ? U : T, void, unknown> { + return this.cursor(); +} + +Object.defineProperties(StagedQuery.prototype, { + run: { configurable: true, value: run, writable: true }, + then: { configurable: true, value: then, writable: true }, + cursor: { configurable: true, value: cursor, writable: true }, + [Symbol.asyncIterator]: { + configurable: true, + value: iterate, + writable: true, + }, +}); + +export { StagedQuery as Query }; diff --git a/src/schema.ts b/src/schema.ts index 6376e92..31d45f4 100644 --- a/src/schema.ts +++ b/src/schema.ts @@ -1,166 +1,27 @@ import { RegisteringProxy } from "@antelopejs/interface-core"; -import { Query } from "./query"; -import { Table } from "./selection"; -import { QueryStage, StagedObject } from "./common"; - -/** - * Secondary table index definition - */ -export interface IndexDefinition { - /** - * Fields to use for compound indexes - */ - fields?: string[]; - - /** - * Whether or not this is a multi index - */ - multi?: boolean; -} - -export type StringFieldType = - | string - | Array - | { [subfield: string]: StringFieldType }; - -interface IoTsCodec { - readonly _A: unknown; - readonly _O: unknown; - readonly _I: unknown; - readonly name: string; - is(u: unknown): boolean; - encode(a: never): unknown; - decode(i: unknown): unknown; -} - -export type FieldType = StringFieldType | IoTsCodec; - -/** - * Schema table definition - */ -export interface TableDefinition { - /** - * Field names and their data types - */ - fields: Record; - - /** - * List of secondary indexes - */ - indexes: Record; -} - -/** - * Table names and their definitions - */ -export interface SchemaDefinition { - [tableName: string]: TableDefinition; -} - -/** - * Sentinel instance id meaning "operate across all instances of the schema". - */ -export const CROSS_INSTANCE: unique symbol = Symbol( - "antelopejs:cross-instance", -); - -/** - * Instance identifier accepted by {@link Schema.instance}. - */ -export type InstanceId = string | typeof CROSS_INSTANCE; +import { Schema as StagedSchema } from "./staged-query/schema"; +import type { SchemaDefinition } from "./staged-query/schema"; //@internal export const Schemas = new RegisteringProxy< (name: string, def: SchemaDefinition) => void >(); -/** - * A schema determines the structure of a database - * - * Each schema can have multiple instances. The internal organization of these - * instances is left up to the module implementation. - */ -export class Schema extends StagedObject { - private static readonly registry = new Map(); - - /** - * Retrieves a previously defined schema by its ID - * - * @param id Schema ID - * @returns The schema instance, or undefined if not found - */ - public static get(id: string): Schema | undefined { - return Schema.registry.get(id); - } - - /** - * Define a new schema with the given ID - * - * @param id ID of this schema, changing this will leave previous data inaccessible - * @param definition Schema definition (tables, fields, indexes..) - */ - public constructor( - public readonly id: string, - public readonly definition: SchemaDefinition, - ) { - super(QueryStage("schema", { id })); +export class Schema extends StagedSchema { + public constructor(id: string, definition: SchemaDefinition) { + super(id, definition); Schemas.register(id, definition); - Schema.registry.set(id, this); - } - - /** - * Gets a specific instance of the schema - * - * @param id Instance ID, or {@link CROSS_INSTANCE} for a cross-instance query - * @returns Schema instance - */ - public instance(id?: InstanceId) { - return this.stage(SchemaInstance, "instance", { id }); - } - - /** - * Creates a new instance of the schema - * - * @param id Instance ID - * @returns Created instance ID - */ - public createInstance(id?: string) { - return this.stage(Query, "createInstance", { id }); - } - - /** - * Destroys an existing instance of the schema - * - * @param id Instance ID - */ - public destroyInstance(id?: string) { - return this.stage(Query, "destroyInstance", { id }); - } - - /** - * Lists the IDs of named instances of this schema. - * - * The default (unnamed) instance is not included. - * - * @returns IDs of named instances of this schema - */ - public listInstances() { - return this.stage(Query, "listInstances"); } } -/** - * Schema instance, could be a database or a filtered portion of one depending on the implementation - */ -export class SchemaInstance extends StagedObject { - /** - * Gets a table from the instance - * - * @param id Table name - * @returns Table - */ - public table(id: TK) { - return this.stage(Table, "table", { id }); - } -} +export { + CROSS_INSTANCE, + type FieldType, + type IndexDefinition, + type InstanceId, + type SchemaDefinition, + SchemaInstance, + type StringFieldType, + type TableDefinition, +} from "./staged-query/schema"; diff --git a/src/selection.ts b/src/selection.ts index c2ad186..fb152e1 100644 --- a/src/selection.ts +++ b/src/selection.ts @@ -1,190 +1 @@ -import { Datum } from "./datum"; -import { Query } from "./query"; -import { Stream } from "./stream"; -import { ValueProxy, type ValueProxyOrValue } from "./valueproxy"; -import type { - Changes, - DeepPartial, - ExtractType, - InsertOptions, -} from "./common"; -import { - AtomicMutationQuery, - ValidateAtomicMutation, - ValidateAtomicMutationTable, - type AtomicMutation, - type AtomicMutationOutcome, -} from "./atomic"; - -type SelectionKey = string | number | boolean; - -/** - * Selection containing a single element - */ -export class SingleSelection extends Datum { - /** - * Update fields of this selection with the given values - * - * @param document Partial document with new values - * @returns Number of modified documents - */ - public update(document: DeepPartial): Query; - public update( - document: (val: ValueProxy) => U, - ): ExtractType extends DeepPartial ? Query : never; - public update(document: DeepPartial | ((val: ValueProxy) => unknown)) { - if (typeof document === "function") { - return this.stage( - Query, - "update", - undefined, - this.callfunc(document, ValueProxy), - ); - } - return this.stage(Query, "update", undefined, document); - } - - /** - * Replace documents of this selection - * - * @param document Partial document to replace with - * @returns Number of modified documents - */ - public replace(document: DeepPartial) { - return this.stage(Query, "replace", undefined, document); - } - - /** - * Delete selected documents - * - * @returns Number of deleted documents - */ - public delete() { - return this.stage(Query, "delete"); - } - - /** - * Turns this selection into a change feed - * - * @returns Change feed - */ - public changes() { - return this.stage(Query[]>, "changes"); - } -} - -/** - * Selection containing any number of elements - */ -export class Selection extends Stream { - /** - * Update fields of this selection with the given values - * - * @param document Partial document with new values - * @returns Number of modified documents - */ - public update(document: DeepPartial): Query; - public update( - document: (val: ValueProxy) => U, - ): ExtractType extends DeepPartial ? Query : never; - public update(document: DeepPartial | ((val: ValueProxy) => unknown)) { - if (typeof document === "function") { - return this.stage( - Query, - "update", - undefined, - this.callfunc(document, ValueProxy), - ); - } - return this.stage(Query, "update", undefined, document); - } - - /** - * Replace documents of this selection - * - * @param document Partial document to replace with - * @returns Number of modified documents - */ - public replace(document: DeepPartial) { - return this.stage(Query, "replace", undefined, document); - } - - /** - * Delete selected documents - * - * @returns Number of deleted documents - */ - public delete() { - return this.stage(Query, "delete"); - } -} - -/** - * Database table - */ -export class Table extends Selection { - /** Atomically checks one record's revision and patches or deletes it without upsert. */ - public atomicMutation( - key: string, - request: AtomicMutation, - ): Query { - ValidateAtomicMutationTable(this.stages); - ValidateAtomicMutation(key, request); - return this.stage( - AtomicMutationQuery, - "atomicMutation", - undefined, - key, - request, - ); - } - - /** - * Inserts one or more documents into this table - * - * @param obj Document(s) to insert - * @returns Inserted IDs - */ - public insert( - obj: DeepPartial | DeepPartial[], - options?: InsertOptions, - ) { - return this.stage(Query, "insert", options, obj); - } - - /** - * Gets a document using its primary key - * - * @param key Primary key value - * @returns Single document selection - */ - public get(key: ValueProxyOrValue) { - return this.stage(SingleSelection, "get", undefined, key); - } - - /** - * Gets multiple documents using a secondary index - * - * @param keys Key value(s) - * @param index Secondary index, will use the primary key if undefined - * @returns Multiple document selection - */ - public getAll( - keys: ValueProxyOrValue | ValueProxyOrValue[], - index?: string, - ) { - return this.stage(Selection, "getAll", { index }, keys); - } - - /** - * Gets multiple documents using a secondary index and bounding values - * - * @param index Secondary index - * @param low Lowest value of the range - * @param high Highest value of the range (excluded) - * @returns Multiple document selection - */ - public between(index: TK, low: T[TK], high: T[TK]) { - return this.stage(Selection, "between", { index }, low, high); - } -} +export { Selection, SingleSelection, Table } from "./staged-query/selection"; diff --git a/src/staged-query/atomic.ts b/src/staged-query/atomic.ts new file mode 100644 index 0000000..3b3f622 --- /dev/null +++ b/src/staged-query/atomic.ts @@ -0,0 +1,162 @@ +import { Query } from "./query"; +import type { QueryStage } from "./common"; + +/** Matches an absent revision field on an existing record, not stored null. */ +export interface MissingRevision { + kind: "missing"; +} + +export interface AtomicCondition { + revisionField: keyof T & string; + expectedRevision: string | MissingRevision; +} + +/** Replaces supplied top-level fields and atomically installs a fresh revision. */ +export interface AtomicUpdate extends AtomicCondition { + type: "update"; + nextRevision: string; + patch: Partial; +} + +export interface AtomicDelete extends AtomicCondition { + type: "delete"; +} + +export type AtomicEqualityValue = string | number | boolean | Date; + +/** Deletes by one observed scalar value; does not provide revision or ABA protection. */ +export interface AtomicDeleteIfEqual { + type: "deleteIfEqual"; + field: keyof T & string; + expectedValue: AtomicEqualityValue; +} + +export type AtomicMutation = + | AtomicUpdate + | AtomicDelete + | AtomicDeleteIfEqual; + +export type AtomicMutationOutcome = "applied" | "not-applied" | "unknown"; + +const tableStages = ["schema", "instance", "table"]; +const identityFields = ["id", "_id"]; + +function isRecord(value: unknown): value is Record { + return ( + value !== null && + typeof value === "object" && + [Object.prototype, null].includes(Object.getPrototypeOf(value)) + ); +} + +function isToken(value: unknown): value is string { + return typeof value === "string" && value.length > 0; +} + +function validateField(field: string) { + if (!isToken(field) || field.includes(".") || field.startsWith("$")) { + throw new TypeError("Atomic mutation fields must be literal field names"); + } +} + +function validateConstant(value: unknown, ancestors = new Set()) { + if (value === null || ["string", "boolean"].includes(typeof value)) return; + if (typeof value === "number" && Number.isFinite(value)) return; + if (value instanceof Date && Number.isFinite(value.getTime())) return; + if ((!Array.isArray(value) && !isRecord(value)) || ancestors.has(value)) { + throw new TypeError("Atomic mutation patches must contain constant data"); + } + if (Array.isArray(value) && Object.keys(value).length !== value.length) { + throw new TypeError("Atomic mutation arrays must not contain holes"); + } + ancestors.add(value); + for (const [field, child] of Object.entries(value)) { + validateField(field); + validateConstant(child, ancestors); + } + ancestors.delete(value); +} + +/** Validates literal input; adapters must additionally supply scope fields. */ +export function ValidateAtomicMutation( + key: string, + request: AtomicMutation, + protectedFields: readonly string[] = [], +): void { + if (!isToken(key) || !isRecord(request)) { + throw new TypeError("Atomic mutation requires one literal record identity"); + } + const forbidden = [...identityFields, ...protectedFields]; + if (request.type === "deleteIfEqual") { + validateEquality(request, forbidden); + return; + } + validateField(request.revisionField); + if (forbidden.includes(request.revisionField)) { + throw new TypeError("Atomic mutation revision cannot be an identity field"); + } + const expected = request.expectedRevision; + const isMissing = + isRecord(expected) && + expected.kind === "missing" && + Object.keys(expected).length === 1; + if (!isToken(expected) && !isMissing) { + throw new TypeError("Expected revision must be a token or missing tag"); + } + if (request.type === "delete") return; + if (request.type !== "update") { + throw new TypeError("Unknown atomic mutation operation"); + } + validateUpdate(request, forbidden); +} + +function validateEquality( + request: AtomicDeleteIfEqual, + forbidden: string[], +) { + validateField(request.field); + if (forbidden.includes(request.field)) { + throw new TypeError("Atomic equality field cannot be an identity field"); + } + const value = request.expectedValue; + if ( + !["string", "number", "boolean"].includes(typeof value) && + !(value instanceof Date) + ) { + throw new TypeError("Atomic equality requires a scalar or Date"); + } + validateConstant(value); +} + +function validateUpdate(request: AtomicUpdate, forbidden: string[]) { + if ( + !isToken(request.nextRevision) || + request.nextRevision === request.expectedRevision + ) { + throw new TypeError("Atomic mutation requires a changed revision token"); + } + if (!isRecord(request.patch)) { + throw new TypeError("Atomic mutation patch must be an object"); + } + forbidden.push(request.revisionField); + if (Object.keys(request.patch).some((field) => forbidden.includes(field))) { + throw new TypeError( + "Atomic mutation patch cannot change identity or revision", + ); + } + validateConstant(request.patch); +} + +/** Rejects selections and cross-instance operations before adapter dispatch. */ +export function ValidateAtomicMutationTable(stages: QueryStage[]): void { + if ( + stages.length !== tableStages.length || + stages.some((stage, index) => stage.stage !== tableStages[index]) || + (stages[1].options?.id !== undefined && + typeof stages[1].options.id !== "string") + ) { + throw new TypeError("Atomic mutation requires one instance-scoped table"); + } +} + +export class AtomicMutationQuery extends Query {} diff --git a/src/staged-query/common.ts b/src/staged-query/common.ts new file mode 100644 index 0000000..b74d0f7 --- /dev/null +++ b/src/staged-query/common.ts @@ -0,0 +1,139 @@ +import type { Datum } from "./datum"; +import type { Query } from "./query"; +import type { ValueProxy, ValueProxyOrValue } from "./valueproxy"; + +export type StagedConstructor = new ( + newStage: QueryStage, + previous?: StagedObject, +) => T; + +/** + * Recursive Partial generic type + */ +export type DeepPartial = { + [K in keyof T]?: T[K] extends Array + ? Array> + : T[K] extends ReadonlyArray + ? ReadonlyArray> + : DeepPartial; +}; + +/** + * Change event + */ +export interface Changes { + /** + * The type of change that occured + * + * Possible values: added, removed, modified + */ + changeType: "added" | "removed" | "modified"; + + /** + * Value prior to the change + */ + oldValue?: T; + + /** + * New value after the change + */ + newValue?: T; +} + +export interface InsertOptions { + conflict?: "update" | "replace"; +} + +export interface QueryStage { + stage: string; + options?: any; + args: any[]; +} + +export function QueryStage(stage: string, options?: any, ...args: any[]) { + return { + stage, + options, + args, + }; +} + +export class StagedObject { + protected readonly stages: QueryStage[]; + + public constructor(newStage: QueryStage, previous?: StagedObject) { + this.stages = previous ? [...previous.stages, newStage] : [newStage]; + } + + //@internal + public build() { + return this.stages; + } + + protected stage( + type: undefined, + stage: string, + options?: any, + ...args: any[] + ): this; + protected stage( + type: StagedConstructor, + stage: string, + options?: any, + ...args: any[] + ): T; + protected stage( + type: StagedConstructor | undefined, + stage: string, + options?: any, + ...args: any[] + ) { + const Constructor = type ?? (this.constructor as StagedConstructor); + return new Constructor( + { + stage, + options, + args, + }, + this, + ); + } + + private static nextargid = 0; + protected callfunc( + func: (...args: T) => any, + ...args: (typeof StagedObject)[] + ): QueryStage { + const argNumbers: number[] = []; + const argValues: StagedObject[] = []; + for (let i = 0; i < args.length; ++i) { + const id = StagedObject.nextargid++; + argNumbers[i] = id; + argValues[i] = new args[i](QueryStage("arg", undefined, id)); + } + return { + stage: "func", // TODO: using the query stage structure here doesnt make any sense + args: [argNumbers, func(...(argValues as T))], + }; + } +} + +// TODO: This adds a lot of complexity to implementations, investigate if we should remove it. +export type Value = Datum | ValueProxyOrValue; + +type UnknownObject = Record; +type ExtractTypeObject = T extends infer O + ? { + [K in keyof O]: ExtractType; + } + : never; +export type ExtractType = + T extends ValueProxy + ? A + : T extends Query + ? A + : T extends UnknownObject + ? ExtractTypeObject + : T extends Array + ? Array> + : T; diff --git a/src/staged-query/datum.ts b/src/staged-query/datum.ts new file mode 100644 index 0000000..e3f4fc0 --- /dev/null +++ b/src/staged-query/datum.ts @@ -0,0 +1,102 @@ +import { Query } from "./query"; +import { ValueProxy } from "./valueproxy"; +import type { ExtractType, Value } from "./common"; +import type { Selection } from "./selection"; + +export class Datum extends Query { + /** + * Changes the type of this datum. + * This does not actually perform any conversion, it only changes the typescript type. + * + * @returns Same datum with a different type + */ + public cast() { + return this as unknown as Datum; + } + + /** + * Indexes the datum. + * + * TODO: Better name? + * TODO: typing for compound keys (a.b.c) + * + * @param key Field name + * @param def Default value + * @returns New datum with the value + */ + public key, U = undefined>(key: K, def?: U) { + return this.stage( + Datum< + U extends undefined + ? NonNullable[K] + : NonNullable[K]> | U + >, + "key", + undefined, + key, + def, + ); + } + + /** + * Defaults the datum to a given value if it is null. + * + * @param value Default value + * @returns Current datum or given value + */ + public default(val: Value) { + return this.stage( + Datum | U>, + "default", + undefined, + val, + ); + } + + /** + * Run a mapping function on the datum. + * + * @param mapper Mapping function + * @returns New datum with the result of the mapper + */ + public do(mapper: (val: ValueProxy) => U) { + return this.stage( + Datum>, + "map", + undefined, + this.callfunc(mapper, ValueProxy), + ); + } + + /** + * Perform a foreign key lookup + * + * @param other Other table + * @param localKey Key in local object + * @param otherKey Key in other table + */ + public lookup( + other: Selection, + localKey: TK, + otherKey: keyof U, + ) { + return this.stage( + Datum & Record>, + "lookup", + { localKey, otherKey }, + other, + ); + } + + /** + * Plucks fields from the documents. + * + * TODO: Better typing + * + * @param fields Fields to keep + * @returns New datum + */ + public pluck(...fields: string[]) { + return this.stage(Datum>, "pluck", undefined, fields); + } +} diff --git a/src/staged-query/index.ts b/src/staged-query/index.ts new file mode 100644 index 0000000..d46336c --- /dev/null +++ b/src/staged-query/index.ts @@ -0,0 +1,18 @@ +export * from "./common"; +export * from "./atomic"; +export { Datum } from "./datum"; +export { Query } from "./query"; +export { + CROSS_INSTANCE, + Schema, + SchemaInstance, + type FieldType, + type IndexDefinition, + type InstanceId, + type SchemaDefinition, + type StringFieldType, + type TableDefinition, +} from "./schema"; +export { Selection, SingleSelection, Table } from "./selection"; +export { Stream } from "./stream"; +export { ValueProxy, type ValueProxyOrValue } from "./valueproxy"; diff --git a/src/staged-query/query.ts b/src/staged-query/query.ts new file mode 100644 index 0000000..dad5a15 --- /dev/null +++ b/src/staged-query/query.ts @@ -0,0 +1,3 @@ +import { StagedObject } from "./common"; + +export class Query extends StagedObject {} diff --git a/src/staged-query/schema.ts b/src/staged-query/schema.ts new file mode 100644 index 0000000..35d69d1 --- /dev/null +++ b/src/staged-query/schema.ts @@ -0,0 +1,163 @@ +import { Query } from "./query"; +import { Table } from "./selection"; +import { QueryStage, StagedObject } from "./common"; + +/** + * Secondary table index definition + */ +export interface IndexDefinition { + /** + * Fields to use for compound indexes + */ + fields?: string[]; + + /** + * Whether or not this is a multi index + */ + multi?: boolean; +} + +export type StringFieldType = + | string + | Array + | { [subfield: string]: StringFieldType }; + +interface IoTsCodec { + readonly _A: unknown; + readonly _O: unknown; + readonly _I: unknown; + readonly name: string; + is(u: unknown): boolean; + encode(a: never): unknown; + decode(i: unknown): unknown; +} + +export type FieldType = StringFieldType | IoTsCodec; + +/** + * Schema table definition + */ +export interface TableDefinition { + /** + * Field names and their data types + */ + fields: Record; + + /** + * List of secondary indexes + */ + indexes: Record; +} + +/** + * Table names and their definitions + */ +export interface SchemaDefinition { + [tableName: string]: TableDefinition; +} + +/** + * Sentinel instance id meaning "operate across all instances of the schema". + */ +export const CROSS_INSTANCE: unique symbol = Symbol( + "antelopejs:cross-instance", +); + +/** + * Instance identifier accepted by {@link Schema.instance}. + */ +export type InstanceId = string | typeof CROSS_INSTANCE; + +/** + * A schema determines the structure of a database + * + * Each schema can have multiple instances. The internal organization of these + * instances is left up to the module implementation. + */ +export class Schema< + T = any, + Definition = SchemaDefinition, +> extends StagedObject { + private static readonly registry = new Map>(); + + /** + * Retrieves a previously defined schema by its ID + * + * @param id Schema ID + * @returns The schema instance, or undefined if not found + */ + public static get( + id: string, + ): Schema | undefined { + return Schema.registry.get(id) as Schema | undefined; + } + + /** + * Define a new schema with the given ID + * + * @param id ID of this schema, changing this will leave previous data inaccessible + * @param definition Schema definition (tables, fields, indexes..) + */ + public constructor( + public readonly id: string, + public readonly definition: Definition, + ) { + super(QueryStage("schema", { id })); + Schema.registry.set(id, this); + } + + /** + * Gets a specific instance of the schema + * + * @param id Instance ID, or {@link CROSS_INSTANCE} for a cross-instance query + * @returns Schema instance + */ + public instance(id?: InstanceId) { + return this.stage(SchemaInstance, "instance", { id }); + } + + /** + * Creates a new instance of the schema + * + * @param id Instance ID + * @returns Created instance ID + */ + public createInstance(id?: string) { + return this.stage(Query, "createInstance", { id }); + } + + /** + * Destroys an existing instance of the schema + * + * @param id Instance ID + */ + public destroyInstance(id?: string) { + return this.stage(Query, "destroyInstance", { id }); + } + + /** + * Lists the IDs of named instances of this schema. + * + * The default (unnamed) instance is not included. + * + * @returns IDs of named instances of this schema + */ + public listInstances() { + return this.stage(Query, "listInstances"); + } +} + +/** + * Schema instance, could be a database or a filtered portion of one depending on the implementation + */ +export class SchemaInstance extends StagedObject { + /** + * Gets a table from the instance + * + * @param id Table name + * @returns Table + */ + public table(id: TK) { + return this.stage(Table, "table", { id }); + } +} diff --git a/src/staged-query/selection.ts b/src/staged-query/selection.ts new file mode 100644 index 0000000..c2ad186 --- /dev/null +++ b/src/staged-query/selection.ts @@ -0,0 +1,190 @@ +import { Datum } from "./datum"; +import { Query } from "./query"; +import { Stream } from "./stream"; +import { ValueProxy, type ValueProxyOrValue } from "./valueproxy"; +import type { + Changes, + DeepPartial, + ExtractType, + InsertOptions, +} from "./common"; +import { + AtomicMutationQuery, + ValidateAtomicMutation, + ValidateAtomicMutationTable, + type AtomicMutation, + type AtomicMutationOutcome, +} from "./atomic"; + +type SelectionKey = string | number | boolean; + +/** + * Selection containing a single element + */ +export class SingleSelection extends Datum { + /** + * Update fields of this selection with the given values + * + * @param document Partial document with new values + * @returns Number of modified documents + */ + public update(document: DeepPartial): Query; + public update( + document: (val: ValueProxy) => U, + ): ExtractType extends DeepPartial ? Query : never; + public update(document: DeepPartial | ((val: ValueProxy) => unknown)) { + if (typeof document === "function") { + return this.stage( + Query, + "update", + undefined, + this.callfunc(document, ValueProxy), + ); + } + return this.stage(Query, "update", undefined, document); + } + + /** + * Replace documents of this selection + * + * @param document Partial document to replace with + * @returns Number of modified documents + */ + public replace(document: DeepPartial) { + return this.stage(Query, "replace", undefined, document); + } + + /** + * Delete selected documents + * + * @returns Number of deleted documents + */ + public delete() { + return this.stage(Query, "delete"); + } + + /** + * Turns this selection into a change feed + * + * @returns Change feed + */ + public changes() { + return this.stage(Query[]>, "changes"); + } +} + +/** + * Selection containing any number of elements + */ +export class Selection extends Stream { + /** + * Update fields of this selection with the given values + * + * @param document Partial document with new values + * @returns Number of modified documents + */ + public update(document: DeepPartial): Query; + public update( + document: (val: ValueProxy) => U, + ): ExtractType extends DeepPartial ? Query : never; + public update(document: DeepPartial | ((val: ValueProxy) => unknown)) { + if (typeof document === "function") { + return this.stage( + Query, + "update", + undefined, + this.callfunc(document, ValueProxy), + ); + } + return this.stage(Query, "update", undefined, document); + } + + /** + * Replace documents of this selection + * + * @param document Partial document to replace with + * @returns Number of modified documents + */ + public replace(document: DeepPartial) { + return this.stage(Query, "replace", undefined, document); + } + + /** + * Delete selected documents + * + * @returns Number of deleted documents + */ + public delete() { + return this.stage(Query, "delete"); + } +} + +/** + * Database table + */ +export class Table extends Selection { + /** Atomically checks one record's revision and patches or deletes it without upsert. */ + public atomicMutation( + key: string, + request: AtomicMutation, + ): Query { + ValidateAtomicMutationTable(this.stages); + ValidateAtomicMutation(key, request); + return this.stage( + AtomicMutationQuery, + "atomicMutation", + undefined, + key, + request, + ); + } + + /** + * Inserts one or more documents into this table + * + * @param obj Document(s) to insert + * @returns Inserted IDs + */ + public insert( + obj: DeepPartial | DeepPartial[], + options?: InsertOptions, + ) { + return this.stage(Query, "insert", options, obj); + } + + /** + * Gets a document using its primary key + * + * @param key Primary key value + * @returns Single document selection + */ + public get(key: ValueProxyOrValue) { + return this.stage(SingleSelection, "get", undefined, key); + } + + /** + * Gets multiple documents using a secondary index + * + * @param keys Key value(s) + * @param index Secondary index, will use the primary key if undefined + * @returns Multiple document selection + */ + public getAll( + keys: ValueProxyOrValue | ValueProxyOrValue[], + index?: string, + ) { + return this.stage(Selection, "getAll", { index }, keys); + } + + /** + * Gets multiple documents using a secondary index and bounding values + * + * @param index Secondary index + * @param low Lowest value of the range + * @param high Highest value of the range (excluded) + * @returns Multiple document selection + */ + public between(index: TK, low: T[TK], high: T[TK]) { + return this.stage(Selection, "between", { index }, low, high); + } +} diff --git a/src/staged-query/stream.ts b/src/staged-query/stream.ts new file mode 100644 index 0000000..c9636ad --- /dev/null +++ b/src/staged-query/stream.ts @@ -0,0 +1,324 @@ +import { Datum } from "./datum"; +import { Query } from "./query"; +import type { Changes, ExtractType, Value } from "./common"; +import type { Selection } from "./selection"; +import { ValueProxy, type ValueProxyOrValue } from "./valueproxy"; + +export class Stream extends Query { + /** + * Changes the type of the value in this stream. + * This does not actually perform any conversion, it only changes the typescript type. + * + * @returns Same stream with a different type + */ + public cast() { + return this as unknown as Stream; + } + + /** + * Indexes the stream value. + * + * TODO: Better name? + * + * @param key Field name + * @param def Default value + * @returns New stream + */ + public key(key: K, def?: U) { + return this.stage( + Stream | U>, + "key", + undefined, + key, + def, + ); + } + + /** + * Defaults the stream value to a given value if it is null. + * + * @param value Default value + * @returns Stream with non-null value + */ + public default(val: Value) { + return this.stage( + Stream | U>, + "default", + undefined, + val, + ); + } + + /** + * Maps the array values using a mapping function. + * + * @param mapper Mapping function + * @returns New stream + */ + public map(mapper: (val: ValueProxy) => U) { + return this.stage( + Stream>, + "map", + undefined, + this.callfunc(mapper, ValueProxy), + ); + } + + /** + * Filters the array using a predicate function. + * + * @param predicate Predicate function. + * @returns Filtered stream + */ + public filter(predicate: (val: ValueProxy) => ValueProxyOrValue) { + return this.stage( + undefined, + "filter", + undefined, + this.callfunc(predicate, ValueProxy), + ); + } + + /** + * Selects specific fields in the documents, discarding the rest. + * + * @param fields Selected fields + * @returns New stream + */ + public pluck(...fields: string[]) { + return this.stage(Stream>, "pluck", undefined, fields); + } + + /** + * Excludes specific fields in the documents + * + * @param fields Excluded fields + * @returns New stream + */ + public without(...fields: string[]) { + return this.stage(Stream>, "without", undefined, fields); + } + + /** + * Concatenates the results of another stream into this stream without deduplication + * + * @param other Stream to concatenate + * @returns New stream containing elements from both streams + */ + public union(other: Stream) { + return this.stage(Stream, "union", undefined, other); + } + + /** + * Perform a left join operation between this stream (left) and another stream (right) + * + * @param right Right stream + * @param predicate Predicate to match elements from the left stream to the right stream + * @param mapper Mapping function for each pair of documents + * @returns New stream with results of the mapping function + */ + public join( + right: Stream, + predicate: ( + left: ValueProxy, + right: ValueProxy, + ) => ValueProxyOrValue, + mapper: (left: ValueProxy, right: ValueProxy) => V, + ) { + return this.stage( + Stream>, + "join", + { innerOnly: false }, + right, + this.callfunc(predicate, ValueProxy, ValueProxy), + this.callfunc(mapper, ValueProxy, ValueProxy), + ); + } + + /** + * Perform an inner join operation between this stream (left) and another stream (right) + * + * @param right Right stream + * @param predicate Predicate to match elements from the left stream to the right stream + * @param mapper Mapping function for each pair of documents + * @returns New stream with results of the mapping function + */ + public joinInner( + right: Stream, + predicate: ( + left: ValueProxy, + right: ValueProxy, + ) => ValueProxyOrValue, + mapper: (left: ValueProxy, right: ValueProxy) => V, + ) { + return this.stage( + Stream>, + "join", + { innerOnly: true }, + right, + this.callfunc(predicate, ValueProxy, ValueProxy), + this.callfunc(mapper, ValueProxy, ValueProxy), + ); + } + + /** + * Transform a foreign key or array of foreign keys into a document from another stream + * + * @param right Stream containing the other documents + * @param localKey Key in the local document to search with and replace + * @param otherKey Key in the other document to match against + * @returns New stream + */ + public lookup( + right: Selection, + localKey: TK, + otherKey: keyof U, + ) { + return this.stage( + Stream & Record>, + "lookup", + { localKey, otherKey }, + right, + ); + } + + /** + * Group the documents using the given index and maps the result using a mapping function + * + * The parameters of this function are: + * - The stream with all the documents inside the group + * - The index value for this group + * + * The result of this function is used as the element in the new stream + * + * @param index Index to group on + * @param mapper Mapping function + * @returns New stream of grouped data + */ + public group( + index: K, + mapper: ( + stream: Stream, + group: ValueProxy, + ) => U, + ) { + return this.stage( + Stream>, + "group", + { index }, + this.callfunc(mapper, Stream, ValueProxy), + ); + } + + /** + * Sort the stream using the given index and direction + * + * @param index Index to sort + * @param direction Sort direction + * @returns New (sorted) stream + */ + public orderBy(index: string, direction?: "asc" | "desc") { + return this.stage(undefined, "orderBy", { index, direction }); + } + + /** + * Obtain a slice (subsection) of the stream + * + * @param offset Offset into the stream + * @param count Number of documents to pick + * @returns New stream + */ + public slice(offset: Value, count?: Value) { + return this.stage(undefined, "slice", undefined, offset, count); + } + + /** + * Obtain the Nth document of the stream + * + * @param n N + * @returns Single document + */ + public nth(n: Value) { + return this.stage(Datum, "nth", undefined, n); + } + + /** + * Gets the count of documents or the count of distinct values of a given field + * + * @param field Field to count distinct entries + * @returns Count + */ + public count(field?: keyof T) { + return this.stage(Datum, "count", { field }); + } + + /** + * Sum of the values on the given field + * + * @param field Field to use + * @returns Sum + */ + public sum(field?: keyof T) { + return this.stage(Datum, "sum", { field }); + } + + /** + * Average of the values on the given field + * + * @param field Field to use + * @returns Average + */ + public avg(field?: keyof T) { + return this.stage(Datum, "avg", { field }); + } + + /** + * Minimum of the values on the given field + * + * @param field Field to use + * @returns Minimum value + */ + public min(field?: keyof T) { + return this.stage(Datum, "min", { field }); + } + + /** + * Maximum of the values on the given field + * + * @param field Field to use + * @returns Maximum value + */ + public max(field?: keyof T) { + return this.stage(Datum, "max", { field }); + } + + /** + * Gets an array of distinct documents in the stream + * + * @returns Array of documents + */ + public distinct(): Datum; + public distinct(field: undefined): Datum; + + /** + * Gets a stream of the distinct values of a field + * + * @param index Field to use + * @returns New stream + */ + public distinct(field: TK): Stream; + public distinct(field?: keyof T) { + return this.stage(field ? Stream : Datum, "distinct", { + field, + }); + } + + /** + * Turns this stream into a change feed + * + * @returns Change feed + */ + public changes() { + return this.stage(Query[]>, "changes"); + } +} diff --git a/src/staged-query/valueproxy.ts b/src/staged-query/valueproxy.ts new file mode 100644 index 0000000..bac97da --- /dev/null +++ b/src/staged-query/valueproxy.ts @@ -0,0 +1,804 @@ +import { QueryStage, StagedObject, type ExtractType } from "./common"; + +export type ValueProxyOrValue = ValueProxy | T; + +export type Is = Left extends Right ? R : never; + +export type ArrayValue = T extends (infer V)[] ? V : never; + +export type IsArray = Right extends (infer A)[] + ? Is + : never; + +export type OnlyObject = T extends Record ? T : never; + +/** + * Proxy to an actual value in the database + * + * Actions performed on this proxy are not executed right away but instead recorded for the actual query + */ +export class ValueProxy extends StagedObject { + //@internal + public static arg(id: number) { + return new ValueProxy(QueryStage("arg", undefined, id)); + } + + /** + * Create a new proxy with a constant value + * + * @param value Value + * @returns Proxy + */ + public static constant(value: T) { + return new ValueProxy(QueryStage("constant", undefined, value)); + } + + /** + * Changes the type of this proxy. + * This does not actually perform any conversion, it only changes the typescript type. + * + * @returns Same proxy with a different type + */ + public cast() { + return this as unknown as ValueProxy; + } + + //#region Any + + /** + * Returns the parameter if the proxy is null. + * + * @param value Value to use in case the proxy is null + * @returns Non-null value. + */ + public default(value: ValueProxyOrValue) { + return this.stage( + ValueProxy | U>, + "default", + undefined, + value, + ); + } + + /** + * AND operator. + * + * @param other Operand B + * @returns A && B + */ + public and(value: unknown) { + return this.stage(ValueProxy, "and", undefined, value); + } + + /** + * OR operator. + * + * @param other Operand B + * @returns A || B + */ + public or(value: unknown) { + return this.stage(ValueProxy, "or", undefined, value); + } + + /** + * NOT operator. + * + * @returns !A + */ + public not() { + return this.stage(ValueProxy, "not"); + } + + /** + * Equality operator. + * + * @param other Operand B + * @returns A == B + */ + public eq(value: unknown) { + return this.stage(ValueProxy, "eq", undefined, value); + } + + /** + * Inequality operator. + * + * @param other Operand B + * @returns A != B + */ + public ne(value: unknown) { + return this.stage(ValueProxy, "ne", undefined, value); + } + + //#endregion + + //#region Date & number arithmethic + + /** + * Addition operator. + * + * @param other Operand B + * @returns New value + */ + public add( + this: Is, + value: ValueProxyOrValue, + ) { + return (>this).stage( + ValueProxy, + "add", + undefined, + value, + ); + } + + /** + * Subtraction operator. + * + * @param other Operand B + * @returns New value + */ + public sub(this: Is, value: ValueProxyOrValue) { + return (>this).stage( + ValueProxy< + Date extends U + ? + | (Date extends T ? number : never) + | (number extends U ? (T extends number ? number : Date) : never) + : Extract + >, + "sub", + undefined, + value, + ); + } + + //#endregion + + //#region Date + + /** + * Determines whether or not the proxy is between the two bounds. + * + * @param left Left bound (inclusive) + * @param right Right bound (exclusive) + * @returns True if the date is within the bounds + */ + public during( + this: Is, + left: ValueProxyOrValue, + right: ValueProxyOrValue, + ) { + return this.stage( + ValueProxy, + "date_during", + undefined, + left, + right, + ); + } + + //@internal + private withTimezone(timezone?: ValueProxyOrValue) { + if (timezone) { + return this.stage( + ValueProxy, + "date_with_timezone", + undefined, + timezone, + ); + } + return this.cast(); + } + + /** + * Number of seconds since the start of the day. + * + * @returns Seconds + */ + public timeofday( + this: Is, + timezone?: ValueProxyOrValue, + ) { + return this.withTimezone(timezone).stage(ValueProxy, "date_tod"); + } + + /** + * Year. + * + * @returns Year + */ + public year(this: Is, timezone?: ValueProxyOrValue) { + return this.withTimezone(timezone).stage(ValueProxy, "date_year"); + } + + /** + * Month. + * + * @returns Month + */ + public month(this: Is, timezone?: ValueProxyOrValue) { + return this.withTimezone(timezone).stage(ValueProxy, "date_month"); + } + + /** + * Day of the month. + * + * @returns Day of the month + */ + public day(this: Is, timezone?: ValueProxyOrValue) { + return this.withTimezone(timezone).stage(ValueProxy, "date_day"); + } + + /** + * Day of the week. + * + * @returns Day of the week + */ + public dayofweek( + this: Is, + timezone?: ValueProxyOrValue, + ) { + return this.withTimezone(timezone).stage(ValueProxy, "date_dow"); + } + + /** + * Day of the year. + * + * @returns Day of the year + */ + public dayofyear( + this: Is, + timezone?: ValueProxyOrValue, + ) { + return this.withTimezone(timezone).stage(ValueProxy, "date_doy"); + } + + /** + * Hour of the day. + * + * @returns Hours + */ + public hours(this: Is, timezone?: ValueProxyOrValue) { + return this.withTimezone(timezone).stage(ValueProxy, "date_hours"); + } + + /** + * Minutes. + * + * @returns Minutes + */ + public minutes( + this: Is, + timezone?: ValueProxyOrValue, + ) { + return this.withTimezone(timezone).stage( + ValueProxy, + "date_minutes", + ); + } + + /** + * Seconds. + * + * @returns Seconds + */ + public seconds( + this: Is, + timezone?: ValueProxyOrValue, + ) { + return this.withTimezone(timezone).stage( + ValueProxy, + "date_seconds", + ); + } + + /** + * Seconds since the UNIX epoch with millisecond precision. + * + * @returns Seconds + */ + public epoch(this: Is) { + return this.stage(ValueProxy, "date_epoch"); + } + + //#endregion + + //#region Number + + /** + * Multiplication operator. + * + * @param other Operand B + * @returns A * B + */ + public mul(this: Is, other: ValueProxyOrValue) { + return this.stage(ValueProxy, "mul", undefined, other); + } + + /** + * Division operator. + * + * @param other Operand B + * @returns A / B + */ + public div(this: Is, other: ValueProxyOrValue) { + return this.stage(ValueProxy, "div", undefined, other); + } + + /** + * Modulo operator. + * + * @param other Operand B + * @returns A % B + */ + public mod(this: Is, other: ValueProxyOrValue) { + return this.stage(ValueProxy, "mod", undefined, other); + } + + /** + * Round to the nearest integer. + * + * @returns Integer + */ + public round(this: Is) { + return this.stage(ValueProxy, "round"); + } + + /** + * Round to the higher integer. + * + * @returns Integer + */ + public ceil(this: Is) { + return this.stage(ValueProxy, "ceil"); + } + + /** + * Round to the lower integer. + * + * @returns Integer + */ + public floor(this: Is) { + return this.stage(ValueProxy, "floor"); + } + + /** + * Bitwise AND operator. + * + * @param other Operand B + * @return A & B + */ + public band(this: Is, other: ValueProxyOrValue) { + return this.stage(ValueProxy, "bit_and", undefined, other); + } + + /** + * Bitwise OR operator. + * + * @param other Operand B + * @return A | B + */ + public bor(this: Is, other: ValueProxyOrValue) { + return this.stage(ValueProxy, "bit_or", undefined, other); + } + + /** + * Bitwise XOR operator. + * + * @param other Operand B + * @return A ^ B + */ + public bxor(this: Is, other: ValueProxyOrValue) { + return this.stage(ValueProxy, "bit_xor", undefined, other); + } + + /** + * Bitwise NOT operator. + * + * @return ~A + */ + public bnot(this: Is) { + return this.stage(ValueProxy, "bit_not"); + } + + /** + * Bitwise left shift operator. + * + * @param other Operand B + * @returns A << B + */ + public blshift(this: Is, other: ValueProxyOrValue) { + return this.stage(ValueProxy, "bit_lshift", undefined, other); + } + + /** + * Bitwise right shift operator. + * + * @param other Operand B + * @returns A >> B + */ + public brshift(this: Is, other: ValueProxyOrValue) { + return this.stage(ValueProxy, "bit_rshift", undefined, other); + } + + /** + * Bitwise right shift operator with sign preservation. + * + * @param other Operand B + * @returns A >> B (sign-preserving) + */ + public brshiftPreserveSign( + this: Is, + other: ValueProxyOrValue, + ) { + return this.stage( + ValueProxy, + "bit_rshift", + { preserveSign: true }, + other, + ); + } + + //#endregion + + //#region Comparison + + /** + * Greater-than operator. + * + * @param other Operand B + * @returns A > B + */ + public gt( + this: Is, + other: ValueProxyOrValue, + ) { + return (>this).stage( + ValueProxy, + "cmp_gt", + undefined, + other, + ); + } + + /** + * Greater-or-equal operator. + * + * @param other Operand B + * @returns A >= B + */ + public ge( + this: Is, + other: ValueProxyOrValue, + ) { + return (>this).stage( + ValueProxy, + "cmp_ge", + undefined, + other, + ); + } + + /** + * Lesser-than operator. + * + * @param other Operand B + * @returns A < B + */ + public lt( + this: Is, + other: ValueProxyOrValue, + ) { + return (>this).stage( + ValueProxy, + "cmp_lt", + undefined, + other, + ); + } + + /** + * Lesser-or-equal operator. + * + * @param other Operand B + * @returns A <= B + */ + public le( + this: Is, + other: ValueProxyOrValue, + ) { + return (>this).stage( + ValueProxy, + "cmp_le", + undefined, + other, + ); + } + + //#endregion + + //#region String + + /** + * Splits the string using a separator. + * + * @param separator Separator string + * @param maxSplits Maximum number of results + * @returns Array of sub-strings + */ + public split( + this: Is, + separator?: string, + maxSplits?: number, + ) { + return this.stage(ValueProxy, "str_split", { + separator, + maxSplits, + }); + } + + /** + * Concatenate the string with another. + * + * @param other Second string + * @returns Concatenated string + */ + public concat(this: Is, other: ValueProxyOrValue) { + return this.stage(ValueProxy, "str_concat", undefined, other); + } + + /** + * Converts the string to all upper case. + * + * @returns New string + */ + public upcase(this: Is) { + return this.stage(ValueProxy, "str_upcase"); + } + + /** + * Converts the string to all lower case. + * + * @returns New string + */ + public downcase(this: Is) { + return this.stage(ValueProxy, "str_downcase"); + } + + /** + * Gets the number of Unicode codepoints in the string. + * + * @returns Number of codepoints + */ + public strlen(this: Is) { + return this.stage(ValueProxy, "str_len"); + } + + /** + * Checks if the string matches a regex. + * + * @param regex Regex + * @returns True if the string matched + */ + public match(this: Is, regex: ValueProxyOrValue) { + return this.stage(ValueProxy, "str_match", undefined, regex); + } + + //#endregion + + //#region Array + + /** + * Indexes the array with a given number position + * + * @param key Position in the array + * @returns Element at the given position + */ + public index( + this: IsArray, + key: ValueProxyOrValue, + ) { + return this.stage(ValueProxy>, "arr_index", undefined, key); + } + + /** + * Tests if the arrays contains a value. + * + * @param val Value to search for. + * @returns True if the value was found. + */ + public includes( + this: IsArray, + val: ValueProxyOrValue>, + ) { + return this.stage(ValueProxy, "arr_includes", undefined, val); + } + + /** + * Returns a slice of the array. + * + * @param start Start index (inclusive, 0-indexed) + * @param end End index (exclusive) + * @returns Sub-array + */ + public slice( + this: IsArray, + start: ValueProxyOrValue, + end?: ValueProxyOrValue, + ) { + return this.stage( + ValueProxy>>, + "arr_slice", + undefined, + start, + end, + ); + } + + /** + * Maps the array values using a mapping function. + * + * @param mapper Mapping function + * @returns New array + */ + public map( + this: IsArray, + mapper: (val: ValueProxy>) => U, + ) { + return this.stage( + ValueProxy[]>, + "arr_map", + undefined, + this.callfunc(mapper, ValueProxy>), + ); + } + + /** + * Filters the array using a predicate function. + * + * @param predicate Predicate function. + * @returns Filtered array + */ + public filter( + this: IsArray, + predicate: (val: ValueProxy>) => ValueProxyOrValue, + ) { + return this.stage( + ValueProxy>>, + "arr_filter", + undefined, + this.callfunc(predicate, ValueProxy>), + ); + } + + /** + * Checks if the array is empty. + * + * @returns True if the array is empty + */ + public isempty(this: IsArray) { + return this.stage(ValueProxy, "arr_empty"); + } + + /** + * Gets the length of the array. + * + * @returns Length + */ + public count(this: IsArray) { + return this.stage(ValueProxy, "arr_count"); + } + + /** + * Gets the sum of a number array. + * + * @returns Sum + */ + public sum(this: IsArray) { + return (>this).stage(ValueProxy, "arr_sum"); + } + + /** + * Gets the average of a number array. + * + * @returns Average + */ + public avg(this: IsArray) { + return (>this).stage(ValueProxy, "arr_avg"); + } + + /** + * Gets the minimum of a number array. + * + * @returns Minimum + */ + public min(this: IsArray) { + return (>this).stage(ValueProxy, "arr_min"); + } + + /** + * Gets the maxmimum of a number array. + * + * @returns Maximum + */ + public max(this: IsArray) { + return (>this).stage(ValueProxy, "arr_max"); + } + + //#endregion + + //#region Object + + /** + * Indexes the object. + * + * TODO: Better name? + * + * @param key Field name + * @param def Default value + * @returns Value of the field + */ + public key, K extends keyof TO = keyof TO, U = undefined>( + this: Is, this>, + key: ValueProxyOrValue, + def?: ValueProxyOrValue, + ) { + return this.stage( + ValueProxy< + U extends undefined ? TO[K] : Exclude | U + >, + "obj_index", + undefined, + key, + def, + ); + } + + /** + * Merges the object with another. + * + * @param value Other object + * @returns `{...A, ...B}` + */ + public merge>( + this: Is, this>, + other: ValueProxyOrValue, + ) { + return this.stage( + ValueProxy & ExtractType>, + "obj_merge", + undefined, + other, + ); + } + + /** + * Gets the keys of the object as an array. + * + * @returns Array of keys + */ + public keys>(this: Is, this>) { + return this.stage(ValueProxy>, "obj_keys"); + } + + /** + * Gets the values of the object as an array. + * + * @returns Array of values + */ + public values>(this: Is, this>) { + return this.stage(ValueProxy>, "obj_values"); + } + + /** + * Tests if the object has the specified fields. + * + * @param fields field list + * @returns True if the object matches + */ + public hasfields(this: Is, this>, ...fields: string[]) { + return this.stage(ValueProxy, "obj_has", undefined, fields); + } + + //#endregion +} diff --git a/src/stream.ts b/src/stream.ts index 88ce743..9323d25 100644 --- a/src/stream.ts +++ b/src/stream.ts @@ -1,324 +1 @@ -import { Datum } from "./datum"; -import { Query } from "./query"; -import type { Selection } from "./selection"; -import type { Changes, ExtractType, Value } from "./common"; -import { ValueProxy, type ValueProxyOrValue } from "./valueproxy"; - -export class Stream extends Query { - /** - * Changes the type of the value in this stream. - * This does not actually perform any conversion, it only changes the typescript type. - * - * @returns Same stream with a different type - */ - public cast() { - return this as unknown as Stream; - } - - /** - * Indexes the stream value. - * - * TODO: Better name? - * - * @param key Field name - * @param def Default value - * @returns New stream - */ - public key(key: K, def?: U) { - return this.stage( - Stream | U>, - "key", - undefined, - key, - def, - ); - } - - /** - * Defaults the stream value to a given value if it is null. - * - * @param value Default value - * @returns Stream with non-null value - */ - public default(val: Value) { - return this.stage( - Stream | U>, - "default", - undefined, - val, - ); - } - - /** - * Maps the array values using a mapping function. - * - * @param mapper Mapping function - * @returns New stream - */ - public map(mapper: (val: ValueProxy) => U) { - return this.stage( - Stream>, - "map", - undefined, - this.callfunc(mapper, ValueProxy), - ); - } - - /** - * Filters the array using a predicate function. - * - * @param predicate Predicate function. - * @returns Filtered stream - */ - public filter(predicate: (val: ValueProxy) => ValueProxyOrValue) { - return this.stage( - undefined, - "filter", - undefined, - this.callfunc(predicate, ValueProxy), - ); - } - - /** - * Selects specific fields in the documents, discarding the rest. - * - * @param fields Selected fields - * @returns New stream - */ - public pluck(...fields: string[]) { - return this.stage(Stream>, "pluck", undefined, fields); - } - - /** - * Excludes specific fields in the documents - * - * @param fields Excluded fields - * @returns New stream - */ - public without(...fields: string[]) { - return this.stage(Stream>, "without", undefined, fields); - } - - /** - * Concatenates the results of another stream into this stream without deduplication - * - * @param other Stream to concatenate - * @returns New stream containing elements from both streams - */ - public union(other: Stream) { - return this.stage(Stream, "union", undefined, other); - } - - /** - * Perform a left join operation between this stream (left) and another stream (right) - * - * @param right Right stream - * @param predicate Predicate to match elements from the left stream to the right stream - * @param mapper Mapping function for each pair of documents - * @returns New stream with results of the mapping function - */ - public join( - right: Stream, - predicate: ( - left: ValueProxy, - right: ValueProxy, - ) => ValueProxyOrValue, - mapper: (left: ValueProxy, right: ValueProxy) => V, - ) { - return this.stage( - Stream>, - "join", - { innerOnly: false }, - right, - this.callfunc(predicate, ValueProxy, ValueProxy), - this.callfunc(mapper, ValueProxy, ValueProxy), - ); - } - - /** - * Perform an inner join operation between this stream (left) and another stream (right) - * - * @param right Right stream - * @param predicate Predicate to match elements from the left stream to the right stream - * @param mapper Mapping function for each pair of documents - * @returns New stream with results of the mapping function - */ - public joinInner( - right: Stream, - predicate: ( - left: ValueProxy, - right: ValueProxy, - ) => ValueProxyOrValue, - mapper: (left: ValueProxy, right: ValueProxy) => V, - ) { - return this.stage( - Stream>, - "join", - { innerOnly: true }, - right, - this.callfunc(predicate, ValueProxy, ValueProxy), - this.callfunc(mapper, ValueProxy, ValueProxy), - ); - } - - /** - * Transform a foreign key or array of foreign keys into a document from another stream - * - * @param right Stream containing the other documents - * @param localKey Key in the local document to search with and replace - * @param otherKey Key in the other document to match against - * @returns New stream - */ - public lookup( - right: Selection, - localKey: TK, - otherKey: keyof U, - ) { - return this.stage( - Stream & Record>, - "lookup", - { localKey, otherKey }, - right, - ); - } - - /** - * Group the documents using the given index and maps the result using a mapping function - * - * The parameters of this function are: - * - The stream with all the documents inside the group - * - The index value for this group - * - * The result of this function is used as the element in the new stream - * - * @param index Index to group on - * @param mapper Mapping function - * @returns New stream of grouped data - */ - public group( - index: K, - mapper: ( - stream: Stream, - group: ValueProxy, - ) => U, - ) { - return this.stage( - Stream>, - "group", - { index }, - this.callfunc(mapper, Stream, ValueProxy), - ); - } - - /** - * Sort the stream using the given index and direction - * - * @param index Index to sort - * @param direction Sort direction - * @returns New (sorted) stream - */ - public orderBy(index: string, direction?: "asc" | "desc") { - return this.stage(undefined, "orderBy", { index, direction }); - } - - /** - * Obtain a slice (subsection) of the stream - * - * @param offset Offset into the stream - * @param count Number of documents to pick - * @returns New stream - */ - public slice(offset: Value, count?: Value) { - return this.stage(undefined, "slice", undefined, offset, count); - } - - /** - * Obtain the Nth document of the stream - * - * @param n N - * @returns Single document - */ - public nth(n: Value) { - return this.stage(Datum, "nth", undefined, n); - } - - /** - * Gets the count of documents or the count of distinct values of a given field - * - * @param field Field to count distinct entries - * @returns Count - */ - public count(field?: keyof T) { - return this.stage(Datum, "count", { field }); - } - - /** - * Sum of the values on the given field - * - * @param field Field to use - * @returns Sum - */ - public sum(field?: keyof T) { - return this.stage(Datum, "sum", { field }); - } - - /** - * Average of the values on the given field - * - * @param field Field to use - * @returns Average - */ - public avg(field?: keyof T) { - return this.stage(Datum, "avg", { field }); - } - - /** - * Minimum of the values on the given field - * - * @param field Field to use - * @returns Minimum value - */ - public min(field?: keyof T) { - return this.stage(Datum, "min", { field }); - } - - /** - * Maximum of the values on the given field - * - * @param field Field to use - * @returns Maximum value - */ - public max(field?: keyof T) { - return this.stage(Datum, "max", { field }); - } - - /** - * Gets an array of distinct documents in the stream - * - * @returns Array of documents - */ - public distinct(): Datum; - public distinct(field: undefined): Datum; - - /** - * Gets a stream of the distinct values of a field - * - * @param index Field to use - * @returns New stream - */ - public distinct(field: TK): Stream; - public distinct(field?: keyof T) { - return this.stage(field ? Stream : Datum, "distinct", { - field, - }); - } - - /** - * Turns this stream into a change feed - * - * @returns Change feed - */ - public changes() { - return this.stage(Query[]>, "changes"); - } -} +export { Stream } from "./staged-query/stream"; diff --git a/src/valueproxy.ts b/src/valueproxy.ts index 2125996..47153e3 100644 --- a/src/valueproxy.ts +++ b/src/valueproxy.ts @@ -1,804 +1 @@ -import { type ExtractType, QueryStage, StagedObject } from "./common"; - -export type ValueProxyOrValue = ValueProxy | T; - -export type Is = Left extends Right ? R : never; - -export type ArrayValue = T extends (infer V)[] ? V : never; - -export type IsArray = Right extends (infer A)[] - ? Is - : never; - -export type OnlyObject = T extends Record ? T : never; - -/** - * Proxy to an actual value in the database - * - * Actions performed on this proxy are not executed right away but instead recorded for the actual query - */ -export class ValueProxy extends StagedObject { - //@internal - public static arg(id: number) { - return new ValueProxy(QueryStage("arg", undefined, id)); - } - - /** - * Create a new proxy with a constant value - * - * @param value Value - * @returns Proxy - */ - public static constant(value: T) { - return new ValueProxy(QueryStage("constant", undefined, value)); - } - - /** - * Changes the type of this proxy. - * This does not actually perform any conversion, it only changes the typescript type. - * - * @returns Same proxy with a different type - */ - public cast() { - return this as unknown as ValueProxy; - } - - //#region Any - - /** - * Returns the parameter if the proxy is null. - * - * @param value Value to use in case the proxy is null - * @returns Non-null value. - */ - public default(value: ValueProxyOrValue) { - return this.stage( - ValueProxy | U>, - "default", - undefined, - value, - ); - } - - /** - * AND operator. - * - * @param other Operand B - * @returns A && B - */ - public and(value: unknown) { - return this.stage(ValueProxy, "and", undefined, value); - } - - /** - * OR operator. - * - * @param other Operand B - * @returns A || B - */ - public or(value: unknown) { - return this.stage(ValueProxy, "or", undefined, value); - } - - /** - * NOT operator. - * - * @returns !A - */ - public not() { - return this.stage(ValueProxy, "not"); - } - - /** - * Equality operator. - * - * @param other Operand B - * @returns A == B - */ - public eq(value: unknown) { - return this.stage(ValueProxy, "eq", undefined, value); - } - - /** - * Inequality operator. - * - * @param other Operand B - * @returns A != B - */ - public ne(value: unknown) { - return this.stage(ValueProxy, "ne", undefined, value); - } - - //#endregion - - //#region Date & number arithmethic - - /** - * Addition operator. - * - * @param other Operand B - * @returns New value - */ - public add( - this: Is, - value: ValueProxyOrValue, - ) { - return (>this).stage( - ValueProxy, - "add", - undefined, - value, - ); - } - - /** - * Subtraction operator. - * - * @param other Operand B - * @returns New value - */ - public sub(this: Is, value: ValueProxyOrValue) { - return (>this).stage( - ValueProxy< - Date extends U - ? - | (Date extends T ? number : never) - | (number extends U ? (T extends number ? number : Date) : never) - : Extract - >, - "sub", - undefined, - value, - ); - } - - //#endregion - - //#region Date - - /** - * Determines whether or not the proxy is between the two bounds. - * - * @param left Left bound (inclusive) - * @param right Right bound (exclusive) - * @returns True if the date is within the bounds - */ - public during( - this: Is, - left: ValueProxyOrValue, - right: ValueProxyOrValue, - ) { - return this.stage( - ValueProxy, - "date_during", - undefined, - left, - right, - ); - } - - //@internal - private withTimezone(timezone?: ValueProxyOrValue) { - if (timezone) { - return this.stage( - ValueProxy, - "date_with_timezone", - undefined, - timezone, - ); - } - return this.cast(); - } - - /** - * Number of seconds since the start of the day. - * - * @returns Seconds - */ - public timeofday( - this: Is, - timezone?: ValueProxyOrValue, - ) { - return this.withTimezone(timezone).stage(ValueProxy, "date_tod"); - } - - /** - * Year. - * - * @returns Year - */ - public year(this: Is, timezone?: ValueProxyOrValue) { - return this.withTimezone(timezone).stage(ValueProxy, "date_year"); - } - - /** - * Month. - * - * @returns Month - */ - public month(this: Is, timezone?: ValueProxyOrValue) { - return this.withTimezone(timezone).stage(ValueProxy, "date_month"); - } - - /** - * Day of the month. - * - * @returns Day of the month - */ - public day(this: Is, timezone?: ValueProxyOrValue) { - return this.withTimezone(timezone).stage(ValueProxy, "date_day"); - } - - /** - * Day of the week. - * - * @returns Day of the week - */ - public dayofweek( - this: Is, - timezone?: ValueProxyOrValue, - ) { - return this.withTimezone(timezone).stage(ValueProxy, "date_dow"); - } - - /** - * Day of the year. - * - * @returns Day of the year - */ - public dayofyear( - this: Is, - timezone?: ValueProxyOrValue, - ) { - return this.withTimezone(timezone).stage(ValueProxy, "date_doy"); - } - - /** - * Hour of the day. - * - * @returns Hours - */ - public hours(this: Is, timezone?: ValueProxyOrValue) { - return this.withTimezone(timezone).stage(ValueProxy, "date_hours"); - } - - /** - * Minutes. - * - * @returns Minutes - */ - public minutes( - this: Is, - timezone?: ValueProxyOrValue, - ) { - return this.withTimezone(timezone).stage( - ValueProxy, - "date_minutes", - ); - } - - /** - * Seconds. - * - * @returns Seconds - */ - public seconds( - this: Is, - timezone?: ValueProxyOrValue, - ) { - return this.withTimezone(timezone).stage( - ValueProxy, - "date_seconds", - ); - } - - /** - * Seconds since the UNIX epoch with millisecond precision. - * - * @returns Seconds - */ - public epoch(this: Is) { - return this.stage(ValueProxy, "date_epoch"); - } - - //#endregion - - //#region Number - - /** - * Multiplication operator. - * - * @param other Operand B - * @returns A * B - */ - public mul(this: Is, other: ValueProxyOrValue) { - return this.stage(ValueProxy, "mul", undefined, other); - } - - /** - * Division operator. - * - * @param other Operand B - * @returns A / B - */ - public div(this: Is, other: ValueProxyOrValue) { - return this.stage(ValueProxy, "div", undefined, other); - } - - /** - * Modulo operator. - * - * @param other Operand B - * @returns A % B - */ - public mod(this: Is, other: ValueProxyOrValue) { - return this.stage(ValueProxy, "mod", undefined, other); - } - - /** - * Round to the nearest integer. - * - * @returns Integer - */ - public round(this: Is) { - return this.stage(ValueProxy, "round"); - } - - /** - * Round to the higher integer. - * - * @returns Integer - */ - public ceil(this: Is) { - return this.stage(ValueProxy, "ceil"); - } - - /** - * Round to the lower integer. - * - * @returns Integer - */ - public floor(this: Is) { - return this.stage(ValueProxy, "floor"); - } - - /** - * Bitwise AND operator. - * - * @param other Operand B - * @return A & B - */ - public band(this: Is, other: ValueProxyOrValue) { - return this.stage(ValueProxy, "bit_and", undefined, other); - } - - /** - * Bitwise OR operator. - * - * @param other Operand B - * @return A | B - */ - public bor(this: Is, other: ValueProxyOrValue) { - return this.stage(ValueProxy, "bit_or", undefined, other); - } - - /** - * Bitwise XOR operator. - * - * @param other Operand B - * @return A ^ B - */ - public bxor(this: Is, other: ValueProxyOrValue) { - return this.stage(ValueProxy, "bit_xor", undefined, other); - } - - /** - * Bitwise NOT operator. - * - * @return ~A - */ - public bnot(this: Is) { - return this.stage(ValueProxy, "bit_not"); - } - - /** - * Bitwise left shift operator. - * - * @param other Operand B - * @returns A << B - */ - public blshift(this: Is, other: ValueProxyOrValue) { - return this.stage(ValueProxy, "bit_lshift", undefined, other); - } - - /** - * Bitwise right shift operator. - * - * @param other Operand B - * @returns A >> B - */ - public brshift(this: Is, other: ValueProxyOrValue) { - return this.stage(ValueProxy, "bit_rshift", undefined, other); - } - - /** - * Bitwise right shift operator with sign preservation. - * - * @param other Operand B - * @returns A >> B (sign-preserving) - */ - public brshiftPreserveSign( - this: Is, - other: ValueProxyOrValue, - ) { - return this.stage( - ValueProxy, - "bit_rshift", - { preserveSign: true }, - other, - ); - } - - //#endregion - - //#region Comparison - - /** - * Greater-than operator. - * - * @param other Operand B - * @returns A > B - */ - public gt( - this: Is, - other: ValueProxyOrValue, - ) { - return (>this).stage( - ValueProxy, - "cmp_gt", - undefined, - other, - ); - } - - /** - * Greater-or-equal operator. - * - * @param other Operand B - * @returns A >= B - */ - public ge( - this: Is, - other: ValueProxyOrValue, - ) { - return (>this).stage( - ValueProxy, - "cmp_ge", - undefined, - other, - ); - } - - /** - * Lesser-than operator. - * - * @param other Operand B - * @returns A < B - */ - public lt( - this: Is, - other: ValueProxyOrValue, - ) { - return (>this).stage( - ValueProxy, - "cmp_lt", - undefined, - other, - ); - } - - /** - * Lesser-or-equal operator. - * - * @param other Operand B - * @returns A <= B - */ - public le( - this: Is, - other: ValueProxyOrValue, - ) { - return (>this).stage( - ValueProxy, - "cmp_le", - undefined, - other, - ); - } - - //#endregion - - //#region String - - /** - * Splits the string using a separator. - * - * @param separator Separator string - * @param maxSplits Maximum number of results - * @returns Array of sub-strings - */ - public split( - this: Is, - separator?: string, - maxSplits?: number, - ) { - return this.stage(ValueProxy, "str_split", { - separator, - maxSplits, - }); - } - - /** - * Concatenate the string with another. - * - * @param other Second string - * @returns Concatenated string - */ - public concat(this: Is, other: ValueProxyOrValue) { - return this.stage(ValueProxy, "str_concat", undefined, other); - } - - /** - * Converts the string to all upper case. - * - * @returns New string - */ - public upcase(this: Is) { - return this.stage(ValueProxy, "str_upcase"); - } - - /** - * Converts the string to all lower case. - * - * @returns New string - */ - public downcase(this: Is) { - return this.stage(ValueProxy, "str_downcase"); - } - - /** - * Gets the number of Unicode codepoints in the string. - * - * @returns Number of codepoints - */ - public strlen(this: Is) { - return this.stage(ValueProxy, "str_len"); - } - - /** - * Checks if the string matches a regex. - * - * @param regex Regex - * @returns True if the string matched - */ - public match(this: Is, regex: ValueProxyOrValue) { - return this.stage(ValueProxy, "str_match", undefined, regex); - } - - //#endregion - - //#region Array - - /** - * Indexes the array with a given number position - * - * @param key Position in the array - * @returns Element at the given position - */ - public index( - this: IsArray, - key: ValueProxyOrValue, - ) { - return this.stage(ValueProxy>, "arr_index", undefined, key); - } - - /** - * Tests if the arrays contains a value. - * - * @param val Value to search for. - * @returns True if the value was found. - */ - public includes( - this: IsArray, - val: ValueProxyOrValue>, - ) { - return this.stage(ValueProxy, "arr_includes", undefined, val); - } - - /** - * Returns a slice of the array. - * - * @param start Start index (inclusive, 0-indexed) - * @param end End index (exclusive) - * @returns Sub-array - */ - public slice( - this: IsArray, - start: ValueProxyOrValue, - end?: ValueProxyOrValue, - ) { - return this.stage( - ValueProxy>>, - "arr_slice", - undefined, - start, - end, - ); - } - - /** - * Maps the array values using a mapping function. - * - * @param mapper Mapping function - * @returns New array - */ - public map( - this: IsArray, - mapper: (val: ValueProxy>) => U, - ) { - return this.stage( - ValueProxy[]>, - "arr_map", - undefined, - this.callfunc(mapper, ValueProxy>), - ); - } - - /** - * Filters the array using a predicate function. - * - * @param predicate Predicate function. - * @returns Filtered array - */ - public filter( - this: IsArray, - predicate: (val: ValueProxy>) => ValueProxyOrValue, - ) { - return this.stage( - ValueProxy>>, - "arr_filter", - undefined, - this.callfunc(predicate, ValueProxy>), - ); - } - - /** - * Checks if the array is empty. - * - * @returns True if the array is empty - */ - public isempty(this: IsArray) { - return this.stage(ValueProxy, "arr_empty"); - } - - /** - * Gets the length of the array. - * - * @returns Length - */ - public count(this: IsArray) { - return this.stage(ValueProxy, "arr_count"); - } - - /** - * Gets the sum of a number array. - * - * @returns Sum - */ - public sum(this: IsArray) { - return (>this).stage(ValueProxy, "arr_sum"); - } - - /** - * Gets the average of a number array. - * - * @returns Average - */ - public avg(this: IsArray) { - return (>this).stage(ValueProxy, "arr_avg"); - } - - /** - * Gets the minimum of a number array. - * - * @returns Minimum - */ - public min(this: IsArray) { - return (>this).stage(ValueProxy, "arr_min"); - } - - /** - * Gets the maxmimum of a number array. - * - * @returns Maximum - */ - public max(this: IsArray) { - return (>this).stage(ValueProxy, "arr_max"); - } - - //#endregion - - //#region Object - - /** - * Indexes the object. - * - * TODO: Better name? - * - * @param key Field name - * @param def Default value - * @returns Value of the field - */ - public key, K extends keyof TO = keyof TO, U = undefined>( - this: Is, this>, - key: ValueProxyOrValue, - def?: ValueProxyOrValue, - ) { - return this.stage( - ValueProxy< - U extends undefined ? TO[K] : Exclude | U - >, - "obj_index", - undefined, - key, - def, - ); - } - - /** - * Merges the object with another. - * - * @param value Other object - * @returns `{...A, ...B}` - */ - public merge>( - this: Is, this>, - other: ValueProxyOrValue, - ) { - return this.stage( - ValueProxy & ExtractType>, - "obj_merge", - undefined, - other, - ); - } - - /** - * Gets the keys of the object as an array. - * - * @returns Array of keys - */ - public keys>(this: Is, this>) { - return this.stage(ValueProxy>, "obj_keys"); - } - - /** - * Gets the values of the object as an array. - * - * @returns Array of values - */ - public values>(this: Is, this>) { - return this.stage(ValueProxy>, "obj_values"); - } - - /** - * Tests if the object has the specified fields. - * - * @param fields field list - * @returns True if the object matches - */ - public hasfields(this: Is, this>, ...fields: string[]) { - return this.stage(ValueProxy, "obj_has", undefined, fields); - } - - //#endregion -} +export * from "./staged-query/valueproxy"; From e59f689b8e5d07a1a59ce98470aa31c4a2a9710d Mon Sep 17 00:00:00 2001 From: Antony Rizzitelli Date: Tue, 15 Sep 2026 16:52:17 +0000 Subject: [PATCH 2/4] fix(query): satisfy staged entry lint rules --- scripts/test-staged-query.cjs | 4 ++-- src/query.ts | 4 ++-- src/schema.ts | 6 ++++-- src/staged-query/datum.ts | 2 +- src/staged-query/query.ts | 1 + src/staged-query/stream.ts | 2 +- 6 files changed, 11 insertions(+), 8 deletions(-) diff --git a/scripts/test-staged-query.cjs b/scripts/test-staged-query.cjs index d5a99d2..d7b1cd6 100644 --- a/scripts/test-staged-query.cjs +++ b/scripts/test-staged-query.cjs @@ -1,7 +1,7 @@ -const assert = require("node:assert/strict"); const fs = require("node:fs"); -const Module = require("node:module"); +const assert = require("node:assert/strict"); const path = require("node:path"); +const Module = require("node:module"); const forbiddenImports = ["@antelopejs/interface-core", "node:async_hooks"]; const originalLoad = Module._load; diff --git a/src/query.ts b/src/query.ts index 7e086db..f3f92a7 100644 --- a/src/query.ts +++ b/src/query.ts @@ -1,7 +1,7 @@ import { InterfaceFunction } from "@antelopejs/interface-core"; -import { Query as StagedQuery } from "./staged-query/query"; import { StagedObject } from "./staged-query/common"; +import { Query as StagedQuery } from "./staged-query/query"; //@internal export const RunQuery = @@ -81,7 +81,6 @@ function run(this: StagedQuery): Promise { return RunQuery(this.build()); } -// oxlint-disable-next-line unicorn/no-thenable -- Query is deliberately PromiseLike so `await query` runs it; the contract requires this method. function then( this: StagedQuery, onfulfilled?: ((value: T) => TResult1 | PromiseLike) | null, @@ -104,6 +103,7 @@ function iterate( Object.defineProperties(StagedQuery.prototype, { run: { configurable: true, value: run, writable: true }, + // oxlint-disable-next-line unicorn/no-thenable -- Query is deliberately PromiseLike so `await query` executes it. then: { configurable: true, value: then, writable: true }, cursor: { configurable: true, value: cursor, writable: true }, [Symbol.asyncIterator]: { diff --git a/src/schema.ts b/src/schema.ts index 31d45f4..436968a 100644 --- a/src/schema.ts +++ b/src/schema.ts @@ -1,7 +1,9 @@ import { RegisteringProxy } from "@antelopejs/interface-core"; -import { Schema as StagedSchema } from "./staged-query/schema"; -import type { SchemaDefinition } from "./staged-query/schema"; +import { + Schema as StagedSchema, + type SchemaDefinition, +} from "./staged-query/schema"; //@internal export const Schemas = new RegisteringProxy< diff --git a/src/staged-query/datum.ts b/src/staged-query/datum.ts index e3f4fc0..d01b10d 100644 --- a/src/staged-query/datum.ts +++ b/src/staged-query/datum.ts @@ -1,7 +1,7 @@ import { Query } from "./query"; import { ValueProxy } from "./valueproxy"; -import type { ExtractType, Value } from "./common"; import type { Selection } from "./selection"; +import type { ExtractType, Value } from "./common"; export class Datum extends Query { /** diff --git a/src/staged-query/query.ts b/src/staged-query/query.ts index dad5a15..0279626 100644 --- a/src/staged-query/query.ts +++ b/src/staged-query/query.ts @@ -1,3 +1,4 @@ import { StagedObject } from "./common"; +// oxlint-disable-next-line typescript/no-unused-vars -- The executable entry consumes T through declaration merging. export class Query extends StagedObject {} diff --git a/src/staged-query/stream.ts b/src/staged-query/stream.ts index c9636ad..88ce743 100644 --- a/src/staged-query/stream.ts +++ b/src/staged-query/stream.ts @@ -1,7 +1,7 @@ import { Datum } from "./datum"; import { Query } from "./query"; -import type { Changes, ExtractType, Value } from "./common"; import type { Selection } from "./selection"; +import type { Changes, ExtractType, Value } from "./common"; import { ValueProxy, type ValueProxyOrValue } from "./valueproxy"; export class Stream extends Query { From 4959484ccf5aa82cb09e58e55c4dc682de5e4b11 Mon Sep 17 00:00:00 2001 From: Antony Rizzitelli Date: Tue, 15 Sep 2026 16:53:37 +0000 Subject: [PATCH 3/4] fix(test): order staged query imports --- scripts/test-staged-query.cjs | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/scripts/test-staged-query.cjs b/scripts/test-staged-query.cjs index d7b1cd6..301f68a 100644 --- a/scripts/test-staged-query.cjs +++ b/scripts/test-staged-query.cjs @@ -1,6 +1,6 @@ const fs = require("node:fs"); -const assert = require("node:assert/strict"); const path = require("node:path"); +const assert = require("node:assert/strict"); const Module = require("node:module"); const forbiddenImports = ["@antelopejs/interface-core", "node:async_hooks"]; From 4d2c7b0fcb0d961210b79e9b5b9814692b5e2914 Mon Sep 17 00:00:00 2001 From: Antony Rizzitelli Date: Tue, 15 Sep 2026 16:54:39 +0000 Subject: [PATCH 4/4] fix(test): complete import ordering --- scripts/test-staged-query.cjs | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/scripts/test-staged-query.cjs b/scripts/test-staged-query.cjs index 301f68a..504d928 100644 --- a/scripts/test-staged-query.cjs +++ b/scripts/test-staged-query.cjs @@ -1,7 +1,7 @@ const fs = require("node:fs"); const path = require("node:path"); -const assert = require("node:assert/strict"); const Module = require("node:module"); +const assert = require("node:assert/strict"); const forbiddenImports = ["@antelopejs/interface-core", "node:async_hooks"]; const originalLoad = Module._load;