Sven (AAMOS AI)
487 lines
16 KiB
TypeScript
487 lines
16 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";
|
||
|
||
/**
|
||
* Å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 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()
|
||
.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;
|
||
|
||
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 = await deriveMemoryUpdates(ctx.aamos, {
|
||
scope: "user",
|
||
scopeId: userId,
|
||
localeContext,
|
||
events: events.map((e) => ({
|
||
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"),
|
||
},
|
||
});
|
||
|
||
for (const proposal of proposals) {
|
||
// 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): exportera korrigeringar MED samtycke till AAMOS. */
|
||
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 result = await ctx.aamos.runTask(
|
||
"EXPORT_TRAINING_SAMPLE",
|
||
{
|
||
marketLocale: "sv-SE",
|
||
samples: eligible.map((c) => ({
|
||
taskType: c.taskType,
|
||
aiOutput: c.aiOutput as Record<string, unknown>,
|
||
userCorrection: c.userCorrection as Record<string, unknown>,
|
||
modelVersion: c.modelVersion ?? null,
|
||
promptVersion: c.promptVersion ?? null,
|
||
})),
|
||
},
|
||
{
|
||
correlationId: `training-export-${Date.now()}`,
|
||
consentFlags: {
|
||
personalization: false,
|
||
anonymizedImprovement: true,
|
||
imageTraining: false,
|
||
},
|
||
},
|
||
);
|
||
|
||
if (result.status !== "ok" || !result.output) {
|
||
throw new Error(`AAMOS training export failed: ${result.error ?? "unknown"}`);
|
||
}
|
||
|
||
const batchId = result.output.batchId;
|
||
for (const correction of eligible) {
|
||
await ctx.db
|
||
.update(schema.aiCorrections)
|
||
.set({ exportedToTraining: new Date(), 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;
|
||
}
|