feat: Use Compiler for Event Definitions

.changeset/compiled-event-plans.md

@@ -0,0 +1,6 @@
+---
+'@powersync/service-sync-rules': minor
+'@powersync/service-core': minor
+---
+
+Compile replication events for every sync-config edition into additive serialized sync-plan data, expose behavioral equality for compiled event models, and normalize legacy sidecar events when loading older plans while preserving raw SQL for older services.

modules/module-mongodb-storage/src/storage/implementation/MongoBucketBatch.ts

@@ -1,5 +1,5 @@
 import { mongo } from '@powersync/lib-service-mongodb';
-import { HydratedSyncConfig, SqlEventDescriptor, SqliteRow, SqliteValue } from '@powersync/service-sync-rules';
+import { HydratedEventDescriptor, HydratedSyncConfig, SqliteRow, SqliteValue } from '@powersync/service-sync-rules';
 import * as bson from 'bson';
 
 import {
@@ -934,9 +934,9 @@ export abstract class MongoBucketBatch
   }
 
   /**
-   * Gets relevant {@link SqlEventDescriptor}s for the given {@link SourceTable}
+   * Gets relevant {@link HydratedEventDescriptor}s for the given {@link SourceTable}
    */
-  protected getTableEvents(table: storage.SourceTable): SqlEventDescriptor[] {
+  protected getTableEvents(table: storage.SourceTable): HydratedEventDescriptor[] {
     return this.sync_rules.eventDescriptors.filter((evt) =>
       [...evt.getSourceTables()].some((sourceTable) => sourceTable.matches(table.ref))
     );

modules/module-postgres-storage/src/storage/batch/PostgresBucketBatch.ts

@@ -1334,10 +1334,10 @@ export class PostgresBucketBatch
   }
 
   /**
-   * Gets relevant {@link SqlEventDescriptor}s for the given {@link SourceTable}
+   * Gets relevant {@link HydratedEventDescriptor}s for the given {@link SourceTable}
    * TODO maybe share this with an abstract class
    */
-  protected getTableEvents(table: storage.SourceTable): sync_rules.SqlEventDescriptor[] {
+  protected getTableEvents(table: storage.SourceTable): sync_rules.HydratedEventDescriptor[] {
     return this.sync_rules.eventDescriptors.filter((evt) =>
       [...evt.getSourceTables()].some((sourceTable) => sourceTable.matches(table.ref))
     );

packages/service-core/src/storage/BucketStorageFactory.ts

@@ -180,11 +180,10 @@ export interface SerializedSyncPlan {
   plan: RawSerializedSyncPlan;
   compatibility: SerializedCompatibilityContext;
   /**
-   * Event descriptors are not currently represented in the sync plan because they don't use the sync streams compiler
-   * yet.
+   * Raw event SQL persisted as a compatibility mirror for compiled {@link plan} events.
    *
-   * We might revisit that in the future, but for now we store SQL text of their definitions here to be able to restore
-   * them.
+   * Compiled events are an additive plan field. Older services ignore that field and restore these descriptors through
+   * the legacy evaluator. Keep dual-writing this field until a future plan version explicitly removes that support.
    */
   eventDescriptors: Record<string, string[]>;
   errors?: ReplicationError[];
@@ -212,13 +211,14 @@ export function updateSyncRulesFromConfig(
   const { config, errors } = parsed;
   if (config instanceof PrecompiledSyncConfig) {
     const eventDescriptors: Record<string, string[]> = {};
-    for (const event of config.eventDescriptors) {
-      eventDescriptors[event.name] = event.sourceQueries.map((q) => q.sql);
+    for (const event of config.plan.events) {
+      eventDescriptors[event.name] = event.sourceQueries.map((query) => query.sql);
     }
 
     plan = {
       compatibility: config.compatibility.serialize(),
       plan: serializeSyncPlan(config.plan),
+      // Dual-write raw SQL so older services can ignore additive compiled plan events without losing event behavior.
       eventDescriptors,
       errors: errors.map((e) => syncConfigYamlErrorToReplicationError(e))
     };

packages/service-core/src/storage/PersistedSyncConfigContent.ts

@@ -2,14 +2,14 @@ import { logger as defaultLogger, ErrorCode, ServiceError } from '@powersync/lib
 import {
   CompatibilityContext,
   CompatibilityOption,
+  compileEventDefinitions,
   DEFAULT_HYDRATION_STATE,
   deserializeSyncPlan,
   ErrorLocation,
   HydratedSyncConfig,
   HydrationState,
   nodeSqlite,
   PrecompiledSyncConfig,
-  SqlEventDescriptor,
   SqlSyncRules,
   SyncConfigWithErrors,
   versionedHydrationState,
@@ -40,17 +40,23 @@ export function parsePersistedSyncConfigContent(options: ParsePersistedSyncConfi
 
   const plan = deserializeSyncPlan(compiledPlan.plan);
   const compatibility = CompatibilityContext.deserialize(compiledPlan.compatibility);
-  const eventDefinitions: SqlEventDescriptor[] = [];
-  for (const [name, queries] of Object.entries(compiledPlan.eventDescriptors)) {
-    const descriptor = new SqlEventDescriptor(name, compatibility);
-    for (const query of queries) {
-      descriptor.addSourceQuery(query, parseOptions);
+  const errors: YamlError[] = [];
+  // Compiled events are additive to plan versions 1 and 2. New readers prefer them when present; when an older plan
+  // does not contain them, normalize the dual-written raw SQL at this loading boundary. This keeps legacy event
+  // evaluators out of PrecompiledSyncConfig while older binaries can continue reading the same persisted config.
+  if (compiledPlan.plan.events == null) {
+    const normalized = compileEventDefinitions(compiledPlan.eventDescriptors, parseOptions);
+    const fatalErrors = normalized.errors.filter((error) => error.type == 'fatal');
+    if (fatalErrors.length != 0) {
+      throw new Error(
+        `Failed to compile persisted replication events: ${fatalErrors.map((error) => error.message).join(', ')}`
+      );
     }
-
-    eventDefinitions.push(descriptor);
+    plan.events = normalized.events;
+    errors.push(...normalized.errors.map((error) => new YamlError(error)));
   }
 
-  const precompiled = new PrecompiledSyncConfig(plan, compatibility, eventDefinitions, {
+  const precompiled = new PrecompiledSyncConfig(plan, compatibility, {
     defaultSchema: parseOptions.defaultSchema,
     sourceText: content
   });
@@ -59,7 +65,6 @@ export function parsePersistedSyncConfigContent(options: ParsePersistedSyncConfi
   // This means asUpdateOptions will not change the storage version, even if the default changes.
   precompiled.storageVersion = storageVersion;
 
-  const errors: YamlError[] = [];
   if (compiledPlan.errors) {
     for (const error of compiledPlan.errors) {
       const location: ErrorLocation | undefined = error.location && {

packages/service-core/src/storage/ReplicationEventPayload.ts

@@ -11,6 +11,6 @@ export type EventData = {
 export type ReplicationEventPayload = {
   batch: BucketStorageBatch;
   data: EventData;
-  event: sync_rules.SqlEventDescriptor;
+  event: sync_rules.HydratedEventDescriptor;
   table: SourceTable;
 };

packages/service-core/test/src/PersistedSyncConfigContent.test.ts

@@ -0,0 +1,99 @@
+import {
+  DEFAULT_HYDRATION_STATE,
+  DEFAULT_TAG,
+  nodeSqlite,
+  PrecompiledSyncConfig,
+  serializeSyncPlan,
+  SqlSyncRules
+} from '@powersync/service-sync-rules';
+import * as sqlite from 'node:sqlite';
+import { describe, expect, test } from 'vitest';
+import {
+  parsePersistedSyncConfigContent,
+  SerializedSyncPlan,
+  updateSyncRulesFromConfig
+} from '../../src/storage/storage-index.js';
+
+const EVENT_QUERY = 'SELECT user_id, checkpoint FROM checkpoints WHERE active = true';
+const yamlWithoutEvents = `
+config:
+  edition: 3
+
+streams:
+  checkpoints:
+    query: SELECT * FROM checkpoints
+`;
+const yamlWithEvents = `${yamlWithoutEvents}
+event_definitions:
+  write_checkpoints:
+    payloads:
+      - ${EVENT_QUERY}
+`;
+
+describe('persisted compiled replication events', () => {
+  test('dual-writes additive compiled events and restores the compiled evaluator', () => {
+    const parsed = SqlSyncRules.fromYaml(yamlWithEvents, { defaultSchema: 'test_schema' });
+    const update = updateSyncRulesFromConfig(parsed);
+    const compiled = update.config.plan!;
+    const originalEvent = compiled.plan.events![0];
+
+    expect(compiled.plan.version).toBeLessThanOrEqual(2);
+    expect(compiled.plan.events).toHaveLength(1);
+    expect(originalEvent.name).toBe('write_checkpoints');
+    expect(compiled.eventDescriptors).toEqual({ write_checkpoints: [EVENT_QUERY] });
+
+    const restored = restore(compiled);
+    expect(restored.config).toBeInstanceOf(PrecompiledSyncConfig);
+    expect(restored.config).not.toHaveProperty('eventDescriptors');
+    expect(restored.config.eventDefinitions).toHaveLength(1);
+    expect(restored.config.eventDefinitions[0].name).toBe('write_checkpoints');
+
+    const hydrated = restored.config.hydrate({
+      hydrationState: DEFAULT_HYDRATION_STATE,
+      sqlite: nodeSqlite(sqlite)
+    });
+    expect(
+      hydrated.eventDescriptors[0].evaluateRowWithErrors({
+        sourceTable: { connectionTag: DEFAULT_TAG, schema: 'test_schema', name: 'checkpoints' },
+        record: { user_id: 'user-1', checkpoint: 4n, active: 1 }
+      })
+    ).toEqual({ result: { data: { user_id: 'user-1', checkpoint: 4n } }, errors: [] });
+
+    // This represents what an older compiler sees after ignoring the additive plan.events field. The raw descriptor
+    // mirror is normalized into the compiled representation by the new loading boundary.
+    const { events: _ignored, ...planWithoutCompiledEvents } = compiled.plan;
+    const legacyView = restore({ ...compiled, plan: planWithoutCompiledEvents });
+    expect(legacyView.config).not.toHaveProperty('eventDescriptors');
+    expect(legacyView.config.eventDefinitions).toHaveLength(1);
+    expect(legacyView.config.eventDefinitions[0].name).toBe('write_checkpoints');
+    // Recompiling the raw SQL mirror produces the same serialized event plan.
+    const legacyEvent = serializeSyncPlan((legacyView.config as PrecompiledSyncConfig).plan).events![0];
+    expect(legacyEvent).toEqual(originalEvent);
+  });
+
+  test('restores raw event descriptors attached to version 1 and 2 plans', () => {
+    const parsed = SqlSyncRules.fromYaml(yamlWithoutEvents, { defaultSchema: 'test_schema' });
+    const update = updateSyncRulesFromConfig(parsed);
+    const legacy: SerializedSyncPlan = {
+      ...update.config.plan!,
+      eventDescriptors: { write_checkpoints: [EVENT_QUERY] }
+    };
+
+    expect(legacy.plan.version).toBeLessThanOrEqual(2);
+    const restored = restore(legacy);
+    expect(restored.config).not.toHaveProperty('eventDescriptors');
+    expect(restored.config.eventDefinitions).toHaveLength(1);
+    expect((restored.config as PrecompiledSyncConfig).plan.events).toMatchObject([
+      { name: 'write_checkpoints', sourceQueries: [{ sql: EVENT_QUERY }] }
+    ]);
+  });
+});
+
+function restore(compiledPlan: SerializedSyncPlan) {
+  return parsePersistedSyncConfigContent({
+    content: yamlWithEvents,
+    compiledPlan,
+    storageVersion: 1,
+    parseOptions: { defaultSchema: 'test_schema' }
+  });
+}

packages/sync-rules/src/HydratedSyncConfig.ts

@@ -8,6 +8,7 @@ import {
   EvaluationError,
   GetBucketParameterQuerierResult,
   GetQuerierOptions,
+  HydratedEventDescriptor,
   HydrateSyncConfigParams,
   HydrationInput,
   isEvaluatedParameters,
@@ -21,7 +22,6 @@ import {
   QuerierError,
   ScopedEvaluateParameterRow,
   ScopedEvaluateRow,
-  SqlEventDescriptor,
   SqliteInputValue,
   SqliteValue,
   SyncConfig,
@@ -52,7 +52,7 @@ export class HydratedSyncConfig {
    */
   private bucketSources: HydratedBucketSource[] = [];
 
-  eventDescriptors: SqlEventDescriptor[] = [];
+  eventDescriptors: HydratedEventDescriptor[] = [];
 
   /**
    * Only a single compatibility context is supported across all merged SyncConfigs.
@@ -118,7 +118,9 @@ export class HydratedSyncConfig {
       this.bucketParameterLookupSources
     ).evaluateParameterRow;
 
-    this.eventDescriptors = definitions.flatMap((definition) => definition.eventDescriptors);
+    this.eventDescriptors = definitions.flatMap((definition) =>
+      definition.eventDefinitions.map((event) => event.createEvaluator(this.hydrationInput))
+    );
 
     if (definitions.length == 1) {
       this.#bucketSourceDefinitions = definitions[0].bucketSources;

packages/sync-rules/src/SyncConfig.ts

@@ -6,7 +6,7 @@ import {
 } from './BucketSource.js';
 import { CompatibilityContext } from './compatibility.js';
 import { YamlError } from './errors.js';
-import { SqlEventDescriptor } from './events/SqlEventDescriptor.js';
+import { EventDefinition } from './events/EventDescriptor.js';
 import { HydratedSyncConfig } from './HydratedSyncConfig.js';
 import { SourceTableRef } from './SourceTableRef.js';
 import { TablePattern } from './TablePattern.js';
@@ -21,14 +21,15 @@ export abstract class SyncConfig {
   bucketDataSources: BucketDataSource[] = [];
   bucketParameterLookupSources: ParameterIndexLookupCreator[] = [];
   bucketSources: BucketSource[] = [];
+  /** Prepared event definitions. Executable event descriptors only exist on {@link HydratedSyncConfig}. */
+  eventDefinitions: EventDefinition[] = [];
   compatibility: CompatibilityContext = CompatibilityContext.FULL_BACKWARDS_COMPATIBILITY;
   /**
    * If not defined, the storage module picks the latest stable version.
    *
    * Only supported storage versions can be set here when parsing from yaml.
    */
   storageVersion: number | undefined;
-  eventDescriptors: SqlEventDescriptor[] = [];
 
   /**
    * The (YAML-based) source contents from which this sync config has been derived.
@@ -71,9 +72,9 @@ export abstract class SyncConfig {
         sourceTables.set(r.key(), r);
       }
     }
-    for (const event of this.eventDescriptors) {
-      for (const r of event.getSourceTables()) {
-        sourceTables.set(r.key(), r);
+    for (const event of this.eventDefinitions) {
+      for (const table of event.getSourceTables()) {
+        sourceTables.set(table.key(), table);
       }
     }
   }
@@ -87,19 +88,17 @@ export abstract class SyncConfig {
   getEventTables(): TablePattern[] {
     const eventTables = new Map<string, TablePattern>();
 
-    if (this.eventDescriptors) {
-      for (const event of this.eventDescriptors) {
-        for (const r of event.getSourceTables()) {
-          eventTables.set(r.key(), r);
-        }
+    for (const event of this.eventDefinitions) {
+      for (const table of event.getSourceTables()) {
+        eventTables.set(table.key(), table);
       }
     }
 
     return [...eventTables.values()];
   }
 
   tableTriggersEvent(table: SourceTableRef): boolean {
-    return this.eventDescriptors.some((bucket) => bucket.tableTriggersEvent(table));
+    return this.eventDefinitions.some((event) => event.tableTriggersEvent(table));
   }
 
   tableSyncsData(table: SourceTableRef): boolean {