315 lines
13 KiB
TypeScript
315 lines
13 KiB
TypeScript
|
// services/worker/src/watermark.service.ts —— 水印历史重生成服务（P4a Task 3，
|
|||
|
|
// 「显式任务」语义的实现体：水印变更默认仅影响新照片，历史照片经本任务重derive）。
|
|||
|
|
// 逐张处理（分批 50）：原图 → composeWatermarkBrowse（浏览图基线 + 文字层）→
|
|||
|
|
// 写版本化水印键 `variants/{photoId}/{configVersion}/watermark.jpg` → upsert
|
|||
|
|
// photo_variants(variant='watermark', version=configVersion)。原子性口径（A20）：
|
|||
|
|
// · 对象先写、行后 upsert——cn 公开侧的选图规则是「有效开关开启且行存在 ⇒
|
|||
|
|
// 用行的键」（行只会指向已写成功的对象），处理中的照片始终有 browse 变体
|
|||
|
|
// 兜底，绝无破图；
|
|||
|
|
// · 原图（originals）与 browse 变体零触碰（e2e A20 断言）；
|
|||
|
|
// · 照片级失败（原图缺失/解码失败）计入 failed 并继续——部分成功允许，
|
|||
|
|
// job 仍 done（failed>0 供前端展示；重试 = 重新 POST 新任务，按照片粒度
|
|||
|
|
// 幂等：同 configVersion+photo 重复 upsert 天然收敛）。失败张**旧图保留**：
|
|||
|
|
// 行不动（cn 继续服务旧键）+ 旧图对象拷贝到目标版本键（ov 模式键在有效
|
|||
|
|
// 版本推进后仍有对象可服务——失败照片保留旧图，A20；见 fallbackKeepOldImage）；
|
|||
|
|
// · 任务级中止（配置缺失/不可解析）→ state failed + errorCode，照片零处理
|
|||
|
|
// （有效版本不推进，ov/cn 继续旧配置，杜绝半切换状态）。
|
|||
|
|
// 关闭版（enabled=false）重生成 = 逐张删除 watermark 变体行（对象保留供历史
|
|||
|
|
// 版本回溯，成本可忽略）；完成后有效位 enabled=false，两侧公开服务回落 browse。
|
|||
|
|
// 完成事务：job 置 done + 推进 albums「有效版本位」（watermarkEffectiveVersion
|
|||
|
|
// /watermarkEnabled，仅更高版本推进——乱序完成的旧任务自然让位）+ entityVersion+1
|
|||
|
|
// + outbox album.updated（载荷经 contracts 共享构建器，ov 副本据此切换模式键）。
|
|||
|
|
import { Readable } from 'node:stream';
|
|||
|
|
import type { Prisma, PrismaClient } from '@photo/db-cn';
|
|||
|
|
import {
|
|||
|
|
buildAlbumUpdatedPayloadCore,
|
|||
|
|
parseStoredWatermarkConfig,
|
|||
|
|
type StoredWatermarkConfig,
|
|||
|
|
} from '@photo/contracts';
|
|||
|
|
import { watermarkVariantObjectKey, type StoragePort } from '@photo/storage';
|
|||
|
|
import type { Logger } from 'pino';
|
|||
|
|
import { composeWatermarkBrowse } from './watermark';
|
|||
|
|
|
|||
|
|
/** 照片分批大小（逐批查询，游标推进；发布中途的新照片自然纳入后续批次）。 */
|
|||
|
|
const PHOTO_BATCH_SIZE = 50;
|
|||
|
|
|
|||
|
|
/** 任务级错误码（照片级失败不使用——只计数）。 */
|
|||
|
|
export const WATERMARK_ERROR_CONFIG_MISSING = 'WATERMARK_CONFIG_MISSING';
|
|||
|
|
export const WATERMARK_ERROR_CONFIG_INVALID = 'WATERMARK_CONFIG_INVALID';
|
|||
|
|
export const WATERMARK_ERROR_ALBUM_MISSING = 'WATERMARK_ALBUM_MISSING';
|
|||
|
|
|
|||
|
|
export type WatermarkJobOutcome =
|
|||
|
|
| 'done' // 全部处理完成（允许 failed>0 部分成功）
|
|||
|
|
| 'failed' // 任务级中止（errorCode 已落行）
|
|||
|
|
| 'skipped'; // 行不存在 / 已 done 的重复投递
|
|||
|
|
|
|||
|
|
export class WatermarkRegenService {
|
|||
|
|
constructor(
|
|||
|
|
private readonly prisma: PrismaClient,
|
|||
|
|
private readonly storage: StoragePort,
|
|||
|
|
private readonly logger?: Logger,
|
|||
|
|
) {}
|
|||
|
|
|
|||
|
|
/** 处理一个重生成任务（BullMQ 投递与 e2e 直驱共用入口）。 */
|
|||
|
|
async processJob(jobRowId: string): Promise<WatermarkJobOutcome> {
|
|||
|
|
const job = await this.prisma.watermark_jobs.findUnique({
|
|||
|
|
where: { id: jobRowId },
|
|||
|
|
});
|
|||
|
|
if (job === null || job.state === 'done') {
|
|||
|
|
return 'skipped';
|
|||
|
|
}
|
|||
|
|
// 认领（含 BullMQ 整任务重试对 failed 行的复活）+ 计数清零重跑（按照片
|
|||
|
|
// 幂等，重复处理无副作用）。
|
|||
|
|
const claimed = await this.prisma.watermark_jobs.updateMany({
|
|||
|
|
where: { id: jobRowId, state: { in: ['pending', 'running', 'failed'] } },
|
|||
|
|
data: { state: 'running', progress: 0, failed: 0, errorCode: null },
|
|||
|
|
});
|
|||
|
|
if (claimed.count === 0) {
|
|||
|
|
return 'skipped';
|
|||
|
|
}
|
|||
|
|
|
|||
|
|
const configRow = await this.prisma.watermark_configs.findUnique({
|
|||
|
|
where: {
|
|||
|
|
albumId_configVersion: {
|
|||
|
|
albumId: job.albumId,
|
|||
|
|
configVersion: job.configVersion,
|
|||
|
|
},
|
|||
|
|
},
|
|||
|
|
});
|
|||
|
|
const config = parseStoredWatermarkConfig(configRow?.config);
|
|||
|
|
if (configRow === null) {
|
|||
|
|
return this.abortJob(jobRowId, WATERMARK_ERROR_CONFIG_MISSING);
|
|||
|
|
}
|
|||
|
|
if (config === null) {
|
|||
|
|
return this.abortJob(jobRowId, WATERMARK_ERROR_CONFIG_INVALID);
|
|||
|
|
}
|
|||
|
|
const albumExists = await this.prisma.albums.findUnique({
|
|||
|
|
where: { id: job.albumId },
|
|||
|
|
select: { id: true },
|
|||
|
|
});
|
|||
|
|
if (albumExists === null) {
|
|||
|
|
return this.abortJob(jobRowId, WATERMARK_ERROR_ALBUM_MISSING);
|
|||
|
|
}
|
|||
|
|
|
|||
|
|
let processed = 0;
|
|||
|
|
let failed = 0;
|
|||
|
|
let cursor: string | undefined;
|
|||
|
|
for (;;) {
|
|||
|
|
const photos = await this.prisma.photos.findMany({
|
|||
|
|
where: {
|
|||
|
|
albumId: job.albumId,
|
|||
|
|
visibility: 'published',
|
|||
|
|
...(cursor === undefined ? {} : { id: { gt: cursor } }),
|
|||
|
|
},
|
|||
|
|
orderBy: { id: 'asc' },
|
|||
|
|
take: PHOTO_BATCH_SIZE,
|
|||
|
|
select: { id: true, objectKey: true },
|
|||
|
|
});
|
|||
|
|
if (photos.length === 0) {
|
|||
|
|
break;
|
|||
|
|
}
|
|||
|
|
for (const photo of photos) {
|
|||
|
|
try {
|
|||
|
|
await this.processPhoto(photo.id, photo.objectKey, job.albumId, job.configVersion, config);
|
|||
|
|
} catch (err) {
|
|||
|
|
failed += 1;
|
|||
|
|
this.logger?.warn(
|
|||
|
|
{ jobId: jobRowId, photoId: photo.id, err: err instanceof Error ? err.message : String(err) },
|
|||
|
|
'水印重生成单张失败（计入 failed，继续）',
|
|||
|
|
);
|
|||
|
|
if (config.enabled) {
|
|||
|
|
// A20「失败照片保留旧图」的 ov 侧保障：旧图对象拷贝到目标版本键，
|
|||
|
|
// 使 done 后 ov 模式键（有效版本）仍有对象可服务。best-effort——
|
|||
|
|
// 拷贝再失败（存储整体故障）只影响 ov 该张，cn 旧行照常服务。
|
|||
|
|
try {
|
|||
|
|
await this.fallbackKeepOldImage(photo.id, job.albumId, job.configVersion);
|
|||
|
|
} catch (fallbackErr) {
|
|||
|
|
this.logger?.warn(
|
|||
|
|
{
|
|||
|
|
jobId: jobRowId,
|
|||
|
|
photoId: photo.id,
|
|||
|
|
err: fallbackErr instanceof Error ? fallbackErr.message : String(fallbackErr),
|
|||
|
|
},
|
|||
|
|
'水印重生成失败张的旧图拷贝亦失败（ov 该张模式键将缺对象）',
|
|||
|
|
);
|
|||
|
|
}
|
|||
|
|
}
|
|||
|
|
}
|
|||
|
|
processed += 1;
|
|||
|
|
cursor = photo.id;
|
|||
|
|
await this.prisma.watermark_jobs.update({
|
|||
|
|
where: { id: jobRowId },
|
|||
|
|
data: { progress: processed, failed },
|
|||
|
|
});
|
|||
|
|
}
|
|||
|
|
}
|
|||
|
|
|
|||
|
|
await this.completeJob(jobRowId, job.albumId, job.configVersion, config);
|
|||
|
|
this.logger?.info(
|
|||
|
|
{ jobId: jobRowId, albumId: job.albumId, configVersion: job.configVersion, total: processed, failed },
|
|||
|
|
'水印重生成任务完成',
|
|||
|
|
);
|
|||
|
|
return 'done';
|
|||
|
|
}
|
|||
|
|
|
|||
|
|
/**
|
|||
|
|
* 单张：enabled → 产水印变体（对象先写、行后 upsert）；disabled → 删行
|
|||
|
|
* （对象保留）。抛错由调用方计数继续。
|
|||
|
|
*/
|
|||
|
|
private async processPhoto(
|
|||
|
|
photoId: string,
|
|||
|
|
objectKey: string,
|
|||
|
|
albumId: string,
|
|||
|
|
configVersion: number,
|
|||
|
|
config: StoredWatermarkConfig,
|
|||
|
|
): Promise<void> {
|
|||
|
|
if (!config.enabled) {
|
|||
|
|
await this.prisma.photo_variants.deleteMany({
|
|||
|
|
where: { photoId, variant: 'watermark' },
|
|||
|
|
});
|
|||
|
|
return;
|
|||
|
|
}
|
|||
|
|
const input = await this.readOriginal(objectKey);
|
|||
|
|
const out = await composeWatermarkBrowse(input, config);
|
|||
|
|
const key = watermarkVariantObjectKey(albumId, photoId, configVersion);
|
|||
|
|
await this.storage.put(key, out.buffer, 'image/jpeg');
|
|||
|
|
await this.prisma.photo_variants.upsert({
|
|||
|
|
where: { photoId_variant: { photoId, variant: 'watermark' } },
|
|||
|
|
update: {
|
|||
|
|
objectKey: key,
|
|||
|
|
bytes: out.buffer.length,
|
|||
|
|
version: configVersion,
|
|||
|
|
width: out.width,
|
|||
|
|
height: out.height,
|
|||
|
|
},
|
|||
|
|
create: {
|
|||
|
|
photoId,
|
|||
|
|
variant: 'watermark',
|
|||
|
|
objectKey: key,
|
|||
|
|
bytes: out.buffer.length,
|
|||
|
|
version: configVersion,
|
|||
|
|
width: out.width,
|
|||
|
|
height: out.height,
|
|||
|
|
},
|
|||
|
|
});
|
|||
|
|
}
|
|||
|
|
|
|||
|
|
/**
|
|||
|
|
* 失败张旧图保留（A20，对象级兜底——行不动）：把照片当前实际展示的旧图
|
|||
|
|
* （旧水印行对象；无水印行则 browse 对象）拷贝到目标版本水印键。done 后
|
|||
|
|
* ov 模式键指向有效版本，此拷贝保证该键有对象（旧图字节）；cn 侧不受影响
|
|||
|
|
* （旧行未覆盖，继续服务旧键）。源=目标（同版本重跑）时跳过。
|
|||
|
|
*/
|
|||
|
|
private async fallbackKeepOldImage(
|
|||
|
|
photoId: string,
|
|||
|
|
albumId: string,
|
|||
|
|
configVersion: number,
|
|||
|
|
): Promise<void> {
|
|||
|
|
const variants = await this.prisma.photo_variants.findMany({
|
|||
|
|
where: { photoId, variant: { in: ['watermark', 'browse'] } },
|
|||
|
|
});
|
|||
|
|
const source =
|
|||
|
|
variants.find((v) => v.variant === 'watermark') ??
|
|||
|
|
variants.find((v) => v.variant === 'browse');
|
|||
|
|
if (source === undefined) {
|
|||
|
|
return; // 连 browse 行都缺（post-publish 不变量破坏）：无可拷贝，放弃。
|
|||
|
|
}
|
|||
|
|
const targetKey = watermarkVariantObjectKey(albumId, photoId, configVersion);
|
|||
|
|
if (source.objectKey === targetKey) {
|
|||
|
|
return;
|
|||
|
|
}
|
|||
|
|
const bytes = await this.readOriginal(source.objectKey);
|
|||
|
|
await this.storage.put(targetKey, bytes, 'image/jpeg');
|
|||
|
|
}
|
|||
|
|
|
|||
|
|
/** 任务级中止：照片零处理，行置 failed + errorCode。 */
|
|||
|
|
private async abortJob(
|
|||
|
|
jobRowId: string,
|
|||
|
|
errorCode: string,
|
|||
|
|
): Promise<WatermarkJobOutcome> {
|
|||
|
|
await this.prisma.watermark_jobs.update({
|
|||
|
|
where: { id: jobRowId },
|
|||
|
|
data: { state: 'failed', errorCode },
|
|||
|
|
});
|
|||
|
|
this.logger?.error({ jobId: jobRowId, errorCode }, '水印重生成任务中止');
|
|||
|
|
return 'failed';
|
|||
|
|
}
|
|||
|
|
|
|||
|
|
/**
|
|||
|
|
* 完成事务：job done + 有效版本推进 + album.updated outbox。乐观重试同
|
|||
|
|
* albums.service.update（并发相册更新败方整事务重跑；3 次仍败抛错交还
|
|||
|
|
* BullMQ 重试——照片处理幂等，完成事务随重投递收敛）。乱序完成的旧任务
|
|||
|
|
* （有效版本已更高）跳过推进、只置 done。
|
|||
|
|
*/
|
|||
|
|
private async completeJob(
|
|||
|
|
jobRowId: string,
|
|||
|
|
albumId: string,
|
|||
|
|
configVersion: number,
|
|||
|
|
config: StoredWatermarkConfig,
|
|||
|
|
): Promise<void> {
|
|||
|
|
let outcome: 'done' | 'retry' = 'retry';
|
|||
|
|
for (let attempt = 0; attempt < 3 && outcome === 'retry'; attempt++) {
|
|||
|
|
outcome = await this.prisma.$transaction(async (tx) => {
|
|||
|
|
await tx.watermark_jobs.updateMany({
|
|||
|
|
where: { id: jobRowId, state: 'running' },
|
|||
|
|
data: { state: 'done', errorCode: null },
|
|||
|
|
});
|
|||
|
|
const album = await tx.albums.findUnique({
|
|||
|
|
where: { id: albumId },
|
|||
|
|
select: { entityVersion: true, watermarkEffectiveVersion: true },
|
|||
|
|
});
|
|||
|
|
if (album === null || album.watermarkEffectiveVersion >= configVersion) {
|
|||
|
|
return 'done' as const;
|
|||
|
|
}
|
|||
|
|
const claimed = await tx.albums.updateMany({
|
|||
|
|
where: {
|
|||
|
|
id: albumId,
|
|||
|
|
entityVersion: album.entityVersion,
|
|||
|
|
watermarkEffectiveVersion: album.watermarkEffectiveVersion,
|
|||
|
|
},
|
|||
|
|
data: {
|
|||
|
|
watermarkEffectiveVersion: configVersion,
|
|||
|
|
watermarkEnabled: config.enabled,
|
|||
|
|
entityVersion: { increment: 1 },
|
|||
|
|
},
|
|||
|
|
});
|
|||
|
|
if (claimed.count === 0) {
|
|||
|
|
return 'retry' as const;
|
|||
|
|
}
|
|||
|
|
const updated = await tx.albums.findUniqueOrThrow({
|
|||
|
|
where: { id: albumId },
|
|||
|
|
});
|
|||
|
|
const translations = await tx.album_translations.findMany({
|
|||
|
|
where: { albumId },
|
|||
|
|
orderBy: { locale: 'asc' },
|
|||
|
|
});
|
|||
|
|
await tx.outbox_events.create({
|
|||
|
|
data: {
|
|||
|
|
type: 'album.updated',
|
|||
|
|
albumId,
|
|||
|
|
entityVersion: updated.entityVersion,
|
|||
|
|
payload: buildAlbumUpdatedPayloadCore(
|
|||
|
|
updated,
|
|||
|
|
translations,
|
|||
|
|
) as unknown as Prisma.InputJsonValue,
|
|||
|
|
},
|
|||
|
|
});
|
|||
|
|
return 'done' as const;
|
|||
|
|
});
|
|||
|
|
}
|
|||
|
|
if (outcome !== 'done') {
|
|||
|
|
throw new Error(`watermark: 完成事务乐观重试耗尽 jobId=${jobRowId}`);
|
|||
|
|
}
|
|||
|
|
}
|
|||
|
|
|
|||
|
|
/** 流式拉取原图并聚合为单缓冲（同 publish.service.readOriginal 约定）。 */
|
|||
|
|
private async readOriginal(objectKey: string): Promise<Buffer> {
|
|||
|
|
const stream = (await this.storage.getStream(objectKey)) 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);
|
|||
|
|
}
|
|||
|
|
}
|