From 5df9b3439bd590bb85ec5209a7f7e632f1b3c4f0 Mon Sep 17 00:00:00 2001 From: Rajat Saxena Date: Wed, 21 Oct 2026 19:40:21 +0530 Subject: [PATCH 1/3] fix(observability): batch and retry telemetry; keep identifier properties Each capture and log line was its own request, aborted after 2 seconds with no retry, so bursts or a slow network silently dropped events (25% of requests in a SendLit test run). Queue them and send batches (PostHog /batch/, one OTLP request per batch) every second or 50 items, with a 10-second timeout and two retries on network errors, 429 and 5xx, and a bounded queue. The long-token mask also hid every UUID property such as team_id. Add contextPolicy.identifierKeys for allowlisted keys that hold opaque ids. --- .changeset/observability-batching.md | 5 ++ packages/observability/README.md | 12 +++ packages/observability/src/create.test.ts | 56 +++++++++++++ packages/observability/src/create.ts | 84 +++++++++---------- packages/observability/src/redaction.ts | 20 +++-- packages/observability/src/transport.test.ts | 73 +++++++++++++++++ packages/observability/src/transport.ts | 85 ++++++++++++++++++++ 7 files changed, 284 insertions(+), 51 deletions(-) create mode 100644 .changeset/observability-batching.md create mode 100644 packages/observability/src/transport.test.ts create mode 100644 packages/observability/src/transport.ts diff --git a/.changeset/observability-batching.md b/.changeset/observability-batching.md new file mode 100644 index 0000000..269d284 --- /dev/null +++ b/.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. diff --git a/packages/observability/README.md b/packages/observability/README.md index 086f820..caeede4 100644 --- a/packages/observability/README.md +++ b/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 diff --git a/packages/observability/src/create.test.ts b/packages/observability/src/create.test.ts index 35c51e7..65141cb 100644 --- a/packages/observability/src/create.test.ts +++ b/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 }>; + 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; diff --git a/packages/observability/src/create.ts b/packages/observability/src/create.ts index 27c7223..a9ca662 100644 --- a/packages/observability/src/create.ts +++ b/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,12 +24,9 @@ function createOtlpLogStream(input: { endpoint: string; headers?: Record; }) { - const pending = new Set>(); - const stream = { - write(message: string) { - const controller = new AbortController(); - const timer = setTimeout(() => controller.abort(), 2_000); - const request = fetch(input.endpoint, { + const sender = createBatchSender({ + send: (messages, signal) => + fetch(input.endpoint, { method: "POST", headers: { "content-type": "application/json", ...input.headers }, body: JSON.stringify({ @@ -36,28 +34,24 @@ function createOtlpLogStream(input: { { 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(), }; } @@ -65,28 +59,22 @@ function createHttpCaptureClient(input: { apiKey: string; host?: string; }): CaptureClient { - const endpoint = `${(input.host ?? "https://app.posthog.com").replace(/\/$/, "")}/capture/`; - const pending = new Set>(); - const send = (payload: Record) => { - 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>({ + 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) => + 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; + /** 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; }; 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, ), }, }); diff --git a/packages/observability/src/redaction.ts b/packages/observability/src/redaction.ts index 6c35ff8..81cc6e6 100644 --- a/packages/observability/src/redaction.ts +++ b/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, allowlist: ReadonlySet, + identifierKeys: ReadonlySet = new Set(), ): Record { const out: Record = {}; 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; diff --git a/packages/observability/src/transport.test.ts b/packages/observability/src/transport.test.ts new file mode 100644 index 0000000..19ba3e2 --- /dev/null +++ b/packages/observability/src/transport.test.ts @@ -0,0 +1,73 @@ +import { describe, expect, it, vi } from "bun:test"; +import { createBatchSender } from "./transport.js"; + +const ok = () => new Response(null, { status: 200 }); + +describe("createBatchSender", () => { + it("sends a burst in batches instead of one request per item", async () => { + const send = vi.fn(async () => ok()); + const sender = createBatchSender({ send, maxBatch: 50 }); + for (let i = 0; i < 120; i++) sender.push(i); + await sender.shutdown(); + const batches = send.mock.calls.map(([items]) => items as number[]); + expect(batches.map((items) => items.length)).toEqual([50, 50, 20]); + expect(batches.flat()).toEqual([...Array(120).keys()]); + }); + + it("sends a partial batch after the flush interval", async () => { + const send = vi.fn(async () => ok()); + const sender = createBatchSender({ send, flushIntervalMs: 10 }); + sender.push("a"); + expect(send).not.toHaveBeenCalled(); + await new Promise((resolve) => setTimeout(resolve, 30)); + expect(send).toHaveBeenCalledTimes(1); + }); + + it("retries on 429, 5xx and network errors, then gives up quietly", async () => { + const responses = [ + () => Promise.reject(new Error("socket hang up")), + () => Promise.resolve(new Response(null, { status: 503 })), + () => Promise.resolve(ok()), + ]; + const send = vi.fn(() => (responses.shift() ?? (() => Promise.resolve(ok())))()); + const sender = createBatchSender({ send, retryDelaysMs: [1, 1] }); + sender.push("event"); + await sender.shutdown(); + expect(send).toHaveBeenCalledTimes(3); + + const failing = vi.fn(async () => new Response(null, { status: 500 })); + const giveUp = createBatchSender({ send: failing, retryDelaysMs: [1, 1] }); + giveUp.push("event"); + await expect(giveUp.shutdown()).resolves.toBeUndefined(); + expect(failing).toHaveBeenCalledTimes(3); + }); + + it("does not retry other client errors", async () => { + const send = vi.fn(async () => new Response(null, { status: 400 })); + const sender = createBatchSender({ send, retryDelaysMs: [1, 1] }); + sender.push("event"); + await sender.shutdown(); + expect(send).toHaveBeenCalledTimes(1); + }); + + it("aborts a request that takes longer than the timeout", async () => { + const send = vi.fn( + (_items: string[], signal: AbortSignal) => + new Promise((_, reject) => + signal.addEventListener("abort", () => reject(new Error("aborted"))), + ), + ); + const sender = createBatchSender({ send, timeoutMs: 5, retryDelaysMs: [] }); + sender.push("event"); + await sender.shutdown(); + expect(send).toHaveBeenCalledTimes(1); + }); + + it("keeps the queue bounded by dropping the oldest items", async () => { + const send = vi.fn(async () => ok()); + const sender = createBatchSender({ send, maxBatch: 100, maxQueue: 3 }); + for (let i = 0; i < 5; i++) sender.push(i); + await sender.shutdown(); + expect(send.mock.calls[0]?.[0]).toEqual([2, 3, 4]); + }); +}); diff --git a/packages/observability/src/transport.ts b/packages/observability/src/transport.ts new file mode 100644 index 0000000..5a84e7c --- /dev/null +++ b/packages/observability/src/transport.ts @@ -0,0 +1,85 @@ +/** + * Sends telemetry in batches with retries, so a burst of captures (or a slow + * network) doesn't drop events. Items wait in a bounded queue and go out when + * `maxBatch` are waiting or after `flushIntervalMs`. A failed batch (network + * error, timeout, 429 or 5xx) is retried with backoff, then dropped; telemetry + * never throws into the caller. + */ +export type BatchSenderOptions = { + send(items: T[], signal: AbortSignal): Promise; + maxBatch?: number; + maxQueue?: number; + flushIntervalMs?: number; + timeoutMs?: number; + retryDelaysMs?: readonly number[]; +}; + +export type BatchSender = { + push(item: T): void; + /** Sends everything queued and waits for in-flight batches. */ + shutdown(): Promise; +}; + +const RETRYABLE_STATUS = (status: number) => status === 429 || status >= 500; +const sleep = (ms: number) => new Promise((resolve) => setTimeout(resolve, ms)); + +export function createBatchSender(options: BatchSenderOptions): BatchSender { + const maxBatch = options.maxBatch ?? 50; + const maxQueue = options.maxQueue ?? 1_000; + const flushIntervalMs = options.flushIntervalMs ?? 1_000; + const timeoutMs = options.timeoutMs ?? 10_000; + const retryDelaysMs = options.retryDelaysMs ?? [500, 2_000]; + const queue: T[] = []; + const inFlight = new Set>(); + let timer: ReturnType | undefined; + + async function deliver(items: T[]) { + for (let attempt = 0; attempt <= retryDelaysMs.length; attempt++) { + const controller = new AbortController(); + const abort = setTimeout(() => controller.abort(), timeoutMs); + try { + const response = await options.send(items, controller.signal); + if (!RETRYABLE_STATUS(response.status)) return; + } catch { + // Network error or timeout: retry below. + } finally { + clearTimeout(abort); + } + const delay = retryDelaysMs[attempt]; + if (delay === undefined) return; + await sleep(delay); + } + } + + function drain() { + if (timer) { + clearTimeout(timer); + timer = undefined; + } + while (queue.length) { + const request = deliver(queue.splice(0, maxBatch)).finally(() => + inFlight.delete(request), + ); + inFlight.add(request); + } + } + + return { + push(item) { + // Keep memory bounded if the endpoint is down: drop the oldest. + if (queue.length >= maxQueue) queue.shift(); + queue.push(item); + if (queue.length >= maxBatch) { + drain(); + } else if (!timer) { + timer = setTimeout(drain, flushIntervalMs); + // Pending telemetry must not keep a CLI or test process alive. + (timer as { unref?: () => void }).unref?.(); + } + }, + async shutdown() { + drain(); + await Promise.allSettled([...inFlight]); + }, + }; +} From 7e59ce1c0b504b59aee07701aeaaafb91479958b Mon Sep 17 00:00:00 2001 From: Rajat Saxena Date: Wed, 21 Oct 2026 19:40:56 +0530 Subject: [PATCH 2/3] fix(oauth-server-kit): show the auth API's error message on the sign-in page post() threw a fixed 'Sign-in failed. Please try again.' for every non-2xx response, so product refusals (disposable email, signup caps) and Better Auth errors never reached the person. Use the response's message when it has one; showError already sets textContent, so it isn't parsed as HTML. Closes #11 --- .changeset/login-page-errors.md | 5 ++ .../oauth-server-kit/src/oauth-pages.test.ts | 56 +++++++++++++++++++ packages/oauth-server-kit/src/oauth-pages.ts | 2 +- 3 files changed, 62 insertions(+), 1 deletion(-) create mode 100644 .changeset/login-page-errors.md diff --git a/.changeset/login-page-errors.md b/.changeset/login-page-errors.md new file mode 100644 index 0000000..674c57a --- /dev/null +++ b/.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. diff --git a/packages/oauth-server-kit/src/oauth-pages.test.ts b/packages/oauth-server-kit/src/oauth-pages.test.ts index 420a78c..9e49ac0 100644 --- a/packages/oauth-server-kit/src/oauth-pages.test.ts +++ b/packages/oauth-server-kit/src/oauth-pages.test.ts @@ -27,6 +27,43 @@ function pagesApp(overrides: Partial = {}) { 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 { + const script = html.match(/