Refactor write checkpoints to make testing easier.

Author: rkistner

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

@@ -56,25 +56,14 @@ export const writeCheckpoint2 = routeDefinition({
 
     const apiHandler = service_context.routerEngine!.getAPI();
 
-    const client_id = payload.params.client_id;
-    const full_user_id = util.checkpointUserId(user_id, client_id);
-
-    const currentCheckpoint = await apiHandler.getReplicationHead();
-    const {
-      storageEngine: { activeBucketStorage }
-    } = service_context;
-
-    const activeSyncRules = await activeBucketStorage.getActiveSyncRulesContent();
-    if (!activeSyncRules) {
-      throw new framework.errors.ValidationError(`Cannot create Write Checkpoint since no sync rules are active.`);
-    }
-
-    using syncBucketStorage = activeBucketStorage.getInstance(activeSyncRules);
-    const writeCheckpoint = await syncBucketStorage.createManagedWriteCheckpoint({
-      user_id: full_user_id,
-      heads: { '1': currentCheckpoint }
+    const { replicationHead, writeCheckpoint } = await util.createWriteCheckpoint({
+      userId: user_id,
+      clientId: payload.params.client_id,
+      api: apiHandler,
+      storage: service_context.storageEngine.activeBucketStorage
     });
-    logger.info(`Write checkpoint 2: ${JSON.stringify({ currentCheckpoint, id: String(full_user_id) })}`);
+
+    logger.info(`Write checkpoint for ${user_id}/${payload.params.client_id}: ${writeCheckpoint} | ${replicationHead}`);
 
     return {
       write_checkpoint: String(writeCheckpoint)

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

@@ -0,0 +1,40 @@
+import { ErrorCode, logger, ServiceError } from '@powersync/lib-services-framework';
+import { RouteAPI } from '../api/RouteAPI.js';
+import { BucketStorageFactory } from '../storage/BucketStorage.js';
+
+export interface CreateWriteCheckpointOptions {
+  userId: string | undefined;
+  clientId: string | undefined;
+  api: RouteAPI;
+  storage: BucketStorageFactory;
+}
+export async function createWriteCheckpoint(options: CreateWriteCheckpointOptions) {
+  const full_user_id = checkpointUserId(options.userId, options.clientId);
+
+  const currentCheckpoint = await options.api.getReplicationHead();
+
+  const activeSyncRules = await options.storage.getActiveSyncRulesContent();
+  if (!activeSyncRules) {
+    throw new ServiceError(ErrorCode.PSYNC_S2302, `Cannot create Write Checkpoint since no sync rules are active.`);
+  }
+
+  using syncBucketStorage = options.storage.getInstance(activeSyncRules);
+  const writeCheckpoint = await syncBucketStorage.createManagedWriteCheckpoint({
+    user_id: full_user_id,
+    heads: { '1': currentCheckpoint }
+  });
+  return {
+    writeCheckpoint: String(writeCheckpoint),
+    replicationHead: currentCheckpoint
+  };
+}
+
+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}`;
+}

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

@@ -5,6 +5,7 @@ export * from './Mutex.js';
 export * from './protocol-types.js';
 export * from './secs.js';
 export * from './utils.js';
+export * from './checkpointing.js';
 
 export * from './config.js';
 export * from './config/compound-config-collector.js';

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

@@ -145,16 +145,6 @@ export function isCompleteRow(storeData: boolean, row: sync_rules.ToastableSqlit
   return !hasToastedValues(row);
 }
 
-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}`;
-}
-
 /**
  * Reduce a bucket to the final state as stored on the client.
  *