API Reference

The public API is the Drasi class exported from @drasi/lib. There are 42 methods across 8 areas. Every method except pluginKinds(), sourceConfigSchema(), reactionConfigSchema(), and bootstrapConfigSchema() is async (returns a Promise).

Method names are shown in JavaScript (camelCase) form. Concrete TypeScript types for every parameter, return value, and callback payload are emitted into the generated index.d.ts — see TypeScript types.

import { Drasi } from '@drasi/lib';

Surface at a glance

¹ static factory   ² synchronous (does not return a Promise)

Construction

Drasi.create(id, options?) → Promise<Drasi>

Static. Create a new, not-yet-started engine instance.

options:

• secrets?: Record<string, string> — seeds an in-memory secret store that cdylib plugins resolve ConfigValue::Secret references against. Non-string values are ignored.
• stateStore?: { kind: 'redb', path: string } — enables a persistent plugin state store (also required by durable reactions).
• indexStore?: { kind: 'rocksdb', path: string, enableArchive?: boolean, directIo?: boolean } — enables a persistent query-index backend so continuous-query indexes and the reaction outbox survive restarts. RocksDB holds a process-exclusive lock on path, so a given path may be used by only one engine at a time.
• identity?: { kind: 'password' | 'token', username?, password?, token? } — wires a built-in identity provider that injects credentials into sources/reactions.

Returns a Drasi instance.

Drasi.fromConfig(config) → Promise<Drasi>

Static. Build an engine from a declarative config object and start it. Equivalent to create + optional loadPlugins + start + adding each declared source, query, and reaction (which auto-start on the running engine).

config (DrasiConfig):

• id?: string (default "drasi")
• secrets?, stateStore?, indexStore?, identity? — forwarded to create.
• pluginsDir?: string — if present, loadPlugins(pluginsDir) runs before start.
• sources?: Array<{ kind, id, config?, autoStart?, bootstrap? }>
• queries?: Array<{ id, query, sources, language?, joins? }>
• reactions?: Array<{ kind, id, queries, config? }>

const drasi = await Drasi.fromConfig({
  id: 'app',
  pluginsDir: './plugins',
  sources: [{ kind: 'mock', id: 'counters', config: { dataType: { type: 'counter' }, intervalMs: 300 } }],
  queries: [{ id: 'big', query: 'MATCH (c:Counter) WHERE c.value > 3 RETURN c.value AS value', sources: ['counters'] }],
});

Plugins

Plugins are self-contained cdylib files (.so/.dylib/.dll) loaded through drasi-host-sdk, exactly like drasi-server. See Working with plugins.

loadPlugins(dir, verify?) → Promise<{ plugins, sources, reactions, bootstrap }>

Discover and load all matching cdylib plugins from dir, registering their descriptors so their kinds become usable by addSource/addReaction.

• dir: string — directory to scan.
• verify?: Record<string, string> — a { filename: sha256hex } integrity allowlist. When provided, only files whose contents hash to the expected value are loaded; mismatches are skipped with a warning.

Returns counts { plugins, sources, reactions, bootstrap } (all number).

watchPlugins(dir) → Promise<void>

Watch dir and hot-(re)load plugins as files are added or changed (1s debounce). Removed files deregister their kinds. Reload failures are logged, not thrown.

listPluginTags(repository) → Promise<string[]>

List available tags for a plugin repo in the configured OCI registry (default ghcr.io/drasi-project), e.g. listPluginTags("source/postgres").

pullPlugin(reference, destDir, filename, options?) → Promise<{ path, verification }>

Download a plugin artifact from an OCI registry to destDir/filename.

• reference: string — full OCI reference, e.g. ghcr.io/drasi-project/source/postgres:0.1.13-windows-msvc-amd64.
• options?: PullPluginOptions — opt-in cosign enforcement:
  • verify?: boolean — run signature verification; a tampered or valid-but-untrusted artifact is rejected.
  • requireSigned?: boolean — additionally reject unsigned artifacts. Implies verify.
  • trustedIdentities?: { issuer, subjectPattern }[] — signer allowlist (defaults to the drasi-project GitHub Actions identity).

Returns { path: string, verification }, where verification is one of:

{ status: 'unsigned' }                    // no signature (or verify not requested)
{ status: 'verified', issuer, subject }   // valid signature chaining to Sigstore
{ status: 'tampered', reason }            // a signature exists but did not verify

After a successful pull, call loadPlugins(destDir) (or watchPlugins) to register it.

sourceConfigSchema(kind) / reactionConfigSchema(kind) / bootstrapConfigSchema(kind) → { name, schema }

Synchronous. Return the config schema a registered plugin kind declares, as { name: string, schema: Record<string, unknown> }. Use it to validate config with a JSON-schema validator (such as ajv) before calling addSource/addReaction. Because these are synchronous, an unregistered kind throws with a typed err.code (UNKNOWN_SOURCE_KIND / UNKNOWN_REACTION_KIND / UNKNOWN_BOOTSTRAP_KIND).

pluginKinds() → { sources, reactions, bootstrap }

Synchronous. Return the currently registered kinds as { sources: string[], reactions: string[], bootstrap: string[] }.

Sources

addSource(kind, id, config, autoStart?, bootstrap?) → Promise<void>

Add a source instance of a registered plugin kind.

addJsSource(id, autoStart?) → Promise<void>

Register a programmatic source that JavaScript pushes changes into. Default autoStart = true. Maintains a current-state snapshot so late-subscribing queries receive a bootstrap replay of live elements. See JavaScript sources.

pushChange(sourceId, change) → Promise<void>

