Remove DataStream, support binary sync lines over HTTP (#821)

Author: simolus3

.changeset/cool-lions-melt.md

@@ -0,0 +1,6 @@
+---
+'@powersync/react-native': minor
+'@powersync/common': minor
+---
+
+Simplify internal handling of HTTP streams.

packages/common/package.json

@@ -64,13 +64,15 @@
     "@rollup/plugin-node-resolve": "catalog:",
     "@types/node": "catalog:",
     "@types/uuid": "catalog:",
+    "bson": "catalog:",
     "buffer": "^6.0.3",
-    "rollup": "catalog:",
-    "rollup-plugin-dts": "catalog:",
     "cross-fetch": "^4.1.0",
+    "estree-walker": "^3.0.3",
     "js-logger": "catalog:",
+    "magic-string": "^0.30.21",
+    "rollup": "catalog:",
+    "rollup-plugin-dts": "catalog:",
     "rsocket-core": "1.0.0-alpha.3",
-    "rsocket-websocket-client": "1.0.0-alpha.3",
-    "bson": "catalog:"
+    "rsocket-websocket-client": "1.0.0-alpha.3"
   }
 }

packages/common/rollup.config.mjs

@@ -3,9 +3,24 @@ import inject from '@rollup/plugin-inject';
 import json from '@rollup/plugin-json';
 import nodeResolve from '@rollup/plugin-node-resolve';
 import { dts } from 'rollup-plugin-dts';
+import MagicString from 'magic-string';
+import { walk } from 'estree-walker';
 
 function defineBuild(isNode) {
   const suffix = isNode ? '.node' : '';
+  const plugins = [
+    [
+      json(),
+      nodeResolve({ preferBuiltins: false, browser: true }),
+      commonjs({}),
+      inject({
+        Buffer: isNode ? ['node:buffer', 'Buffer'] : ['buffer/', 'Buffer']
+      })
+    ]
+  ];
+  if (!isNode) {
+    plugins.push(applyAsyncIteratorPolyfill());
+  }
 
   return {
     input: 'lib/index.js',
@@ -21,16 +36,7 @@ function defineBuild(isNode) {
         sourcemap: true
       }
     ],
-    plugins: [
-      [
-        json(),
-        nodeResolve({ preferBuiltins: false, browser: true }),
-        commonjs({}),
-        inject({
-          Buffer: isNode ? ['node:buffer', 'Buffer'] : ['buffer/', 'Buffer']
-        })
-      ]
-    ],
+    plugins,
     external: ['bson', isNode ? 'event-iterator' : undefined]
   };
 }
@@ -49,3 +55,71 @@ export default () => {
     }
   ];
 };
+
+/**
+ * Replaces `Symbol.asyncIterator` with a polyfill in bundled dependencies (specifically `EventIterator`).
+ *
+ * This is derived from an example in Rollup docs: https://rollupjs.org/plugin-development/#resolveid
+ */
+function applyAsyncIteratorPolyfill() {
+  // A fake file we import into every file using Symbol.asyncIterator
+  const POLYFILL_ID = '\0symbol-async-iterator';
+
+  return {
+    name: 'applyAsyncIteratorPonyfill',
+    async resolveId(source) {
+      if (source === POLYFILL_ID) {
+        return { id: POLYFILL_ID };
+      }
+      return null;
+    },
+    load(id) {
+      if (id === POLYFILL_ID) {
+        // Outside of Node.JS, we replace Symbol.asyncIterator with an import to this symbol. This allows us to use
+        // Symbol.asyncIterator internally. If users install the recommended polyfill, https://github.com/Azure/azure-sdk-for-js/blob/%40azure/core-asynciterator-polyfill_1.0.2/sdk/core/core-asynciterator-polyfill/src/index.ts#L4-L6
+        // they can also use our async iterables on React Native.
+        // Of course, we could also inline this definition with a simple replacement. But putting this in a fake module
+        // de-duplicates it.
+        return `
+export const symbolAsyncIterator = Symbol.asyncIterator ?? Symbol.for('Symbol.asyncIterator');
+        `;
+      }
+    },
+    transform(code, id) {
+      if (id === POLYFILL_ID) return null;
+      if (!code.includes('Symbol.asyncIterator')) return null;
+
+      const ast = this.parse(code);
+      const s = new MagicString(code);
+
+      let replaced = false;
+      const symbolName = 'symbolAsyncIterator';
+
+      walk(ast, {
+        enter(node) {
+          // Replace Symbol.asyncIterator
+          if (
+            node.type === 'MemberExpression' &&
+            !node.computed &&
+            node.object.type === 'Identifier' &&
+            node.object.name === 'Symbol' &&
+            node.property.type === 'Identifier' &&
+            node.property.name === 'asyncIterator'
+          ) {
+            replaced = true;
+            s.overwrite(node.start, node.end, symbolName);
+          }
+        }
+      });
+
+      if (!replaced) return null;
+
+      s.prepend(`import { ${symbolName} } from ${JSON.stringify(POLYFILL_ID)};\n`);
+
+      return {
+        code: s.toString(),
+        map: s.generateMap({ hires: true })
+      };
+    }
+  };
+}

packages/common/src/client/AbstractPowerSyncDatabase.ts

@@ -13,7 +13,7 @@ import { UploadQueueStats } from '../db/crud/UploadQueueStatus.js';
 import { Schema } from '../db/schema/Schema.js';
 import { BaseObserver } from '../utils/BaseObserver.js';
 import { ControlledExecutor } from '../utils/ControlledExecutor.js';
-import { symbolAsyncIterator, throttleTrailing } from '../utils/async.js';
+import { throttleTrailing } from '../utils/async.js';
 import {
   ConnectionManager,
   CreateSyncImplementationOptions,
@@ -704,7 +704,7 @@ export abstract class AbstractPowerSyncDatabase extends BaseObserver<PowerSyncDB
    * @returns A transaction of CRUD operations to upload, or null if there are none
    */
   async getNextCrudTransaction(): Promise<CrudTransaction | null> {
-    const iterator = this.getCrudTransactions()[symbolAsyncIterator]();
+    const iterator = this.getCrudTransactions()[Symbol.asyncIterator]();
     return (await iterator.next()).value;
   }
 
@@ -741,7 +741,7 @@ export abstract class AbstractPowerSyncDatabase extends BaseObserver<PowerSyncDB
    */
   getCrudTransactions(): AsyncIterable<CrudTransaction, null> {
     return {
-      [symbolAsyncIterator]: () => {
+      [Symbol.asyncIterator]: () => {
         let lastCrudItemId = -1;
         const sql = `
 WITH RECURSIVE crud_entries AS (