refactor: abstract the stats syncer to support general-purpose synchr…

…onization workers

.changeset/long-turkeys-impress.md

@@ -0,0 +1,11 @@
+---
+"@blobscan/rest-api-server": minor
+"@blobscan/syncers": minor
+---
+
+Refactored the stats syncer package to support general-purpose synchronization workers/queues.
+
+Key changes include:
+
+    •	Renamed the package to syncers.
+    •	Exported each syncer directly, removing the StatsSyncer managing entity.

apps/rest-api-server/package.json

@@ -18,8 +18,7 @@
     "@blobscan/api": "workspace:^0.9.0",
     "@blobscan/logger": "workspace:^0.1.0",
     "@blobscan/open-telemetry": "workspace:^0.0.7",
-    "@blobscan/stats-syncer": "workspace:^0.1.8",
-    "@blobscan/swarm-syncer": "workspace:^0.0.1",
+    "@blobscan/syncers": "workspace:^0.1.9",
     "@blobscan/zod": "workspace:^0.1.0",
     "@opentelemetry/instrumentation-express": "^0.33.0",
     "@sentry/node": "^7.109.0",

apps/rest-api-server/src/index.ts

@@ -14,38 +14,16 @@ import {
   gracefulShutdown as apiGracefulShutdown,
 } from "@blobscan/api";
 import { collectDefaultMetrics } from "@blobscan/open-telemetry";
-import { StatsSyncer } from "@blobscan/stats-syncer";
-import { SwarmStampSyncer } from "@blobscan/swarm-syncer";
 
 import { env } from "./env";
 import { logger } from "./logger";
 import { morganMiddleware } from "./morgan";
 import { openApiDocument } from "./openapi";
-import { getNetworkDencunForkSlot } from "./utils";
+import { setUpSyncers } from "./syncers";
 
 collectDefaultMetrics();
 
-const statsSyncer = new StatsSyncer({
-  redisUri: env.REDIS_URI,
-  lowestSlot:
-    env.DENCUN_FORK_SLOT ?? getNetworkDencunForkSlot(env.NETWORK_NAME),
-});
-
-statsSyncer.start({
-  cronPatterns: {
-    daily: env.STATS_SYNCER_DAILY_CRON_PATTERN,
-    overall: env.STATS_SYNCER_OVERALL_CRON_PATTERN,
-  },
-});
-
-
-if (env.SWARM_STORAGE_ENABLED && env.SWARM_BATCH_ID) {
-  const swarmSyncer = new SwarmStampSyncer({
-    redisUri: env.REDIS_URI,
-    env.BEE_ENDPOINT, //FIXME
-  );
-  swarmSyncer.start(env.SWARM_SYNCER_CRON);
-}
+const closeSyncers = setUpSyncers();
 
 const app = express();
 
@@ -83,7 +61,7 @@ async function gracefulShutdown(signal: string) {
   logger.debug(`Received ${signal}. Shutting down...`);
 
   await apiGracefulShutdown()
-    .finally(() => statsSyncer.close())
+    .finally(() => closeSyncers())
     .finally(() => {
       server.close(() => {
         logger.debug("Server shut down successfully");

apps/rest-api-server/src/syncers.ts

@@ -0,0 +1,30 @@
+import {
+  DailyStatsSyncer,
+  OverallStatsSyncer,
+  createRedisConnection,
+} from "@blobscan/syncers";
+
+import { env } from "./env";
+import { getNetworkDencunForkSlot } from "./utils";
+
+export function setUpSyncers() {
+  const connection = createRedisConnection(env.REDIS_URI);
+
+  const dailyStatsSyncer = new DailyStatsSyncer({
+    cronPattern: env.STATS_SYNCER_DAILY_CRON_PATTERN,
+    redisUriOrConnection: connection,
+  });
+
+  const overallStatsSyncer = new OverallStatsSyncer({
+    cronPattern: env.STATS_SYNCER_OVERALL_CRON_PATTERN,
+    redisUriOrConnection: connection,
+    lowestSlot:
+      env.DENCUN_FORK_SLOT ?? getNetworkDencunForkSlot(env.NETWORK_NAME),
+  });
+
+  Promise.all([dailyStatsSyncer.start(), overallStatsSyncer.start()]);
+
+  return () => {
+    return dailyStatsSyncer.close().finally(() => overallStatsSyncer.close());
+  };
+}