Skip to content

data-engine.mdx

Latest commit

 

History

History
390 lines (300 loc) · 12 KB

data-engine.mdx

File metadata and controls

390 lines (300 loc) · 12 KB
Raw
title IDataEngine Contract
description Reference for the Data Engine contract — the core data persistence layer for CRUD operations, queries, aggregations, and transactions

IDataEngine Contract

The Data Engine is the core persistence layer of ObjectStack. Every data operation — inserts, finds, updates, deletes, counts, and aggregations — flows through this contract. The Kernel delegates to the Data Engine after applying security, validation, and hooks.

**Source:** `packages/spec/src/contracts/data-engine.ts` **Schema:** `packages/spec/src/data/data-engine.zod.ts` **Service name:** `data` (registered via `CoreServiceName`)

Interface Definition

The canonical IDataEngine interface uses QueryAST-aligned parameter names (where, fields, orderBy, limit, offset, expand) — no mechanical translation is needed between the Engine and Driver layers.

import type {
  EngineQueryOptions,
  DataEngineInsertOptions,
  EngineUpdateOptions,
  EngineDeleteOptions,
  EngineAggregateOptions,
  EngineCountOptions,
  DataEngineRequest,
} from '@objectstack/spec/data';

export interface IDataEngine {
  // Query
  find(objectName: string, query?: EngineQueryOptions): Promise<any[]>;
  findOne(objectName: string, query?: EngineQueryOptions): Promise<any>;
  count(objectName: string, query?: EngineCountOptions): Promise<number>;
  aggregate(objectName: string, query: EngineAggregateOptions): Promise<any[]>;

  // Mutation (write ops also accept in-process WriteObservabilityOptions — see `update`)
  insert(objectName: string, data: any | any[], options?: DataEngineInsertOptions & WriteObservabilityOptions): Promise<any>;
  update(objectName: string, data: any, options?: EngineUpdateOptions & WriteObservabilityOptions): Promise<any>;
  delete(objectName: string, options?: EngineDeleteOptions): Promise<any>;

  // AI / Vector Search (optional)
  vectorFind?(objectName: string, vector: number[], options?: {
    where?: any; limit?: number; fields?: string[]; threshold?: number;
  }): Promise<any[]>;

  // Batch Operations (optional, transactional)
  batch?(requests: DataEngineRequest[], options?: { transaction?: boolean }): Promise<any[]>;

  // Raw Command Escape Hatch (optional)
  execute?(command: any, options?: Record<string, any>): Promise<any>;
}

Query Operations

All query methods use canonical QueryAST parameter names: where, fields, orderBy, limit, offset, expand.

find

Executes a structured query with filtering, sorting, pagination, and field selection. Returns an array of records.

const tasks = await engine.find('task', {
  where: {
    status: 'open',
    priority: { $in: ['high', 'critical'] },
  },
  orderBy: [{ field: 'due_date', order: 'asc' }],
  limit: 20,
  offset: 0,
  fields: ['id', 'title', 'status', 'priority', 'due_date'],
});

EngineQueryOptions

Defined by EngineQueryOptionsSchema in @objectstack/spec:

interface EngineQueryOptions {
  where?: FilterCondition;                   // WHERE clause — MongoDB-style $op
  fields?: FieldNode[];                      // SELECT fields
  orderBy?: SortNode[];                      // ORDER BY
  limit?: number;                            // LIMIT
  offset?: number;                           // OFFSET
  top?: number;                              // Alias for limit (OData compat)
  cursor?: Record<string, unknown>;          // Keyset pagination
  search?: FullTextSearch;                   // Full-text search
  expand?: Record<string, QueryAST>;         // Recursive relation loading
  distinct?: boolean;                        // SELECT DISTINCT
  context?: ExecutionContext;                 // Identity, tenant, transaction
}

FilterCondition (where)

Filters use the canonical where + MongoDB-style $op object syntax from FilterConditionSchema:

// Implicit equality
where: { status: 'active' }

// Explicit operators
where: { amount: { $gt: 50000 } }

// Logical combinations
where: {
  $and: [
    { status: 'open' },
    { priority: { $in: ['high', 'critical'] } },
  ],
}

// Logical OR
where: {
  $or: [
    { role: 'admin' },
    { email: { $contains: '@company.com' } },
  ],
}

// Nested relation filter
where: {
  account: { industry: 'tech' },
}

Supported operators: $eq, $ne, $gt, $gte, $lt, $lte, $in, $nin, $between, $contains, $notContains, $startsWith, $endsWith, $null, $exists

Logical operators: $and, $or, $not

findOne

Convenience method that returns the first record matching a query, or null.

const task = await engine.findOne('task', {
  where: { title: 'Implement login page' },
  fields: ['id', 'title', 'status'],
});

count

Returns the number of records matching a filter without fetching data.

const openTasks = await engine.count('task', {
  where: { status: 'open' },
});
// 23

Mutation Operations

insert

Creates one or more records. Returns the inserted record(s) with system-generated fields (id, created_at, etc.).

const task = await engine.insert('task', {
  title: 'Implement login page',
  status: 'open',
  priority: 'high',
  assigned_to: 'usr_01HQ3V5K8N2M4P6R7T9W',
});

console.log(task.id);         // "tsk_01HQ4A7B9D3F5G8J2K4L"
console.log(task.created_at); // "2025-01-20T10:30:00.000Z"

Bulk insert:

const tasks = await engine.insert('task', [
  { title: 'Task 1', status: 'open', priority: 'high' },
  { title: 'Task 2', status: 'open', priority: 'medium' },
  { title: 'Task 3', status: 'open', priority: 'low' },
]);

DataEngineInsertOptions

