feat: custom checkpoint requests

README.md

@@ -33,17 +33,26 @@ The endpoints are as follows:
 
    - PowerSync uses this endpoint to sync delete events that occurred on the client application.
 
-7. PUT `/api/attachments/:id`
+7. POST `/api/data/checkpoint-request`
+
+   - PowerSync clients using custom checkpoint requests use this endpoint to record a checkpoint request after upload processing.
+   - Request body: `{ "user_id": "...", "client_id": "...", "checkpoint_request_id": "..." }`
+   - Response body: `{ "checkpoint_request_id": "..." }`
+   - The returned value is the effective checkpoint request state accepted by the backend. If a newer request is already recorded for the same user/client pair, that newer value is returned.
+   - The `checkpoints.checkpoint` column should be a 64-bit integer type. Add a nullable `checkpoint_requested_at` timestamp column; this route stamps it when it records a request-derived checkpoint, while the legacy `PUT /api/data/checkpoint` route clears it.
+   - In custom write checkpoint sync rules, project `checkpoint_request_id` only for request-derived rows, for example: `SELECT user_id, checkpoint, client_id, CASE WHEN checkpoint_requested_at IS NOT NULL THEN checkpoint END AS checkpoint_request_id FROM checkpoints`. This lets the PowerSync service mark those checkpoints as safe for timely cleanup.
+
+8. PUT `/api/attachments/:id`
 
    - PowerSync uses this endpoint to upload a file for use with the [attachments](https://docs.powersync.com/usage/use-case-examples/attachments-files) API.
 
-8. GET `/api/attachments/:id`
+9. GET `/api/attachments/:id`
 
    - PowerSync uses this endpoint to download a previously uploaded attachment.
 
-9. DELETE `/api/attachments/:id`
+10. DELETE `/api/attachments/:id`
 
-   - PowerSync uses this endpoint to delete a previously uploaded attachment.
+    - PowerSync uses this endpoint to delete a previously uploaded attachment.
 
 > Attachments are stored on the local filesystem under `attachments/` for demo purposes only — there is no auth on these endpoints and the directory is ephemeral under Docker. For production, back the client's storage adapter with an object store (S3, Cloudflare R2, Supabase Storage, etc.).
 

src/api/data.js

@@ -6,7 +6,40 @@ const router = express.Router();
 
 const persistenceFactory = factories[config.database.type];
 
-const { updateBatch, createCheckpoint } = await persistenceFactory(config.database.uri);
+const { updateBatch, createCheckpoint, createCheckpointRequest } = await persistenceFactory(config.database.uri);
+
+const CHECKPOINT_REQUEST_ID_MAX = 9_223_372_036_854_775_807n;
+
+function parseCheckpointRequestId(value) {
+  if (typeof value === 'bigint') {
+    return value;
+  }
+  if (typeof value === 'number') {
+    if (!Number.isSafeInteger(value)) {
+      throw new Error('checkpoint_request_id must be a safe integer when sent as a number');
+    }
+    return BigInt(value);
+  }
+  if (typeof value === 'string' && /^[0-9]+$/.test(value)) {
+    return BigInt(value);
+  }
+  throw new Error('checkpoint_request_id must be an integer');
+}
+
+function validateCheckpointRequestId(value) {
+  const parsed = parseCheckpointRequestId(value);
+  if (parsed <= 0n || parsed > CHECKPOINT_REQUEST_ID_MAX) {
+    throw new Error(`checkpoint_request_id must be between 1 and ${CHECKPOINT_REQUEST_ID_MAX}`);
+  }
+  return parsed;
+}
+
+function validateString(value, name) {
+  if (typeof value !== 'string' || value.length === 0) {
+    throw new Error(`${name} must be a non-empty string`);
+  }
+  return value;
+}
 
 /**
  * Handle a batch of events.
@@ -74,6 +107,51 @@ router.put('/checkpoint', async (req, res) => {
   });
 });
 
+/**
+ * Handle custom write checkpoint requests created by newer SDKs.
+ */
+router.post('/checkpoint-request', async (req, res) => {
+  if (!req.body) {
+    res.status(400).send({
+      message: 'Invalid body provided'
+    });
+    return;
+  }
+
+  let user_id;
+  let client_id;
+  let checkpoint_request_id;
+
+  try {
+    user_id = validateString(req.body.user_id, 'user_id');
+    client_id = validateString(req.body.client_id, 'client_id');
+    checkpoint_request_id = validateCheckpointRequestId(req.body.checkpoint_request_id);
+  } catch (e) {
+    res.status(400).send({
+      message: e.message
+    });
+    return;
+  }
+
+  try {
+    const acceptedCheckpointRequestId = await createCheckpointRequest(
+      user_id,
+      client_id,
+      checkpoint_request_id,
+      new Date()
+    );
+
+    res.status(200).send({
+      checkpoint_request_id: String(acceptedCheckpointRequestId)
+    });
+  } catch (e) {
+    console.error(e.stack ?? e.message);
+    res.status(400).send({
+      message: `Request failed: ${e.message}`
+    });
+  }
+});
+
 /**
  * Handle all PATCH events sent to the server by the client PowerSync application
  */

src/persistance/mongo/mongo-persistance.js

@@ -26,12 +26,50 @@ export const createMongoPersister = async (uri) => {
         {
           $inc: {
             checkpoint: 1n
+          },
+          $unset: {
+            checkpoint_requested_at: ''
           }
         },
         { upsert: true, returnDocument: 'after' }
       );
       return doc.checkpoint;
     },
