feat: recheck content assets

apps/worker/package.json

@@ -9,9 +9,12 @@
     "build": "node -e \"fs.rmSync('dist',{recursive:true,force:true})\" && tsc -p tsconfig.json",
     "check": "tsc -p tsconfig.json --noEmit",
     "crm:once": "tsx src/index.ts --once --job crm",
-    "commerce:once": "tsx src/index.ts --once --job commerce"
+    "commerce:once": "tsx src/index.ts --once --job commerce",
+    "assets:once": "tsx src/index.ts --once --job assets"
   },
   "dependencies": {
+    "@supabase/storage-js": "^2.108.2",
+    "ali-oss": "^6.23.0",
     "pg": "^8.16.3"
   },
   "devDependencies": {

apps/worker/src/config.ts

@@ -13,6 +13,27 @@ export interface WorkerConfig {
   commerceBatchSize: number;
   commerceMinAgeSeconds: number;
   commerceRequestTimeoutMs: number;
+  assetBatchSize: number;
+  assetMinAgeSeconds: number;
+  assetRecheckIntervalSeconds: number;
+  assetRequestTimeoutMs: number;
+  storageMaxUploadBytes: number;
+  storageAllowedMimePrefixes: string[];
+  storageAllowedMimeTypes: string[];
+  storageRequireTenantPrefix: boolean;
+  aliyunOssRegion: string;
+  aliyunOssEndpoint: string;
+  aliyunOssAccessKeyId: string;
+  aliyunOssAccessKeySecret: string;
+  aliyunOssStsToken: string;
+  aliyunOssInternal: boolean;
+  tencentCosRegion: string;
+  tencentCosAppId: string;
+  tencentCosSecretId: string;
+  tencentCosSecretKey: string;
+  tencentCosSecurityToken: string;
+  supabaseStorageUrl: string;
+  supabaseStorageServiceKey: string;
 }
 
 export const config: WorkerConfig = {
@@ -28,4 +49,43 @@ export const config: WorkerConfig = {
   commerceBatchSize: envNumber('WORKER_COMMERCE_BATCH_SIZE', 20),
   commerceMinAgeSeconds: envNumber('WORKER_COMMERCE_MIN_AGE_SECONDS', 300),
   commerceRequestTimeoutMs: envNumber('WORKER_COMMERCE_REQUEST_TIMEOUT_MS', 10_000),
+  assetBatchSize: envNumber('WORKER_ASSET_BATCH_SIZE', 50),
+  assetMinAgeSeconds: envNumber('WORKER_ASSET_MIN_AGE_SECONDS', 300),
+  assetRecheckIntervalSeconds: envNumber('WORKER_ASSET_RECHECK_INTERVAL_SECONDS', 60 * 60 * 24),
+  assetRequestTimeoutMs: envNumber('WORKER_ASSET_REQUEST_TIMEOUT_MS', 10_000),
+  storageMaxUploadBytes: envNumber('STORAGE_MAX_UPLOAD_BYTES', 1024 * 1024 * 500),
+  storageAllowedMimePrefixes: envList('STORAGE_ALLOWED_MIME_PREFIXES', 'image/,video/,audio/'),
+  storageAllowedMimeTypes: envList(
+    'STORAGE_ALLOWED_MIME_TYPES',
+    [
+      'application/pdf',
+      'application/json',
+      'application/zip',
+      'application/x-zip-compressed',
+      'application/msword',
+      'application/vnd.openxmlformats-officedocument.wordprocessingml.document',
+      'application/vnd.ms-excel',
+      'application/vnd.openxmlformats-officedocument.spreadsheetml.sheet',
+      'application/vnd.ms-powerpoint',
+      'application/vnd.openxmlformats-officedocument.presentationml.presentation',
+      'application/octet-stream',
+      'text/plain',
+      'text/markdown',
+      'text/csv',
+    ].join(','),
+  ),
+  storageRequireTenantPrefix: envBoolean('STORAGE_REQUIRE_TENANT_PREFIX', true),
+  aliyunOssRegion: envString('ALIYUN_OSS_REGION', ''),
+  aliyunOssEndpoint: envString('ALIYUN_OSS_ENDPOINT', ''),
+  aliyunOssAccessKeyId: envString('ALIYUN_OSS_ACCESS_KEY_ID', ''),
+  aliyunOssAccessKeySecret: envString('ALIYUN_OSS_ACCESS_KEY_SECRET', ''),
+  aliyunOssStsToken: envString('ALIYUN_OSS_STS_TOKEN', ''),
+  aliyunOssInternal: envBoolean('ALIYUN_OSS_INTERNAL', false),
+  tencentCosRegion: envString('TENCENT_COS_REGION', ''),
+  tencentCosAppId: envString('TENCENT_COS_APP_ID', ''),
+  tencentCosSecretId: envString('TENCENT_COS_SECRET_ID', ''),
+  tencentCosSecretKey: envString('TENCENT_COS_SECRET_KEY', ''),
+  tencentCosSecurityToken: envString('TENCENT_COS_SECURITY_TOKEN', ''),
+  supabaseStorageUrl: envString('SUPABASE_STORAGE_URL', ''),
+  supabaseStorageServiceKey: envString('SUPABASE_STORAGE_SERVICE_KEY', ''),
 };

apps/worker/src/index.ts

@@ -2,6 +2,7 @@ import { closePool } from './db.js';
 import { config } from './config.js';
 import { processCrmBatch } from './jobs/crm.js';
 import { processCommerceBatch } from './jobs/commerce.js';
+import { processAssetBatch } from './jobs/assets.js';
 
 function hasArg(name: string) {
   return process.argv.includes(name);
@@ -28,6 +29,14 @@ async function runOnce() {
     );
     return;
   }
+  if (job === 'assets') {
+    const result = await processAssetBatch();
+    console.log(
+      `[worker] assets batch processed=${result.processed}`
+        + ` verified=${result.verified} failed=${result.failed} skipped=${result.skipped} errors=${result.errors}`,
+    );
+    return;
+  }
   throw new Error(`Unsupported worker job: ${job}`);
 }
 

apps/worker/src/jobs/assets.ts

@@ -0,0 +1,715 @@
+import crypto from 'node:crypto';
+import type pg from 'pg';
+import { pool } from '../db.js';
+import { config } from '../config.js';
+
+type StorageProviderName = 'external_url' | 'supabase_storage' | 'aliyun_oss' | 'tencent_cos' | 'qiniu_kodo' | 'local_dev';
+
+interface AssetCandidate {
+  id: string;
+  tenantId: string;
+  title: string | null;
+  status: string;
+  uploadStatus: string;
+  storageProvider: StorageProviderName;
+  bucket: string | null;
+  objectKey: string | null;
+  mimeType: string | null;
+  fileSizeBytes: number | string | null;
+  checksumSha256: string | null;
+  verifiedSizeBytes: number | string | null;
+  verifiedChecksumSha256: string | null;
+  verificationDetails: Record<string, unknown>;
+  securityFlags: Record<string, unknown>;
+}
+
+interface StorageObjectMetadata {
+  provider: StorageProviderName;
+  bucket: string | null;
+  objectKey: string | null;
+  exists: boolean;
+  sizeBytes: number | null;
+  mimeType: string | null;
+  checksumSha256: string | null;
+  etag: string | null;
+  lastModified: string | null;
+  rawHeaders: Record<string, string>;
+  verificationSource: string;
+}
+
+interface AssetWorkerResult {
+  processed: number;
+  verified: number;
+  failed: number;
+  skipped: number;
+  errors: number;
+}
+
+class AssetWorkerError extends Error {
+  readonly code: string;
+
+  constructor(message: string, code = 'ASSET_WORKER_ERROR') {
+    super(message);
+    this.code = code;
+  }
+}
+
+const UPLOADABLE_PROVIDERS = new Set<StorageProviderName>(['local_dev', 'supabase_storage', 'aliyun_oss', 'tencent_cos']);
+const SAFE_OBJECT_KEY_RE = /^[A-Za-z0-9][A-Za-z0-9._~!$&'()+,;=@/-]{0,1023}$/;
+
+function objectValue(value: unknown): Record<string, unknown> {
+  return value && typeof value === 'object' && !Array.isArray(value) ? value as Record<string, unknown> : {};
+}
+
+function nowIso() {
+  return new Date().toISOString();
+}
+
+function canonicalObjectKey(objectKey: string) {
+  return objectKey.replace(/^\/+/, '').replace(/\/{2,}/g, '/');
+}
+
+function validateObjectKey(tenantId: string, objectKey: string) {
+  const clean = canonicalObjectKey(objectKey);
+  if (!clean || clean.includes('..') || clean.includes('\\') || clean.includes('%2f') || clean.includes('%2F')) {
+    throw new AssetWorkerError('Invalid objectKey', 'INVALID_OBJECT_KEY');
+  }
+  if (!SAFE_OBJECT_KEY_RE.test(clean)) {
+    throw new AssetWorkerError('objectKey contains unsafe characters', 'INVALID_OBJECT_KEY');
+  }
+  if (config.storageRequireTenantPrefix && !clean.startsWith(`${tenantId}/`)) {
+    throw new AssetWorkerError('objectKey must be scoped by tenantId prefix', 'OBJECT_KEY_TENANT_PREFIX_REQUIRED');
+  }
+  return clean;
+}
+
+function normalizeSafeInteger(value: number | string | null) {
+  if (value === null) return null;
+  const numberValue = typeof value === 'number' ? value : Number(value);
+  if (!Number.isSafeInteger(numberValue)) {
+    throw new AssetWorkerError('fileSizeBytes must be a non-negative safe integer', 'INVALID_FILE_SIZE');
+  }
+  return numberValue;
+}
+
+function validateFileSize(fileSizeBytes: number | string | null) {
+  if (fileSizeBytes === null) return null;
+  const normalized = normalizeSafeInteger(fileSizeBytes);
+  if (normalized === null || normalized < 0) {
+    throw new AssetWorkerError('fileSizeBytes must be a non-negative safe integer', 'INVALID_FILE_SIZE');
+  }
+  if (normalized > config.storageMaxUploadBytes) {
+    throw new AssetWorkerError('file exceeds STORAGE_MAX_UPLOAD_BYTES', 'FILE_TOO_LARGE');
+  }
+  return normalized;
+}
+
+function validateMimeType(mimeType: string | null) {
+  if (!mimeType) return null;
+  const normalized = mimeType.trim().toLowerCase();
+  if (!normalized || normalized.length > 160 || normalized.includes('\r') || normalized.includes('\n')) {
+    throw new AssetWorkerError('Invalid mimeType', 'INVALID_MIME_TYPE');
+  }
+  const exact = config.storageAllowedMimeTypes.map(item => item.toLowerCase());
+  const prefixes = config.storageAllowedMimePrefixes.map(item => item.toLowerCase());
+  if (!exact.includes(normalized) && !prefixes.some(prefix => normalized.startsWith(prefix))) {
+    throw new AssetWorkerError(`mimeType is not allowed: ${normalized}`, 'MIME_TYPE_NOT_ALLOWED');
+  }
+  return normalized;
+}
+
+function normalizeHeaderMap(headers: Headers | Record<string, unknown>) {
+  const normalized: Record<string, string> = {};
+  if (headers instanceof Headers) {
+    headers.forEach((value, key) => {
+      normalized[key.toLowerCase()] = value;
+    });
+    return normalized;
+  }
+  for (const [key, value] of Object.entries(headers)) {
+    if (value === undefined || value === null) continue;
+    normalized[key.toLowerCase()] = Array.isArray(value) ? String(value[0] || '') : String(value);
+  }
+  return normalized;
+}
+
+function numberHeader(headers: Record<string, string>, key: string) {
+  const value = Number(headers[key]);
+  return Number.isFinite(value) && value >= 0 ? Math.trunc(value) : null;
+}
+
+function firstHeader(headers: Record<string, string>, keys: string[]) {
+  for (const key of keys) {
+    const value = headers[key.toLowerCase()];
+    if (value) return value;
+  }
+  return null;
+}
+
+function cleanEtag(value: string | null) {
+  return value ? value.replace(/^"+|"+$/g, '') : null;
+}
+
+function metadataFromHeaders(input: {
+  provider: StorageProviderName;
+  bucket: string | null;
+  objectKey: string | null;
+  headers: Record<string, string>;
+  verificationSource: string;
+}): StorageObjectMetadata {
+  const checksumSha256 = firstHeader(input.headers, [
+    'x-oss-meta-sha256',
+    'x-oss-meta-checksum-sha256',
+    'x-cos-meta-sha256',
+    'x-cos-meta-checksum-sha256',
+    'x-amz-meta-sha256',
+    'x-amz-meta-checksum-sha256',
+  ]);
+
+  return {
+    provider: input.provider,
+    bucket: input.bucket,
+    objectKey: input.objectKey,
+    exists: true,
+    sizeBytes: numberHeader(input.headers, 'content-length'),
+    mimeType: firstHeader(input.headers, ['content-type']),
+    checksumSha256: checksumSha256?.trim().toLowerCase() || null,
+    etag: cleanEtag(firstHeader(input.headers, ['etag'])),
+    lastModified: firstHeader(input.headers, ['last-modified']),
+    rawHeaders: input.headers,
+    verificationSource: input.verificationSource,
+  };
+}
+
+function hmacSha1Hex(key: string | Buffer, value: string) {
+  return crypto.createHmac('sha1', key).update(value).digest('hex');
+}
+
+function sha1Hex(value: string) {
+  return crypto.createHash('sha1').update(value).digest('hex');
+}
+
+function cosEncodePath(objectKey: string) {
+  return objectKey
+    .split('/')
+    .map(part => encodeURIComponent(part).replace(/[!'()*]/g, char => `%${char.charCodeAt(0).toString(16).toUpperCase()}`))
+    .join('/');
+}
+
+function requireConfigured(condition: unknown, provider: StorageProviderName, missing: string) {
+  if (!condition) {
+    throw new AssetWorkerError(`${provider} is not configured: ${missing}`, 'STORAGE_PROVIDER_NOT_CONFIGURED');
+  }
+}
+
+function cosHost(bucket: string) {
+  requireConfigured(config.tencentCosRegion, 'tencent_cos', 'TENCENT_COS_REGION');
+  const bucketWithAppId = config.tencentCosAppId && !bucket.endsWith(`-${config.tencentCosAppId}`)
+    ? `${bucket}-${config.tencentCosAppId}`
+    : bucket;
+  return `${bucketWithAppId}.cos.${config.tencentCosRegion}.myqcloud.com`;
+}
+
+function nowSeconds() {
+  return Math.floor(Date.now() / 1000);
+}
+
+function signTencentCosHead(input: { bucket: string; objectKey: string; expiresInSec: number }) {
+  requireConfigured(config.tencentCosSecretId, 'tencent_cos', 'TENCENT_COS_SECRET_ID');
+  requireConfigured(config.tencentCosSecretKey, 'tencent_cos', 'TENCENT_COS_SECRET_KEY');
+  const host = cosHost(input.bucket);
+  const start = nowSeconds();
+  const end = start + input.expiresInSec;
+  const keyTime = `${start};${end}`;
+  const pathname = `/${cosEncodePath(input.objectKey)}`;
+  const signedHeaders: Record<string, string> = { host };
+  const headerKeys = Object.keys(signedHeaders).sort();
+  const headerList = headerKeys.join(';');
+  const httpHeaders = headerKeys
+    .map(key => `${encodeURIComponent(key)}=${encodeURIComponent(signedHeaders[key]).toLowerCase()}`)
+    .join('&');
+  const signedQuery: Record<string, string> = {};
+  if (config.tencentCosSecurityToken) signedQuery['x-cos-security-token'] = config.tencentCosSecurityToken;
+  const queryKeys = Object.keys(signedQuery).sort();
+  const urlParamList = queryKeys.join(';');
+  const httpParameters = queryKeys
+    .map(key => `${encodeURIComponent(key)}=${encodeURIComponent(signedQuery[key])}`)
+    .join('&');
+  const httpString = `head\n${pathname}\n${httpParameters}\n${httpHeaders}\n`;
+  const stringToSign = `sha1\n${keyTime}\n${sha1Hex(httpString)}\n`;
+  const signKey = hmacSha1Hex(config.tencentCosSecretKey, keyTime);
+  const signature = hmacSha1Hex(signKey, stringToSign);
+  const query = new URLSearchParams();
+  query.set('q-sign-algorithm', 'sha1');
+  query.set('q-ak', config.tencentCosSecretId);
+  query.set('q-sign-time', keyTime);
+  query.set('q-key-time', keyTime);
+  query.set('q-header-list', headerList);
+  query.set('q-url-param-list', urlParamList);
+  query.set('q-signature', signature);
+  for (const key of queryKeys) query.set(key, signedQuery[key]);
+  return `https://${host}${pathname}?${query.toString()}`;
+}
+
+async function fetchWithTimeout(url: string, init: RequestInit, timeoutMs: number) {
+  const controller = new AbortController();
+  const timeout = setTimeout(() => controller.abort(), timeoutMs);
+  try {
+    return await fetch(url, { ...init, signal: controller.signal });
+  } finally {
+    clearTimeout(timeout);
+  }
+}
+
+async function headAliyunOssObject(asset: AssetCandidate): Promise<StorageObjectMetadata> {
+  if (!asset.bucket || !asset.objectKey) {
+    throw new AssetWorkerError('Aliyun OSS asset requires bucket and objectKey', 'ASSET_OBJECT_LOCATION_REQUIRED');
+  }
+  requireConfigured(config.aliyunOssAccessKeyId, 'aliyun_oss', 'ALIYUN_OSS_ACCESS_KEY_ID');
+  requireConfigured(config.aliyunOssAccessKeySecret, 'aliyun_oss', 'ALIYUN_OSS_ACCESS_KEY_SECRET');
+  requireConfigured(config.aliyunOssRegion || config.aliyunOssEndpoint, 'aliyun_oss', 'ALIYUN_OSS_REGION or ALIYUN_OSS_ENDPOINT');
+
+  const { default: OSS } = await import('ali-oss');
+  const client = new OSS({
+    region: config.aliyunOssRegion || undefined,
+    endpoint: config.aliyunOssEndpoint || undefined,
+    accessKeyId: config.aliyunOssAccessKeyId,
+    accessKeySecret: config.aliyunOssAccessKeySecret,
+    stsToken: config.aliyunOssStsToken || undefined,
+    bucket: asset.bucket,
+    internal: config.aliyunOssInternal,
+    secure: true,
+  });
+
+  try {
+    const response = await client.head(asset.objectKey);
+    const headers = normalizeHeaderMap(response.res?.headers || response);
+    return metadataFromHeaders({
+      provider: 'aliyun_oss',
+      bucket: asset.bucket,
+      objectKey: asset.objectKey,
+      headers,
+      verificationSource: 'aliyun-oss-head-object',
+    });
+  } catch (error) {
+    const status = Number((error as { status?: number; statusCode?: number }).status || (error as { statusCode?: number }).statusCode || 0);
+    if (status === 404) {
+      throw new AssetWorkerError('Object was not found in Aliyun OSS', 'STORAGE_OBJECT_NOT_FOUND');
+    }
+    throw error;
+  }
+}
+
+async function headTencentCosObject(asset: AssetCandidate): Promise<StorageObjectMetadata> {
+  if (!asset.bucket || !asset.objectKey) {
+    throw new AssetWorkerError('Tencent COS asset requires bucket and objectKey', 'ASSET_OBJECT_LOCATION_REQUIRED');
+  }
+  const url = signTencentCosHead({ bucket: asset.bucket, objectKey: asset.objectKey, expiresInSec: 60 });
+  const response = await fetchWithTimeout(url, { method: 'HEAD' }, config.assetRequestTimeoutMs);
+  if (response.status === 404) {
+    throw new AssetWorkerError('Object was not found in Tencent COS', 'STORAGE_OBJECT_NOT_FOUND');
+  }
+  if (!response.ok) {
+    throw new AssetWorkerError(`Tencent COS object metadata check failed: ${response.status}`, 'STORAGE_HEAD_FAILED');
+  }
+  return metadataFromHeaders({
+    provider: 'tencent_cos',
+    bucket: asset.bucket,
+    objectKey: asset.objectKey,
+    headers: normalizeHeaderMap(response.headers),
+    verificationSource: 'tencent-cos-head-object',
+  });
+}
+
+async function headSupabaseStorageObject(asset: AssetCandidate): Promise<StorageObjectMetadata> {
+  if (!asset.bucket || !asset.objectKey) {
+    throw new AssetWorkerError('Supabase Storage asset requires bucket and objectKey', 'ASSET_OBJECT_LOCATION_REQUIRED');
+  }
+  requireConfigured(config.supabaseStorageUrl, 'supabase_storage', 'SUPABASE_STORAGE_URL');
+  requireConfigured(config.supabaseStorageServiceKey, 'supabase_storage', 'SUPABASE_STORAGE_SERVICE_KEY');
+  const { StorageClient } = await import('@supabase/storage-js');
+  const client = new StorageClient(config.supabaseStorageUrl.replace(/\/+$/, ''), {
+    apikey: config.supabaseStorageServiceKey,
+    authorization: `Bearer ${config.supabaseStorageServiceKey}`,
+  });
+  const parts = asset.objectKey.split('/');
+  const name = parts.pop() || '';
+  const folder = parts.join('/');
+  const response = await client.from(asset.bucket).list(folder || undefined, {
+    search: name,
+    limit: 20,
+  });
+  if (response.error) {
+    throw new AssetWorkerError(response.error.message || 'Supabase Storage metadata check failed', 'STORAGE_HEAD_FAILED');
+  }
+  const file = response.data?.find(item => item.name === name);
+  if (!file) {
+    throw new AssetWorkerError('Object was not found in Supabase Storage', 'STORAGE_OBJECT_NOT_FOUND');
+  }
+  const metadata = (file as { metadata?: Record<string, unknown> }).metadata || {};
+  const headers: Record<string, string> = {};
+  for (const [key, value] of Object.entries(metadata)) {
+    if (value !== undefined && value !== null) headers[key.toLowerCase()] = String(value);
+  }
+  const size = Number(metadata.size ?? metadata.contentLength);
+  return {
+    provider: 'supabase_storage',
+    bucket: asset.bucket,
+    objectKey: asset.objectKey,
+    exists: true,
+    sizeBytes: Number.isFinite(size) && size >= 0 ? Math.trunc(size) : null,
+    mimeType: typeof metadata.mimetype === 'string' ? metadata.mimetype : typeof metadata.contentType === 'string' ? metadata.contentType : null,
+    checksumSha256: typeof metadata.sha256 === 'string' ? metadata.sha256.trim().toLowerCase() : null,
+    etag: typeof metadata.eTag === 'string' ? metadata.eTag : typeof metadata.etag === 'string' ? metadata.etag : null,
+    lastModified: typeof file.updated_at === 'string' ? file.updated_at : null,
+    rawHeaders: headers,
+    verificationSource: 'supabase-storage-list-metadata',
+  };
+}
+
+async function headStorageObject(asset: AssetCandidate): Promise<StorageObjectMetadata> {
+  if (!asset.objectKey) {
+    throw new AssetWorkerError('Asset requires objectKey', 'ASSET_LOCATION_REQUIRED');
+  }
+  const objectKey = validateObjectKey(asset.tenantId, asset.objectKey);
+  const fileSizeBytes = validateFileSize(asset.fileSizeBytes);
+  validateFileSize(asset.verifiedSizeBytes);
+  const mimeType = validateMimeType(asset.mimeType);
+  const checksumSha256 = asset.checksumSha256?.trim().toLowerCase() || null;
+
+  if (asset.storageProvider === 'local_dev') {
+    return {
+      provider: asset.storageProvider,
+      bucket: asset.bucket,
+      objectKey,
+      exists: true,
+      sizeBytes: fileSizeBytes,
+      mimeType,
+      checksumSha256,
+      etag: checksumSha256,
+      lastModified: nowIso(),
+      rawHeaders: {},
+      verificationSource: 'local-dev-declared-metadata',
+    };
+  }
+  if (asset.storageProvider === 'aliyun_oss') {
+    return headAliyunOssObject({ ...asset, objectKey });
+  }
+  if (asset.storageProvider === 'tencent_cos') {
+    return headTencentCosObject({ ...asset, objectKey });
+  }
+  if (asset.storageProvider === 'supabase_storage') {
+    return headSupabaseStorageObject({ ...asset, objectKey });
+  }
+  throw new AssetWorkerError(`${asset.storageProvider} does not support managed upload recheck`, 'UPLOAD_RECHECK_PROVIDER_NOT_SUPPORTED');
+}
+
+function canonicalMime(value: string | null) {
+  return value?.split(';')[0]?.trim().toLowerCase() || null;
+}
+
+function compareAssetMetadata(asset: AssetCandidate, metadata: StorageObjectMetadata) {
+  const issues: string[] = [];
+  const expectedSize = validateFileSize(asset.verifiedSizeBytes ?? asset.fileSizeBytes);
+  const expectedMime = canonicalMime(validateMimeType(asset.mimeType));
+  const observedMime = canonicalMime(metadata.mimeType);
+  const expectedChecksum = (asset.verifiedChecksumSha256 || asset.checksumSha256 || '').trim().toLowerCase() || null;
+
+  if (expectedSize !== null && metadata.sizeBytes !== null && expectedSize !== metadata.sizeBytes) {
+    issues.push('file_size_mismatch');
+  }
+  if (expectedMime && observedMime && expectedMime !== observedMime) {
+    issues.push('mime_type_mismatch');
+  }
+  if (expectedChecksum && metadata.checksumSha256 && expectedChecksum !== metadata.checksumSha256) {
+    issues.push('checksum_mismatch');
+  }
+
+  return {
+    issues,
+    expected: {
+      fileSizeBytes: expectedSize,
+      mimeType: expectedMime,
+      checksumSha256: expectedChecksum,
+    },
+    observed: {
+      fileSizeBytes: metadata.sizeBytes,
+      mimeType: observedMime,
+      checksumSha256: metadata.checksumSha256,
+      etag: metadata.etag,
+      lastModified: metadata.lastModified,
+      verificationSource: metadata.verificationSource,
+    },
+    checksumVerified: Boolean(expectedChecksum && metadata.checksumSha256 && expectedChecksum === metadata.checksumSha256),
+    checksumUnavailable: Boolean(expectedChecksum && !metadata.checksumSha256),
+  };
+}
+
+function eventError(error: unknown) {
+  if (error instanceof AssetWorkerError) {
+    return { code: error.code, message: error.message };
+  }
+  if (error instanceof Error) {
+    return { code: 'ASSET_WORKER_ERROR', message: error.message };
+  }
+  return { code: 'ASSET_WORKER_ERROR', message: String(error) };
+}
+
+function mergeVerificationDetails(asset: AssetCandidate, assetWorker: Record<string, unknown>) {
+  return {
+    ...asset.verificationDetails,
+    assetWorker: {
+      ...objectValue(asset.verificationDetails.assetWorker),
+      ...assetWorker,
+    },
+  };
+}
+
+async function recordAudit(
+  client: pg.PoolClient,
+  input: {
+    tenantId: string;
+    action: string;
+    targetId: string;
+    details: Record<string, unknown>;
+  },
+) {
+  await client.query(
+    `
+    insert into public.audit_logs (tenant_id, actor_user_id, action, target_type, target_id, details)
+    values ($1, null, $2, 'content_asset', $3, $4::jsonb)
+    `,
+    [input.tenantId, input.action, input.targetId, JSON.stringify(input.details)],
+  );
+}
+
+async function claimAssetCandidates(client: pg.PoolClient, limit: number, claimId: string) {
+  const result = await client.query<AssetCandidate>(
+    `
+    with candidates as (
+      select id
+      from public.content_assets
+      where storage_provider = any($1::text[])
+        and object_key is not null
+        and upload_status in ('pending', 'verified')
+        and (
+          (upload_status = 'pending' and updated_at <= now() - ($2::int * interval '1 second'))
+          or (
+            upload_status = 'verified'
+            and coalesce((verification_details #>> '{assetWorker,lastCheckedAt}')::timestamptz, verified_at, updated_at, created_at)
+              <= now() - ($3::int * interval '1 second')
+          )
+        )
+        and coalesce(verification_details #>> '{assetWorker,claimId}', '') <> $5
+      order by
+        case when upload_status = 'pending' then 0 else 1 end,
+        coalesce((verification_details #>> '{assetWorker,lastCheckedAt}')::timestamptz, verified_at, updated_at, created_at) asc
+      limit $4
+      for update skip locked
+    )
+    update public.content_assets ca
+    set verification_details = jsonb_set(
+          coalesce(ca.verification_details, '{}'::jsonb),
+          '{assetWorker}',
+          coalesce(ca.verification_details->'assetWorker', '{}'::jsonb) || $6::jsonb,
+          true
+        ),
+        updated_at = now()
+    from candidates
+    where ca.id = candidates.id
+    returning ca.id,
+              ca.tenant_id as "tenantId",
+              ca.title,
+              ca.status,
+              ca.upload_status as "uploadStatus",
+              ca.storage_provider as "storageProvider",
+              ca.bucket,
+              ca.object_key as "objectKey",
+              ca.mime_type as "mimeType",
+              ca.file_size_bytes as "fileSizeBytes",
+              ca.checksum_sha256 as "checksumSha256",
+              ca.verified_size_bytes as "verifiedSizeBytes",
+              ca.verified_checksum_sha256 as "verifiedChecksumSha256",
+              ca.verification_details as "verificationDetails",
+              ca.security_flags as "securityFlags"
+    `,
+    [
+      Array.from(UPLOADABLE_PROVIDERS),
+      config.assetMinAgeSeconds,
+      config.assetRecheckIntervalSeconds,
+      limit,
+      claimId,
+      JSON.stringify({ claimId, claimedAt: nowIso() }),
+    ],
+  );
+  return result.rows;
+}
+
+async function markAssetVerified(client: pg.PoolClient, asset: AssetCandidate, metadata: StorageObjectMetadata, comparison: ReturnType<typeof compareAssetMetadata>) {
+  const details = mergeVerificationDetails(asset, {
+    lastCheckedAt: nowIso(),
+    lastResult: 'verified',
+    issues: [],
+    observed: comparison.observed,
+    expected: comparison.expected,
+    checksumVerified: comparison.checksumVerified,
+    checksumUnavailable: comparison.checksumUnavailable,
+  });
+
+  await client.query(
+    `
+    update public.content_assets
+    set upload_status = 'verified',
+        verified_at = coalesce(verified_at, now()),
+        verified_size_bytes = coalesce($3::bigint, verified_size_bytes),
+        verified_checksum_sha256 = coalesce($4, verified_checksum_sha256),
+        file_size_bytes = coalesce(file_size_bytes, $3::bigint),
+        mime_type = coalesce(mime_type, $5),
+        verification_details = $6::jsonb,
+        security_flags = coalesce(security_flags, '{}'::jsonb) - 'assetRecheckFailed',
+        updated_at = now()
+    where tenant_id = $1 and id = $2
+    `,
+    [
+      asset.tenantId,
+      asset.id,
+      metadata.sizeBytes ?? comparison.expected.fileSizeBytes,
+      metadata.checksumSha256 ?? comparison.expected.checksumSha256,
+      comparison.observed.mimeType || comparison.expected.mimeType,
+      JSON.stringify(details),
+    ],
+  );
+
+  await recordAudit(client, {
+    tenantId: asset.tenantId,
+    action: 'content.asset.rechecked',
+    targetId: asset.id,
+    details: {
+      provider: asset.storageProvider,
+      bucket: asset.bucket,
+      objectKey: asset.objectKey,
+      result: 'verified',
+      checksumUnavailable: comparison.checksumUnavailable,
+    },
+  });
+}
+
+async function markAssetFailed(
+  client: pg.PoolClient,
+  asset: AssetCandidate,
+  issues: string[],
+  metadata: StorageObjectMetadata | null,
+  error: Record<string, unknown> | null,
+) {
+  const details = mergeVerificationDetails(asset, {
+    lastCheckedAt: nowIso(),
+    lastResult: 'failed',
+    issues,
+    observed: metadata
+      ? {
+        fileSizeBytes: metadata.sizeBytes,
+        mimeType: canonicalMime(metadata.mimeType),
+        checksumSha256: metadata.checksumSha256,
+        etag: metadata.etag,
+        lastModified: metadata.lastModified,
+        verificationSource: metadata.verificationSource,
+      }
+      : null,
+    error,
+  });
+
+  await client.query(
+    `
+    update public.content_assets
+    set upload_status = 'failed',
+        status = case when status = 'active' then 'draft' else status end,
+        verification_details = $3::jsonb,
+        security_flags = coalesce(security_flags, '{}'::jsonb)
+          || jsonb_build_object('assetRecheckFailed', true, 'assetRecheckFailedAt', now()),
+        updated_at = now()
+    where tenant_id = $1 and id = $2
+    `,
+    [asset.tenantId, asset.id, JSON.stringify(details)],
+  );
+
+  await recordAudit(client, {
+    tenantId: asset.tenantId,
+    action: 'content.asset.recheck_failed',
+    targetId: asset.id,
+    details: {
+      provider: asset.storageProvider,
+      bucket: asset.bucket,
+      objectKey: asset.objectKey,
+      result: 'failed',
+      issues,
+      error,
+      unpublished: asset.status === 'active',
+    },
+  });
+}
+
+async function processAsset(asset: AssetCandidate) {
+  const client = await pool.connect();
+  try {
+    const metadata = await headStorageObject(asset);
+    const comparison = compareAssetMetadata(asset, metadata);
+    await client.query('begin');
+    if (comparison.issues.length) {
+      await markAssetFailed(client, asset, comparison.issues, metadata, null);
+      await client.query('commit');
+      return 'failed';
+    }
+    await markAssetVerified(client, asset, metadata, comparison);
+    await client.query('commit');
+    return 'verified';
+  } catch (error) {
+    await client.query('rollback').catch(() => {});
+    try {
+      await client.query('begin');
+      await markAssetFailed(client, asset, [eventError(error).code], null, eventError(error));
+      await client.query('commit');
+    } catch {
+      await client.query('rollback').catch(() => {});
+    }
+    return 'error';
+  } finally {
+    client.release();
+  }
+}
+
+export async function processAssetBatch(limit = config.assetBatchSize): Promise<AssetWorkerResult> {
+  const claimId = crypto.randomUUID();
+  const client = await pool.connect();
+  let assets: AssetCandidate[] = [];
+  try {
+    await client.query('begin');
+    assets = await claimAssetCandidates(client, limit, claimId);
+    await client.query('commit');
+  } catch (error) {
+    await client.query('rollback');
+    throw error;
+  } finally {
+    client.release();
+  }
+
+  const result: AssetWorkerResult = {
+    processed: assets.length,
+    verified: 0,
+    failed: 0,
+    skipped: 0,
+    errors: 0,
+  };
+
+  for (const asset of assets) {
+    if (!UPLOADABLE_PROVIDERS.has(asset.storageProvider)) {
+      result.skipped += 1;
+      continue;
+    }
+    const status = await processAsset(asset);
+    if (status === 'verified') result.verified += 1;
+    else if (status === 'failed') result.failed += 1;
+    else result.errors += 1;
+  }
+
+  return result;
+}

apps/worker/src/types/ali-oss.d.ts

@@ -0,0 +1,25 @@
+declare module 'ali-oss' {
+  interface ClientOptions {
+    region?: string;
+    endpoint?: string;
+    accessKeyId: string;
+    accessKeySecret: string;
+    stsToken?: string;
+    bucket?: string;
+    internal?: boolean;
+    secure?: boolean;
+  }
+
+  interface HeadObjectResult {
+    res?: {
+      headers?: Record<string, string | string[]>;
+      status?: number;
+    };
+    [key: string]: unknown;
+  }
+
+  export default class OSS {
+    constructor(options: ClientOptions);
+    head(name: string): Promise<HeadObjectResult>;
+  }
+}