feat(observability): add otel ingest and span sanitization

- add a landing otlp traces proxy with token-based rate limiting and redaction
- attach provider, thread, and user telemetry attributes to server spans
- switch analytics service to the otel-backed implementation

Author: tyulyukov

.codex/config.toml

@@ -0,0 +1,7 @@
+[mcp_servers.Railway]
+args = ["@railway/mcp-server"]
+command = "npx"
+
+[mcp_servers.chrome-devtools]
+args = ["chrome-devtools-mcp@latest", "--browserUrl", "http://127.0.0.1:9222"]
+command = "npx"

apps/landing/src/app/api/otel/traces/route.ts

@@ -0,0 +1,111 @@
+import { createHash } from "node:crypto";
+
+import { redactOtlpBody, upsertResourceAttribute, type OtlpTraceBody } from "~/lib/otelRedact";
+import { allow, type RateLimitTier } from "~/lib/otelRateLimit";
+import { tokenHash, verifyJiraToken } from "~/lib/otelVerifier";
+
+export const runtime = "nodejs";
+
+const SERVICE_NAME = "marcode";
+
+function readBearerToken(request: Request): string | undefined {
+  const header = request.headers.get("authorization") ?? request.headers.get("Authorization");
+  if (!header) return undefined;
+  const match = /^Bearer\s+(.+)$/i.exec(header.trim());
+  return match?.[1]?.trim() || undefined;
+}
+
+function readClientIp(request: Request): string {
+  const forwardedFor = request.headers.get("x-forwarded-for");
+  if (forwardedFor) {
+    const first = forwardedFor.split(",")[0]?.trim();
+    if (first) return first;
+  }
+  return request.headers.get("x-real-ip") ?? "unknown";
+}
+
+function parseExtraHeaders(input: string | undefined): Record<string, string> {
+  if (!input) return {};
+  const out: Record<string, string> = {};
+  for (const part of input.split(",")) {
+    const [name, ...rest] = part.split("=");
+    if (!name) continue;
+    const value = rest.join("=").trim();
+    if (!value) continue;
+    out[name.trim()] = value;
+  }
+  return out;
+}
+
+function isOtlpTraceBody(value: unknown): value is OtlpTraceBody {
+  return (
+    typeof value === "object" &&
+    value !== null &&
+    "resourceSpans" in value &&
+    Array.isArray((value as { resourceSpans: unknown }).resourceSpans)
+  );
+}
+
+export async function POST(request: Request): Promise<Response> {
+  const endpoint = process.env.OTEL_EXPORTER_OTLP_TRACES_ENDPOINT;
+  if (!endpoint) {
+    return Response.json({ error: "OTEL ingest not configured" }, { status: 503 });
+  }
+
+  const ip = readClientIp(request);
+  const ipHash = createHash("sha256").update(ip).digest("hex").slice(0, 16);
+  const token = readBearerToken(request);
+  const limitKey = `${ip}:${token ? tokenHash(token) : "anon"}`;
+
+  const verification = token ? await verifyJiraToken(token) : { valid: false, isGenesis: false };
+  const isGenesis = verification.isGenesis;
+  const tier: RateLimitTier = verification.valid ? "authed" : "anonymous";
+
+  if (!isGenesis) {
+    const result = allow(limitKey, tier);
+    if (!result.allowed) {
+      return new Response(JSON.stringify({ error: "Rate limit exceeded" }), {
+        status: 429,
+        headers: {
+          "Content-Type": "application/json",
+          "Retry-After": Math.max(1, Math.ceil((result.resetAt - Date.now()) / 1000)).toString(),
+        },
+      });
+    }
+  }
+
+  let body: unknown;
+  try {
+    body = await request.json();
+  } catch {
+    return Response.json({ error: "Invalid JSON body" }, { status: 400 });
+  }
+  if (!isOtlpTraceBody(body)) {
+    return Response.json({ error: "Body must be OTLP/JSON with resourceSpans" }, { status: 400 });
+  }
+
+  upsertResourceAttribute(body, "service.name", { stringValue: SERVICE_NAME });
+  upsertResourceAttribute(body, "user.is_genesis", { boolValue: isGenesis });
+  upsertResourceAttribute(body, "ingest.client_ip_hash", { stringValue: ipHash });
+  redactOtlpBody(body);
+
+  const extraHeaders = parseExtraHeaders(process.env.OTEL_EXPORTER_OTLP_TRACES_HEADERS);
+  let upstream: Response;
+  try {
+    upstream = await fetch(endpoint, {
+      method: "POST",
+      headers: { "Content-Type": "application/json", ...extraHeaders },
+      body: JSON.stringify(body),
+    });
+  } catch (cause) {
+    return Response.json({ error: "Trace export failed", detail: String(cause) }, { status: 502 });
+  }
+  if (!upstream.ok) {
+    const text = await upstream.text().catch(() => "");
+    return new Response(JSON.stringify({ error: "Tempo rejected the trace", detail: text }), {
+      status: 502,
+      headers: { "Content-Type": "application/json" },
+    });
+  }
+  return new Response(null, { status: 204 });
+}

