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;
  assetType: string;
  storageProvider: StorageProviderName;
  bucket: string | null;
  objectKey: string | null;
  fileName: string | null;
  mimeType: string | null;
  fileSizeBytes: number | string | null;
  checksumSha256: string | null;
  verifiedSizeBytes: number | string | null;
  verifiedChecksumSha256: string | null;
  verificationDetails: Record<string, unknown>;
  securityScanStatus: string;
  securityScanSummary: 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;
}

interface SecurityScanResult {
  status: 'passed' | 'failed' | 'skipped';
  provider: string;
  riskLevel: 'none' | 'low' | 'medium' | 'high' | 'critical';
  issueCodes: string[];
  details: Record<string, unknown>;
}

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}$/;
const EXTENSION_MIME_HINTS: Record<string, string[]> = {
  pdf: ['application/pdf'],
  png: ['image/png'],
  jpg: ['image/jpeg'],
  jpeg: ['image/jpeg'],
  gif: ['image/gif'],
  webp: ['image/webp'],
  svg: ['image/svg+xml'],
  mp4: ['video/mp4'],
  mov: ['video/quicktime'],
  mp3: ['audio/mpeg'],
  wav: ['audio/wav', 'audio/x-wav'],
  doc: ['application/msword'],
  docx: ['application/vnd.openxmlformats-officedocument.wordprocessingml.document'],
  xls: ['application/vnd.ms-excel'],
  xlsx: ['application/vnd.openxmlformats-officedocument.spreadsheetml.sheet'],
  ppt: ['application/vnd.ms-powerpoint'],
  pptx: ['application/vnd.openxmlformats-officedocument.presentationml.presentation'],
  zip: ['application/zip', 'application/x-zip-compressed'],
  json: ['application/json'],
  txt: ['text/plain'],
  md: ['text/markdown', 'text/plain'],
  csv: ['text/csv', 'text/plain'],
};

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 extensionFromName(value: string | null) {
  if (!value) return '';
  const withoutQuery = value.split('?')[0]?.split('#')[0] || value;
  const last = withoutQuery.split('/').pop() || withoutQuery;
  const dot = last.lastIndexOf('.');
  return dot >= 0 ? last.slice(dot + 1).trim().toLowerCase() : '';
}

function metadataBoolean(value: unknown) {
  return value === true || value === 'true' || value === 1 || value === '1';
}

const RISK_LEVEL_ORDER: Record<SecurityScanResult['riskLevel'], number> = {
  none: 0,
  low: 1,
  medium: 2,
  high: 3,
  critical: 4,
};

function normalizeRiskLevel(value: unknown, fallback: SecurityScanResult['riskLevel'] = 'high'): SecurityScanResult['riskLevel'] {
  const normalized = typeof value === 'string' ? value.trim().toLowerCase() : '';
  if (normalized === 'none' || normalized === 'low' || normalized === 'medium' || normalized === 'high' || normalized === 'critical') {
    return normalized;
  }
  return fallback;
}

function highestRiskLevel(results: SecurityScanResult[]): SecurityScanResult['riskLevel'] {
  return results.reduce<SecurityScanResult['riskLevel']>((highest, result) => (
    RISK_LEVEL_ORDER[result.riskLevel] > RISK_LEVEL_ORDER[highest] ? result.riskLevel : highest
  ), 'none');
}

function normalizeIssueCodes(value: unknown, fallback: string[] = []) {
  const rawItems = Array.isArray(value) ? value : fallback;
  return Array.from(new Set(
    rawItems
      .map(item => String(item || '').trim())
      .filter(Boolean)
      .map(item => item.replace(/[^A-Za-z0-9_.:-]/g, '_').slice(0, 96))
      .filter(Boolean),
  ));
}

function scannerModes() {
  return Array.from(new Set(
    config.assetSecurityScanner
      .split(',')
      .map(item => item.trim().toLowerCase())
      .filter(Boolean),
  ));
}