interface DataEngineInsertOptions {
  returning?: boolean;          // Return inserted record(s)? Default: true
  context?: ExecutionContext;    // Identity, tenant, transaction
}

update

Updates specific fields on matched record(s). The where clause in options identifies target records.

const updated = await engine.update('task', 
  { status: 'in_progress', estimated_hours: 12 },
  { where: { id: 'tsk_01HQ4A7B9D3F5G8J2K4L' } }
);
**Partial Updates:** Only fields included in the `data` parameter are modified. Omitted fields retain their current values.

EngineUpdateOptions

interface EngineUpdateOptions {
  where?: FilterCondition;      // Filter to identify records (WHERE)
  upsert?: boolean;             // Insert if not found? Default: false
  multi?: boolean;              // Update multiple records? Default: false
  returning?: boolean;          // Return updated record(s)? Default: false
  context?: ExecutionContext;
}

WriteObservabilityOptions

The write methods (insert / update) additionally accept an in-process onFieldsDropped listener. The engine invokes it when caller-supplied write fields are legally stripped from the payload before the driver write — static readonly fields or a TRUE readonlyWhen predicate. The write still succeeds; the listener exists so callers that report per-field success (e.g. the flow engine's update_record step) can surface a warning instead of a silent success.

interface WriteObservabilityOptions {
  onFieldsDropped?: (event: DroppedFieldsEvent) => void;
}

interface DroppedFieldsEvent {
  object: string;                        // resolved object name
  fields: string[];                      // caller-supplied fields that were dropped
  reason: 'readonly' | 'readonly_when';  // why they were dropped
}
`onFieldsDropped` is a **TS-contract-level, in-process-only** channel. It is deliberately not part of the serializable Zod options schemas: a function is unrepresentable in JSON Schema and cannot cross the RPC (Virtual Data Engine) boundary, so remote callers never receive these events. A listener that throws never breaks the write — the engine catches and logs.

delete

Deletes record(s) matching the where condition.

await engine.delete('task', {
  where: { id: 'tsk_01HQ4A7B9D3F5G8J2K4L' },
});

EngineDeleteOptions

interface EngineDeleteOptions {
  where?: FilterCondition;      // Filter to identify records (WHERE)
  multi?: boolean;              // Delete multiple records? Default: false
  context?: ExecutionContext;
}

Aggregation

aggregate

Runs aggregation queries with grouping for analytics and reporting.

const result = await engine.aggregate('task', {
  where: { project: 'prj_01HQ3V5K8N' },
  groupBy: ['status'],
  aggregations: [
    { function: 'count', alias: 'task_count' },
    { function: 'sum', field: 'estimated_hours', alias: 'total_hours' },
    { function: 'avg', field: 'estimated_hours', alias: 'avg_hours' },
  ],
});

// result:
// [
//   { status: 'open', task_count: 15, total_hours: 120, avg_hours: 8 },
//   { status: 'in_progress', task_count: 8, total_hours: 96, avg_hours: 12 },
//   { status: 'done', task_count: 24, total_hours: 192, avg_hours: 8 },
// ]

EngineAggregateOptions

interface EngineAggregateOptions {
  where?: FilterCondition;                 // Pre-aggregation filter (WHERE)
  groupBy?: string[];                      // GROUP BY fields
  aggregations?: AggregationNode[];        // Aggregation definitions
  context?: ExecutionContext;
}

interface AggregationNode {
  function: 'count' | 'sum' | 'avg' | 'min' | 'max'
          | 'count_distinct' | 'array_agg' | 'string_agg';
  field?: string;           // Field to aggregate (optional for COUNT(*))
  alias: string;            // Result column alias
  distinct?: boolean;       // Apply DISTINCT before aggregation
  filter?: FilterCondition; // Per-aggregation FILTER WHERE
}

Optional Capabilities

vectorFind (AI/RAG)

Perform similarity search using vector embeddings:

const results = await engine.vectorFind?.('document', embeddingVector, {
  where: { category: 'technical' },
  fields: ['title', 'content'],
  limit: 5,
  threshold: 0.8,
});

batch (Transactional)

Execute multiple operations in a single transaction:

const results = await engine.batch?.([
  { method: 'insert', object: 'project', data: { name: 'Website Redesign', status: 'active' } },
  { method: 'insert', object: 'task', data: { title: 'Design mockups', status: 'open' } },
  { method: 'insert', object: 'task', data: { title: 'Implement frontend', status: 'open' } },
], { transaction: true });

execute (Raw Command)

Escape hatch for raw driver-specific commands:

const result = await engine.execute?.(
  'SELECT * FROM tasks WHERE status = $1',
  { params: ['open'] }
);

Error Codes

Code HTTP Description
RECORD_NOT_FOUND 404 No record exists with the given ID
VALIDATION_FAILED 400 Data does not match the object schema
DELETE_RESTRICTED 409 Cannot delete; dependent child records reference it via a restrict delete behavior

Legacy Compatibility

**Deprecated Parameter Names**

The following legacy parameter names are accepted by the RPC layer for backward compatibility but should not be used in new code. The protocol normalizer resolves conflicts with canonical names taking precedence.

Legacy (Deprecated) Canonical Notes
filter where FilterCondition object
select fields Array of FieldNode
sort orderBy Array of { field, order }
skip offset Number
populate expand Record of field → QueryAST

The deprecated DataEngineQueryOptionsSchema, DataEngineUpdateOptionsSchema, DataEngineDeleteOptionsSchema, and DataEngineAggregateOptionsSchema are maintained in @objectstack/spec for backward compatibility but will be removed in a future major version. Migrate to the QueryAST-aligned equivalents: EngineQueryOptionsSchema, EngineUpdateOptionsSchema, EngineDeleteOptionsSchema, EngineAggregateOptionsSchema.