Hugging Face's logo

Spaces:
teaql
/
browser-sqlite-playground
Running

App Files Community
browser-sqlite-playground / source /teaql-ts /src /sql /core.ts
philipgreat's picture
philipgreat
Deploy traceable order management playground
aa89cb5 verified
Raw History Blame Contribute Delete
97 kB
import {
observeRuntimeOperation,
RuntimeTelemetry,
RuntimeExecutionCorrelation,
RuntimeTelemetryScope,
startRuntimeOperation,
} from '../core/telemetry';
import { CheckException, EntityChecker } from '../core/checker';
import { UserContext } from '../core/context';
import { mergeRuntimeBootstrap } from '../core/runtime-module';
import type { BootstrapEntity, RuntimeBootstrap } from '../core/runtime-module';
import { contextSchemaCapability } from '../core/schema-capability';
import { OrderBy, SelectQuery } from '../core/ast';
import { GraphCommittedError, GraphMutationSession, MutationIntent, MutationRequest, QueryIntent, QueryRequest } from '../core/request-intent';
import type { QueryOperation } from '../core/request-intent';
import { canonicalSQLTracePath, cloneTraceNodes, queryTraceSource, TraceNode } from '../core/trace-chain';
import { projectSQLLog, projectSafeSQLLog, logValueStrings, scrubLogText, credentialName, inheritSQLLogBindings, retainSQLLogProvenance } from '../core/log-privacy';
import { sqlDiagnosticEvent } from '../core/sql-diagnostic-event';
import { telemetryNotice } from '../core/telemetry-startup';
import type { SQLLogBindingSource } from '../core/log-privacy';
import { SQLDatabaseKind, SQLParameterLogPolicy } from './log-rendering';
import { queryDiagnosticOrigin, retainQueryDiagnosticOrigin } from '../core/query-snapshot';
import { executeRelationFacets } from '../core/facet';
import type { FacetAggregate } from '../core/facet';
import { SmartList } from '../core/smart-list';
export type LogicalColumnType =
| 'boolean'
| 'double'
| 'decimal'
| 'date'
| 'datetime'
| 'json'
| 'integer'
| 'text';
export type ColumnSchema = {
columnName: string;
modelName?: string;
logicalType: LogicalColumnType;
decode: 'string' | 'number' | 'date' | 'native';
nullable?: boolean;
/** Trusted model/application log policy, never taken from a query payload. */
logPolicy?: SQLParameterLogPolicy;
};
export type EntitySchema = {
table: string;
columns: Record<string, ColumnSchema>;
relations?: Record<string, RelationSchema>;
auditMaskFields?: readonly string[];
};
export type RelationSchema = {
targetEntity: string;
localKey: string;
foreignKey: string;
many: boolean;
};
export type CanonicalRelationIndex = {
name: string;
table: string;
foreignColumn: string;
idColumn: string;
};
/** Canonical index for recent-child Top-N. Custom ordering needs an explicit model index. */
export function canonicalRelationIndexes(
schemas: Record<string, EntitySchema>,
): CanonicalRelationIndex[] {
const indexes = new Map<string, CanonicalRelationIndex>();
for (const schema of Object.values(schemas)) {
for (const relation of Object.values(schema.relations || {})) {
if (!relation.many) continue;
const target = schemas[relation.targetEntity];
const foreignColumn = target?.columns[relation.foreignKey]?.columnName;
const idColumn = target?.columns.id?.columnName;
if (!target || !foreignColumn || !idColumn) continue;
const raw = `idx_${target.table}_${foreignColumn}_${idColumn}`;
let hash = 2166136261;
for (let i = 0; i < raw.length; i += 1) {
hash = Math.imul(hash ^ raw.charCodeAt(i), 16777619);
}
const suffix = (hash >>> 0).toString(36);
const name = raw.length <= 30
? raw
: `${raw.slice(0, Math.max(1, 29 - suffix.length))}_${suffix}`;
indexes.set(name, { name, table: target.table, foreignColumn, idColumn });
}
}
return [...indexes.values()];
}
export type SqlQueryResult = {
rows: any[];
rowCount: number;
};
export type MutationResult = {
success: boolean;
id: string;
version?: number;
deleted?: boolean;
persistedRecord?: Record<string, unknown>;
/** Trusted internal result; use policy projection before diagnostics. */
metadata?: SQLExecutionMetadata;
};
export interface SqlSession {
query(sql: string, values?: any[]): Promise<SqlQueryResult>;
}
export interface TeaQLSqlDriver extends SqlSession {
readonly databaseKind: SQLDatabaseKind;
readonly topNRelationPlanPolicy?: 'window' | 'alwaysProbe';
stream(sql: string, values?: any[]): AsyncIterable<any>;
identifier(value: string): string;
placeholder(index: number): string;
encode(value: any, column?: ColumnSchema): any;
contains(columnSql: string, placeholder: string): string;
aggregateFunction(name: string): string;
ensureSchema(schemas: Record<string, EntitySchema>): Promise<void>;
transaction<T>(work: (session: SqlSession) => Promise<T>): Promise<T>;
nextId(session: SqlSession, entity: string): Promise<string>;
ensureIdFloor(session: SqlSession, entity: string, floor: string): Promise<void>;
close(): Promise<void>;
}
export async function ensureOptimisticIdFloor(
session: SqlSession,
placeholder: (index: number) => string,
entity: string,
floor: string,
): Promise<void> {
const numericFloor = Number(floor);
if (!Number.isSafeInteger(numericFloor) || numericFloor < 0) {
throw new Error(`Invalid ID space floor ${floor} for ${entity}`);
}
for (let attempt = 1; attempt <= 100; attempt += 1) {
const currentResult = await session.query(
`SELECT current_level AS id FROM teaql_id_space WHERE type_name = ${placeholder(1)}`,
[entity],
);
if (!currentResult.rowCount) {
try {
const inserted = await session.query(
`INSERT INTO teaql_id_space(type_name, current_level) VALUES (${placeholder(1)}, ${placeholder(2)})`,
[entity, numericFloor],
);
if (inserted.rowCount === 1) return;
} catch (error) {
const winner = await session.query(
`SELECT current_level FROM teaql_id_space WHERE type_name = ${placeholder(1)}`,
[entity],
);
if (!winner.rowCount) throw error;
}
continue;
}
const current = Number(currentResult.rows[0].id);
if (current >= numericFloor) return;
const updated = await session.query(
`UPDATE teaql_id_space SET current_level = ${placeholder(1)} ` +
`WHERE type_name = ${placeholder(2)} AND current_level = ${placeholder(3)}`,
[numericFloor, entity, current],
);
if (updated.rowCount === 1) return;
if (updated.rowCount !== 0) throw new Error(
`ID space floor update for ${entity} changed ${updated.rowCount} rows on attempt ${attempt}`);
}
throw new Error(
`Unable to synchronize ID space floor for ${entity} after 100 optimistic-lock attempts`);
}
function sameBootstrapValue(left: unknown, right: unknown): boolean {
if (left === right) return true;
if (left instanceof Date) left = left.getTime();
if (right instanceof Date) right = right.getTime();
return left !== null && left !== undefined && right !== null && right !== undefined &&
String(left) === String(right);
}
export interface TeaQLDataService {
withQueryRequest?<T>(query: SelectQuery, work: () => Promise<T>): Promise<T>;
executeGraphSave<T>(intent: MutationIntent, work: (graph: GraphMutationSession) => Promise<T>): Promise<T>;
preflightMutation(mutation: any): any;
afterGraphCommit(work: () => void): void;
afterGraphRollback(work: () => void): void;
executeMutation(mutation: any): Promise<MutationResult>;
executeQuery<T = any>(query: any): Promise<T[]>;
executeCount(query: any, alias?: string, operation?: QueryOperation): Promise<number>;
executeForStream<T = any>(query: any, chunkSize?: number): AsyncIterable<T[]>;
executeFacetMembership?(outerQuery: any, relationName: string, countAliases?: readonly string[]): Promise<Map<string, number>>;
close?(): Promise<void>;
}
export type SQLExecutionOperation = 'select' | 'insert' | 'update' | 'delete';
/** Statement/cursor completion, not transaction commit or business success. */
export type SQLExecutionOutcome = 'success' | 'failure' | 'cancelled';
type SQLLogIntent = { comment?: string; purpose?: string; auditReason?: string; tracePath?: readonly SQLTraceFrame[];
traceChain?: readonly TraceNode[];
mutationLineage?: readonly TraceNode[];
inheritedBindings?: SQLLogBindingSource;
/** Compiler-only target ID for free-text projection; never included in log metadata. */
targetID?: unknown };
export type SQLTraceFrame = TraceNode & Readonly<{
level: number;
}>;
function sqlTraceFrames(source: readonly TraceNode[]): readonly SQLTraceFrame[] {
return Object.freeze(source.map((node, level) => Object.freeze({ ...node, level })));
}
function mutationLogIntent(request: MutationRequest<any>, mutation: any, provider: string,
operation: SQLExecutionOperation, id: string, localBindings: SQLLogBindingSource): SQLLogIntent {
const lineage = request.traceFor({ entity: String(mutation.entity), id });
const path = canonicalSQLTracePath([...lineage, { kind: 'entity', name: String(mutation.entity),
entityId: id, detail: '' }], provider, operation);
return { auditReason: request.comment, tracePath: sqlTraceFrames(path.tracePath),
mutationLineage: lineage, targetID: id,
inheritedBindings: inheritSQLLogBindings(localBindings, request.graphSession?.logBindings) };
}
export type SQLExecutionMetadata = Readonly<{
/** Ordered physical children of a logical mutation result. */
statements?: readonly SQLExecutionMetadata[];
operation: SQLExecutionOperation;
executionOutcome?: SQLExecutionOutcome;
comment?: string;
purpose?: string;
auditReason?: string;
/** Business intent lineage: root purpose followed by each request's comment. */
traceChain?: readonly TraceNode[];
/** Physical execution route: operation/request/relation/provider/sql. */
tracePath: readonly SQLTraceFrame[];
/** Business responsibility, not the physical SQL route. */
mutationLineage?: readonly TraceNode[];
parameterizedSQL: string;
parameters: readonly unknown[];
/** SQL with bind values rendered as literals, intended only for diagnostics. */
debugSQL: string;
databaseKind?: SQLDatabaseKind;
parameterLogPolicies?: readonly SQLParameterLogPolicy[];
/** Set by the runtime SQL compiler, not by the incoming Q/TFP request. */
sqlOrigin?: 'generated';
maskedParameters?: readonly boolean[];
logMode?: 'masked' | 'debug-plaintext';
omissionReason?: 'untrusted-literal-sql' | 'policy-count-mismatch' | 'unsupported-or-mismatched-bindings';
elapsedMicros: number;
resultCount?: number;
affectedRows?: number;
resultSummary: string;
}>;
export { debugSQL } from './log-rendering';
export type { SQLDatabaseKind } from './log-rendering';
export interface RuntimeTelemetrySink {
record(metadata: SQLExecutionMetadata): void;
}
/**
* Policy-projected SQL diagnostic surface. Values are redacted by default;
* selecting a custom sink does not grant plaintext access.
*/
export interface DiagnosticSQLLogSink {
write(metadata: SQLExecutionMetadata): void;
}
export class TextDiagnosticSQLLogSink implements DiagnosticSQLLogSink {
constructor(private readonly writer: (text: string) => void = text => console.debug(text)) {}
write(metadata: SQLExecutionMetadata): void {
metadata = projectSQLLog(metadata);
this.writer(
`[TeaQL SQL][${metadata.operation}][${metadata.elapsedMicros}us] ${metadata.resultSummary}` +
`${metadata.executionOutcome ? ` outcome=${metadata.executionOutcome}` : ''}\n` +
`comment=${metadata.comment ?? ''} purpose=${metadata.purpose ?? ''} ` +
`auditReason=${metadata.auditReason ?? ''} tracePath=${diagnosticJSON(metadata.tracePath)}` +
` traceChain=${diagnosticJSON(metadata.traceChain ?? [])}` +
` mutationLineage=${diagnosticJSON(metadata.mutationLineage ?? [])}\n` +
(metadata.omissionReason ? `SQL omitted: ${metadata.omissionReason}\n` : '') +
`Debug SQL: ${metadata.debugSQL}`,
);
}
}
function diagnosticJSON(value: unknown): string {
return JSON.stringify(value, (_key, item) =>
typeof item === 'bigint' ? item.toString() : item);
}
export class SQLExecutionEvidenceStore implements RuntimeTelemetrySink {
private mode: 'all' | 'select' | 'mutation' | 'disabled' = 'all';
private entries: SQLExecutionMetadata[] = [];
record(metadata: SQLExecutionMetadata): void {
const isSelect = metadata.operation === 'select';
if (this.mode === 'disabled' ||
(this.mode === 'select' && !isSelect) ||
(this.mode === 'mutation' && isSelect)) return;
this.entries.push(Object.freeze({
...metadata,
...(metadata.traceChain ? { traceChain: cloneTraceNodes(metadata.traceChain) } : {}),
parameters: Object.freeze([...metadata.parameters]),
}));
}
private setMode(mode: 'all' | 'select' | 'mutation' | 'disabled'): this {
this.mode = mode;
this.entries = [];
return this;
}
enableAll(): this { return this.setMode('all'); }
enableSelect(): this { return this.setMode('select'); }
enableMutation(): this { return this.setMode('mutation'); }
disable(): this { return this.setMode('disabled'); }
snapshot(): readonly SQLExecutionMetadata[] { return [...this.entries]; }
}
type NormalizedAggregate = { func: string; field: string; retName: string };
type NormalizedOrder = { field: string; direction: string };
// Private to one relation load. Keys never enter records, wire requests,
// mutation state or UserContext; null hydration cannot erase membership.
type RelationAssembly = { field: string; keys: WeakMap<object, unknown> };
export abstract class AbstractSQLTeaQLClient implements TeaQLDataService {
private schemaReady?: Promise<void>;
private bootstrapTail: Promise<void> = Promise.resolve();
public readonly sqlTrace: string[] = [];
private readonly internalQueryToken = Symbol('teaql-internal-query');
private readonly bindLogPolicies = new WeakMap<readonly unknown[], SQLParameterLogPolicy[]>();
// Compiler-owned operands before LIKE decoration, never SQL parameters or
// serialized metadata. Lifetime follows this execution's binding array.
private readonly bindOperandSources = new WeakMap<readonly unknown[], SQLLogBindingSource>();
private readonly derivedQueryBindings = new WeakMap<object, SQLLogBindingSource>();
private readonly derivedRelationAssembly = new WeakMap<object, RelationAssembly>();
private fieldLogPolicy(schema: EntitySchema, field: string): SQLParameterLogPolicy {
const column = schema.columns[field];
const name = column?.modelName ?? column?.columnName ?? field;
if (credentialName(name) || credentialName(field)) return 'credential';
if (!schema.auditMaskFields) return 'unknown';
if (schema.auditMaskFields?.includes(name) || schema.auditMaskFields?.includes(field)) return 'masked';
return column?.logPolicy ?? 'unknown';
}
private bindValue(values: any[], value: unknown, policy: SQLParameterLogPolicy, sourceOperand?: string): void {
const policies = this.bindLogPolicies.get(values) ?? values.map(() => 'unknown' as const);
values.push(value);
policies.push(policy);
this.bindLogPolicies.set(values, policies);
if (sourceOperand !== undefined) {
this.bindOperandSources.set(values, inheritSQLLogBindings({
parameterizedSQL: '', parameters: [sourceOperand], parameterLogPolicies: [policy], sqlOrigin: 'generated',
}, this.bindOperandSources.get(values)));
}
}
private queryLogBindings(sql: string, values: readonly unknown[]): SQLLogBindingSource {
return inheritSQLLogBindings({ parameterizedSQL: sql, parameters: values,
parameterLogPolicies: this.bindLogPolicies.get(values), sqlOrigin: 'generated' }, this.bindOperandSources.get(values));
}
private readonly auditEvents: Readonly<Record<string, unknown>>[] = [];
private auditSink?: (event: Readonly<Record<string, unknown>>) => void | Promise<void>;
private telemetrySink?: RuntimeTelemetrySink;
private diagnosticSQLLogSink?: DiagnosticSQLLogSink = new TextDiagnosticSQLLogSink();
private queryLoggingEnabled = true;
private mutationLoggingEnabled = true;
private runtimeTelemetry?: RuntimeTelemetry;
private telemetryStartup?: Promise<void>;
private standardTelemetryHandle?: import('../telemetry/standard-node').StandardTelemetryHandle;
private customTelemetry = false;
private readonly checkers: Record<string, EntityChecker> = {};
private userContext = new UserContext();
private bootstrap: RuntimeBootstrap = {};
private graphMutationSession?: SqlSession;
private activeGraph?: GraphMutationSession;
private graphAuditActions: Array<() => Promise<void>> = [];
private graphCommitActions: Array<() => void> = [];
private graphRollbackActions: Array<() => void> = [];
private graphSaveTail: Promise<void> = Promise.resolve();
private readonly schemas: Record<string, EntitySchema>;
protected constructor(
protected readonly driver: TeaQLSqlDriver,
schemas: Record<string, EntitySchema>,
) { this.schemas = { ...schemas }; }
/** Installs metadata only. Call context.ensureSchema() explicitly when schema changes are intended. */
install(module: import('../core/runtime-module').RuntimeModule): this {
Object.assign(this.schemas, module.schemas);
Object.assign(this.checkers, module.checkers);
this.bootstrap = mergeRuntimeBootstrap(this.bootstrap, module.bootstrap);
return this;
}
setUserContext(context: UserContext): this { this.userContext = context; return this; }
private schema(entity: string): EntitySchema {
const schema = this.schemas[entity];
if (!schema) throw new Error(`Unknown TeaQL entity: ${entity}`);
return schema;
}
private toRuntimeMutationRecord(
schema: EntitySchema,
canonicalRecord: Record<string, unknown>,
): Record<string, unknown> {
const result: Record<string, unknown> = {};
for (const [canonicalName, value] of Object.entries(canonicalRecord)) {
const runtimeName = schema.columns[canonicalName]
? canonicalName
: Object.keys(schema.columns).find(
name => schema.columns[name].modelName === canonicalName ||
schema.columns[name].columnName === canonicalName,
);
if (runtimeName) result[runtimeName] = value;
}
return result;
}
private encode(value: any, column?: ColumnSchema): any {
const normalized = value?.id ?? value;
if (normalized === undefined) return undefined;
return this.driver.encode(normalized, column);
}
private decodeRow(entity: string, row: any, aggregateNames: string[] = []): any {
const schema = this.schema(entity);
const result = { ...row };
for (const [field, column] of Object.entries(schema.columns)) {
const value = result[field];
if (value === null || value === undefined) continue;
if (column.decode === 'number') result[field] = Number(value);
if (column.decode === 'string') result[field] = String(value);
if (column.decode === 'date' && value instanceof Date) {
result[field] = value.toISOString();
}
if (column.logicalType === 'boolean') result[field] = Boolean(value);
}
for (const name of aggregateNames) {
if (result[name] !== null && result[name] !== undefined) {
result[name] = Number(result[name]);
}
}
return result;
}
get auditTrace(): ReadonlyArray<Readonly<Record<string, unknown>>> {
return [...this.auditEvents];
}
setAuditSink(sink: (event: Readonly<Record<string, unknown>>) => void | Promise<void>): this {
this.auditSink = sink;
return this;
}
setRuntimeTelemetrySink(sink: RuntimeTelemetrySink | undefined): this {
this.telemetrySink = sink;
return this;
}
setDiagnosticSQLLogSink(sink: DiagnosticSQLLogSink | undefined): this {
this.diagnosticSQLLogSink = sink;
return this;
}
setQueryLoggingEnabled(enabled: boolean): this {
this.queryLoggingEnabled = enabled;
return this;
}
setMutationLoggingEnabled(enabled: boolean): this {
this.mutationLoggingEnabled = enabled;
return this;
}
setRuntimeTelemetry(telemetry: RuntimeTelemetry | undefined): this {
this.customTelemetry = true;
this.runtimeTelemetry = telemetry;
return this;
}
/** Node provider startup hook. Optional module initialization remains fail-open. */
protected attachStandardTelemetryStartup(
startup: Promise<import('../telemetry/standard-node').StandardTelemetryHandle | undefined>,
): void {
this.telemetryStartup = startup.then(async handle => {
if (!handle) return;
if (this.customTelemetry) { await handle.release(); return; }
this.standardTelemetryHandle = handle;
this.runtimeTelemetry = handle.telemetry;
}).catch(() => { telemetryNotice('TeaQL telemetry: initialization failed; business continues without export.', true); });
}
async flushTelemetry(): Promise<void> {
await this.telemetryStartup;
try { await this.standardTelemetryHandle?.flush(); }
catch { telemetryNotice('TeaQL telemetry: export flush failed; business outcome unchanged.', true); }
}
async withQueryRequest<T>(query: SelectQuery, work: () => Promise<T>): Promise<T> {
new QueryRequest(query); // Reject invalid intent before starting execution.
await this.telemetryStartup;
return observeRuntimeOperation(this.runtimeTelemetry, { family: 'query',
name: `${String(query.entity)}.request`, attributes: { 'teaql.entity.type': String(query.entity) } }, work);
}
private recordSQL(
operation: SQLExecutionOperation, parameterizedSQL: string, parameters: readonly unknown[],
startedAt: number, resultCount?: number, affectedRows?: number,
intent: SQLLogIntent = {},
executionOutcome: SQLExecutionOutcome = 'success',
inherited?: SQLLogBindingSource,
correlation?: RuntimeExecutionCorrelation,
): SQLExecutionMetadata {
const isSelect = operation === 'select';
const logsEnabled = isSelect ? this.queryLoggingEnabled : this.mutationLoggingEnabled;
const { targetID, inheritedBindings, ...visibleIntent } = intent;
const metadata = Object.freeze({
operation, ...visibleIntent, executionOutcome,
traceChain: cloneTraceNodes(intent.traceChain ?? []),
tracePath: sqlTraceFrames(intent.tracePath ?? []),
...(intent.traceChain ? { traceChain: cloneTraceNodes(intent.traceChain) } : {}),
...(intent.mutationLineage ? { mutationLineage: cloneTraceNodes(intent.mutationLineage) } : {}),
parameterizedSQL, parameters: Object.freeze([...parameters]),
// Never build a plaintext SQL copy before the log policy boundary.
debugSQL: '', databaseKind: this.driver.databaseKind, sqlOrigin: 'generated' as const,
parameterLogPolicies: Object.freeze([...(this.bindLogPolicies.get(parameters) ?? parameters.map(() => 'unknown' as const))]),
elapsedMicros: Math.max(0, (Date.now() - startedAt) * 1_000),
resultCount, affectedRows,
resultSummary: resultCount !== undefined
? `${resultCount} rows returned` : affectedRows !== undefined ? `${affectedRows} rows affected`
: `statement ${executionOutcome}; row count unknown`,
});
const ancestors = inheritedBindings && inherited
? inheritSQLLogBindings(inherited, inheritedBindings) : inheritedBindings ?? inherited;
const operands = this.bindOperandSources.get(parameters);
const provenance = operands ? inheritSQLLogBindings(operands, ancestors) : ancestors;
retainSQLLogProvenance(metadata, provenance, targetID === undefined ? [] : [targetID]);
if (this.runtimeTelemetry?.recordDiagnostic) {
try {
if (this.runtimeTelemetry.recordsSQLDiagnostics?.() !== false)
this.runtimeTelemetry.recordDiagnostic(sqlDiagnosticEvent(projectSafeSQLLog(metadata), correlation));
}
catch { telemetryNotice('TeaQL telemetry: SQL diagnostic rejected; business outcome unchanged.', true); }
}
if (!this.telemetrySink && !(logsEnabled && this.diagnosticSQLLogSink)) return metadata;
try {
const projected = projectSQLLog(metadata);
// Runtime diagnostics are fail-open and each sink is independent. A
// broken application sink must not roll back a successful SQL mutation.
try { this.telemetrySink?.record(projected); } catch { /* diagnostic sink failed */ }
if (logsEnabled) {
try { this.diagnosticSQLLogSink?.write(projected); } catch { /* diagnostic sink failed */ }
}
} catch { /* projection failure must not alter database execution */ }
return metadata;
}
private queryLogIntent(query: any): SQLLogIntent {
const request = new QueryRequest(query);
const path = canonicalSQLTracePath(request.traceSource, this.driver.databaseKind, 'select');
return { comment: request.intent.comment, purpose: request.purpose,
traceChain: request.traceChain, tracePath: sqlTraceFrames(path.tracePath) };
}
private async executeLoggedSQL(
operation: SQLExecutionOperation, sql: string, values: any[],
intent: SQLLogIntent,
execute: () => Promise<SqlQueryResult>,
inherited?: SQLLogBindingSource,
statements?: SQLExecutionMetadata[],
): Promise<SqlQueryResult> {
await this.telemetryStartup;
const scope = startRuntimeOperation(this.runtimeTelemetry, { family: 'provider',
name: `${this.driver.databaseKind}.${operation}`,
attributes: { 'teaql.provider.kind': this.driver.databaseKind,
'teaql.provider.operation': operation === 'select' ? 'query' : operation } });
const correlation = scope.correlation?.();
const startedAt = Date.now();
let result: SqlQueryResult;
try { result = await (scope.run ? scope.run(execute) : execute()); }
catch (error) {
// No exception text/cause/driver object enters a diagnostic sink. Preserve
// the original error for the caller, even if a sink itself is broken.
try {
const metadata = this.recordSQL(operation, sql, values, startedAt, undefined, undefined, intent, 'failure', inherited, correlation);
statements?.push(metadata);
}
finally { scope.failure(error); throw error; }
}
const metadata = this.recordSQL(operation, sql, values, startedAt,
operation === 'select' ? result.rowCount : undefined,
operation === 'select' ? undefined : result.rowCount, intent, 'success', inherited, correlation);
statements?.push(metadata);
scope.success({ attributes: { 'teaql.result.cardinality': result.rowCount } });
return result;
}
/** Package-internal physical capability used only by UserContext.ensureSchema(). */
async [contextSchemaCapability](context: UserContext): Promise<void> {
await this.telemetryStartup;
if (!this.schemaReady) {
this.schemaReady = this.driver.ensureSchema(this.schemas);
}
await this.schemaReady;
const predecessor = this.bootstrapTail;
let release!: () => void;
this.bootstrapTail = new Promise<void>(resolve => { release = resolve; });
await predecessor;
this.userContext = context;
const previousActor = context.getResource<string>('bootstrapActor');
const previousCategory = context.getResource<string>('bootstrapCategory');
context.insertResource('bootstrapActor', 'teaql-generated-bootstrap');
context.insertResource('bootstrapCategory', 'runtime-bootstrap');
try {
if (this.bootstrap.ensure) {
await this.bootstrap.ensure(context);
} else {
await this.ensureBootstrapData();
}
} finally {
if (previousActor === undefined) context.removeResource('bootstrapActor');
else context.insertResource('bootstrapActor', previousActor);
if (previousCategory === undefined) context.removeResource('bootstrapCategory');
else context.insertResource('bootstrapCategory', previousCategory);
release();
}
}
/** Allows a provider which replaces its physical store to require explicit schema reconciliation again. */
protected invalidateSchemaState(): void {
this.schemaReady = undefined;
}
private async ensureBootstrapData(): Promise<void> {
const records = [
...(this.bootstrap.defaultDomainRoot ? [this.bootstrap.defaultDomainRoot] : []),
...(this.bootstrap.constants ?? []),
];
if (!records.length) return;
await this.executeGraphSave(new MutationIntent('reconcile model bootstrap data'), async graph => {
const session = this.graphMutationSession!;
const planned: Array<{ record: BootstrapEntity; request: MutationRequest<any> }> = [];
for (const record of records) {
const mutation = await this.planBootstrapMutation(session, record);
if (mutation) planned.push({ record, request: graph.request(mutation) });
}
for (const { request } of planned) this.preflightMutation(request);
for (const { request } of planned) await this.executeMutation(request);
for (const record of records) await this.driver.ensureIdFloor(session, record.entity, record.id);
});
}
private async planBootstrapMutation(session: SqlSession, record: BootstrapEntity): Promise<any | undefined> {
const schema = this.schema(record.entity);
const entries = Object.entries(record.values ?? {}).map(([field, value]) => {
const column = schema.columns[field];
if (!column) throw new Error(`Unknown bootstrap field ${record.entity}.${field}`);
return {
column: column.columnName,
value: this.driver.encode(value, column),
};
});
const query = new SelectQuery(record.entity).filter({ id: { $eq: record.id } }).limit(1)
.comment('inspect model bootstrap record').purpose('ensure model-defined root and constants');
query.prepareForList();
const request = new QueryRequest(query);
// Reconciliation must see an existing tombstone too. Normal list SQL adds
// version > 0 and would attempt a duplicate INSERT for a deleted constant.
// This fixed primary-key lookup still owns validated query intent and logs.
const sql = `SELECT * FROM ${this.driver.identifier(schema.table)} WHERE ` +
`${this.driver.identifier(schema.columns.id?.columnName ?? 'id')} = ${this.driver.placeholder(1)}`;
const values = [record.id];
this.bindLogPolicies.set(values, [this.fieldLogPolicy(schema, 'id')]);
const current = await this.executeLoggedSQL('select', sql, values, this.queryLogIntent(request.query),
() => session.query(sql, values));
if (!current.rowCount) {
return { entity: record.entity, action: 'Create', id: record.id, version: 0,
payload: { ...record.values } };
}
const row = current.rows[0];
const changed = entries.filter(entry => !sameBootstrapValue(row[entry.column], entry.value));
if (!changed.length) return undefined;
// Checker sees the fully loaded object, not a partial update projection.
return { entity: record.entity, action: 'Update', id: record.id,
version: Number(row[schema.columns.version?.columnName ?? 'version']),
payload: { ...this.decodeRowForSchema(schema, row), ...record.values } };
}
async executeGraphSave<T>(intent: MutationIntent, work: (graph: GraphMutationSession) => Promise<T>): Promise<T> {
// Root intent is mandatory before lifecycle initialization or provider work.
const graph = new GraphMutationSession(intent);
await this.telemetryStartup;
const save = async () => {
// Generated child saves call their within-graph entry point directly. Every
// public/root save comes through here and is queued, so an unrelated async
// save cannot accidentally join another graph's transaction merely because
// this client currently has an active session.
const predecessor = this.graphSaveTail;
let release!: () => void;
this.graphSaveTail = new Promise<void>(resolve => { release = resolve; });
await predecessor;
let fixEvidenceStarted = false;
let mutationPolicyGraphStarted = false;
let committed = false;
try {
this.graphCommitActions = [];
this.graphRollbackActions = [];
this.graphAuditActions = [];
this.userContext.insertResource('fixTime', new Date());
this.userContext.beginFixEvidence();
fixEvidenceStarted = true;
this.userContext.beginMutationPolicyGraph();
mutationPolicyGraphStarted = true;
const result = await this.driver.transaction(async session => {
this.graphMutationSession = session;
this.activeGraph = graph;
try {
const value = await work(graph);
this.userContext.ensureMutationPolicyGraphComplete();
return value;
}
finally { this.graphMutationSession = undefined; this.activeGraph = undefined; }
});
committed = true;
let failure: unknown;
let failed = false;
// Cleanup and other audit events must continue after one sink fails.
for (const action of [...this.graphCommitActions, ...this.graphAuditActions]) {
try { await action(); } catch (error) { if (!failed) failure = error; failed = true; }
}
if (failed) throw new GraphCommittedError(failure);
return result;
} catch (error) {
if (!committed) {
// Keep the original execution failure even if a restore callback fails.
for (const action of [...this.graphRollbackActions].reverse()) {
try { action(); } catch { /* independent rollback restorations continue */ }
}
}
throw error;
} finally {
this.graphCommitActions = [];
this.graphRollbackActions = [];
this.graphAuditActions = [];
if (mutationPolicyGraphStarted) this.userContext.endMutationPolicyGraph();
this.userContext.removeResource('fixTime');
if (fixEvidenceStarted) this.userContext.finishFixEvidence();
release();
}
};
return observeRuntimeOperation(this.runtimeTelemetry, { family: 'mutation', name: 'graph.save',
attributes: { 'teaql.entity.type': 'graph', 'teaql.mutation.kind': 'graph' } }, save);
}
afterGraphCommit(work: () => void): void {
if (!this.graphMutationSession) throw new Error('No graph save is active');
this.graphCommitActions.push(work);
}
afterGraphRollback(work: () => void): void {
if (!this.graphMutationSession) throw new Error('No graph save is active');
this.graphRollbackActions.push(work);
}
private requireGraphOwnership(request: MutationRequest<any>): void {
if (request.graphSession !== this.activeGraph ||
(this.graphMutationSession !== undefined && request.graphSession === undefined)) {
throw new Error('GRAPH_MUTATION_SESSION_REQUIRED: use the explicit active graph request capability');
}
}
private async withMutationSession<T>(request: MutationRequest<any>, work: (session: SqlSession) => Promise<T>): Promise<T> {
this.requireGraphOwnership(request);
return request.graphSession ? work(this.graphMutationSession!) : this.driver.transaction(work);
}
preflightMutation(mutation: any): any {
const request = mutation instanceof MutationRequest ? mutation : new MutationRequest(mutation);
this.requireGraphOwnership(request);
mutation = this.checkAndFixMutation(request);
if (request.graphSession) {
const schema = this.schema(mutation.entity);
request.graphSession.captureLogBindings(this.mutationLogBindings(request, mutation, schema));
}
this.userContext.recordMutationPolicyPreflight(mutation);
return mutation;
}
private mutationLogBindings(request: MutationRequest<any>, mutation: any, schema: EntitySchema): SQLLogBindingSource {
const source = (values: Record<string, unknown>): SQLLogBindingSource => {
const record = this.toRuntimeMutationRecord(schema, values);
const fields = Object.keys(record);
return { parameterizedSQL: '', sqlOrigin: 'generated', parameters: fields.map(field => record[field]),
parameterLogPolicies: fields.map(field => this.fieldLogPolicy(schema, field)) };
};
return inheritSQLLogBindings(source(mutation.payload || {}), source(request.loadedValues()));
}
private checkAndFixMutation(mutation: any): any {
const request = mutation instanceof MutationRequest ? mutation : new MutationRequest(mutation);
this.requireGraphOwnership(request);
mutation = request.mutation;
// The ledger exposes immutable snapshots. Fixers receive a mutable working
// payload and every derived value is copied back into the graph ledger.
mutation = { ...mutation, payload: { ...(mutation?.payload ?? {}) } };
Object.defineProperty(mutation, 'comment', { value: request.comment, enumerable: true });
const checker = this.checkers[String(mutation.entity)];
if (!checker) return mutation;
const results: import('../core/i18n').CheckResult[] = [];
const ownsFixTime = this.userContext.getResource<Date>('fixTime') === undefined;
if (ownsFixTime) this.userContext.insertResource('fixTime', new Date()).beginFixEvidence();
try {
checker.checkAndFix(this.userContext, mutation, results);
if (mutation.ledgerKey && mutation.ledgerRoot) {
for (const [field, value] of Object.entries(mutation.payload || {})) {
mutation.ledgerRoot.set(mutation.ledgerKey, field, value);
}
}
this.userContext.translateCheckResults(results);
if (results.length) throw new CheckException(results);
return mutation;
} finally {
if (ownsFixTime) this.userContext.removeResource('fixTime').finishFixEvidence();
}
}
async executeMutation(mutation: any): Promise<MutationResult> {
await this.telemetryStartup;
// Validate before Checker, policy, telemetry and transaction/provider work.
const request = mutation instanceof MutationRequest ? mutation : new MutationRequest(mutation);
this.requireGraphOwnership(request);
mutation = request.mutation;
const scope = startRuntimeOperation(this.runtimeTelemetry, {
family: 'mutation',
name: `${String(mutation?.entity || 'unknown')}.${String(mutation?.action || 'unknown').toLowerCase()}`,
attributes: {
'teaql.entity.type': String(mutation?.entity || 'unknown'),
'teaql.mutation.kind': String(mutation?.action || 'unknown').toLowerCase(),
},
});
const work = async () => { try {
mutation = this.checkAndFixMutation(request);
const mutationGovernance = this.userContext.enterMutationPolicy(mutation);
const schema = this.schema(mutation.entity);
const mutationRecord = this.toRuntimeMutationRecord(schema, mutation.payload || {});
const diagnosticBindings = this.mutationLogBindings(request, mutation, schema);
const table = this.driver.identifier(schema.table);
const statements: SQLExecutionMetadata[] = [];
const result = await observeRuntimeOperation(this.runtimeTelemetry, {
family: 'provider',
name: `${this.driver.databaseKind}.mutation`,
attributes: {
'teaql.provider.kind': this.driver.databaseKind,
'teaql.provider.operation': String(mutation.action).toLowerCase(),
},
}, () => this.withMutationSession(request, async session => {
if (mutation.action === 'Create') {
const id = mutation.id
? String(mutation.id)
: await this.driver.nextId(session, mutation.entity);
if (mutation.id) {
await this.driver.ensureIdFloor(session, mutation.entity, id);
}
const version = Number(mutation.version || 0) + 1;
const record: Record<string, unknown> = { ...mutationRecord, id, version };
const fields = Object.keys(schema.columns)
.filter(field => record[field] !== undefined);
const columns = fields.map(field =>
this.driver.identifier(schema.columns[field].columnName),
).join(', ');
const placeholders = fields.map((_, index) =>
this.driver.placeholder(index + 1),
).join(', ');
const values = fields.map(field =>
this.encode(record[field], schema.columns[field]),
);
this.bindLogPolicies.set(values, fields.map(field => this.fieldLogPolicy(schema, field)));
const sql = `INSERT INTO ${table} (${columns}) VALUES (${placeholders})`;
const intent = mutationLogIntent(request, mutation, this.driver.databaseKind, 'insert', id, diagnosticBindings);
await this.executeLoggedSQL('insert', sql, values, intent, () => session.query(sql, values), undefined, statements);
return {
success: true,
id,
version,
persistedRecord: await this.readPersistedRecord(session, schema, id, intent, sql, values, statements),
};
}
if (mutation.action === 'Update') {
const fields = Object.keys(schema.columns).filter(field =>
field !== 'id' && field !== 'version' &&
mutationRecord[field] !== undefined,
);
const values = fields.map(field =>
this.encode(mutationRecord[field], schema.columns[field]),
);
this.bindLogPolicies.set(values, fields.map(field => this.fieldLogPolicy(schema, field)));
const assignments = fields.map((field, index) =>
`${this.driver.identifier(schema.columns[field].columnName)} = ` +
this.driver.placeholder(index + 1),
);
const versionColumn = this.driver.identifier('version');
assignments.push(`${versionColumn} = ${versionColumn} + 1`);
this.bindValue(values, String(mutation.id), this.fieldLogPolicy(schema, 'id'));
const predicates = [
`${this.driver.identifier('id')} = ${this.driver.placeholder(values.length)}`,
];
if (mutation.version !== undefined && mutation.version !== null) {
this.bindValue(values, Number(mutation.version), this.fieldLogPolicy(schema, 'version'));
predicates.push(
`${versionColumn} = ${this.driver.placeholder(values.length)}`,
);
}
const sql = `UPDATE ${table} SET ${assignments.join(', ')} ` +
`WHERE ${predicates.join(' AND ')}`;
const intent = mutationLogIntent(request, mutation, this.driver.databaseKind, 'update', String(mutation.id), diagnosticBindings);
const result = await this.executeLoggedSQL('update', sql, values, intent, () => session.query(sql, values), undefined, statements);
if (result.rowCount !== 1) {
throw new Error(
`Optimistic lock failed or ${mutation.entity}(${mutation.id}) does not exist`,
);
}
const persistedRecord = await this.readPersistedRecord(
session, schema, String(mutation.id), intent, sql, values, statements,
);
return {
success: true,
id: String(mutation.id),
version: Number(persistedRecord.version),
persistedRecord,
};
}
if (mutation.action === 'Delete') {
const versionColumn = this.driver.identifier('version');
const values: any[] = [String(mutation.id)];
this.bindLogPolicies.set(values, [this.fieldLogPolicy(schema, 'id')]);
const predicates = [
`${this.driver.identifier('id')} = ${this.driver.placeholder(1)}`,
];
if (mutation.version !== undefined && mutation.version !== null) {
this.bindValue(values, Number(mutation.version), this.fieldLogPolicy(schema, 'version'));
predicates.push(
`${this.driver.identifier('version')} = ` +
this.driver.placeholder(values.length),
);
}
const sql = `UPDATE ${table} SET ${versionColumn} = -(${versionColumn} + 1) ` +
`WHERE ${predicates.join(' AND ')}`;
const intent = mutationLogIntent(request, mutation, this.driver.databaseKind, 'delete', String(mutation.id), diagnosticBindings);
const result = await this.executeLoggedSQL('delete', sql, values, intent, () => session.query(sql, values), undefined, statements);
if (result.rowCount !== 1) {
throw new Error(
`Optimistic lock failed or ${mutation.entity}(${mutation.id}) does not exist`,
);
}
const persistedRecord = await this.readPersistedRecord(
session, schema, String(mutation.id), intent, sql, values, statements,
);
return {
success: true,
id: String(mutation.id),
version: Number(persistedRecord.version),
deleted: true,
persistedRecord,
};
}
throw new Error(`Unsupported mutation action: ${mutation.action}`);
}));
const auditProjection = request.auditProjection({ entity: String(mutation.entity), id: result.id }, mutation.payload, diagnosticBindings);
const event = Object.freeze({
entity: mutation.entity,
action: mutation.action,
id: String(result.id),
...auditProjection,
recordedAt: new Date().toISOString(),
actor: this.userContext.getResource<string>('bootstrapActor'),
category: this.userContext.getResource<string>('bootstrapCategory'),
changedFields: Object.freeze(Object.keys(mutation.payload || {}).sort()),
version: result.version,
mutationGovernance,
});
const publish = async () => {
this.auditEvents.push(event);
if (this.auditSink) await observeRuntimeOperation(this.runtimeTelemetry, {
family: 'audit',
name: `${mutation.entity}.audit`,
attributes: {
'teaql.entity.type': String(mutation.entity),
'teaql.mutation.kind': String(mutation.action).toLowerCase(),
'teaql.audit.changed_field_count': Object.keys(mutation.payload || {}).length,
},
}, async () => this.auditSink!(event));
};
if (request.graphSession) this.graphAuditActions.push(publish);
else {
try { await publish(); } catch (error) { throw new GraphCommittedError(error); }
}
scope.success();
const metadata = Object.freeze({ ...statements[0], statements: Object.freeze([...statements]) });
retainSQLLogProvenance(metadata, inheritSQLLogBindings(diagnosticBindings, request.graphSession?.logBindings), [result.id]);
return { ...result, metadata };
} catch (error) {
scope.failure(error);
throw error;
} };
return scope.run ? scope.run(work) : work();
}
private async readPersistedRecord(
session: SqlSession, schema: EntitySchema, id: string,
intent: SQLLogIntent, writeSQL: string, writeValues: any[],
statements: SQLExecutionMetadata[],
): Promise<Record<string, unknown>> {
const projection = Object.entries(schema.columns).map(([field, column]) =>
`${this.driver.identifier(column.columnName)} AS ${this.driver.identifier(field)}`,
).join(', ');
const sql = `SELECT ${projection} FROM ${this.driver.identifier(schema.table)} WHERE ` +
`${this.driver.identifier('id')} = ${this.driver.placeholder(1)}`;
const values = [id];
this.bindLogPolicies.set(values, [this.fieldLogPolicy(schema, 'id')]);
const scope = startRuntimeOperation(this.runtimeTelemetry, { family: 'provider',
name: `${this.driver.databaseKind}.readback`, attributes: {
'teaql.provider.kind': this.driver.databaseKind, 'teaql.provider.operation': 'query' } });
const startedAt = Date.now();
let result: SqlQueryResult | undefined;
const inherited: SQLLogBindingSource = { parameterizedSQL: writeSQL, parameters: writeValues,
parameterLogPolicies: this.bindLogPolicies.get(writeValues), sqlOrigin: 'generated' };
const root = intent.tracePath?.find(node => node.kind === 'operation')?.name ?? 'unknown';
const path = canonicalSQLTracePath([
...queryTraceSource(root, intent.auditReason!, 'verify persisted mutation result'),
...(intent.tracePath ?? []).filter(node => node.kind === 'relation'),
], this.driver.databaseKind, 'select');
const readIntent = { ...intent, tracePath: sqlTraceFrames(path.tracePath),
comment: intent.auditReason, purpose: 'verify persisted mutation result' };
const recordReadback = (outcome: SQLExecutionOutcome) => statements.push(
this.recordSQL('select', sql, values, startedAt, result?.rowCount, undefined, readIntent, outcome, inherited, scope.correlation?.()));
let record: Record<string, unknown>;
try {
result = await (scope.run ? scope.run(() => session.query(sql, values)) : session.query(sql, values));
scope.success({ attributes: { 'teaql.result.cardinality': result.rowCount } });
if (result.rowCount !== 1) throw new Error(`Persisted ${schema.table}(${id}) could not be read back`);
record = this.decodeRowForSchema(schema, result.rows[0]);
} catch (error) {
if (!result) scope.failure(error);
// Write success remains a separate statement fact. A snapshot validation
// failure means SELECT success with its real count, not a driver failure.
const outcome: SQLExecutionOutcome = result ? 'success'
: error instanceof Error && (error.name === 'AbortError' || error.name === 'TimeoutError')
? 'cancelled' : 'failure';
try { recordReadback(outcome); } finally { throw error; }
}
recordReadback('success');
return record;
}
private decodeRowForSchema(schema: EntitySchema, row: any): Record<string, unknown> {
const result = { ...row };
for (const [field, column] of Object.entries(schema.columns)) {
const value = result[field];
if (value === null || value === undefined) continue;
if (column.decode === 'number') result[field] = Number(value);
if (column.decode === 'string') result[field] = String(value);
if (column.decode === 'date' && value instanceof Date) result[field] = value.toISOString();
if (column.logicalType === 'boolean') result[field] = Boolean(value);
}
return result;
}
private compileExpression(
expression: any,
schema: EntitySchema,
values: any[],
): string {
if (Array.isArray(expression?.$and)) {
const parts = expression.$and.map((item: any) =>
this.compileExpression(item, schema, values),
);
return `(${parts.join(' AND ')})`;
}
const parts = Object.entries(expression || {}).map(
([field, predicate]: [string, any]) => {
const column = schema.columns[field];
if (!column) throw new Error(`Unknown field ${field} for SQL query`);
const quotedField = this.driver.identifier(column.columnName);
const logPolicy = this.fieldLogPolicy(schema, field);
if (predicate?.$eq !== undefined) {
const value = predicate.$eq?.id ?? predicate.$eq;
if (value === null) return `${quotedField} IS NULL`;
this.bindValue(values, this.encode(value, column), logPolicy);
return `${quotedField} = ${this.driver.placeholder(values.length)}`;
}
if (predicate?.$ne !== undefined) {
const value = predicate.$ne?.id ?? predicate.$ne;
if (value === null) return `${quotedField} IS NOT NULL`;
this.bindValue(values, this.encode(value, column), logPolicy);
return `${quotedField} <> ${this.driver.placeholder(values.length)}`;
}
if (predicate?.$soundLike !== undefined) {
this.bindValue(values, String(predicate.$soundLike), logPolicy);
return `SOUNDEX(${quotedField}) = SOUNDEX(${this.driver.placeholder(values.length)})`;
}
if (predicate?.$contains !== undefined) {
this.bindValue(values, String(predicate.$contains), logPolicy);
return this.driver.contains(
quotedField,
this.driver.placeholder(values.length),
);
}
if (predicate?.$notContains !== undefined) {
this.bindValue(values, String(predicate.$notContains), logPolicy);
return `NOT (${this.driver.contains(quotedField, this.driver.placeholder(values.length))})`;
}
for (const [operator, prefix, suffix, negative] of [
['$startsWith', '', '%', false],
['$notStartsWith', '', '%', true],
['$endsWith', '%', '', false],
['$notEndsWith', '%', '', true],
] as const) {
if (predicate?.[operator] !== undefined) {
// Retain the exact original value; stripping SQL wildcard text
// would misclassify literal %, _ or escape characters in intent.
const operand = String(predicate[operator]);
this.bindValue(values, `${prefix}${operand}${suffix}`, logPolicy, operand);
const like = `${quotedField} LIKE ${this.driver.placeholder(values.length)}`;
return negative ? `NOT (${like})` : like;
}
}
if (Array.isArray(predicate?.$in)) {
if (!predicate.$in.length) return 'FALSE';
const placeholders = predicate.$in.map((value: any) => {
this.bindValue(values, this.encode(value?.id ?? value, column), logPolicy);
return this.driver.placeholder(values.length);
});
return `${quotedField} IN (${placeholders.join(', ')})`;
}
if (Array.isArray(predicate?.$notIn)) {
if (!predicate.$notIn.length) return 'TRUE';
const placeholders = predicate.$notIn.map((value: any) => {
this.bindValue(values, this.encode(value?.id ?? value, column), logPolicy);
return this.driver.placeholder(values.length);
});
return `${quotedField} NOT IN (${placeholders.join(', ')})`;
}
for (const [operator, negative] of [
['$inSubquery', false],
['$notInSubquery', true],
] as const) {
const subquery = predicate?.[operator];
if (subquery !== undefined) {
const childQuery = subquery.query;
const childSchema = this.schema(childQuery?.entity);
const projectedField = String(subquery.field);
const projectedColumn = childSchema.columns[projectedField];
if (!projectedColumn) {
throw new Error(
`Unknown subquery projection ${projectedField} for ${childQuery?.entity}`,
);
}
const childPredicates = this.filters(childQuery).map(item =>
this.compileExpression(item, childSchema, values),
);
if (childSchema.columns.version) {
childPredicates.push(
`${this.driver.identifier(childSchema.columns.version.columnName)} > 0`,
);
}
// A NULL projected by a NOT IN subquery makes the predicate
// UNKNOWN for every outer row. Orphan relations are handled by
// the explicit unknown predicate and must not poison HaveNo or
// negative relation matching.
if (negative) {
childPredicates.push(
`${this.driver.identifier(projectedColumn.columnName)} IS NOT NULL`,
);
}
const where = childPredicates.length
? ` WHERE ${childPredicates.join(' AND ')}`
: '';
const sql = `SELECT ${this.driver.identifier(projectedColumn.columnName)} ` +
`FROM ${this.driver.identifier(childSchema.table)}${where}`;
return `${quotedField} ${negative ? 'NOT IN' : 'IN'} (${sql})`;
}
}
if (Array.isArray(predicate?.$between) && predicate.$between.length === 2) {
this.bindValue(values, this.encode(predicate.$between[0], column), logPolicy);
const lower = this.driver.placeholder(values.length);
this.bindValue(values, this.encode(predicate.$between[1], column), logPolicy);
const upper = this.driver.placeholder(values.length);
return `${quotedField} BETWEEN ${lower} AND ${upper}`;
}
if (predicate?.$isNull === true) return `${quotedField} IS NULL`;
if (predicate?.$isNull === false) return `${quotedField} IS NOT NULL`;
if (predicate?.$gte !== undefined) {
this.bindValue(values, this.encode(predicate.$gte, column), logPolicy);
return `${quotedField} >= ${this.driver.placeholder(values.length)}`;
}
if (predicate?.$lte !== undefined) {
this.bindValue(values, this.encode(predicate.$lte, column), logPolicy);
return `${quotedField} <= ${this.driver.placeholder(values.length)}`;
}
if (predicate?.$gt !== undefined) {
this.bindValue(values, this.encode(predicate.$gt, column), logPolicy);
return `${quotedField} > ${this.driver.placeholder(values.length)}`;
}
if (predicate?.$lt !== undefined) {
this.bindValue(values, this.encode(predicate.$lt, column), logPolicy);
return `${quotedField} < ${this.driver.placeholder(values.length)}`;
}
throw new Error(`Unsupported query predicate for ${field}: ${JSON.stringify(predicate)}`);
},
);
return parts.length ? `(${parts.join(' AND ')})` : 'TRUE';
}
private filters(query: any): any[] {
if (Array.isArray(query._filters) && query._filters.length) return query._filters;
return query.filterCondition ? [query.filterCondition] : [];
}
private groupBy(query: any): string[] {
return query._groupBy || query.groupByItems || [];
}
private aggregates(query: any): NormalizedAggregate[] {
if (Array.isArray(query._aggregates)) return query._aggregates;
return (query.aggregateItems || []).map((aggregate: any) => ({
func: aggregate.function,
field: aggregate.field,
retName: aggregate.alias,
}));
}
private orders(query: any): NormalizedOrder[] {
if (Array.isArray(query._orderBy)) {
return query._orderBy.map((order: any) => ({
field: order.f,
direction: order.d,
}));
}
return (query.orderItems || []).map((order: any) => ({
field: order.field,
direction: order.direction,
}));
}
private async compileQuery(query: any, recordTrace = true, assemblyFields: readonly string[] = []): Promise<{
sql: string;
values: any[];
aggregateNames: string[];
}> {
// Internal execution changes query shape/limits, never the intent contract.
new QueryIntent(query?.commentText ?? query?._comment, query?.purposeText ?? query?._purpose);
const schema = this.schema(query.entity);
const values: any[] = [];
const groupProperties = this.groupBy(query);
const groupFields = groupProperties.map(field => {
if (!schema.columns[field]) throw new Error(`Unknown group field: ${field}`);
return this.driver.identifier(schema.columns[field].columnName);
});
const groupProjection = groupProperties.map(field =>
`${this.driver.identifier(schema.columns[field].columnName)} AS ` +
this.driver.identifier(field),
);
const aggregateNames: string[] = [];
const requestedFields = Array.isArray(query.selectItems) && query.selectItems.length
? [...new Set(['id', 'version', ...query.selectItems])]
: Object.keys(schema.columns);
// Explicitly requested relations need their real membership keys, even
// with a minimal parent projection. Missing projection is not SQL NULL.
if (!groupProperties.length && !this.aggregates(query).length) {
for (const field of assemblyFields) if (!requestedFields.includes(field)) requestedFields.push(field);
const relationNames = [...(query.relations ?? []).map((load: any) => load.name),
...(query.relationAggregates ?? []).map((aggregate: any) => aggregate.relationName)];
for (const name of relationNames) {
const relation = schema.relations?.[name];
if (!relation) throw new Error(`Missing relation ${query.entity}.${name}`);
if (!requestedFields.includes(relation.localKey)) requestedFields.push(relation.localKey);
}
}
for (const field of requestedFields) {
if (!schema.columns[field]) throw new Error(`Unknown selected field: ${field}`);
}
let projection = requestedFields.map(field => {
const column = schema.columns[field];
return `${this.driver.identifier(column.columnName)} AS ${this.driver.identifier(field)}`;
}).join(', ');
const aggregates = this.aggregates(query);
if (aggregates.length) {
const aggregateProjection = aggregates.map(aggregate => {
if (!schema.columns[aggregate.field]) {
throw new Error(`Unknown aggregate field: ${aggregate.field}`);
}
aggregateNames.push(aggregate.retName);
return `${this.driver.aggregateFunction(aggregate.func)}(` +
`${this.driver.identifier(schema.columns[aggregate.field].columnName)}) AS ` +
this.driver.identifier(aggregate.retName);
});
projection = [...groupProjection, ...aggregateProjection].join(', ');
}
const partitionBy = query.__teaqlPartitionBy as string | undefined;
const orders = this.orders(query);
const orderClauses = orders.map(order => {
if (!schema.columns[order.field]) {
throw new Error(`Unknown order field: ${order.field}`);
}
const direction = String(order.direction).toLowerCase() === 'desc'
? 'DESC'
: 'ASC';
return `${this.driver.identifier(schema.columns[order.field].columnName)} ${direction}`;
});
if (partitionBy) {
const partitionColumn = schema.columns[partitionBy];
if (!partitionColumn) throw new Error(`Unknown partition field: ${partitionBy}`);
const windowOrder = orderClauses.length ? ` ORDER BY ${orderClauses.join(', ')}` : '';
projection += `, ROW_NUMBER() OVER (PARTITION BY ` +
`${this.driver.identifier(partitionColumn.columnName)}${windowOrder}) AS ` +
this.driver.identifier('__teaql_partition_rank');
}
let sql = `SELECT ${projection} FROM ${this.driver.identifier(schema.table)}`;
const filters = this.filters(query);
const predicates = filters.map(expression =>
this.compileExpression(expression, schema, values),
);
if (schema.columns.version) {
predicates.push(
`${this.driver.identifier(schema.columns.version.columnName)} > 0`,
);
}
if (predicates.length) {
sql += ` WHERE ${predicates.join(' AND ')}`;
}
if (groupFields.length) sql += ` GROUP BY ${groupFields.join(', ')}`;
const limit = query._limit !== undefined
? Number(query._limit)
: Number(query.limitValue || 0);
const offset = query._offset !== undefined
? Number(query._offset)
: Number(query.offsetValue || 0);
if (partitionBy) {
const rank = this.driver.identifier('__teaql_partition_rank');
const predicates: string[] = [];
this.bindValue(values, offset, 'plain');
predicates.push(`${rank} > ${this.driver.placeholder(values.length)}`);
if (limit > 0) {
this.bindValue(values, offset + limit, 'plain');
predicates.push(`${rank} <= ${this.driver.placeholder(values.length)}`);
}
sql = `SELECT * FROM (${sql}) AS ${this.driver.identifier('__teaql_partitioned')} ` +
`WHERE ${predicates.join(' AND ')} ORDER BY ${rank}`;
} else if (orders.length) {
sql += ` ORDER BY ${orderClauses.join(', ')}`;
}
if (!partitionBy && limit > 0) {
this.bindValue(values, limit, 'plain');
sql += ` LIMIT ${this.driver.placeholder(values.length)}`;
}
if (!partitionBy && offset > 0) {
this.bindValue(values, offset, 'plain');
sql += ` OFFSET ${this.driver.placeholder(values.length)}`;
}
if (recordTrace) this.sqlTrace.push(sql);
return { sql, values, aggregateNames };
}
async executeQuery<T = any>(query: any): Promise<T[]> {
const request = query instanceof QueryRequest ? query : new QueryRequest(query);
// Keep the existing builder hard-limit normalization contract. Intent is
// already validated; the private execution snapshot is normalized again.
if (!(query instanceof QueryRequest) && query?.[this.internalQueryToken] !== true
&& typeof query?.prepareForList === 'function') query.prepareForList();
return this.executeQueryWithIntent<T>(request.query, this.derivedQueryBindings.get(query),
this.derivedRelationAssembly.get(query));
}
private async executeDerivedQuery<T>(query: object, inherited: SQLLogBindingSource | undefined,
intent: QueryIntent, assembly?: RelationAssembly): Promise<T[]> {
// Only fresh execution-local child requests are registered. Keep virtual
// executeQuery dispatch for existing subclasses; no provenance on wire data.
const captured = new QueryRequest(query, intent).query;
if (inherited) this.derivedQueryBindings.set(captured, inherited);
if (assembly) this.derivedRelationAssembly.set(captured, assembly);
try { return await this.executeQuery<T>(captured); }
finally {
this.derivedQueryBindings.delete(captured);
this.derivedRelationAssembly.delete(captured);
}
}
private descendantBindings(query: any, sql: string, values: any[], inherited?: SQLLogBindingSource): SQLLogBindingSource | undefined {
if (!query.relations?.length && !query.relationAggregates?.length) return undefined;
return inheritSQLLogBindings(this.queryLogBindings(sql, values), inherited);
}
private async queryTreeBindings(query: any, sql: string, values: any[], inherited?: SQLLogBindingSource): Promise<SQLLogBindingSource> {
let source = inheritSQLLogBindings(this.queryLogBindings(sql, values), inherited);
const intent = new QueryRequest(query).intent;
const stack: any[] = [query];
const visited = new Set<object>();
while (stack.length) {
const current = stack.pop();
if (!current || visited.has(current)) continue;
visited.add(current);
if (current !== query) {
// Pure compilation, not a physical query or an sqlTrace entry.
const compiled = await this.compileQuery(new QueryRequest(current, intent).query, false);
source = inheritSQLLogBindings(this.queryLogBindings(compiled.sql, compiled.values), source);
}
const origin = queryDiagnosticOrigin(current);
if (origin) stack.push(origin);
for (const child of [...(current.relations ?? []), ...(current.relationAggregates ?? [])]) {
if (child.query) stack.push(child.query);
}
for (const facet of current.facets ?? []) {
if (!facet.query) continue;
// Facet aggregates describe membership rows from the outer entity.
// The nested query only selects the facet value object itself, so its
// diagnostic compilation must mirror execution and omit those measures.
const facetValueQuery = facet.query.clone();
facetValueQuery.aggregateItems = [];
facetValueQuery.groupByItems = [];
stack.push(facetValueQuery);
}
}
return source;
}
private async executeQueryWithIntent<T = any>(query: any, inherited?: SQLLogBindingSource,
assembly?: RelationAssembly): Promise<T[]> {
await this.telemetryStartup;
const scope = startRuntimeOperation(this.runtimeTelemetry, {
family: 'query',
name: `${String(query?.entity || 'unknown')}.list`,
attributes: { 'teaql.entity.type': String(query?.entity || 'unknown') },
});
const work = async () => { try {
const internal = query?.[this.internalQueryToken] === true;
if (!internal) {
if (typeof query?.prepareForList !== 'function') {
throw new Error('TeaQL list execution requires the formal runtime SelectQuery');
}
query.prepareForList();
}
const idSetPrepared = internal ? { query, execution: undefined } : await this.prepareIdSetPage(query, inherited);
const prepared = internal ? idSetPrepared : await this.prepareContinuousPage(idSetPrepared.query);
query = new QueryRequest(query).withQuery(prepared.query).query;
const { sql, values, aggregateNames } = await this.compileQuery(query, true, assembly ? [assembly.field] : []);
inherited = await this.queryTreeBindings(query, sql, values, inherited);
const result = await this.executeLoggedSQL('select', sql, values, this.queryLogIntent(query),
() => this.driver.query(sql, values), inherited);
const rows = result.rows.map(row =>
this.decodeRow(query.entity, row, aggregateNames),
);
if (idSetPrepared.execution?.pageIds) {
const positions = new Map<string, number>(
idSetPrepared.execution.pageIds.map(
(id: any, index: number): [string, number] => [String(id), index],
),
);
rows.sort((left: any, right: any) =>
(positions.get(String(left.id)) ?? Number.MAX_SAFE_INTEGER)
- (positions.get(String(right.id)) ?? Number.MAX_SAFE_INTEGER));
}
const descendantBindings = this.descendantBindings(query, sql, values, inherited);
if (assembly) for (const row of rows) assembly.keys.set(row, row[assembly.field]);
await this.enhanceQueryRows(rows, query, descendantBindings);
if (assembly && Array.isArray(query.selectItems) && query.selectItems.length
&& !['id', 'version', ...query.selectItems, ...this.groupBy(query)].includes(assembly.field)) {
const schema = this.schema(query.entity);
const selectedRelations = [...(query.relations ?? []).map((load: any) => load.name),
...(query.relationAggregates ?? []).map((aggregate: any) => aggregate.relationName)];
if (!selectedRelations.some(name => schema.relations?.[name]?.localKey === assembly.field))
for (const row of rows) delete row[assembly.field];
}
if (!internal) await this.registerContinuousPage(query, prepared.execution, rows);
scope.success({ attributes: { 'teaql.result.cardinality': rows.length } });
return rows as T[];
} catch (error) {
scope.failure(error);
throw error;
} };
return scope.run ? scope.run(work) : work();
}
private async prepareIdSetPage(query: any, inherited?: SQLLogBindingSource): Promise<{ query: any; execution?: any }> {
const options = query.localIdSetPaginationOptions?.();
const runtime = query.localIdSetPaginationRuntime?.();
if (!options || !runtime) { runtime?.observe('ID_SET_DISABLED'); return { query }; }
const limit = this.queryLimit(query);
if (!limit || query.localContinuousPageOptions?.() || this.aggregates(query).length
|| this.groupBy(query).length || query.__teaqlPartitionBy) {
runtime.observe('ID_SET_FALLBACK_UNSUPPORTED_SHAPE');
return { query };
}
if ((query.orderItems || []).some((order: any) => order.expr != null)) {
runtime.observe('ID_SET_FALLBACK_NON_DETERMINISTIC_ORDER');
return { query };
}
if (!this.orders(query).some(order => order.field === 'id')) {
query = query.clone().order(OrderBy.asc('id'));
}
const normalized = query.clone();
normalized.offsetValue = 0;
normalized.limitValue = 0;
normalized.selectItems = [];
normalized.relations = [];
normalized.commentText = undefined;
normalized.purposeText = undefined;
normalized.clearIdSetPaginationRuntime();
normalized.clearContinuousPageRuntime();
const rawKey = `${options.namespace}|${runtime.scope}|${JSON.stringify(normalized)}`;
let hash = 2166136261;
for (let i = 0; i < rawKey.length; i++) hash = Math.imul(hash ^ rawKey.charCodeAt(i), 16777619);
const queryKey = `teaql:id-set:v1:${(hash >>> 0).toString(16)}`;
let retained: any;
let cacheHit = false;
try { retained = runtime.get(queryKey); }
catch { runtime.observe('ID_SET_FALLBACK_STORE_UNAVAILABLE'); return { query }; }
if (!retained) {
let buildResult: any;
try {
buildResult = await runtime.build(queryKey, async () => {
const existing = runtime.get(queryKey);
if (existing) return existing;
const idQuery = query.clone();
idQuery.selectItems = ['id'];
idQuery.relations = [];
idQuery.relationAggregates = [];
idQuery.facets = [];
idQuery.offsetValue = 0;
idQuery.limitValue = options.maxIds + 1;
idQuery.clearIdSetPaginationRuntime();
idQuery.clearContinuousPageRuntime();
(idQuery as any)[this.internalQueryToken] = true;
const helper = new QueryRequest(query).deriveHelper(idQuery, `collect ${query.entity} IDs for pagination`);
// Classification is pure compilation, not another SQL execution.
// The collector drops relations, but its inherited prose can still
// mention a secret bound only by one of those future requests.
const diagnostic = await this.compileQuery(query, false);
const helperBindings = await this.queryTreeBindings(query, diagnostic.sql, diagnostic.values, inherited);
const idRows = await this.executeDerivedQuery<any>(helper.query,
helperBindings, helper.intent);
if (idRows.length > options.maxIds) return undefined;
let ids: BigUint64Array;
try { ids = BigUint64Array.from(idRows.map(row => BigInt(row.id))); }
catch { throw new Error('ID_SET_UNSUPPORTED_ID'); }
const value = { ids, expiresAt: Date.now() + options.ttlSeconds * 1000 };
runtime.put(queryKey, value);
return value;
});
} catch (error) {
runtime.observe(error instanceof Error && error.message === 'ID_SET_UNSUPPORTED_ID'
? 'ID_SET_FALLBACK_UNSUPPORTED_SHAPE'
: 'ID_SET_FALLBACK_STORE_UNAVAILABLE');
return { query };
}
retained = buildResult.value;
if (!retained) {
runtime.observe('ID_SET_FALLBACK_LIMIT_EXCEEDED', options.maxIds + 1);
return { query };
}
cacheHit = !buildResult.built;
runtime.observe(cacheHit ? 'ID_SET_HIT' : 'ID_SET_BUILD', retained.ids.length);
} else {
cacheHit = true;
runtime.observe('ID_SET_HIT', retained.ids.length);
}
const offset = Number(query.offsetValue || query._offset || 0);
const pageIds = Array.from(retained.ids.slice(offset, offset + limit));
const page = query.clone();
page.offsetValue = 0;
page.limitValue = Math.max(pageIds.length, 1);
page.filterCondition = page.filterCondition
? { $and: [page.filterCondition, { id: { $in: pageIds } }] }
: { id: { $in: pageIds } };
page.clearIdSetPaginationRuntime();
runtime.observe(cacheHit ? 'ID_SET_HIT' : 'ID_SET_BUILD', retained.ids.length);
return { query: page, execution: { pageIds, totalCount: retained.ids.length } };
}
async executeFacetMembership(
outerQuery: SelectQuery,
relationName: string,
countAliases: readonly string[] = [],
): Promise<Map<string, number>> {
const aggregates = await this.executeFacetAggregates(outerQuery, relationName, [{
function: 'Count', field: 'id', alias: '__teaql_facet_count',
}]);
return new Map([...aggregates].map(([key, values]) => [
key, Number(values.__teaql_facet_count ?? 0),
]));
}
async executeFacetAggregates(
outerQuery: SelectQuery,
relationName: string,
aggregates: readonly FacetAggregate[],
): Promise<Map<string, Record<string, unknown>>> {
const request = new QueryRequest(outerQuery);
outerQuery = request.query;
const query = request.withQuery(outerQuery.clone()).query;
retainQueryDiagnosticOrigin(outerQuery, query);
query.facets = [];
query.relations = [];
query.relationAggregates = [];
query.orderItems = [];
query.offsetValue = 0;
query.limitValue = 0;
query.selectItems = [];
query.groupByItems = [relationName];
query.aggregateItems = aggregates.map(aggregate => ({ ...aggregate }));
(query as any)[this.internalQueryToken] = true;
const rows = await this.executeQuery<Record<string, unknown>>(query);
return new Map(rows.flatMap(row => {
const value = row[relationName];
if (value === null || value === undefined) return [];
const result: Record<string, unknown> = {};
for (const aggregate of aggregates) result[aggregate.alias] = row[aggregate.alias];
return [[String(value), result]];
}));
}
async executeCount(query: any, countAlias?: string, operation?: QueryOperation): Promise<number> {
const request = query instanceof QueryRequest ? query : new QueryRequest(query);
query = request.query;
if (typeof query?.forExactCount !== 'function') {
throw new Error('TeaQL exact count requires the formal runtime SelectQuery');
}
const alias = countAlias ?? '__teaql_total';
const countRequest = request.deriveOperation(query.forExactCount(alias), operation ?? {
kind: 'named-count', ...(countAlias !== undefined ? { alias: countAlias } : {}),
});
const rows = await this.executeQuery<Record<string, unknown>>(countRequest.query);
const value = rows[0]?.[alias];
if (typeof value !== 'number' || !Number.isFinite(value)) {
throw new Error(`TeaQL provider did not return exact count alias ${alias}`);
}
return value;
}
private async prepareContinuousPage(query: any): Promise<{ query: any; execution?: any }> {
const options = query.localContinuousPageOptions?.();
const runtime = query.localContinuousPageRuntime?.();
if (!options || !runtime) { runtime?.observe('DISABLED'); return { query }; }
const orders = this.orders(query);
const limit = this.queryLimit(query);
if (!limit || orders.length !== 1 || orders[0].field !== 'id' || this.aggregates(query).length || this.groupBy(query).length || query.__teaqlPartitionBy) {
runtime.observe('OFFSET_FALLBACK:UNSUPPORTED_QUERY_SHAPE'); return { query };
}
const clone = Object.assign(Object.create(Object.getPrototypeOf(query)), query);
clone.orderItems = [...(query.orderItems || [])]; clone.relations = [...(query.relations || [])];
Object.defineProperty(clone, 'continuousPageFetchOptions', { enumerable: false, writable: true, value: options });
Object.defineProperty(clone, 'continuousPageRuntimeContext', { enumerable: false, writable: true, value: runtime });
const offset = Number(query.offsetValue || query._offset || 0);
const normalized = { ...JSON.parse(JSON.stringify(clone)), offsetValue: 0, _offset: 0, commentText: undefined, purposeText: undefined };
const rawKey = `${options.namespace}|${runtime.owner}|${JSON.stringify(normalized)}`;
let hash = 2166136261; for (let i = 0; i < rawKey.length; i++) hash = Math.imul(hash ^ rawKey.charCodeAt(i), 16777619);
const queryKey = `teaql:continuous-page:v1:${(hash >>> 0).toString(16)}`;
const execution: any = { queryKey, offset, limit, direction: String(orders[0].direction).toLowerCase(), ttlSeconds: options.ttlSeconds, runtime, optimized: false };
if (offset === 0) { runtime.observe('OFFSET_FALLBACK:FIRST_PAGE'); return { query: clone, execution }; }
let cursor: any; try {
cursor = await observeRuntimeOperation(this.runtimeTelemetry, {
family: 'cache',
name: 'continuous_page.get',
attributes: { 'teaql.cache.operation': 'get' },
}, () => runtime.get(queryKey, offset));
} catch { runtime.observe('OFFSET_FALLBACK:STORE_UNAVAILABLE'); return { query: clone, execution }; }
if (!cursor) { runtime.observe('OFFSET_FALLBACK:CACHE_MISS'); return { query: clone, execution }; }
const seek = { id: { [execution.direction === 'desc' ? '$lt' : '$gt']: cursor.boundary } };
clone.filterCondition = clone.filterCondition ? { $and: [clone.filterCondition, seek] } : seek;
clone.offsetValue = 0; if (clone._offset !== undefined) clone._offset = 0;
execution.optimized = true; execution.cursorId = cursor.cursorId;
runtime.observe('CURSOR_SEEK', cursor.cursorId);
return { query: clone, execution };
}
private async registerContinuousPage(_query: any, execution: any, rows: any[]): Promise<void> {
if (!execution || rows.length !== execution.limit || !rows.length || rows[rows.length - 1].id === undefined) return;
try {
await observeRuntimeOperation(this.runtimeTelemetry, {
family: 'cache',
name: 'continuous_page.put',
attributes: { 'teaql.cache.operation': 'put' },
}, () => execution.runtime.put(execution.queryKey, execution.offset + rows.length, {
cursorId: `cpg_${Date.now().toString(16)}_${Math.random().toString(16).slice(2)}`,
boundary: rows[rows.length - 1].id,
expiresAt: Date.now() + execution.ttlSeconds * 1000,
}));
} catch { execution.runtime.observe('OFFSET_FALLBACK:STORE_UNAVAILABLE'); return; }
if (execution.optimized) execution.runtime.observe('CURSOR_SEEK', execution.cursorId);
}
executeForStream<T = any>(query: any, chunkSize = 1000): AsyncIterable<T[]> {
// Capture before returning the lazy iterator. No SQL/cursor is opened here.
// Preserve the asynchronous failure contract, but never re-read a builder
// that the caller may "repair" or repurpose before the first poll.
try {
const request = query instanceof QueryRequest
? new QueryRequest(query.query, query.intent) : new QueryRequest(query);
return this.executeCapturedStream<T>(request.query, chunkSize);
} catch (error) {
return (async function* (): AsyncIterable<T[]> { throw error; })();
}
}
private async *executeCapturedStream<T>(query: any, chunkSize: number): AsyncIterable<T[]> {
await this.telemetryStartup;
if (!Number.isInteger(chunkSize) || chunkSize <= 0) {
throw new Error('stream chunk size must be a positive integer');
}
if (Array.isArray(query.facets) && query.facets.length > 0) {
throw new Error('QRY-F01_STREAM_UNSUPPORTED: execute facets with executeForList');
}
const { sql, values, aggregateNames } = await this.compileQuery(query);
const inherited = await this.queryTreeBindings(query, sql, values);
const queryScope = startRuntimeOperation(this.runtimeTelemetry, { family: 'query',
name: `${String(query.entity)}.stream`, attributes: { 'teaql.entity.type': String(query.entity) } });
let providerScope!: RuntimeTelemetryScope;
const startProvider = async () => { providerScope = startRuntimeOperation(this.runtimeTelemetry, {
family: 'provider', name: `${this.driver.databaseKind}.stream`, attributes: {
'teaql.provider.kind': this.driver.databaseKind, 'teaql.provider.operation': 'query' } }); };
await (queryScope.run ? queryScope.run(startProvider) : startProvider());
const correlation = providerScope.correlation?.();
const iterator = this.driver.stream(sql, values)[Symbol.asyncIterator]();
const startedAt = Date.now();
let outcome: SQLExecutionOutcome = 'cancelled';
let delivered = 0;
let chunk: any[] = [];
const descendantBindings = this.descendantBindings(query, sql, values, inherited);
try {
while (true) {
const next = await (providerScope.run ? providerScope.run(() => iterator.next()) : iterator.next());
if (next.done) break;
chunk.push(this.decodeRow(query.entity, next.value, aggregateNames));
if (chunk.length === chunkSize) {
await this.enhanceQueryRows(chunk, query, descendantBindings);
delivered += chunk.length;
yield chunk as T[];
chunk = [];
}
}
if (chunk.length) {
await this.enhanceQueryRows(chunk, query, descendantBindings);
delivered += chunk.length;
yield chunk as T[];
}
outcome = 'success';
} catch (error) {
outcome = 'failure';
providerScope.failure(error); queryScope.failure(error);
throw error;
} finally {
try { await iterator.return?.(); }
catch (error) {
if (outcome !== 'failure') {
outcome = 'failure';
providerScope.failure(error); queryScope.failure(error);
throw error;
}
} finally {
// Generator.return()/consumer break also closes the driver's cursor. Report
// only delivered rows, not prefetched rows, and never stringify an error.
if (outcome === 'failure') {
try { this.recordSQL('select', sql, values, startedAt, delivered, undefined,
this.queryLogIntent(query), outcome, inherited, correlation); } catch { /* preserve driver failure */ }
} else {
this.recordSQL('select', sql, values, startedAt, delivered, undefined,
this.queryLogIntent(query), outcome, inherited, correlation);
providerScope.success({ attributes: { 'teaql.result.cardinality': delivered } });
queryScope.success({ attributes: { 'teaql.result.cardinality': delivered } });
}
}
}
}
private async enhanceQueryRows(rows: any[], query: any, inherited?: SQLLogBindingSource): Promise<void> {
if (!rows.length) return;
const names = [...(query.relations ?? []).map((load: any) => load.name),
...(query.relationAggregates ?? []).map((aggregate: any) => aggregate.relationName)];
if (!names.length) return;
const schema = this.schema(query.entity);
const keys = new Map<string, unknown[]>();
for (const name of names) {
const relation = schema.relations?.[name];
if (!relation) throw new Error(`Missing relation ${query.entity}.${name}`);
if (!keys.has(relation.localKey)) keys.set(relation.localKey, rows.map(row => row[relation.localKey]));
}
// Lists and each streamed chunk capture before aliases or sibling loads can
// replace any scalar key. Keep existing execution order and trace sources.
await this.enhanceRelationAggregates(rows, query, keys, inherited);
await this.enhanceRelations(rows, query, keys, inherited);
}
private async enhanceRelations(parents: any[], query: any, keys: ReadonlyMap<string, readonly unknown[]>,
inherited?: SQLLogBindingSource): Promise<void> {
if (!parents.length || !Array.isArray(query.relations) || !query.relations.length) return;
const parentSchema = this.schema(query.entity);
const intent = new QueryRequest(query).intent;
for (const load of query.relations) {
const relation = parentSchema.relations?.[load.name];
if (!relation) throw new Error(`Missing relation ${query.entity}.${load.name}`);
const parentKeys = keys.get(relation.localKey)!;
const parentIds = parentKeys
.filter(value => value !== undefined && value !== null);
const limit = load.query ? this.queryLimit(load.query) : undefined;
const threshold = load.query?.localTopNProbeParentThreshold?.();
const boundedTopN = relation.many && limit !== undefined;
const requestedChildFields = Array.isArray(load.query?.selectItems)
? load.query.selectItems
: [];
const implicitRelationKey = requestedChildFields.length > 0
&& !requestedChildFields.includes(relation.foreignKey);
const providerAlwaysProbe = this.driver.topNRelationPlanPolicy === 'alwaysProbe';
const useProbes = boundedTopN && (threshold === undefined
? providerAlwaysProbe
: threshold > 0 && parentIds.length <= threshold);
const selectedPlan = useProbes ? 'bounded_probes' : 'window';
const relationScope = startRuntimeOperation(this.runtimeTelemetry, {
family: 'relation_load',
name: `${query.entity}.${String(load.name)}`,
attributes: {
'teaql.entity.type': String(query.entity),
'teaql.relation.name': String(load.name),
'teaql.relation.parent_count': parentIds.length,
'teaql.relation.per_parent_limit': limit ?? 0,
'teaql.relation.configured_probe_threshold': threshold ?? 'provider-default',
'teaql.relation.selected_plan': selectedPlan,
'teaql.relation.probe_count': useProbes ? parentIds.length : 0,
},
});
try {
if (!parentIds.length && !(relation.many && load.query?.facets?.length)) {
for (const parent of parents) parent[load.name] = relation.many ? [] : null;
relationScope.success({ attributes: { 'teaql.result.cardinality': 0 } });
continue;
}
const childQuery = {
...(load.query || {}),
entity: relation.targetEntity,
// Relation assembly keys belong to the runtime protocol. Callers should
// not have to expose selectId()/selectOrderId() in their domain query.
selectItems: implicitRelationKey
? [...requestedChildFields, relation.foreignKey]
: requestedChildFields,
_filters: [...this.filters(load.query || {})],
relations: [...(load.query?.relations || [])],
orderItems: [...(load.query?.orderItems || [])],
__teaqlPartitionBy: boundedTopN && !useProbes
? relation.foreignKey
: undefined,
};
if (boundedTopN) {
// Aggregate rows have group identity, not source-row identity. An
// ungrouped id is invalid on strict SQL engines and cannot break ties.
const groups = this.groupBy(childQuery);
const stableFields = groups.length || this.aggregates(childQuery).length ? groups : ['id'];
for (const field of stableFields) {
if (!childQuery.orderItems.some((order: any) => order.field === field)) {
childQuery.orderItems.push(OrderBy.asc(field));
}
}
}
if (typeof childQuery.clearContinuousPageRuntime === 'function') childQuery.clearContinuousPageRuntime();
childQuery[this.internalQueryToken] = true;
const children: any[] = [];
const assembly: RelationAssembly = { field: relation.foreignKey, keys: new WeakMap() };
if (useProbes) {
for (const parentId of parentIds) {
const probeQuery = {
...childQuery,
_filters: [...childQuery._filters, { [relation.foreignKey]: { $eq: parentId } }],
__teaqlPartitionBy: undefined,
};
children.push(...await this.executeDerivedQuery<any>(
new QueryRequest(query).derive(probeQuery, load.name, load.operation).query, inherited, intent, assembly));
}
} else {
childQuery._filters.push({ [relation.foreignKey]: { $in: parentIds } });
children.push(...await this.executeDerivedQuery<any>(
new QueryRequest(query).derive(childQuery, load.name, load.operation).query, inherited, intent, assembly));
}
for (const child of children) delete child.__teaql_partition_rank;
const buckets = new Map<any, any[]>();
for (const child of children) {
if (!assembly.keys.has(child)) throw new Error('Relation assembly lost its captured row identity');
const key = assembly.keys.get(child);
const bucket = buckets.get(key) || [];
bucket.push(child);
buckets.set(key, bucket);
}
if (implicitRelationKey) {
for (const child of children) delete child[relation.foreignKey];
}
for (const [index, parent] of parents.entries()) {
const related = buckets.get(parentKeys[index]) || [];
if (relation.many && load.query?.facets?.length) {
// Each loaded collection owns its own facet membership. The full
// filtered child set, not its Top-N page or other parents, is counted.
const facetQuery: SelectQuery = load.query.clone();
facetQuery.entity = relation.targetEntity;
// A missing local key means no relationship, never orphan rows with
// a NULL foreign key. Empty IN compiles to an unsatisfiable predicate.
const key = parentKeys[index];
const membership = { [relation.foreignKey]: key === null || key === undefined
? { $in: [] } : { $eq: key } };
facetQuery.filterCondition = facetQuery.filterCondition
? { $and: [facetQuery.filterCondition, membership] } : membership;
retainQueryDiagnosticOrigin(query, facetQuery);
const derived = new QueryRequest(query).derive(facetQuery, load.name, load.operation).query;
parent[load.name] = new SmartList(related, { facets:
await executeRelationFacets(this, item => item, derived, facetQuery.facets) });
} else {
// A failed detail selection does not erase the captured FK. Missing
// target properties remain absent (NotLoaded), never loaded-null.
const key = parentKeys[index];
parent[load.name] = relation.many ? related : (related[0] ??
(key === null || key === undefined ? null : { [relation.foreignKey]: key }));
}
}
relationScope.success({ attributes: { 'teaql.result.cardinality': children.length } });
} catch (error) {
relationScope.failure(error);
throw error;
}
}
}
private async enhanceRelationAggregates(parents: any[], query: any, keys: ReadonlyMap<string, readonly unknown[]>,
inherited?: SQLLogBindingSource): Promise<void> {
const aggregates = query.relationAggregates;
if (!parents.length || !Array.isArray(aggregates) || !aggregates.length) return;
const parentSchema = this.schema(query.entity);
const intent = new QueryRequest(query).intent;
for (const aggregate of aggregates) {
const relation = parentSchema.relations?.[aggregate.relationName];
if (!relation) throw new Error(`Missing relation ${query.entity}.${aggregate.relationName}`);
const parentKeys = keys.get(relation.localKey)!;
const parentIds = parentKeys
.filter(value => value !== undefined && value !== null);
if (!parentIds.length) {
for (const parent of parents) {
parent[aggregate.alias] = aggregate.singleResult
? this.emptyAggregateValue(aggregate.query)
: {};
}
continue;
}
const childQuery: SelectQuery = aggregate.query.clone();
childQuery.entity = relation.targetEntity;
childQuery.selectItems = [];
childQuery.properties = [];
childQuery.orderItems = [];
childQuery.limitValue = 0;
childQuery.offsetValue = 0;
childQuery.relations = [];
childQuery.relationAggregates = [];
if (!childQuery.aggregateItems.length) {
childQuery.aggregate('Count', 'id', aggregate.alias);
}
if (!childQuery.groupByItems.includes(relation.foreignKey)) {
childQuery.groupBy(relation.foreignKey);
}
const membership = { [relation.foreignKey]: { $in: parentIds } };
childQuery.filterCondition = childQuery.filterCondition
? { $and: [childQuery.filterCondition, membership] } : membership;
(childQuery as any)[this.internalQueryToken] = true;
const rows = await this.executeDerivedQuery<any>(
new QueryRequest(query).deriveAggregate(childQuery, aggregate.relationName, aggregate.operation ?? {
kind: 'relation-count', relation: aggregate.relationName, alias: aggregate.alias,
}).query, inherited, intent);
const buckets = new Map<any, any>();
for (const row of rows) buckets.set(row[relation.foreignKey], row);
for (const [index, parent] of parents.entries()) {
const row = buckets.get(parentKeys[index]);
if (!row) {
parent[aggregate.alias] = aggregate.singleResult
? this.emptyAggregateValue(aggregate.query)
: {};
} else if (aggregate.singleResult) {
parent[aggregate.alias] = row[childQuery.aggregateItems[0].alias] ?? null;
} else {
parent[aggregate.alias] = Object.fromEntries(
Object.entries(row).filter(([key]) => key !== relation.foreignKey),
);
}
}
}
}
private emptyAggregateValue(query: SelectQuery): unknown {
const aggregate = query.aggregateItems[0];
return !aggregate || String(aggregate.function).toLowerCase() === 'count' ? 0 : null;
}
private queryLimit(query: any): number | undefined {
if (query._limit !== undefined) return Number(query._limit);
if (query.limitValue !== undefined && Number(query.limitValue) > 0) {
return Number(query.limitValue);
}
return undefined;
}
async close(): Promise<void> {
try { await this.driver.close(); }
finally {
await this.telemetryStartup;
await this.standardTelemetryHandle?.release();
this.standardTelemetryHandle = undefined;
}
}
}
export function assertSafeIdentifier(value: string): string {
if (!/^[A-Za-z_][A-Za-z0-9_]*$/.test(value)) {
throw new Error(`Unsafe SQL identifier: ${value}`);
}
return value;
}
export function standardAggregateFunction(name: string): string {
const functions: Record<string, string> = {
count: 'COUNT',
sum: 'SUM',
avg: 'AVG',
min: 'MIN',
max: 'MAX',
};
const sqlFunction = functions[String(name).toLowerCase()];
if (!sqlFunction) throw new Error(`Unsupported aggregate: ${name}`);
return sqlFunction;
}