+    createCheckpointRequest: async (user_id, client_id, checkpoint_request_id, checkpoint_requested_at) => {
+      const doc = await db.collection('checkpoints').findOneAndUpdate(
+        {
+          user_id,
+          client_id
+        },
+        [
+          {
+            $set: {
+              checkpoint: {
+                $cond: [
+                  { $gt: [checkpoint_request_id, { $ifNull: ['$checkpoint', 0n] }] },
+                  checkpoint_request_id,
+                  '$checkpoint'
+                ]
+              },
+              checkpoint_requested_at: {
+                $cond: [
+                  { $gt: [checkpoint_request_id, { $ifNull: ['$checkpoint', 0n] }] },
+                  checkpoint_requested_at,
+                  '$checkpoint_requested_at'
+                ]
+              },
+              user_id,
+              client_id
+            }
+          }
+        ],
+        {
+          upsert: true,
+          returnDocument: 'after'
+        }
+      );
+      return doc.checkpoint;
+    },
     updateBatch: async (batch) => {
       // TODO: Use batches & transactions.
       // TODO: Do type conversion. This currently persists data from the client as is,

src/persistance/mssql/mssql-persistance.js

@@ -1,7 +1,6 @@
 import { URL } from 'url';
 import sql from 'mssql';
 
-
 function escapeIdentifier(identifier) {
   return `[${identifier}]`;
 }
@@ -41,7 +40,6 @@ export const createMSSQLPersister = async (uri) => {
     updateBatch: async (batch) => {
       const transaction = await pool.transaction();
       try {
-
         await transaction.begin();
 
         for (let op of batch) {
@@ -57,7 +55,7 @@ export const createMSSQLPersister = async (uri) => {
             const columns = columnsEscaped.join(', ');
             // Parameter names should not have brackets: @id, @col1, @col3
             const columnParamaters = columnNames.map((c) => `@${c}`).join(', ');
-            const sourceColumns = columnsEscaped.map(column => `source.${column}`).join(', ');
+            const sourceColumns = columnsEscaped.map((column) => `source.${column}`).join(', ');
 
             let updateClauses = [];
             // [[col1] = source.col1, [col3] = source.col3]
@@ -69,7 +67,8 @@ export const createMSSQLPersister = async (uri) => {
             }
 
             // Update clause is omitted if there are no fields to update (Can only happen if the only record in data is the id)
-            const updateClause = updateClauses.length > 0 ? `WHEN MATCHED THEN UPDATE SET ${updateClauses.join(', ')}` : null;
+            const updateClause =
+              updateClauses.length > 0 ? `WHEN MATCHED THEN UPDATE SET ${updateClauses.join(', ')}` : null;
             const insertClause = `WHEN NOT MATCHED THEN INSERT (${columns}) VALUES (${sourceColumns})`;
 
             const statement = `
@@ -86,7 +85,6 @@ export const createMSSQLPersister = async (uri) => {
               request.input(column, with_id[column]);
             }
             await request.query(statement);
-
           } else if (op.op == 'PATCH') {
             const data = op.data;
             const with_id = { ...data, id: op.id ?? data.id };
@@ -128,32 +126,68 @@ export const createMSSQLPersister = async (uri) => {
     async createCheckpoint(user_id, client_id) {
       const transaction = await pool.transaction();
       try {
-      await transaction.begin();
+        await transaction.begin();
 
-      const statement = `
+        const statement = `
         MERGE INTO checkpoints AS t
         USING (VALUES (@user_id, @client_id, @checkpoint)) AS source (user_id, client_id, checkpoint)
           ON t.user_id = source.user_id AND t.client_id = source.client_id
         WHEN MATCHED THEN 
-          UPDATE SET checkpoint = t.checkpoint + 1
+          UPDATE SET
+            checkpoint = t.checkpoint + 1,
+            checkpoint_requested_at = NULL
         WHEN NOT MATCHED THEN 
           INSERT (user_id, client_id, checkpoint)
           VALUES (source.user_id, source.client_id, source.checkpoint)
         OUTPUT INSERTED.checkpoint;
       `;
-      const request = transaction.request();
-      request.input('user_id', user_id);
-      request.input('client_id', client_id);
-      request.input('checkpoint', 1);
-      const response = await request.query(statement);
-
-      await transaction.commit();
-      return response.recordset[0].checkpoint;
+        const request = transaction.request();
+        request.input('user_id', user_id);
+        request.input('client_id', client_id);
+        request.input('checkpoint', 1);
+        const response = await request.query(statement);
+
+        await transaction.commit();
+        return response.recordset[0].checkpoint;
+      } catch (e) {
+        await transaction.rollback();
+        throw e;
+      }
+    },
+    async createCheckpointRequest(user_id, client_id, checkpoint_request_id, checkpoint_requested_at) {
+      const transaction = await pool.transaction();
+      try {
+        await transaction.begin();
+
+        const statement = `
+          MERGE INTO checkpoints AS t
+          USING (VALUES (@user_id, @client_id, @checkpoint, @checkpoint_requested_at)) AS source (user_id, client_id, checkpoint, checkpoint_requested_at)
+            ON t.user_id = source.user_id AND t.client_id = source.client_id
+          WHEN MATCHED AND source.checkpoint > t.checkpoint THEN
+            UPDATE SET
+              checkpoint = source.checkpoint,
+              checkpoint_requested_at = source.checkpoint_requested_at
+          WHEN NOT MATCHED THEN
+            INSERT (user_id, client_id, checkpoint, checkpoint_requested_at)
+            VALUES (source.user_id, source.client_id, source.checkpoint, source.checkpoint_requested_at);
+
+          SELECT checkpoint
+          FROM checkpoints
+          WHERE user_id = @user_id AND client_id = @client_id;
+        `;
+        const request = transaction.request();
+        request.input('user_id', user_id);
+        request.input('client_id', client_id);
+        request.input('checkpoint', sql.BigInt, checkpoint_request_id.toString());
+        request.input('checkpoint_requested_at', sql.DateTimeOffset, checkpoint_requested_at);
+        const response = await request.query(statement);
+
+        await transaction.commit();
+        return response.recordset[0].checkpoint;
       } catch (e) {
         await transaction.rollback();
         throw e;
       }
-
     }
   };
   return persister;

src/persistance/mysql/mysql-persistance.js

@@ -27,7 +27,8 @@ export const createMySQLPersister = (uri) => {
          (user_id, client_id, checkpoint)
       VALUES (?, ?, 1)
       ON DUPLICATE KEY UPDATE 
-        checkpoint = checkpoint + 1;
+        checkpoint = checkpoint + 1,
+        checkpoint_requested_at = NULL;
       `,
           [user_id, client_id]
         );
@@ -50,6 +51,37 @@ export const createMySQLPersister = (uri) => {
         connection.release();
       }
     },
+    async createCheckpointRequest(user_id, client_id, checkpoint_request_id, checkpoint_requested_at) {
+      const connection = await pool.getConnection();
+      try {
+        await connection.beginTransaction();
+        await connection.query(
+          `
+      INSERT INTO checkpoints
+         (user_id, client_id, checkpoint, checkpoint_requested_at)
+      VALUES (?, ?, ?, ?)
+      ON DUPLICATE KEY UPDATE
+        checkpoint_requested_at = IF(VALUES(checkpoint) > checkpoint, VALUES(checkpoint_requested_at), checkpoint_requested_at),
+        checkpoint = GREATEST(checkpoint, VALUES(checkpoint));
+      `,
+          [user_id, client_id, checkpoint_request_id.toString(), checkpoint_requested_at]
+        );
+        const [rows] = await connection.query(
+          `
+           SELECT checkpoint FROM checkpoints WHERE user_id = ? AND client_id = ?;
+           `,
+          [user_id, client_id]
+        );
+
+        await connection.commit();
+        return rows[0].checkpoint;
+      } catch (ex) {
+        await connection.rollback();
+        throw ex;
+      } finally {
+        connection.release();
+      }
+    },
     /**
      * @type {import('../persister-factories.js').BatchPersister}
      */

src/persistance/persister-factories.js

@@ -32,10 +32,18 @@ import { createMSSQLPersister } from './mssql/mssql-persistance.js';
  * @param {string} user_id
  * @param {string} client_id
  * @returns {Promise<bigint>} checkpoint
+ *
+ * @callback CreateCheckpointRequest
+ * @param {string} user_id
+ * @param {string} client_id
+ * @param {bigint} checkpoint_request_id
+ * @param {Date} checkpoint_requested_at
+ * @returns {Promise<bigint|string|number>} checkpoint_request_id
  * 
  * @typedef {Object} Persister
  * @prop {BatchPersister} updateBatch
  * @prop {CreateCheckpoint} createCheckpoint
+ * @prop {CreateCheckpointRequest} createCheckpointRequest
 
  * @callback PersisterFactory
  * @param {string} URI -