Alert Monitoring Pipeline

def utc(text):
    # utcTime has no "Z"; it is UTC.
    return dt.datetime.fromisoformat(text).replace(tzinfo=dt.timezone.utc)


def iso_z(moment):
    return moment.strftime("%Y-%m-%dT%H:%M:%SZ")


def record_key(rec):
    return f'{rec["id"]}|{rec["vehicleId"]}|{rec["utcTime"]}|{rec["lat"]}|{rec["lng"]}'


def load_state():
    try:
        with open(STATE_FILE) as f:
            return json.load(f)
    except FileNotFoundError:
        return {"last_completed_start": None, "seen": {}}


def save_state(state):
    # Write to a temporary file, then rename it over the old one, so a crash never leaves a half-written state file.
    folder = os.path.dirname(os.path.abspath(STATE_FILE))
    with tempfile.NamedTemporaryFile("w", dir=folder, delete=False) as f:
        json.dump(state, f)
    os.replace(f.name, STATE_FILE)


def poll_once(state):
    started = dt.datetime.now(dt.timezone.utc)
    last = state["last_completed_start"]
    window_from = (utc(last) if last else started) - OVERLAP
    records = get_all_pages(f"/v2/api/companies/{COMPANY_ID}/alert-records",
                            {"from": iso_z(window_from), "num": NUM})

    per_type = {}
    for rec in records:
        per_type[rec["alertType"]] = per_type.get(rec["alertType"], 0) + 1
    for alert_type, count in per_type.items():
        if count >= NUM:
            print(f"Warning: alert type {alert_type} returned {NUM} records; older ones may have been cut. Raise NUM.")

    for rec in sorted(records, key=lambda r: r["utcTime"]):
        key = record_key(rec)
        if key in state["seen"]:
            continue
        name = TYPE_NAMES.get(rec["alertType"], f"type {rec['alertType']}")
        try:
            process_alert(rec, name)
        except Exception as e:                    # not saved, so it is retried on the next poll
            print(f"Handler failed for {key}: {e}")
            continue
        state["seen"][key] = rec["utcTime"]

    # The next poll starts from this poll's start − OVERLAP. Keep keys newer than that (minus 1 s, because
    # utcTime is cut to whole seconds while the server compares the full time).
    next_from = max(started, utc(last) if last else started) - OVERLAP
    state["seen"] = {k: t for k, t in state["seen"].items() if utc(t) >= next_from - dt.timedelta(seconds=1)}
    state["last_completed_start"] = max(started, utc(last)).isoformat() if last else started.isoformat()
    save_state(state)

// Save as alert_poller.mjs and run with Node 18+ (an ES module: it uses import and top-level await).
import { readFile, writeFile, rename } from "node:fs/promises";

const API_BASE = process.env.TELEMAX_API_BASE ?? "https://api.telemax.com.au";
const API_KEY = process.env.TELEMAX_API_KEY;
const COMPANY_ID = Number(process.env.TELEMAX_COMPANY_ID);              // your level-0 company
const OVERLAP_MS = Number(process.env.ALERT_OVERLAP_SECONDS ?? 7200) * 1000;
const NUM = Number(process.env.ALERT_NUM ?? 500);
const PAGE_SIZE = Number(process.env.ALERT_PAGE_SIZE ?? 100);
const POLL_MS = Number(process.env.ALERT_POLL_SECONDS ?? 60) * 1000;
const STATE_FILE = process.env.ALERT_STATE_FILE ?? "alert-poller-state.json";
const MAX_POLLS = Number(process.env.ALERT_MAX_POLLS ?? 0);           // 0 = run until stopped

let token = null, tokenExpiresAt = 0;

async function getToken() {
  if (token && Date.now() < tokenExpiresAt - 300_000) return token;
  const res = await fetch(`${API_BASE}/v2/api/authentication/token/api-key`, {
    method: "POST",
    headers: { "Content-Type": "application/x-www-form-urlencoded" },
    body: new URLSearchParams({ apiKey: API_KEY }),
  });
  if (!res.ok) throw new Error(`Token request failed: HTTP ${res.status}`);
  const body = await res.json();
  token = body.access_token;
  tokenExpiresAt = Date.now() + body.expires_in * 1000;
  return token;
}

class BudgetExhaustedError extends Error {}

