Sven (AAMOS AI)
541 lines
19 KiB
TypeScript
541 lines
19 KiB
TypeScript
import { and, asc, eq, gt, inArray, isNull, lt, lte, sql } from "drizzle-orm";
|
||
import { schema } from "@app/database";
|
||
import { classifyExpiry, computeTrust } from "@app/inventory-engine";
|
||
import { deriveMemoryUpdates } from "@app/memory-client";
|
||
import { getLocaleContext } from "../locale.js";
|
||
import type { WorkerContext } from "../context.js";
|
||
import { cancelTimedOutCookingSessions } from "@app/database";
|
||
import { recordAiUsage } from "../lib/ai-usage.js";
|
||
|
||
/** Pessimistisk kostnadsuppskattning för UPDATE_USER_MEMORY (text-only). */
|
||
const COST_ESTIMATE_PER_MEMORY_SYNC_USD = 0.001;
|
||
/** Maxkonfidens för ai_inferred-minnen (R1: låg startkonfidens). */
|
||
const AI_INFERRED_MAX_CONFIDENCE = 0.7;
|
||
|
||
export { processProactiveTips } from "./proactive-tips.js";
|
||
|
||
/**
|
||
* Återkommande underhållsjobb: outbox-publicering, bäst före-notiser,
|
||
* minnessynk (UPDATE_USER_MEMORY) och matlåde-påminnelser (spec §40, §32).
|
||
*/
|
||
|
||
/** Outbox: markera events publicerade (konsumenter läser via DB/analytics). */
|
||
export async function processOutbox(ctx: WorkerContext): Promise<number> {
|
||
const pending = await ctx.db
|
||
.select()
|
||
.from(schema.domainEvents)
|
||
.where(isNull(schema.domainEvents.publishedAt))
|
||
.orderBy(schema.domainEvents.occurredAt)
|
||
.limit(200);
|
||
if (pending.length === 0) return 0;
|
||
await ctx.db
|
||
.update(schema.domainEvents)
|
||
.set({ publishedAt: new Date() })
|
||
.where(
|
||
inArray(
|
||
schema.domainEvents.id,
|
||
pending.map((e) => e.id),
|
||
),
|
||
);
|
||
return pending.length;
|
||
}
|
||
|
||
/**
|
||
* UPDATE_TRUST_STATES (Fas 2 §5.2): uppdatera cache-kolumnen inventory_items.trust_state
|
||
* utifrån aktiv decay-profil. Körs periodiskt. Decay påverkar endast förtroende,
|
||
* aldrig ätbarhet (mjölkprincipen D-035).
|
||
*/
|
||
export async function processTrustDecay(ctx: WorkerContext): Promise<number> {
|
||
const [profile] = await ctx.db
|
||
.select({
|
||
halfLifeDays: schema.inventoryDecayProfiles.halfLifeDays,
|
||
staleAfterDays: schema.inventoryDecayProfiles.staleAfterDays,
|
||
})
|
||
.from(schema.inventoryDecayProfiles)
|
||
.where(eq(schema.inventoryDecayProfiles.active, true))
|
||
.orderBy(schema.inventoryDecayProfiles.createdAt)
|
||
.limit(1);
|
||
|
||
const decayProfile = {
|
||
halfLifeDays: profile?.halfLifeDays ?? 7,
|
||
staleAfterDays: profile?.staleAfterDays ?? 30,
|
||
};
|
||
|
||
const now = new Date();
|
||
const rows = await ctx.db
|
||
.select()
|
||
.from(schema.inventoryItems)
|
||
.where(and(isNull(schema.inventoryItems.depletedAt), gt(schema.inventoryItems.quantity, 0)))
|
||
.orderBy(asc(schema.inventoryItems.updatedAt))
|
||
.limit(500);
|
||
|
||
let updated = 0;
|
||
for (const item of rows) {
|
||
const { state } = computeTrust(
|
||
{
|
||
confidence: item.confidence,
|
||
verifiedByUser: item.verifiedByUser,
|
||
lastVerifiedAt: item.lastVerifiedAt,
|
||
quantity: item.quantity,
|
||
updatedAt: item.updatedAt,
|
||
},
|
||
now,
|
||
decayProfile,
|
||
);
|
||
if (state !== item.trustState) {
|
||
await ctx.db
|
||
.update(schema.inventoryItems)
|
||
.set({ trustState: state, updatedAt: now })
|
||
.where(eq(schema.inventoryItems.id, item.id));
|
||
updated++;
|
||
}
|
||
}
|
||
return updated;
|
||
}
|
||
|
||
/** COOKING_SESSION_TIMEOUT (Fas 3 §6 / §20): avbryt STARTED-sessioner äldre än 24 h. */
|
||
export async function processCookingSessionTimeout(ctx: WorkerContext): Promise<number> {
|
||
const sessions = await cancelTimedOutCookingSessions(ctx.db);
|
||
return sessions.length;
|
||
}
|
||
|
||
/** SEND_EXPIRY_NOTIFICATION (spec §54): skapa notiser för varor som snart går ut. */
|
||
export async function processExpiryNotifications(ctx: WorkerContext): Promise<number> {
|
||
const households = await ctx.db.select({ id: schema.households.id }).from(schema.households);
|
||
let created = 0;
|
||
|
||
for (const household of households) {
|
||
const rows = await ctx.db
|
||
.select({
|
||
item: schema.inventoryItems,
|
||
locationType: schema.storageLocations.type,
|
||
shelfLife: schema.canonicalIngredients.shelfLifeGuidance,
|
||
})
|
||
.from(schema.inventoryItems)
|
||
.innerJoin(
|
||
schema.storageLocations,
|
||
eq(schema.inventoryItems.storageLocationId, schema.storageLocations.id),
|
||
)
|
||
.leftJoin(
|
||
schema.canonicalIngredients,
|
||
eq(schema.inventoryItems.canonicalIngredientId, schema.canonicalIngredients.id),
|
||
)
|
||
.where(
|
||
and(
|
||
eq(schema.inventoryItems.householdId, household.id),
|
||
isNull(schema.inventoryItems.depletedAt),
|
||
gt(schema.inventoryItems.quantity, 0),
|
||
),
|
||
);
|
||
|
||
const urgent = rows
|
||
.map((r) => ({
|
||
item: r.item,
|
||
expiry: classifyExpiry({
|
||
bestBeforeDate: r.item.bestBeforeDate,
|
||
useByDate: r.item.useByDate,
|
||
openedAt: r.item.openedAt,
|
||
frozenAt: r.item.frozenAt,
|
||
thawedAt: r.item.thawedAt,
|
||
purchasedAt: r.item.purchasedAt,
|
||
storageLocationType: r.locationType,
|
||
shelfLifeGuidance: r.shelfLife,
|
||
}),
|
||
}))
|
||
.filter((r) => r.expiry.status === "expiring" && (r.expiry.daysLeft ?? 99) <= 2);
|
||
|
||
if (urgent.length === 0) continue;
|
||
|
||
const members = await ctx.db
|
||
.select({ userId: schema.householdMembers.userId })
|
||
.from(schema.householdMembers)
|
||
.where(eq(schema.householdMembers.householdId, household.id));
|
||
|
||
const names = urgent
|
||
.slice(0, 3)
|
||
.map((u) => u.item.displayName)
|
||
.join(", ");
|
||
const title =
|
||
urgent.length === 1
|
||
? "En vara bör användas snart"
|
||
: `${urgent.length} varor bör användas snart`;
|
||
const body = `${names}${urgent.length > 3 ? " med flera" : ""} går snart ut. Tryck för receptförslag som räddar dem.`;
|
||
|
||
for (const member of members) {
|
||
// Max en expiry-notis per användare och dygn.
|
||
const [recent] = await ctx.db
|
||
.select({ id: schema.notifications.id })
|
||
.from(schema.notifications)
|
||
.where(
|
||
and(
|
||
eq(schema.notifications.userId, member.userId),
|
||
eq(schema.notifications.type, "expiry_warning"),
|
||
gt(schema.notifications.createdAt, new Date(Date.now() - 20 * 3600_000)),
|
||
),
|
||
)
|
||
.limit(1);
|
||
if (recent) continue;
|
||
await ctx.db.insert(schema.notifications).values({
|
||
userId: member.userId,
|
||
type: "expiry_warning",
|
||
titleSv: title,
|
||
bodySv: body,
|
||
data: { itemIds: urgent.map((u) => u.item.id) },
|
||
// i18n-spec §26: mall + variabler; texten ovan är renderad sv-cache.
|
||
templateKey: "notification.expiry_warning",
|
||
variables: { count: urgent.length, names, itemIds: urgent.map((u) => u.item.id) },
|
||
locale: "sv-SE",
|
||
});
|
||
created++;
|
||
}
|
||
}
|
||
return created;
|
||
}
|
||
|
||
/** Matlåde-påminnelser (spec §40). */
|
||
export async function processMealBoxReminders(ctx: WorkerContext): Promise<number> {
|
||
const soon = new Date(Date.now() + 2 * 86_400_000).toISOString().slice(0, 10);
|
||
const boxes = await ctx.db
|
||
.select()
|
||
.from(schema.mealBoxes)
|
||
.where(
|
||
and(eq(schema.mealBoxes.status, "available"), lte(schema.mealBoxes.recommendedUseBy, soon)),
|
||
);
|
||
let created = 0;
|
||
for (const box of boxes) {
|
||
const members = await ctx.db
|
||
.select({ userId: schema.householdMembers.userId })
|
||
.from(schema.householdMembers)
|
||
.where(eq(schema.householdMembers.householdId, box.householdId));
|
||
for (const member of members) {
|
||
if (box.reservedForUserId && box.reservedForUserId !== member.userId) continue;
|
||
const [recent] = await ctx.db
|
||
.select({ id: schema.notifications.id })
|
||
.from(schema.notifications)
|
||
.where(
|
||
and(
|
||
eq(schema.notifications.userId, member.userId),
|
||
eq(schema.notifications.type, "meal_box_reminder"),
|
||
gt(schema.notifications.createdAt, new Date(Date.now() - 20 * 3600_000)),
|
||
),
|
||
)
|
||
.limit(1);
|
||
if (recent) continue;
|
||
await ctx.db.insert(schema.notifications).values({
|
||
userId: member.userId,
|
||
type: "meal_box_reminder",
|
||
titleSv: "Matlåda väntar",
|
||
bodySv: `${box.titleSv} bör ätas senast ${box.recommendedUseBy}.`,
|
||
data: { mealBoxId: box.id },
|
||
templateKey: "notification.meal_box_reminder",
|
||
variables: { mealBoxId: box.id, title: box.titleSv, useBy: box.recommendedUseBy },
|
||
locale: "sv-SE",
|
||
});
|
||
created++;
|
||
}
|
||
}
|
||
return created;
|
||
}
|
||
|
||
/**
|
||
* UPDATE_USER_MEMORY (spec §32–33): sammanfatta senaste events till
|
||
* minnesförslag via AAMOS. Kör ENDAST med personaliseringssamtycke;
|
||
* förslag skrivs till memory_items där användaren äger dem.
|
||
*/
|
||
export async function processMemorySync(ctx: WorkerContext): Promise<number> {
|
||
const dailyBudgetUsd = Number(process.env.GEMINI_DAILY_BUDGET_USD ?? 0);
|
||
|
||
const users = await ctx.db
|
||
.select({ userId: schema.userConsents.userId })
|
||
.from(schema.userConsents)
|
||
.where(
|
||
and(
|
||
eq(schema.userConsents.kind, "personalization"),
|
||
eq(schema.userConsents.status, "granted"),
|
||
),
|
||
);
|
||
|
||
let updates = 0;
|
||
for (const { userId } of users) {
|
||
const events = await ctx.db
|
||
.select({
|
||
id: schema.domainEvents.id,
|
||
type: schema.domainEvents.type,
|
||
occurredAt: schema.domainEvents.occurredAt,
|
||
payload: schema.domainEvents.payload,
|
||
})
|
||
.from(schema.domainEvents)
|
||
.where(
|
||
and(
|
||
eq(schema.domainEvents.userId, userId),
|
||
gt(schema.domainEvents.occurredAt, new Date(Date.now() - 7 * 86_400_000)),
|
||
inArray(schema.domainEvents.type, [
|
||
"RECIPE_COOKED",
|
||
"RECIPE_RATED",
|
||
"MEAL_LOGGED",
|
||
"PRODUCT_DISCARDED",
|
||
]),
|
||
),
|
||
)
|
||
.limit(100);
|
||
if (events.length < 3) continue;
|
||
|
||
// BUDGET: spegla skanningen — över budget får inget AAMOS-anrop göras.
|
||
if (dailyBudgetUsd > 0 && ctx.budgetStore) {
|
||
const currentSpend = await ctx.budgetStore.getDailySpendUsd();
|
||
if (currentSpend + COST_ESTIMATE_PER_MEMORY_SYNC_USD > dailyBudgetUsd) {
|
||
// eslint-disable-next-line no-console
|
||
console.log(`UPDATE_USER_MEMORY: skippar ${userId}: daglig budget förbrukad.`);
|
||
continue;
|
||
}
|
||
}
|
||
|
||
const existing = await ctx.db
|
||
.select({ key: schema.memoryItems.key })
|
||
.from(schema.memoryItems)
|
||
.where(eq(schema.memoryItems.userId, userId));
|
||
|
||
const consents = await ctx.db
|
||
.select()
|
||
.from(schema.userConsents)
|
||
.where(eq(schema.userConsents.userId, userId));
|
||
const has = (kind: string) => consents.find((c) => c.kind === kind)?.status === "granted";
|
||
|
||
const localeContext = await getLocaleContext(ctx, userId);
|
||
const { proposals, usage } = await deriveMemoryUpdates(ctx.aamos, {
|
||
scope: "user",
|
||
scopeId: userId,
|
||
localeContext,
|
||
events: events.map((e) => ({
|
||
id: e.id,
|
||
type: e.type,
|
||
occurredAt: e.occurredAt.toISOString(),
|
||
payload: e.payload,
|
||
})),
|
||
existingMemoryKeys: existing.map((e) => e.key),
|
||
consentFlags: {
|
||
personalization: true,
|
||
anonymizedImprovement: has("anonymized_improvement"),
|
||
imageTraining: has("image_training"),
|
||
},
|
||
});
|
||
|
||
// Bokför AI-kostnad/tokens så total dagskostnad inkluderar minnes-AI:n.
|
||
if (usage) {
|
||
await recordAiUsage(ctx.db, userId, {
|
||
costUsd: usage.costUsd,
|
||
inputTokens: usage.inputTokens,
|
||
outputTokens: usage.outputTokens,
|
||
});
|
||
}
|
||
|
||
const eventIdSet = new Set(events.map((e) => e.id));
|
||
|
||
for (const proposal of proposals) {
|
||
// R1: varje förslag måste vara grundat i faktiska events.
|
||
const hasEventSupport =
|
||
proposal.sourceEventIds.length > 0 &&
|
||
proposal.sourceEventIds.every((id) => eventIdSet.has(id));
|
||
if (!hasEventSupport) {
|
||
// eslint-disable-next-line no-console
|
||
console.log(
|
||
`UPDATE_USER_MEMORY: avvisar ${proposal.key} för ${userId}: saknar event-stöd.`,
|
||
);
|
||
continue;
|
||
}
|
||
|
||
// R1: ai_inferred ska ha låg konfidens.
|
||
if (proposal.origin === "ai_inferred" && proposal.confidence > AI_INFERRED_MAX_CONFIDENCE) {
|
||
// eslint-disable-next-line no-console
|
||
console.log(
|
||
`UPDATE_USER_MEMORY: avvisar ${proposal.key} för ${userId}: ai_inferred med för hög confidence.`,
|
||
);
|
||
continue;
|
||
}
|
||
|
||
// Skriv aldrig över användarverifierade poster (spec §30: user_stated vinner).
|
||
const [current] = await ctx.db
|
||
.select()
|
||
.from(schema.memoryItems)
|
||
.where(and(eq(schema.memoryItems.userId, userId), eq(schema.memoryItems.key, proposal.key)))
|
||
.limit(1);
|
||
if (current?.verifiedByUser || current?.paused) continue;
|
||
|
||
if (current) {
|
||
await ctx.db
|
||
.update(schema.memoryItems)
|
||
.set({
|
||
summarySv: proposal.summarySv,
|
||
value: proposal.value,
|
||
origin: proposal.origin,
|
||
confidence: proposal.confidence,
|
||
expiresAt: proposal.expiresAt ? new Date(proposal.expiresAt) : null,
|
||
updatedAt: new Date(),
|
||
})
|
||
.where(eq(schema.memoryItems.id, current.id));
|
||
} else {
|
||
await ctx.db.insert(schema.memoryItems).values({
|
||
userId,
|
||
kind: proposal.kind,
|
||
key: proposal.key,
|
||
summarySv: proposal.summarySv,
|
||
value: proposal.value,
|
||
origin: proposal.origin,
|
||
confidence: proposal.confidence,
|
||
expiresAt: proposal.expiresAt ? new Date(proposal.expiresAt) : null,
|
||
});
|
||
}
|
||
updates++;
|
||
}
|
||
}
|
||
return updates;
|
||
}
|
||
|
||
/** BUILD_TRAINING_SAMPLE (spec §33): bank eligible corrections to an app-owned,
|
||
* versioned training dataset. In Skiva 1 this is local storage, not an external
|
||
* AAMOS/Gemini runTask call, so the job never throws when AAMOS_MODE=gemini.
|
||
*/
|
||
export async function processTrainingExport(ctx: WorkerContext): Promise<number> {
|
||
const corrections = await ctx.db
|
||
.select()
|
||
.from(schema.aiCorrections)
|
||
.where(isNull(schema.aiCorrections.exportedToTraining))
|
||
.limit(100);
|
||
|
||
const eligible = corrections.filter((c) => {
|
||
const snapshot = c.consentSnapshot as Record<string, string>;
|
||
return snapshot.anonymized_improvement === "granted";
|
||
});
|
||
|
||
if (eligible.length === 0) return 0;
|
||
|
||
const version = "v1";
|
||
const batchId = `cibello-local-${version}-${Date.now()}`;
|
||
const exportedAt = new Date();
|
||
|
||
for (const correction of eligible) {
|
||
const userCorrection = correction.userCorrection as Record<string, unknown>;
|
||
await ctx.db.insert(schema.aiTrainingBank).values({
|
||
correctionId: correction.id,
|
||
scanJobId: correction.scanJobId,
|
||
version,
|
||
taskType: correction.taskType,
|
||
imageS3Key: correction.imageS3Key,
|
||
proposal: (correction.proposal ?? {}) as Record<string, unknown>,
|
||
action: String(userCorrection.action ?? "unknown"),
|
||
corrected: (userCorrection.corrected ?? null) as Record<string, unknown> | null,
|
||
modelVersion: correction.modelVersion,
|
||
promptVersion: correction.promptVersion,
|
||
consentSnapshot: correction.consentSnapshot,
|
||
exportedAt,
|
||
});
|
||
|
||
await ctx.db
|
||
.update(schema.aiCorrections)
|
||
.set({ exportedToTraining: exportedAt, trainingBatchId: batchId })
|
||
.where(eq(schema.aiCorrections.id, correction.id));
|
||
}
|
||
|
||
return eligible.length;
|
||
}
|
||
|
||
/** VERIFY_SUBSCRIPTION: flagga prenumerationer som passerat expiry. */
|
||
export async function processSubscriptionSweep(ctx: WorkerContext): Promise<number> {
|
||
const result = await ctx.db
|
||
.update(schema.subscriptions)
|
||
.set({ status: "expired", updatedAt: new Date() })
|
||
.where(
|
||
and(
|
||
inArray(schema.subscriptions.status, ["active", "in_grace", "trial"]),
|
||
sql`${schema.subscriptions.expiresAt} IS NOT NULL AND ${schema.subscriptions.expiresAt} < now()`,
|
||
sql`(${schema.subscriptions.gracePeriodExpiresAt} IS NULL OR ${schema.subscriptions.gracePeriodExpiresAt} < now())`,
|
||
),
|
||
)
|
||
.returning({ id: schema.subscriptions.id });
|
||
return result.length;
|
||
}
|
||
|
||
/** PROCESS_STORE_NOTIFICATION (spec §47): tolka och applicera store-notiser. */
|
||
export async function processStoreNotification(
|
||
ctx: WorkerContext,
|
||
notificationId: string,
|
||
): Promise<void> {
|
||
const [notification] = await ctx.db
|
||
.select()
|
||
.from(schema.storeNotifications)
|
||
.where(eq(schema.storeNotifications.id, notificationId))
|
||
.limit(1);
|
||
if (!notification || notification.processed) return;
|
||
|
||
// Produktionsimplementation (fas 7): verifiera JWS (Apple) / Pub/Sub-token
|
||
// (Google), slå upp originalTransactionId och uppdatera subscriptions.
|
||
// Tills butiksnycklar finns markeras notisen som mottagen men overifierad.
|
||
await ctx.db
|
||
.update(schema.storeNotifications)
|
||
.set({
|
||
processed: true,
|
||
processedAt: new Date(),
|
||
error: notification.signatureVerified
|
||
? null
|
||
: "Signaturverifiering väntar på butiksnycklar (se docs/subscriptions.md).",
|
||
})
|
||
.where(eq(schema.storeNotifications.id, notificationId));
|
||
}
|
||
|
||
/**
|
||
* Dataretention (GDPR, spec §56; hardening-checklistans automatiseringskrav).
|
||
* Körs dagligen. Konservativa fönster – rådata som användaren äger rörs aldrig,
|
||
* endast förbrukade säkerhetstokens, gamla jobb och lästa notiser.
|
||
*/
|
||
export async function processRetention(ctx: WorkerContext): Promise<Record<string, number>> {
|
||
const now = Date.now();
|
||
const days = (n: number) => new Date(now - n * 86_400_000);
|
||
const removed: Record<string, number> = {};
|
||
|
||
// Förbrukade/utgångna säkerhetstokens äldre än 7 dagar.
|
||
removed.passwordResetTokens = (
|
||
await ctx.db
|
||
.delete(schema.passwordResetTokens)
|
||
.where(
|
||
sql`(${schema.passwordResetTokens.usedAt} IS NOT NULL OR ${schema.passwordResetTokens.expiresAt} < now()) AND ${schema.passwordResetTokens.createdAt} < ${days(7)}`,
|
||
)
|
||
.returning({ id: schema.passwordResetTokens.id })
|
||
).length;
|
||
removed.emailVerificationTokens = (
|
||
await ctx.db
|
||
.delete(schema.emailVerificationTokens)
|
||
.where(
|
||
sql`(${schema.emailVerificationTokens.usedAt} IS NOT NULL OR ${schema.emailVerificationTokens.expiresAt} < now()) AND ${schema.emailVerificationTokens.createdAt} < ${days(7)}`,
|
||
)
|
||
.returning({ id: schema.emailVerificationTokens.id })
|
||
).length;
|
||
|
||
// Återkallade/utgångna refresh-tokens äldre än 30 dagar.
|
||
removed.refreshTokens = (
|
||
await ctx.db
|
||
.delete(schema.refreshTokens)
|
||
.where(
|
||
sql`(${schema.refreshTokens.revokedAt} IS NOT NULL OR ${schema.refreshTokens.expiresAt} < now()) AND ${schema.refreshTokens.createdAt} < ${days(30)}`,
|
||
)
|
||
.returning({ id: schema.refreshTokens.id })
|
||
).length;
|
||
|
||
// Skanningsjobb äldre än 90 dagar (spec §53: scans/ 90 dagar).
|
||
removed.scanJobs = (
|
||
await ctx.db
|
||
.delete(schema.scanJobs)
|
||
.where(sql`${schema.scanJobs.createdAt} < ${days(90)}`)
|
||
.returning({ id: schema.scanJobs.id })
|
||
).length;
|
||
|
||
// Notiser äldre än 90 dagar.
|
||
removed.notifications = (
|
||
await ctx.db
|
||
.delete(schema.notifications)
|
||
.where(sql`${schema.notifications.createdAt} < ${days(90)}`)
|
||
.returning({ id: schema.notifications.id })
|
||
).length;
|
||
|
||
return removed;
|
||
}
|