Add test client.

Author: rkistner

pnpm-lock.yaml

@@ -2,5 +2,6 @@ packages:
   - 'packages/*'
   - 'libs/*'
   - 'service'
+  - 'test-client'
   # exclude packages that are inside test directories
   - '!**/test/**'

pnpm-workspace.yaml

@@ -0,0 +1,18 @@
+# Test Client
+
+This is a minimal client demonstrating direct usage of the sync api.
+
+For a full implementation, see our client SDKs.
+
+## Usage
+
+```sh
+# In project root
+pnpm install
+pnpm build:packages
+# Here
+pnpm build
+node dist/bin.js fetch-operations --token <token> --endpoint http://localhost:8080
+```
+
+The script will normalize the data in each bucket to a single CLEAR operation, followed by the latest PUT operation for each row.

test-client/README.md

@@ -0,0 +1,23 @@
+{
+  "name": "test-client",
+  "repository": "https://github.com/powersync-ja/powersync-service",
+  "private": true,
+  "version": "0.1.0",
+  "main": "dist/index.js",
+  "bin": "dist/bin.js",
+  "license": "Apache-2.0",
+  "type": "module",
+  "scripts": {
+    "fetch-operations": "tsc -b && node dist/bin.js fetch-operations",
+    "build": "tsc -b",
+    "clean": "rm -rf ./dist && tsc -b --clean"
+  },
+  "dependencies": {
+    "@powersync/service-core": "workspace:*",
+    "commander": "^12.0.0"
+  },
+  "devDependencies": {
+    "typescript": "^5.2.2",
+    "@types/node": "18.11.11"
+  }
+}

test-client/package.json

@@ -0,0 +1,14 @@
+import { program } from 'commander';
+import { getCheckpointData } from './client.js';
+
+program
+  .command('fetch-operations')
+  .option('-t, --token [token]')
+  .option('-e, --endpoint [endpoint]')
+  .option('--raw')
+  .action(async (options) => {
+    const data = await getCheckpointData({ endpoint: options.endpoint, token: options.token, raw: options.raw });
+    console.log(JSON.stringify(data, null, 2));
+  });
+
+await program.parseAsync();

test-client/src/bin.ts

@@ -0,0 +1,42 @@
+import { ndjsonStream } from './ndjson.js';
+import type * as types from '@powersync/service-core';
+import { isCheckpoint, isCheckpointComplete, isStreamingSyncData, normalizeData } from './util.js';
+
+export interface GetCheckpointOptions {
+  endpoint: string;
+  token: string;
+  raw?: boolean;
+}
+
+export async function getCheckpointData(options: GetCheckpointOptions) {
+  const response = await fetch(`${options.endpoint}/sync/stream`, {
+    method: 'POST',
+    headers: {
+      'Content-Type': 'application/json',
+      Authorization: `Token ${options.token}`
+    },
+    body: JSON.stringify({
+      raw_data: true,
+      include_checksum: true,
+      parameters: {}
+    } satisfies types.StreamingSyncRequest)
+  });
+  if (!response.ok) {
+    throw new Error(response.statusText + '\n' + (await response.text()));
+  }
+
+  let data: types.StreamingSyncData[] = [];
+  let checkpoint: types.StreamingSyncCheckpoint;
+
+  for await (let chunk of ndjsonStream<types.StreamingSyncLine>(response.body!)) {
+    if (isStreamingSyncData(chunk)) {
+      data.push(chunk);
+    } else if (isCheckpoint(chunk)) {
+      checkpoint = chunk;
+    } else if (isCheckpointComplete(chunk)) {
+      break;
+    }
+  }
+
+  return normalizeData(checkpoint!, data, { raw: options.raw ?? false });
+}

test-client/src/client.ts

@@ -0,0 +1,62 @@
+// import { ReadableStream } from 'node:stream/web';
+
+export function ndjsonStream<T>(response: ReadableStream<Uint8Array>): ReadableStream<T> & AsyncIterable<T> {
+  // For cancellation
+  var is_reader: any,
+    cancellationRequest = false;
+  return new ReadableStream<T>({
+    start: function (controller) {
+      var reader = response.getReader();
+      is_reader = reader;
+      var decoder = new TextDecoder();
+      var data_buf = '';
+
+      reader.read().then(function processResult(result): void | Promise<any> {
+        if (result.done) {
+          if (cancellationRequest) {
+            // Immediately exit
+            return;
+          }
+
+          data_buf = data_buf.trim();
+          if (data_buf.length !== 0) {
+            try {
+              var data_l = JSON.parse(data_buf);
+              controller.enqueue(data_l);
+            } catch (e) {
+              controller.error(e);
+              return;
+            }
+          }
+          controller.close();
+          return;
+        }
+
+        var data = decoder.decode(result.value, { stream: true });
+        data_buf += data;
+        var lines = data_buf.split('\n');
+        for (var i = 0; i < lines.length - 1; ++i) {
+          var l = lines[i].trim();
+          if (l.length > 0) {
+            try {
+              var data_line = JSON.parse(l);
+              controller.enqueue(data_line);
+            } catch (e) {
+              controller.error(e);
+              cancellationRequest = true;
+              reader.cancel();
+              return;
+            }
+          }
+        }
+        data_buf = lines[lines.length - 1];
+
+        return reader.read().then(processResult);
+      });
+    },
+    cancel: function (reason) {
+      cancellationRequest = true;
+      is_reader.cancel();
+    }
+  }) as any;
+}

