Writing Plugins

A plugin is a class that implements the DatrixPlugin interface. Datrix provides a BasePlugin abstract class that covers the optional hooks with no-op defaults, so you only need to implement what you need.

BasePlugin vs DatrixPlugin

import { BasePlugin } from "@datrix/core"
import type { PluginContext } from "@datrix/core"

export class MyPlugin extends BasePlugin<MyPluginOptions> {
  readonly name    = "my-plugin"
  readonly version = "1.0.0"

  async init(context: PluginContext): Promise<void> {
    this.context = context
    // setup work here
  }

  async destroy(): Promise<void> {
    // cleanup here
  }
}

Register it in defineConfig:

import { defineConfig } from "@datrix/core"

export default defineConfig(() => ({
  adapter: ...,
  schemas: [...],
  plugins: [new MyPlugin({ ... })],
}))

Initialization order

When defineConfig starts up, plugins are processed in this order:

1. init(context) — called once with adapter, schemas, and config
2. getSchemas() — plugin schemas registered into the registry
3. extendSchemas(ctx) — field/index extensions applied to existing schemas
4. Registry finalized (relations resolved, junction tables created)
5. onSchemaLoad(schemas) — called after registry is fully ready

destroy() is called on datrix.shutdown().

Schema hooks

getSchemas

Return new SchemaDefinition objects the plugin needs. They are registered alongside user schemas before the registry is finalized.

override async getSchemas(): Promise<SchemaDefinition[]> {
  return [
    defineSchema({
      name: "audit_log",
      fields: {
        action:     { type: "string", required: true },
        recordId:   { type: "number", required: true },
        schemaName: { type: "string", required: true },
        userId:     { type: "number" },
        diff:       { type: "json" },
      },
    }),
  ]
}

extendSchemas

Add fields or indexes to existing user schemas. Receives a SchemaExtensionContext with helper methods.

override async extendSchemas(ctx: SchemaExtensionContext): Promise<SchemaExtension[]> {
  // Add createdBy to every schema that has a permission block
  return ctx.extendWhere(
    (schema) => schema.permission !== undefined,
    () => ({
      fields: {
        createdBy: { type: "number" },
      },
    }),
  )
}

SchemaExtensionContext helpers:

onSchemaLoad

Called after the registry is fully finalized. Use this for setup that needs the complete schema list, such as creating prepared statements or building caches.

override async onSchemaLoad(schemas: ISchemaRegistry): Promise<void> {
  const names = schemas.getNames()
  // e.g. pre-warm a schema cache
}

Query hooks

Query hooks are called for every non-raw operation, in this order per query:

onCreateQueryContext  →  onBeforeQuery  →  [query runs]  →  onAfterQuery

All hooks run serially across plugins in registration order. Hooks are skipped for datrix.raw.* calls.

onCreateQueryContext

Called first, before the query is dispatched. Use it to enrich the context — for example, inject the authenticated user so it's available in onBeforeQuery and permission functions.

This is how ApiPlugin injects the request user:

override async onCreateQueryContext(context: QueryContext): Promise<QueryContext> {
  if (this.currentUser) {
    context.user = this.currentUser
  }
  return context
}

context.metadata is a plain mutable object. Anything you write here is available in onBeforeQuery and onAfterQuery for the same operation.

onBeforeQuery

Receives the QueryObject and must return it (modified or unchanged). Use it to inject additional WHERE conditions, rewrite queries, or set metadata for onAfterQuery.

override async onBeforeQuery<T extends DatrixEntry>(
  query: QueryObject<T>,
  context: QueryContext,
): Promise<QueryObject<T>> {
  // Inject a soft-delete filter on every SELECT
  if (query.type === "select") {
    return {
      ...query,
      where: { ...query.where, deletedAt: null } as WhereClause<T>,
    }
  }
  return query
}

onAfterQuery

Receives the result and must return it (modified or unchanged). Use it to strip sensitive fields, transform output, or trigger side effects.

override async onAfterQuery<T extends DatrixEntry>(
  result: T,
  context: QueryContext,
): Promise<T> {
  // Strip internal fields from read results
  if (context.action === "findMany" || context.action === "findOne") {
    const rows = Array.isArray(result) ? result : [result]
    const stripped = rows.map(({ internalField: _, ...rest }) => rest)
    return (Array.isArray(result) ? stripped : stripped[0]) as T
  }
  return result
}

Full example: audit log plugin

A plugin that writes an audit log entry after every create, update, or delete operation.

import { BasePlugin } from "@datrix/core"
import { defineSchema } from "@datrix/core"
import type { PluginContext, SchemaDefinition } from "@datrix/core"
import type { QueryContext } from "@datrix/core"
import type { DatrixEntry } from "@datrix/core"
import type { QueryObject } from "@datrix/core"

interface AuditLogOptions {
  readonly schemas?: readonly string[]  // limit to specific schemas — undefined = all
}

export class AuditLogPlugin extends BasePlugin<AuditLogOptions> {
  readonly name    = "audit-log"
  readonly version = "1.0.0"

  async init(context: PluginContext): Promise<void> {
    this.context = context
  }

  async destroy(): Promise<void> {}

  override async getSchemas(): Promise<SchemaDefinition[]> {
    return [
      defineSchema({
        name: "auditLog",
        fields: {
          action:     { type: "string", required: true },
          schemaName: { type: "string", required: true },
          recordId:   { type: "number" },
          userId:     { type: "number" },
          diff:       { type: "json" },
        },
      }),
    ]
  }

  override async onBeforeQuery<T extends DatrixEntry>(
    query: QueryObject<T>,
    context: QueryContext,
  ): Promise<QueryObject<T>> {
    // Snapshot write data before the operation so we can log it after
    if (query.type === "insert" || query.type === "update") {
      context.metadata.auditData = query.type === "insert"
        ? (query as { data?: unknown }).data
        : (query as { data?: unknown }).data
    }
    return query
  }

  override async onAfterQuery<T extends DatrixEntry>(
    result: T,
    context: QueryContext,
  ): Promise<T> {
    const tracked = this.options.schemas
    const schemaName = context.schema.name

    if (tracked && !tracked.includes(schemaName)) return result
    if (schemaName === "auditLog") return result  // avoid infinite loop

    const writeActions = ["create", "createMany", "update", "updateMany", "delete", "deleteMany"]
    if (!writeActions.includes(context.action)) return result

    const ctx = this.getContext()
    const rows = Array.isArray(result) ? result : [result]

    for (const row of rows) {
      await ctx.adapter.executeQuery({
        type: "insert",
        table: "audit_logs",
        data: [{
          action:     context.action,
          schemaName,
          recordId:   (row as DatrixEntry).id ?? null,
          userId:     (context.user as { id?: number } | undefined)?.id ?? null,
          diff:       context.metadata.auditData ?? null,
        }],
      })
    }

    return result
  }
}

Usage:

export default defineConfig(() => ({
  adapter: new PostgresAdapter({ ... }),
  schemas: [userSchema, postSchema],
  plugins: [
    new AuditLogPlugin({ schemas: ["post", "user"] }),
  ],
}))