fix production issues

apps/api/src/workers/index.ts

@@ -0,0 +1,57 @@
+import http from 'node:http'
+import { assertStorageConfiguration } from '../lib/storage'
+import { prisma } from '../lib/prisma'
+import { redis } from '../lib/redis'
+import { renderPrometheusText, setGauge } from '../lib/opsMetrics'
+import { startOutboxWorker, startScheduledJobs } from './jobs'
+
+assertStorageConfiguration()
+
+const workerId = process.env.WORKER_ID ?? String(process.pid)
+const metricsPort = Number(process.env.WORKER_METRICS_PORT ?? 0)
+
+console.log(`[worker] starting jobs worker id=${workerId}`)
+
+startOutboxWorker()
+startScheduledJobs()
+
+if (Number.isFinite(metricsPort) && metricsPort > 0) {
+  const server = http.createServer(async (req, res) => {
+    if (req.url === '/health') {
+      res.writeHead(200, { 'Content-Type': 'application/json' })
+      res.end(JSON.stringify({ status: 'ok', role: 'worker', workerId }))
+      return
+    }
+    if (req.url === '/metrics') {
+      try {
+        const pending = await prisma.notificationOutbox.count({ where: { status: 'PENDING' } })
+        setGauge('notification_outbox_pending', pending)
+      } catch {
+        /* leave previous gauge */
+      }
+      const body = renderPrometheusText()
+      res.writeHead(200, { 'Content-Type': 'text/plain; version=0.0.4; charset=utf-8' })
+      res.end(body)
+      return
+    }
+    res.writeHead(404)
+    res.end()
+  })
+  server.listen(metricsPort, () => {
+    console.log(`[worker] metrics listening on :${metricsPort}`)
+  })
+}
+
+async function shutdown(signal: string) {
+  console.log(`[worker] ${signal} received, shutting down`)
+  try {
+    await redis.quit()
+  } catch {
+    redis.disconnect()
+  }
+  await prisma.$disconnect()
+  process.exit(0)
+}
+
+process.on('SIGTERM', () => void shutdown('SIGTERM'))
+process.on('SIGINT', () => void shutdown('SIGINT'))

apps/api/src/workers/jobs.ts

