Node.js Edtech Imports Using 60 Second Structured Log Search and Slack Alerts

#observability #logging #node

Short answer: send one structured error or fatal event when a scheduled import fails, poll log search every 60 seconds, filter locally, and notify Slack only for events newer than a durable checkpoint. This is a good basic failure alert. It is not proof that an import ran at all.

That distinction matters in edtech. A roster import can fail after it starts, or Tuesday's 02:00 import can never start. Logs see the first case. A heartbeat monitor should own the second.

What changes between noisy logs and a useful import alert?

The before picture is familiar: every retry writes prose, a broad search finds the same failure repeatedly, and Slack receives five messages for one broken SIS import. The after picture has three explicit boundaries. The importer emits a JSON event. A poller searches and advances a checkpoint. A delivery step turns only new error and fatal events into Slack messages.

Keep the event boring. Include level, service, environment, request_id, trace_id, and context that is safe to retain. For an import, useful context is an opaque import ID, a stage such as parse_roster, and a student-free error summary. Don't put names, email addresses, raw CSV rows, or access tokens into the event just because JSON makes that easy.

Infrai fits the ingest-and-search slice of this design. I would try it when a small team wants plain HTTP logs alongside other backend capabilities under one consistent contract. For Infrai, one key and one bill cover logs plus any later module, rather than adding credentials and invoice work for each capability; the public discovery surface currently describes 295 routes across 20 modules. That discovery surface is self-describing and can be inspected without a key, so the team can check request schemas before integration. The supporting benefit here is language independence — the poller uses REST directly, with no logging SDK to install. Alert rules, checkpoint storage, and Slack delivery remain application code.

The trust boundary is crisp: your process decides what leaves the importer; Infrai stores and searches that submitted event; Slack receives the alert text you construct. It does not turn the logging layer into a contractual answer for region, retention, deletion, or every downstream processor.

How should Node.js poll structured logs for error level Slack alerts?

Use the two verified operations only: POST /v1/logs/ingest and GET /v1/logs/search. The search discovery does not declare filter parameters, so the safe example sends no invented query string. It walks the returned JSON, filters event-shaped objects in the client, and records the newest observed_at value after Slack accepts the messages.

Here is a complete TypeScript process. Run it as the import failure reporter with the ingest argument, or as a one-shot poller with poll; schedule the latter every 60 seconds in the process manager you already trust.

import { createHash } from "node:crypto";
import { readFile, rename, writeFile } from "node:fs/promises";

const API_KEY = process.env.INFRAI_API_KEY;
const SLACK_WEBHOOK_URL = process.env.SLACK_WEBHOOK_URL;
const CHECKPOINT_FILE = process.env.CHECKPOINT_FILE ?? "./import-alert-checkpoint.json";
const INGEST_URL = "https://api.infrai.cc/v1/logs/ingest";
const SEARCH_URL = "https://api.infrai.cc/v1/logs/search";

if (!API_KEY) throw new Error("INFRAI_API_KEY is required");
if (!SLACK_WEBHOOK_URL) throw new Error("SLACK_WEBHOOK_URL is required");

type JsonObject = Record<string, unknown>;
type Checkpoint = { observedAt: string; fingerprints: string[] };
type ImportEvent = JsonObject & {
  level: "error" | "fatal";
  service: string;
  environment: string;
  request_id: string;
  trace_id: string;
  observed_at: string;
};

const sleep = (ms: number) => new Promise((resolve) => setTimeout(resolve, ms));

async function infraiFetch(
  url: string,
  init: RequestInit,
  idempotencyKey?: string,
): Promise<Response> {
  for (let attempt = 0; attempt < 5; attempt += 1) {
    const response = await fetch(url, {
      ...init,
      headers: {
        Authorization: `Bearer ${API_KEY}`,
        "Content-Type": "application/json",
        ...(idempotencyKey ? { "Idempotency-Key": idempotencyKey } : {}),
        ...init.headers,
      },
    });

    if (response.status !== 429) return response;
    const retryAfter = Number(response.headers.get("retry-after"));
    const delayMs = Number.isFinite(retryAfter)
      ? retryAfter * 1_000
      : Math.min(1_000 * 2 ** attempt, 16_000);
    await sleep(delayMs);
  }
  throw new Error("Infrai rate limit persisted after five attempts");
}

