Write Checkpoints

src/api/auth.js

@@ -56,12 +56,16 @@ router.get('/token', async (req, res) => {
   await ensureKeys();
   const powerSyncKey = keys.privateKey;
 
+  const { user_id = 'UserID ' } = req.query;
+
+  console.log('UserID for token is', user_id);
+
   const token = await new SignJWT({})
     .setProtectedHeader({
       alg: powerSyncKey.alg,
       kid: powerSyncKey.kid
     })
-    .setSubject('UserID')
+    .setSubject(user_id)
     .setIssuedAt()
     .setIssuer(config.powersync.jwtIssuer)
     .setAudience(config.powersync.url)

src/api/data.js

@@ -6,7 +6,7 @@ const router = express.Router();
 
 const persistenceFactory = factories[config.database.type];
 
-const { updateBatch } = await persistenceFactory(config.database.uri);
+const { updateBatch, createCheckpoint } = await persistenceFactory(config.database.uri);
 
 /**
  * Handle a batch of events.
@@ -58,6 +58,22 @@ router.put('/', async (req, res) => {
   }
 });
 
+router.put('/checkpoint', async (req, res) => {
+  if (!req.body) {
+    res.status(400).send({
+      message: 'Invalid body provided'
+    });
+    return;
+  }
+  const { user_id = 'UserID', client_id = '1' } = req.body;
+
+  const checkpoint = await createCheckpoint(user_id, client_id);
+
+  res.status(200).send({
+    checkpoint
+  });
+});
+
 /**
  * Handle all PATCH events sent to the server by the client PowerSync application
  */

src/persistance/mongo/mongo-persistance.js

@@ -13,10 +13,25 @@ export const createMongoPersister = async (uri) => {
   const db = client.db();
   await client.connect();
 
-  return {
-    /**
-     * @type {import('../persister-factories.js').BatchPersister}
-     */
+  /**
+   * @type {import('../persister-factories.js').Persister}
+   */
+  const persister = {
+    createCheckpoint: async (user_id, client_id) => {
+      const doc = await db.collection('checkpoints').findOneAndUpdate(
+        {
+          user_id,
+          client_id
+        },
+        {
+          $inc: {
+            checkpoint: 1n
+          }
+        },
+        { 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,
@@ -53,4 +68,6 @@ export const createMongoPersister = async (uri) => {
       }
     }
   };
+
+  return persister;
 };

src/persistance/mysql/mysql-persistance.js

@@ -13,8 +13,43 @@ export const createMySQLPersister = (uri) => {
   console.debug('Using MySQL Persister');
 
   const pool = mysql.createPool(uri);
+  /**
+   * @type {import('../persister-factories.js').Persister}
+   */
+  const persister = {
+    async createCheckpoint(user_id, client_id) {
+      const connection = await pool.getConnection();
+      try {
+        await connection.beginTransaction();
+        await connection.query(
+          `
+      INSERT INTO checkpoints
+         (user_id, client_id, checkpoint)
+      VALUES (?, ?, 1)
+      ON DUPLICATE KEY UPDATE 
+        checkpoint = checkpoint + 1;
+      `,
+          [user_id, client_id]
+        );
+        const [rows] = await connection.query(
+          `
+           SELECT checkpoint FROM checkpoints WHERE user_id = ? AND client_id = ?;
+           `,
+          [user_id, client_id]
+        );
 
-  return {
+        await connection.commit();
+        /**
+         * @type {bigint}
+         */
+        const checkpoint = rows[0].checkpoint;
+        return checkpoint;
+      } catch (ex) {
+        await connection.rollback();
+      } finally {
+        connection.release();
+      }
+    },
     /**
      * @type {import('../persister-factories.js').BatchPersister}
      */
@@ -82,4 +117,5 @@ export const createMySQLPersister = (uri) => {
       }
     }
   };
+  return persister;
 };

src/persistance/persister-factories.js

@@ -27,8 +27,14 @@ import { createPostgresPersister } from './postgres/postgres-persistance.js';
  * @param {(DeleteOp | PutOp | PatchOp)[]} batch
  * @returns {Promise<void>}
  * 
+ * @callback CreateCheckpoint
+ * @param {string} user_id
+ * @param {string} client_id
+ * @returns {Promise<bigint>} checkpoint
+ * 
  * @typedef {Object} Persister
  * @prop {BatchPersister} updateBatch
+ * @prop {CreateCheckpoint} createCheckpoint
 
  * @callback PersisterFactory
  * @param {string} URI -

src/persistance/postgres/postgres-persistance.js

@@ -30,10 +30,10 @@ export const createPostgresPersister = (uri) => {
     console.error('Pool connection failure to postgres:', err, client);
   });
 
-  return {
-    /**
-     * @type {import('../persister-factories.js').BatchPersister}
-     */
+  /**
+   * @type {import('../persister-factories.js').Persister}
+   */
+  const persister = {
     updateBatch: async (batch) => {
       const client = await pool.connect();
       try {
@@ -109,6 +109,27 @@ export const createPostgresPersister = (uri) => {
       } finally {
         client.release();
       }
+    },
+    async createCheckpoint(user_id, client_id) {
+      const response = await pool.query(
+        `
+    INSERT INTO checkpoints(user_id, client_id, checkpoint)
+    VALUES 
+        ($1, $2, '1')
+    ON 
+        CONFLICT (user_id, client_id)
+    DO 
+        UPDATE SET checkpoint = checkpoints.checkpoint + 1
+    RETURNING checkpoint;
+    `,
+        [user_id, client_id]
+      );
+      /**
+       * @type {bigint}
+       */
+      const checkpoint = response.rows[0].checkpoint;
+      return checkpoint;
     }
   };
+  return persister;
 };