function sanitizeProviderDetails(value: unknown, depth = 0): unknown {
  if (depth > 5) return '[max_depth]';
  if (value === null || value === undefined) return value;
  if (typeof value === 'string') return value.length > 2000 ? `${value.slice(0, 2000)}...` : value;
  if (typeof value === 'number' || typeof value === 'boolean') return value;
  if (Array.isArray(value)) return value.slice(0, 50).map(item => sanitizeProviderDetails(item, depth + 1));
  if (typeof value !== 'object') return String(value);

  const sanitized: Record<string, unknown> = {};
  for (const [key, child] of Object.entries(value as Record<string, unknown>).slice(0, 100)) {
    const normalizedKey = key.toLowerCase().replace(/[-_\s]/g, '');
    if (
      normalizedKey.includes('secret') ||
      normalizedKey.includes('token') ||
      normalizedKey.includes('password') ||
      normalizedKey.includes('privatekey') ||
      normalizedKey.includes('apikey') ||
      normalizedKey.includes('authorization')
    ) {
      sanitized[key] = '[redacted]';
      continue;
    }
    sanitized[key] = sanitizeProviderDetails(child, depth + 1);
  }
  return sanitized;
}

function endpointHost(value: string) {
  try {
    return new URL(value).host;
  } catch {
    return 'invalid_endpoint';
  }
}

function securityScanRules(asset: AssetCandidate, metadata: StorageObjectMetadata): SecurityScanResult {
  const issueCodes: string[] = [];
  const details: Record<string, unknown> = {
    observed: {
      fileSizeBytes: metadata.sizeBytes,
      mimeType: canonicalMime(metadata.mimeType),
      checksumSha256: metadata.checksumSha256,
      verificationSource: metadata.verificationSource,
    },
    declared: {
      assetType: asset.assetType,
      fileName: asset.fileName,
      objectKey: asset.objectKey,
      mimeType: canonicalMime(asset.mimeType),
      fileSizeBytes: validateFileSize(asset.fileSizeBytes),
      checksumSha256: asset.checksumSha256,
    },
  };

  try {
    if (asset.objectKey) validateObjectKey(asset.tenantId, asset.objectKey);
  } catch (error) {
    issueCodes.push(eventError(error).code);
  }

  try {
    validateFileSize(metadata.sizeBytes ?? asset.verifiedSizeBytes ?? asset.fileSizeBytes);
  } catch (error) {
    issueCodes.push(eventError(error).code);
  }

  const declaredMime = canonicalMime(asset.mimeType);
  const observedMime = canonicalMime(metadata.mimeType);
  const effectiveMime = observedMime || declaredMime;
  try {
    validateMimeType(effectiveMime);
  } catch (error) {
    issueCodes.push(eventError(error).code);
  }

  const fileExtension = extensionFromName(asset.fileName) || extensionFromName(asset.objectKey);
  details.fileExtension = fileExtension || null;
  if (fileExtension && effectiveMime && EXTENSION_MIME_HINTS[fileExtension]) {
    const allowedForExtension = EXTENSION_MIME_HINTS[fileExtension];
    if (!allowedForExtension.includes(effectiveMime)) {
      issueCodes.push('file_extension_mime_mismatch');
      details.extensionMimeExpected = allowedForExtension;
    }
  }

  const metadataFlags = objectValue(asset.securityScanSummary);
  const assetMetadata = objectValue(asset.verificationDetails?.metadata);
  const explicitFlag = [
    metadataFlags.forceFail,
    metadataFlags.securityScanForceFail,
    assetMetadata.securityScanForceFail,
    assetMetadata.forceSecurityScanFail,
  ].some(metadataBoolean);
  if (explicitFlag) {
    issueCodes.push('security_scan_forced_failure');
  }

  const forcedIssue = typeof metadataFlags.forceIssueCode === 'string'
    ? metadataFlags.forceIssueCode
    : typeof assetMetadata.forceIssueCode === 'string'
      ? assetMetadata.forceIssueCode
      : '';
  if (forcedIssue) issueCodes.push(forcedIssue);

  const uniqueIssues = Array.from(new Set(issueCodes));
  const highRiskIssues = new Set([
    'INVALID_OBJECT_KEY',
    'OBJECT_KEY_TENANT_PREFIX_REQUIRED',
    'MIME_TYPE_NOT_ALLOWED',
    'FILE_TOO_LARGE',
    'security_scan_forced_failure',
  ]);
  const riskLevel = uniqueIssues.some(code => highRiskIssues.has(code))
    ? 'high'
    : uniqueIssues.length
      ? 'medium'
      : 'none';

  return {
    status: uniqueIssues.length ? 'failed' : 'passed',
    provider: 'metadata_rules',
    riskLevel,
    issueCodes: uniqueIssues,
    details,
  };
}