@@ -0,0 +1,230 @@
+import cron from 'node-cron'
+import { prisma } from '../lib/prisma'
+import { redis } from '../lib/redis'
+import { processNotificationOutbox, sendNotification } from '../services/notificationService'
+import { observeOutboxProcessed } from '../lib/opsMetrics'
+import {
+  runTrialExpirationJob,
+  runPeriodEndCancellationJob,
+} from '../modules/subscriptions/subscription.service'
+import { runCollectionsWorker } from '../modules/subscriptions/subscription.collections.service'
+
+const WORKER_ID = process.env.WORKER_ID ?? `jobs:${process.pid}`
+const LEADER_KEY = 'rentaldrivego:jobs:leader'
+const LEADER_TTL_SECONDS = 30
+
+async function withLeaderLock<T>(fn: () => Promise<T>): Promise<T | null> {
+  const acquired = await redis.set(LEADER_KEY, WORKER_ID, 'EX', LEADER_TTL_SECONDS, 'NX')
+  if (acquired !== 'OK') {
+    const current = await redis.get(LEADER_KEY)
+    if (current !== WORKER_ID) return null
+  } else {
+    // refresh TTL periodically while holding
+  }
+  try {
+    return await fn()
+  } finally {
+    const current = await redis.get(LEADER_KEY)
+    if (current === WORKER_ID) await redis.del(LEADER_KEY)
+  }
+}
+
+async function renewLeaderIfOwned() {
+  const current = await redis.get(LEADER_KEY)
+  if (current === WORKER_ID) await redis.expire(LEADER_KEY, LEADER_TTL_SECONDS)
+}
+
+export async function runLicenseExpiryJob() {
+  const customers = await prisma.customer.findMany({ where: { licenseExpiry: { not: null } } })
+  for (const c of customers) {
+    if (!c.licenseExpiry) continue
+    const daysLeft = Math.ceil((c.licenseExpiry.getTime() - Date.now()) / (1000 * 60 * 60 * 24))
+    const expired = c.licenseExpiry <= new Date()
+    const expiring = !expired && daysLeft < 90
+    if (expired !== c.licenseExpired || expiring !== c.licenseExpiringSoon) {
+      await prisma.customer.update({
+        where: { id: c.id },
+        data: {
+          licenseExpired: expired,
+          licenseExpiringSoon: expiring,
+          licenseValidationStatus: expired ? 'EXPIRED' : expiring ? 'EXPIRING' : 'VALID',
+        },
+      })
+    }
+  }
+}
+
+export async function runTrialEndingRemindersJob() {
+  const soon = new Date(Date.now() + 3 * 24 * 60 * 60 * 1000)
+  const subscriptions = await prisma.subscription.findMany({
+    where: { status: 'TRIALING', trialEndAt: { lte: soon, gte: new Date() } },
+    include: { company: { include: { employees: { where: { role: 'OWNER' } } } } },
+  })
+  for (const sub of subscriptions) {
+    const owner = sub.company.employees[0]
+    if (!owner) continue
+    await sendNotification({
+      type: 'SUBSCRIPTION_TRIAL_ENDING',
+      companyId: sub.companyId,
+      employeeId: owner.id,
+      channels: ['IN_APP'],
+      templateKey: 'subscription.trial_ending',
+      templateVariables: {
+        trialEndDate: sub.trialEndAt ?? new Date(Date.now() + 3 * 24 * 60 * 60 * 1000),
+      },
+    }).catch((err) => {
+      console.error('[Notifications] Failed to create trial ending reminder:', err?.message ?? String(err))
+    })
+  }
+}
+
+export async function runMaintenanceRemindersJob() {
+  const now = new Date()
+  const in30Days = new Date(now.getTime() + 30 * 24 * 60 * 60 * 1000)
+
+  const allCandidates = await prisma.maintenanceLog.findMany({
+    where: {
+      OR: [
+        { nextDueAt: { lte: in30Days } },
+        { nextDueMileage: { not: null } },
+      ],
+    },
+    include: {
+      vehicle: {
+        include: {
+          company: {
+            include: {
+              employees: { where: { role: { in: ['OWNER', 'MANAGER'] }, isActive: true }, take: 1 },
+            },
+          },
+        },
+      },
+    },
+    orderBy: { performedAt: 'desc' },
+  })
+
+  const latestByKey = new Map<string, (typeof allCandidates)[number]>()
+  for (const log of allCandidates) {
+    const key = `${log.vehicleId}:${log.type}`
+    if (!latestByKey.has(key)) latestByKey.set(key, log)
+  }
+
+  for (const log of latestByKey.values()) {
+    const vehicle = log.vehicle
+    const company = vehicle.company
+    const recipient = company.employees[0]
+    if (!recipient) continue
+
+    let isOverdueByDate = false
+    let daysLeft: number | null = null
+    let dueSoonByDate = false
+    if (log.nextDueAt) {
+      daysLeft = Math.ceil((log.nextDueAt.getTime() - now.getTime()) / (1000 * 60 * 60 * 24))
+      isOverdueByDate = log.nextDueAt <= now
+      dueSoonByDate = !isOverdueByDate && daysLeft <= 30
+    }
+
+    let isOverdueByOdometer = false
+    let kmLeft: number | null = null
+    let dueSoonByOdometer = false
+    if (log.nextDueMileage != null && vehicle.mileage != null) {
+      kmLeft = log.nextDueMileage - vehicle.mileage
+      isOverdueByOdometer = kmLeft <= 0
+      dueSoonByOdometer = !isOverdueByOdometer && kmLeft <= 500
+    }
+
+    const isOverdue = isOverdueByDate || isOverdueByOdometer
+    const isDueSoon = !isOverdue && (dueSoonByDate || dueSoonByOdometer)
+    if (!isOverdue && !isDueSoon) continue
+
+    const dueParts: string[] = []
+    if (isOverdueByDate) dueParts.push(`overdue since ${log.nextDueAt!.toLocaleDateString()}`)
+    else if (dueSoonByDate && daysLeft != null) dueParts.push(`due in ${daysLeft} day${daysLeft === 1 ? '' : 's'}`)
+    if (isOverdueByOdometer) dueParts.push(`overdue by odometer (${Math.abs(kmLeft!).toLocaleString()} km ago)`)
+    else if (dueSoonByOdometer && kmLeft != null) dueParts.push(`${kmLeft.toLocaleString()} km remaining`)
+
+    const title = isOverdue
+      ? `Overdue: ${log.type} — ${vehicle.make} ${vehicle.model}`
+      : `${log.type} due soon — ${vehicle.make} ${vehicle.model}`
+    const body = `${log.type} for ${vehicle.make} ${vehicle.model} (${vehicle.licensePlate}): ${dueParts.join('; ')}. Please log the service to dismiss this reminder.`
+
+    const reminderDate = now.toISOString().slice(0, 10)
+    await sendNotification({
+      type: 'VEHICLE_MAINTENANCE_DUE',
+      title,
+      body,
+      data: {
+        vehicleId: vehicle.id,
+        maintenanceLogId: log.id,
+        maintenanceType: log.type,
+        isOverdue,
+        daysLeft,
+        kmLeft,
+        isOverdueByDate,
+        isOverdueByOdometer,
+      },
+      companyId: company.id,
+      employeeId: recipient.id,
+      channels: ['IN_APP'],
+      sourceType: 'maintenance_log',
+      sourceId: log.id,
+      idempotencyKey: `maintenance:${log.id}:${recipient.id}:${reminderDate}`,
+    })
+  }
+}
+
+/** Outbox dispatch does its own DB leasing — safe across workers without Redis leader. */
+export function startOutboxWorker() {
+  cron.schedule('* * * * *', async () => {
+    try {
+      const n = await processNotificationOutbox(50, WORKER_ID)
+      if (n > 0) {
+        observeOutboxProcessed(n)
+        console.log(`[notifications] outbox: ${n} events completed`)
+      }
+    } catch (err: any) {
+      console.error('[notifications] outbox worker failed:', err?.message ?? err)
+    }
+  })
+}
+
+/** Scheduled domain jobs — only one leader executes them. */
+export function startScheduledJobs() {
+  setInterval(() => {
+    void renewLeaderIfOwned()
+  }, 10_000)
+
+  cron.schedule('0 8 * * *', async () => {
+    await withLeaderLock(async () => {
+      await runLicenseExpiryJob()
+      await runMaintenanceRemindersJob()
+    })
+  })
+
+  cron.schedule('0 * * * *', async () => {
+    await withLeaderLock(async () => {
+      const n = await runTrialExpirationJob()
+      if (n > 0) console.log(`[subscription] trial_expiration: ${n} expired`)
+    })
+  })
+
+  cron.schedule('0 1 * * *', async () => {
+    await withLeaderLock(async () => {
+      const nPeriod = await runPeriodEndCancellationJob()
+      if (nPeriod > 0) console.log(`[subscription] period_end_cancel: ${nPeriod} cancelled`)
+    })
+  })
+
+  cron.schedule('*/15 * * * *', async () => {
+    await withLeaderLock(async () => {
+      const n = await runCollectionsWorker()
+      if (n > 0) console.log(`[subscription] collections: ${n} cases processed`)
+    })
+  })
+
+  cron.schedule('0 9 * * *', async () => {
+    await withLeaderLock(async () => {
+      await runTrialEndingRemindersJob()
+    })
+  })
+}