Skip to content

Separa classificação do refresh e evita digest duplicado #143

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
paulocarinhena merged 1 commit into from
Nov 12, 2026
Merged

Separa classificação do refresh e evita digest duplicado #143

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
32 changes: 28 additions & 4 deletions src/lib/ai/classify.ts
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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(
Expand All @@ -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;

Expand All @@ -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.
Expand Down Expand Up @@ -76,15 +73,42 @@ ${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.
await applyRulesToArticles(scored, "classified", userId).catch((err) => console.error(`[rules] user=${userId}`, errorMessage(err)));
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;
Expand Down
28 changes: 22 additions & 6 deletions src/lib/ai/digest.ts
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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({
Expand All @@ -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");
Expand All @@ -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) {
Expand Down
22 changes: 12 additions & 10 deletions src/lib/jobs/scheduler.ts
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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";
Expand Down Expand Up @@ -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 () => {
Expand Down
Loading