refactor(jobs): expose a single entry point

Author: PJColombo

.env.example

@@ -11,6 +11,7 @@ DIRECT_URL=postgresql://blobscan:s3cr3t@localhost:5432/blobscan_dev?schema=publi
 
 BLOBSCAN_WEB_TAG=next
 BLOBSCAN_API_TAG=next
+BLOBSCAN_JOBS_TAG=next
 INDEXER_TAG=master
 
 ### blobscan website

Dockerfile

@@ -145,9 +145,11 @@ WORKDIR /app
 
 ENV NODE_ENV=production
 
+COPY --from=jobs-builder /app/node_modules/.prisma ./node_modules/.prisma
 COPY --from=jobs-builder /app/node_modules/prisma ./node_modules/prisma
 COPY --from=jobs-builder /app/node_modules/@prisma ./node_modules/@prisma
 COPY --from=jobs-builder /app/apps/jobs/dist ./
 
 ADD docker-entrypoint.sh /
-ENTRYPOINT ["/docker-entrypoint.sh"] 
+ENTRYPOINT ["/docker-entrypoint.sh"]
+CMD ["jobs"]

apps/jobs/esbuild.config.mjs

@@ -1,22 +1,17 @@
 // esbuild.config.js
 import * as esbuild from "esbuild";
-import glob from "fast-glob";
 import fs from "fs";
 
