MikrotikManager/backend/src/services/alert-engine/source-watermark.ts

import { desc, eq, inArray } from "drizzle-orm"
import { db } from "../../db/index.js"
import { alertEngineCursor, schedulerRuns } from "../../db/schema.js"

const SOURCE_JOBS = [
  "traffic",
  "servers_rest_ping",
  "uptime_resources",
  "uptime_ping",
  "uptime_speed",
  "gre_bgp",
] as const

export async function getLatestSourceFinishedAt(): Promise<string | null> {
  const row = (await db
    .select({ finishedAt: schedulerRuns.finishedAt })
    .from(schedulerRuns)
    .where(inArray(schedulerRuns.jobKey, [...SOURCE_JOBS]))
    .orderBy(desc(schedulerRuns.finishedAt))
    .limit(1))[0]
  return row?.finishedAt ?? null
}

export async function getSourceWatermark(): Promise<string | null> {
  const row = (await db.select().from(alertEngineCursor).where(eq(alertEngineCursor.id, 1)).limit(1))[0]
  return row?.lastSourceFinishedAt ?? null
}

export async function updateSourceWatermark(lastSourceFinishedAt: string | null): Promise<void> {
  const existing = (await db
    .select()
    .from(alertEngineCursor)
    .where(eq(alertEngineCursor.id, 1))
    .limit(1))[0]
  if (existing) {
    await db.update(alertEngineCursor)
      .set({
        lastSourceFinishedAt,
        updatedAt: new Date().toISOString(),
      })
      .where(eq(alertEngineCursor.id, 1))
    return
  }
  await db.insert(alertEngineCursor)
    .values({
      id: 1,
      lastSourceFinishedAt,
      updatedAt: new Date().toISOString(),
    })
}