feat: create jobs app

.changeset/funny-pigs-admire.md

@@ -0,0 +1,5 @@
+---
+"@blobscan/rest-api-server": minor
+---
+
+Removed syncers

.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

.github/workflows/docker.yml

@@ -84,6 +84,32 @@ jobs:
             DATABASE_URL=${{ env.DATABASE_URL }}
             DIRECT_URL=${{ env.DIRECT_URL }}
 
+      - name: Extract metadata (tags, labels) for Docker (jobs)
+        id: meta_jobs
+        uses: docker/metadata-action@v5
+        with:
+          images: |
+            blossomlabs/blobscan-jobs
+          tags: |
+            type=semver,pattern={{version}}
+            type=semver,pattern={{major}}.{{minor}}
+            type=semver,pattern={{major}}
+
+      - name: Build and push (jobs)
+        uses: docker/[email protected]
+        with:
+          context: .
+          push: true
+          target: jobs
+          tags: ${{ steps.meta_api.outputs.tags }}
+          labels: ${{ steps.meta_api.outputs.labels }}
+          build-args: |
+            NEXT_PUBLIC_BLOBSCAN_RELEASE=${{ env.NEXT_PUBLIC_BLOBSCAN_RELEASE }}
+            BUILD_TIMESTAMP=${{ env.BUILD_TIMESTAMP }}
+            GIT_COMMIT=${{ env.GIT_COMMIT }}
+            DATABASE_URL=${{ env.DATABASE_URL }}
+            DIRECT_URL=${{ env.DIRECT_URL }}
+
       - name: Extract metadata (tags, labels) for Docker (Web)
         id: meta_web
         uses: docker/metadata-action@v5

.github/workflows/docker_dev.yml

@@ -77,6 +77,21 @@ jobs:
             DATABASE_URL=${{ env.DATABASE_URL }}
             DIRECT_URL=${{ env.DIRECT_URL }}
 
+      - name: Build and push (jobs)
+        uses: docker/[email protected]
+        with:
+          context: .
+          push: true
+          target: jobs
+          tags: |
+            blossomlabs/blobscan-jobs:development
+          build-args: |
+            NEXT_PUBLIC_BLOBSCAN_RELEASE=${{ env.NEXT_PUBLIC_BLOBSCAN_RELEASE }}
+            BUILD_TIMESTAMP=${{ env.BUILD_TIMESTAMP }}
+            GIT_COMMIT=${{ env.GIT_COMMIT }}
+            DATABASE_URL=${{ env.DATABASE_URL }}
+            DIRECT_URL=${{ env.DIRECT_URL }}
+
       - name: Build and push (web)
         uses: docker/[email protected]
         with:

Dockerfile

@@ -29,6 +29,7 @@ WORKDIR /prepare
 
 RUN turbo prune @blobscan/web --docker --out-dir /prepare/web
 RUN turbo prune @blobscan/rest-api-server --docker --out-dir /prepare/api
+RUN turbo prune @blobscan/jobs --docker --out-dir /prepare/jobs
 
 # stage: web-builder
 FROM deps AS web-builder
@@ -116,3 +117,39 @@ ENV PORT=3001
 ADD docker-entrypoint.sh /
 ENTRYPOINT ["/docker-entrypoint.sh"]
 CMD ["api"]
+
+# stage: jobs-builder
+FROM deps AS jobs-builder
+
+WORKDIR /app
+
+ARG DATABASE_URL
+ARG DIRECT_URL
+
+COPY --from=deps /prepare/jobs/json .
+COPY --from=deps /prepare/jobs/pnpm-lock.yaml .
+
+RUN --mount=type=cache,id=pnpm,target=/pnpm/store pnpm fetch -r
+RUN --mount=type=cache,id=pnpm,target=/pnpm/store pnpm install --frozen-lockfile
+
+COPY --from=deps /prepare/jobs/full .
+
+# Copy original which includes pipelines
+COPY --from=deps /prepare/turbo.json .
+RUN --mount=type=cache,id=pnpm,target=/pnpm/store DATABASE_URL=${DATABASE_URL} DIRECT_URL=${DIRECT_URL} pnpm build --filter=@blobscan/jobs
+
+# stage: jobs
+FROM base AS jobs
+
+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"]
+CMD ["jobs"]

apps/docs/package.json

@@ -21,7 +21,7 @@
     "@tailwindcss/typography": "^0.5.7",
     "autoprefixer": "^10.4.14",
     "clsx": "^2.1.0",
