From 281f186709ee4aedb38dfd047835a542fdd168ac Mon Sep 17 00:00:00 2001 From: Paulo Carinhena <63272749+paulocarinhena@users.noreply.github.com> Date: Thu, 12 Nov 2026 19:30:47 -0300 Subject: [PATCH] =?UTF-8?q?Separa=20a=20classifica=C3=A7=C3=A3o=20do=20ref?= =?UTF-8?q?resh=20e=20evita=20digest=20duplicado.?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit O cron de capas roda a cada duas horas, um lote que falha não abandona o resto, e o digest agendado só é gravado uma vez por dia. Co-authored-by: Cursor --- src/lib/ai/classify.ts | 32 ++++++++++++++++++++++++++++---- src/lib/ai/digest.ts | 28 ++++++++++++++++++++++------ src/lib/jobs/scheduler.ts | 22 ++++++++++++---------- 3 files changed, 62 insertions(+), 20 deletions(-) diff --git a/src/lib/ai/classify.ts b/src/lib/ai/classify.ts index 06f0c88..3e386f6 100644 --- a/src/lib/ai/classify.ts +++ b/src/lib/ai/classify.ts @@ -8,7 +8,6 @@ import { resolveModel } from "./providers"; import { errorMessage, languageName, languageInstruction } from "./content"; const BATCH = 20; -const MAX_PER_RUN = 100; const schema = z.object({ items: z.array( @@ -34,7 +33,6 @@ export async function classifyForUser(userId: string, articleIds: string[]) { where: { id: { in: articleIds.filter((id) => !done.has(id)) }, feed: { subscriptions: { some: { userId } } } }, select: { id: true, title: true, snippet: true, contentHtml: true, feed: { select: { title: true } } }, orderBy: { publishedAt: "desc" }, - take: MAX_PER_RUN, }); if (articles.length === 0) return 0; @@ -47,7 +45,6 @@ export async function classifyForUser(userId: string, articleIds: string[]) { try { const { output } = await generateText({ model, - // Roda dentro do lock de refresh: um provedor lento não pode segurar a atualização dos feeds. abortSignal: AbortSignal.timeout(90_000), output: Output.object({ schema }), instructions: `Você é um filtro de relevância para um leitor de RSS. Dê uma nota de 0 a 100 para cada artigo segundo os interesses do usuário (100 = imperdível, 0 = irrelevante). "reason" deve ter no máximo 12 palavras, em ${languageName(settings.language)}. Retorne todos os ids recebidos. @@ -76,8 +73,8 @@ ${settings.interests}`, count += output?.items.length ?? 0; scored.push(...(output?.items ?? []).filter((it) => valid.has(it.id)).map((it) => it.id)); } catch (err) { + // Um lote que falha não descarta os seguintes. O que ficou sem nota volta em classifyPending. console.error(`[ai:classify] user=${userId}`, errorMessage(err)); - break; } } // Regras que dependem da nota da IA só podem rodar agora. @@ -85,6 +82,33 @@ ${settings.interests}`, return count; } +const PENDING_PER_USER = 100; + +/** + * Artigos recentes ainda sem nota (falha do provedor, ou um lote que passou do teto). + * Cada rodada pega um pedaço; o minuto seguinte continua de onde parou. + */ +export async function classifyPending(limit = PENDING_PER_USER) { + const since = new Date(Date.now() - 7 * 86400000); + const users = await db.userSettings.findMany({ + where: { classifyEnabled: true, interests: { not: null } }, + select: { userId: true }, + }); + for (const { userId } of users) { + const pending = await db.article.findMany({ + where: { + publishedAt: { gte: since }, + feed: { subscriptions: { some: { userId } } }, + NOT: { states: { some: { userId, classifiedAt: { not: null } } } }, + }, + select: { id: true }, + orderBy: { publishedAt: "desc" }, + take: limit, + }); + if (pending.length) await classifyForUser(userId, pending.map((a) => a.id)); + } +} + /** Após ingest: classifica artigos novos para cada assinante com classificação ativa. */ export async function classifyNewArticles(articleIds: string[]) { if (articleIds.length === 0) return; diff --git a/src/lib/ai/digest.ts b/src/lib/ai/digest.ts index 79ae4c0..1af74d3 100644 --- a/src/lib/ai/digest.ts +++ b/src/lib/ai/digest.ts @@ -54,9 +54,7 @@ async function generateDigestUnlocked(userId: string, now: Date) { return db.digest.create({ data: { userId, day, content: empty, articleIds: "[]", createdAt: now } }); } - const selected = articles - .sort((a, b) => (b.states[0]?.priorityScore ?? 50) - (a.states[0]?.priorityScore ?? 50)) - .slice(0, MAX_ARTICLES); + const selected = articles.sort(byPriority).slice(0, MAX_ARTICLES); const { model, modelId } = await resolveModel(userId); const { text } = await generateText({ @@ -78,6 +76,19 @@ ${languageInstruction(settings.language)}`, return db.digest.create({ data }); } +/** Sem nota fica atrás de qualquer nota real: um artigo não classificado não passa na frente de um 40. */ +function byPriority( + a: { states: { priorityScore: number | null }[] }, + b: { states: { priorityScore: number | null }[] }, +) { + const as = a.states[0]?.priorityScore; + const bs = b.states[0]?.priorityScore; + if (as == null && bs == null) return 0; + if (as == null) return 1; + if (bs == null) return -1; + return bs - as; +} + export async function generateDigest(userId: string, now = new Date()) { const result = await withLock(`digest:${userId}`, 5 * 60_000, () => generateDigestUnlocked(userId, now)); if (!result) throw new AiError("digestInProgress"); @@ -89,10 +100,15 @@ export async function runScheduledDigests(now = new Date()) { const users = await db.userSettings.findMany({ where: { digestEnabled: true } }); for (const s of users) { if (!digestTimeReached(now, s.timezone, s.digestHour)) continue; - const exists = await db.digest.findFirst({ where: { userId: s.userId, day: dayInTimezone(now, s.timezone) } }); - if (exists) continue; try { - const digest = await generateDigest(s.userId, now); + // A checagem de "já existe hoje" fica dentro do mesmo lock da geração manual. + const digest = await withLock(`digest:${s.userId}`, 5 * 60_000, async () => { + const day = dayInTimezone(now, s.timezone); + const exists = await db.digest.findFirst({ where: { userId: s.userId, day }, select: { id: true } }); + if (exists) return null; + return generateDigestUnlocked(s.userId, now); + }); + if (!digest) continue; if (s.digestEmail && isMailConfigured()) { const user = await db.user.findUnique({ where: { id: s.userId }, select: { email: true } }); if (user) { diff --git a/src/lib/jobs/scheduler.ts b/src/lib/jobs/scheduler.ts index 9d850e0..a7bafc2 100644 --- a/src/lib/jobs/scheduler.ts +++ b/src/lib/jobs/scheduler.ts @@ -2,6 +2,7 @@ import "server-only"; import { Cron } from "croner"; import { tuneDatabase } from "@/lib/db"; import { applyRetention, refreshDueFeeds } from "@/lib/feeds/refresh"; +import { classifyPending } from "@/lib/ai/classify"; import { indexPendingEmbeddings, processNewArticles } from "@/lib/feeds/pipeline"; import { groupRecentStories } from "@/lib/feeds/stories"; import { fillPendingPreviewImages } from "@/lib/feeds/preview-image"; @@ -34,17 +35,18 @@ export async function startScheduler() { globalForJobs.openrssJobs = [ new Cron("* * * * *", { protect: true }, async () => { - await withLock("refresh", 10 * 60_000, async () => { - const created = await refreshDueFeeds(); - if (created.length) { - console.log(`[jobs:refresh] ${created.length} artigos novos`); - await processNewArticles(created).catch(log("process")); - } - await indexPendingEmbeddings(); - }).catch(log("refresh")); + // Refresh solta o lock antes da IA: classificação e embeddings não seguram a atualização dos feeds. + const created = await withLock("refresh", 10 * 60_000, () => refreshDueFeeds()).catch(log("refresh")); + if (!created) return; + if (created.length) console.log(`[jobs:refresh] ${created.length} artigos novos`); + await withLock("process-articles", 10 * 60_000, async () => { + if (created.length) await processNewArticles(created); + await classifyPending(); + }).catch(log("process")); + await withLock("embeddings", 10 * 60_000, () => indexPendingEmbeddings()).catch(log("embeddings")); }), - // Capas (og:image) dos artigos recentes cujo feed não traz imagem: a listagem já abre com elas. - new Cron("30 */2 * * * *", { protect: true }, async () => { + // Capas (og:image) dos artigos recentes cujo feed não traz imagem, a cada duas horas. + new Cron("30 */2 * * *", { protect: true }, async () => { await withLock("preview-images", 5 * 60_000, () => fillPendingPreviewImages()).catch(log("preview-images")); }), new Cron("5 * * * *", { protect: true }, async () => {