fix: batch and retry telemetry, keep id properties, show sign-in errors

.changeset/login-page-errors.md

@@ -0,0 +1,5 @@
+---
+"@codelitdev/oauth-server-kit": patch
+---
+
+The hosted sign-in page now shows the auth API's error `message` (for example a product's signup refusal or an expired code) instead of always "Sign-in failed. Please try again.", which stays the fallback when the response has no message.

.changeset/observability-batching.md

@@ -0,0 +1,5 @@
+---
+"@codelitdev/observability": patch
+---
+
+Batch and retry telemetry: PostHog captures go to `/batch/` and OTLP logs are sent many records per request, every second or 50 items, with a 10-second timeout and two retries on network errors, 429 and 5xx. Previously each item was its own request, aborted after 2 seconds without a retry, so bursts or a slow network silently dropped events. Add `contextPolicy.identifierKeys` so allowlisted id properties (UUIDs) aren't masked as tokens.

packages/oauth-server-kit/src/oauth-pages.test.ts

@@ -27,6 +27,43 @@ function pagesApp(overrides: Partial<CreateOAuthPagesRouterOptions> = {}) {
   return app;
 }
 
+/** Runs the login page's inline script against stub DOM and fetch, submits
+ * the email form, and returns the error banner text. */
+async function submitEmail(html: string, response: Response): Promise<string> {
+  const script = html.match(/<script>([\s\S]*?)<\/script>/)?.[1];
+  if (!script) throw new Error("login_script_missing");
+  const listeners = new Map<string, (event: unknown) => Promise<void>>();
+  const element = (id: string) => ({
+    id,
+    hidden: false,
+    textContent: "",
+    style: { display: "" },
+    value: "person@example.com",
+    addEventListener(type: string, listener: (event: unknown) => Promise<void>) {
+      listeners.set(`${id}:${type}`, listener);
+    },
+    querySelector: () => ({ value: "person@example.com" }),
+  });
+  const elements = new Map(
+    ["error", "email-form", "otp-form", "otp", "change-email"].map((id) => [
+      id,
+      element(id),
+    ]),
+  );
+  const document = {
+    getElementById: (id: string) => elements.get(id) ?? null,
+    querySelectorAll: () => [],
+  };
+  const location = { search: "", href: "https://auth.example.com/login" };
+  new Function("document", "location", "fetch", script)(
+    document,
+    location,
+    async () => response,
+  );
+  await listeners.get("email-form:submit")?.({ preventDefault() {} });
+  return elements.get("error")?.textContent ?? "";
+}
+
 describe("createOAuthPagesRouter", () => {
   it("renders configured email and social login using the custom auth path", async () => {
     await withHttpServer(pagesApp(), async (baseUrl) => {
@@ -64,6 +101,25 @@ describe("createOAuthPagesRouter", () => {
     });
   });
 
+  it("shows the auth API's error message, or a generic one without it", async () => {
+    await withHttpServer(pagesApp(), async (baseUrl) => {
+      const html = await (await fetch(`${baseUrl}/login`)).text();
+      const refused = await submitEmail(
+        html,
+        Response.json(
+          { code: "DISPOSABLE_EMAIL", message: "Use your work or personal email." },
+          { status: 403 },
+        ),
+      );
+      expect(refused).toBe("Use your work or personal email.");
+      const broken = await submitEmail(
+        html,
+        new Response("<html>Bad gateway</html>", { status: 502 }),
+      );
+      expect(broken).toBe("Sign-in failed. Please try again.");
+    });
+  });
+
   it("falls back from a malicious or malformed ordinary-login redirect", async () => {
     for (const redirect of ["https://evil.example/phish", "not-a-url"]) {
       await withHttpServer(pagesApp(), async (baseUrl) => {

packages/oauth-server-kit/src/oauth-pages.ts

@@ -240,7 +240,7 @@ function loginScript(options: NormalizedOptions, oauth: boolean, redirect?: stri
   return `<script>(function(){
 var authBasePath=${authBasePath};var oauth=${oauth};var oauthQuery=location.search.slice(1);var redirectTarget=${redirectTarget};var errorEl=document.getElementById("error");
 function showError(message){errorEl.textContent=message||"Sign-in failed. Please try again.";errorEl.style.display="block"}function clearError(){errorEl.style.display="none"}
-async function post(path,body){var response=await fetch(authBasePath+path,{method:"POST",headers:{"content-type":"application/json",accept:"application/json"},body:JSON.stringify(body)});var data={};try{data=await response.json()}catch{}if(!response.ok)throw new Error("Sign-in failed. Please try again.");return data}
+async function post(path,body){var response=await fetch(authBasePath+path,{method:"POST",headers:{"content-type":"application/json",accept:"application/json"},body:JSON.stringify(body)});var data={};try{data=await response.json()}catch{}if(!response.ok)throw new Error(typeof data.message==="string"&&data.message?data.message:"Sign-in failed. Please try again.");return data}
 function follow(data){var target=data.url||data.redirect_uri||redirectTarget;if(!target)throw new Error("Sign-in failed. Please try again.");location.assign(target)}
 var emailForm=document.getElementById("email-form"),otpForm=document.getElementById("otp-form"),email="";
 if(emailForm){emailForm.addEventListener("submit",async function(event){event.preventDefault();clearError();email=emailForm.querySelector("input").value;try{await post("/email-otp/send-verification-otp",{email:email,type:"sign-in",...(oauth?{oauth_query:oauthQuery}:{})});emailForm.hidden=true;otpForm.hidden=false}catch(error){showError(error.message)}});otpForm.addEventListener("submit",async function(event){event.preventDefault();clearError();try{var data=await post("/sign-in/email-otp",{email:email,otp:document.getElementById("otp").value,name:email.split("@")[0],...(oauth?{oauth_query:oauthQuery}:{})});follow(data)}catch(error){showError(error.message)}});document.getElementById("change-email").addEventListener("click",function(){otpForm.hidden=true;emailForm.hidden=false;clearError()})}

packages/observability/README.md

@@ -10,6 +10,18 @@ client in tests. `logs.otlp` adds an OTLP/HTTP log fan-out while structured
 stdout remains enabled; both remote pipelines are failure-isolated and
 timeout-bounded during shutdown.
 
+Both pipelines queue what they send and post it in batches (PostHog
+`/batch/`, one OTLP request per batch): every second or once 50 items are
+waiting, with a 10-second request timeout and two retries on network errors,
+429 and 5xx. The queue holds at most 1,000 items, dropping the oldest. Call
+`shutdown()` before a process exits, or queued items are lost.
+
+`contextPolicy.propertyAllowlist` decides which properties are sent. String
+values are masked for emails, secrets and anything that looks like a token
+(24 or more letters, digits, `-` or `_`). List keys that hold opaque record
+ids, such as `team_id`, in `contextPolicy.identifierKeys` so their UUIDs are
+not masked too. Never list a key that can hold a session or access token.
+
 Next.js (and other browser) apps use `createBrowserObservability` from
 `@codelitdev/observability/browser`. Pass `apiKey` from request-time env
 (`POSTHOG_API_KEY`), not `NEXT_PUBLIC_*`, so OSS images can enable telemetry

packages/observability/src/create.test.ts

@@ -215,6 +215,62 @@ describe("createObservability", () => {
     expect(Date.now() - started).toBeLessThan(500);
   });
 
+  it("keeps identifier values that the long-token mask would hide", () => {
+    const client = fakeClient();
+    const obs = createObservability({
+      serviceName: "reference-api",
+      environment: "test",
+      posthog: { apiKey: "phc_test", client },
+      contextPolicy: {
+        propertyAllowlist: new Set(["team_id", "note", "owner_id"]),
+        identifierKeys: new Set(["team_id", "owner_id"]),
+      },
+    });
+    const teamId = "01a15eb2-1861-7bbe-9b50-6a8f0bf0ef6e";
+    obs.captureEvent({
+      event: "team_paused",
+      source: "abuse",
+      properties: {
+        team_id: teamId,
+        owner_id: "someone@example.com",
+        note: "abcdefghijklmnopqrstuvwxyz0123",
+      },
+    });
+    const [event] = client.events as Array<{ properties: Record<string, unknown> }>;
+    expect(event?.properties.team_id).toBe(teamId);
+    expect(event?.properties.owner_id).toBe("[redacted-email]");
+    expect(event?.properties.note).toBe("[redacted-token]");
+  });
+
+  it("batches events from the default HTTP client with their timestamps", async () => {
+    const fetchMock = vi.fn(async () => new Response(null, { status: 200 }));
+    const previousFetch = globalThis.fetch;
+    globalThis.fetch = fetchMock as unknown as typeof fetch;
+    try {
+      const obs = createObservability({
+        serviceName: "reference-api",
+        environment: "test",
+        posthog: { apiKey: "phc_test", host: "https://posthog.example" },
+        contextPolicy: { propertyAllowlist: new Set() },
+        dedupeTtlMs: 0,
+        perSourceCap: Number.POSITIVE_INFINITY,
+      });
+      for (let i = 0; i < 10; i++) {
+        obs.captureEvent({ event: `step_${i}`, source: "worker", subjectId: `t${i}` });
+      }
+      await obs.shutdown(500);
+      const calls = fetchMock.mock.calls as unknown as Array<[string, RequestInit]>;
+      expect(calls).toHaveLength(1);
+      expect(calls[0]?.[0]).toBe("https://posthog.example/batch/");
+      const body = JSON.parse(String(calls[0]?.[1].body));
+      expect(body.api_key).toBe("phc_test");
+      expect(body.batch).toHaveLength(10);
+      expect(typeof body.batch[0].timestamp).toBe("string");
+    } finally {
+      globalThis.fetch = previousFetch;
+    }
+  });
+
   it("fans structured logs to OTLP without disabling stdout", async () => {
     const fetchMock = vi.fn(async () => new Response(null, { status: 200 }));
     const previousFetch = globalThis.fetch;

packages/observability/src/create.ts

@@ -2,6 +2,7 @@ import type { NextFunction, Request, Response } from "express";
 import pino, { type Logger } from "pino";
 import { type Clock, createDedupe, createSourceCap } from "./limits.js";
 import { opaqueSubjectId, pickAllowlisted, redactText } from "./redaction.js";
+import { createBatchSender } from "./transport.js";
 
 export type ObservabilityClock = Clock;
 
@@ -23,70 +24,57 @@ function createOtlpLogStream(input: {
   endpoint: string;
   headers?: Record<string, string>;
 }) {
-  const pending = new Set<Promise<void>>();
-  const stream = {
-    write(message: string) {
-      const controller = new AbortController();
-      const timer = setTimeout(() => controller.abort(), 2_000);
-      const request = fetch(input.endpoint, {
+  const sender = createBatchSender<string>({
+    send: (messages, signal) =>
+      fetch(input.endpoint, {
         method: "POST",
         headers: { "content-type": "application/json", ...input.headers },
         body: JSON.stringify({
           resourceLogs: [
             {
               scopeLogs: [
                 {
-                  logRecords: [{ body: { stringValue: message.trim() } }],
+                  logRecords: messages.map((message) => ({
+                    body: { stringValue: message },
+                  })),
                 },
               ],
             },
           ],
         }),
-        signal: controller.signal,
-      })
-        .then(() => undefined)
-        .catch(() => undefined)
-        .finally(() => {
-          clearTimeout(timer);
-          pending.delete(request);
-        });
-      pending.add(request);
-    },
-  };
+        signal,
+      }),
+  });
   return {
-    stream,
-    async shutdown() {
-      await Promise.allSettled([...pending]);
+    stream: {
+      write(message: string) {
+        sender.push(message.trim());
+      },
     },
+    shutdown: () => sender.shutdown(),
   };
 }
 
 function createHttpCaptureClient(input: {
   apiKey: string;
   host?: string;
 }): CaptureClient {
-  const endpoint = `${(input.host ?? "https://app.posthog.com").replace(/\/$/, "")}/capture/`;
-  const pending = new Set<Promise<void>>();
-  const send = (payload: Record<string, unknown>) => {
-    const controller = new AbortController();
-    const timer = setTimeout(() => controller.abort(), 2_000);
-    const request = fetch(endpoint, {
-      method: "POST",
-      headers: {
-        "content-type": "application/json",
-        authorization: `Bearer ${input.apiKey}`,
-      },
-      body: JSON.stringify({ api_key: input.apiKey, ...payload }),
-      signal: controller.signal,
-    })
-      .then(() => undefined)
-      .catch(() => undefined)
-      .finally(() => {
-        clearTimeout(timer);
-        pending.delete(request);
-      });
-    pending.add(request);
-  };
+  const endpoint = `${(input.host ?? "https://app.posthog.com").replace(/\/$/, "")}/batch/`;
+  const sender = createBatchSender<Record<string, unknown>>({
+    send: (batch, signal) =>
+      fetch(endpoint, {
+        method: "POST",
+        headers: {
+          "content-type": "application/json",
+          authorization: `Bearer ${input.apiKey}`,
+        },
+        body: JSON.stringify({ api_key: input.apiKey, batch }),
+        signal,
+      }),
+  });
+  // Events wait in a batch, so each carries the time it happened.
+  const send = (payload: Record<string, unknown>) =>
+    sender.push({ ...payload, timestamp: new Date().toISOString() });
   return {
     captureException(error, distinctId, properties) {
       send({
@@ -107,14 +95,16 @@ function createHttpCaptureClient(input: {
         properties: input.properties,
       });
     },
-    async shutdown() {
-      await Promise.allSettled([...pending]);
-    },
+    shutdown: () => sender.shutdown(),
   };
 }
 
 export type ContextPolicy = {
   propertyAllowlist: ReadonlySet<string>;
+  /** Allowlisted keys holding opaque record ids (for example `team_id`).
+   * Their values are sent as-is apart from email and secret masking. Don't
+   * list keys that can hold session or access tokens. */
+  identifierKeys?: ReadonlySet<string>;
 };
 
 export type CreateObservabilityOptions = {
@@ -268,6 +258,7 @@ export function createObservability(
         ...pickAllowlisted(
           input.context ?? {},
           options.contextPolicy.propertyAllowlist,
+          options.contextPolicy.identifierKeys,
         ),
       });
     } catch {
@@ -297,6 +288,7 @@ export function createObservability(
             ...pickAllowlisted(
               input.properties ?? {},
               options.contextPolicy.propertyAllowlist,
+              options.contextPolicy.identifierKeys,
             ),
           },
         });

packages/observability/src/redaction.ts

@@ -6,12 +6,16 @@ const SENSITIVE_ASSIGNMENT =
   /\b(authorization|cookie|password|secret|token|api[_-]?key|pepper|digest|card|cvv|pan)\b\s*([=:])\s*(?:"[^"]*"|'[^']*'|[^\s,;&})\]]+)/gi;
 const BEARER_TOKEN = /\bBearer\s+[A-Za-z0-9._~+/-]+=*/gi;
 
-export function redactText(input: string, limit = 500): string {
-  const masked = input
+export function redactText(
+  input: string,
+  limit = 500,
+  options: { keepLongTokens?: boolean } = {},
+): string {
+  let masked = input
     .replace(SENSITIVE_ASSIGNMENT, "$1$2[redacted]")
     .replace(BEARER_TOKEN, "Bearer [redacted]")
-    .replace(EMAIL, "[redacted-email]")
-    .replace(TOKEN, "[redacted-token]");
+    .replace(EMAIL, "[redacted-email]");
+  if (!options.keepLongTokens) masked = masked.replace(TOKEN, "[redacted-token]");
   if (masked.length <= limit) return masked;
   return `${masked.slice(0, limit)}...`;
 }
@@ -20,9 +24,15 @@ export function isSensitiveKey(key: string): boolean {
   return SECRET_KEYS.test(key);
 }
 
+/**
+ * Keeps allowlisted keys only, with secrets and emails masked. Values of
+ * `identifierKeys` (opaque record ids such as UUIDs) skip the long-token mask,
+ * which would otherwise hide every id of 24 or more characters.
+ */
 export function pickAllowlisted(
   input: Record<string, unknown>,
   allowlist: ReadonlySet<string>,
+  identifierKeys: ReadonlySet<string> = new Set(),
 ): Record<string, string | number | boolean> {
   const out: Record<string, string | number | boolean> = {};
   for (const [key, value] of Object.entries(input)) {
@@ -32,7 +42,7 @@ export function pickAllowlisted(
       continue;
     }
     if (typeof value === "string") {
-      out[key] = redactText(value, 500);
+      out[key] = redactText(value, 500, { keepLongTokens: identifierKeys.has(key) });
     }
   }
   return out;