refactor(retention): move log stripping to gateway

apps/gateway/src/chat/chat.ts

@@ -1981,6 +1981,13 @@ chat.openapi(completions, async (c) => {
 		});
 	}
 
+	// Organization data retention level. Captured here (right after the org is
+	// resolved) so every log path — including the early rejections below and the
+	// insertLog wrapper further down — can decide whether payload fields should
+	// be persisted. Orgs backing end-user wallets are always regular PAYG orgs,
+	// so the withWalletCredits substitution below never changes this value.
+	const retentionLevel = organization.retentionLevel ?? "none";
+
 	// Note: the end-user-wallet credits substitution (withWalletCredits) happens
 	// further below — orgs backing end-user wallets are always regular
 	// PAYG/credits orgs, never dev-plan orgs, so it cannot affect the dev-plan
@@ -2140,7 +2147,7 @@ chat.openapi(completions, async (c) => {
 						usedServiceTier: null,
 						dataStorageCost: "0",
 					},
-					{ syncInsert: syncLogInsert },
+					{ syncInsert: syncLogInsert, retentionLevel },
 				);
 			} catch (error) {
 				logger.error("Failed to log unsupported service tier rejection", {
@@ -2615,7 +2622,7 @@ chat.openapi(completions, async (c) => {
 						usedServiceTier: null,
 						dataStorageCost: "0",
 					},
-					{ syncInsert: syncLogInsert },
+					{ syncInsert: syncLogInsert, retentionLevel },
 				);
 			} catch (error) {
 				logger.error("Failed to log budget-thinking rejection", {
@@ -2640,9 +2647,6 @@ chat.openapi(completions, async (c) => {
 	let usedRegion: string | undefined = requestedRegion;
 	let routingMetadata: RoutingMetadata | undefined;
 
-	// Extract retention level for data storage cost calculation
-	const retentionLevel = organization?.retentionLevel ?? "none";
-
 	// Get image size limits from environment variables or use defaults
 	const freeLimitMB = Number(process.env.IMAGE_SIZE_LIMIT_FREE_MB) || 50;
 	const proLimitMB = Number(process.env.IMAGE_SIZE_LIMIT_PRO_MB) || 100;
@@ -5087,7 +5091,10 @@ chat.openapi(completions, async (c) => {
 				gatewayContentFilterResponse:
 					logData.gatewayContentFilterResponse ?? gatewayContentFilterResponse,
 			},
-			options,
+			// Default the retention level from the resolved organization so payload
+			// fields are stripped before publishing to the log queue for
+			// non-retaining orgs. A caller-supplied value still wins.
+			{ retentionLevel, ...options },
 		);
 
 	if (contentFilterBlocked) {

apps/gateway/src/lib/logs-retention.spec.ts

@@ -0,0 +1,86 @@
+import { beforeEach, describe, expect, it, vi } from "vitest";
+
+import { insertLog } from "./logs.js";
+
+import type { LogInsertData } from "@llmgateway/db";
+
+const publishToQueue = vi.fn();
+
+vi.mock(import("@llmgateway/cache"), async (importOriginal) => {
+	const actual = await importOriginal();
+	return {
+		...actual,
+		publishToQueue: (...args: unknown[]) => publishToQueue(...args),
+	};
+});
+
+vi.mock("@llmgateway/instrumentation", () => ({
+	recordChatCompletionMetrics: vi.fn(),
+}));
+
+function baseLogData(overrides: Partial<LogInsertData>): LogInsertData {
+	return {
+		requestId: "req-1",
+		organizationId: "org-1",
+		projectId: "project-1",
+		apiKeyId: "api-key-1",
+		duration: 100,
+		requestedModel: "some-model",
+		usedModel: "some-model",
+		usedProvider: "openai",
+		responseSize: 0,
+		mode: "credits",
+		usedMode: "credits",
+		finishReason: "stop",
+		hasError: false,
+		messages: [{ role: "user", content: "secret prompt" }],
+		content: "secret completion",
+		reasoningContent: "secret reasoning",
+		tools: [{ type: "function", function: { name: "f" } }],
+		toolChoice: "auto",
+		toolResults: [{ id: "1" }],
+		responsesApiData: { foo: "bar" },
+		...overrides,
+	} as LogInsertData;
+}
+
+describe("insertLog retention stripping", () => {
+	beforeEach(() => {
+		publishToQueue.mockClear();
+	});
+
+	it("strips payload fields before publishing when retention is none", async () => {
+		await insertLog(baseLogData({}), { retentionLevel: "none" });
+
+		expect(publishToQueue).toHaveBeenCalledTimes(1);
+		const published = publishToQueue.mock.calls[0][1] as LogInsertData;
+		expect(published.messages).toBeNull();
+		expect(published.content).toBeNull();
+		expect(published.reasoningContent).toBeNull();
+		expect(published.tools).toBeNull();
+		expect(published.toolChoice).toBeNull();
+		expect(published.toolResults).toBeNull();
+		expect(published.responsesApiData).toBeNull();
+		// Metadata is preserved.
+		expect(published.requestId).toBe("req-1");
+		expect(published.organizationId).toBe("org-1");
+		expect(JSON.stringify(published)).not.toContain("secret");
+	});
+
+	it("keeps payload fields when retention is retain", async () => {
+		await insertLog(baseLogData({}), { retentionLevel: "retain" });
+
+		const published = publishToQueue.mock.calls[0][1] as LogInsertData;
+		expect(published.content).toBe("secret completion");
+		expect(published.messages).toEqual([
+			{ role: "user", content: "secret prompt" },
+		]);
+	});
+
+	it("keeps payload fields when retention is unspecified", async () => {
+		await insertLog(baseLogData({}));
+
+		const published = publishToQueue.mock.calls[0][1] as LogInsertData;
+		expect(published.content).toBe("secret completion");
+	});
+});

apps/gateway/src/lib/logs.ts

@@ -2,6 +2,7 @@ import { publishToQueue, LOG_QUEUE } from "@llmgateway/cache";
 import {
 	db,
 	log,
+	stripRetentionSensitiveLogFields,
 	UnifiedFinishReason,
 	type LogInsertData,
 } from "@llmgateway/db";
@@ -285,8 +286,17 @@ export type LogData = InferInsertModel<typeof log>;
 
 export async function insertLog(
 	logData: LogInsertData,
-	options?: { syncInsert?: boolean },
+	options?: { syncInsert?: boolean; retentionLevel?: "retain" | "none" | null },
 ): Promise<unknown> {
+	// When the organization doesn't retain request/response data, strip the
+	// payload fields here — before the row is ever published to the log queue —
+	// so large prompts, completions, and tool payloads never travel through
+	// Redis. The worker performs the same strip as a safety net for any log whose
+	// retention level wasn't resolved at publish time.
+	if (options?.retentionLevel === "none") {
+		logData = stripRetentionSensitiveLogFields(logData);
+	}
+
 	// Stealth providers: the raw upstream error may reveal the underlying
 	// platform, so it survives only in internalErrorDetails — a column excluded
 	// from public API routes and the UI — while the public errorDetails keeps

apps/worker/src/worker.ts

@@ -15,7 +15,6 @@ import {
 	addApiKeyPeriodDuration,
 	and,
 	apiKey,
-	cdb,
 	closeDatabase,
 	db,
 	enqueueWebhookDeliveries,
@@ -122,11 +121,6 @@ const LOG_QUEUE_CONCURRENCY = Math.max(
 	1,
 	Number(process.env.LOG_QUEUE_CONCURRENCY) || 4,
 );
-// Cache organization retention levels to avoid a serial Postgres round-trip
-// before every log batch insert. retentionLevel changes rarely; the short TTL
-// bounds how long a stale value can keep retaining or stripping log payloads.
-const ORG_RETENTION_CACHE_TTL_MS =
-	Number(process.env.ORG_RETENTION_CACHE_TTL_MS) || 60_000;
 const CREDIT_BATCH_SIZE = Number(process.env.CREDIT_BATCH_SIZE) || 100;
 const BATCH_PROCESSING_INTERVAL_SECONDS =
 	Number(process.env.CREDIT_BATCH_INTERVAL) || 5;
@@ -1686,50 +1680,6 @@ function recordLogInsertSuccess(): void {
 	logInsertCircuit.nextAttemptAt = 0;
 }
 
-const orgRetentionCache = new Map<
-	string,
-	{ retentionLevel: "retain" | "none"; expiresAt: number }
->();
-
-// Resolve organization retention levels, serving from the in-memory cache when
-// fresh and only querying Postgres for the ids that are missing or expired.
-async function getOrganizationRetentionLevels(
-	organizationIds: string[],
-): Promise<Map<string, "retain" | "none">> {
-	const now = Date.now();
-	const result = new Map<string, "retain" | "none">();
-	const missing: string[] = [];
-
-	for (const id of organizationIds) {
-		const cached = orgRetentionCache.get(id);
-		if (cached && cached.expiresAt > now) {
-			result.set(id, cached.retentionLevel);
-		} else {
-			missing.push(id);
-		}
-	}
-
-	if (missing.length > 0) {
-		const organizations = await cdb
-			.select({
-				id: organization.id,
-				retentionLevel: organization.retentionLevel,
-			})
-			.from(organization)
-			.where(inArray(organization.id, missing));
-
-		for (const org of organizations) {
-			result.set(org.id, org.retentionLevel);
-			orgRetentionCache.set(org.id, {
-				retentionLevel: org.retentionLevel,
-				expiresAt: now + ORG_RETENTION_CACHE_TTL_MS,
-			});
-		}
-	}
-
-	return result;
-}
-
 // Returns the number of messages successfully inserted, so the drain loop can
 // decide whether to sleep (partial batch) or immediately fetch the next batch
 // (full batch, queue likely still backed up).
@@ -1747,49 +1697,21 @@ export async function processLogQueue(): Promise<number> {
 	const MAX_RETRIES = 5;
 
 	try {
+		// The gateway decides what to persist: it strips request/response payload
+		// fields before publishing for orgs that don't retain data, so the worker
+		// inserts the queued rows as-is with no per-batch org retention lookup.
 		const logData = message.map((i) => JSON.parse(i) as LogInsertData);
-		const organizationIds = Array.from(
-			new Set(logData.map((data) => data.organizationId)),
-		);
-		const selectStart = Date.now();
-		const retentionByOrg =
-			organizationIds.length > 0
-				? await getOrganizationRetentionLevels(organizationIds)
-				: new Map<string, "retain" | "none">();
-		const selectMs = Date.now() - selectStart;
-
-		const processedLogData: (
-			| LogInsertData
-			| Omit<LogInsertData, "messages" | "content">
-		)[] = logData.map((data) => {
-			if (retentionByOrg.get(data.organizationId) === "none") {
-				const {
-					messages: _messages,
-					content: _content,
-					reasoningContent: _reasoningContent,
-					tools: _tools,
-					toolChoice: _toolChoice,
-					toolResults: _toolResults,
-					responsesApiData: _responsesApiData,
-					...metadataOnly
-				} = data;
-				return metadataOnly;
-			}
-
-			return data;
-		});
 
 		// Insert logs with retry logic
 		let lastError: Error | undefined;
 		for (let attempt = 0; attempt <= MAX_RETRIES; attempt++) {
 			try {
-				// Type assertion is safe here as both LogInsertData and its subset are compatible with the log insert schema
 				const insertStart = Date.now();
-				await db.insert(log).values(processedLogData as LogInsertData[]);
+				await db.insert(log).values(logData);
 				const insertMs = Date.now() - insertStart;
 				recordLogInsertSuccess();
 				logger.info(
-					`Processed log batch: ${message.length} rows (org lookup ${selectMs}ms, insert ${insertMs}ms)`,
+					`Processed log batch: ${message.length} rows (insert ${insertMs}ms)`,
 				);
 				return message.length; // Success, exit function
 			} catch (insertError) {

packages/db/src/index.ts

@@ -10,6 +10,7 @@ export * from "./email-recipients.js";
 export * from "./rate-limit-helpers.js";
 export * from "./schema.js";
 export * from "./log-payloads.js";
+export * from "./log-retention.js";
 export * from "./types.js";
 export * from "./migrate.js";
 export * from "./relations.js";

packages/db/src/log-retention.ts

@@ -0,0 +1,38 @@
+import type { LogInsertData } from "./types.js";
+
+/**
+ * Log columns that hold request/response payload data (the actual prompt,
+ * completion, reasoning, tool definitions/results and Responses API payload).
+ * Organizations with `retentionLevel` "none" do not retain any of this, so it
+ * is cleared before the row is persisted.
+ */
+export const RETENTION_SENSITIVE_LOG_FIELDS = [
+	"messages",
+	"content",
+	"reasoningContent",
+	"tools",
+	"toolChoice",
+	"toolResults",
+	"responsesApiData",
+] as const;
+
+/**
+ * Return a copy of the log data with every retention-sensitive payload field
+ * cleared. Used by the gateway so payloads for non-retaining orgs never travel
+ * through Redis (or reach the database) in the first place — the gateway is the
+ * sole decider of what gets persisted, so the worker inserts queued rows as-is.
+ */
+export function stripRetentionSensitiveLogFields<T extends LogInsertData>(
+	logData: T,
+): T {
+	return {
+		...logData,
+		messages: null,
+		content: null,
+		reasoningContent: null,
+		tools: null,
+		toolChoice: null,
+		toolResults: null,
+		responsesApiData: null,
+	};
+}