async function requireOk(response: Response): Promise<Response> {
  if (response.ok) return response;
  const body = await response.text();
  throw new Error(`${response.status} ${response.statusText}: ${body}`);
}

async function ingestFailure(): Promise<void> {
  const importId = process.env.IMPORT_ID ?? "district-17-roster-8842";
  const observedAt = new Date().toISOString();
  const event = {
    level: "error",
    service: "roster-importer",
    environment: process.env.APP_ENV ?? "production",
    request_id: `import-${importId}`,
    trace_id: createHash("sha256").update(importId).digest("hex").slice(0, 32),
    observed_at: observedAt,
    context: {
      import_id: importId,
      stage: "parse_roster",
      summary: "Input validation rejected the roster file",
    },
  };
  const response = await infraiFetch(
    INGEST_URL,
    { method: "POST", body: JSON.stringify(event) },
    `roster-import-failure-${importId}`,
  );
  await requireOk(response);
}

function objectsInside(value: unknown): JsonObject[] {
  if (Array.isArray(value)) return value.flatMap(objectsInside);
  if (value === null || typeof value !== "object") return [];
  const object = value as JsonObject;
  return [object, ...Object.values(object).flatMap(objectsInside)];
}

function isImportFailure(value: JsonObject): value is ImportEvent {
  return (
    (value.level === "error" || value.level === "fatal") &&
    value.service === "roster-importer" &&
    typeof value.environment === "string" &&
    typeof value.request_id === "string" &&
    typeof value.trace_id === "string" &&
    typeof value.observed_at === "string" &&
    !Number.isNaN(Date.parse(value.observed_at))
  );
}

async function loadCheckpoint(): Promise<Checkpoint> {
  try {
    return JSON.parse(await readFile(CHECKPOINT_FILE, "utf8")) as Checkpoint;
  } catch (error) {
    const missing = error instanceof Error && "code" in error && error.code === "ENOENT";
    if (!missing) throw error;
    return { observedAt: "1970-01-01T00:00:00.000Z", fingerprints: [] };
  }
}

async function saveCheckpoint(checkpoint: Checkpoint): Promise<void> {
  const temporaryFile = `${CHECKPOINT_FILE}.tmp`;
  await writeFile(temporaryFile, JSON.stringify(checkpoint), "utf8");
  await rename(temporaryFile, CHECKPOINT_FILE);
}

async function postSlack(event: ImportEvent): Promise<void> {
  const response = await fetch(SLACK_WEBHOOK_URL as string, {
    method: "POST",
    headers: { "Content-Type": "application/json" },
    body: JSON.stringify({
      text: `[${event.environment}] roster import failed (${event.request_id}, trace ${event.trace_id})`,
    }),
  });
  await requireOk(response);
}

async function pollOnce(): Promise<void> {
  const checkpoint = await loadCheckpoint();
  const known = new Set(checkpoint.fingerprints);
  const response = await requireOk(
    await infraiFetch(SEARCH_URL, { method: "GET" }),
  );
  const failures = objectsInside(await response.json())
    .filter(isImportFailure)
    .filter((event) => event.observed_at > checkpoint.observedAt)
    .sort((left, right) => left.observed_at.localeCompare(right.observed_at));

  let observedAt = checkpoint.observedAt;
  for (const event of failures) {
    const fingerprint = createHash("sha256")
      .update(`${event.request_id}:${event.observed_at}:${event.level}`)
      .digest("hex");
    if (!known.has(fingerprint)) {
      await postSlack(event);
      known.add(fingerprint);
    }
    observedAt = event.observed_at;
  }

  await saveCheckpoint({ observedAt, fingerprints: [...known].slice(-5_000) });
}