function mergeSecurityScanResults(results: SecurityScanResult[]): SecurityScanResult {
  const uniqueIssueCodes = Array.from(new Set(results.flatMap(result => result.issueCodes)));
  const failed = results.find(result => result.status === 'failed');
  const skipped = results.find(result => result.status === 'skipped');
  return {
    status: failed ? 'failed' : skipped ? 'skipped' : 'passed',
    provider: results.map(result => result.provider).join('+'),
    riskLevel: highestRiskLevel(results),
    issueCodes: uniqueIssueCodes,
    details: {
      results: results.map(result => ({
        provider: result.provider,
        status: result.status,
        riskLevel: result.riskLevel,
        issueCodes: result.issueCodes,
        details: result.details,
      })),
    },
  };
}

async function externalHttpSecurityScan(
  asset: AssetCandidate,
  metadata: StorageObjectMetadata,
  metadataRulesResult: SecurityScanResult,
): Promise<SecurityScanResult> {
  const endpoint = config.assetSecurityScanHttpEndpoint.trim();
  const provider = 'http_security_scan';
  const unavailableIssue = ['external_security_scan_unavailable'];
  const unavailableDetails = (error: unknown) => ({
    endpointHost: endpoint ? endpointHost(endpoint) : null,
    failOpen: config.assetSecurityScanFailOpen,
    error: eventError(error),
  });

  if (!endpoint) {
    return {
      status: config.assetSecurityScanFailOpen ? 'passed' : 'failed',
      provider,
      riskLevel: config.assetSecurityScanFailOpen ? 'low' : 'high',
      issueCodes: unavailableIssue,
      details: unavailableDetails(new AssetWorkerError('WORKER_ASSET_SECURITY_SCAN_HTTP_ENDPOINT is required', 'ASSET_SECURITY_SCAN_HTTP_ENDPOINT_REQUIRED')),
    };
  }

  const payload = {
    assetId: asset.id,
    tenantId: asset.tenantId,
    assetType: asset.assetType,
    storageProvider: asset.storageProvider,
    bucket: asset.bucket,
    objectKey: asset.objectKey,
    fileName: asset.fileName,
    mimeType: canonicalMime(metadata.mimeType) || canonicalMime(asset.mimeType),
    fileSizeBytes: metadata.sizeBytes ?? normalizeSafeInteger(asset.fileSizeBytes),
    checksumSha256: metadata.checksumSha256 || asset.verifiedChecksumSha256 || asset.checksumSha256,
    metadata: {
      observed: {
        fileSizeBytes: metadata.sizeBytes,
        mimeType: canonicalMime(metadata.mimeType),
        checksumSha256: metadata.checksumSha256,
        etag: metadata.etag,
        lastModified: metadata.lastModified,
        verificationSource: metadata.verificationSource,
      },
      declared: objectValue(metadataRulesResult.details).declared || {},
    },
    requestedAt: nowIso(),
  };

  try {
    const response = await fetchWithTimeout(endpoint, {
      method: 'POST',
      headers: {
        'content-type': 'application/json',
        accept: 'application/json',
        ...(config.assetSecurityScanHttpToken ? { authorization: `Bearer ${config.assetSecurityScanHttpToken}` } : {}),
      },
      body: JSON.stringify(payload),
    }, config.assetSecurityScanHttpTimeoutMs);

    if (!response.ok) {
      throw new AssetWorkerError(`External security scanner returned HTTP ${response.status}`, 'ASSET_SECURITY_SCAN_HTTP_STATUS');
    }

    const body = await response.json() as Record<string, unknown>;
    const status = typeof body.status === 'string' ? body.status.trim().toLowerCase() : '';
    if (status !== 'passed' && status !== 'failed') {
      throw new AssetWorkerError('External security scanner returned invalid status', 'ASSET_SECURITY_SCAN_INVALID_RESPONSE');
    }

    const bodyIssueCodes = normalizeIssueCodes(body.issueCodes, status === 'failed' ? ['external_security_scan_failed'] : []);
    return {
      status,
      provider,
      riskLevel: normalizeRiskLevel(body.riskLevel, status === 'failed' ? 'high' : 'none'),
      issueCodes: bodyIssueCodes,
      details: {
        endpointHost: endpointHost(endpoint),
        httpStatus: response.status,
        scannerProvider: typeof body.provider === 'string' ? body.provider.slice(0, 120) : null,
        scannerDetails: sanitizeProviderDetails(body.details || {}),
      },
    };
  } catch (error) {
    return {
      status: config.assetSecurityScanFailOpen ? 'passed' : 'failed',
      provider,
      riskLevel: config.assetSecurityScanFailOpen ? 'low' : 'high',
      issueCodes: unavailableIssue,
      details: unavailableDetails(error),
    };
  }
}

