- Use own interval sync

- Remove interval endpoint

Author: digitalmio

README.md

@@ -28,8 +28,6 @@ The `/v2` and `/v3` endpoints are more complex. Turso Edge POP will run all `exe
 
 The app also exposes a `/health` and `/version` informational endpoints.
 
-Lastly, you can use `/sync` endpoint to manually trigger a sync of the _local database_ with the origin database. Please note that this will sync local POP only (ie won't send pubsub message to other POPs).
-
 Turso Edge POP supports HTTP requests only (no websocket support etc), hence you need to use `http`/`https` url when connecting to POP.
 
 When you point your client to the POP, it will either respond with local data or seamlessly proxy your request to the origin server. This process is automatic and does not require any special setup from the client. On successful origin response, POP will also process a `sync` update to fetch a fresh copy if any writes were made.

bun.lock

@@ -12,6 +12,7 @@
         "ioredis": "^5.4.2",
         "node-sql-parser": "^5.3.6",
         "pino": "^9.6.0",
+        "set-interval-async": "^3.0.3",
         "zod": "^3.24.1",
       },
       "devDependencies": {
@@ -150,6 +151,8 @@
 
     "safe-stable-stringify": ["[email protected]", "", {}, "sha512-b3rppTKm9T+PsVCBEOUR46GWI7fdOs00VKZ1+9c1EWDaDMvjQc6tUwuFyIprgGgTcWoVHSKrU8H31ZHA2e0RHA=="],
 
+    "set-interval-async": ["[email protected]", "", {}, "sha512-o4DyBv6mko+A9cH3QKek4SAAT5UyJRkfdTi6JHii6ZCKUYFun8SwgBmQrOXd158JOwBQzA+BnO8BvT64xuCaSw=="],
+
     "sonic-boom": ["[email protected]", "", { "dependencies": { "atomic-sleep": "^1.0.0" } }, "sha512-INb7TM37/mAcsGmc9hyyI6+QR3rR1zVRu36B0NeGXKnOOLiZOfER5SA+N7X7k3yUYRzLWafduTDvJAfDswwEww=="],
 
     "split2": ["[email protected]", "", {}, "sha512-UcjcJOWknrNkF6PLX83qcHM6KHgVKNkV62Y8a5uYDVv9ydGQVwAHMKqHdJje1VTWpljG0WYpCDhrCdAOYH4TWg=="],

package.json

@@ -22,6 +22,7 @@
     "ioredis": "^5.4.2",
     "node-sql-parser": "^5.3.6",
     "pino": "^9.6.0",
+    "set-interval-async": "^3.0.3",
     "zod": "^3.24.1"
   }
 }

src/helpers/env.ts

@@ -7,7 +7,7 @@ export const env = createEnv({
     // Turso
     TURSO_DATABASE_URL: z.string().url(),
     TURSO_AUTH_TOKEN: z.string().startsWith("ey"),
-    TURSO_SYNC_INTERVAL: z.coerce.number().default(60),
+    TURSO_SYNC_INTERVAL: z.coerce.number().min(3).default(60),
 
     // Local SQL database file
     DB_FILEPATH: z.string().default("/app/data/local.db"),

src/helpers/redis.ts

@@ -1,6 +1,6 @@
 import Redis from "ioredis";
 import { env } from "./env";
-import { tursoClient } from "./turso-client";
+import { tursoClientSync } from "./turso-client";
 
 // why 2 instances?
 // - _redisPub_ is used for publishing sync events
@@ -23,6 +23,6 @@ export const publishRedisSyncCommand = async () => {
   if (redisPub) {
     redisPub.publish(env.REDIS_SYNC_CHANNEL, "sync");
   } else {
-    await tursoClient.sync();
+    await tursoClientSync("api");
   }
 };

src/helpers/truso-proxy.ts

@@ -1,6 +1,5 @@
 import type { ContentfulStatusCode } from "hono/utils/http-status";
 import { env } from "./env";
-import { tursoClient } from "./turso-client";
 
 export const tursoProxy = async (requests: string) => {
   const response = await fetch(`${env.TURSO_DATABASE_URL}/v3/pipeline`, {
@@ -14,11 +13,6 @@ export const tursoProxy = async (requests: string) => {
 
   const status = response.status as ContentfulStatusCode;
 
-  // Sync to local database
-  if (status === 200) {
-    await tursoClient.sync();
-  }
-
   return {
     status,
     result: status === 200 ? await response.json() : await response.text(),

src/helpers/turso-client.ts

@@ -1,11 +1,33 @@
 import { createClient } from "@libsql/client";
 import { env } from "./env";
+import { logger } from "./logger";
 
+let isSyncing = false;
+
+const setIsSyncing = (value = true) => {
+  isSyncing = value;
+};
+
+// we are not using "built in" syncInterval, as this can corrupt the database if run in parallel to manual syncs
+// due to this fact, interval syncing is done manually, check src/jobs/interval-sync.ts
 export const tursoClient = createClient({
   url: `file:${env.DB_FILEPATH}`,
   syncUrl: env.TURSO_DATABASE_URL,
   authToken: env.TURSO_AUTH_TOKEN,
-  syncInterval: env.TURSO_SYNC_INTERVAL,
   encryptionKey: env.DB_ENCRYPTION_KEY,
   fetch, // use Bun native fetch
 });
+
+export const tursoClientSync = async (type: "interval" | "pubsub" | "api") => {
+  const typeUpperFirst = type.charAt(0).toUpperCase() + type.slice(1);
+  if (!isSyncing) {
+    setIsSyncing(true);
+    await tursoClient.sync();
+    logger.info(`${typeUpperFirst} Sync triggered`);
+    setIsSyncing(false);
+  } else {
+    logger.info(
+      `${typeUpperFirst} Sync skipped, parallel sync already running`,
+    );
+  }
+};

src/index.ts

@@ -3,11 +3,10 @@ import type { Env as HonoPinoEnv } from "hono-pino";
 import { cors } from "hono/cors";
 import { appVersion, region } from "./helpers/app-vars";
 import { env } from "./helpers/env";
-import { redisSub } from "./helpers/redis";
-import { syncJob } from "./jobs/sync";
+import { intervalSyncJob } from "./jobs/interval-sync";
+import { syncJobPubsub } from "./jobs/pubsub-sync";
 import { appendLoggerInfo, registerLogger } from "./middlewares/logger";
 import healthRoute from "./routes/health";
-import syncRoute from "./routes/sync";
 import queryRoute from "./routes/v0-query";
 import pipelineRoute from "./routes/v2-v3-pipeline";
 import versionRoute from "./routes/version";
@@ -17,8 +16,13 @@ const app = new Hono<HonoPinoEnv>()
   .use(appendLoggerInfo)
   .use(cors());
 
-syncJob();
+// listen for pubsub sync
+syncJobPubsub();
 
+// run interval sync
+intervalSyncJob();
+
+// routes
 app.route("/", queryRoute);
 
 app.get("/v2", (c) => c.body(null, 200));
@@ -28,15 +32,14 @@ app.route("/v3/pipeline", pipelineRoute);
 
 app.route("/version", versionRoute);
 app.route("/health", healthRoute);
-app.route("/sync", syncRoute);
 
 if (!env.QUIET || process.env.NODE_ENV !== "production") {
   console.log(`🚀 Turso Edge Pop server running on port ${env.PORT}`);
   console.log(`📦 Version: ${appVersion}`);
   console.log(`🌎 Region: ${region}`);
   console.log(`💾 Database path: ${env.DB_FILEPATH}`);
   console.log(`⏱️ Sync Internal: ${env.TURSO_SYNC_INTERVAL} seconds`);
-  console.log(`🔄 Redis sync: ${Boolean(redisSub)}`);
+  console.log(`🔄 Redis sync: ${Boolean(env.REDIS_CONNECTION_STRING)}`);
   console.log(`------------------------------------------------------------------
 `);
 }

src/jobs/interval-sync.ts

@@ -0,0 +1,11 @@
+import { setIntervalAsync } from "set-interval-async";
+import { env } from "../helpers/env";
+import { tursoClientSync } from "../helpers/turso-client";
+
+export const intervalSyncJob = async () => {
+  await tursoClientSync("interval");
+
+  setIntervalAsync(async () => {
+    await tursoClientSync("interval");
+  }, env.TURSO_SYNC_INTERVAL * 1000);
+};

src/jobs/sync.ts renamed to src/jobs/pubsub-sync.ts

@@ -2,11 +2,11 @@ import debounce from "debounce";
 import { env } from "../helpers/env";
 import { logger } from "../helpers/logger";
 import { redisSub } from "../helpers/redis";
-import { tursoClient } from "../helpers/turso-client";
+import { tursoClientSync } from "../helpers/turso-client";
 
 export let lastSyncTimestamp = 0;
 
-export const syncJob = () => {
+export const syncJobPubsub = () => {
   if (redisSub) {
     // subscribe
     redisSub.subscribe(env.REDIS_SYNC_CHANNEL, (err, count) => {
@@ -32,7 +32,7 @@ export const syncJob = () => {
             "Received 'sync' message on channel %s. Syncing database.",
             channel,
           );
-          await tursoClient.sync();
+          await tursoClientSync("pubsub");
           lastSyncTimestamp = Math.floor(Date.now() / 1000);
         }, env.REDIS_SYNC_DEBOUNCE * 1000);
       }