Support new sync stream syntax

Author: simolus3

packages/service-core/src/routes/endpoints/sync-rules.ts

@@ -1,5 +1,5 @@
 import { ErrorCode, errors, router, schema } from '@powersync/lib-services-framework';
-import { SqlSyncRules, SyncRulesErrors } from '@powersync/service-sync-rules';
+import { SqlBucketDescriptor, SqlSyncRules, SyncRulesErrors } from '@powersync/service-sync-rules';
 import type { FastifyPluginAsync } from 'fastify';
 import * as t from 'ts-codec';
 

packages/sync-rules/src/BaseSqlDataQuery.ts

@@ -4,14 +4,29 @@ import { ColumnDefinition } from './ExpressionType.js';
 import { SourceTableInterface } from './SourceTableInterface.js';
 import { SqlTools } from './sql_filters.js';
 import { TablePattern } from './TablePattern.js';
-import { QueryParameters, QuerySchema, SourceSchema, SourceSchemaTable, SqliteJsonRow, SqliteRow } from './types.js';
+import {
+  EvaluationResult,
+  QueryParameters,
+  QuerySchema,
+  SourceSchema,
+  SourceSchemaTable,
+  SqliteJsonRow,
+  SqliteRow
+} from './types.js';
 import { filterJsonRow } from './utils.js';
+import { castAsText } from './sql_functions.js';
 
 export interface RowValueExtractor {
   extract(tables: QueryParameters, into: SqliteRow): void;
   getTypes(schema: QuerySchema, into: Record<string, ColumnDefinition>): void;
 }
 
