feat(api-water): 支持上传水印图片
This commit is contained in:
R524809
@@ -0,0 +1,537 @@
|
||||
import { createHash, randomUUID } from 'node:crypto';
|
||||
import {
|
||||
CreateBucketCommand,
|
||||
GetObjectCommand,
|
||||
HeadBucketCommand,
|
||||
HeadObjectCommand,
|
||||
PutObjectCommand,
|
||||
S3Client,
|
||||
} from '@aws-sdk/client-s3';
|
||||
import { getSignedUrl } from '@aws-sdk/s3-request-presigner';
|
||||
import { Inject, Injectable } from '@nestjs/common';
|
||||
import type {
|
||||
CompleteMediaUploadIntentRequest,
|
||||
CreateMediaUploadIntentRequest,
|
||||
} from '@atlas/contracts';
|
||||
import type { Database, DatabaseTransaction } from '@atlas/database';
|
||||
import { ApiError } from '../common/api-error.js';
|
||||
import { IdempotencyService } from '../common/idempotency.service.js';
|
||||
import { APP_CONFIG, type AppConfig } from '../config/app-config.module.js';
|
||||
import { DATABASE } from '../database/database.module.js';
|
||||
|
||||
type IntentRow = {
|
||||
id: string;
|
||||
storage_provider: string;
|
||||
bucket: string;
|
||||
object_key: string;
|
||||
expected_content_type: string;
|
||||
expected_byte_size: string;
|
||||
expected_checksum_sha256: string;
|
||||
status: 'pending' | 'ready' | 'failed' | 'expired';
|
||||
expires_at: Date | string;
|
||||
asset_id: string | null;
|
||||
};
|
||||
type AssetRow = {
|
||||
id: string;
|
||||
content_type: string;
|
||||
byte_size: string;
|
||||
width_px: number | null;
|
||||
height_px: number | null;
|
||||
created_at: Date | string;
|
||||
};
|
||||
|
||||
@Injectable()
|
||||
export class MediaService {
|
||||
private readonly s3: S3Client;
|
||||
private bucketReady: Promise<void> | null = null;
|
||||
|
||||
public constructor(
|
||||
@Inject(IdempotencyService)
|
||||
private readonly idempotency: IdempotencyService,
|
||||
@Inject(DATABASE) private readonly database: Database,
|
||||
@Inject(APP_CONFIG) private readonly config: AppConfig,
|
||||
) {
|
||||
this.s3 = new S3Client({
|
||||
endpoint: config.OBJECT_STORAGE_ENDPOINT,
|
||||
region: 'us-east-1',
|
||||
forcePathStyle: config.objectStorageForcePathStyle,
|
||||
credentials: {
|
||||
accessKeyId: config.objectStorageAccessKey,
|
||||
secretAccessKey: config.objectStorageSecretKey,
|
||||
},
|
||||
});
|
||||
}
|
||||
|
||||
public async createIntent(
|
||||
workspaceId: string,
|
||||
userId: string,
|
||||
key: string,
|
||||
input: CreateMediaUploadIntentRequest,
|
||||
) {
|
||||
return this.idempotency.run(
|
||||
{
|
||||
workspaceId,
|
||||
userId,
|
||||
method: 'POST',
|
||||
path: '/api/media/upload-intents',
|
||||
key,
|
||||
request: input,
|
||||
},
|
||||
async (tx) => {
|
||||
await this.ensureBucket();
|
||||
const id = randomUUID();
|
||||
const objectKey = `uploads/${workspaceId}/watermark-images/${id}`;
|
||||
const expiresAt = new Date(Date.now() + 10 * 60_000);
|
||||
await tx.query(
|
||||
`INSERT INTO atlas.media_upload_intents
|
||||
(id, workspace_id, purpose, storage_provider, bucket, object_key, expected_content_type,
|
||||
expected_byte_size, expected_checksum_sha256, expires_at, created_by_user_id)
|
||||
VALUES ($1, current_setting('app.workspace_id')::uuid, 'watermark_image', 'local_minio',
|
||||
$2, $3, $4, $5, $6, $7, $8)`,
|
||||
[
|
||||
id,
|
||||
this.config.OBJECT_STORAGE_BUCKET,
|
||||
objectKey,
|
||||
input.content_type,
|
||||
input.byte_size,
|
||||
input.checksum_sha256,
|
||||
expiresAt,
|
||||
userId,
|
||||
],
|
||||
);
|
||||
const url = await getSignedUrl(
|
||||
this.s3,
|
||||
new PutObjectCommand({
|
||||
Bucket: this.config.OBJECT_STORAGE_BUCKET,
|
||||
Key: objectKey,
|
||||
ContentType: input.content_type,
|
||||
}),
|
||||
{ expiresIn: 600 },
|
||||
);
|
||||
return {
|
||||
status: 201,
|
||||
body: {
|
||||
data: {
|
||||
upload_id: id,
|
||||
expires_at: expiresAt.toISOString(),
|
||||
upload: {
|
||||
method: 'PUT',
|
||||
url,
|
||||
headers: { 'content-type': input.content_type },
|
||||
},
|
||||
},
|
||||
},
|
||||
};
|
||||
},
|
||||
);
|
||||
}
|
||||
|
||||
public async completeIntent(
|
||||
workspaceId: string,
|
||||
userId: string,
|
||||
intentId: string,
|
||||
key: string,
|
||||
input: CompleteMediaUploadIntentRequest,
|
||||
) {
|
||||
try {
|
||||
return await this.idempotency.run(
|
||||
{
|
||||
workspaceId,
|
||||
userId,
|
||||
method: 'POST',
|
||||
path: `/api/media/upload-intents/${intentId}/complete`,
|
||||
key,
|
||||
request: input,
|
||||
},
|
||||
async (tx) => {
|
||||
const intent = await this.lockIntent(tx, intentId);
|
||||
if (intent.status === 'ready' && intent.asset_id) {
|
||||
return {
|
||||
status: 201,
|
||||
body: {
|
||||
data: {
|
||||
asset: await this.requireAsset(
|
||||
tx,
|
||||
intent.asset_id,
|
||||
),
|
||||
},
|
||||
},
|
||||
};
|
||||
}
|
||||
if (intent.status === 'failed') {
|
||||
throw new ApiError(
|
||||
422,
|
||||
'UPLOAD_VALIDATION_FAILED',
|
||||
'该上传意图的文件未通过安全校验，请重新上传',
|
||||
);
|
||||
}
|
||||
if (intent.status === 'expired') {
|
||||
throw new ApiError(
|
||||
422,
|
||||
'UPLOAD_EXPIRED',
|
||||
'上传意图已过期，请重新发起上传',
|
||||
);
|
||||
}
|
||||
if (new Date(intent.expires_at).getTime() <= Date.now()) {
|
||||
throw new ApiError(
|
||||
422,
|
||||
'UPLOAD_EXPIRED',
|
||||
'上传意图已过期，请重新发起上传',
|
||||
);
|
||||
}
|
||||
if (
|
||||
input.checksum_sha256 !==
|
||||
intent.expected_checksum_sha256
|
||||
) {
|
||||
throw new ApiError(
|
||||
422,
|
||||
'UPLOAD_CHECKSUM_MISMATCH',
|
||||
'完成请求的文件摘要与上传意图不一致',
|
||||
);
|
||||
}
|
||||
const inspected = await this.inspectObject(intent);
|
||||
if (
|
||||
inspected.byteSize !==
|
||||
Number(intent.expected_byte_size) ||
|
||||
inspected.checksum !==
|
||||
intent.expected_checksum_sha256 ||
|
||||
inspected.contentType !== intent.expected_content_type
|
||||
) {
|
||||
throw new ApiError(
|
||||
422,
|
||||
'UPLOAD_VALIDATION_FAILED',
|
||||
'上传文件未通过安全校验',
|
||||
);
|
||||
}
|
||||
const existing = await tx.query<{ id: string }>(
|
||||
`SELECT id FROM atlas.media_assets WHERE checksum_sha256 = $1`,
|
||||
[inspected.checksum],
|
||||
);
|
||||
const assetId = existing.rows[0]?.id ?? randomUUID();
|
||||
if (!existing.rows[0]) {
|
||||
await tx.query(
|
||||
`INSERT INTO atlas.media_assets
|
||||
(id, workspace_id, storage_provider, bucket, object_key, content_type, byte_size,
|
||||
checksum_sha256, width_px, height_px)
|
||||
VALUES ($1, current_setting('app.workspace_id')::uuid, 'local_minio', $2, $3, $4, $5, $6, $7, $8)`,
|
||||
[
|
||||
assetId,
|
||||
intent.bucket,
|
||||
intent.object_key,
|
||||
inspected.contentType,
|
||||
inspected.byteSize,
|
||||
inspected.checksum,
|
||||
inspected.width,
|
||||
inspected.height,
|
||||
],
|
||||
);
|
||||
}
|
||||
await tx.query(
|
||||
`UPDATE atlas.media_upload_intents SET status = 'ready', asset_id = $2,
|
||||
completed_at = now(), error_code = NULL, error_message = NULL WHERE id = $1`,
|
||||
[intentId, assetId],
|
||||
);
|
||||
return {
|
||||
status: 201,
|
||||
body: {
|
||||
data: {
|
||||
asset: await this.requireAsset(tx, assetId),
|
||||
},
|
||||
},
|
||||
};
|
||||
},
|
||||
);
|
||||
} catch (error) {
|
||||
const code = apiErrorCode(error);
|
||||
if (
|
||||
code === 'UPLOAD_EXPIRED' ||
|
||||
code === 'UPLOAD_OBJECT_MISSING' ||
|
||||
code === 'UPLOAD_VALIDATION_FAILED'
|
||||
) {
|
||||
await this.persistTerminalIntentStatus(
|
||||
workspaceId,
|
||||
intentId,
|
||||
code,
|
||||
);
|
||||
}
|
||||
throw error;
|
||||
}
|
||||
}
|
||||
|
||||
public async accessUrl(workspaceId: string, assetId: string) {
|
||||
return this.database.workspaceTransaction(workspaceId, async (tx) => {
|
||||
const asset = await tx.query<{
|
||||
bucket: string;
|
||||
object_key: string;
|
||||
}>(
|
||||
`SELECT bucket, object_key FROM atlas.media_assets WHERE id = $1`,
|
||||
[assetId],
|
||||
);
|
||||
if (!asset.rows[0])
|
||||
throw new ApiError(404, 'RESOURCE_NOT_FOUND', '素材不存在');
|
||||
const expiresAt = new Date(Date.now() + 10 * 60_000);
|
||||
const url = await getSignedUrl(
|
||||
this.s3,
|
||||
new GetObjectCommand({
|
||||
Bucket: asset.rows[0].bucket,
|
||||
Key: asset.rows[0].object_key,
|
||||
}),
|
||||
{ expiresIn: 600 },
|
||||
);
|
||||
return { url, expires_at: expiresAt.toISOString() };
|
||||
});
|
||||
}
|
||||
|
||||
private async lockIntent(
|
||||
tx: DatabaseTransaction,
|
||||
intentId: string,
|
||||
): Promise<IntentRow> {
|
||||
const rows = await tx.query<IntentRow>(
|
||||
`SELECT id, storage_provider, bucket, object_key, expected_content_type, expected_byte_size,
|
||||
expected_checksum_sha256, status, expires_at, asset_id
|
||||
FROM atlas.media_upload_intents WHERE id = $1 FOR UPDATE`,
|
||||
[intentId],
|
||||
);
|
||||
if (!rows.rows[0])
|
||||
throw new ApiError(404, 'RESOURCE_NOT_FOUND', '上传意图不存在');
|
||||
return rows.rows[0];
|
||||
}
|
||||
|
||||
private async requireAsset(tx: DatabaseTransaction, assetId: string) {
|
||||
const rows = await tx.query<AssetRow>(
|
||||
`SELECT id, content_type, byte_size, width_px, height_px, created_at
|
||||
FROM atlas.media_assets WHERE id = $1`,
|
||||
[assetId],
|
||||
);
|
||||
if (!rows.rows[0])
|
||||
throw new ApiError(404, 'RESOURCE_NOT_FOUND', '素材不存在');
|
||||
const asset = rows.rows[0];
|
||||
return {
|
||||
id: asset.id,
|
||||
content_type: asset.content_type,
|
||||
byte_size: Number(asset.byte_size),
|
||||
width_px: asset.width_px,
|
||||
height_px: asset.height_px,
|
||||
created_at:
|
||||
asset.created_at instanceof Date
|
||||
? asset.created_at.toISOString()
|
||||
: asset.created_at,
|
||||
};
|
||||
}
|
||||
|
||||
private async inspectObject(intent: IntentRow) {
|
||||
let head;
|
||||
try {
|
||||
head = await this.s3.send(
|
||||
new HeadObjectCommand({
|
||||
Bucket: intent.bucket,
|
||||
Key: intent.object_key,
|
||||
}),
|
||||
);
|
||||
} catch {
|
||||
throw new ApiError(
|
||||
422,
|
||||
'UPLOAD_OBJECT_MISSING',
|
||||
'尚未找到已上传文件',
|
||||
);
|
||||
}
|
||||
if (!head.ContentLength || head.ContentLength > 5 * 1024 * 1024) {
|
||||
throw new ApiError(
|
||||
422,
|
||||
'UPLOAD_VALIDATION_FAILED',
|
||||
'上传文件大小不符合要求',
|
||||
);
|
||||
}
|
||||
const declaredContentType = head.ContentType?.split(
|
||||
';',
|
||||
1,
|
||||
)[0]?.toLowerCase();
|
||||
if (declaredContentType !== intent.expected_content_type) {
|
||||
throw new ApiError(
|
||||
422,
|
||||
'UPLOAD_VALIDATION_FAILED',
|
||||
'上传文件声明类型不符合要求',
|
||||
);
|
||||
}
|
||||
const object = await this.s3.send(
|
||||
new GetObjectCommand({
|
||||
Bucket: intent.bucket,
|
||||
Key: intent.object_key,
|
||||
}),
|
||||
);
|
||||
if (!object.Body)
|
||||
throw new ApiError(
|
||||
422,
|
||||
'UPLOAD_OBJECT_MISSING',
|
||||
'尚未找到已上传文件',
|
||||
);
|
||||
const chunks: Buffer[] = [];
|
||||
let byteSize = 0;
|
||||
for await (const chunk of object.Body as AsyncIterable<Uint8Array>) {
|
||||
byteSize += chunk.byteLength;
|
||||
// HeadObject 和 GetObject 之间对象可能被替换；读取时也必须有硬上限。
|
||||
if (byteSize > 5 * 1024 * 1024) {
|
||||
throw new ApiError(
|
||||
422,
|
||||
'UPLOAD_VALIDATION_FAILED',
|
||||
'上传文件大小不符合要求',
|
||||
);
|
||||
}
|
||||
chunks.push(Buffer.from(chunk));
|
||||
}
|
||||
const bytes = Buffer.concat(chunks, byteSize);
|
||||
const detected = detectImage(bytes);
|
||||
if (
|
||||
!detected ||
|
||||
detected.contentType !== intent.expected_content_type
|
||||
) {
|
||||
throw new ApiError(
|
||||
422,
|
||||
'UPLOAD_VALIDATION_FAILED',
|
||||
'上传文件类型或内容不符合要求',
|
||||
);
|
||||
}
|
||||
return {
|
||||
contentType: detected.contentType,
|
||||
width: detected.width,
|
||||
height: detected.height,
|
||||
byteSize,
|
||||
checksum: createHash('sha256').update(bytes).digest('hex'),
|
||||
};
|
||||
}
|
||||
|
||||
private async ensureBucket() {
|
||||
if (!this.bucketReady)
|
||||
this.bucketReady = (async () => {
|
||||
try {
|
||||
await this.s3.send(
|
||||
new HeadBucketCommand({
|
||||
Bucket: this.config.OBJECT_STORAGE_BUCKET,
|
||||
}),
|
||||
);
|
||||
} catch {
|
||||
await this.s3.send(
|
||||
new CreateBucketCommand({
|
||||
Bucket: this.config.OBJECT_STORAGE_BUCKET,
|
||||
}),
|
||||
);
|
||||
}
|
||||
})();
|
||||
return this.bucketReady;
|
||||
}
|
||||
|
||||
private async persistTerminalIntentStatus(
|
||||
workspaceId: string,
|
||||
intentId: string,
|
||||
code:
|
||||
| 'UPLOAD_EXPIRED'
|
||||
| 'UPLOAD_OBJECT_MISSING'
|
||||
| 'UPLOAD_VALIDATION_FAILED',
|
||||
) {
|
||||
await this.database.workspaceTransaction(workspaceId, async (tx) => {
|
||||
const expired = code === 'UPLOAD_EXPIRED';
|
||||
await tx.query(
|
||||
`UPDATE atlas.media_upload_intents
|
||||
SET status = $2, error_code = $3, error_message = $4, completed_at = now()
|
||||
WHERE id = $1 AND status = 'pending'`,
|
||||
[
|
||||
intentId,
|
||||
expired ? 'expired' : 'failed',
|
||||
code,
|
||||
expired ? '上传意图已过期' : '上传文件未通过安全校验',
|
||||
],
|
||||
);
|
||||
});
|
||||
}
|
||||
}
|
||||
|
||||
function apiErrorCode(error: unknown): string | undefined {
|
||||
if (!(error instanceof ApiError)) return undefined;
|
||||
const response = error.getResponse();
|
||||
if (typeof response !== 'object' || !response || !('error' in response))
|
||||
return undefined;
|
||||
const apiError = response.error;
|
||||
return typeof apiError === 'object' &&
|
||||
apiError &&
|
||||
'code' in apiError &&
|
||||
typeof apiError.code === 'string'
|
||||
? apiError.code
|
||||
: undefined;
|
||||
}
|
||||
|
||||
function detectImage(bytes: Buffer): {
|
||||
contentType: 'image/png' | 'image/jpeg' | 'image/webp';
|
||||
width: number;
|
||||
height: number;
|
||||
} | null {
|
||||
if (
|
||||
bytes.length >= 24 &&
|
||||
bytes
|
||||
.subarray(0, 8)
|
||||
.equals(Buffer.from([137, 80, 78, 71, 13, 10, 26, 10]))
|
||||
) {
|
||||
return dimensions(
|
||||
'image/png',
|
||||
bytes.readUInt32BE(16),
|
||||
bytes.readUInt32BE(20),
|
||||
);
|
||||
}
|
||||
if (
|
||||
bytes.length >= 12 &&
|
||||
bytes.subarray(0, 4).toString() === 'RIFF' &&
|
||||
bytes.subarray(8, 12).toString() === 'WEBP'
|
||||
) {
|
||||
const type = bytes.subarray(12, 16).toString();
|
||||
if (type === 'VP8X' && bytes.length >= 30) {
|
||||
return dimensions(
|
||||
'image/webp',
|
||||
1 + bytes.readUIntLE(24, 3),
|
||||
1 + bytes.readUIntLE(27, 3),
|
||||
);
|
||||
}
|
||||
if (
|
||||
type === 'VP8 ' &&
|
||||
bytes.length >= 30 &&
|
||||
bytes[23] === 0x9d &&
|
||||
bytes[24] === 0x01 &&
|
||||
bytes[25] === 0x2a
|
||||
) {
|
||||
return dimensions(
|
||||
'image/webp',
|
||||
bytes.readUInt16LE(26) & 0x3fff,
|
||||
bytes.readUInt16LE(28) & 0x3fff,
|
||||
);
|
||||
}
|
||||
}
|
||||
if (bytes.length > 4 && bytes[0] === 0xff && bytes[1] === 0xd8) {
|
||||
for (let offset = 2; offset + 9 < bytes.length;) {
|
||||
if (bytes[offset] !== 0xff) return null;
|
||||
const marker = bytes[offset + 1];
|
||||
const length = bytes.readUInt16BE(offset + 2);
|
||||
if (length < 2 || offset + 2 + length > bytes.length) return null;
|
||||
if (
|
||||
[
|
||||
0xc0, 0xc1, 0xc2, 0xc3, 0xc5, 0xc6, 0xc7, 0xc9, 0xca, 0xcb,
|
||||
0xcd, 0xce, 0xcf,
|
||||
].includes(marker)
|
||||
) {
|
||||
return dimensions(
|
||||
'image/jpeg',
|
||||
bytes.readUInt16BE(offset + 7),
|
||||
bytes.readUInt16BE(offset + 5),
|
||||
);
|
||||
}
|
||||
offset += 2 + length;
|
||||
}
|
||||
}
|
||||
return null;
|
||||
}
|
||||
function dimensions(
|
||||
contentType: 'image/png' | 'image/jpeg' | 'image/webp',
|
||||
width: number,
|
||||
height: number,
|
||||
) {
|
||||
return width >= 1 && height >= 1 && width <= 8192 && height <= 8192
|
||||
? { contentType, width, height }
|
||||
: null;
|
||||
}
|
||||
Reference in New Issue
Block a user