From 6c737d46ecf650bfb9f94b0f938b28c5eee98a8b Mon Sep 17 00:00:00 2001
From: iza <59828082+izadoesdev@users.noreply.github.com>
Date: Sun, 2 Aug 2026 13:39:13 +0300
Subject: [PATCH 1/3] fix(basket): preserve retries after interrupted admission

---
 apps/basket/src/lib/event-service.ts          | 13 +----
 apps/basket/src/lib/security.test.ts          | 30 +++++++++--
 apps/basket/src/lib/security.ts               |  7 ++-
 apps/basket/src/lib/structured-errors.test.ts | 15 ++++++
 apps/basket/src/lib/structured-errors.ts      | 25 +++++++++
 apps/basket/src/routes/basket.ts              | 16 +++---
 apps/basket/src/routes/integration.test.ts    | 54 ++++++++++++++++---
 7 files changed, 128 insertions(+), 32 deletions(-)

diff --git a/apps/basket/src/lib/event-service.ts b/apps/basket/src/lib/event-service.ts
index b0abb5d4d..723c8f1f3 100644
--- a/apps/basket/src/lib/event-service.ts
+++ b/apps/basket/src/lib/event-service.ts
@@ -16,6 +16,7 @@ import {
 	reserveDuplicateBatch,
 	shouldAnonymizeVisitorIds,
 } from "@lib/security";
+import { deliveryUnavailable } from "@lib/structured-errors";
 import { record } from "@lib/tracing";
 import { extractTrustedClientIp, getGeo } from "@utils/ip-geo";
 import { parseUserAgent } from "@utils/user-agent";
@@ -27,7 +28,6 @@ import {
 	validateSessionId,
 } from "@utils/validation";
 import { randomUUIDv7 } from "bun";
-import { createError } from "evlog";
 import { useLogger } from "evlog/elysia";
 import { createHash } from "node:crypto";
 
@@ -127,17 +127,6 @@ export function stableBatchDeliveryId(
 		.digest("hex");
 }
 
-function deliveryUnavailable(cause: unknown) {
-	return createError({
-		code: "basket.DELIVERY_UNAVAILABLE",
-		message: "Analytics delivery temporarily unavailable",
-		status: 503,
-		why: "Databuddy could not durably accept the event.",
-		fix: "Retry the same event after a short delay.",
-		cause: cause instanceof Error ? cause : new Error(String(cause)),
-	});
-}
-
 function directEventIdentity(eventId: unknown, generateFn: () => string) {
 	const sourceEventId =
 		typeof eventId === "string" && eventId.trim()
diff --git a/apps/basket/src/lib/security.test.ts b/apps/basket/src/lib/security.test.ts
index 361cc87a3..70e962cfa 100644
--- a/apps/basket/src/lib/security.test.ts
+++ b/apps/basket/src/lib/security.test.ts
@@ -174,7 +174,7 @@ describe("duplicate reservations", () => {
 			"dedup:track:evt_1",
 			expect.stringMatching(/^pending:/),
 			"EX",
-			120,
+			30,
 			"NX"
 		);
 	});