apps/landing/src/lib/otelRateLimit.ts

@@ -0,0 +1,70 @@
+/**
+ * Sliding-window rate limiter for the OTEL ingest endpoint.
+ *
+ * - Authed (valid Jira token): 120 req/min per (IP, tokenHash) bucket.
+ * - Anonymous (no token / invalid): 30 req/min per IP bucket.
+ * - Genesis: bypass entirely (caller skips the check).
+ *
+ * In-memory single-instance store. Adequate for the current Dockerized
+ * single-replica landing deploy; swap for Redis if we ever scale horizontally.
+ */
+const RATE_WINDOW_MS = 60 * 1000;
+const AUTHED_LIMIT_PER_WINDOW = 120;
+const ANONYMOUS_LIMIT_PER_WINDOW = 30;
+const PRUNE_INTERVAL_MS = 10 * 1000;
+
+const buckets = new Map<string, number[]>();
+
+let janitorStarted = false;
+function ensureJanitor() {
+  if (janitorStarted) return;
+  janitorStarted = true;
+  const interval = setInterval(() => {
+    const cutoff = Date.now() - RATE_WINDOW_MS;
+    for (const [key, timestamps] of buckets) {
+      const fresh = timestamps.filter((t) => t > cutoff);
+      if (fresh.length === 0) {
+        buckets.delete(key);
+      } else if (fresh.length !== timestamps.length) {
+        buckets.set(key, fresh);
+      }
+    }
+  }, PRUNE_INTERVAL_MS);
+  if (typeof interval === "object" && interval !== null && "unref" in interval) {
+    (interval as { unref: () => void }).unref();
+  }
+}
+
+export type RateLimitTier = "authed" | "anonymous";
+
+export interface AllowResult {
+  readonly allowed: boolean;
+  readonly remaining: number;
+  readonly resetAt: number;
+}
+
+export function allow(key: string, tier: RateLimitTier): AllowResult {
+  ensureJanitor();
+  const limit = tier === "authed" ? AUTHED_LIMIT_PER_WINDOW : ANONYMOUS_LIMIT_PER_WINDOW;
+  const now = Date.now();
+  const cutoff = now - RATE_WINDOW_MS;
+
+  const timestamps = (buckets.get(key) ?? []).filter((t) => t > cutoff);
+  if (timestamps.length >= limit) {
+    buckets.set(key, timestamps);
+    const oldest = timestamps[0] ?? now;
+    return {
+      allowed: false,
+      remaining: 0,
+      resetAt: oldest + RATE_WINDOW_MS,
+    };
+  }
+
+  timestamps.push(now);
+  buckets.set(key, timestamps);
+  return {
+    allowed: true,
+    remaining: limit - timestamps.length,
+    resetAt: now + RATE_WINDOW_MS,
+  };
+}

apps/landing/src/lib/otelRedact.ts

