Merge pull request #64 from powersync-ja/client-id

Support client_id and User-Agent

Author: rkistner

.changeset/rude-timers-matter.md

@@ -0,0 +1,6 @@
+---
+'@powersync/service-core': minor
+'@powersync/service-image': minor
+---
+
+Support client_id parameter and User-Agent headers.

packages/service-core/src/routes/configure-rsocket.ts

@@ -23,7 +23,7 @@ export function configureRSocket(router: ReactiveSocketRouter<Context>, options:
 
   router.applyWebSocketEndpoints(server, {
     contextProvider: async (data: Buffer) => {
-      const { token } = RSocketContextMeta.decode(deserialize(data) as any);
+      const { token, user_agent } = RSocketContextMeta.decode(deserialize(data) as any);
 
       if (!token) {
         throw new errors.AuthorizationError('No token provided');
@@ -38,6 +38,7 @@ export function configureRSocket(router: ReactiveSocketRouter<Context>, options:
           }
           return {
             token,
+            user_agent,
             ...context,
             token_errors: token_errors,
             system

packages/service-core/src/routes/endpoints/checkpointing.ts

@@ -5,7 +5,9 @@ import * as util from '../../util/util-index.js';
 import { authUser } from '../auth.js';
 import { routeDefinition } from '../router.js';
 
-const WriteCheckpointRequest = t.object({});
+const WriteCheckpointRequest = t.object({
+  client_id: t.string.optional()
+});
 
 export const writeCheckpoint = routeDefinition({
   path: '/write-checkpoint.json',
@@ -30,8 +32,10 @@ export const writeCheckpoint2 = routeDefinition({
   validator: schema.createTsCodecValidator(WriteCheckpointRequest, { allowAdditional: true }),
   handler: async (payload) => {
     const { user_id, system } = payload.context;
+    const client_id = payload.params.client_id;
+    const full_user_id = util.checkpointUserId(user_id, client_id);
     const storage = system.storage;
-    const write_checkpoint = await util.createWriteCheckpoint(system.requirePgPool(), storage, user_id!);
+    const write_checkpoint = await util.createWriteCheckpoint(system.requirePgPool(), storage, full_user_id);
     return {
       write_checkpoint: String(write_checkpoint)
     };

packages/service-core/src/routes/endpoints/socket-route.ts

@@ -124,6 +124,7 @@ export const syncStreamReactive: SocketRouteGenerator = (router) =>
         disposer();
         logger.info(`Sync stream complete`, {
           user_id: syncParams.user_id,
+          user_agent: context.user_agent,
           operations_synced: tracker.operationsSynced,
           data_synced_bytes: tracker.dataSyncedBytes
         });

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

@@ -21,6 +21,9 @@ export const syncStreamed = routeDefinition({
   validator: schema.createTsCodecValidator(util.StreamingSyncRequest, { allowAdditional: true }),
   handler: async (payload) => {
     const system = payload.context.system;
+    const headers = payload.request.headers;
+    const userAgent = headers['x-user-agent'] ?? headers['user-agent'];
+    const clientId = payload.params.client_id;
 
     if (system.closed) {
       throw new errors.JourneyError({
@@ -92,6 +95,8 @@ export const syncStreamed = routeDefinition({
           Metrics.getInstance().concurrent_connections.add(-1);
           logger.info(`Sync stream complete`, {
             user_id: syncParams.user_id,
+            client_id: clientId,
+            user_agent: userAgent,
             operations_synced: tracker.operationsSynced,
             data_synced_bytes: tracker.dataSyncedBytes
           });

packages/service-core/src/routes/router-socket.ts

@@ -9,5 +9,6 @@ import { Context } from './router.js';
 export type SocketRouteGenerator = (router: ReactiveSocketRouter<Context>) => IReactiveStream;
 
 export const RSocketContextMeta = t.object({
-  token: t.string
+  token: t.string,
+  user_agent: t.string.optional()
 });

packages/service-core/src/routes/router.ts

@@ -11,6 +11,10 @@ export type Context = {
 
   token_payload?: auth.JwtPayload;
   token_errors?: string[];
+  /**
+   * Only on websocket endpoints.
+   */
+  user_agent?: string;
 };
 
 export type BasicRouterRequest = {

packages/service-core/src/sync/sync.ts

@@ -93,7 +93,8 @@ async function* streamResponseInner(
     }
   }
 
-  const stream = storage.watchWriteCheckpoint(syncParams.token_parameters.user_id as string, signal);
+  const checkpointUserId = util.checkpointUserId(syncParams.token_parameters.user_id as string, params.client_id);
+  const stream = storage.watchWriteCheckpoint(checkpointUserId, signal);
   for await (const next of stream) {
     const { base, writeCheckpoint } = next;
     const checkpoint = base.checkpoint;

packages/service-core/src/util/protocol-types.ts

@@ -94,7 +94,12 @@ export const StreamingSyncRequest = t.object({
   /**
    * Client parameters to be passed to the sync rules.
    */
-  parameters: t.record(t.any).optional()
+  parameters: t.record(t.any).optional(),
+
+  /**
+   * Unique client id.
+   */
+  client_id: t.string.optional()
 });
 
 export type StreamingSyncRequest = t.Decoded<typeof StreamingSyncRequest>;

packages/service-core/src/util/utils.ts

@@ -120,3 +120,13 @@ export async function createWriteCheckpoint(
   logger.info(`Write checkpoint 2: ${JSON.stringify({ lsn, id: String(id) })}`);
   return id;
 }
+
+export function checkpointUserId(user_id: string | undefined, client_id: string | undefined) {
+  if (user_id == null) {
+    throw new Error('user_id is required');
+  }
+  if (client_id == null) {
+    return user_id;
+  }
+  return `${user_id}/${client_id}`;
+}