@@ -248,7 +248,7 @@ describe("duplicate reservations", () => {
 			expect.stringMatching(/^ambiguous-pending:\d+:/),
 			"delivered",
 			"ambiguous",
-			120,
+			30,
 			expect.any(Number),
 			"ambiguous-pending:",
 			86_400,
@@ -322,11 +322,35 @@ describe("duplicate reservations", () => {
 			"dedup:track:stable-delivery-id",
 			expect.stringMatching(/^pending:/),
 			"EX",
-			120,
+			30,
 			"NX"
 		);
 	});
 
+	test("recovers a stale pending lease without shortening delivered retention", async () => {
+		mockRedisSet.mockResolvedValueOnce(null).mockResolvedValueOnce("OK");
+		mockRedisGet.mockResolvedValue("pending:crashed-owner");
+
+		expect(await reserveDuplicate("evt_1", "track")).toEqual({
+			duplicate: false,
+			retryable: true,
+		});
+
+		expect(await reserveDuplicate("evt_1", "track")).toMatchObject({
+			deliveredTtl: 86_400,
+			duplicate: false,
+			key: "dedup:track:evt_1",
+			token: expect.stringMatching(/^pending:/),
+		});
+		expect(mockRedisSet.mock.calls[1]).toEqual([
+			"dedup:track:evt_1",
+			expect.stringMatching(/^pending:/),
+			"EX",
+			30,
+			"NX",
+		]);
+	});
+
 	test("retries a Redis error before acquiring the reservation", async () => {
 		mockRedisSet
 			.mockRejectedValueOnce(new Error("stale connection"))
diff --git a/apps/basket/src/lib/security.ts b/apps/basket/src/lib/security.ts
index 2411a21bf..67ae505e1 100644
--- a/apps/basket/src/lib/security.ts
+++ b/apps/basket/src/lib/security.ts
@@ -8,7 +8,10 @@ import { useLogger } from "evlog/elysia";
 
 const EXIT_EVENT_TTL = 172_800;
 const STANDARD_EVENT_TTL = 86_400;
-const PENDING_DEDUP_TTL = 120;
+// Pending ownership only guards an in-flight admission. It must outlive the
+// 20-second Railway shutdown budget while still letting a crashed owner expire
+// before client retries are suppressed for minutes.
+const PENDING_DEDUP_TTL = 30;
 const DEDUP_RETRY_DELAY_MS = 25;
 export const DEDUP_RESERVATION_TIMEOUT_MS = 750;
 const DEDUP_FAILURE_COOLDOWN_MS = 5000;
@@ -393,7 +396,7 @@ export function reserveDuplicate(
 			if (error instanceof DeduplicationDeadlineError) {
 				// Redis commands cannot be cancelled. If SET NX succeeds after the
 				// caller returns retryable, conditionally reconcile only this attempt's
-				// token so it cannot strand an ownerless 120-second reservation.
+				// token so it cannot strand an ownerless pending reservation.
 				reservationOperation
 					.then(async (state) => {
 						if (state === "ambiguous-acquired") {
diff --git a/apps/basket/src/lib/structured-errors.test.ts b/apps/basket/src/lib/structured-errors.test.ts
index 2b87c36ba..67c4befb0 100644
--- a/apps/basket/src/lib/structured-errors.test.ts
+++ b/apps/basket/src/lib/structured-errors.test.ts
@@ -4,6 +4,8 @@ import {
 	basketErrors,
 	buildBasketErrorPayload,
 	createIngestSchemaValidationError,
+	deliveryUnavailable,
+	isDeliveryUnavailableError,
 	isIngestSchemaValidationError,
 	rethrowOrWrap,
 } from "./structured-errors";
@@ -86,6 +88,19 @@ describe("IngestSchemaValidationError", () => {
 	});
 });
 
+describe("deliveryUnavailable", () => {
+	test("creates a retryable structured 503", () => {
+		const error = deliveryUnavailable(new Error("Redis unavailable"));
+
+		expect(error).toMatchObject({
+			code: "basket.DELIVERY_UNAVAILABLE",
+			status: 503,
+		});
+		expect(isDeliveryUnavailableError(error)).toBe(true);
+		expect(isDeliveryUnavailableError(new Error("nope"))).toBe(false);
+	});
+});
+
 // ── rethrowOrWrap ──
 
 describe("rethrowOrWrap", () => {
diff --git a/apps/basket/src/lib/structured-errors.ts b/apps/basket/src/lib/structured-errors.ts
index c6bc3e819..7c78ac2c5 100644
--- a/apps/basket/src/lib/structured-errors.ts
+++ b/apps/basket/src/lib/structured-errors.ts
@@ -227,6 +227,31 @@ export function createIngestSchemaValidationError(
 	return Object.assign(err, { issues });
 }
 
+/**
+ * A request must not report success while its telemetry could not be durably
+ * admitted. Callers return this 503 so SDKs and queueing clients retry.
+ */
+export function deliveryUnavailable(cause: unknown) {
+	return createError({
+		code: "basket.DELIVERY_UNAVAILABLE",
+		message: "Analytics delivery temporarily unavailable",
+		status: 503,
+		why: "Databuddy could not durably accept the event.",
+		fix: "Retry the same event after a short delay.",
+		cause: cause instanceof Error ? cause : new Error(String(cause)),
+	});
+}
+
+export function isDeliveryUnavailableError(
+	error: unknown
+): error is EvlogError {
+	return (
+		error instanceof EvlogError &&
+		error.status === 503 &&
+		error.code === "basket.DELIVERY_UNAVAILABLE"
+	);
+}
+
 export function isIngestSchemaValidationError(
 	error: unknown
 ): error is IngestSchemaValidationError {
diff --git a/apps/basket/src/routes/basket.ts b/apps/basket/src/routes/basket.ts
index 34e7d95ec..a4a212432 100644
--- a/apps/basket/src/routes/basket.ts
+++ b/apps/basket/src/routes/basket.ts
@@ -42,6 +42,8 @@ import {
 import {
 	basketErrors,
 	createIngestSchemaValidationError,
+	deliveryUnavailable,
+	isDeliveryUnavailableError,
 	rethrowOrWrap,
 } from "@lib/structured-errors";
 import { record } from "@lib/tracing";
@@ -621,13 +623,13 @@ const app = new Elysia()
 						});
 					}
 				} catch (error) {
-					log.error(error instanceof Error ? error : new Error(String(error)));
-					results.push({
-						status: "error",
-						message: "Processing failed",
-						code: "EVENT_PROCESSING_FAILED",
-						eventType,
-					});
+					if (isDeliveryUnavailableError(error)) {
+						throw error;
+					}
+					const processingError =
+						error instanceof Error ? error : new Error(String(error));
+					log.error(processingError);
+					throw deliveryUnavailable(processingError);
 				}
 			}
 
diff --git a/apps/basket/src/routes/integration.test.ts b/apps/basket/src/routes/integration.test.ts
index 4de96b340..4bceaf552 100644
--- a/apps/basket/src/routes/integration.test.ts
+++ b/apps/basket/src/routes/integration.test.ts
@@ -13,6 +13,7 @@ const {
 	mockInsertIndividualVitals,
 	mockInsertErrorSpans,
 	mockInsertCustomEvents,
+	mockGetGeo,
 	mockCheckAutumnUsage,
 	mockGetApiKeyFromHeader,
 	mockHasKeyScope,
@@ -63,6 +64,14 @@ const {
 		mockInsertIndividualVitals: vi.fn(() => Promise.resolve()),
 		mockInsertErrorSpans: vi.fn(() => Promise.resolve()),
 		mockInsertCustomEvents: vi.fn(() => Promise.resolve()),
+		mockGetGeo: vi.fn(() =>
+			Promise.resolve({
+				anonymizedIP: "abc123",
+				country: "US",
+				region: "CA",
+				city: "SF",
+			})
+		),
 		mockCheckAutumnUsage: vi.fn(() => Promise.resolve({ allowed: true })),
 		mockGetApiKeyFromHeader: vi.fn(() => Promise.resolve(defaultApiKey)),
 		mockHasKeyScope: vi.fn(() => true),
@@ -114,14 +123,7 @@ vi.mock("@lib/security", () => ({
 }));
 
 vi.mock("@utils/ip-geo", () => ({
-	getGeo: vi.fn(() =>
-		Promise.resolve({
-			anonymizedIP: "abc123",
-			country: "US",
-			region: "CA",
-			city: "SF",
-		})
-	),
+	getGeo: mockGetGeo,
 	extractIpFromRequest: vi.fn(() => "1.2.3.4"),
 	extractTrustedClientIp: vi.fn(() => "1.2.3.4"),
 	getVisitorCountryForAutoMode: vi.fn((events: Array<{ anonymizeVisitorIds?: unknown }>) =>
@@ -578,6 +580,42 @@ describe("POST /batch", () => {
 		expect(await json(res)).toMatchObject({ batch: true, processed: 1 });
 	});
 
+	test("returns 503 instead of accepting an event that could not be prepared", async () => {
+		mockInsertTrackEventsBatch.mockClear();
+		mockInsertOutgoingLinksBatch.mockClear();
+		mockGetGeo
+			.mockResolvedValueOnce({
+				anonymizedIP: "abc123",
+				country: "US",
+				region: "CA",
+				city: "SF",
+			})
+			.mockRejectedValueOnce(new Error("GeoIP unavailable"));
+
+		const result = await post(basketApp, "/batch", [
+			{
+				type: "track",
+				eventId: "evt_1",
+				name: "pageview",
+				path: "https://example.com/a",
+			},
+			{
+				type: "track",
+				eventId: "evt_2",
+				name: "click",
+				path: "https://example.com/b",
+			},
+		]);
+
+		expect(result.status).toBe(503);
+		expect(await json(result)).toMatchObject({
+			code: "basket.DELIVERY_UNAVAILABLE",
+			retryable: true,
+		});
+		expect(mockInsertTrackEventsBatch).not.toHaveBeenCalled();
+		expect(mockInsertOutgoingLinksBatch).not.toHaveBeenCalled();
+	});
+
 	test("not an array → 400", async () => {
 		const res = await post(basketApp, "/batch", { not: "array" });
 		expect(res.status).toBe(400);

From 9c3ff98d2785167fc31ed82deb8713effd82ff20 Mon Sep 17 00:00:00 2001
From: iza <59828082+izadoesdev@users.noreply.github.com>
Date: Sun, 2 Aug 2026 14:50:08 +0300
Subject: [PATCH 2/3] fix(basket): reserve events at delivery handoff

---
 .../src/lib/event-service.delivery.test.ts    | 76 +++++++++++++++++++
 apps/basket/src/lib/event-service.ts          | 70 +++++++++--------
 apps/basket/src/lib/security.test.ts          |  9 +--
 apps/basket/src/lib/security.ts               |  6 +-
 apps/basket/src/lib/structured-errors.test.ts |  3 -
 apps/basket/src/lib/structured-errors.ts      | 10 ---
 apps/basket/src/routes/basket.ts              | 16 ++--
 apps/basket/src/routes/integration.test.ts    | 38 +++++++++-
 8 files changed, 167 insertions(+), 61 deletions(-)

diff --git a/apps/basket/src/lib/event-service.delivery.test.ts b/apps/basket/src/lib/event-service.delivery.test.ts
index a35edf5be..303b59aeb 100644
--- a/apps/basket/src/lib/event-service.delivery.test.ts
+++ b/apps/basket/src/lib/event-service.delivery.test.ts
@@ -181,6 +181,82 @@ describe("event-service producer handoff", () => {
 		expect(mockMarkDuplicateReservationDelivered).toHaveBeenCalledOnce();
 	});
 
+	test("reserves track events only after enrichment completes", async () => {
+		let resolveGeo!: (value: {
+			anonymizedIP: string;
+			city: string;
+			country: string;
+			region: string;
+		}) => void;
+		mockGetGeo.mockImplementationOnce(
+			() =>
+				new Promise((resolve) => {
+					resolveGeo = resolve;
+				})
+		);
+
+		const pending = insertTrackEvent(
+			{ eventId: "evt_1", name: "pageview", path: "/" },
+			"ws_1",
+			"Mozilla/5.0",
+			"1.2.3.4",
+			new Request("https://basket.example/px.jpg")
+		);
+
+		expect(mockGetGeo).toHaveBeenCalledOnce();
+		expect(mockReserveDuplicate).not.toHaveBeenCalled();
+		resolveGeo({
+			anonymizedIP: "1.2.3.0",
+			city: "San Francisco",
+			country: "US",
+			region: "CA",
+		});
+		await pending;
+
+		expect(mockReserveDuplicate.mock.invocationCallOrder[0]).toBeLessThan(
+			mockSend.mock.invocationCallOrder[0] as number
+		);
+	});
+
+	test("reserves outgoing links only after GeoIP enrichment completes", async () => {
+		let resolveGeo!: (value: {
+			anonymizedIP: string;
+			city: string;
+			country: string;
+			region: string;
+		}) => void;
+		mockGetGeo.mockImplementationOnce(
+			() =>
+				new Promise((resolve) => {
+					resolveGeo = resolve;
+				})
+		);
+
+		const pending = insertOutgoingLink(
+			{
+				anonymizeVisitorIds: "auto",
+				eventId: "evt_link_1",
+				href: "https://external.example",
+			},
+			"ws_1",
+			new Request("https://basket.example/px.jpg")
+		);
+
+		expect(mockGetGeo).toHaveBeenCalledOnce();
+		expect(mockReserveDuplicate).not.toHaveBeenCalled();
+		resolveGeo({
+			anonymizedIP: "1.2.3.0",
+			city: "San Francisco",
+			country: "US",
+			region: "CA",
+		});
+		await pending;
+
+		expect(mockReserveDuplicate.mock.invocationCallOrder[0]).toBeLessThan(
+			mockSend.mock.invocationCallOrder[0] as number
+		);
+	});
+
 	test("propagates outgoing-link producer admission failures", async () => {
 		const error = new Error("buffer full");
 		mockRunPromise.mockRejectedValueOnce(error);
diff --git a/apps/basket/src/lib/event-service.ts b/apps/basket/src/lib/event-service.ts
index 723c8f1f3..31259f3ef 100644
--- a/apps/basket/src/lib/event-service.ts
+++ b/apps/basket/src/lib/event-service.ts
@@ -238,20 +238,7 @@ export function insertTrackEvent(
 		);
 
 		const deliveryId = stableAnalyticsEventId(clientId, "track", sourceEventId);
-		const reservation = await reserveDuplicate(
-			deliveryId,
-			"track",
-			storedEventId
-		);
-		if (reservation.duplicate) {
-			return;
-		}
-		if (reservation.retryable) {
-			throw deliveryUnavailable(
-				new Error("A concurrent attempt owns this analytics event")
-			);
-		}
-
+		let trackEvent: EventsInsert;
 		try {
 			const geoData = await getGeo(ip, request);
 			const trustedCountry = extractTrustedClientIp(request)
@@ -287,7 +274,7 @@ export function insertTrackEvent(
 
 			const now = Date.now();
 
-			const trackEvent = buildTrackEvent(trackData, {
+			trackEvent = buildTrackEvent(trackData, {
 				clientId,
 				eventId: sourceEventId,
 				anonymousId,
@@ -295,7 +282,25 @@ export function insertTrackEvent(
 				ua,
 				now,
 			});
+		} catch (error) {
+			throw deliveryUnavailable(error);
+		}
+
+		const reservation = await reserveDuplicate(
+			deliveryId,
+			"track",
+			storedEventId
+		);
+		if (reservation.duplicate) {
+			return;
+		}
+		if (reservation.retryable) {
+			throw deliveryUnavailable(
+				new Error("A concurrent attempt owns this analytics event")
+			);
+		}
 
+		try {
 			await runPromise(
 				send("analytics-events", trackEvent, undefined, {
 					allowDirectFallback: reservation.ambiguous !== true,
@@ -327,20 +332,7 @@ export function insertOutgoingLink(
 			"outgoing_link",
 			sourceEventId
 		);
-		const reservation = await reserveDuplicate(
-			deliveryId,
-			"outgoing_link",
-			storedEventId
-		);
-		if (reservation.duplicate) {
-			return;
-		}
-		if (reservation.retryable) {
-			throw deliveryUnavailable(
-				new Error("A concurrent attempt owns this analytics event")
-			);
-		}
-
+		let outgoingLinkEvent: OutgoingLinksInsert;
 		try {
 			log.set({
 				event: {
@@ -363,7 +355,7 @@ export function insertOutgoingLink(
 			);
 			const salt = anonymizeVisitorIds ? await getDailySalt() : undefined;
 
-			const outgoingLinkEvent: OutgoingLinksInsert = {
+			outgoingLinkEvent = {
 				id: deliveryId,
 				client_id: clientId,
 				anonymous_id: applyVisitorIdPrivacy(
@@ -380,7 +372,25 @@ export function insertOutgoingLink(
 				timestamp:
 					typeof linkData.timestamp === "number" ? linkData.timestamp : now,
 			};
+		} catch (error) {
+			throw deliveryUnavailable(error);
+		}
+
+		const reservation = await reserveDuplicate(
+			deliveryId,
+			"outgoing_link",
+			storedEventId
+		);
+		if (reservation.duplicate) {
+			return;
+		}
+		if (reservation.retryable) {
+			throw deliveryUnavailable(
+				new Error("A concurrent attempt owns this analytics event")
+			);
+		}
 
+		try {
 			await runPromise(
 				send("analytics-outgoing-links", outgoingLinkEvent, undefined, {
 					allowDirectFallback: reservation.ambiguous !== true,
diff --git a/apps/basket/src/lib/security.test.ts b/apps/basket/src/lib/security.test.ts
index 70e962cfa..95c7ae6ca 100644
--- a/apps/basket/src/lib/security.test.ts
+++ b/apps/basket/src/lib/security.test.ts
@@ -327,7 +327,7 @@ describe("duplicate reservations", () => {
 		);
 	});
 
-	test("recovers a stale pending lease without shortening delivered retention", async () => {
+	test("allows a later retry after a stale pending lease", async () => {
 		mockRedisSet.mockResolvedValueOnce(null).mockResolvedValueOnce("OK");
 		mockRedisGet.mockResolvedValue("pending:crashed-owner");
 
@@ -342,13 +342,6 @@ describe("duplicate reservations", () => {
 			key: "dedup:track:evt_1",
 			token: expect.stringMatching(/^pending:/),
 		});
-		expect(mockRedisSet.mock.calls[1]).toEqual([
-			"dedup:track:evt_1",
-			expect.stringMatching(/^pending:/),
-			"EX",
-			30,
-			"NX",
-		]);
 	});
 
 	test("retries a Redis error before acquiring the reservation", async () => {
diff --git a/apps/basket/src/lib/security.ts b/apps/basket/src/lib/security.ts
index 67ae505e1..5e34fed76 100644
--- a/apps/basket/src/lib/security.ts
+++ b/apps/basket/src/lib/security.ts
@@ -8,9 +8,9 @@ import { useLogger } from "evlog/elysia";
 
 const EXIT_EVENT_TTL = 172_800;
 const STANDARD_EVENT_TTL = 86_400;
-// Pending ownership only guards an in-flight admission. It must outlive the
-// 20-second Railway shutdown budget while still letting a crashed owner expire
-// before client retries are suppressed for minutes.
+// Reservations begin immediately before the bounded producer handoff. This
+// exceeds the 20-second Railway shutdown budget while letting a crashed owner
+// expire before client retries are suppressed for minutes.
 const PENDING_DEDUP_TTL = 30;
 const DEDUP_RETRY_DELAY_MS = 25;
 export const DEDUP_RESERVATION_TIMEOUT_MS = 750;
diff --git a/apps/basket/src/lib/structured-errors.test.ts b/apps/basket/src/lib/structured-errors.test.ts
index 67c4befb0..2fa714aaa 100644
--- a/apps/basket/src/lib/structured-errors.test.ts
+++ b/apps/basket/src/lib/structured-errors.test.ts
@@ -5,7 +5,6 @@ import {
 	buildBasketErrorPayload,
 	createIngestSchemaValidationError,
 	deliveryUnavailable,
-	isDeliveryUnavailableError,
 	isIngestSchemaValidationError,
 	rethrowOrWrap,
 } from "./structured-errors";
@@ -96,8 +95,6 @@ describe("deliveryUnavailable", () => {
 			code: "basket.DELIVERY_UNAVAILABLE",
 			status: 503,
 		});
-		expect(isDeliveryUnavailableError(error)).toBe(true);
-		expect(isDeliveryUnavailableError(new Error("nope"))).toBe(false);
 	});
 });
 
diff --git a/apps/basket/src/lib/structured-errors.ts b/apps/basket/src/lib/structured-errors.ts
index 7c78ac2c5..810b5c99b 100644
--- a/apps/basket/src/lib/structured-errors.ts
+++ b/apps/basket/src/lib/structured-errors.ts
@@ -242,16 +242,6 @@ export function deliveryUnavailable(cause: unknown) {
 	});
 }
 
-export function isDeliveryUnavailableError(
-	error: unknown
-): error is EvlogError {
-	return (
-		error instanceof EvlogError &&
-		error.status === 503 &&
-		error.code === "basket.DELIVERY_UNAVAILABLE"
-	);
-}
-
 export function isIngestSchemaValidationError(
 	error: unknown
 ): error is IngestSchemaValidationError {
diff --git a/apps/basket/src/routes/basket.ts b/apps/basket/src/routes/basket.ts
index a4a212432..a9c236fb0 100644
--- a/apps/basket/src/routes/basket.ts
+++ b/apps/basket/src/routes/basket.ts
@@ -43,7 +43,6 @@ import {
 	basketErrors,
 	createIngestSchemaValidationError,
 	deliveryUnavailable,
-	isDeliveryUnavailableError,
 	rethrowOrWrap,
 } from "@lib/structured-errors";
 import { record } from "@lib/tracing";
@@ -520,7 +519,10 @@ const app = new Elysia()
 			};
 
 			for (const event of body) {
-				const eventType = event.type || "track";
+				const isEventObject =
+					event !== null && typeof event === "object" && !Array.isArray(event);
+				const eventType = isEventObject ? event.type || "track" : "track";
+				const eventId = isEventObject ? event.eventId : undefined;
 
 				try {
 					if (eventType === "track") {
@@ -549,7 +551,7 @@ const app = new Elysia()
 								batchSchemaItemFailure(
 									parseResult.error.issues,
 									eventType,
-									event.eventId
+									eventId
 								)
 							);
 							continue;
@@ -594,7 +596,7 @@ const app = new Elysia()
 								batchSchemaItemFailure(
 									parseResult.error.issues,
 									eventType,
-									event.eventId
+									eventId
 								)
 							);
 							continue;
@@ -623,7 +625,11 @@ const app = new Elysia()
 						});
 					}
 				} catch (error) {
-					if (isDeliveryUnavailableError(error)) {
+					if (
+						error instanceof EvlogError &&
+						error.status === 503 &&
+						error.code === "basket.DELIVERY_UNAVAILABLE"
+					) {
 						throw error;
 					}
 					const processingError =
diff --git a/apps/basket/src/routes/integration.test.ts b/apps/basket/src/routes/integration.test.ts
index 4bceaf552..4a6d3957f 100644
--- a/apps/basket/src/routes/integration.test.ts
+++ b/apps/basket/src/routes/integration.test.ts
@@ -89,6 +89,7 @@ vi.mock("evlog/elysia", () => ({
 vi.mock("@lib/tracing", () => ({
 	record: (_n: string, fn: Function) => Promise.resolve().then(() => fn()),
 	captureError: noop,
+	mergeWideEvent: noop,
 }));
 
 vi.mock("@lib/request-validation", () => ({
@@ -616,6 +617,39 @@ describe("POST /batch", () => {
 		expect(mockInsertOutgoingLinksBatch).not.toHaveBeenCalled();
 	});
 
+	test("keeps malformed items as schema failures without dropping valid events", async () => {
+		mockInsertTrackEventsBatch.mockClear();
+		mockInsertOutgoingLinksBatch.mockClear();
+
+		const result = await post(basketApp, "/batch", [
+			{
+				type: "track",
+				eventId: "evt_1",
+				name: "pageview",
+				path: "https://example.com/a",
+			},
+			null,
+		]);
+
+		expect(result.status).toBe(200);
+		expect(await json(result)).toMatchObject({
+			status: "partial",
+			batch: true,
+			processed: 2,
+			batched: { track: 1, outgoing_link: 0 },
+			results: [
+				{ status: "success", type: "track", eventId: "evt_1" },
+				{
+					status: "error",
+					code: "INVALID_EVENT_SCHEMA",
+					eventType: "track",
+				},
+			],
+		});
+		expect(mockInsertTrackEventsBatch).toHaveBeenCalledOnce();
+		expect(mockInsertOutgoingLinksBatch).toHaveBeenCalledOnce();
+	});
+
 	test("not an array → 400", async () => {
 		const res = await post(basketApp, "/batch", { not: "array" });
 		expect(res.status).toBe(400);
@@ -632,7 +666,7 @@ describe("POST /batch", () => {
 		expect(res.status).toBe(400);
 	});
 
-	test("mixed valid + unknown types → partial results", async () => {
+	test("mixed valid + invalid types → partial results", async () => {
 		const res = await post(basketApp, "/batch", [
 			{
 				type: "track",
@@ -640,7 +674,7 @@ describe("POST /batch", () => {
 				name: "pageview",
 				path: "https://example.com/a",
 			},
-			{ type: "bogus_type" },
+			{ type: 1 },
 		]);
 		expect(res.status).toBe(200);
 		const body = await json(res);

From 188d967017661f49361d15d4f4ed2e59b8eda5f0 Mon Sep 17 00:00:00 2001
From: iza <59828082+izadoesdev@users.noreply.github.com>
Date: Sun, 2 Aug 2026 14:50:20 +0300
Subject: [PATCH 3/3] fix(ai): keep Redis mock compatible

---
 packages/ai/src/ai/agents/cache.test.ts | 1 +
 1 file changed, 1 insertion(+)

diff --git a/packages/ai/src/ai/agents/cache.test.ts b/packages/ai/src/ai/agents/cache.test.ts
index ba742d2b3..03fe4864c 100644
--- a/packages/ai/src/ai/agents/cache.test.ts
+++ b/packages/ai/src/ai/agents/cache.test.ts
@@ -169,6 +169,7 @@ vi.mock("@databuddy/redis", () => ({
 	redis: mockRedisClient,
 	setActiveStream: vi.fn(async () => undefined),
 	setCachedLink: vi.fn(async () => undefined),
+	setCachedLinkIfAbsent: vi.fn(async () => true),
 	setCachedLinkNotFound: vi.fn(async () => undefined),
 	shouldRecordClick: vi.fn(async () => true),
 	shutdownRedis: vi.fn(async () => undefined),