feat(worker): add scheduled job to refresh zalo oauth tokens daily (#552)

Co-authored-by: Deathgiver <anonymous@users.noreply.github.com>
Co-authored-by: Claude Sonnet 4.6 <noreply@anthropic.com>

Author: 3 people

apps/worker/src/schedule/handlers/refresh-zalo-tokens.ts

@@ -0,0 +1,40 @@
+import { zaloIntegrationService } from "@chatbotx.io/business"
+import {
+  calculateExpiresAt,
+  refreshAccessToken,
+  type ZaloAuthValue,
+} from "@chatbotx.io/integration-zalo"
+import { logger } from "../../lib/logger"
+
+export async function refreshZaloTokens(): Promise<void> {
+  const integrations = await zaloIntegrationService.findAll()
+  logger.info(`[refreshZaloTokens] found=${integrations.length}`)
+
+  for (const integration of integrations) {
+    try {
+      const auth = integration.auth as ZaloAuthValue
+      if (!auth.tokens.refreshToken) {
+        logger.warn(
+          `[refreshZaloTokens] id=${integration.id} skipped: no refreshToken`,
+        )
+        continue
+      }
+
+      const newTokens = await refreshAccessToken(auth, auth.tokens.refreshToken)
+
+      await zaloIntegrationService.updateAuth(integration.id, {
+        ...auth,
+        tokens: {
+          ...auth.tokens,
+          accessToken: newTokens.access_token,
+          refreshToken: newTokens.refresh_token,
+          expiresAt: calculateExpiresAt(newTokens.expires_in),
+        },
+      })
+
+      logger.info(`[refreshZaloTokens] id=${integration.id} refreshed`)
+    } catch (error) {
+      logger.error(error, `[refreshZaloTokens] id=${integration.id} failed`)
+    }
+  }
+}

apps/worker/src/schedule/handlers/register-schedules.ts

@@ -116,4 +116,18 @@ export const registerSchedules = async () => {
       },
     },
   )
+
+  await scheduleQueue.upsertJobScheduler(
+    ScheduleJobData.refreshZaloTokens,
+    {
+      pattern: "0 2 * * *",
+    },
+    {
+      name: ScheduleJobData.refreshZaloTokens,
+      data: {
+        type: ScheduleJobData.refreshZaloTokens,
+        data: {},
+      },
+    },
+  )
 }

apps/worker/src/schedule/worker.ts

@@ -18,6 +18,7 @@ import { maintainMacPartitions } from "./handlers/maintain-mac-partitions"
 import { prepareBroadcast } from "./handlers/prepare-broadcast"
 import { processBroadcastContacts } from "./handlers/process-broadcast-contacts"
 import { purgeCoexistStaging } from "./handlers/purge-coexist-staging"
+import { refreshZaloTokens } from "./handlers/refresh-zalo-tokens"
 import { registerSchedules } from "./handlers/register-schedules"
 import { scanCoexistRuns } from "./handlers/scan-coexist-runs"
 import { scanSmartDelay } from "./handlers/scan-smart-delay"
@@ -90,6 +91,10 @@ async function startScheduleWorker() {
           await purgeCoexistStaging()
           return
 
+        case ScheduleJobData.refreshZaloTokens:
+          await refreshZaloTokens()
+          return
+
         default:
           logger.warn("Unknown schedule job type")
       }

packages/business/src/integration-zalo/index.ts

@@ -1 +1,2 @@
 export * from "./schema"
+export * from "./service"

packages/business/src/integration-zalo/service.ts

@@ -0,0 +1,22 @@
+import { db, eq } from "@chatbotx.io/database/client"
+import { integrationZaloModel } from "@chatbotx.io/database/schema"
+import { BaseService } from "../base.service"
+
+class ZaloIntegrationService extends BaseService {
+  async findAll(): Promise<
+    Array<{ id: string; auth: Record<string, unknown> }>
+  > {
+    return await db
+      .select({ id: integrationZaloModel.id, auth: integrationZaloModel.auth })
+      .from(integrationZaloModel)
+  }
+
+  async updateAuth(id: string, auth: Record<string, unknown>): Promise<void> {
+    await db
+      .update(integrationZaloModel)
+      .set({ auth })
+      .where(eq(integrationZaloModel.id, id))
+  }
+}
+
+export const zaloIntegrationService = new ZaloIntegrationService()

packages/worker-config/src/queues/schedule/index.ts

@@ -19,6 +19,7 @@ export const ScheduleJobData = {
   maintainMacPartitions: "maintainMacPartitions",
   scanCoexistRuns: "scanCoexistRuns",
   purgeCoexistStaging: "purgeCoexistStaging",
+  refreshZaloTokens: "refreshZaloTokens",
 } as const
 
 export type ScheduleJobBroadcast = {
@@ -87,6 +88,11 @@ export type ScheduleJobPurgeCoexistStaging = {
   data: Record<string, never>
 }
 
+export type ScheduleJobRefreshZaloTokens = {
+  type: typeof ScheduleJobData.refreshZaloTokens
+  data: Record<string, never>
+}
+
 export type ScheduleJobData =
   | ScheduleJobBroadcast
   | ScheduleJobEnqueueBroadcast
@@ -100,6 +106,7 @@ export type ScheduleJobData =
   | ScheduleJobMaintainMacPartitions
   | ScheduleJobScanCoexistRuns
   | ScheduleJobPurgeCoexistStaging
+  | ScheduleJobRefreshZaloTokens
 
 export const scheduleQueue =
   process.env.NEXT_PHASE === "phase-production-build"