-const entryPoints = await glob("src/**/worker.ts");
-
 const result = await esbuild.build({
-  entryPoints,
+  entryPoints: ["src/index.ts"],
   outdir: "dist",
   outbase: "src",
   platform: "node",
   target: "node20",
   format: "cjs",
   metafile: !!process.env.BUILD_METADATA_ENABLED,
   bundle: true,
-  sourcemap: true,
-  treeShaking: true,
-  external: ["ioredis", "prisma", "@prisma/client"],
+  external: [".prisma", "prisma", "@prisma/client"],
 });
 
 if (process.env.BUILD_METADATA_ENABLED) {

apps/jobs/package.json

@@ -6,16 +6,10 @@
     "build": "node esbuild.config.mjs",
     "build:metadata": "BUILD_METADATA_ENABLED=true pnpm build",
     "clean": "git clean -xdf node_modules",
-    "dev": "concurrently -n eth-price,stats,swarm-stamp -c green,blue,magenta \"pnpm dev:eth-price\" \"pnpm dev:stats\" \"pnpm dev:swarm-stamp\"",
-    "dev:eth-price": "pnpm with-env tsx src/eth-price/worker.ts",
-    "dev:stats": "pnpm with-env tsx src/stats/worker.ts",
-    "dev:swarm-stamp": "pnpm with-env tsx src/swarm-stamp/worker.ts",
+    "dev": "pnpm with-env tsx src/index.ts",
     "lint": "eslint .",
     "lint:fix": "pnpm lint --fix",
-    "start": "concurrently -n eth-price,stats,swarm-stamp -c green,blue,magenta \"pnpm start:eth-price\" \"pnpm start:stats\" \"pnpm start:swarm-stamp\"",
-    "start:eth-price": "pnpm with-env node dist/eth-price/worker.js",
-    "start:stats": "pnpm with-env node dist/stats/worker.js",
-    "start:swarm-stamp": "pnpm with-env node dist/swarm-stamp/worker.js",
+    "start": "pnpm with-env node dist/index.js",
     "test": "pnpm with-env:test vitest",
     "test:ui": "pnpm with-env:test vitest --ui",
     "with-env:test": "dotenv -e ../../.env.test --",

apps/jobs/src/BaseCronJob.ts renamed to apps/jobs/src/cron-jobs/BaseCronJob.ts

@@ -1,12 +1,13 @@
 /* eslint-disable @typescript-eslint/no-misused-promises */
+import type { Processor } from "bullmq";
 import { Queue, Worker } from "bullmq";
 import type { Redis } from "ioredis";
 
 import { createModuleLogger } from "@blobscan/logger";
 import type { Logger } from "@blobscan/logger";
 
-import { ErrorException } from "./errors";
-import { createRedis } from "./redis";
+import { ErrorException } from "../errors";
+import { createRedis } from "../redis";
 
 export interface CommonCronJobConfig {
   redisUriOrConnection: Redis | string;
@@ -15,11 +16,12 @@ export interface CommonCronJobConfig {
 
 export interface BaseCronJobConfig extends CommonCronJobConfig {
   name: string;
-  jobFn: () => Promise<void>;
+  processor: Processor;
+  jobData?: Record<string, unknown>;
 }
 
 export class CronJobError extends ErrorException {
-  constructor(cronJobName: string, message: string, cause: unknown) {
+  constructor(cronJobName: string, message: string, cause?: unknown) {
     super(`Cron job "${cronJobName}" failed: ${message}`, cause);
   }
 }
@@ -28,22 +30,25 @@ export class BaseCronJob {
   name: string;
   cronPattern: string;
 
-  protected jobFn: () => Promise<void>;
   protected logger: Logger;
 
   protected connection: Redis;
-  protected worker: Worker | undefined;
-  protected queue: Queue | undefined;
+  protected worker?: Worker;
+  protected queue?: Queue;
+
+  protected jobData?: Record<string, unknown>;
 
   constructor({
     name,
     cronPattern,
     redisUriOrConnection,
-    jobFn,
+    processor: processorFile,
+    jobData,
   }: BaseCronJobConfig) {
     this.name = `${name}-cron-job`;
     this.cronPattern = cronPattern;
     this.logger = createModuleLogger(this.name);
+    this.jobData = jobData;
 
     let connection: Redis;
 
@@ -57,7 +62,7 @@ export class BaseCronJob {
       connection,
     });
 
-    this.worker = new Worker(this.queue.name, jobFn, {
+    this.worker = new Worker(this.queue.name, processorFile, {
       connection,
     });
 
@@ -70,13 +75,12 @@ export class BaseCronJob {
     });
 
     this.connection = connection;
-    this.jobFn = jobFn;
   }
 
   async start() {
     try {
       const jobName = `${this.name}-job`;
-      const repeatableJob = await this.queue?.add(jobName, null, {
+      const repeatableJob = await this.queue?.add(jobName, this.jobData, {
         repeat: {
           pattern: this.cronPattern,
         },

apps/jobs/src/cron-jobs/daily-stats/DailyStatsCronJob.ts

@@ -0,0 +1,46 @@
+import { formatDate } from "../../utils";
+import { BaseCronJob } from "../BaseCronJob";
+import type { CommonCronJobConfig } from "../BaseCronJob";
+import dailyStats from "./processor";
+import type { DailyStatsJobResult } from "./types";
+
+export type DailyStatsCronJobConfig = CommonCronJobConfig;
+
+export class DailyStatsCronJob extends BaseCronJob {
+  constructor({ redisUriOrConnection, cronPattern }: DailyStatsCronJobConfig) {
+    const name = "daily-stats";
+
+    super({
+      name,
+      redisUriOrConnection,
+      cronPattern,
+      processor: dailyStats,
+    });
+
+    this.worker?.on("completed", (_, result) => {
+      const result_ = result as DailyStatsJobResult;
+
+      if (!result_) {
+        this.logger.info(
+          "Daily stats aggregation skipped: no blocks indexed yet"
+        );
+
+        return;
+      }
+
+      const { fromDate, toDate, totalAggregationsCreated } = result_;
+
+      if (fromDate === toDate) {
+        this.logger.info(`Daily stats aggregation skipped: already up to date`);
+
+        return;
+      }
+
+      this.logger.info(
+        `Daily data up to day ${formatDate(
+          toDate
+        )} aggregated. ${totalAggregationsCreated} daily stats created successfully.`
+      );
+    });
+  }
+}

apps/jobs/src/cron-jobs/daily-stats/processor.ts

@@ -0,0 +1,45 @@
+import { normalizeDate, toDailyDate } from "@blobscan/dayjs";
+import { prisma } from "@blobscan/db";
+
+import type { DailyStatsJobResult } from "./types";
+
+export default async (): Promise<DailyStatsJobResult> => {
+  const lastIndexedBlock = await prisma.block.findLatest();
+
+  if (!lastIndexedBlock) {
+    return;
+  }
+  const targetDate = normalizeDate(lastIndexedBlock.timestamp).subtract(
+    1,
+    "day"
+  );
+  const targetDay = toDailyDate(targetDate);
+
+  const rawLastDailyStatsDay = await prisma.dailyStats.findFirst({
+    select: { day: true },
+    where: { category: null, rollup: null },
+    orderBy: { day: "desc" },
+  });
+  const lastDailyStatsDay = rawLastDailyStatsDay?.day
+    ? normalizeDate(rawLastDailyStatsDay.day)
+    : undefined;
+
+  if (lastDailyStatsDay && lastDailyStatsDay.isSame(targetDay, "day")) {
+    return {
+      fromDate: lastDailyStatsDay.utc().toISOString(),
+      toDate: targetDay.utc().toISOString(),
+      totalAggregationsCreated: 0,
+    };
+  }
+
+  const [blobDailyStats] = await prisma.dailyStats.aggregate({
+    from: lastDailyStatsDay?.add(1, "day"),
+    to: targetDay,
+  });
+
+  return {
+    fromDate: lastDailyStatsDay?.utc().toISOString(),
+    toDate: targetDate.utc().toISOString(),
+    totalAggregationsCreated: blobDailyStats.length,
+  };
+};

apps/jobs/src/cron-jobs/daily-stats/types.ts

@@ -0,0 +1,14 @@
+import type { SandboxedJob } from "bullmq";
+
+export type DailyStatsJobResult =
+  | {
+      fromDate?: string;
+      toDate: string;
+      totalAggregationsCreated: number;
+    }
+  | undefined;
+
+export type DailyStatsSanboxedJob = SandboxedJob<
+  undefined,
+  DailyStatsJobResult
+>;

apps/jobs/src/cron-jobs/eth-price/EthPriceCronJob.ts

@@ -0,0 +1,39 @@
+import type { PriceFeed } from "@blobscan/price-feed";
+
+import type { CommonCronJobConfig } from "../BaseCronJob";
+import { BaseCronJob } from "../BaseCronJob";
+import ethPrice from "./processor";
+
+export interface EthPriceCronJobConfig extends CommonCronJobConfig {
+  ethUsdPriceFeed: PriceFeed;
+}
+
+export class EthPriceCronJob extends BaseCronJob {
+  constructor({ ...restConfig }: CommonCronJobConfig) {
+    super({
+      ...restConfig,
+      name: "eth-price",
+      processor: ethPrice,
+    });
+
+    this.worker?.on("completed", (_, result?) => {
+      const { price, timestamp, roundId } = result as {
+        price?: number;
+        timestamp: string;
+        roundId?: string;
+      };
+
+      if (!price && !roundId) {
+        this.logger.warn(
+          `Skipping eth price update: No price data found for datetime ${timestamp}`
+        );
+
+        return;
+      }
+
+      this.logger.info(
+        `ETH price indexed: $${price} at ${timestamp} recorded (retrieved from round ${roundId})`
+      );
+    });
+  }
+}

apps/jobs/src/cron-jobs/eth-price/price-feed.ts

@@ -0,0 +1,31 @@
+import * as chains from "viem/chains";
+
+import { PriceFeed } from "@blobscan/price-feed";
+
+import { env } from "../../env";
+import { client } from "./viem";
+
+let priceFeed: PriceFeed | undefined;
+
+export async function getEthUsdPriceFeed() {
+  if (!priceFeed) {
+    const chain = Object.values(chains).find(
+      (c) => c.id === env.ETH_PRICE_SYNCER_CHAIN_ID
+    );
+
+    if (!chain) {
+      throw new Error(
+        `Can't initialize ETH price syncer: chain with id ${env.ETH_PRICE_SYNCER_CHAIN_ID} not found`
+      );
+    }
+
+    priceFeed = await PriceFeed.create({
+      client,
+      dataFeedContractAddress:
+        env.ETH_PRICE_SYNCER_ETH_USD_PRICE_FEED_CONTRACT_ADDRESS as `0x${string}`,
+      timeTolerance: env.ETH_PRICE_SYNCER_TIME_TOLERANCE,
+    });
+  }
+
+  return priceFeed;
+}