fix: harden analytics delivery

.agents/skills/databuddy-internal/SKILL.md

@@ -57,9 +57,11 @@ Keep additions **minimal**: one bullet, a new `rg` hint, or a routing note—eno
 - Slack-reachable shared packages (`@databuddy/ai`, `@databuddy/rpc`) must not import `evlog/elysia`; use host-injected request logger providers from the API and plain evlog fallbacks elsewhere.
 - AI link tools must assign link folders by existing folder `id` or `slug` only; folder names are display text and must not be used for routing or dedupe.
 - `apps/basket`: ingest and LLM tracking service, Elysia app on port `4000`
+- A legacy Basket dedupe value of `"1"` is an admission reservation, not delivery confirmation; claim or retry it rather than suppressing the event.
 - `apps/docs`: Next.js + Fumadocs docs app on port `3005`
 - When a user drops a prototype, remove only prototype-specific wiring and preserve the existing product surfaces it temporarily reused.
 - `apps/links`: redirect/link service
+- `apps/links` tests share Bun `mock.module` state; mocks of `../lib/logging` must include every export used by later test files, including `setAttributes` for producer imports.
 - `apps/uptime`: uptime monitoring service
 - `apps/uptime` BullMQ worker concurrency defaults high for Bun async I/O; do not lower it just because `10_000` looks large. Verify downstream saturation or lock/timeout evidence first.
 - Public status pages render from `apps/status`; `apps/dashboard` owns status-page management/config UI only. When cleaning public status UX, update shared `@databuddy/ui/uptime` pieces or `apps/status` wrappers instead of redesigning dashboard-only route remnants.

apps/basket/package.json

@@ -24,7 +24,6 @@
     "@maxmind/geoip2-node": "^6.3.4",
     "@types/ua-parser-js": "^0.7.39",
     "async-mutex": "^0.5.0",
-    "effect": "^4.0.0-beta.90",
     "elysia": "catalog:",
     "evlog": "catalog:",
     "kafkajs": "^2.2.4",

apps/basket/src/index.ts

@@ -8,7 +8,7 @@ import {
 import { shutdownPostgres } from "@databuddy/db";
 import { clickHouse } from "@databuddy/db/clickhouse";
 import { getRedisCache } from "@databuddy/redis/redis";
-import { disconnect, disposeRuntime, runPromise } from "@lib/producer";
+import { disconnect, ShutdownDrainError } from "@lib/producer";
 import { Kafka } from "kafkajs";
 import { databuddyEvlogRedaction } from "@databuddy/shared/evlog-redaction";
 import {
@@ -79,13 +79,28 @@ async function gracefulShutdown(signal: string, exitCode = 0) {
 				error_message: error instanceof Error ? error.message : String(error),
 			});
 		const { shutdownRedis } = await import("@databuddy/redis");
+		// Wait for acknowledged delivery before tearing down its dependencies.
+		try {
+			await disconnect();
+		} catch (error) {
+			finalExitCode = 1;
+			if (error instanceof ShutdownDrainError) {
+				log.error({
+					lifecycle: "producerDrain",
+					error_message:
+						"Basket producer drain timed out waiting for in-flight delivery",
+					in_flight: error.inFlight,
+					drain_timeout_ms: error.deadlineMs,
+				});
+			} else {
+				logErr("producerDrain")(error);
+			}
+		}
 		await Promise.all([
 			shutdownRedis().catch(logErr("redisShutdown")),
 			shutdownPostgres().catch(logErr("postgresShutdown")),
-			flushBatchedAxiomDrain().catch(logErr("drainFlush")),
-			runPromise(disconnect).catch(logErr("shutdown")),
-			disposeRuntime().catch(logErr("runtimeDispose")),
 		]);
+		await flushBatchedAxiomDrain().catch(logErr("drainFlush"));
 		closeGeoIPReader();
 	} catch (error) {
 		finalExitCode = 1;