async function runSecurityScans(asset: AssetCandidate, metadata: StorageObjectMetadata): Promise<SecurityScanResult> {
  const metadataRulesResult = securityScanRules(asset, metadata);
  if (metadataRulesResult.status !== 'passed') {
    return metadataRulesResult;
  }

  const modes = scannerModes();
  if (modes.length === 0 || (modes.length === 1 && modes[0] === 'metadata_rules')) {
    return metadataRulesResult;
  }

  const unsupportedModes = modes.filter(mode => mode !== 'metadata_rules' && mode !== 'http');
  if (unsupportedModes.length > 0) {
    return mergeSecurityScanResults([
      metadataRulesResult,
      {
        status: 'failed',
        provider: 'asset_security_scanner_config',
        riskLevel: 'high',
        issueCodes: ['asset_security_scanner_unsupported'],
        details: { configuredModes: unsupportedModes },
      },
    ]);
  }

  const results = [metadataRulesResult];
  if (modes.includes('http')) {
    results.push(await externalHttpSecurityScan(asset, metadata, metadataRulesResult));
  }
  return mergeSecurityScanResults(results);
}

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 recordSecurityScanEvent(
  client: pg.PoolClient,
  asset: AssetCandidate,
  scan: SecurityScanResult,
) {
  await client.query(
    `
    insert into public.content_asset_security_scan_events (
      tenant_id, asset_id, provider, scan_status, risk_level, issue_codes, details
    )
    values ($1, $2, $3, $4, $5, $6::text[], $7::jsonb)
    `,
    [
      asset.tenantId,
      asset.id,
      scan.provider,
      scan.status,
      scan.riskLevel,
      scan.issueCodes,
      JSON.stringify(scan.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 security_scan_status = 'pending'
            and updated_at <= now() - ($2::int * interval '1 second')
          )
          or (
            upload_status = 'verified'
            and security_scan_status in ('passed', 'not_required')
            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
        ),
        security_scan_status = case
          when ca.upload_status = 'verified' and ca.security_scan_status = 'pending' then 'scanning'
          else ca.security_scan_status
        end,
        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.asset_type as "assetType",
              ca.storage_provider as "storageProvider",
              ca.bucket,
              ca.object_key as "objectKey",
              ca.file_name as "fileName",
              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_scan_status as "securityScanStatus",
              ca.security_scan_summary as "securityScanSummary",
              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 markAssetSecurityScanPassed(
  client: pg.PoolClient,
  asset: AssetCandidate,
  scan: SecurityScanResult,
) {
  await client.query(
    `
    update public.content_assets
    set security_scan_status = 'passed',
        security_scanned_at = now(),
        security_scan_provider = $3,
        security_scan_summary = $4::jsonb,
        security_flags = coalesce(security_flags, '{}'::jsonb) - 'assetSecurityScanFailed',
        updated_at = now()
    where tenant_id = $1 and id = $2
    `,
    [
      asset.tenantId,
      asset.id,
      scan.provider,
      JSON.stringify({
        riskLevel: scan.riskLevel,
        issueCodes: scan.issueCodes,
        provider: scan.provider,
        scannedAt: nowIso(),
        details: scan.details,
      }),
    ],
  );

  await recordSecurityScanEvent(client, asset, scan);
  await recordAudit(client, {
    tenantId: asset.tenantId,
    action: 'content.asset.security_scan_passed',
    targetId: asset.id,
    details: {
      provider: scan.provider,
      riskLevel: scan.riskLevel,
      issueCodes: scan.issueCodes,
    },
  });
}

async function markAssetSecurityScanFailed(
  client: pg.PoolClient,
  asset: AssetCandidate,
  scan: SecurityScanResult,
) {
  await client.query(
    `
    update public.content_assets
    set security_scan_status = 'failed',
        security_scanned_at = now(),
        security_scan_provider = $3,
        security_scan_summary = $4::jsonb,
        status = case when status = 'active' then 'draft' else status end,
        security_flags = coalesce(security_flags, '{}'::jsonb)
          || jsonb_build_object('assetSecurityScanFailed', true, 'assetSecurityScanFailedAt', now()),
        updated_at = now()
    where tenant_id = $1 and id = $2
    `,
    [
      asset.tenantId,
      asset.id,
      scan.provider,
      JSON.stringify({
        riskLevel: scan.riskLevel,
        issueCodes: scan.issueCodes,
        provider: scan.provider,
        scannedAt: nowIso(),
        details: scan.details,
      }),
    ],
  );

  await recordSecurityScanEvent(client, asset, scan);
  await recordAudit(client, {
    tenantId: asset.tenantId,
    action: 'content.asset.security_scan_failed',
    targetId: asset.id,
    details: {
      provider: scan.provider,
      riskLevel: scan.riskLevel,
      issueCodes: scan.issueCodes,
      unpublished: asset.status === 'active',
    },
  });
}

async function markAssetFailed(
  client: pg.PoolClient,
  asset: AssetCandidate,
  issues: string[],
  metadata: StorageObjectMetadata | null,
  error: Record<string, unknown> | null,
) {
  const uniqueIssues = Array.from(new Set(issues));
  const details = mergeVerificationDetails(asset, {
    lastCheckedAt: nowIso(),
    lastResult: 'failed',
    issues: uniqueIssues,
    observed: metadata
      ? {
        fileSizeBytes: metadata.sizeBytes,
        mimeType: canonicalMime(metadata.mimeType),
        checksumSha256: metadata.checksumSha256,
        etag: metadata.etag,
        lastModified: metadata.lastModified,
        verificationSource: metadata.verificationSource,
      }
      : null,
    error,
  });
  const scanSkipped: SecurityScanResult = {
    status: 'skipped',
    provider: 'metadata_rules',
    riskLevel: 'medium',
    issueCodes: uniqueIssues,
    details: {
      skippedReason: 'upload_recheck_failed',
      observed: metadata
        ? {
          fileSizeBytes: metadata.sizeBytes,
          mimeType: canonicalMime(metadata.mimeType),
          checksumSha256: metadata.checksumSha256,
          verificationSource: metadata.verificationSource,
        }
        : null,
      error,
    },
  };

  await client.query(
    `
    update public.content_assets
    set upload_status = 'failed',
        security_scan_status = 'skipped',
        security_scanned_at = null,
        security_scan_provider = $4,
        security_scan_summary = $5::jsonb,
        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),
      scanSkipped.provider,
      JSON.stringify({
        riskLevel: scanSkipped.riskLevel,
        issueCodes: scanSkipped.issueCodes,
        provider: scanSkipped.provider,
        skippedReason: 'upload_recheck_failed',
        details: scanSkipped.details,
      }),
    ],
  );

  await recordSecurityScanEvent(client, asset, scanSkipped);
  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: uniqueIssues,
      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);
    const scan = comparison.issues.length ? null : await runSecurityScans(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);
    if (scan?.status === 'failed') {
      await markAssetSecurityScanFailed(client, asset, scan);
      await client.query('commit');
      return 'failed';
    }
    if (scan?.status === 'passed') {
      await markAssetSecurityScanPassed(client, asset, scan);
    }
    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;
}