-    "fast-glob": "^3.2.12",
+    "fast-glob": "^3.3.2",
     "flexsearch": "^0.7.31",
     "js-yaml": "^4.1.0",
     "next": "^14.2.30",

apps/docs/src/app/docs/background-jobs/page.md

@@ -8,10 +8,10 @@ nextjs:
 
 Blobscan requires [BullMQ](https://bullmq.io/) to run the following tasks in the background:
 
-- `DailyStatsSyncer` - calculates metrics for the blobs, block and transactions charts.
-- `OverallStatsSyncer` - calculates other metrics such as Total Tx Fees Saved, Total Blocks and Total Blobs.
-- `SwarmStampSyncer` - updates TTL for Ethereum Swarm batches.
+- `DailyStatsCronJob` - calculates metrics for the blobs, block and transactions charts.
+- `OverallStatsCronJob` - calculates other metrics such as Total Tx Fees Saved, Total Blocks and Total Blobs.
+- `SwarmStampCronJob` - updates TTL for Ethereum Swarm batches.
 
-For more information, check out the [@blobscan/syncers](https://github.com/Blobscan/blobscan/tree/main/packages/syncers/src/syncers) package.
+For more information, check out the [@blobscan/jobs](https://github.com/Blobscan/blobscan/tree/main/apps/jobs) app.
 
 The BullMQ queue is also used to upload blobs in parallel to different Storages.

apps/docs/src/app/docs/codebase-overview/page.md

@@ -14,6 +14,7 @@ Blobscan.com is comprised of the following major components:
 
 - **Web App**: A [Next.js](https://nextjs.org/) application hosted on [Vercel](https://vercel.com/) that spins up a [tRPC API](https://trpc.io) that communicates with the database via [Prisma](https://www.prisma.io/). It also uses [Tailwind CSS](https://tailwindcss.com/) for styling.
 - **REST API**: An express app that runs the tRPC API with [OpenAPI](https://www.openapis.org/) support. It exposes some of the tRPC API endpoints as REST endpoints for the public and external services such as the indexer.
+- **Jobs**: A Node.js app running the following BullMQ background cron jobs: eth usd price syncing, stats aggregation and swarm stamp syncing.
 - **Indexer**: A Rust service that listens to the Ethereum blockchain looking for blocks and transactions containing blobs and forwards them to the REST API to be indexed.
 
 {% figure  src="/architecture.svg" appendCurrentTheme=true /%}
@@ -33,6 +34,7 @@ Blobscan is composed of the following apps:
 |  [`@blobscan/docs`](https://github.com/Blobscan/blobscan/tree/main/apps/docs)                       | Nextjs app that contains the documentation. |
 |  [`@blobscan/web`](https://github.com/Blobscan/blobscan/tree/main/apps/web)                         | Nextjs app that contains the web app.       |
 |  [`@blobscan/rest-api-server`](https://github.com/Blobscan/blobscan/tree/main/apps/rest-api-server) | Express app that contains the REST API.     |
+|  [`@blobscan/jobs`](https://github.com/Blobscan/blobscan/tree/main/apps/jobs)                       | Node.js app running BullMQ cron jobs.       |
 
 ### CLI
 

packages/syncers/CHANGELOG.md → apps/jobs/CHANGELOG.md

@@ -1,4 +1,4 @@
-# @blobscan/syncers
+# @blobscan/jobs
 
 ## 0.5.0
 

apps/jobs/esbuild.config.mjs

@@ -0,0 +1,19 @@
+// esbuild.config.js
+import * as esbuild from "esbuild";
+import fs from "fs";
+
+const result = await esbuild.build({
+  entryPoints: ["src/index.ts"],
+  outdir: "dist",
+  outbase: "src",
+  platform: "node",
+  target: "node20",
+  format: "cjs",
+  metafile: !!process.env.BUILD_METADATA_ENABLED,
+  bundle: true,
+  external: [".prisma", "prisma", "@prisma/client"],
+});
+
+if (process.env.BUILD_METADATA_ENABLED) {
+  fs.writeFileSync("dist/meta.json", JSON.stringify(result.metafile));
+}

apps/jobs/package.json

@@ -0,0 +1,45 @@
+{
+  "name": "@blobscan/jobs",
+  "version": "0.5.0",
+  "private": true,
+  "scripts": {
+    "build": "node esbuild.config.mjs",
+    "build:metadata": "BUILD_METADATA_ENABLED=true pnpm build",
+    "clean": "git clean -xdf node_modules",
+    "dev": "pnpm with-env tsx src/index.ts",
+    "lint": "eslint .",
+    "lint:fix": "pnpm lint --fix",
+    "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 --",
+    "with-env": "dotenv -e ../../.env --",
+    "type-check": "tsc --noEmit"
+  },
+  "keywords": [],
+  "license": "MIT",
+  "dependencies": {
+    "@blobscan/dates": "workspace:^0.0.1",
+    "@blobscan/dayjs": "workspace:^0.1.0",
+    "@blobscan/db": "workspace:^0.18.0",
+    "@blobscan/logger": "workspace:^0.1.2",
+    "@blobscan/price-feed": "workspace:^0.1.0",
+    "@blobscan/zod": "workspace:^0.1.0",
+    "axios": "^1.7.2",
+    "bullmq": "^4.13.2",
+    "ioredis": "^5.3.2",
+    "viem": "^2.17.4"
+  },
+  "devDependencies": {
+    "concurrently": "^9.2.0",
+    "esbuild": "0.25.5",
+    "fast-glob": "^3.3.2",
+    "tsx": "^4.19.2"
+  },
+  "eslintConfig": {
+    "root": true,
+    "extends": [
+      "@blobscan/eslint-config/base"
+    ]
+  }
+}

packages/syncers/src/BaseSyncer.ts → apps/jobs/src/cron-jobs/BaseCronJob.ts

@@ -1,48 +1,59 @@
 /* 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, SyncerError } from "./errors";
-import { createRedisConnection } from "./utils";
+import { ErrorException } from "../errors";
+import { createRedis } from "../redis";
 
-export interface CommonSyncerConfig {
+export interface CommonCronJobConfig {
   redisUriOrConnection: Redis | string;
   cronPattern: string;
 }
 
-export interface BaseSyncerConfig extends CommonSyncerConfig {
+export interface BaseCronJobConfig extends CommonCronJobConfig {
   name: string;
-  syncerFn: () => Promise<void>;
+  processor: Processor;
+  jobData?: Record<string, unknown>;
 }
 
-export class BaseSyncer {
+export class CronJobError extends ErrorException {
+  constructor(cronJobName: string, message: string, cause?: unknown) {
+    super(`Cron job "${cronJobName}" failed: ${message}`, cause);
+  }
+}
+
+export class BaseCronJob {
   name: string;
   cronPattern: string;
 
-  protected syncerFn: () => 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,
-    syncerFn,
-  }: BaseSyncerConfig) {
-    this.name = `${name}-syncer`;
+    processor: processorFile,
+    jobData,
+  }: BaseCronJobConfig) {
+    this.name = `${name}-cron-job`;
     this.cronPattern = cronPattern;
     this.logger = createModuleLogger(this.name);
+    this.jobData = jobData;
 
     let connection: Redis;
 
     if (typeof redisUriOrConnection === "string") {
-      connection = createRedisConnection(redisUriOrConnection);
+      connection = createRedis(redisUriOrConnection);
     } else {
       connection = redisUriOrConnection;
     }
@@ -51,7 +62,7 @@ export class BaseSyncer {
       connection,
     });
 
-    this.worker = new Worker(this.queue.name, syncerFn, {
+    this.worker = new Worker(this.queue.name, processorFile, {
       connection,
     });
 
@@ -64,25 +75,24 @@ export class BaseSyncer {
     });
 
     this.connection = connection;
-    this.syncerFn = syncerFn;
   }
 
   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,
         },
       });
 
-      this.logger.info("Syncer started successfully");
+      this.logger.debug("Cron job started successfully");
 
       return repeatableJob;
     } catch (err) {
-      throw new SyncerError(
+      throw new CronJobError(
         this.name,
-        "An error ocurred when starting syncer",
+        "An error ocurred when starting cron job",
         err
       );
     }
@@ -107,15 +117,15 @@ export class BaseSyncer {
           this.queue?.removeAllListeners().close()
         );
 
-        this.logger.info("Syncer closed successfully");
+        this.logger.info("Cron job closed successfully");
       });
   }
 
   async #performClosingOperation(operation: () => Promise<void> | undefined) {
     try {
       await operation();
     } catch (err) {
-      const err_ = new SyncerError(
+      const err_ = new CronJobError(
         this.name,
         "An error ocurred when performing closing operation",
         err