浅念
469 lines
19 KiB
TypeScript
469 lines
19 KiB
TypeScript
// services/worker/test/watermark.e2e-spec.ts —— 水印历史重生成 e2e（P4a Task 3，
|
||
// 先写后实现）。真实 MySQL（直种 album/photos/watermark_configs/watermark_jobs，
|
||
// 形状对齐 api WatermarkService.create 的落库结果）+ 真实 MinIO（原图/变体对象）
|
||
// + 真实 Redis（BullMQ）。分层驱动同 publish.e2e-spec：核心断言直驱
|
||
// WatermarkRegenService，一组真实入队用例覆盖 Worker 包装。覆盖：
|
||
// ① 重生成（enabled）：逐张产水印变体（对象 + 行 version=configVersion）、
|
||
// progress→total、job done、**有效版本推进**（watermarkEffectiveVersion/
|
||
// watermarkEnabled + entityVersion+1）+ outbox album.updated（载荷水印位）；
|
||
// ② A20：原图对象键/字节零触碰；browse 变体行不动；
|
||
// ③ 失败注入：单张原图被删 → failed=1、job 仍 done（部分成功）、失败张无水印行
|
||
// （cn 保留旧图=无行回落 browse）、旧图对象拷贝到目标版本键（ov 模式键
|
||
// 存在性保障——A20 失败照片保留旧图）；
|
||
// ④ 关闭版重生成：水印行逐张删除（对象保留）、有效位 enabled=false、outbox
|
||
// 载荷 enabled=false + 新版本；
|
||
// ⑤ 任务级中止：配置行缺失 → state=failed + WATERMARK_CONFIG_MISSING、照片
|
||
// 零处理、有效位不动；
|
||
// ⑥ 真实 BullMQ Worker：api 生产者同参入队 → 消费 → done；幽灵投递 skipped。
|
||
import { randomBytes } from 'node:crypto';
|
||
import { Readable } from 'node:stream';
|
||
import { DeleteObjectCommand, S3Client } from '@aws-sdk/client-s3';
|
||
import { Queue, type Worker } from 'bullmq';
|
||
import IORedis from 'ioredis';
|
||
import type { Logger } from 'pino';
|
||
import sharp from 'sharp';
|
||
import { PrismaClient } from '@photo/db-cn';
|
||
import type { Prisma } from '@photo/db-cn';
|
||
import { QUEUE_WATERMARK, normalizeWatermarkConfig } from '@photo/contracts';
|
||
import {
|
||
createStorageFromEnv,
|
||
watermarkVariantObjectKey,
|
||
type MinioStorageAdapter,
|
||
type StoragePort,
|
||
} from '@photo/storage';
|
||
import {
|
||
WATERMARK_REGEN_JOB_NAME,
|
||
WATERMARK_JOB_ATTEMPTS,
|
||
WATERMARK_JOB_BACKOFF,
|
||
createWatermarkRegenWorker,
|
||
type WatermarkRegenJobData,
|
||
type WatermarkRegenWorkerOptions,
|
||
} from '../src/watermark.processor';
|
||
import {
|
||
WatermarkRegenService,
|
||
WATERMARK_ERROR_CONFIG_MISSING,
|
||
type WatermarkJobOutcome,
|
||
} from '../src/watermark.service';
|
||
import { resetDb } from './reset';
|
||
|
||
jest.setTimeout(120_000);
|
||
|
||
/** 轮询等待：e2e 对 BullMQ 异步路径的断言前置。 */
|
||
async function waitFor(
|
||
check: () => Promise<boolean>,
|
||
timeoutMs = 30_000,
|
||
stepMs = 100,
|
||
): Promise<void> {
|
||
const deadline = Date.now() + timeoutMs;
|
||
for (;;) {
|
||
if (await check()) return;
|
||
if (Date.now() > deadline) {
|
||
throw new Error(`waitFor 超时（${timeoutMs}ms）`);
|
||
}
|
||
await new Promise((resolve) => setTimeout(resolve, stepMs));
|
||
}
|
||
}
|
||
|
||
function cuidLike(alias: string): string {
|
||
return `c${Date.now().toString(36)}${alias}${randomBytes(6).toString('hex')}`;
|
||
}
|
||
|
||
/** 测试内生成真实 JPEG（带纹理，水印差异可观测）。 */
|
||
async function makeJpeg(width: number, height: number): Promise<Buffer> {
|
||
const svg = Buffer.from(
|
||
`<svg xmlns="http://www.w3.org/2000/svg" width="${width}" height="${height}">` +
|
||
Array.from(
|
||
{ length: 12 },
|
||
(_, i) =>
|
||
`<rect x="${(i * 83) % width}" y="${(i * 47) % height}" width="${50 + i * 9}" height="${35 + i * 7}" fill="#${(0x50 + i * 7).toString(16).padStart(2, '0')}6030" fill-opacity="0.85"/>`,
|
||
).join('') +
|
||
'</svg>',
|
||
);
|
||
return sharp(svg).jpeg({ quality: 85 }).toBuffer();
|
||
}
|
||
|
||
const KEY_ROOT = 'photo-live/dev';
|
||
|
||
/** 流式读取对象为单缓冲（与 worker readOriginal 同约定）。 */
|
||
async function readObject(storage: StoragePort, key: string): Promise<Buffer> {
|
||
const stream = (await storage.getStream(key)) as Readable;
|
||
const chunks: Buffer[] = [];
|
||
for await (const chunk of stream) {
|
||
chunks.push(Buffer.isBuffer(chunk) ? chunk : Buffer.from(chunk as Uint8Array));
|
||
}
|
||
return Buffer.concat(chunks);
|
||
}
|
||
|
||
/** S3 协议直删单对象（StoragePort 无删除——YAGNI，仅测试隔离/注入用）。 */
|
||
async function deleteObject(key: string): Promise<void> {
|
||
const client = new S3Client({
|
||
endpoint: process.env.S3_ENDPOINT,
|
||
region: process.env.S3_REGION,
|
||
credentials: {
|
||
accessKeyId: process.env.S3_ACCESS_KEY as string,
|
||
secretAccessKey: process.env.S3_SECRET_KEY as string,
|
||
},
|
||
forcePathStyle: true,
|
||
});
|
||
try {
|
||
await client.send(new DeleteObjectCommand({ Bucket: process.env.S3_BUCKET, Key: key }));
|
||
} finally {
|
||
await client.destroy();
|
||
}
|
||
}
|
||
|
||
describe('watermark regen (e2e)', () => {
|
||
let prisma: PrismaClient;
|
||
let storage: MinioStorageAdapter;
|
||
let service: WatermarkRegenService;
|
||
let queue: Queue;
|
||
let queueConnection: IORedis;
|
||
|
||
/** 直种 album（水印位可调，形状对齐 api create 提交后的状态）。 */
|
||
const seedAlbum = async (
|
||
alias: string,
|
||
watermark: { version: number; effective: number; enabled: boolean },
|
||
): Promise<string> => {
|
||
const suffix = randomBytes(4).toString('hex');
|
||
return (
|
||
await prisma.albums.create({
|
||
data: {
|
||
publicToken: `pt-${alias}-${suffix}`,
|
||
ownerId: `u-${alias}`,
|
||
startAt: new Date(),
|
||
defaultLocale: 'zh-CN',
|
||
watermarkVersion: watermark.version,
|
||
watermarkEffectiveVersion: watermark.effective,
|
||
watermarkEnabled: watermark.enabled,
|
||
},
|
||
select: { id: true },
|
||
})
|
||
).id;
|
||
};
|
||
|
||
/** 直种已发布照片（原图对象 + browse/thumb 变体行与对象——发布后形态）。 */
|
||
const seedPublishedPhoto = async (
|
||
albumId: string,
|
||
alias: string,
|
||
options: { withOriginalObject?: boolean } = {},
|
||
): Promise<{
|
||
photoId: string;
|
||
objectKey: string;
|
||
browseKey: string;
|
||
original: Buffer;
|
||
browse: Buffer;
|
||
}> => {
|
||
const photoId = cuidLike(alias);
|
||
const objectKey = `${KEY_ROOT}/albums/${albumId}/originals/${photoId}.jpg`;
|
||
const browseKey = `${KEY_ROOT}/albums/${albumId}/variants/${photoId}/1/browse.jpg`;
|
||
const thumbKey = `${KEY_ROOT}/albums/${albumId}/variants/${photoId}/1/thumb.jpg`;
|
||
const original = await makeJpeg(900, 600);
|
||
if (options.withOriginalObject !== false) {
|
||
await storage.put(objectKey, original, 'image/jpeg');
|
||
}
|
||
const browse = await sharp(original)
|
||
.resize({ width: 1920, withoutEnlargement: true })
|
||
.jpeg({ quality: 82 })
|
||
.toBuffer();
|
||
const thumb = await sharp(original)
|
||
.resize({ width: 480, withoutEnlargement: true })
|
||
.jpeg({ quality: 75 })
|
||
.toBuffer();
|
||
await storage.put(browseKey, browse, 'image/jpeg');
|
||
await storage.put(thumbKey, thumb, 'image/jpeg');
|
||
await prisma.photos.create({
|
||
data: {
|
||
id: photoId,
|
||
albumId,
|
||
uploaderId: `u-${alias}`,
|
||
objectKey,
|
||
bytes: original.length,
|
||
processingState: 'ready',
|
||
visibility: 'published',
|
||
publishedAt: new Date(),
|
||
width: 900,
|
||
height: 600,
|
||
},
|
||
});
|
||
await prisma.photo_variants.createMany({
|
||
data: [
|
||
{ photoId, variant: 'thumb', objectKey: thumbKey, bytes: thumb.length, version: 1, width: 480, height: 320 },
|
||
{ photoId, variant: 'browse', objectKey: browseKey, bytes: browse.length, version: 1, width: 900, height: 600 },
|
||
],
|
||
});
|
||
return { photoId, objectKey, browseKey, original, browse };
|
||
};
|
||
|
||
/** 直种配置行 + 任务行（形状对齐 api WatermarkService.create 落库结果）。 */
|
||
const seedWatermarkJob = async (
|
||
albumId: string,
|
||
configVersion: number,
|
||
enabled: boolean,
|
||
total: number,
|
||
): Promise<string> => {
|
||
await prisma.watermark_configs.create({
|
||
data: {
|
||
albumId,
|
||
configVersion,
|
||
config: normalizeWatermarkConfig({
|
||
enabled,
|
||
...(enabled ? { text: { content: `WM v${configVersion}` } } : {}),
|
||
}) as unknown as Prisma.InputJsonValue,
|
||
},
|
||
});
|
||
return (
|
||
await prisma.watermark_jobs.create({
|
||
data: { albumId, configVersion, state: 'pending', total },
|
||
select: { id: true },
|
||
})
|
||
).id;
|
||
};
|
||
|
||
beforeAll(async () => {
|
||
await resetDb();
|
||
prisma = new PrismaClient();
|
||
storage = createStorageFromEnv();
|
||
service = new WatermarkRegenService(prisma, storage);
|
||
queueConnection = new IORedis(
|
||
process.env.REDIS_URL_CN ?? 'redis://localhost:6379',
|
||
{ maxRetriesPerRequest: null },
|
||
);
|
||
queueConnection.on('error', () => undefined);
|
||
queue = new Queue(QUEUE_WATERMARK, { connection: queueConnection });
|
||
});
|
||
|
||
afterAll(async () => {
|
||
await queue.obliterate({ force: true }).catch(() => undefined);
|
||
await queue.close().catch(() => undefined);
|
||
await queueConnection.quit().catch(() => undefined);
|
||
await prisma.$disconnect();
|
||
});
|
||
|
||
test('重生成（enabled）：变体/进度/有效位推进/outbox，A20 原图与 browse 零触碰', async () => {
|
||
const albumId = await seedAlbum('wm1', { version: 1, effective: 0, enabled: false });
|
||
const photos = [
|
||
await seedPublishedPhoto(albumId, 'wm1a'),
|
||
await seedPublishedPhoto(albumId, 'wm1b'),
|
||
await seedPublishedPhoto(albumId, 'wm1c'),
|
||
];
|
||
const jobId = await seedWatermarkJob(albumId, 1, true, photos.length);
|
||
|
||
const outcome = await service.processJob(jobId);
|
||
expect(outcome).toBe('done');
|
||
|
||
const job = await prisma.watermark_jobs.findUniqueOrThrow({ where: { id: jobId } });
|
||
expect(job.state).toBe('done');
|
||
expect(job.progress).toBe(3);
|
||
expect(job.total).toBe(3);
|
||
expect(job.failed).toBe(0);
|
||
expect(job.errorCode).toBeNull();
|
||
|
||
// 逐张：水印行（version=1）+ 对象存在且为非空 jpeg；与 browse 字节不同。
|
||
for (const { photoId, objectKey, original } of photos) {
|
||
const row = await prisma.photo_variants.findUnique({
|
||
where: { photoId_variant: { photoId, variant: 'watermark' } },
|
||
});
|
||
expect(row?.version).toBe(1);
|
||
expect(row?.objectKey).toBe(watermarkVariantObjectKey(albumId, photoId, 1));
|
||
const wmObject = await readObject(storage, row!.objectKey);
|
||
expect(wmObject.length).toBeGreaterThan(0);
|
||
expect((await sharp(wmObject).metadata()).format).toBe('jpeg');
|
||
const browseRow = await prisma.photo_variants.findUniqueOrThrow({
|
||
where: { photoId_variant: { photoId, variant: 'browse' } },
|
||
});
|
||
expect(browseRow.version).toBe(1); // browse 变体行零触碰
|
||
expect(wmObject.equals(await readObject(storage, browseRow.objectKey))).toBe(false);
|
||
// A20：originals 键与字节零触碰。
|
||
expect(await readObject(storage, objectKey)).toEqual(original);
|
||
}
|
||
|
||
// 有效位推进 + 实体版本 +1。
|
||
const album = await prisma.albums.findUniqueOrThrow({ where: { id: albumId } });
|
||
expect(album.watermarkEffectiveVersion).toBe(1);
|
||
expect(album.watermarkEnabled).toBe(true);
|
||
expect(album.watermarkVersion).toBe(1);
|
||
expect(album.entityVersion).toBe(2); // 种子 1 → 推进后 2
|
||
|
||
// outbox album.updated：载荷水印位 = 推进后的有效状态。
|
||
const event = await prisma.outbox_events.findFirstOrThrow({
|
||
where: { albumId, type: 'album.updated' },
|
||
});
|
||
expect(event.entityVersion).toBe(2);
|
||
expect(event.priority).toBe('normal');
|
||
expect(event.payload).toMatchObject({
|
||
watermarkEnabled: true,
|
||
watermarkConfigVersion: 1,
|
||
});
|
||
});
|
||
|
||
test('失败注入：单张原图缺失 → failed=1 仍 done，失败张无行 + 旧图拷贝到目标键', async () => {
|
||
const albumId = await seedAlbum('wm2', { version: 1, effective: 0, enabled: false });
|
||
const ok = await seedPublishedPhoto(albumId, 'wm2a');
|
||
const broken = await seedPublishedPhoto(albumId, 'wm2b');
|
||
await deleteObject(broken.objectKey); // 模拟历史原图对象丢失（DB 行保留）
|
||
expect(await storage.head(broken.objectKey)).toBeNull();
|
||
|
||
const jobId = await seedWatermarkJob(albumId, 1, true, 2);
|
||
const outcome = await service.processJob(jobId);
|
||
expect(outcome).toBe('done');
|
||
|
||
const job = await prisma.watermark_jobs.findUniqueOrThrow({ where: { id: jobId } });
|
||
expect(job.state).toBe('done');
|
||
expect(job.failed).toBe(1);
|
||
expect(job.progress).toBe(2);
|
||
|
||
// 成功张：水印行/对象齐备。
|
||
const okRow = await prisma.photo_variants.findUnique({
|
||
where: { photoId_variant: { photoId: ok.photoId, variant: 'watermark' } },
|
||
});
|
||
expect(okRow?.version).toBe(1);
|
||
|
||
// 失败张：cn 无水印行（保留旧图=该张此前无水印，回落 browse）；ov 模式键
|
||
// 保障——旧图（browse 对象）已拷贝到目标版本键，字节一致。
|
||
const brokenRow = await prisma.photo_variants.findUnique({
|
||
where: { photoId_variant: { photoId: broken.photoId, variant: 'watermark' } },
|
||
});
|
||
expect(brokenRow).toBeNull();
|
||
const fallbackKey = watermarkVariantObjectKey(albumId, broken.photoId, 1);
|
||
const fallbackObject = await readObject(storage, fallbackKey);
|
||
expect(fallbackObject.equals(broken.browse)).toBe(true);
|
||
|
||
// 部分成功仍推进有效位（done 即切换；失败张 ov 拿旧图、cn 回落 browse）。
|
||
const album = await prisma.albums.findUniqueOrThrow({ where: { id: albumId } });
|
||
expect(album.watermarkEffectiveVersion).toBe(1);
|
||
expect(album.watermarkEnabled).toBe(true);
|
||
});
|
||
|
||
test('关闭版重生成：水印行删除（对象保留）、有效位 enabled=false、outbox 同步', async () => {
|
||
const albumId = await seedAlbum('wm3', { version: 2, effective: 0, enabled: false });
|
||
const photo = await seedPublishedPhoto(albumId, 'wm3a');
|
||
// 前置：v1 开启版先行生效（行在 v1 键、有效位推进）。
|
||
expect(await service.processJob(await seedWatermarkJob(albumId, 1, true, 1))).toBe('done');
|
||
const v1Key = watermarkVariantObjectKey(albumId, photo.photoId, 1);
|
||
const v1Row = await prisma.photo_variants.findUniqueOrThrow({
|
||
where: { photoId_variant: { photoId: photo.photoId, variant: 'watermark' } },
|
||
});
|
||
expect(v1Row.objectKey).toBe(v1Key);
|
||
|
||
// 关闭版（v2 enabled=false）重生成：删行，对象保留。
|
||
const jobId = await seedWatermarkJob(albumId, 2, false, 1);
|
||
expect(await service.processJob(jobId)).toBe('done');
|
||
|
||
expect(
|
||
await prisma.photo_variants.findUnique({
|
||
where: { photoId_variant: { photoId: photo.photoId, variant: 'watermark' } },
|
||
}),
|
||
).toBeNull();
|
||
expect(await storage.head(v1Key)).not.toBeNull(); // 对象保留（历史可溯）
|
||
|
||
const album = await prisma.albums.findUniqueOrThrow({ where: { id: albumId } });
|
||
expect(album.watermarkEffectiveVersion).toBe(2);
|
||
expect(album.watermarkEnabled).toBe(false);
|
||
expect(album.watermarkVersion).toBe(2);
|
||
|
||
const events = await prisma.outbox_events.findMany({
|
||
where: { albumId, type: 'album.updated' },
|
||
orderBy: { entityVersion: 'asc' },
|
||
});
|
||
expect(events.at(-1)?.payload).toMatchObject({
|
||
watermarkEnabled: false,
|
||
watermarkConfigVersion: 2,
|
||
});
|
||
});
|
||
|
||
test('任务级中止：配置行缺失 → failed + WATERMARK_CONFIG_MISSING，零处理零推进', async () => {
|
||
const albumId = await seedAlbum('wm4', { version: 1, effective: 0, enabled: false });
|
||
const photo = await seedPublishedPhoto(albumId, 'wm4a');
|
||
// 不种 watermark_configs —— 任务指向的版本无配置。
|
||
const jobId = (
|
||
await prisma.watermark_jobs.create({
|
||
data: { albumId, configVersion: 1, state: 'pending', total: 1 },
|
||
select: { id: true },
|
||
})
|
||
).id;
|
||
|
||
expect(await service.processJob(jobId)).toBe('failed');
|
||
const job = await prisma.watermark_jobs.findUniqueOrThrow({ where: { id: jobId } });
|
||
expect(job.state).toBe('failed');
|
||
expect(job.errorCode).toBe(WATERMARK_ERROR_CONFIG_MISSING);
|
||
expect(job.progress).toBe(0);
|
||
expect(
|
||
await prisma.photo_variants.findUnique({
|
||
where: { photoId_variant: { photoId: photo.photoId, variant: 'watermark' } },
|
||
}),
|
||
).toBeNull();
|
||
const album = await prisma.albums.findUniqueOrThrow({ where: { id: albumId } });
|
||
expect(album.watermarkEffectiveVersion).toBe(0);
|
||
expect(album.watermarkEnabled).toBe(false);
|
||
expect(album.entityVersion).toBe(1); // 未推进
|
||
});
|
||
|
||
describe('真实 BullMQ Worker', () => {
|
||
let worker: Worker<WatermarkRegenJobData, WatermarkJobOutcome, string>;
|
||
let workerConnection: IORedis;
|
||
|
||
beforeAll(async () => {
|
||
workerConnection = new IORedis(
|
||
process.env.REDIS_URL_CN ?? 'redis://localhost:6379',
|
||
{ maxRetriesPerRequest: null },
|
||
);
|
||
workerConnection.on('error', () => undefined);
|
||
const options: WatermarkRegenWorkerOptions = {
|
||
prisma,
|
||
storage,
|
||
connection: workerConnection,
|
||
logger: console as unknown as Logger,
|
||
};
|
||
worker = createWatermarkRegenWorker(options);
|
||
});
|
||
|
||
afterAll(async () => {
|
||
await worker.close();
|
||
await workerConnection.quit().catch(() => undefined);
|
||
});
|
||
|
||
/** 与 api 生产者同参入队（jobId=行 id 去重 + attempts:3 指数退避 1s 起）。 */
|
||
const enqueue = (jobId: string) =>
|
||
queue.add(WATERMARK_REGEN_JOB_NAME, { jobId }, {
|
||
jobId,
|
||
attempts: WATERMARK_JOB_ATTEMPTS,
|
||
backoff: WATERMARK_JOB_BACKOFF,
|
||
});
|
||
|
||
test('真实入队 → 消费 → done（有效位推进）', async () => {
|
||
const albumId = await seedAlbum('wm5', { version: 1, effective: 0, enabled: false });
|
||
const photo = await seedPublishedPhoto(albumId, 'wm5a');
|
||
const jobId = await seedWatermarkJob(albumId, 1, true, 1);
|
||
await enqueue(jobId);
|
||
|
||
await waitFor(async () =>
|
||
(await prisma.watermark_jobs.findUnique({ where: { id: jobId } }))?.state === 'done',
|
||
);
|
||
const job = await prisma.watermark_jobs.findUniqueOrThrow({ where: { id: jobId } });
|
||
expect(job.progress).toBe(1);
|
||
expect(job.failed).toBe(0);
|
||
const row = await prisma.photo_variants.findUnique({
|
||
where: { photoId_variant: { photoId: photo.photoId, variant: 'watermark' } },
|
||
});
|
||
expect(row?.version).toBe(1);
|
||
expect(
|
||
await storage.head(watermarkVariantObjectKey(albumId, photo.photoId, 1)),
|
||
).not.toBeNull();
|
||
const album = await prisma.albums.findUniqueOrThrow({ where: { id: albumId } });
|
||
expect(album.watermarkEnabled).toBe(true);
|
||
expect(album.watermarkEffectiveVersion).toBe(1);
|
||
});
|
||
|
||
test('行不存在的幽灵投递 → skipped，不产生任何行/对象痕迹', async () => {
|
||
const rowsBefore = await prisma.watermark_jobs.count();
|
||
const ghost = cuidLike('ghost');
|
||
await enqueue(ghost);
|
||
await waitFor(async () => {
|
||
const job = await queue.getJob(ghost);
|
||
return job?.returnvalue === 'skipped';
|
||
});
|
||
expect(await prisma.watermark_jobs.count()).toBe(rowsBefore);
|
||
});
|
||
});
|
||
});
|