Merge pull request #54 from powersync-ja/test-client

Add test client

pnpm-workspace.yaml

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

test-client/README.md

@@ -0,0 +1,40 @@
+# Test Client
+
+This is a minimal client demonstrating direct usage of the HTTP stream sync api.
+
+For a full implementation, see our client SDKs.
+
+## Usage
+
+```sh
+# In project root
+pnpm install
+pnpm build:packages
+# In this folder
+pnpm build
+node dist/bin.js fetch-operations --token <token> --endpoint http://localhost:8080
+
+# More examples:
+
+# If the endpoint is present in token aud field, it can be omitted from args:
+node dist/bin.js fetch-operations --token <token>
+
+# If a local powersync.yaml is present with a configured HS256 key, this can be used:
+node dist/bin.js fetch-operations --config path/to/powersync.yaml --endpoint http://localhost:8080
+
+# Without endpoint, it defaults to http://127.0.0.1:<port> from the config:
+node dist/bin.js fetch-operations --config path/to/powersync.yaml
+
+# Use --sub to specify a user id in the generated token:
+node dist/bin.js fetch-operations --config path/to/powersync.yaml --sub test-user
+```
+
+The `fetch-operations` command downloads data for a single checkpoint, and outputs a normalized form: one CLEAR operation, followed by the latest PUT operation for each row. This normalized form is still split per bucket. The output is not affected by compacting, but can be affected by replication order.
+
+To avoid normalizing the data, use the `--raw` option. This may include additional CLEAR, MOVE, REMOVE and duplicate PUT operations.
+
+To generate a token without downloading data, use the `generate-token` command:
+
+```sh
+node dist/bin.js generate-token --config path/to/powersync.yaml --sub test-user
+```

test-client/package.json

@@ -0,0 +1,26 @@
+{
+  "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",
+    "generate-token": "tsc -b && node dist/bin.js generate-token",
+    "build": "tsc -b",
+    "clean": "rm -rf ./dist && tsc -b --clean"
+  },
+  "dependencies": {
+    "@powersync/service-core": "workspace:*",
+    "commander": "^12.0.0",
+    "jose": "^4.15.1",
+    "yaml": "^2.5.0"
+  },
+  "devDependencies": {
+    "@types/node": "18.11.11",
+    "typescript": "^5.2.2"
+  }
+}

test-client/src/auth.ts

@@ -0,0 +1,60 @@
+import * as jose from 'jose';
+import * as fs from 'node:fs/promises';
+import * as yaml from 'yaml';
+
+export interface CredentialsOptions {
+  token?: string;
+  endpoint?: string;
+  config?: string;
+  sub?: string;
+}
+
+export async function getCredentials(options: CredentialsOptions): Promise<{ endpoint: string; token: string }> {
+  if (options.token != null) {
+    if (options.endpoint != null) {
+      return { token: options.token, endpoint: options.endpoint };
+    } else {
+      const parsed = jose.decodeJwt(options.token);
+      const aud = Array.isArray(parsed.aud) ? parsed.aud[0] : parsed.aud;
+      if (!(aud ?? '').startsWith('http')) {
+        throw new Error(`Specify endpoint, or aud in the token`);
+      }
+      return {
+        token: options.token,
+        endpoint: aud!
+      };
+    }
+  } else if (options.config != null) {
+    const file = await fs.readFile(options.config, 'utf-8');
+    const parsed = await yaml.parse(file);
+    const keys = (parsed.client_auth?.jwks?.keys ?? []).filter((key: any) => key.alg == 'HS256');
+    if (keys.length == 0) {
+      throw new Error('No HS256 key found in the config');
+    }
+
+    let endpoint = options.endpoint;
+    if (endpoint == null) {
+      endpoint = `http://127.0.0.1:${parsed.port ?? 8080}`;
+    }
+
+    const aud = parsed.client_auth?.audience?.[0] ?? endpoint;
+
+    const rawKey = keys[0];
+    const key = await jose.importJWK(rawKey);
+
+    const sub = options.sub ?? 'test_user';
+
+    const token = await new jose.SignJWT({})
+      .setProtectedHeader({ alg: rawKey.alg, kid: rawKey.kid })
+      .setSubject(sub)
+      .setIssuedAt()
+      .setIssuer('test-client')
+      .setAudience(aud)
+      .setExpirationTime('1h')
+      .sign(key);
+
+    return { token, endpoint };
+  } else {
+    throw new Error(`Specify token or config path`);
+  }
+}

test-client/src/bin.ts

@@ -0,0 +1,32 @@
+import { program } from 'commander';
+import { getCheckpointData } from './client.js';
+import { getCredentials } from './auth.js';
+import * as jose from 'jose';
+
+program
+  .command('fetch-operations')
+  .option('-t, --token [token]', 'JWT to use for authentication')
+  .option('-e, --endpoint [endpoint]', 'endpoint URI')
+  .option('-c, --config [config]', 'path to powersync.yaml, to auto-generate a token from a HS256 key')
+  .option('-u, --sub [sub]', 'sub field for auto-generated token')
+  .option('--raw', 'output operations as received, without normalizing')
+  .action(async (options) => {
+    const credentials = await getCredentials(options);
+    const data = await getCheckpointData({ ...credentials, raw: options.raw });
+    console.log(JSON.stringify(data, null, 2));
+  });
+
+program
+  .command('generate-token')
+  .description('Generate a JWT from for a given powersync.yaml config file')
+  .option('-c, --config [config]', 'path to powersync.yaml')
+  .option('-u, --sub [sub]', 'sub field for auto-generated token')
+  .action(async (options) => {
+    const credentials = await getCredentials(options);
+    const decoded = await jose.decodeJwt(credentials.token);
+
+    console.error(`Payload:\n${JSON.stringify(decoded, null, 2)}\nToken:`);
+    console.log(credentials.token);
+  });
+
+await program.parseAsync();

test-client/src/client.ts

@@ -0,0 +1,45 @@
+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,
+      // Client parameters can be specified here
+      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)) {
+      // Collect data
+      data.push(chunk);
+    } else if (isCheckpoint(chunk)) {
+      checkpoint = chunk;
+    } else if (isCheckpointComplete(chunk)) {
+      // Stop on the first checkpoint_complete message.
+      break;
+    }
+  }
+
+  return normalizeData(checkpoint!, data, { raw: options.raw ?? false });
+}

test-client/src/index.ts

@@ -0,0 +1,3 @@
+export * from './client.js';
+export * from './ndjson.js';
+export * from './util.js';

test-client/src/ndjson.ts

@@ -0,0 +1,59 @@
+export function ndjsonStream<T>(response: ReadableStream<Uint8Array>): ReadableStream<T> & AsyncIterable<T> {
+  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;
+}