// One API call. Waits out a short 429; stops when an hourly or daily budget is exhausted.
async function request(path, params = {}, headers = {}) {
  for (;;) {
    const url = `${API_BASE}${path}?${new URLSearchParams(params)}`;
    const res = await fetch(url, { headers: { ...headers, Authorization: `Bearer ${await getToken()}` } });
    if (res.status === 429) {
      const wait = Number(res.headers.get("Retry-After") ?? 1);
      if (wait >= 3600) throw new BudgetExhaustedError(`Rate-limit budget exhausted (Retry-After: ${wait} s); try again later`);
      await new Promise((r) => setTimeout(r, wait * 1000));
      continue;
    }
    if (!res.ok) throw new Error(`HTTP ${res.status} for ${path}`);
    return res;
  }
}

// Every page; follow-up pages reuse the pagination session, so they do not count against the rate limit.
async function getAllPages(path, params) {
  const items = [];
  let page = 1, sessionToken = null;
  for (;;) {
    const res = await request(path, { ...params, page, pageSize: PAGE_SIZE },
                              sessionToken ? { "X-Session-Token": sessionToken } : {});
    const data = await res.json();
    items.push(...(data.items ?? []));
    sessionToken = res.headers.get("X-Session-Token") ?? sessionToken;
    if (page >= (data.numberOfPages ?? 1)) return items;
    page++;
  }
}

const utc = (text) => new Date(text + "Z");                            // utcTime has no "Z"; it is UTC
const isoZ = (ms) => new Date(ms).toISOString().replace(/\.\d{3}Z$/, "Z");
const recordKey = (r) => `${r.id}|${r.vehicleId}|${r.utcTime}|${r.lat}|${r.lng}`;

async function loadState() {
  try { return JSON.parse(await readFile(STATE_FILE, "utf8")); }
  catch { return { lastCompletedStart: null, seen: {} }; }
}

async function saveState(state) {
  await writeFile(STATE_FILE + ".tmp", JSON.stringify(state));        // write, then rename over the old file
  await rename(STATE_FILE + ".tmp", STATE_FILE);
}

// Names for the alertType values alert-records returns (3–12). Route by the record's own alertType.
const TYPE_NAMES = { 3: "Engine Fault", 4: "Fatigue", 5: "Geofence", 6: "Geofence Speed", 7: "Idle", 8: "Ignition",
                     9: "Low Battery", 10: "Exceed Maximum Speed", 11: "Mileage", 12: "Above Speed Limit" };

async function pollOnce(state) {
  const started = Date.now();
  const last = state.lastCompletedStart ? Date.parse(state.lastCompletedStart) : null;
  const windowFrom = (last ?? started) - OVERLAP_MS;
  const records = await getAllPages(`/v2/api/companies/${COMPANY_ID}/alert-records`,
                                    { from: isoZ(windowFrom), num: NUM });

  const perType = new Map();
  for (const r of records) perType.set(r.alertType, (perType.get(r.alertType) ?? 0) + 1);
  for (const [type, count] of perType)
    if (count >= NUM) console.warn(`Alert type ${type} returned ${NUM} records; older ones may have been cut. Raise NUM.`);

  for (const r of [...records].sort((a, b) => a.utcTime.localeCompare(b.utcTime))) {
    const key = recordKey(r);
    if (key in state.seen) continue;
    try {
      await handleAlert(r, TYPE_NAMES[r.alertType] ?? `type ${r.alertType}`);
    } catch (e) {                                                       // not saved, so retried next poll
      console.error(`Handler failed for ${key}: ${e.message}`);
      continue;
    }
    state.seen[key] = r.utcTime;
  }

  const nextFrom = Math.max(started, last ?? started) - OVERLAP_MS;
  for (const [key, t] of Object.entries(state.seen))
    if (utc(t).getTime() < nextFrom - 1000) delete state.seen[key];
  state.lastCompletedStart = new Date(Math.max(started, last ?? started)).toISOString();
  await saveState(state);
}

async function handleAlert(r, typeName) {               // replace with your own (async) handling
  console.log(`[${typeName}] ${r.vehicleName} at ${r.address} (${r.utcTime} UTC)`);
}

// A self-scheduling loop: the next poll starts only after this one has finished.
const state = await loadState();
for (let polls = 1; ; polls++) {
  try {
    await pollOnce(state);
  } catch (e) {
    if (e instanceof BudgetExhaustedError) throw e;                   // stop: retrying cannot succeed before the window resets
    console.error(`Poll failed, will retry: ${e.message}`);
  }
  if (MAX_POLLS && polls >= MAX_POLLS) break;
  await new Promise((r) => setTimeout(r, POLL_MS));
}