From 8f20b2ef1b2e67e8dcd31c64978615623adea014 Mon Sep 17 00:00:00 2001 From: "Sven (AAMOS AI)" Date: Mon, 10 Aug 2026 05:47:43 +0700 Subject: [PATCH] feat(ops): /ops/v1/summary med ai/aktivering/engagemang/betalning/jobb/sakerhet + canary MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - Nya toppnivåblock i /ops/v1/summary: ai_scan, aktivering, engagemang, betalning, jobb, sakerhet. Alla värden läses från DB/cache; null där data saknas, inga påhittade värden. - AI-kostnad i USD mikrocent (native); intäkter fortsatt SEK-öre. - product_analytics_events är källa för volym, latens, lyckandegrad, felfrekvens, aktivering, retention och engagemang. - Nya händelser: scan_started (API) och scan_failed/scan_completed med latencyMs + felkod (worker). latencyMs flödar nu in i scan_completed. - Safety canary-jobb varje timme: re-härleder allergener för alla recept, räknar överifierade publika recept och food-safety-lint; skriver EN rad till ops_safety_canary. Endpointen läser endast sista raden. - Cache-refresh-jobb var 60 s skriver hela summariet till Redis; endpointen serverar cachen med 503 vid cache-miss. - Bearer-token-skydd med OPS_TOKEN; HTTPS-tvång i produktion; ingen PII. - Tester för endpoint, auth, cache-miss och safety canary. --- apps/api/src/config.ts | 9 + apps/api/src/lib/helpers.ts | 32 +- apps/api/src/routes/ops.ts | 62 +++ apps/api/src/routes/scans.ts | 16 +- apps/api/src/server.ts | 2 + apps/api/test/ops.test.ts | 126 ++++++ apps/api/test/setup-env.ts | 4 +- apps/worker/package.json | 2 + apps/worker/src/context.ts | 10 +- apps/worker/src/index.ts | 40 +- apps/worker/src/processors/ops-summary.ts | 77 ++++ apps/worker/src/processors/safety-canary.ts | 185 +++++++++ apps/worker/src/processors/scans.ts | 63 ++- apps/worker/test/safety-canary.test.ts | 35 ++ .../migrations/0023_ops_safety_canary.sql | 12 + infrastructure/migrations/meta/_journal.json | 9 +- packages/analytics/src/builders.ts | 1 + packages/database/package.json | 3 + packages/database/src/analytics.ts | 42 ++ packages/database/src/index.ts | 1 + packages/database/src/ops-summary.ts | 383 ++++++++++++++++++ packages/database/src/schema/index.ts | 1 + packages/database/src/schema/ops.ts | 17 + packages/shared-types/src/analytics.ts | 7 +- packages/shared-types/src/index.ts | 1 + packages/shared-types/src/ops.ts | 64 +++ pnpm-lock.yaml | 6 + 27 files changed, 1169 insertions(+), 41 deletions(-) create mode 100644 apps/api/src/routes/ops.ts create mode 100644 apps/api/test/ops.test.ts create mode 100644 apps/worker/src/processors/ops-summary.ts create mode 100644 apps/worker/src/processors/safety-canary.ts create mode 100644 apps/worker/test/safety-canary.test.ts create mode 100644 infrastructure/migrations/0023_ops_safety_canary.sql create mode 100644 packages/database/src/analytics.ts create mode 100644 packages/database/src/ops-summary.ts create mode 100644 packages/database/src/schema/ops.ts create mode 100644 packages/shared-types/src/ops.ts diff --git a/apps/api/src/config.ts b/apps/api/src/config.ts index cf45753..dad2d8f 100644 --- a/apps/api/src/config.ts +++ b/apps/api/src/config.ts @@ -52,6 +52,9 @@ const configSchema = z.object({ GEMINI_TIMEOUT_MS: z.coerce.number().int().default(60_000), GEMINI_DAILY_BUDGET_USD: z.coerce.number().default(0), + /** Strong, rotated ops token for /ops/v1/summary. Set via SSM in production. */ + OPS_TOKEN: z.string().min(1).default("dev-ops-token-change-me"), + APP_STORE_MODE: z.enum(["production", "sandbox"]).default("sandbox"), EMAIL_MODE: z.enum(["log", "smtp"]).default("log"), SMTP_HOST: z.string().default(""), @@ -98,6 +101,12 @@ export function loadConfig(env: NodeJS.ProcessEnv = process.env): AppConfig { console.error("SÄKERHETSSTOPP: AAMOS_MODE=gemini kräver GEMINI_API_KEY i produktion."); process.exit(1); } + if (!cfg.OPS_TOKEN || cfg.OPS_TOKEN.length < 32 || cfg.OPS_TOKEN.startsWith("dev-ops-token")) { + console.error( + "SÄKERHETSSTOPP: OPS_TOKEN måste vara en stark, icke-default hemlighet i produktion.", + ); + process.exit(1); + } } return cfg; } diff --git a/apps/api/src/lib/helpers.ts b/apps/api/src/lib/helpers.ts index fd83a57..fc1e0be 100644 --- a/apps/api/src/lib/helpers.ts +++ b/apps/api/src/lib/helpers.ts @@ -2,7 +2,6 @@ import { createHash, randomBytes, randomUUID } from "node:crypto"; import { and, eq } from "drizzle-orm"; import type { Database } from "@app/database"; import { schema } from "@app/database"; -import type { AnalyticsEvent } from "@app/analytics"; import type { DecayProfile } from "@app/inventory-engine"; import type { EventType } from "@app/shared-types"; import type { NewDomainEvent } from "@app/events"; @@ -118,34 +117,9 @@ export async function emitEvent( }); } -/** Track product analytics server-side if user opted in. */ -export async function trackProductAnalytics( - db: Database, - userId: string, - event: AnalyticsEvent, -): Promise { - const optedIn = await db - .select({ status: schema.userConsents.status }) - .from(schema.userConsents) - .where(and(eq(schema.userConsents.userId, userId), eq(schema.userConsents.kind, "product_analytics"))) - .limit(1); - if (optedIn[0] && optedIn[0].status !== "granted") return; - - await db.insert(schema.productAnalyticsEvents).values({ - occurredAt: event.occurredAt ? new Date(event.occurredAt) : new Date(), - receivedAt: new Date(), - eventName: event.name, - anonymousId: event.anonymousId ?? null, - sessionId: event.sessionId ?? null, - userId, - householdId: event.householdId ?? null, - appVersion: event.appVersion ?? null, - platform: event.platform ?? null, - locale: event.locale ?? null, - experimentVariant: event.experimentVariant ?? null, - properties: event.properties ?? {}, - }); -} +// Track product analytics server-side if user opted in. +// Implementation lives in @app/database so workers can reuse it. +export { trackProductAnalytics } from "@app/database/analytics"; /** Audit-logg (spec §56). */ export async function audit( diff --git a/apps/api/src/routes/ops.ts b/apps/api/src/routes/ops.ts new file mode 100644 index 0000000..2165637 --- /dev/null +++ b/apps/api/src/routes/ops.ts @@ -0,0 +1,62 @@ +import { timingSafeEqual } from "node:crypto"; +import type { FastifyInstance, FastifyReply, FastifyRequest } from "fastify"; +import { errors } from "../lib/errors.js"; + +const CACHE_KEY = "ops:summary:cache"; +const CACHE_TS_KEY = "ops:summary:computed_at"; + +function requireOpsToken(app: FastifyInstance) { + return async (req: FastifyRequest, reply: FastifyReply): Promise => { + if (app.config.NODE_ENV === "production" && req.protocol !== "https") { + return reply + .status(403) + .send({ error: { code: "HTTPS_REQUIRED", message: "Ops-endpointen kräver HTTPS." } }); + } + + const header = req.headers.authorization ?? ""; + const match = /^Bearer\s+(.+)$/.exec(header); + const provided = match?.[1] ?? ""; + const expected = app.config.OPS_TOKEN ?? ""; + + if (!expected || expected.length < 20 || provided.length !== expected.length) { + throw errors.unauthorized(); + } + + const ok = timingSafeEqual(Buffer.from(provided), Buffer.from(expected)); + if (!ok) throw errors.unauthorized(); + }; +} + +/** + * Operations summary endpoint: read-only, token-protected, served from cache. + * Heavy computation is done by the REFRESH_OPS_SUMMARY_CACHE worker job. + */ +export async function opsRoutes(app: FastifyInstance) { + const preHandler = [requireOpsToken(app)]; + + app.get("/ops/v1/summary", { preHandler }, async (_req, reply) => { + const [json, ts] = await app.redis.mget(CACHE_KEY, CACHE_TS_KEY); + if (!json) { + return reply.status(503).send({ + error: { + code: "CACHE_MISS", + message: "Sammanfattningen är inte färdigberäknad än. Vänta på nästa canary-/cache-jobb.", + }, + }); + } + + let summary: Record; + try { + summary = JSON.parse(json) as Record; + } catch { + return reply + .status(500) + .send({ error: { code: "CACHE_INVALID", message: "Cachen innehåller ogiltig JSON." } }); + } + + return reply.send({ + ...summary, + cached_at: ts ?? summary.as_of, + }); + }); +} diff --git a/apps/api/src/routes/scans.ts b/apps/api/src/routes/scans.ts index 970e90a..2580f68 100644 --- a/apps/api/src/routes/scans.ts +++ b/apps/api/src/routes/scans.ts @@ -1,6 +1,7 @@ import type { FastifyInstance } from "fastify"; import { and, eq, gt, isNull } from "drizzle-orm"; -import { markMilestone, schema } from "@app/database"; +import { markMilestone, schema, trackProductAnalytics } from "@app/database"; +import { scanStarted } from "@app/analytics"; import type { JobType, ScanType } from "@app/shared-types"; import { confirmScanInputSchema, createScanInputSchema, idParamSchema } from "@app/validation"; import { errors, parse } from "../lib/errors.js"; @@ -108,6 +109,16 @@ export async function scanRoutes(app: FastifyInstance) { jobType: job.jobType, correlationId: req.correlationId, }); + + await trackProductAnalytics(app.db, req.userId, { + ...scanStarted(), + householdId: job.householdId ?? undefined, + properties: { + scanType: job.scanType, + jobType: job.jobType, + }, + }); + return { ok: true, status: "queued" }; }); @@ -353,7 +364,8 @@ function extractProposals(job: { result: unknown }): ProposalItem[] { unit: it.unit ?? null, bestBeforeDate: it.bestBeforeDate ?? null, confidence: typeof it.confidence === "number" ? it.confidence : null, - requiresConfirmation: typeof it.requiresConfirmation === "boolean" ? it.requiresConfirmation : null, + requiresConfirmation: + typeof it.requiresConfirmation === "boolean" ? it.requiresConfirmation : null, })); } diff --git a/apps/api/src/server.ts b/apps/api/src/server.ts index 8090ed5..0144eab 100644 --- a/apps/api/src/server.ts +++ b/apps/api/src/server.ts @@ -30,6 +30,7 @@ import { adminWorkersRoutes } from "./routes/admin-workers.js"; import { analyticsRoutes } from "./routes/analytics.js"; import { onboardingRoutes } from "./routes/onboarding.js"; import { activationRoutes } from "./routes/activation.js"; +import { opsRoutes } from "./routes/ops.js"; declare module "fastify" { interface FastifyInstance { @@ -98,6 +99,7 @@ export async function buildServer(config: AppConfig) { await app.register(analyticsRoutes); await app.register(onboardingRoutes); await app.register(activationRoutes); + await app.register(opsRoutes); return app; } diff --git a/apps/api/test/ops.test.ts b/apps/api/test/ops.test.ts new file mode 100644 index 0000000..5015cc2 --- /dev/null +++ b/apps/api/test/ops.test.ts @@ -0,0 +1,126 @@ +import "./setup-env.js"; +import { describe, expect, it, beforeAll, afterAll } from "vitest"; +import { buildServer } from "../src/server.js"; +import { loadConfig } from "../src/config.js"; +import { createDatabase, schema } from "@app/database"; +import { computeOpsSummary, type OpsQueueSummary } from "@app/database/ops-summary"; + +const testDb = createDatabase(process.env.TEST_DATABASE_URL!); +const config = loadConfig(); + +const queueSummary: OpsQueueSummary = { + vantande: 2, + aktiva: 1, + misslyckade_24h: 0, + aldsta_vantande_sek: null, + workers_ok: true, +}; + +describe("/ops/v1/summary", () => { + let app: Awaited>; + + beforeAll(async () => { + app = await buildServer(config); + await app.ready(); + await cleanup(); + }); + + afterAll(async () => { + await cleanup(); + await app.close(); + await testDb.pool.end(); + }); + + async function cleanup() { + await testDb.db.delete(schema.opsSafetyCanary); + await testDb.db.delete(schema.productAnalyticsEvents); + await app.redis.del("ops:summary:cache", "ops:summary:computed_at"); + } + + it("avvisar anrop utan token", async () => { + const res = await app.inject({ method: "GET", url: "/ops/v1/summary" }); + expect(res.statusCode).toBe(401); + }); + + it("avvisar felaktig token", async () => { + const res = await app.inject({ + method: "GET", + url: "/ops/v1/summary", + headers: { authorization: "Bearer wrong-token" }, + }); + expect(res.statusCode).toBe(401); + }); + + it("returnerar 503 när cachen saknas", async () => { + await app.redis.del("ops:summary:cache", "ops:summary:computed_at"); + const res = await app.inject({ + method: "GET", + url: "/ops/v1/summary", + headers: { authorization: `Bearer ${config.OPS_TOKEN}` }, + }); + expect(res.statusCode).toBe(503); + const body = JSON.parse(res.body) as { error: { code: string } }; + expect(body.error.code).toBe("CACHE_MISS"); + }); + + it("serverar cachad summary med rätt block", async () => { + const now = new Date(); + await testDb.db.insert(schema.productAnalyticsEvents).values([ + { eventName: "scan_started", occurredAt: now, properties: {}, userId: null }, + { + eventName: "scan_completed", + occurredAt: now, + properties: { latencyMs: 120 }, + userId: null, + }, + { + eventName: "scan_completed", + occurredAt: now, + properties: { latencyMs: 240 }, + userId: null, + }, + { + eventName: "cooking_session_completed", + occurredAt: now, + properties: {}, + userId: null, + }, + ]); + await testDb.db.insert(schema.opsSafetyCanary).values({ + allergenInvariantBrott: 0, + overifieradeVisade: 3, + foodSafetyLintAvvisade7d: 1, + }); + + const summary = await computeOpsSummary({ + db: testDb.db, + budgetUsd: 0, + queueSummary, + }); + await app.redis.set("ops:summary:cache", JSON.stringify(summary), "EX", 60); + await app.redis.set("ops:summary:computed_at", summary.as_of, "EX", 60); + + const res = await app.inject({ + method: "GET", + url: "/ops/v1/summary", + headers: { authorization: `Bearer ${config.OPS_TOKEN}` }, + }); + + expect(res.statusCode).toBe(200); + const body = JSON.parse(res.body) as { + ai_scan: { scans_24h: number; latens_p50_ms: number | null }; + engagemang: { lagade_maltider_24h: number }; + sakerhet: { overifierade_visade: number; senaste_kontroll: string | null }; + jobb: { vantande: number; workers_ok: boolean }; + cached_at: string; + }; + expect(body.ai_scan.scans_24h).toBe(2); + expect(body.ai_scan.latens_p50_ms).toBeGreaterThan(0); + expect(body.engagemang.lagade_maltider_24h).toBe(1); + expect(body.sakerhet.overifierade_visade).toBe(3); + expect(body.sakerhet.senaste_kontroll).toBeTruthy(); + expect(body.jobb.vantande).toBe(2); + expect(body.jobb.workers_ok).toBe(true); + expect(body.cached_at).toBe(summary.as_of); + }); +}); diff --git a/apps/api/test/setup-env.ts b/apps/api/test/setup-env.ts index 94483c1..9af3964 100644 --- a/apps/api/test/setup-env.ts +++ b/apps/api/test/setup-env.ts @@ -8,8 +8,10 @@ process.env.AAMOS_MODE = "mock"; process.env.EMAIL_MODE = "log"; process.env.S3_MODE = "mock"; process.env.LOG_LEVEL = "error"; +process.env.OPS_TOKEN = "test-ops-token-not-for-production"; // Local test database fallback – tests still need a Postgres instance, but // the connection string is not a secret and the value is predictable. -process.env.TEST_DATABASE_URL ||= "postgres://app_user:app_dev_password@localhost:5432/cibello_test"; +process.env.TEST_DATABASE_URL ||= + "postgres://app_user:app_dev_password@localhost:5432/cibello_test"; process.env.DATABASE_URL = process.env.TEST_DATABASE_URL; diff --git a/apps/worker/package.json b/apps/worker/package.json index 244997f..cc596b6 100644 --- a/apps/worker/package.json +++ b/apps/worker/package.json @@ -15,6 +15,7 @@ }, "dependencies": { "@app/ai-contracts": "workspace:*", + "@app/analytics": "workspace:*", "@app/database": "workspace:*", "@app/events": "workspace:*", "@app/inventory-engine": "workspace:*", @@ -37,6 +38,7 @@ "tsup": { "noExternal": [ "@app/ai-contracts", + "@app/analytics", "@app/database", "@app/events", "@app/inventory-engine", diff --git a/apps/worker/src/context.ts b/apps/worker/src/context.ts index 61e3304..db87085 100644 --- a/apps/worker/src/context.ts +++ b/apps/worker/src/context.ts @@ -18,6 +18,7 @@ import type { Redis } from "ioredis"; export interface WorkerContext { db: Database; aamos: AamosClient; + redis?: Redis; /** Bygger läs-URL för lagrade bilder (samma signaturlogik som API:ts mock-S3). */ readUrl: (key: string) => string; apiBaseUrl: string; @@ -45,7 +46,9 @@ class RedisBudgetStore implements BudgetStore { private secondsUntilMidnightUtc(): number { const now = new Date(); - const midnight = new Date(Date.UTC(now.getUTCFullYear(), now.getUTCMonth(), now.getUTCDate() + 1)); + const midnight = new Date( + Date.UTC(now.getUTCFullYear(), now.getUTCMonth(), now.getUTCDate() + 1), + ); return Math.max(1, Math.floor((midnight.getTime() - now.getTime()) / 1000)); } } @@ -56,7 +59,9 @@ export function createContext(redis?: Redis): WorkerContext { throw new Error("Missing DATABASE_URL. Worker must connect explicitly to the app database."); } const { db, pool } = createDatabase(databaseUrl); - const budgetStore = redis ? new RedisBudgetStore(redis, `gemini:daily:budget:${new Date().toISOString().slice(0, 10)}`) : undefined; + const budgetStore = redis + ? new RedisBudgetStore(redis, `gemini:daily:budget:${new Date().toISOString().slice(0, 10)}`) + : undefined; const aamos = createAamosClient(undefined, { budgetStore }); const apiBaseUrl = process.env.API_BASE_URL ?? "http://localhost:4000"; const secret = process.env.ENTITLEMENT_SIGNING_SECRET ?? "dev-only-change-me-three"; @@ -64,6 +69,7 @@ export function createContext(redis?: Redis): WorkerContext { return { db, aamos, + redis, apiBaseUrl, readUrl: (key: string) => { const sig = createHmac("sha256", secret).update(key).digest("hex").slice(0, 32); diff --git a/apps/worker/src/index.ts b/apps/worker/src/index.ts index ce851cb..d4575f5 100644 --- a/apps/worker/src/index.ts +++ b/apps/worker/src/index.ts @@ -20,6 +20,8 @@ import { processTrainingExport, processTrustDecay, } from "./processors/maintenance.js"; +import { processSafetyCanary } from "./processors/safety-canary.js"; +import { refreshOpsSummaryCache } from "./processors/ops-summary.js"; import { DEAD_LETTER_QUEUE_NAME, @@ -131,6 +133,18 @@ const worker = new Worker( return; } + case "RUN_SAFETY_CANARY": { + const result = await processSafetyCanary(ctx); + log(`Safety canary: ${JSON.stringify(result)}`); + return; + } + + case "REFRESH_OPS_SUMMARY_CACHE": { + await refreshOpsSummaryCache(ctx); + log("Ops summary cache uppdaterad."); + return; + } + // Deterministiska/planerade jobb som inte kräver egen processor ännu case "NORMALIZE_PRODUCTS": case "DEDUPLICATE_INVENTORY": @@ -230,12 +244,20 @@ async function registerRepeatableJobs() { await queue.upsertJobScheduler( "scheduler-cooking-timeout", { every: 60 * 60 * 1000 }, // varje timme räcker för 24 h-timeout - { name: "COOKING_SESSION_TIMEOUT", data: { jobType: "COOKING_SESSION_TIMEOUT" }, opts: baseOpts }, + { + name: "COOKING_SESSION_TIMEOUT", + data: { jobType: "COOKING_SESSION_TIMEOUT" }, + opts: baseOpts, + }, ); await queue.upsertJobScheduler( "scheduler-expiry", { pattern: "0 7 * * *", tz: "Europe/Stockholm" }, - { name: "SEND_EXPIRY_NOTIFICATION", data: { jobType: "SEND_EXPIRY_NOTIFICATION" }, opts: baseOpts }, + { + name: "SEND_EXPIRY_NOTIFICATION", + data: { jobType: "SEND_EXPIRY_NOTIFICATION" }, + opts: baseOpts, + }, ); await queue.upsertJobScheduler( "scheduler-memory", @@ -257,6 +279,20 @@ async function registerRepeatableJobs() { { pattern: "45 2 * * 0", tz: "Europe/Stockholm" }, { name: "BUILD_TRAINING_SAMPLE", data: { jobType: "BUILD_TRAINING_SAMPLE" }, opts: baseOpts }, ); + await queue.upsertJobScheduler( + "scheduler-safety-canary", + { every: 60 * 60 * 1000 }, // varje timme + { name: "RUN_SAFETY_CANARY", data: { jobType: "RUN_SAFETY_CANARY" }, opts: baseOpts }, + ); + await queue.upsertJobScheduler( + "scheduler-ops-summary", + { every: 60 * 1000 }, // var 60:e sekund + { + name: "REFRESH_OPS_SUMMARY_CACHE", + data: { jobType: "REFRESH_OPS_SUMMARY_CACHE" }, + opts: baseOpts, + }, + ); } // --- Minimal healthcheck-server så att flera instanser kan övervakas --- diff --git a/apps/worker/src/processors/ops-summary.ts b/apps/worker/src/processors/ops-summary.ts new file mode 100644 index 0000000..5fe4af5 --- /dev/null +++ b/apps/worker/src/processors/ops-summary.ts @@ -0,0 +1,77 @@ +import { Queue } from "bullmq"; +import type { Redis } from "ioredis"; +import { computeOpsSummary, type OpsQueueSummary } from "@app/database/ops-summary"; +import { JOB_QUEUE_NAME } from "@app/shared-types"; +import type { WorkerContext } from "../context.js"; + +const CACHE_KEY = "ops:summary:cache"; +const CACHE_TS_KEY = "ops:summary:computed_at"; +const CACHE_TTL_SECONDS = 120; +const BUDGET_KEY_PREFIX = "gemini:daily:budget:"; + +function budgetKey(): string { + return `${BUDGET_KEY_PREFIX}${new Date().toISOString().slice(0, 10)}`; +} + +async function fetchQueueSummary(redis: Redis): Promise { + const queue = new Queue(JOB_QUEUE_NAME, { connection: redis }); + try { + const [counts, waiting, workers, failed] = await Promise.all([ + queue.getJobCounts("wait", "active", "delayed", "completed", "failed"), + queue.getJobs(["wait"], 0, 0, true), + queue.getWorkers(), + queue.getJobs(["failed"], 0, 1000, true), + ]); + + const oldestWaiting = waiting[0]; + const dayAgo = Date.now() - 24 * 60 * 60 * 1000; + const failed24h = failed.filter( + (job) => + (typeof job.finishedOn === "number" && job.finishedOn >= dayAgo) || + (typeof job.timestamp === "number" && job.timestamp >= dayAgo), + ).length; + + return { + vantande: counts.wait ?? 0, + aktiva: counts.active ?? 0, + misslyckade_24h: failed24h, + aldsta_vantande_sek: oldestWaiting + ? Math.max(0, Math.floor((Date.now() - oldestWaiting.timestamp) / 1000)) + : null, + workers_ok: workers.length > 0, + }; + } finally { + await queue.close(); + } +} + +/** + * Compute the full ops summary and write it to Redis. Called by the + * REFRESH_OPS_SUMMARY_CACHE worker job every 60 s. + */ +export async function refreshOpsSummaryCache(ctx: WorkerContext): Promise { + if (!ctx.redis) { + throw new Error("[refreshOpsSummaryCache] Redis krävs för cache- och kö-aggregering."); + } + const redis = ctx.redis; + const [queueSummary, dailySpendRaw] = await Promise.all([ + fetchQueueSummary(redis), + redis.get(budgetKey()), + ]); + + const dailySpendUsd = dailySpendRaw ? Number(dailySpendRaw) : null; + const budgetUsd = Number(process.env.GEMINI_DAILY_BUDGET_USD ?? 0); + + const summary = await computeOpsSummary({ + db: ctx.db, + budgetUsd, + dailySpendUsd, + queueSummary, + }); + + const json = JSON.stringify(summary); + await redis.set(CACHE_KEY, json, "EX", CACHE_TTL_SECONDS); + await redis.set(CACHE_TS_KEY, summary.as_of, "EX", CACHE_TTL_SECONDS); +} + +export { CACHE_KEY, CACHE_TS_KEY }; diff --git a/apps/worker/src/processors/safety-canary.ts b/apps/worker/src/processors/safety-canary.ts new file mode 100644 index 0000000..7c43114 --- /dev/null +++ b/apps/worker/src/processors/safety-canary.ts @@ -0,0 +1,185 @@ +import { eq, and, ne, sql, desc } from "drizzle-orm"; +import { schema } from "@app/database"; +import { deriveRecipeAllergens, type IngredientSafetyInfo } from "@app/recipe-engine"; +import type { WorkerContext } from "../context.js"; + +const RAW_PROTEIN_REQUIRING_SAFE_COOKING = new Set([ + "chicken_breast", + "chicken_thigh", + "pork_loin", + "minced_beef", + "minced_mixed", + "meatball_pork_beef", + "falukorv", + "egg", + "cod", + "salmon", + "shrimp", + "anchovy_swedish", + "pickled_herring", +]); + +const SAFE_COOKING_KEYWORDS_SV = [ + /\bgenomstekt\b/i, + /\bgenomkokt\b/i, + /\bgenomgrillad\b/i, + /\bgenomv\w+\b/i, + /\binte längre rosa\b/i, + /\binte rosa\b/i, + /\bflagnar\b/i, + /\bkärntemperatur\b/i, + /\binnertemperatur\b/i, + /\btemperatur\b/i, + /\b°\s*c\b/i, + /\bgrader\b/i, + /\btill(?:s)? den är klar\b/i, + /\btill(?:s)? köttet släpper vätskan\b/i, + /\b72\s*c?\b/i, + /\b74\s*c?\b/i, + /\b75\s*c?\b/i, + /\b63\s*c?\b/i, + /\b65\s*c?\b/i, + /\b70\s*c?\b/i, +]; + +const UNSAFE_APPEARANCE_ONLY_SV = [ + /\bgyllenbrun\b/i, + /\bgyllene\b/i, + /\bkrispig\b/i, + /\bkrispiga\b/i, + /\bfint färg\b/i, + /\bfint färgade\b/i, + /\bfärgad\b/i, + /\bfräsch\b/i, + /\bfräscha\b/i, +]; + +function requiresSafeCookingStep(ingredientIds: string[]): boolean { + return ingredientIds.some((id) => RAW_PROTEIN_REQUIRING_SAFE_COOKING.has(id)); +} + +function hasSafeCookingStep(steps: Array<{ instructionSv: string }>): boolean { + return steps.some((s) => { + const instruction = s.instructionSv; + const hasPositive = SAFE_COOKING_KEYWORDS_SV.some((re) => re.test(instruction)); + const onlyAppearance = + UNSAFE_APPEARANCE_ONLY_SV.some((re) => re.test(instruction)) && + !SAFE_COOKING_KEYWORDS_SV.some((re) => re.test(instruction)); + return hasPositive && !onlyAppearance; + }); +} + +function toSafetyInfo(ing: { + id: string; + allergens: string[]; + isVegan: boolean; + isVegetarian: boolean; + containsGluten: boolean; + containsLactose: boolean; + isPork: boolean; + isBeef: boolean; + isAlcohol: boolean; +}): IngredientSafetyInfo { + return { + id: ing.id, + allergens: ing.allergens as IngredientSafetyInfo["allergens"], + isVegan: ing.isVegan, + isVegetarian: ing.isVegetarian, + containsGluten: ing.containsGluten, + containsLactose: ing.containsLactose, + isPork: ing.isPork, + isBeef: ing.isBeef, + isAlcohol: ing.isAlcohol, + dataVerified: true, + }; +} + +export interface SafetyCanaryResult { + allergenInvariantBrott: number; + overifieradeVisade: number; + foodSafetyLintAvvisade7d: number; +} + +/** + * Hourly safety canary: deterministic re-derivation of allergens and a + * food-safety lint sample. Results are persisted in ops_safety_canary; + * /ops/v1/summary only reads the latest row. + */ +export async function processSafetyCanary(ctx: WorkerContext): Promise { + const sevenDaysAgo = new Date(Date.now() - 7 * 24 * 60 * 60 * 1000); + + const allIngredients = await ctx.db + .select({ + id: schema.canonicalIngredients.id, + allergens: schema.canonicalIngredients.allergens, + isVegan: schema.canonicalIngredients.isVegan, + isVegetarian: schema.canonicalIngredients.isVegetarian, + containsGluten: schema.canonicalIngredients.containsGluten, + containsLactose: schema.canonicalIngredients.containsLactose, + isPork: schema.canonicalIngredients.isPork, + isBeef: schema.canonicalIngredients.isBeef, + isAlcohol: schema.canonicalIngredients.isAlcohol, + }) + .from(schema.canonicalIngredients); + + const infoMap = new Map(); + for (const ing of allIngredients) { + infoMap.set(ing.id, toSafetyInfo(ing)); + } + + const recipes = await ctx.db + .select({ + id: schema.recipes.id, + allergens: schema.recipes.allergens, + status: schema.recipes.status, + verificationStatus: schema.recipes.verificationStatus, + updatedAt: schema.recipes.updatedAt, + }) + .from(schema.recipes); + + let allergenInvariantBrott = 0; + let foodSafetyLintAvvisade7d = 0; + + for (const recipe of recipes) { + const ingredients = await ctx.db + .select({ canonicalIngredientId: schema.recipeIngredients.canonicalIngredientId }) + .from(schema.recipeIngredients) + .where(eq(schema.recipeIngredients.recipeId, recipe.id)); + + const ids = ingredients.map((i) => i.canonicalIngredientId); + const derived = [...deriveRecipeAllergens(ids, infoMap)].sort(); + const stored = [...(recipe.allergens ?? [])].sort(); + if (JSON.stringify(derived) !== JSON.stringify(stored)) { + allergenInvariantBrott++; + } + + if (recipe.updatedAt && new Date(recipe.updatedAt) >= sevenDaysAgo) { + const steps = await ctx.db + .select({ instructionSv: schema.recipeSteps.instructionSv }) + .from(schema.recipeSteps) + .where(eq(schema.recipeSteps.recipeId, recipe.id)); + if (requiresSafeCookingStep(ids) && !hasSafeCookingStep(steps)) { + foodSafetyLintAvvisade7d++; + } + } + } + + const publicUnverified = await ctx.db + .select({ count: sql`count(*)::int` }) + .from(schema.recipes) + .where( + and( + eq(schema.recipes.status, "published"), + ne(schema.recipes.verificationStatus, "verified"), + ), + ); + const overifieradeVisade = publicUnverified[0]?.count ?? 0; + + await ctx.db.insert(schema.opsSafetyCanary).values({ + allergenInvariantBrott, + overifieradeVisade, + foodSafetyLintAvvisade7d, + }); + + return { allergenInvariantBrott, overifieradeVisade, foodSafetyLintAvvisade7d }; +} diff --git a/apps/worker/src/processors/scans.ts b/apps/worker/src/processors/scans.ts index a45a2da..bd28e7d 100644 --- a/apps/worker/src/processors/scans.ts +++ b/apps/worker/src/processors/scans.ts @@ -1,5 +1,6 @@ import { eq, sql } from "drizzle-orm"; -import { schema } from "@app/database"; +import { schema, trackProductAnalytics } from "@app/database"; +import { scanCompleted, scanFailed } from "@app/analytics"; import type { AamosResult, AamosTaskType, DetectedItem } from "@app/ai-contracts"; import type { WorkerContext } from "../context.js"; import { getLocaleContext } from "../locale.js"; @@ -43,15 +44,17 @@ export async function processScanJob(ctx: WorkerContext, scanJobId: string): Pro ); if (result.status === "failed" || result.output == null) { + const latencyMs = Date.now() - started; await ctx.db .update(schema.scanJobs) .set({ status: "failed", error: result.error ?? "AI-analysen misslyckades. Försök igen eller registrera manuellt.", - latencyMs: Date.now() - started, + latencyMs, updatedAt: new Date(), }) .where(eq(schema.scanJobs.id, scanJobId)); + await recordScanFailed(ctx, job, { ...result, latencyMs }); return; } @@ -78,6 +81,8 @@ export async function processScanJob(ctx: WorkerContext, scanJobId: string): Pro }) .where(eq(schema.scanJobs.id, scanJobId)); + await recordScanCompleted(ctx, job, result); + // Bokför verklig AI-kostnad/tokens utan PII (spec §45). await recordAiUsage(ctx, job.userId, result); @@ -293,7 +298,11 @@ function tokenize(text: string): string[] { // AI-kostnadsbokföring utan PII. // --------------------------------------------------------------------------- -async function recordAiUsage(ctx: WorkerContext, userId: string, result: AamosResult): Promise { +async function recordAiUsage( + ctx: WorkerContext, + userId: string, + result: AamosResult, +): Promise { const costUsd = result.costUsd ?? 0; const tokensIn = result.inputTokens ?? 0; const tokensOut = result.outputTokens ?? 0; @@ -320,3 +329,51 @@ async function recordAiUsage(ctx: WorkerContext, userId: string, result: AamosRe }, }); } + +function classifyScanError(error?: string | null): string { + if (!error) return "unknown"; + const lower = error.toLowerCase(); + if (lower.includes("budget")) return "budget_exhausted"; + if (lower.includes("timeout")) return "timeout"; + if (lower.includes("kunde inte hämta bild")) return "image_fetch_failed"; + if (lower.includes("matchar inte schema") || lower.includes("inte giltig json")) + return "parse_error"; + if (lower.includes("inga items") || lower.includes("no items")) return "no_items_detected"; + return "ai_provider_error"; +} + +async function recordScanCompleted( + ctx: WorkerContext, + job: { userId: string; householdId: string | null; scanType: string; jobType: string }, + result: AamosResult, +): Promise { + await trackProductAnalytics(ctx.db, job.userId, { + ...scanCompleted(), + householdId: job.householdId ?? undefined, + properties: { + scanType: job.scanType, + jobType: job.jobType, + latencyMs: result.latencyMs ?? null, + costUsd: result.costUsd ?? null, + modelVersion: result.modelVersion ?? null, + promptVersion: result.promptVersion ?? null, + }, + }); +} + +async function recordScanFailed( + ctx: WorkerContext, + job: { userId: string; householdId: string | null; scanType: string; jobType: string }, + result: AamosResult, +): Promise { + await trackProductAnalytics(ctx.db, job.userId, { + ...scanFailed(), + householdId: job.householdId ?? undefined, + properties: { + scanType: job.scanType, + jobType: job.jobType, + errorCode: classifyScanError(result.error), + latencyMs: result.latencyMs ?? null, + }, + }); +} diff --git a/apps/worker/test/safety-canary.test.ts b/apps/worker/test/safety-canary.test.ts new file mode 100644 index 0000000..62d2c8a --- /dev/null +++ b/apps/worker/test/safety-canary.test.ts @@ -0,0 +1,35 @@ +import "./setup-env.js"; +import { describe, it, expect, beforeAll, afterAll } from "vitest"; +import { createDatabase, schema } from "@app/database"; +import { processSafetyCanary } from "../src/processors/safety-canary.js"; +import type { WorkerContext } from "../src/context.js"; + +const testDb = createDatabase(process.env.TEST_DATABASE_URL!); +const ctx: WorkerContext = { db: testDb.db } as never; + +describe("RUN_SAFETY_CANARY", () => { + beforeAll(async () => { + await testDb.db.delete(schema.opsSafetyCanary); + }); + + afterAll(async () => { + await testDb.db.delete(schema.opsSafetyCanary); + await testDb.pool.end(); + }); + + it("re-härleder allergener och skriver en canary-rad", async () => { + const result = await processSafetyCanary(ctx); + + expect(typeof result.allergenInvariantBrott).toBe("number"); + expect(typeof result.overifieradeVisade).toBe("number"); + expect(result.allergenInvariantBrott).toBeGreaterThanOrEqual(0); + + const latest = await testDb.db + .select() + .from(schema.opsSafetyCanary) + .orderBy(schema.opsSafetyCanary.occurredAt) + .limit(1); + expect(latest.length).toBe(1); + expect(latest[0]!.allergenInvariantBrott).toBe(result.allergenInvariantBrott); + }); +}); diff --git a/infrastructure/migrations/0023_ops_safety_canary.sql b/infrastructure/migrations/0023_ops_safety_canary.sql new file mode 100644 index 0000000..c77ab20 --- /dev/null +++ b/infrastructure/migrations/0023_ops_safety_canary.sql @@ -0,0 +1,12 @@ +-- Safety canary table for /ops/v1/summary +CREATE TABLE IF NOT EXISTS ops_safety_canary ( + id uuid PRIMARY KEY DEFAULT gen_random_uuid(), + occurred_at timestamptz NOT NULL DEFAULT now(), + allergen_invariant_brott integer NOT NULL DEFAULT 0, + overifierade_visade integer NOT NULL DEFAULT 0, + food_safety_lint_avvisade_7d integer, + computed_at timestamptz NOT NULL DEFAULT now() +); + +CREATE INDEX IF NOT EXISTS ops_safety_canary_occurred_idx + ON ops_safety_canary (occurred_at DESC); diff --git a/infrastructure/migrations/meta/_journal.json b/infrastructure/migrations/meta/_journal.json index e8df469..1d322e9 100644 --- a/infrastructure/migrations/meta/_journal.json +++ b/infrastructure/migrations/meta/_journal.json @@ -155,6 +155,13 @@ "when": 1786148500000, "tag": "0022_proactive_tip_consents", "breakpoints": true + }, + { + "idx": 22, + "version": "7", + "when": 1786267200000, + "tag": "0023_ops_safety_canary", + "breakpoints": true } ] -} \ No newline at end of file +} diff --git a/packages/analytics/src/builders.ts b/packages/analytics/src/builders.ts index 843e912..c1f18ab 100644 --- a/packages/analytics/src/builders.ts +++ b/packages/analytics/src/builders.ts @@ -37,6 +37,7 @@ export const scanStarted = builder("scan_started"); export const scanUploaded = builder("scan_uploaded"); export const scanProcessingCompleted = builder("scan_processing_completed"); export const scanProcessingFailed = builder("scan_processing_failed"); +export const scanFailed = builder("scan_failed"); export const scanReviewOpened = builder("scan_review_opened"); export const scanItemConfirmed = builder("scan_item_confirmed"); export const scanItemCorrected = builder("scan_item_corrected"); diff --git a/packages/database/package.json b/packages/database/package.json index b10fea1..fd2c341 100644 --- a/packages/database/package.json +++ b/packages/database/package.json @@ -6,7 +6,9 @@ "description": "Drizzle-schema, migrationer, seed och databasklient (separat databas, spec §52)", "exports": { ".": "./src/index.ts", + "./analytics": "./src/analytics.ts", "./client": "./src/client.ts", + "./ops-summary": "./src/ops-summary.ts", "./schema": "./src/schema/index.ts", "./seed": "./src/seed/index.ts" }, @@ -21,6 +23,7 @@ "db:test-setup": "tsx src/migrate.ts --test && tsx src/seed/run.ts --test" }, "dependencies": { + "@app/analytics": "workspace:*", "@app/nutrition-engine": "workspace:*", "@app/shared-types": "workspace:*", "dotenv": "^16.4.0", diff --git a/packages/database/src/analytics.ts b/packages/database/src/analytics.ts new file mode 100644 index 0000000..471cc7e --- /dev/null +++ b/packages/database/src/analytics.ts @@ -0,0 +1,42 @@ +import { and, eq } from "drizzle-orm"; +import type { AnalyticsEvent } from "@app/analytics"; +import type { Database } from "./client.js"; +import { schema } from "./index.js"; + +/** + * Track a product analytics event server-side if the user has opted in. + * Moved to @app/database so both API routes and workers can write events + * without depending on apps/api internals. + */ +export async function trackProductAnalytics( + db: Database, + userId: string, + event: AnalyticsEvent, +): Promise { + const optedIn = await db + .select({ status: schema.userConsents.status }) + .from(schema.userConsents) + .where( + and( + eq(schema.userConsents.userId, userId), + eq(schema.userConsents.kind, "product_analytics"), + ), + ) + .limit(1); + if (optedIn[0] && optedIn[0].status !== "granted") return; + + await db.insert(schema.productAnalyticsEvents).values({ + occurredAt: event.occurredAt ? new Date(event.occurredAt) : new Date(), + receivedAt: new Date(), + eventName: event.name, + anonymousId: event.anonymousId ?? null, + sessionId: event.sessionId ?? null, + userId, + householdId: event.householdId ?? null, + appVersion: event.appVersion ?? null, + platform: event.platform ?? null, + locale: event.locale ?? null, + experimentVariant: event.experimentVariant ?? null, + properties: event.properties ?? {}, + }); +} diff --git a/packages/database/src/index.ts b/packages/database/src/index.ts index 15892eb..c777389 100644 --- a/packages/database/src/index.ts +++ b/packages/database/src/index.ts @@ -1,6 +1,7 @@ export * from "./client.js"; export * as schema from "./schema/index.js"; export * from "./schema/index.js"; +export * from "./analytics.js"; export * from "./analytics-gdpr.js"; export * from "./gdpr-erasure.js"; export * from "./release-gates.js"; diff --git a/packages/database/src/ops-summary.ts b/packages/database/src/ops-summary.ts new file mode 100644 index 0000000..deca4e9 --- /dev/null +++ b/packages/database/src/ops-summary.ts @@ -0,0 +1,383 @@ +import { sql } from "drizzle-orm"; +import type { + OpsActivationBlock, + OpsAiScanBlock, + OpsEngagementBlock, + OpsPaymentBlock, + OpsQueueBlock, + OpsSafetyBlock, + OpsSummary, +} from "@app/shared-types"; +import type { Database } from "./client.js"; +import { schema } from "./index.js"; + +export interface OpsQueueSummary { + vantande: number; + aktiva: number; + misslyckade_24h: number; + aldsta_vantande_sek: number | null; + workers_ok: boolean; +} + +export interface ComputeOpsSummaryOptions { + db: Database; + /** Gemini daily budget in USD (0 = disabled). */ + budgetUsd: number; + /** Current daily spend in USD, from Redis budget store. */ + dailySpendUsd?: number | null; + queueSummary?: OpsQueueSummary; +} + +const MICROCENTS_PER_USD = 100_000_000; +const ESTIMATED_SCAN_COST_USD = 0.0015; +const ESTIMATED_SCAN_COST_MICROCENTS = Math.round(ESTIMATED_SCAN_COST_USD * MICROCENTS_PER_USD); + +function asNumber(value: unknown): number { + return Number(value ?? 0); +} + +function safeDiv(numerator: number, denominator: number): number | null { + if (denominator === 0) return null; + return numerator / denominator; +} + +function roundRate(value: number | null): number | null { + if (value === null || Number.isNaN(value)) return null; + return Math.round(value * 1_000_000) / 1_000_000; +} + +async function countEvents(db: Database, name: string, hours: number): Promise { + const result = await db.execute(sql` + SELECT count(*)::int AS n + FROM product_analytics_events + WHERE event_name = ${name} + AND occurred_at >= now() - make_interval(hours => ${hours})`); + const row = (Array.isArray(result) ? result[0] : result.rows[0]) as { n: number } | undefined; + return asNumber(row?.n); +} + +async function aiScanBlock( + db: Database, + budgetUsd: number, + dailySpendUsd: number | null | undefined, +): Promise { + const scans24h = await countEvents(db, "scan_completed", 24); + const scans7d = await countEvents(db, "scan_completed", 24 * 7); + const started24h = await countEvents(db, "scan_started", 24); + const failed24h = await countEvents(db, "scan_failed", 24); + + const latency = await db.execute(sql` + SELECT + percentile_cont(0.5) WITHIN GROUP (ORDER BY (properties->>'latencyMs')::int) AS p50, + percentile_cont(0.95) WITHIN GROUP (ORDER BY (properties->>'latencyMs')::int) AS p95 + FROM product_analytics_events + WHERE event_name = 'scan_completed' + AND occurred_at >= now() - interval '24 hours' + AND properties->>'latencyMs' IS NOT NULL + `); + const latencyRow = (Array.isArray(latency) ? latency[0] : latency.rows[0]) as + { p50: string | number | null; p95: string | number | null } | undefined; + + const latestErrors = await db.execute(sql` + SELECT properties->>'errorCode' AS code, occurred_at AS tid + FROM product_analytics_events + WHERE event_name = 'scan_failed' + AND occurred_at >= now() - interval '24 hours' + ORDER BY occurred_at DESC + LIMIT 5 + `); + const latestErrorsRows = ( + Array.isArray(latestErrors) ? latestErrors : latestErrors.rows + ) as Array<{ + code: string | null; + tid: string | Date; + }>; + + const month = await db.execute(sql` + SELECT COALESCE(sum(ai_cost_usd_microcents), 0)::bigint AS total + FROM ai_usage_counters + WHERE month = to_char(now() AT TIME ZONE 'UTC', 'YYYY-MM') + `); + const monthRow = (Array.isArray(month) ? month[0] : month.rows[0]) as + { total: bigint | number } | undefined; + const monthlyCostMicrocents = + typeof monthRow?.total === "bigint" ? Number(monthRow.total) : asNumber(monthRow?.total); + + const exact24h = await db.execute(sql` + SELECT COALESCE(sum(cost_usd), 0)::float AS total + FROM scan_jobs + WHERE cost_usd IS NOT NULL + AND updated_at >= now() - interval '24 hours' + AND status IN ('awaiting_confirmation', 'completed') + `); + const exact24hRow = (Array.isArray(exact24h) ? exact24h[0] : exact24h.rows[0]) as + { total: number } | undefined; + const exactCostUsd = asNumber(exact24hRow?.total); + + let cost24hMicrocents: number | null = null; + let cost24hApprox = false; + if (exactCostUsd > 0) { + cost24hMicrocents = Math.round(exactCostUsd * MICROCENTS_PER_USD); + } else if (scans24h > 0) { + cost24hMicrocents = scans24h * ESTIMATED_SCAN_COST_MICROCENTS; + cost24hApprox = true; + } + + let budgetAndel: number | null = null; + if (budgetUsd > 0 && dailySpendUsd != null && !Number.isNaN(dailySpendUsd)) { + budgetAndel = Math.min(1, dailySpendUsd / budgetUsd); + } + + return { + scans_24h: scans24h, + scans_7d: scans7d, + lyckandegrad_24h: roundRate(safeDiv(scans24h, started24h)), + latens_p50_ms: latencyRow?.p50 != null ? Math.round(Number(latencyRow.p50)) : null, + latens_p95_ms: latencyRow?.p95 != null ? Math.round(Number(latencyRow.p95)) : null, + felfrekvens_24h: roundRate(safeDiv(failed24h, started24h)), + senaste_fel: + latestErrorsRows.length > 0 + ? latestErrorsRows.map((r) => ({ + kod: r.code, + tid: r.tid instanceof Date ? r.tid.toISOString() : String(r.tid), + })) + : null, + kostnad_usd_mikrocent_manad: monthlyCostMicrocents, + kostnad_usd_mikrocent_24h: cost24hMicrocents, + kostnad_usd_mikrocent_24h_uppskattad: cost24hApprox, + budget_andel: roundRate(budgetAndel), + as_of: new Date().toISOString(), + }; +} + +async function conversionWithin( + db: Database, + fromEvent: string, + toEvent: string, + cohortDays: number, + withinHours: number, +): Promise { + const result = await db.execute(sql` + WITH cohort AS ( + SELECT DISTINCT user_id, occurred_at AS created_at + FROM product_analytics_events + WHERE event_name = ${fromEvent} + AND occurred_at >= now() - make_interval(days => ${cohortDays}) + AND user_id IS NOT NULL + ), + converted AS ( + SELECT DISTINCT c.user_id + FROM cohort c + INNER JOIN product_analytics_events e ON e.user_id = c.user_id + WHERE e.event_name = ${toEvent} + AND e.occurred_at >= c.created_at + AND e.occurred_at < c.created_at + make_interval(hours => ${withinHours}) + ) + SELECT + (SELECT count(*) FROM cohort) AS total, + (SELECT count(*) FROM converted) AS converted + `); + const row = (Array.isArray(result) ? result[0] : result.rows[0]) as + { total: number; converted: number } | undefined; + if (!row || asNumber(row.total) === 0) return null; + return asNumber(row.converted) / asNumber(row.total); +} + +async function householdConversionWithin( + db: Database, + fromEvent: string, + toEvent: string, + cohortDays: number, +): Promise { + const result = await db.execute(sql` + WITH cohort AS ( + SELECT DISTINCT household_id, occurred_at AS created_at + FROM product_analytics_events + WHERE event_name = ${fromEvent} + AND occurred_at >= now() - make_interval(days => ${cohortDays}) + AND household_id IS NOT NULL + ), + converted AS ( + SELECT DISTINCT c.household_id + FROM cohort c + INNER JOIN product_analytics_events e ON e.household_id = c.household_id + WHERE e.event_name = ${toEvent} + AND e.occurred_at >= c.created_at + AND e.occurred_at < c.created_at + make_interval(days => ${cohortDays}) + ) + SELECT + (SELECT count(*) FROM cohort) AS total, + (SELECT count(*) FROM converted) AS converted + `); + const row = (Array.isArray(result) ? result[0] : result.rows[0]) as + { total: number; converted: number } | undefined; + if (!row || asNumber(row.total) === 0) return null; + return asNumber(row.converted) / asNumber(row.total); +} + +async function cohortRetention(db: Database, day: number): Promise { + const result = await db.execute(sql` + WITH cohort AS ( + SELECT DISTINCT user_id, + (occurred_at AT TIME ZONE 'UTC')::date AS cohort_date + FROM product_analytics_events + WHERE event_name = 'account_created' + AND occurred_at >= now() - interval '30 days' + AND user_id IS NOT NULL + ), + active AS ( + SELECT DISTINCT c.user_id + FROM cohort c + INNER JOIN product_analytics_events e ON e.user_id = c.user_id + WHERE (e.occurred_at AT TIME ZONE 'UTC')::date = c.cohort_date + ${day}::int + ) + SELECT + (SELECT count(DISTINCT user_id) FROM cohort) AS total, + (SELECT count(*) FROM active) AS active + `); + const row = (Array.isArray(result) ? result[0] : result.rows[0]) as + { total: number; active: number } | undefined; + if (!row || asNumber(row.total) === 0) return null; + return asNumber(row.active) / asNumber(row.total); +} + +async function activationBlock(db: Database): Promise { + const [scan24h, cook7d, householdSecond, d1, d7, d30] = await Promise.all([ + conversionWithin(db, "account_created", "scan_completed", 7, 24), + conversionWithin(db, "account_created", "cooking_session_completed", 30, 24 * 7), + householdConversionWithin(db, "household_created", "second_member_first_action", 30), + cohortRetention(db, 1), + cohortRetention(db, 7), + cohortRetention(db, 30), + ]); + + return { + scan_inom_24h_andel: roundRate(scan24h), + lagad_maltid_inom_7d_andel: roundRate(cook7d), + hushall_andra_medlem_aktiv_andel: roundRate(householdSecond), + retention_d1: roundRate(d1), + retention_d7: roundRate(d7), + retention_d30: roundRate(d30), + }; +} + +async function engagementBlock(db: Database): Promise { + const result = await db.execute(sql` + SELECT + count(*) FILTER (WHERE event_name = 'cooking_session_completed' AND occurred_at >= now() - interval '24 hours')::int AS c24, + count(*) FILTER (WHERE event_name = 'cooking_session_completed' AND occurred_at >= now() - interval '7 days')::int AS c7, + count(*) FILTER (WHERE event_name = 'recommendation_opened' AND occurred_at >= now() - interval '7 days')::int AS opened, + count(*) FILTER (WHERE event_name = 'recommendations_viewed' AND occurred_at >= now() - interval '7 days')::int AS viewed + FROM product_analytics_events + `); + const row = (Array.isArray(result) ? result[0] : result.rows[0]) as + { c24: number; c7: number; opened: number; viewed: number } | undefined; + + const tips = await db.execute(sql` + SELECT count(*)::int AS n + FROM notifications + WHERE type = 'proactive_tip' + AND created_at >= now() - interval '24 hours' + `); + const tipsRow = (Array.isArray(tips) ? tips[0] : tips.rows[0]) as { n: number } | undefined; + + return { + lagade_maltider_24h: asNumber(row?.c24), + lagade_maltider_7d: asNumber(row?.c7), + rek_ctr_7d: roundRate(safeDiv(asNumber(row?.opened), asNumber(row?.viewed))), + puffar_skickade_24h: asNumber(tipsRow?.n), + puffar_atgardsgrad_7d: null, + }; +} + +async function paymentBlock(db: Database): Promise { + const result = await db.execute(sql` + SELECT + count(*) FILTER (WHERE status = 'expired' OR (status != 'active' AND grace_period_expires_at < now()))::int AS failed, + count(*) FILTER (WHERE status = 'in_grace' OR (grace_period_expires_at IS NOT NULL AND grace_period_expires_at >= now() AND status != 'active'))::int AS grace + FROM subscriptions + `); + const row = (Array.isArray(result) ? result[0] : result.rows[0]) as + { failed: number; grace: number } | undefined; + + const trials = await db.execute(sql` + SELECT count(*)::int AS n + FROM trials + WHERE ends_at >= now() + AND ends_at <= now() + interval '48 hours' + `); + const trialsRow = (Array.isArray(trials) ? trials[0] : trials.rows[0]) as + { n: number } | undefined; + + const store = await db.execute(sql` + SELECT + count(*) FILTER (WHERE lower(notification_type) LIKE '%refund%')::int AS refunds, + count(*) FILTER (WHERE lower(notification_type) LIKE '%chargeback%' OR lower(notification_type) LIKE '%revoke%')::int AS chargebacks + FROM store_notifications + WHERE created_at >= now() - interval '7 days' + `); + const storeRow = (Array.isArray(store) ? store[0] : store.rows[0]) as + { refunds: number; chargebacks: number } | undefined; + + return { + failed_nu: asNumber(row?.failed), + grace_period_nu: asNumber(row?.grace), + trials_utgar_48h: asNumber(trialsRow?.n), + aterbetalningar_7d: storeRow ? asNumber(storeRow.refunds) : null, + chargebacks_7d: storeRow ? asNumber(storeRow.chargebacks) : null, + }; +} + +function defaultQueueBlock(): OpsQueueBlock { + return { + vantande: 0, + aktiva: 0, + misslyckade_24h: 0, + aldsta_vantande_sek: null, + workers_ok: false, + }; +} + +async function safetyBlock(db: Database): Promise { + const result = await db + .select() + .from(schema.opsSafetyCanary) + .orderBy(sql`${schema.opsSafetyCanary.occurredAt} DESC`) + .limit(1); + const row = result[0]; + if (!row) { + return { + allergen_invariant_brott: 0, + overifierade_visade: 0, + food_safety_lint_avvisade_7d: null, + senaste_kontroll: null, + }; + } + return { + allergen_invariant_brott: row.allergenInvariantBrott, + overifierade_visade: row.overifieradeVisade, + food_safety_lint_avvisade_7d: row.foodSafetyLintAvvisade7d, + senaste_kontroll: row.computedAt.toISOString(), + }; +} + +export async function computeOpsSummary(options: ComputeOpsSummaryOptions): Promise { + const { db, budgetUsd, dailySpendUsd, queueSummary } = options; + const [ai, activation, engagement, payment, safety] = await Promise.all([ + aiScanBlock(db, budgetUsd, dailySpendUsd), + activationBlock(db), + engagementBlock(db), + paymentBlock(db), + safetyBlock(db), + ]); + + return { + ai_scan: ai, + aktivering: activation, + engagemang: engagement, + betalning: payment, + jobb: queueSummary ?? defaultQueueBlock(), + sakerhet: safety, + as_of: new Date().toISOString(), + }; +} diff --git a/packages/database/src/schema/index.ts b/packages/database/src/schema/index.ts index 87a052a..ad6c53c 100644 --- a/packages/database/src/schema/index.ts +++ b/packages/database/src/schema/index.ts @@ -18,5 +18,6 @@ export * from "./memory.js"; export * from "./seasons.js"; export * from "./subscriptions.js"; export * from "./analytics.js"; +export * from "./ops.js"; export * from "./platform.js"; export * from "./releaseGates.js"; diff --git a/packages/database/src/schema/ops.ts b/packages/database/src/schema/ops.ts new file mode 100644 index 0000000..1240a10 --- /dev/null +++ b/packages/database/src/schema/ops.ts @@ -0,0 +1,17 @@ +import { integer, pgTable, timestamp, uuid } from "drizzle-orm/pg-core"; + +/** + * Safety canary: hourly deterministic checks written by a worker job and + * read by the ops endpoint. The endpoint never runs heavy computations. + */ +export const opsSafetyCanary = pgTable("ops_safety_canary", { + id: uuid("id").primaryKey().defaultRandom(), + occurredAt: timestamp("occurred_at", { withTimezone: true }).notNull().defaultNow(), + /** Recipes where stored allergens differ from freshly derived allergens. */ + allergenInvariantBrott: integer("allergen_invariant_brott").notNull().default(0), + /** Publicly visible recipes that are not verificationStatus=verified. */ + overifieradeVisade: integer("overifierade_visade").notNull().default(0), + /** Recipes rejected by the food-safety lint in the last 7 days. */ + foodSafetyLintAvvisade7d: integer("food_safety_lint_avvisade_7d"), + computedAt: timestamp("computed_at", { withTimezone: true }).notNull().defaultNow(), +}); diff --git a/packages/shared-types/src/analytics.ts b/packages/shared-types/src/analytics.ts index 422711c..5e30125 100644 --- a/packages/shared-types/src/analytics.ts +++ b/packages/shared-types/src/analytics.ts @@ -29,6 +29,7 @@ export const ANALYTICS_INVENTORY_EVENT_NAMES = [ "scan_uploaded", "scan_processing_completed", "scan_processing_failed", + "scan_failed", "scan_review_opened", "scan_item_confirmed", "scan_item_corrected", @@ -129,7 +130,11 @@ export const FUNNELS = { cooking_to_updated_inventory: ["cooking_session_started", "cooking_session_completed"], activated_to_trial: ["cooking_session_completed", "trial_started"], trial_to_paid: ["trial_started", "subscription_started"], - single_to_two_members: ["household_invite_sent", "household_invite_accepted", "second_member_first_action"], + single_to_two_members: [ + "household_invite_sent", + "household_invite_accepted", + "second_member_first_action", + ], } as const; export type FunnelName = keyof typeof FUNNELS; diff --git a/packages/shared-types/src/index.ts b/packages/shared-types/src/index.ts index 4406577..e4d5902 100644 --- a/packages/shared-types/src/index.ts +++ b/packages/shared-types/src/index.ts @@ -7,4 +7,5 @@ export * from "./locale.js"; export * from "./money.js"; export * from "./measurement.js"; export * from "./analytics.js"; +export * from "./ops.js"; export * from "./release-gates.js"; diff --git a/packages/shared-types/src/ops.ts b/packages/shared-types/src/ops.ts new file mode 100644 index 0000000..aa006ab --- /dev/null +++ b/packages/shared-types/src/ops.ts @@ -0,0 +1,64 @@ +export interface OpsAiScanBlock { + scans_24h: number; + scans_7d: number; + lyckandegrad_24h: number | null; + latens_p50_ms: number | null; + latens_p95_ms: number | null; + felfrekvens_24h: number | null; + senaste_fel: Array<{ kod: string | null; tid: string }> | null; + kostnad_usd_mikrocent_manad: number; + kostnad_usd_mikrocent_24h: number | null; + kostnad_usd_mikrocent_24h_uppskattad: boolean; + budget_andel: number | null; + as_of: string; +} + +export interface OpsActivationBlock { + scan_inom_24h_andel: number | null; + lagad_maltid_inom_7d_andel: number | null; + hushall_andra_medlem_aktiv_andel: number | null; + retention_d1: number | null; + retention_d7: number | null; + retention_d30: number | null; +} + +export interface OpsEngagementBlock { + lagade_maltider_24h: number; + lagade_maltider_7d: number; + rek_ctr_7d: number | null; + puffar_skickade_24h: number; + puffar_atgardsgrad_7d: number | null; +} + +export interface OpsPaymentBlock { + failed_nu: number; + grace_period_nu: number; + trials_utgar_48h: number; + aterbetalningar_7d: number | null; + chargebacks_7d: number | null; +} + +export interface OpsQueueBlock { + vantande: number; + aktiva: number; + misslyckade_24h: number; + aldsta_vantande_sek: number | null; + workers_ok: boolean; +} + +export interface OpsSafetyBlock { + allergen_invariant_brott: number; + overifierade_visade: number; + food_safety_lint_avvisade_7d: number | null; + senaste_kontroll: string | null; +} + +export interface OpsSummary { + ai_scan: OpsAiScanBlock; + aktivering: OpsActivationBlock; + engagemang: OpsEngagementBlock; + betalning: OpsPaymentBlock; + jobb: OpsQueueBlock; + sakerhet: OpsSafetyBlock; + as_of: string; +} diff --git a/pnpm-lock.yaml b/pnpm-lock.yaml index e101827..5244d15 100644 --- a/pnpm-lock.yaml +++ b/pnpm-lock.yaml @@ -266,6 +266,9 @@ importers: '@app/ai-contracts': specifier: workspace:* version: link:../../packages/ai-contracts + '@app/analytics': + specifier: workspace:* + version: link:../../packages/analytics '@app/database': specifier: workspace:* version: link:../../packages/database @@ -342,6 +345,9 @@ importers: packages/database: dependencies: + '@app/analytics': + specifier: workspace:* + version: link:../analytics '@app/nutrition-engine': specifier: workspace:* version: link:../nutrition-engine