fix(news): persist publication requests and retry cache delivery

src/lib/services/news-cache.test.ts

@@ -18,7 +18,11 @@ vi.mock("@/lib/cache", () => ({
 	},
 }));
 
-import { cacheNews, invalidateNewsCache } from "./news-cache";
+import {
+	cacheNews,
+	invalidateNewsCache,
+	refreshNewsCacheForDelivery,
+} from "./news-cache";
 
 beforeEach(() => {
 	state.values.clear();
@@ -65,3 +69,12 @@ it("reads fresh data when Redis is unavailable", async () => {
 		"fresh",
 	]);
 });
+
+it("keeps failed durable cache invalidations retryable", async () => {
+	state.set.mockRejectedValue(Error("offline"));
+	await expect(refreshNewsCacheForDelivery()).rejects.toThrow("offline");
+});
+it("does not acknowledge an ended Redis connection as refreshed", async () => {
+	state.status = "end";
+	await expect(refreshNewsCacheForDelivery()).rejects.toThrow();
+});

src/lib/services/news-cache.ts

@@ -30,3 +30,11 @@ export async function invalidateNewsCache(): Promise<void> {
 		});
 	}
 }
+
+/** Worker delivery must fail visibly so the outbox can retry the invalidation. */
+export async function refreshNewsCacheForDelivery(): Promise<void> {
+	if (!redis) return;
+	if (redis.status === "end")
+		throw new Error("News cache connection is closed");
+	await redis.set(REVISION_KEY, randomUUID());
+}

src/lib/services/news-scheduler.test.ts

@@ -0,0 +1,130 @@
+import type { SQL } from "drizzle-orm";
+import { MySqlDialect } from "drizzle-orm/mysql-core";
+import { beforeEach, expect, it, vi } from "vitest";
+
+const state = vi.hoisted(() => ({
+	articles: [] as Array<{
+		id: string;
+		userId: number | null;
+		status: string;
+		publishAt: Date;
+	}>,
+	operations: [] as unknown[][],
+	effects: [] as unknown[][],
+	failOutbox: false,
+	loseRace: false,
+	refresh: vi.fn(),
+}));
+vi.mock("@/lib/logger", () => ({ logger: { error: vi.fn() } }));
+vi.mock("./news-cache", () => ({ invalidateNewsCache: state.refresh }));
+vi.mock("@/lib/db", () => {
+	const dialect = new MySqlDialect();
+	const tx = {
+		execute: async (query: SQL) => {
+			const { sql, params } = dialect.sqlToQuery(query);
+			if (sql.startsWith("SELECT"))
+				return [
+					state.articles
+						.filter(
+							(a) =>
+								a.status === "scheduled" && a.publishAt <= (params[0] as Date),
+						)
+						.slice(0, 100),
+				];
+			if (sql.startsWith("UPDATE website_articles")) {
+				const article = state.articles.find((a) => a.id === params[2]);
+				if (
+					state.loseRace ||
+					!article ||
+					article.status !== "scheduled" ||
+					article.publishAt > (params[3] as Date)
+				)
+					return [{ affectedRows: 0 }];
+				article.status = "published";
+				return [{ affectedRows: 1 }];
+			}
+			if (sql.startsWith("INSERT INTO cms_operations"))
+				state.operations.push(params);
+			else if (sql.startsWith("INSERT INTO cms_outbox")) {
+				if (state.failOutbox) throw Error("outbox unavailable");
+				state.effects.push(params);
+			} else throw Error(`Unexpected SQL ${sql}`);
+			return [[]];
+		},
+	};
+	return {
+		db: {
+			transaction: async (work: (tx: unknown) => Promise<unknown>) => {
+				const before = structuredClone({
+					articles: state.articles,
+					operations: state.operations,
+					effects: state.effects,
+				});
+				try {
+					return await work(tx);
+				} catch (error) {
+					Object.assign(state, before);
+					throw error;
+				}
+			},
+		},
+	};
+});
+
+import { publishDueArticles } from "./news-scheduler";
+
+const now = new Date("2030-01-02T00:00:00Z");
+beforeEach(() => {
+	state.articles = [
+		{
+			id: "1",
+			userId: 0,
+			status: "scheduled",
+			publishAt: new Date("2030-01-01T00:00:00Z"),
+		},
+	];
+	state.operations = [];
+	state.effects = [];
+	state.failOutbox = false;
+	state.loseRace = false;
+	state.refresh.mockReset();
+});
+it("publishes once with the original legacy author and a durable cache refresh", async () => {
+	expect(await publishDueArticles(now)).toBe(1);
+	expect(await publishDueArticles(now)).toBe(0);
+	expect(state.articles[0].status).toBe("published");
+	expect(state.operations).toHaveLength(1);
+	expect(state.operations[0][1]).toBe(0);
+	expect(state.effects).toHaveLength(1);
+	expect(state.effects[0][2]).toBe("news.refresh");
+});
+it("rolls publication back when durable cache enqueue fails", async () => {
+	state.failOutbox = true;
+	await expect(publishDueArticles(now)).rejects.toThrow("outbox unavailable");
+	expect(state.articles[0].status).toBe("scheduled");
+	expect(state.operations).toEqual([]);
+	expect(state.effects).toEqual([]);
+});
+it("leaves future articles and concurrent changed articles alone", async () => {
+	state.articles[0].publishAt = new Date("2031-01-01T00:00:00Z");
+	expect(await publishDueArticles(now)).toBe(0);
+	state.articles[0].publishAt = new Date("2030-01-01T00:00:00Z");
+	state.loseRace = true;
+	expect(await publishDueArticles(now)).toBe(0);
+	expect(state.operations).toEqual([]);
+	expect(state.effects).toEqual([]);
+});
+it("records real authors and represents missing legacy authors as system", async () => {
+	state.articles[0].userId = null;
+	state.articles.push({ ...state.articles[0], id: "2", userId: 42 });
+	expect(await publishDueArticles(now)).toBe(2);
+	expect(state.operations.map((row) => row[1])).toEqual([0, 42]);
+});
+
+it("reports committed publication even when immediate cache refresh fails", async () => {
+	state.refresh.mockRejectedValue(Error("cache offline"));
+	expect(await publishDueArticles(now)).toBe(1);
+	expect(state.articles[0].status).toBe("published");
+	expect(state.effects).toHaveLength(1);
+	expect(state.refresh).toHaveBeenCalledOnce();
+});