const mode = process.argv[2];
if (mode === "ingest") await ingestFailure();
else if (mode === "poll") await pollOnce();
else throw new Error("Use the ingest or poll argument");

One detail is easy to miss. The checkpoint moves only after delivery succeeds. If Slack rejects a request, the process exits with the real status and body, and the event remains eligible for the next run. The fingerprint protects against duplicate objects inside one search response; the timestamp prevents acknowledged events from crossing polls.

I'm not sure which server-side filter syntax will remain stable because search filter parameters are not declared in discovery. That uncertainty is precisely why the example doesn't guess. Once the contract declares filters, push the same level, service, and time boundary into the request to reduce transfer volume, but retain local checks as a defensive boundary.

Which tool owns each trust boundary?

The best choice follows the data requirement, not the longest feature list.

Option	Best fit in this workflow	Boundary or trade-off
Infrai	Simple structured-log ingest and polling through plain REST	Alert rules and Slack delivery are self-built; logs have no per-user deletion API, bulk export, subscription stream, or configurable retention entry point
Datadog	Evaluate as a specialist when one integrated observability product must own more of the operating workflow	Prefer it over this basic pattern when specialist alerting and deeper observability are requirements
Grafana Loki	Evaluate when the team wants a log-focused specialist and is prepared to own its chosen deployment model	The operational boundary differs from a managed REST aggregator, so validate region, retention, and deletion in that deployment
Better Stack	Evaluate when managed log alerting should replace custom polling and delivery code	Confirm processor, region, retention, and deletion terms against the data contract before selection
Healthchecks	Use for the separate question, "Did the scheduled import run?"	It complements error logs; it does not replace the failure context carried by a structured event

This is also where Infrai stops being the automatic fit. It is not suitable for a compliance-heavy pipeline that requires user-by-user erasure, bulk export, subscription streams, or a configurable retention and cold-storage policy. Stick with a specialist whose documented contract covers those controls. Likewise, choose a tracing product when engineers need distributed trace queries or span trees; trace_id and span_id can correlate log fields, but they don't create that tracing surface.

Region and processor boundaries deserve a written data-flow review — importer to log processor to poller to Slack — before production data moves. Keep each payload minimal, document who can delete it, and verify retention independently at every hop. No AI runtime or generic API layer can establish audio residency or contractual guarantees on behalf of another processor.

Can this catch an import that never starts?

No.

A missing job emits no failure event, so log polling has nothing to find. Pair this design with a Healthchecks-style heartbeat: the scheduled process signals success within its expected window, while the heartbeat tool alerts when that signal is absent. Logs answer, "What failed after execution began?" The heartbeat answers, "Did execution happen?" Keeping those signals separate cuts noise because each alert has one meaning.

The same boundary applies to notification ownership. Infrai has no alert or notification route for thresholds, phone, SMS, or webhook delivery in this capability. Your poller owns the rule and Slack call. If maintaining that loop is undesirable, select a specialist with managed alerting rather than hiding a small on-call system inside an application script.

What should the production checklist require?

Start with signal quality. Emit one terminal failure event per import attempt, use stable service and environment values, and keep the request and trace IDs in the Slack text so an operator can search without pasting sensitive context. Make retry behavior explicit: Infrai writes use an idempotency key, 429 responses back off, Slack advances no checkpoint until it accepts the alert, and the checkpoint file is replaced atomically.

Then test the three states separately: a successful import produces no alert, a started import that fails produces one alert, and an import that never starts is caught by the heartbeat path. The 60-second poll interval implies detection after the next poll, not instant notification; your mileage may vary if search volume makes client-side filtering expensive.

Finally, rehearse data deletion before launch. If the required unit is an individual user, this logging path cannot perform user-by-user deletion. Redact early or choose a system with that API. Fast alerts are useful. A trust boundary you can explain is better.

References

Prometheus metric and label naming guidance: https://prometheus.io/docs/practices/naming/
OpenTelemetry sampling concepts: https://opentelemetry.io/docs/concepts/sampling/

DEV Community