fix(news): preserve recoverable reads and verify publication against real services

integration/database.test.ts

@@ -1,10 +1,11 @@
 import { execFile } from "node:child_process";
 import { randomUUID } from "node:crypto";
+import { once } from "node:events";
 import { copyFile, mkdir, mkdtemp, rm } from "node:fs/promises";
 import { join, resolve } from "node:path";
 import { pathToFileURL } from "node:url";
 import { promisify } from "node:util";
-import { sql } from "drizzle-orm";
+import { eq, sql } from "drizzle-orm";
 import type { RowDataPacket } from "mysql2/promise";
 import mysql from "mysql2/promise";
 import {
@@ -12,10 +13,39 @@ import {
 	type StartedTestContainer,
 	Wait,
 } from "testcontainers";
-import { afterAll, beforeAll, beforeEach, describe, expect, it } from "vitest";
+import {
+	afterAll,
+	beforeAll,
+	beforeEach,
+	describe,
+	expect,
+	it,
+	vi,
+} from "vitest";
+import { WebsiteArticles } from "@/db/schema";
+import { articleEditToken } from "@/lib/article-edit-token";
+
+// Only request/framework boundaries are replaced; persistence, caches and workers are real.
+const boundaries = vi.hoisted(() => ({ notify: vi.fn() }));
+vi.mock("@/lib/admin/guard", () => ({
+	requirePermission: async () => ({ id: 7, username: "Integration editor" }),
+}));
+vi.mock("next-intl/server", () => ({
+	getTranslations: async () => (key: string) => key,
+}));
+vi.mock("next/cache", () => ({ revalidatePath: vi.fn() }));
+vi.mock("next/navigation", () => ({
+	redirect: (url: string) => {
+		throw Error(`Unexpected integration redirect: ${url}`);
+	},
+}));
+vi.mock("@/lib/services/webhook", () => ({ notify: boundaries.notify }));
 
 const exec = promisify(execFile);
+const NEWS_REVISION_KEY = "cms:news:revision";
 const migrations = [
+	"0025_article_publication.sql",
+	"0026_article_editor_recovery.sql",
 	"0027_catalog_packages.sql",
 	"0028_history_snapshots.sql",
 	"0029_admin_table_views.sql",
@@ -29,6 +59,10 @@ let appRedis: typeof import("@/lib/redis").redis;
 let commands: typeof import("@/features/catalog/server/bulk-offers");
 let operations: typeof import("@/features/operations/server");
 let cache: typeof import("@/lib/cache");
+let articles: typeof import("@/actions/admin-articles");
+let publicNews: typeof import("@/lib/services/news-detail");
+let scheduler: typeof import("@/lib/services/news-scheduler");
+let worker: typeof import("@/features/operations/worker");
 let migrationRoot: string | undefined;
 let databaseUrl: string;
 
@@ -85,7 +119,19 @@ beforeAll(async () => {
 		.withStartupTimeout(120_000)
 		.start();
 	databaseUrl = `mysql://integration:${databasePassword}@${maria.getHost()}:${maria.getMappedPort(3306)}/integration`;
-	connection = await mysql.createConnection(databaseUrl);
+	// Inspect stored TIMESTAMP values as UTC independently of the host timezone.
+	// The application pool retains its production configuration.
+	connection = await mysql.createConnection({
+		host: maria.getHost(),
+		port: maria.getMappedPort(3306),
+		user: "integration",
+		password: databasePassword,
+		database: "integration",
+		timezone: "Z",
+		supportBigNumbers: true,
+		bigNumberStrings: true,
+		charset: "utf8mb4",
+	});
 	redisContainer = await new GenericContainer("redis:7.4.2-alpine")
 		.withCommand(["redis-server", "--requirepass", redisPassword])
 		.withExposedPorts(6379)
@@ -107,6 +153,9 @@ beforeAll(async () => {
 	await connection.query(
 		"CREATE TABLE admin_audit_log (id INT AUTO_INCREMENT PRIMARY KEY, user_id INT NOT NULL, action VARCHAR(191) NOT NULL DEFAULT '', target VARCHAR(191) NOT NULL DEFAULT '', target_id INT NULL, details TEXT NULL, `before` TEXT NULL, `after` TEXT NULL, diff TEXT NULL, ip_address VARCHAR(45) NULL, created_at VARCHAR(64) NOT NULL DEFAULT '', updated_at VARCHAR(64) NULL) ENGINE=InnoDB",
 	);
+	await connection.query(
+		"CREATE TABLE website_articles (id BIGINT UNSIGNED AUTO_INCREMENT PRIMARY KEY, slug VARCHAR(255) NOT NULL UNIQUE, title VARCHAR(255) NOT NULL, short_story VARCHAR(255) NOT NULL, full_story LONGTEXT NOT NULL, user_id INT NULL, image VARCHAR(255) NOT NULL, created_at TIMESTAMP NULL DEFAULT NULL, updated_at TIMESTAMP NULL DEFAULT NULL) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4",
+	);
 	// The emulator owns core tables. Exercise the real CMS migration CLI over this baseline.
 	migrationRoot = await mkdtemp(join(resolve("integration"), ".migration-"));
 	await mkdir(join(migrationRoot, "scripts"));
@@ -135,6 +184,10 @@ beforeAll(async () => {
 	commands = await import("@/features/catalog/server/bulk-offers");
 	cache = await import("@/lib/cache");
 	operations = await import("@/features/operations/server");
+	articles = await import("@/actions/admin-articles");
+	publicNews = await import("@/lib/services/news-detail");
+	scheduler = await import("@/lib/services/news-scheduler");
+	worker = await import("@/features/operations/worker");
 });
 
 afterAll(async () => {
@@ -163,6 +216,15 @@ afterAll(async () => {
 
 beforeEach(async () => {
 	if (!connection) throw Error("Integration database is not connected");
+	vi.clearAllMocks();
+	await appRedis?.set(NEWS_REVISION_KEY, randomUUID());
+	await connection.query("DROP TRIGGER IF EXISTS reject_news_effect");
+	await connection.query(
+		"DROP TRIGGER IF EXISTS reject_second_scheduled_effect",
+	);
+	await connection.query("DELETE FROM website_article_revisions");
+	await connection.query("DELETE FROM website_article_drafts");
+	await connection.query("DELETE FROM website_articles");
 	await connection.query("DELETE FROM cms_outbox");
 	await connection.query("DELETE FROM cms_operations");
 	await connection.query("DROP TRIGGER IF EXISTS reject_second_history");
@@ -201,7 +263,7 @@ describe("MariaDB migrations and catalog transactions", () => {
 			{ name: "Saved" },
 		]);
 		expect((await migrate("--status")).stdout).toContain(
-			"4/4 applied, 0 pending",
+			`${migrations.length}/${migrations.length} applied, 0 pending`,
 		);
 		const columns = await rows(
 			"SELECT COLUMN_NAME, DATA_TYPE FROM information_schema.COLUMNS WHERE TABLE_SCHEMA=DATABASE() AND TABLE_NAME='admin_audit_log' AND COLUMN_NAME IN ('before','after') ORDER BY COLUMN_NAME",
@@ -419,3 +481,367 @@ describe("operation idempotency and transactional outbox", () => {
 		]);
 	});
 });
+
+function articleForm(extra: Record<string, string> = {}) {
+	const form = new FormData();
+	for (const [key, value] of Object.entries({
+		requestKey: randomUUID(),
+		title: "Integration news",
+		slug: "integration-news",
+		shortStory: "A real database publication",
+		fullStory: "<p>Original draft body</p>",
+		image: "/images/news.png",
+		status: "draft",
+		...extra,
+	}))
+		form.set(key, value);
+	return form;
+}
+
+async function savedArticle(slug = "integration-news") {
+	if (!appDb) throw Error("Integration database is not connected");
+	const [article] = await appDb
+		.select()
+		.from(WebsiteArticles)
+		.where(eq(WebsiteArticles.slug, slug))
+		.limit(1);
+	if (!article) throw Error(`Missing article fixture: ${slug}`);
+	return article;
+}
+
+async function seedScheduledArticles(now: Date) {
+	if (!appDb) throw Error("Integration database is not connected");
+	const before = new Date(now.getTime() - 60_000);
+	const after = new Date(now.getTime() + 60_000);
+	await appDb.insert(WebsiteArticles).values(
+		[
+			{
+				id: 1n,
+				slug: "due-earlier",
+				status: "scheduled",
+				publishAt: before,
+				userId: 7,
+			},
+			{
+				id: 9007199254740993n,
+				slug: "due-now",
+				status: "scheduled",
+				publishAt: now,
+				userId: null,
+			},
+			{
+				id: 3n,
+				slug: "future",
+				status: "scheduled",
+				publishAt: after,
+				userId: 7,
+			},
+			{ id: 4n, slug: "draft", status: "draft", publishAt: before, userId: 7 },
+			{
+				id: 5n,
+				slug: "already-public",
+				status: "published",
+				publishAt: before,
+				userId: 7,
+			},
+		].map((article) => ({
+			...article,
+			title: article.slug,
+			shortStory: "Scheduled integration fixture",
+			fullStory: `<p>${article.slug}</p>`,
+			image: "",
+			createdAt: before,
+			updatedAt: before,
+			publishedAt: article.status === "published" ? before : null,
+		})),
+	);
+}
+
+describe("real news publication, scheduling and cache delivery", () => {
+	it("creates one draft and publishes once under duplicate submits, preserving revision, audit and public content", async () => {
+		const draft = articleForm();
+		const created = await Promise.all([
+			articles.createArticle(draft),
+			articles.createArticle(draft),
+		]);
+		expect(created[0]).toMatchObject({ ok: true });
+		expect(created[1]).toEqual(created[0]);
+		expect(await rows("SELECT id FROM website_articles")).toHaveLength(1);
+		expect(await rows("SELECT kind FROM cms_operations")).toEqual([
+			{ kind: "news.create" },
+		]);
+		expect(await rows("SELECT id FROM website_article_revisions")).toHaveLength(
+			0,
+		);
+		expect(boundaries.notify).not.toHaveBeenCalled();
+		const existing = await savedArticle();
+		expect(existing.status).toBe("draft");
+		expect(existing.publishedAt).toBeNull();
+		expect(await publicNews.getPublishedArticle(existing.slug)).toBeNull();
+		const negativeRevision = await appRedis?.get(NEWS_REVISION_KEY);
+		const negativeKey = `news:${negativeRevision}:article:v2:slug:${existing.slug}`;
+		expect(await appRedis?.get(negativeKey)).toBe("null");
+		expect(await appRedis?.ttl(negativeKey)).toBeGreaterThan(0);
+
+		const body = `<p>${"Contenuto completo è 📰 ".repeat(4000)}</p>`;
+		const publish = articleForm({
+			id: String(existing.id),
+			baseToken: articleEditToken(existing),
+			status: "published",
+			fullStory: body,
+		});
+		const published = await Promise.all([
+			articles.updateArticle(publish),
+			articles.updateArticle(publish),
+		]);
+		expect(published[0]).toMatchObject({ ok: true });
+		expect(published[1]).toEqual(published[0]);
+		const saved = await savedArticle();
+		expect(saved).toMatchObject({
+			status: "published",
+			fullStory: body,
+			publishAt: null,
+		});
+		expect(saved.publishedAt).toBeInstanceOf(Date);
+		const revisions = await rows(
+			"SELECT payload FROM website_article_revisions",
+		);
+		expect(revisions).toHaveLength(1);
+		expect(JSON.parse(revisions[0].payload)).toMatchObject({
+			status: "draft",
+			fullStory: existing.fullStory,
+		});
+		const history = await rows(
+			"SELECT target,`before`,`after` FROM admin_audit_log",
+		);
+		expect(history).toHaveLength(1);
+		expect(history[0].target).toBe("news");
+		expect(JSON.parse(history[0].before)).toMatchObject({
+			status: "draft",
+			fullStory: existing.fullStory,
+		});
+		expect(JSON.parse(history[0].after)).toMatchObject({
+			status: "published",
+			fullStory: body,
+		});
+		expect(await rows("SELECT kind FROM cms_operations ORDER BY kind")).toEqual(
+			[{ kind: "news.create" }, { kind: "news.update" }],
+		);
+		expect(await rows("SELECT topic,status FROM cms_outbox")).toEqual([
+			{ topic: "news.refresh", status: "pending" },
+			{ topic: "news.refresh", status: "pending" },
+		]);
+		expect(boundaries.notify).toHaveBeenCalledOnce();
+		await worker.drainOperationEffects();
+		expect(await rows("SELECT status FROM cms_outbox")).toEqual([
+			{ status: "done" },
+			{ status: "done" },
+		]);
+		expect(await appRedis?.get(NEWS_REVISION_KEY)).not.toBe(negativeRevision);
+		expect(await publicNews.getPublishedArticle(existing.slug)).toMatchObject({
+			id: existing.id,
+			fullStory: body,
+			slug: existing.slug,
+		});
+		const publicKey = `news:${await appRedis?.get(NEWS_REVISION_KEY)}:article:v2:slug:${existing.slug}`;
+		expect(JSON.parse(String(await appRedis?.get(publicKey)))).toMatchObject({
+			id: String(existing.id),
+			fullStory: body,
+		});
+		cache.invalidateMemory(publicKey);
+		const fromRedis = await publicNews.getPublishedArticle(existing.slug);
+		expect(fromRedis?.id).toBe(existing.id);
+		expect(fromRedis?.publishedAt).toEqual(saved.publishedAt);
+	});
+
+	it("rolls back publication, revision, audit and operation when queuing its effect fails, then retries the same submit", async () => {
+		expect(await articles.createArticle(articleForm())).toMatchObject({
+			ok: true,
+		});
+		const existing = await savedArticle();
+		const publish = articleForm({
+			id: String(existing.id),
+			baseToken: articleEditToken(existing),
+			status: "published",
+		});
+		const operationsBefore = await rows("SELECT * FROM cms_operations");
+		const effectsBefore = await rows("SELECT * FROM cms_outbox");
+		await connection?.query(
+			"CREATE TRIGGER reject_news_effect BEFORE INSERT ON cms_outbox FOR EACH ROW BEGIN IF NEW.topic='news.refresh' THEN SIGNAL SQLSTATE '45000' SET MESSAGE_TEXT='forced news effect failure'; END IF; END",
+		);
+		expect(await articles.updateArticle(publish)).toMatchObject({ ok: false });
+		expect(await savedArticle()).toEqual(existing);
+		expect(await rows("SELECT id FROM website_article_revisions")).toHaveLength(
+			0,
+		);
+		expect(await rows("SELECT id FROM admin_audit_log")).toHaveLength(0);
+		expect(await rows("SELECT * FROM cms_operations")).toEqual(
+			operationsBefore,
+		);
+		expect(await rows("SELECT * FROM cms_outbox")).toEqual(effectsBefore);
+		expect(boundaries.notify).not.toHaveBeenCalled();
+		await connection?.query("DROP TRIGGER reject_news_effect");
+		expect(await articles.updateArticle(publish)).toMatchObject({ ok: true });
+		expect(await articles.updateArticle(publish)).toMatchObject({ ok: true });
+		expect((await savedArticle()).status).toBe("published");
+		expect(await rows("SELECT id FROM website_article_revisions")).toHaveLength(
+			1,
+		);
+		expect(await rows("SELECT id FROM admin_audit_log")).toHaveLength(1);
+		expect(await rows("SELECT id FROM cms_operations")).toHaveLength(2);
+		expect(await rows("SELECT id FROM cms_outbox")).toHaveLength(2);
+		expect(boundaries.notify).toHaveBeenCalledOnce();
+	});
+
+	it("rolls back the entire scheduled batch if its second effect cannot be queued", async () => {
+		const now = new Date(Math.floor(Date.now() / 1000) * 1000);
+		await seedScheduledArticles(now);
+		const before = await rows("SELECT * FROM website_articles ORDER BY id");
+		await connection?.query(
+			"CREATE TRIGGER reject_second_scheduled_effect BEFORE INSERT ON cms_outbox FOR EACH ROW BEGIN IF NEW.topic='news.refresh' AND (SELECT COUNT(*) FROM cms_operations WHERE kind='news.schedule.publish')=2 THEN SIGNAL SQLSTATE '45000' SET MESSAGE_TEXT='forced second scheduled effect failure'; END IF; END",
+		);
+		await expect(scheduler.publishDueArticles(now)).rejects.toThrow();
+		expect(await rows("SELECT * FROM website_articles ORDER BY id")).toEqual(
+			before,
+		);
+		expect(await rows("SELECT id FROM cms_operations")).toHaveLength(0);
+		expect(await rows("SELECT id FROM cms_outbox")).toHaveLength(0);
+		await connection?.query("DROP TRIGGER reject_second_scheduled_effect");
+		expect(await scheduler.publishDueArticles(now)).toBe(2);
+		expect(await rows("SELECT id FROM cms_operations")).toHaveLength(2);
+		expect(await rows("SELECT topic,status FROM cms_outbox")).toEqual([
+			{ topic: "news.refresh", status: "pending" },
+			{ topic: "news.refresh", status: "pending" },
+		]);
+	});
+
+	it("serializes competing scheduler ticks without duplicate effects or early publication", async () => {
+		const now = new Date(Math.floor(Date.now() / 1000) * 1000);
+		await seedScheduledArticles(now);
+		const untouched = await rows(
+			"SELECT * FROM website_articles WHERE id IN (3,4,5) ORDER BY id",
+		);
+		expect(await publicNews.getPublishedArticle("due-earlier")).toBeNull();
+		expect(await publicNews.getPublishedArticle("due-now")).toBeNull();
+		const results = await Promise.all([
+			scheduler.publishDueArticles(now),
+			scheduler.publishDueArticles(now),
+		]);
+		expect(results.sort()).toEqual([0, 2]);
+		expect(await scheduler.publishDueArticles(now)).toBe(0);
+		expect(
+			await rows(
+				"SELECT * FROM website_articles WHERE id IN (3,4,5) ORDER BY id",
+			),
+		).toEqual(untouched);
+		const published = await rows(
+			"SELECT CAST(id AS CHAR) AS id,status,published_at FROM website_articles WHERE id IN (1,9007199254740993) ORDER BY id",
+		);
+		expect(published).toEqual([
+			{ id: "1", status: "published", published_at: now },
+			{ id: "9007199254740993", status: "published", published_at: now },
+		]);
+		const recorded = await rows(
+			"SELECT actor_id,kind,result_json FROM cms_operations ORDER BY actor_id",
+		);
+		expect(recorded).toHaveLength(2);
+		expect(
+			recorded.map((entry) => ({
+				actor: entry.actor_id,
+				kind: entry.kind,
+				result: JSON.parse(entry.result_json),
+			})),
+		).toEqual([
+			{
+				actor: 0,
+				kind: "news.schedule.publish",
+				result: { articleId: "9007199254740993", published: true },
+			},
+			{
+				actor: 7,
+				kind: "news.schedule.publish",
+				result: { articleId: "1", published: true },
+			},
+		]);
+		expect(await rows("SELECT topic,status FROM cms_outbox")).toEqual([
+			{ topic: "news.refresh", status: "pending" },
+			{ topic: "news.refresh", status: "pending" },
+		]);
+		await worker.drainOperationEffects();
+		expect(await publicNews.getPublishedArticle("due-earlier")).toMatchObject({
+			id: 1n,
+		});
+		expect(await publicNews.getPublishedArticle("due-now")).toMatchObject({
+			id: 9007199254740993n,
+		});
+		expect(await publicNews.getPublishedArticle("future")).toBeNull();
+		expect(await publicNews.getPublishedArticle("draft")).toBeNull();
+	});
+
+	it("keeps a committed publication readable during Redis disconnect and repairs cached absence through a retried delivery", async () => {
+		if (!appRedis) throw Error("Redis must be enabled in integration tests");
+		const redis = appRedis;
+		expect(await articles.createArticle(articleForm())).toMatchObject({
+			ok: true,
+		});
+		await worker.drainOperationEffects();
+		const existing = await savedArticle();
+		expect(await publicNews.getPublishedArticle(existing.slug)).toBeNull();
+		const negativeRevision = await redis.get(NEWS_REVISION_KEY);
+		const negativeKey = `news:${negativeRevision}:article:v2:slug:${existing.slug}`;
+		expect(await redis.get(negativeKey)).toBe("null");
+		const publish = articleForm({
+			id: String(existing.id),
+			baseToken: articleEditToken(existing),
+			status: "published",
+		});
+		const disconnected = once(redis, "end");
+		redis.disconnect();
+		await disconnected;
+		try {
+			expect(redis.status).toBe("end");
+			expect(await articles.updateArticle(publish)).toMatchObject({ ok: true });
+			expect((await savedArticle()).status).toBe("published");
+			expect(await publicNews.getPublishedArticle(existing.slug)).toMatchObject(
+				{ id: existing.id },
+			);
+			await worker.drainOperationEffects();
+			const [pending] = await rows(
+				"SELECT status,attempts,last_error FROM cms_outbox WHERE status<>'done'",
+			);
+			expect(pending).toMatchObject({ status: "pending", attempts: 1 });
+			expect(pending.last_error).toBeTruthy();
+			await redis.connect();
+			expect(await redis.ping()).toBe("PONG");
+			expect(await redis.get(NEWS_REVISION_KEY)).toBe(negativeRevision);
+			expect(await publicNews.getPublishedArticle(existing.slug)).toBeNull();
+			// Make the real queued retry due without a wall-clock sleep.
+			await connection?.query(
+				"UPDATE cms_outbox SET available_at=UTC_TIMESTAMP(3) WHERE status='pending'",
+			);
+			await worker.drainOperationEffects();
+			expect(
+				await rows(
+					"SELECT status,attempts,last_error FROM cms_outbox ORDER BY attempts",
+				),
+			).toEqual([
+				{ status: "done", attempts: 1, last_error: null },
+				{ status: "done", attempts: 2, last_error: null },
+			]);
+			expect(await redis.get(NEWS_REVISION_KEY)).not.toBe(negativeRevision);
+			expect(await publicNews.getPublishedArticle(existing.slug)).toMatchObject(
+				{ id: existing.id, fullStory: existing.fullStory },
+			);
+			expect(await articles.updateArticle(publish)).toMatchObject({ ok: true });
+			expect(await rows("SELECT id FROM website_articles")).toHaveLength(1);
+			expect(
+				await rows("SELECT id FROM website_article_revisions"),
+			).toHaveLength(1);
+			expect(await rows("SELECT id FROM cms_operations")).toHaveLength(2);
+			expect(await rows("SELECT id FROM cms_outbox")).toHaveLength(2);
+			expect(boundaries.notify).toHaveBeenCalledOnce();
+		} finally {
+			if (redis.status === "end") await redis.connect();
+		}
+	});
+});