src/lib/services/news-scheduler.ts

@@ -0,0 +1,52 @@
+import "server-only";
+import { randomUUID } from "node:crypto";
+import { sql } from "drizzle-orm";
+import { operationHash } from "@/features/operations/model";
+import { enqueueEffect } from "@/features/operations/server";
+import { db } from "@/lib/db";
+import { logger } from "@/lib/logger";
+import { invalidateNewsCache } from "./news-cache";
+
+/** Publish a bounded batch and persist its refresh work before committing. */
+export async function publishDueArticles(now = new Date()): Promise<number> {
+	const published = await db.transaction(async (tx) => {
+		const [rows] = await tx.execute(
+			sql`SELECT CAST(id AS CHAR) AS id,user_id AS userId,publish_at AS publishAt FROM website_articles WHERE status='scheduled' AND publish_at<=${now} ORDER BY publish_at,id LIMIT 100 FOR UPDATE`,
+		);
+		let published = 0;
+		for (const article of rows as unknown as Array<{
+			id: string;
+			userId: number | null;
+			publishAt: Date | string;
+		}>) {
+			const [result] = await tx.execute(
+				sql`UPDATE website_articles SET status='published',published_at=${now},updated_at=${now} WHERE id=${article.id} AND status='scheduled' AND publish_at<=${now}`,
+			);
+			if ((result as unknown as { affectedRows: number }).affectedRows !== 1)
+				continue;
+			const operationId = randomUUID();
+			const hash = operationHash({
+				articleId: article.id,
+				publishAt: new Date(article.publishAt).toISOString(),
+			});
+			// Legacy articles without an author remain attributed to system (0).
+			await tx.execute(
+				sql`INSERT INTO cms_operations (id,actor_id,kind,request_key,request_hash,result_json) VALUES (${operationId},${article.userId ?? 0},${"news.schedule.publish"},${randomUUID()},${hash},${JSON.stringify({ articleId: article.id, published: true })})`,
+			);
+			await enqueueEffect(tx, operationId, "news.refresh");
+			published++;
+		}
+		return published;
+	});
+	if (published > 0) {
+		try {
+			await invalidateNewsCache();
+		} catch (error) {
+			logger.error("Scheduled news published; cache refresh remains queued", {
+				module: "news",
+				error,
+			});
+		}
+	}
+	return published;
+}