Support streams

Author: simolus3

packages/sync-rules/src/SqlSyncRules.ts

@@ -241,7 +241,8 @@ export class SqlSyncRules implements SyncRules {
         ...options,
         accept_potentially_dangerous_queries,
         priority: rules.parsePriority(value),
-        auto_subscribe: value.get('auto_subscribe', true)?.value == true
+        auto_subscribe: value.get('auto_subscribe', true)?.value == true,
+        fixedQuirks
       };
 
       const data = value.get('query', true) as unknown;

packages/sync-rules/src/streams/from_sql.ts

@@ -41,6 +41,7 @@ import {
   Statement
 } from 'pgsql-ast-parser';
 import { STREAM_FUNCTIONS } from './functions.js';
+import { CompatibilityContext, CompatibilityLevel } from '../quirks.js';
 
 export function syncStreamFromSql(
   descriptorName: string,
@@ -91,7 +92,8 @@ class SyncStreamCompiler {
 
     const stream = new SyncStream(
       this.descriptorName,
-      new BaseSqlDataQuery(this.compileDataQuery(tools, query, alias, sourceTable))
+      new BaseSqlDataQuery(this.compileDataQuery(tools, query, alias, sourceTable)),
+      new CompatibilityContext(CompatibilityLevel.SYNC_STREAMS, this.options.fixedQuirks)
     );
     stream.subscribedToByDefault = this.options.auto_subscribe ?? false;
     if (filter.isValid(tools)) {

packages/sync-rules/src/streams/parameter.ts

@@ -7,7 +7,8 @@ import {
   RequestParameters,
   SqliteJsonValue,
   SqliteRow,
-  SqliteValue
+  SqliteValue,
+  TableRow
 } from '../types.js';
 
 /**
@@ -31,7 +32,7 @@ export interface BucketParameter {
    * When a user connects, {@link StaticLookup.fromRequest} would return the user ID from the token. A matching bucket would
    * then contain the oplog data for assets with the matching `owner` column.
    */
-  filterRow(options: EvaluateRowOptions): SqliteJsonValue[];
+  filterRow(row: TableRow): SqliteJsonValue[];
 }
 
 export interface SubqueryEvaluator {

packages/sync-rules/src/streams/stream.ts

@@ -3,6 +3,7 @@ import { BucketInclusionReason, BucketPriority, DEFAULT_BUCKET_PRIORITY } from '
 import { BucketParameterQuerier, PendingQueriers } from '../BucketParameterQuerier.js';
 import { BucketSource, BucketSourceType, ResultSetDescription } from '../BucketSource.js';
 import { ColumnDefinition } from '../ExpressionType.js';
+import { CompatibilityContext } from '../quirks.js';
 import { SourceTableInterface } from '../SourceTableInterface.js';
 import { GetQuerierOptions, RequestedStream } from '../SqlSyncRules.js';
 import { TablePattern } from '../TablePattern.js';
@@ -12,8 +13,10 @@ import {
   EvaluationResult,
   RequestParameters,
   SourceSchema,
-  SqliteRow
+  SqliteRow,
+  TableRow
 } from '../types.js';
+import { applyRowContext } from '../utils.js';
 import { StreamVariant } from './variant.js';
 
 export class SyncStream implements BucketSource {
@@ -23,7 +26,11 @@ export class SyncStream implements BucketSource {
   variants: StreamVariant[];
   data: BaseSqlDataQuery;
 
-  constructor(name: string, data: BaseSqlDataQuery) {
+  constructor(
+    name: string,
+    data: BaseSqlDataQuery,
+    private readonly compatibility: CompatibilityContext
+  ) {
     this.name = name;
     this.subscribedToByDefault = false;
     this.priority = DEFAULT_BUCKET_PRIORITY;
@@ -165,13 +172,19 @@ export class SyncStream implements BucketSource {
     }
 
     const stream = this;
+    const mappedRow = applyRowContext(options.record, this.compatibility);
+    const row: TableRow = {
+      sourceTable: options.sourceTable,
+      record: mappedRow
+    };
+
     return this.data.evaluateRowWithOptions({
       table: options.sourceTable,
-      row: options.record,
+      row: applyRowContext(options.record, this.compatibility),
       bucketIds() {
         const bucketIds: string[] = [];
         for (const variant of stream.variants) {
-          bucketIds.push(...variant.bucketIdsForRow(stream.name, options));
+          bucketIds.push(...variant.bucketIdsForRow(stream.name, row));
         }
 
         return bucketIds;

packages/sync-rules/src/streams/variant.ts

@@ -6,7 +6,8 @@ import {
   EvaluateRowOptions,
   RequestParameters,
   SqliteJsonValue,
-  SqliteRow
+  SqliteRow,
+  TableRow
 } from '../types.js';
 import { isJsonValue, JSONBucketNameSerialize, normalizeParameterValue } from '../utils.js';
 import { BucketParameter, SubqueryEvaluator } from './parameter.js';
@@ -45,7 +46,7 @@ export class StreamVariant {
    *
    * This is introduced for streams like `SELECT * FROM assets WHERE LENGTH(assets.name < 10)`.
    */
-  additionalRowFilters: ((options: EvaluateRowOptions) => boolean)[];
+  additionalRowFilters: ((row: TableRow) => boolean)[];
 
   /**
    * Additional filters that are evaluated against the request of the stream subscription.
@@ -66,7 +67,7 @@ export class StreamVariant {
   /**
    * Given a row in the table this stream selects from, returns all ids of buckets to which that row belongs to.
    */
-  bucketIdsForRow(streamName: string, options: EvaluateRowOptions): string[] {
+  bucketIdsForRow(streamName: string, options: TableRow): string[] {
     return this.instantiationsForRow(options).map((values) => this.buildBucketId(streamName, values));
   }
 
@@ -75,7 +76,7 @@ export class StreamVariant {
    *
    * The inner arrays will have a length equal to the amount of parameters in this variant.
    */
-  instantiationsForRow(options: EvaluateRowOptions): SqliteJsonValue[][] {
+  instantiationsForRow(options: TableRow): SqliteJsonValue[][] {
     for (const additional of this.additionalRowFilters) {
       if (!additional(options)) {
         return [];

packages/sync-rules/src/types.ts

@@ -8,7 +8,7 @@ import { BucketPriority } from './BucketDescription.js';
 import { ParameterLookup } from './BucketParameterQuerier.js';
 import { DateTimeValue } from './types/time.js';
 import { CustomSqliteValue } from './types/custom_sqlite_value.js';
-import { CompatibilityContext } from './quirks.js';
+import { CompatibilityContext, Quirk } from './quirks.js';
 
 export interface SyncRules {
   evaluateRow(options: EvaluateRowOptions): EvaluationResult[];
@@ -23,6 +23,7 @@ export interface QueryParseOptions extends SyncRulesOptions {
 
 export interface StreamParseOptions extends QueryParseOptions {
   auto_subscribe?: boolean;
+  fixedQuirks: Quirk[];
 }
 
 export interface EvaluatedParameters {
@@ -283,9 +284,14 @@ export interface InputParameter {
   parametersToLookupValue(parameters: ParameterValueSet): SqliteValue;
 }
 
-export interface EvaluateRowOptions {
+export interface EvaluateRowOptions extends TableRow<SqliteInputRow> {}
+
+/**
+ * A row associated with the table it's coming from.
+ */
+export interface TableRow<R = SqliteRow> {
   sourceTable: SourceTableInterface;
-  record: SqliteInputRow;
+  record: R;
 }
 
 /**

packages/sync-rules/test/src/quirks.test.ts

@@ -57,6 +57,29 @@ fixed_quirks:
         { bucket: 'mybucket[]', data: { description: '2025-08-19T09:21:00Z', id: 'id' }, id: 'id', table: 'assets' }
       ]);
     });
+
+    test('streams use new format by default', () => {
+      const rules = SqlSyncRules.fromYaml(
+        `
+streams:
+  stream:
+    query: SELECT id, description FROM assets
+    `,
+        PARSE_OPTIONS
+      );
+
+      expect(
+        rules.evaluateRow({
+          sourceTable: ASSETS,
+          record: {
+            id: 'id',
+            description: value
+          }
+        })
+      ).toStrictEqual([
+        { bucket: 'stream|0[]', data: { description: '2025-08-19T09:21:00Z', id: 'id' }, id: 'id', table: 'assets' }
+      ]);
+    });
   });
 
   test('warning for unknown quirk', () => {

packages/sync-rules/test/src/streams.test.ts

@@ -11,6 +11,7 @@ import {
   SqliteJsonRow,
   SqliteRow,
   StaticSchema,
+  StreamParseOptions,
   SyncStream,
   syncStreamFromSql
 } from '../../src/index.js';
@@ -610,7 +611,11 @@ const schema = new StaticSchema([
   }
 ]);
 
-const options = { schema: schema, ...PARSE_OPTIONS };
+const options: StreamParseOptions = {
+  schema: schema,
+  ...PARSE_OPTIONS,
+  fixedQuirks: []
+};
 
 function evaluateBucketIds(stream: SyncStream, sourceTable: SourceTableInterface, record: SqliteRow) {
   return stream.evaluateRow({ sourceTable, record }).map((r) => {