Setup scheduler with trigger.dev

.github/workflows/trigger-deploy.yml

@@ -0,0 +1,25 @@
+name: Deploy to Trigger.dev (Production)
+
+on:
+  push:
+    tags:
+      - "v[0-9]+.[0-9]+.[0-9]+"
+      - "v[0-9]+.[0-9]+.[0-9]+-beta.[0-9]+"
+
+jobs:
+  deploy:
+    runs-on: ubuntu-latest
+    steps:
+      - uses: actions/checkout@v4
+      - name: Use Node.js 20.x
+        uses: actions/setup-node@v4
+        with:
+          node-version: "20.x"
+          cache: "pnpm"
+      - name: Install dependencies
+        run: pnpm install
+      - name: 🚀 Deploy Trigger.dev
+        env:
+          TRIGGER_ACCESS_TOKEN: ${{ secrets.TRIGGER_ACCESS_TOKEN }}
+        run: |
+          pnpm trigger:deploy

.gitignore

@@ -2,3 +2,5 @@
 # sst
 .sst
 sst-env.d.ts
+
+.trigger

package.json

@@ -6,12 +6,15 @@
     "dev": "sst dev --stage dev",
     "test": "vitest",
     "test:coverage": "vitest run --coverage",
-    "migrate": "drizzle-kit migrate"
+    "migrate": "drizzle-kit migrate",
+    "trigger:dev": "pnpm dlx trigger.dev dev",
+    "trigger:deploy": "pnpm dlx trigger.dev deploy"
   },
   "author": "Ru Chern Chong <[email protected]>",
   "license": "MIT",
   "dependencies": {
     "@neondatabase/serverless": "^0.10.4",
+    "@trigger.dev/sdk": "^3.3.7",
     "@upstash/redis": "^1.34.3",
     "adm-zip": "^0.5.16",
     "drizzle-orm": "^0.37.0",

src/config/redis.ts

@@ -2,6 +2,6 @@ import { Redis } from "@upstash/redis";
 import { Resource } from "sst";
 
 export const redis = new Redis({
-  url: Resource.UPSTASH_REDIS_REST_URL.value,
-  token: Resource.UPSTASH_REDIS_REST_TOKEN.value,
+  url: process.env.UPSTASH_REDIS_REST_URL,
+  token: process.env.UPSTASH_REDIS_REST_TOKEN,
 });

src/config/schedulers.ts

@@ -0,0 +1,13 @@
+type SchedulerName = "cars" | "coe";
+
+type ScheduleOptions =
+  | string
+  | {
+      pattern: string;
+      timezone?: string;
+    };
+
+export const schedulers: Record<SchedulerName, ScheduleOptions> = {
+  cars: "*/60 0-10 * * 1-5",
+  coe: "*/60 0-10 * * 1-5",
+};

src/db.ts

@@ -2,5 +2,5 @@ import { neon } from "@neondatabase/serverless";
 import { drizzle } from "drizzle-orm/neon-http";
 import { Resource } from "sst";
 
-const sql = neon(Resource.DATABASE_URL.value);
+const sql = neon(process.env.DATABASE_URL);
 export const db = drizzle({ client: sql });

src/lib/updater.ts

@@ -6,6 +6,7 @@ import { createUniqueKey } from "@/utils/createUniqueKey";
 import { downloadFile } from "@/utils/downloadFile";
 import { processCSV } from "@/utils/processCSV";
 import { cacheChecksum, getCachedChecksum } from "@/utils/redisCache";
+import { getTableName } from "drizzle-orm";
 import type { PgTable } from "drizzle-orm/pg-core";
 
 export interface UpdaterConfig<T extends PgTable> {
@@ -15,7 +16,8 @@ export interface UpdaterConfig<T extends PgTable> {
   keyFields: string[];
 }
 
-export interface UpdaterResult {
+export interface UpdaterResult<T extends PgTable> {
+  table: T["_"]["name"];
   recordsProcessed: number;
   message: string;
   timestamp: string;
@@ -28,7 +30,9 @@ export const updater = async <T extends PgTable>({
   zipFileName,
   zipUrl,
   keyFields,
-}: UpdaterConfig<T>): Promise<UpdaterResult> => {
+}: UpdaterConfig<T>): Promise<UpdaterResult<T>> => {
+  const tableName = getTableName(table);
+
   // Download and extract file
   const extractedFileName = await downloadFile(zipUrl);
   const destinationPath = path.join(AWS_LAMBDA_TEMP_DIR, extractedFileName);
@@ -55,6 +59,7 @@ export const updater = async <T extends PgTable>({
       `File have not been changed since the last update. Checksum: ${checksum}`,
     );
     return {
+      table: tableName,
       recordsProcessed: 0,
       message: `File have not been changed since the last update. Checksum: ${checksum}`,
       timestamp: new Date().toISOString(),
@@ -86,7 +91,7 @@ export const updater = async <T extends PgTable>({
   // Return early when there are no new records to be added to the database
   if (newRecords.length === 0) {
     return {
-      // table: table.$name,
+      table: tableName,
       recordsProcessed: 0,
       message:
         "No new data to insert. The provided data matches the existing records.",
@@ -111,7 +116,7 @@ export const updater = async <T extends PgTable>({
   await cacheChecksum(extractedFileName, checksum);
 
   return {
-    // table: table.$name,
+    table: tableName,
     recordsProcessed: totalInserted,
     message: `${totalInserted} record(s) inserted in ${Math.round(end - start)}ms`,
     timestamp: new Date().toISOString(),

src/trigger/update-cars.ts

@@ -0,0 +1,26 @@
+import { schedulers } from "@/config/schedulers";
+import { updateCars } from "@/lib/updateCars";
+import { logger, schedules } from "@trigger.dev/sdk/v3";
+
+export const updateCarsTask = schedules.task({
+  id: "update-cars",
+  cron: schedulers.cars,
+  run: async (payload: any, { ctx }) => {
+    try {
+      logger.log("Starting Cars Update Task", { payload, ctx });
+
+      const response = await updateCars();
+
+      logger.log("Cars Update Completed", {
+        recordsProcessed: response.recordsProcessed,
+        message: response.message,
+        timestamp: response.timestamp,
+      });
+
+      return response;
+    } catch (error) {
+      logger.error("Cars Update Task Failed", { error });
+      throw error;
+    }
+  },
+});

src/trigger/update-coe.ts

@@ -0,0 +1,26 @@
+import { schedulers } from "@/config/schedulers";
+import { updateCOE } from "@/lib/updateCOE";
+import { logger, schedules } from "@trigger.dev/sdk/v3";
+
+export const updateCOETask = schedules.task({
+  id: "update-coe",
+  cron: schedulers.coe,
+  run: async (payload: any, { ctx }) => {
+    try {
+      logger.log("Starting COE Update Task", { payload, ctx });
+
+      const response = await updateCOE();
+
+      logger.log("COE Update Completed", {
+        recordsProcessed: response.recordsProcessed,
+        message: response.message,
+        timestamp: response.timestamp,
+      });
+
+      return response;
+    } catch (error) {
+      logger.error("COE Update Task Failed", { error });
+      throw error;
+    }
+  },
+});

sst.config.ts

@@ -1,7 +1,5 @@
 /// <reference path="./.sst/platform/config.d.ts" />
 
-import type { CronArgs } from "./.sst/platform/src/components/aws/cron";
-
 export default $config({
   app(input) {
     return {
@@ -16,56 +14,30 @@ export default $config({
     };
   },
   async run() {
-    const DATABASE_URL = new sst.Secret(
-      "DATABASE_URL",
-      process.env.DATABASE_URL,
-    );
-    const UPSTASH_REDIS_REST_URL = new sst.Secret(
-      "UPSTASH_REDIS_REST_URL",
-      process.env.UPSTASH_REDIS_REST_URL,
-    );
-    const UPSTASH_REDIS_REST_TOKEN = new sst.Secret(
-      "UPSTASH_REDIS_REST_TOKEN",
-      process.env.UPSTASH_REDIS_REST_TOKEN,
-    );
-
-    const updateCars = new sst.aws.Function("UpdateCars", {
-      link: [DATABASE_URL, UPSTASH_REDIS_REST_URL, UPSTASH_REDIS_REST_TOKEN],
-      handler: "src/lib/updateCars.handler",
-    });
-
-    const updateCOE = new sst.aws.Function("UpdateCOE", {
-      link: [DATABASE_URL, UPSTASH_REDIS_REST_URL, UPSTASH_REDIS_REST_TOKEN],
-      handler: "src/lib/updateCOE.handler",
-    });
-
-    const CRON_SCHEDULERS: Record<string, CronArgs> = {
-      cars: {
-        schedule: "cron(0/60 0-10 ? * MON-FRI *)",
-        job: updateCars.arn,
-      },
-      coe: {
-        schedule: "cron(0/60 0-10 ? * MON-FRI *)",
-        job: updateCOE.arn,
-      },
-      "coe-1st-bidding": {
-        schedule: "cron(0/10 8-10 ? * 4#1 *)",
-        job: updateCOE.arn,
-      },
-      "coe-2nd-bidding": {
-        schedule: "cron(0/10 8-10 ? * 4#3 *)",
-        job: updateCOE.arn,
-      },
+    const environment = {
+      DATABASE_URL: process.env.DATABASE_URL,
+      UPSTASH_REDIS_REST_URL: process.env.UPSTASH_REDIS_REST_URL,
+      UPSTASH_REDIS_REST_TOKEN: process.env.UPSTASH_REDIS_REST_TOKEN,
     };
 
-    for (const [key, { schedule, job }] of Object.entries(CRON_SCHEDULERS)) {
-      new sst.aws.Cron(key, { job, schedule });
-    }
+    // TODO: Temporary remove Secrets
+    // const DATABASE_URL = new sst.Secret(
+    //   "DATABASE_URL",
+    //   environment.DATABASE_URL,
+    // );
+    // const UPSTASH_REDIS_REST_URL = new sst.Secret(
+    //   "UPSTASH_REDIS_REST_URL",
+    //   environment.UPSTASH_REDIS_REST_URL,
+    // );
+    // const UPSTASH_REDIS_REST_TOKEN = new sst.Secret(
+    //   "UPSTASH_REDIS_REST_TOKEN",
+    //   environment.UPSTASH_REDIS_REST_TOKEN,
+    // );
 
     // TODO: Might create an API in the future
     new sst.aws.Function("Updater", {
-      link: [DATABASE_URL, UPSTASH_REDIS_REST_URL, UPSTASH_REDIS_REST_TOKEN],
       handler: "src/index.handler",
+      environment,
       url: true,
     });
   },

trigger.config.ts

@@ -0,0 +1,19 @@
+import { defineConfig } from "@trigger.dev/sdk/v3";
+
+export default defineConfig({
+  project: "proj_hdsezzyvvwvwzpyzlrat",
+  runtime: "node",
+  logLevel: "log",
+  maxDuration: 60,
+  retries: {
+    enabledInDev: true,
+    default: {
+      maxAttempts: 3,
+      minTimeoutInMs: 1000,
+      maxTimeoutInMs: 10000,
+      factor: 2,
+      randomize: true,
+    },
+  },
+  dirs: ["./src/trigger"],
+});

tsconfig.json

@@ -15,6 +15,12 @@
     },
     "types": ["node"]
   },
-  "include": ["src/**/*", "drizzle.config.ts", "sst.config.ts", "sst-env.d.ts"],
+  "include": [
+    "src/**/*",
+    "drizzle.config.ts",
+    "sst.config.ts",
+    "sst-env.d.ts",
+    "trigger.config.ts"
+  ],
   "exclude": ["node_modules", "dist"]
 }