@@ -0,0 +1,131 @@
+/**
+ * OTLP body redactor.
+ *
+ * Walks the standard OTLP/JSON shape and drops attribute entries whose key
+ * matches a deny-list of suffixes. Mirrors the server-side
+ * `OTEL_DENY_SUFFIXES` in `apps/server/src/observability/Attributes.ts` so
+ * sensitive data has two lines of defense.
+ */
+export const OTEL_DENY_SUFFIXES: ReadonlyArray<string> = [
+  "cwd",
+  "path",
+  "file_path",
+  "filename",
+  "directory",
+  "command_line",
+  "argv",
+  "cmdline",
+  "body",
+  "content",
+  "text",
+  "message_text",
+  "prompt",
+  "completion",
+  "diff",
+  "patch",
+  "email",
+  "username",
+  "account_id",
+  "url",
+  "token",
+  "secret",
+  "key",
+  "password",
+  "authorization",
+  "title",
+  "summary",
+  "description",
+];
+
+export function isDeniedAttributeKey(key: string): boolean {
+  for (const suffix of OTEL_DENY_SUFFIXES) {
+    if (key === suffix || key.endsWith(`.${suffix}`)) return true;
+  }
+  return false;
+}
+
+interface OtlpKeyValue {
+  key: string;
+  value?: unknown;
+}
+
+function filterAttributes(attrs: unknown): OtlpKeyValue[] {
+  if (!Array.isArray(attrs)) return [];
+  return (attrs as OtlpKeyValue[]).filter(
+    (entry) =>
+      typeof entry === "object" &&
+      entry !== null &&
+      typeof entry.key === "string" &&
+      !isDeniedAttributeKey(entry.key),
+  );
+}
+
+interface OtlpSpan {
+  attributes?: OtlpKeyValue[];
+  events?: Array<{ attributes?: OtlpKeyValue[] }>;
+}
+
+interface OtlpScopeSpans {
+  spans?: OtlpSpan[];
+}
+
+interface OtlpResource {
+  attributes?: OtlpKeyValue[];
+}
+
+interface OtlpResourceSpans {
+  resource?: OtlpResource;
+  scopeSpans?: OtlpScopeSpans[];
+}
+
+export interface OtlpTraceBody {
+  resourceSpans?: OtlpResourceSpans[];
+}
+
+/**
+ * redactOtlpBody - mutates and returns the body with deny-listed attributes
+ * removed from resource, span, and span event attribute arrays.
+ */
+export function redactOtlpBody(body: OtlpTraceBody): OtlpTraceBody {
+  for (const rs of body.resourceSpans ?? []) {
+    if (rs.resource) {
+      rs.resource.attributes = filterAttributes(rs.resource.attributes);
+    }
+    for (const ss of rs.scopeSpans ?? []) {
+      for (const span of ss.spans ?? []) {
+        span.attributes = filterAttributes(span.attributes);
+        for (const event of span.events ?? []) {
+          event.attributes = filterAttributes(event.attributes);
+        }
+      }
+    }
+  }
+  return body;
+}
+
+/**
+ * upsertResourceAttribute - inserts or replaces a key/value entry on every
+ * resourceSpans[i].resource.attributes. Used to OVERRIDE `service.name` and
+ * stamp `user.is_genesis`/`ingest.client_ip_hash` server-side regardless of
+ * what the client sent.
+ */
+export function upsertResourceAttribute(
+  body: OtlpTraceBody,
+  key: string,
+  value: { stringValue?: string; boolValue?: boolean; intValue?: string },
+): void {
+  for (const rs of body.resourceSpans ?? []) {
+    rs.resource = rs.resource ?? { attributes: [] };
+    const attrs: OtlpKeyValue[] = Array.isArray(rs.resource.attributes)
+      ? rs.resource.attributes
+      : [];
+    const existingIndex = attrs.findIndex((entry) => entry.key === key);
+    const next: OtlpKeyValue = { key, value };
+    if (existingIndex >= 0) {
+      attrs[existingIndex] = next;
+    } else {
+      attrs.push(next);
+    }
+    rs.resource.attributes = attrs;
+  }
+}

apps/landing/src/lib/otelVerifier.ts

@@ -0,0 +1,54 @@
+import { createHash } from "node:crypto";
+
+const ATLASSIAN_ME_URL = "https://api.atlassian.com/me";
+const VERIFY_CACHE_TTL_MS = 5 * 60 * 1000;
+const GENESIS_EMAIL_SUFFIXES = ["@gen.tech", "@obrio.co"] as const;
+
+export interface VerifyJiraTokenResult {
+  readonly valid: boolean;
+  readonly email?: string;
+  readonly isGenesis: boolean;
+}
+
+interface CachedEntry {
+  readonly result: VerifyJiraTokenResult;
+  readonly expiresAt: number;
+}
+
+const cache = new Map<string, CachedEntry>();
+
+export function tokenHash(token: string): string {
+  return createHash("sha256").update(token).digest("hex").slice(0, 16);
+}
+
+function isGenesisEmail(email: string | undefined): boolean {
+  if (!email) return false;
+  const normalized = email.trim().toLowerCase();
+  return GENESIS_EMAIL_SUFFIXES.some((suffix) => normalized.endsWith(suffix));
+}
+
+export async function verifyJiraToken(token: string): Promise<VerifyJiraTokenResult> {
+  const key = tokenHash(token);
+  const now = Date.now();
+  const cached = cache.get(key);
+  if (cached && cached.expiresAt > now) {
+    return cached.result;
+  }
+
+  let result: VerifyJiraTokenResult = { valid: false, isGenesis: false };
+  try {
+    const response = await fetch(ATLASSIAN_ME_URL, {
+      headers: { Authorization: `Bearer ${token}`, Accept: "application/json" },
+    });
+    if (response.ok) {
+      const body = (await response.json()) as { email?: unknown };
+      const email = typeof body.email === "string" ? body.email : undefined;
+      result = { valid: true, isGenesis: isGenesisEmail(email), ...(email ? { email } : {}) };
+    }
+  } catch {
+    // Network failure → treat as anonymous; don't bubble up.
+  }
+
+  cache.set(key, { result, expiresAt: now + VERIFY_CACHE_TTL_MS });
+  return result;
+}