Remove DataStream, support binary sync lines over HTTP

.changeset/cool-lions-melt.md

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

packages/common/rollup.config.mjs

@@ -3,9 +3,22 @@ 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,17 +34,76 @@ function defineBuild(isNode) {
         sourcemap: true
       }
     ],
-    plugins: [
-      [
-        json(),
-        nodeResolve({ preferBuiltins: false, browser: true }),
-        commonjs({}),
-        inject({
-          Buffer: isNode ? ['node:buffer', 'Buffer'] : ['buffer/', 'Buffer']
-        })
-      ]
-    ],
-    external: ['async-mutex', 'bson', isNode ? 'event-iterator' : undefined]
+    plugins: [plugins],
+    external: ['async-mutex', 'bson']
+  };
+}
+
+/**
+ * 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 SymbolasyncIterator
+  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

@@ -14,7 +14,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 (