test-client/src/index.ts

@@ -0,0 +1,104 @@
+import type * as types from '@powersync/service-core';
+
+export type BucketData = Record<string, types.OplogEntry[]>;
+
+export function normalizeData(
+  checkpoint: types.StreamingSyncCheckpoint,
+  chunks: types.StreamingSyncData[],
+  options: { raw: boolean }
+) {
+  const lastOpId = BigInt(checkpoint.checkpoint.last_op_id);
+  let buckets: BucketData = {};
+  for (let {
+    data: { bucket, data }
+  } of chunks) {
+    buckets[bucket] ??= [];
+    for (let entry of data) {
+      if (BigInt(entry.op_id) > lastOpId) {
+        continue;
+      }
+      buckets[bucket].push(entry);
+    }
+  }
+
+  if (options.raw) {
+    return buckets;
+  }
+
+  return Object.fromEntries(
+    Object.entries(buckets).map(([bucket, entries]) => {
+      return [bucket, reduceBucket(entries)];
+    })
+  );
+}
+
+export function isStreamingSyncData(line: types.StreamingSyncLine): line is types.StreamingSyncData {
+  return (line as types.StreamingSyncData).data != null;
+}
+
+export function isCheckpointComplete(line: types.StreamingSyncLine): line is types.StreamingSyncCheckpointComplete {
+  return (line as types.StreamingSyncCheckpointComplete).checkpoint_complete != null;
+}
+export function isCheckpoint(line: types.StreamingSyncLine): line is types.StreamingSyncCheckpoint {
+  return (line as types.StreamingSyncCheckpoint).checkpoint != null;
+}
+
+/**
+ * Reduce a bucket to the final state as stored on the client.
+ *
+ * This keeps the final state for each row as a PUT operation.
+ *
+ * All other operations are replaced with a single CLEAR operation,
+ * summing their checksums, and using a 0 as an op_id.
+ *
+ * This is the function $r(B)$, as described in /docs/bucket-properties.md.
+ */
+export function reduceBucket(operations: types.OplogEntry[]) {
+  let rowState = new Map<string, types.OplogEntry>();
+  let otherChecksum = 0;
+
+  for (let op of operations) {
+    const key = rowKey(op);
+    if (op.op == 'PUT') {
+      const existing = rowState.get(key);
+      if (existing) {
+        otherChecksum = addChecksums(otherChecksum, existing.checksum as number);
+      }
+      rowState.set(key, op);
+    } else if (op.op == 'REMOVE') {
+      const existing = rowState.get(key);
+      if (existing) {
+        otherChecksum = addChecksums(otherChecksum, existing.checksum as number);
+      }
+      rowState.delete(key);
+      otherChecksum = addChecksums(otherChecksum, op.checksum as number);
+    } else if (op.op == 'CLEAR') {
+      rowState.clear();
+      otherChecksum = op.checksum as number;
+    } else if (op.op == 'MOVE') {
+      otherChecksum = addChecksums(otherChecksum, op.checksum as number);
+    } else {
+      throw new Error(`Unknown operation ${op.op}`);
+    }
+  }
+
+  const puts = [...rowState.values()].sort((a, b) => {
+    return Number(BigInt(a.op_id) - BigInt(b.op_id));
+  });
+
+  let finalState: types.OplogEntry[] = [
+    // Special operation to indiciate the checksum remainder
+    { op_id: '0', op: 'CLEAR', checksum: otherChecksum },
+    ...puts
+  ];
+
+  return finalState;
+}
+
+function rowKey(entry: types.OplogEntry) {
+  return `${entry.object_type}/${entry.object_id}/${entry.subkey}`;
+}
+
+export function addChecksums(a: number, b: number) {
+  return (a + b) & 0xffffffff;
+}

test-client/src/ndjson.ts

@@ -0,0 +1,12 @@
+{
+  "extends": "../tsconfig.base.json",
+  "compilerOptions": {
+    "lib": ["ES2023", "DOM"],
+    "rootDir": "src",
+    "outDir": "dist",
+    "esModuleInterop": true,
+    "sourceMap": true
+  },
+  "include": ["src"],
+  "references": []
+}