Add Kafka analytics event ingestion endpoint

netlify.toml

@@ -63,6 +63,18 @@
   to = "/.netlify/functions/reference-context"
   status = 200
 
+[[redirects]]
+  from = "/api/analytics"
+  to = "/.netlify/functions/analytics"
+  status = 200
+
+[[headers]]
+  for = "/api/analytics"
+  [headers.values]
+    Access-Control-Allow-Origin = "*"
+    Access-Control-Allow-Methods = "GET, POST, OPTIONS"
+    Access-Control-Allow-Headers = "Content-Type, Authorization"
+
 # Swagger UI
 [[redirects]]
   from = "/docs"

netlify/functions/analytics.js

@@ -0,0 +1,219 @@
+/**
+ * POST /api/analytics  — ingest one or more events
+ * GET  /api/analytics  — query events from blob store (blob backend only)
+ *
+ * Backend is selected via ANALYTICS_BACKEND env var:
+ *   "blob"  — Netlify Blobs (default, zero extra infra)
+ *   "kafka" — Upstash Kafka REST API
+ *   "both"  — write to both simultaneously
+ *
+ * Event shape:
+ *   { event: "page_view", properties: { ... }, ts?: number }
+ *
+ * Kafka env vars (required when backend is "kafka" or "both"):
+ *   KAFKA_REST_URL       — broker REST endpoint (Upstash, Confluent, Redpanda, etc.)
+ *   KAFKA_REST_USERNAME
+ *   KAFKA_REST_PASSWORD
+ *   KAFKA_TOPIC  (default: "analytics")
+ *
+ * Blob env vars:
+ *   NETLIFY_AUTH_TOKEN  (already used by this service)
+ *   ANALYTICS_BLOB_STORE  (default: "analytics")
+ *
+ * GET query params (blob backend):
+ *   ?event=page_view     filter by event name
+ *   ?since=<unix_ms>     only events after this timestamp
+ *   ?limit=100           max results (default 100, max 1000)
+ *   ?date=YYYY-MM-DD     restrict to a specific date partition
+ */
+
+const { json, error, options, putBlob, listBlobs, getBlob } = require("./helpers");
+
+const BACKEND = process.env.ANALYTICS_BACKEND || "blob";
+const KAFKA_TOPIC = process.env.KAFKA_TOPIC || "analytics";
+const BLOB_STORE = process.env.ANALYTICS_BLOB_STORE || "analytics";
+
+// ─── Kafka backend ───────────────────────────────────────────────────────────
+
+async function publishToKafka(messages) {
+  const url = process.env.KAFKA_REST_URL;
+  const user = process.env.KAFKA_REST_USERNAME;
+  const pass = process.env.KAFKA_REST_PASSWORD;
+
+  if (!url || !user || !pass) {
+    throw new Error("Kafka env vars not configured (KAFKA_REST_URL, _USERNAME, _PASSWORD)");
+  }
+
+  const resp = await fetch(`${url}/produce/${KAFKA_TOPIC}`, {
+    method: "POST",
+    headers: {
+      "Content-Type": "application/json",
+      Authorization: `Basic ${Buffer.from(`${user}:${pass}`).toString("base64")}`,
+    },
+    body: JSON.stringify(messages),
+  });
+
+  if (!resp.ok) {
+    const text = await resp.text();
+    throw new Error(`Kafka produce failed (${resp.status}): ${text}`);
+  }
+
+  return resp.json();
+}
+
+// ─── Blob backend ────────────────────────────────────────────────────────────
+
+// Event names form the first key segment, so "/" would corrupt the layout
+// (and make events unqueryable). Restrict to a safe, flat charset.
+const EVENT_NAME_RE = /^[A-Za-z0-9._-]{1,128}$/;
+
+// Key layout: {event_name}/{YYYY-MM-DD}/{unix_ms}-{uniq}
+// The uniq suffix guarantees distinct keys for same-named events written in
+// the same millisecond (batch ingest, or concurrent requests), preventing
+// silent overwrites. unix_ms stays the sort key prefix.
+function blobKey(eventName, ts, uniq) {
+  const date = new Date(ts).toISOString().slice(0, 10);
+  return `${eventName}/${date}/${ts}-${uniq}`;
+}
+
+function uniqSuffix() {
+  return `${process.hrtime.bigint().toString(36)}${Math.random().toString(36).slice(2, 8)}`;
+}
+
+async function writeToBlobs(enrichedEvents) {
+  await Promise.all(
+    enrichedEvents.map((e) =>
+      putBlob(BLOB_STORE, blobKey(e.event, e.received_at, uniqSuffix()), e)
+    )
+  );
+}
+
+// Key timestamp lives in the third segment as "{unix_ms}-{uniq}".
+function keyTs(key) {
+  return parseInt(key.split("/")[2], 10); // parseInt stops at "-", ignores uniq
+}
+
+async function queryBlobs({ eventFilter, since, limit, date }) {
+  const allBlobs = await listBlobs(BLOB_STORE);
+
+  let filtered = allBlobs.filter((b) => {
+    const [name, blobDate] = b.key.split("/");
+    if (eventFilter && name !== eventFilter) return false;
+    if (date && blobDate !== date) return false;
+    if (since && keyTs(b.key) <= since) return false;
+    return true;
+  });
+
+  // Sort ascending by timestamp (embedded in key)
+  filtered.sort((a, b) => keyTs(a.key) - keyTs(b.key));
+
+  filtered = filtered.slice(0, limit);
+
+  const events = await Promise.all(filtered.map((b) => getBlob(BLOB_STORE, b.key)));
+
+  return { count: events.length, events };
+}
+
+// ─── Alarm / alerting stubs ──────────────────────────────────────────────────
+
+// TODO: poll GET /api/analytics, check event rates, fire to Slack/PagerDuty
+async function pollAndAlert() {}
+
+// TODO: ship to Datadog via Netlify Log Drain (configure in Netlify UI)
+async function datadogLogDrain() {}
+
+// TODO: PUT metric to CloudWatch via aws4-signed fetch, then define Alarm in CDK/console
+async function putCloudWatchMetric() {}
+
+// TODO: consume Kafka topic via Upstash trigger → Lambda → CloudWatch PutMetricData
+async function kafkaToCloudWatch() {}
+
+// ─── Handler ─────────────────────────────────────────────────────────────────
+
+exports.handler = async (event) => {
+  if (event.httpMethod === "OPTIONS") return options();
+
+  if (event.httpMethod === "GET") {
+    if (BACKEND === "kafka") {
+      return error("GET is only available with the blob backend", 400);
+    }
+    const p = event.queryStringParameters || {};
+    try {
+      const result = await queryBlobs({
+        eventFilter: p.event || null,
+        since: p.since ? parseInt(p.since, 10) : null,
+        date: p.date || null,
+        limit: Math.min(parseInt(p.limit, 10) || 100, 1000),
+      });
+      return json({ ok: true, backend: "blob", ...result });
+    } catch (err) {
+      return error(`Query failed: ${err.message}`, 502);
+    }
+  }
+
+  if (event.httpMethod !== "POST") return error("Method not allowed", 405);
+
+  let body;
+  try {
+    body = JSON.parse(event.body || "{}");
+  } catch {
+    return error("Invalid JSON body", 400);
+  }
+
+  const rawEvents = Array.isArray(body) ? body : [body];
+  if (!rawEvents.length) return error("No events provided", 400);
+
+  for (const e of rawEvents) {
+    if (!e.event || typeof e.event !== "string") {
+      return error('Each event must have a string "event" field', 400);
+    }
+    if (!EVENT_NAME_RE.test(e.event)) {
+      return error(
+        `Invalid event name "${e.event}": use 1-128 chars of [A-Za-z0-9._-]`,
+        400
+      );
+    }
+  }
+
+  const now = Date.now();
+  const sourceIp = event.headers?.["x-forwarded-for"]?.split(",")[0]?.trim() || null;
+
+  const enrichedEvents = rawEvents.map((e) => ({
+    ...e,
+    ts: e.ts || now,
+    received_at: now,
+    source_ip: sourceIp,
+  }));
+
+  const useKafka = BACKEND === "kafka" || BACKEND === "both";
+  const useBlob = BACKEND === "blob" || BACKEND === "both";
+
+  const results = {};
+  const errors = [];
+
+  await Promise.all([
+    useKafka &&
+      publishToKafka(
+        enrichedEvents.map((e) => ({ value: JSON.stringify(e), key: e.event }))
+      )
+        .then((r) => { results.kafka = r; })
+        .catch((err) => { errors.push(`kafka: ${err.message}`); }),
+
+    useBlob &&
+      writeToBlobs(enrichedEvents)
+        .then(() => { results.blob = { written: enrichedEvents.length }; })
+        .catch((err) => { errors.push(`blob: ${err.message}`); }),
+  ].filter(Boolean));
+
+  if (errors.length && Object.keys(results).length === 0) {
+    return error(`All backends failed: ${errors.join("; ")}`, 502);
+  }
+
+  return json({
+    ok: true,
+    published: enrichedEvents.length,
+    backend: BACKEND,
+    results,
+    ...(errors.length ? { warnings: errors } : {}),
+  });
+};