feat(ops): /ops/v1/summary med ai/aktivering/engagemang/betalning/jobb/sakerhet + canary

- 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.
2026-08-10 05:47:43 +07:00
parent f4a603e977
commit 8f20b2ef1b
27 changed files with 1169 additions and 41 deletions
@@ -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",
@@ -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);
@@ -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 ---
@@ -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<OpsQueueSummary> {
+  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<void> {
+  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 };
@@ -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<SafetyCanaryResult> {
+  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<string, IngredientSafetyInfo>();
+  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<number>`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 };
+}
@@ -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<AamosTaskType>): Promise<void> {
+async function recordAiUsage(
+  ctx: WorkerContext,
+  userId: string,
+  result: AamosResult<AamosTaskType>,
+): Promise<void> {
  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<AamosTaskType>,
+): Promise<void> {
+  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<AamosTaskType>,
+): Promise<void> {
+  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,
+    },
+  });
+}
@@ -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);
+  });
+});