+export interface EvaluateRowOptions {
+  table: SourceTableInterface;
+  row: SqliteRow;
+  bucketIds: (params: QueryParameters) => string[];
+}
+
 export interface BaseSqlDataQueryOptions {
   sourceTable: TablePattern;
   table: string;
@@ -21,7 +36,6 @@ export interface BaseSqlDataQueryOptions {
   descriptorName: string;
   bucketParameters: string[];
   tools: SqlTools;
-
   errors?: SqlRuleError[];
 }
 
@@ -149,6 +163,39 @@ export class BaseSqlDataQuery {
     return result;
   }
 
+  evaluateRowWithOptions(options: EvaluateRowOptions): EvaluationResult[] {
+    try {
+      const { table, row, bucketIds } = options;
+
+      const tables = { [this.table]: this.addSpecialParameters(table, row) };
+      const resolvedBucketIds = bucketIds(tables);
+
+      const data = this.transformRow(tables);
+      let id = data.id;
+      if (typeof id != 'string') {
+        // While an explicit cast would be better, this covers against very common
+        // issues when initially testing out sync, for example when the id column is an
+        // auto-incrementing integer.
+        // If there is no id column, we use a blank id. This will result in the user syncing
+        // a single arbitrary row for this table - better than just not being able to sync
+        // anything.
+        id = castAsText(id) ?? '';
+      }
+      const outputTable = this.getOutputName(table.name);
+
+      return resolvedBucketIds.map((bucketId) => {
+        return {
+          bucket: bucketId,
+          table: outputTable,
+          id: id,
+          data
+        } as EvaluationResult;
+      });
+    } catch (e) {
+      return [{ error: e.message ?? `Evaluating data query failed` }];
+    }
+  }
+
   protected transformRow(tables: QueryParameters): SqliteJsonRow {
     let result: SqliteRow = {};
     for (let extractor of this.extractors) {

packages/sync-rules/src/SqlDataQuery.ts

@@ -186,36 +186,15 @@ export class SqlDataQuery extends BaseSqlDataQuery {
   }
 
   evaluateRow(table: SourceTableInterface, row: SqliteRow): EvaluationResult[] {
-    try {
-      const tables = { [this.table]: this.addSpecialParameters(table, row) };
-      const bucketParameters = this.filter.filterRow(tables);
-      const bucketIds = bucketParameters.map((params) =>
-        getBucketId(this.descriptorName, this.bucketParameters, params)
-      );
-
-      const data = this.transformRow(tables);
-      let id = data.id;
-      if (typeof id != 'string') {
-        // While an explicit cast would be better, this covers against very common
-        // issues when initially testing out sync, for example when the id column is an
-        // auto-incrementing integer.
-        // If there is no id column, we use a blank id. This will result in the user syncing
-        // a single arbitrary row for this table - better than just not being able to sync
-        // anything.
-        id = castAsText(id) ?? '';
+    const query = this;
+
+    return this.evaluateRowWithOptions({
+      table,
+      row,
+      bucketIds(tables) {
+        const bucketParameters = query.filter.filterRow(tables);
+        return bucketParameters.map((params) => getBucketId(query.descriptorName, query.bucketParameters, params));
       }
-      const outputTable = this.getOutputName(table.name);
-
-      return bucketIds.map((bucketId) => {
-        return {
-          bucket: bucketId,
-          table: outputTable,
-          id: id,
-          data
-        } as EvaluationResult;
-      });
-    } catch (e) {
-      return [{ error: e.message ?? `Evaluating data query failed` }];
-    }
+    });
   }
 }

packages/sync-rules/src/SqlSyncRules.ts

@@ -22,9 +22,12 @@ import {
   SourceSchema,
   SqliteJsonRow,
   SqliteRow,
+  StreamParseOptions,
   SyncRules
 } from './types.js';
 import { BucketSource } from './BucketSource.js';
+import { SyncStream } from './streams/stream.js';
+import { syncStreamFromSql } from './streams/from_sql.js';
 
 const ACCEPT_POTENTIALLY_DANGEROUS_QUERIES = Symbol('ACCEPT_POTENTIALLY_DANGEROUS_QUERIES');
 
@@ -137,6 +140,7 @@ export class SqlSyncRules implements SyncRules {
 
     // Bucket definitions using explicit parameter and data queries.
     const bucketMap = parsed.get('bucket_definitions') as YAMLMap;
+    const streamMap = parsed.get('streams') as YAMLMap | null;
     const definitionNames = new Set<string>();
     const checkUniqueName = (name: string, literal: Scalar) => {
       if (definitionNames.has(name)) {
@@ -209,6 +213,39 @@ export class SqlSyncRules implements SyncRules {
       rules.bucketSources.push(descriptor);
     }
 
+    for (const entry of streamMap?.items ?? []) {
+      const { key: keyScalar, value } = entry as { key: Scalar; value: YAMLMap };
+      const key = keyScalar.toString();
+      if (!checkUniqueName(key, keyScalar)) {
+        continue;
+      }
+
+      const accept_potentially_dangerous_queries =
+        value.get('accept_potentially_dangerous_queries', true)?.value == true;
+
+      const queryOptions: StreamParseOptions = {
+        ...options,
+        accept_potentially_dangerous_queries,
+        priority: rules.parsePriority(value),
+        default: value.get('default', true)?.value == true
+      };
+
+      const data = value.get('query', true) as unknown;
+      if (data instanceof Scalar) {
+        rules.withScalar(data, (q) => {
+          const [parsed, errors] = syncStreamFromSql(key, q, options);
+          rules.bucketSources.push(parsed);
+          return {
+            parsed: true,
+            errors
+          };
+        });
+      } else {
+        rules.errors.push(this.tokenError(data, 'Must be a string.'));
+        continue;
+      }
+    }
+
     const eventMap = parsed.get('event_definitions') as YAMLMap;
     for (const event of eventMap?.items ?? []) {
       const { key, value } = event as { key: Scalar; value: YAMLSeq };

packages/sync-rules/src/index.ts

@@ -12,10 +12,13 @@ export * from './schema-generators/schema-generators.js';
 export * from './SourceTableInterface.js';
 export * from './sql_filters.js';
 export * from './sql_functions.js';
+export { SqlBucketDescriptor } from './SqlBucketDescriptor.js';
 export * from './SqlDataQuery.js';
 export * from './SqlParameterQuery.js';
 export * from './SqlSyncRules.js';
 export * from './StaticSchema.js';
+export { SyncStream } from './streams/stream.js';
+export { syncStreamFromSql } from './streams/from_sql.js';
 export * from './TablePattern.js';
 export * from './types.js';
 export * from './utils.js';

packages/sync-rules/src/request_functions.ts

@@ -28,7 +28,7 @@ const request_parameters: SqlParameterFunction = {
   usesUnauthenticatedRequestParameters: true
 };
 
-const request_jwt: SqlParameterFunction = {
+export const request_jwt: SqlParameterFunction = {
   debugName: 'request.jwt',
   call(parameters: ParameterValueSet) {
     return parameters.rawTokenPayload;
@@ -42,7 +42,7 @@ const request_jwt: SqlParameterFunction = {
   usesUnauthenticatedRequestParameters: false
 };
 
-const request_user_id: SqlParameterFunction = {
+export const request_user_id: SqlParameterFunction = {
   debugName: 'request.user_id',
   call(parameters: ParameterValueSet) {
     return parameters.userId;