Push one change into a JS source. Uses a bounded channel (capacity 1024) — the returned promise resolves once the change is buffered, applying backpressure when the buffer is full.

change (SourceChangeInput):

• op: 'insert' | 'update' | 'delete' (also accepts add/remove, case-insensitive) — required.
• id: string — required.
• labels?: string[] | string
• properties?: Record<string, unknown>
• startId / endId (aliases inId / outId) — presence of both makes the change a relation (edge); supplying only one errors.
• effectiveFrom?: number — epoch ms; defaults to now.

updateSource(kind, id, config, autoStart?) → Promise<void>

Replace a source’s configuration in place (same id).

startSource(id) / stopSource(id) → Promise<void>

Start or stop a source by id.

removeSource(id, cleanup?) → Promise<void>

Remove a source; also drops any JS-source sender. cleanup = true tears down external state (default false).

listSources() → Promise<Array<{ id, status }>>

List sources. status is a ComponentStatus string such as "Running" or "Stopped".

Queries

addQuery(id, query, sources, language?, joins?) → Promise<void>

Add a continuous query.

updateQuery(id, query, sources, language?, joins?) → Promise<void>

Replace a query definition in place. Same parameters as addQuery.

startQuery(id) / stopQuery(id) → Promise<void>

Start or stop a query by id.

getQueryResults(id) → Promise<unknown[]>

Return the current result set as an array of row objects.

removeQuery(id) → Promise<void>

Remove a query by id.

listQueries() → Promise<Array<{ id, status }>>

List queries with their status strings.

Reactions

addReaction(kind, id, queryIds, config) → Promise<void>

Add a reaction of a registered plugin kind, subscribing to queryIds.

addJsReaction(id, queryIds, callback) → Promise<void>

Add a JavaScript-defined reaction whose logic is a callback.

• callback is a value-only function (result: QueryResultEvent) => void. It is invoked once per non-empty result batch; empty batches are skipped. The callback is registered as an unref’d (weak) threadsafe function, so it does not keep the Node event loop alive on its own.

See JavaScript reactions.

addDurableJsReaction(id, queryIds, callback, options?) → Promise<void>

Add a durable, checkpointed JavaScript reaction.

• callback is an async function (result: QueryResultEvent) => Promise<void>. The reaction awaits its promise and then persists a per-query checkpoint, so a restart resumes after the last checkpointed sequence.
• options?: { recoveryPolicy?: 'skipGap' | 'strict' } — on a detected gap, skipGap (default) resumes from the latest available sequence; strict fails if the checkpointed position is unavailable.

Requires a durable state store ({ stateStore: { kind: 'redb', path } }), otherwise throws DURABLE_REQUIRES_STATE_STORE synchronously. Checkpoint progress is observable via getReactionMetrics(id).

updateReaction(kind, id, queryIds, config) → Promise<void>

Replace a reaction’s configuration in place.

startReaction(id) / stopReaction(id) → Promise<void>

Start or stop a reaction by id.

removeReaction(id, cleanup?) → Promise<void>

Remove a reaction; cleanup default false.

listReactions() → Promise<Array<{ id, status }>>

List reactions with their status strings.

Metrics

getQueryMetrics(id) → Promise<QueryMetrics>

{
  outboxSize, outboxEarliestSeq, outboxLatestSeq, resultSeqAdvances,
  liveResultsCount, outerTransactionDurationNsLast,
  outerTransactionDurationNsMax, snapshotFetchCount,   // all number
}

getReactionMetrics(id) → Promise<Record<string, ReactionQueryMetrics>>

Keyed by query id:

{
  checkpointSequence, checkpointLag, dedupSkipCount, gapDetectionCount,
  recoveryStrictCount, recoveryAutoResetCount, recoveryAutoSkipGapCount,
  fetchSnapshotCount, fetchOutboxCount,   // all number
}

getLifecycleMetrics() → Promise<LifecycleMetrics>

{
  startupRejectionDurableNoStore, startupRejectionDurableOnVolatile,
  startupRejectionSnapshotSkipGap, startupRejectionNoSnapshotAutoReset,
  autoResetCompletions, hashMismatchCount,   // all number
}

Streaming (events & logs)

All streaming methods take a callback and return Promise<void>. Callbacks are unref’d (weak) and invoked in non-blocking mode.

Component events

onAllEvents(cb), onQueryEvents(id, cb), onSourceEvents(id, cb), onReactionEvents(id, cb) — forward ComponentEvent objects.

Logs

onSourceLogs(id, cb), onQueryLogs(id, cb), onReactionLogs(id, cb) — first replay the buffered log history, then stream live LogMessages:

{ timestamp, level, message, instance_id, component_id, component_type }   // all string

Lifecycle

start() → Promise<void>

Start the engine; auto-start components begin running. Components added to an already-running engine auto-start individually, so either ordering works.

stop() → Promise<void>

Stop the engine. No-op if not running.

close() → Promise<void>

Stop (if running) and release host resources: plugin watchers, JS-source channels, and the config-resolver thread. The instance must not be used after close().

Data shapes

QueryResultEvent

Delivered to JavaScript reactions:

{ query_id: string, sequence: number, timestamp: string,
  results: ResultDiff[], metadata: Record<string, unknown> }

ResultDiff is a tagged union on type: ADD / DELETE ({ data }), UPDATE ({ before, after }), aggregation ({ before?, after }), noop.

Component status

listSources / listQueries / listReactions return { id, status } where status is a string such as "Running" or "Stopped".

Errors

Argument-validation errors throw synchronously with a stable, machine-readable err.code from the exported DrasiErrorCode enum. See Error handling for the full list and the sync-vs-async distinction.