video-workflow/cloud-functions/15-fileAssetManager.js

/**
 * 云函数：fileAssetManager
 *
 * 记录七牛文件的归属和生命周期元数据。文件本体仍存放在七牛，
 * 元数据主存储为 Parse 项目命名空间 Class：VideoWorkflowFileAsset。
 */
const { requireSession, requireAdmin, assertRequestedUserMatchesSession } = require('./_session');
const {
  VIDEO_WORKFLOW_CLASSES,
  createParseClassStore,
  parseUserFields,
  parseUserWhere,
} = require('./_parseClassStore');

async function handler(request, response) {
  try {
    const action = pickParam(request, 'action') || 'list';
    if (action === 'adminInspect') {
      const adminSession = await requireAdmin(request, Psql);
      const store = createParseClassStore({ sessionToken: adminSession.sessionToken });
      return adminInspect(request, response, adminSession, store);
    }

    const session = await requireSession(request, Psql);
    assertRequestedUserMatchesSession(request, session);
    const store = createParseClassStore({ sessionToken: session.sessionToken });
    const userId = session.userId;

    if (action === 'register') return registerAsset(request, response, userId, store);
    if (action === 'get') return getAsset(request, response, userId, store);
    if (action === 'list') return listAssets(request, response, userId, store);
    if (action === 'stats') return getStats(request, response, userId, store);
    if (action === 'bind') return bindAsset(request, response, userId, store);
    if (action === 'delete') return deleteAsset(request, response, userId, store);
    if (action === 'purge') return purgeAsset(request, response, userId, store);

    return response.json({ code: 400, success: false, error: `Unknown action: ${action}` });
  } catch (error) {
    console.error('fileAssetManager failed:', error && error.message);
    return response.json({ code: error.status || 500, success: false, error: error.message || 'fileAssetManager failed' });
  }
}

async function registerAsset(request, response, userId, store) {
  const qiniuKey = requiredText(request, 'qiniuKey', 'key');
  assertUserScopedKey(qiniuKey, userId);
  const data = readAssetInput(request);
  const where = parseUserWhere(userId, { qiniuKey });
  const row = await store.upsertByQuery(VIDEO_WORKFLOW_CLASSES.fileAsset, where, {
    ...parseUserFields(userId),
    spaceId: data.spaceId,
    qiniuKey,
    url: data.url,
    bucket: data.bucket,
    mimeType: data.mimeType,
    kind: data.kind,
    sizeBytes: data.sizeBytes,
    sha256: data.sha256,
    sourceModule: data.sourceModule,
    bizType: data.bizType,
    bizId: data.bizId,
    status: data.status,
    metadata: data.metadata,
  });
  return response.json({ code: 200, success: true, data: rowToAsset(row) });
}

async function getAsset(request, response, userId, store) {
  const assetId = requiredText(request, 'assetId', 'id');
  const row = await store.findFirst(VIDEO_WORKFLOW_CLASSES.fileAsset, parseUserWhere(userId, { objectId: assetId }), { limit: 1 });
  return response.json({ code: 200, success: true, data: row ? rowToAsset(row) : null });
}

async function listAssets(request, response, userId, store) {
  const kind = String(pickParam(request, 'kind') || '').trim();
  const sourceModule = String(pickParam(request, 'sourceModule') || '').trim();
  const bizType = String(pickParam(request, 'bizType') || '').trim();
  const bizId = String(pickParam(request, 'bizId') || '').trim();
  const status = String(pickParam(request, 'status') || 'active').trim();
  const limit = safeLimit(pickParam(request, 'limit'), 50, 200);
  const where = parseUserWhere(userId, {
    ...(kind ? { kind } : {}),
    ...(sourceModule ? { sourceModule } : {}),
    ...(bizType ? { bizType } : {}),
    ...(bizId ? { bizId } : {}),
    ...(status ? { status } : {}),
  });
  const rows = await store.list(VIDEO_WORKFLOW_CLASSES.fileAsset, where, { order: '-createdAt', limit });
  return response.json({ code: 200, success: true, data: rows.map(rowToAsset) });
}

async function getStats(request, response, userId, store) {
  const rows = await listAll(store, VIDEO_WORKFLOW_CLASSES.fileAsset, parseUserWhere(userId, {}), 5000);
  return response.json({ code: 200, success: true, data: buildStats(rows) });
}

async function adminInspect(request, response, adminSession, store) {
  const targetUserId = String(pickParam(request, 'targetUserId', 'inspectUserId') || '').trim();
  const limit = safeLimit(pickParam(request, 'limit'), 100, 500);
  const where = targetUserId
    ? { projectKey: 'video-workflow', ownerId: targetUserId }
    : { projectKey: 'video-workflow' };
  const rows = await listAll(store, VIDEO_WORKFLOW_CLASSES.fileAsset, where, limit * 20);
  const users = {};
  for (const row of rows) {
    const userId = row.ownerId || '';
    if (!userId) continue;
    if (!users[userId]) {
      users[userId] = {
        userId,
        totalFiles: 0,
        totalBytes: 0,
        byKind: {},
        latestCreatedAt: '',
      };
    }
    addAssetStats(users[userId], row);
  }
  return response.json({
    code: 200,
    success: true,
    data: {
      adminUserId: adminSession.userId,
      targetUserId,
      users: Object.values(users)
        .sort((a, b) => b.totalBytes - a.totalBytes || b.totalFiles - a.totalFiles)
        .slice(0, limit),
      generatedAt: new Date().toISOString(),
    },
  });
}

async function bindAsset(request, response, userId, store) {
  const assetId = requiredText(request, 'assetId', 'id');
  const row = await store.findFirst(VIDEO_WORKFLOW_CLASSES.fileAsset, parseUserWhere(userId, { objectId: assetId }), { limit: 1 });
  if (!row?.objectId) return response.json({ code: 200, success: true, data: null });
  const bizType = String(pickParam(request, 'bizType') || '').trim();
  const bizId = String(pickParam(request, 'bizId') || '').trim();
  const sourceModule = String(pickParam(request, 'sourceModule') || '').trim();
  const patch = {};
  if (bizType) patch.bizType = bizType;
  if (bizId) patch.bizId = bizId;
  if (sourceModule) patch.sourceModule = sourceModule;
  const updated = Object.keys(patch).length
    ? await store.update(VIDEO_WORKFLOW_CLASSES.fileAsset, row.objectId, patch)
    : row;
  return response.json({ code: 200, success: true, data: rowToAsset(updated) });
}

async function deleteAsset(request, response, userId, store) {
  const assetId = requiredText(request, 'assetId', 'id');
  const row = await store.findFirst(VIDEO_WORKFLOW_CLASSES.fileAsset, parseUserWhere(userId, { objectId: assetId }), { limit: 1 });
  if (!row?.objectId) return response.json({ code: 200, success: true, data: null });
  const updated = await store.update(VIDEO_WORKFLOW_CLASSES.fileAsset, row.objectId, { status: 'deleted' });
  return response.json({ code: 200, success: true, data: rowToAsset(updated) });
}

async function purgeAsset(request, response, userId, store) {
  const assetId = requiredText(request, 'assetId', 'id');
  const reason = String(pickParam(request, 'reason') || '').trim();
  if (!reason) return response.json({ code: 400, success: false, error: '清理文件资产必须提供 reason' });
  const row = await store.findFirst(VIDEO_WORKFLOW_CLASSES.fileAsset, parseUserWhere(userId, { objectId: assetId }), { limit: 1 });
  if (!row?.objectId) return response.json({ code: 200, success: true, data: null });
  const updated = await store.update(VIDEO_WORKFLOW_CLASSES.fileAsset, row.objectId, {
    status: 'purged',
    metadata: { ...normalizeObject(row.metadata), purged: true, purgedAt: new Date().toISOString(), reason },
  });
  return response.json({ code: 200, success: true, data: rowToAsset(updated) });
}

function readAssetInput(request) {
  return {
    spaceId: String(pickParam(request, 'spaceId') || ''),
    url: String(pickParam(request, 'url') || ''),
    bucket: String(pickParam(request, 'bucket') || ''),
    mimeType: String(pickParam(request, 'mimeType', 'contentType') || ''),
    kind: normalizeKind(String(pickParam(request, 'kind') || '')),
    sizeBytes: Math.max(0, Number(pickParam(request, 'sizeBytes', 'size') || 0)),
    sha256: String(pickParam(request, 'sha256') || '').slice(0, 128),
    sourceModule: String(pickParam(request, 'sourceModule') || ''),
    bizType: String(pickParam(request, 'bizType') || ''),
    bizId: String(pickParam(request, 'bizId') || ''),
    status: String(pickParam(request, 'status') || 'active'),
    metadata: normalizeObject(pickParam(request, 'metadata') || {}),
  };
}

function rowToAsset(row) {
  return {
    assetId: row.objectId,
    id: row.objectId,
    userId: row.ownerId || '',
    spaceId: row.spaceId || '',
    qiniuKey: row.qiniuKey,
    url: row.url || '',
    bucket: row.bucket || '',
    mimeType: row.mimeType || '',
    kind: row.kind || '',
    sizeBytes: Number(row.sizeBytes || 0),
    sha256: row.sha256 || '',
    sourceModule: row.sourceModule || '',
    bizType: row.bizType || '',
    bizId: row.bizId || '',
    status: row.status || 'active',
    metadata: normalizeObject(row.metadata),
    createdAt: row.createdAt,
    updatedAt: row.updatedAt,
  };
}

async function listAll(store, className, where, maxRows) {
  const all = [];
  const pageSize = 1000;
  for (let skip = 0; skip < maxRows; skip += pageSize) {
    const rows = await store.list(className, where, { order: '-createdAt', limit: Math.min(pageSize, maxRows - skip), skip });
    all.push(...rows);
    if (rows.length < pageSize || all.length >= maxRows) break;
  }
  return all.slice(0, maxRows);
}

function buildStats(rows) {
  const byKind = {};
  let total = 0;
  let totalBytes = 0;
  for (const row of rows) {
    const aggregate = { totalFiles: 0, totalBytes: 0, byKind, latestCreatedAt: '' };
    addAssetStats(aggregate, row);
    total += 1;
    totalBytes += Number(row.sizeBytes || 0);
  }
  return { total, totalBytes, byKind, generatedAt: new Date().toISOString() };
}

function addAssetStats(target, row) {
  const kind = row.kind || 'unknown';
  const status = row.status || 'active';
  const bytes = Number(row.sizeBytes || 0);
  if (!target.byKind[kind]) target.byKind[kind] = { total: 0, bytes: 0, byStatus: {} };
  target.byKind[kind].total += 1;
  target.byKind[kind].bytes += bytes;
  target.byKind[kind].byStatus[status] = (target.byKind[kind].byStatus[status] || 0) + 1;
  target.totalFiles += 1;
  target.totalBytes += bytes;
  const latest = row.createdAt || '';
  if (latest && String(latest) > String(target.latestCreatedAt || '')) target.latestCreatedAt = latest;
}

function assertUserScopedKey(qiniuKey, userId) {
  const expected = `users/${userId}/`;
  if (!String(qiniuKey || '').startsWith(expected)) {
    const error = new Error('File key does not belong to current user partition');
    error.status = 403;
    throw error;
  }
}

function normalizeKind(kind) {
  if (['image', 'video', 'audio', 'export', 'snapshot'].includes(kind)) return kind;
  return 'image';
}

function requiredText(request, ...names) {
  const value = pickParam(request, ...names);
  const text = String(value || '').trim();
  if (!text) {
    const error = new Error(`Missing ${names[0]}`);
    error.status = 400;
    throw error;
  }
  return text;
}

function pickParam(request, ...names) {
  const sources = [request.params, request.body, request.query, request];
  for (const src of sources) {
    if (!src || typeof src !== 'object') continue;
    for (const name of names) {
      const value = src[name];
      if (value !== undefined && value !== null && value !== '') return value;
    }
  }
  return null;
}

function normalizeObject(value) {
  if (typeof value === 'string') {
    try { return JSON.parse(value); } catch { return {}; }
  }
  if (!value || typeof value !== 'object' || Array.isArray(value)) return {};
  return value;
}

function safeLimit(value, fallback, max) {
  const parsed = parseInt(value || fallback, 10);
  return Number.isFinite(parsed) ? Math.min(Math.max(parsed, 1), max) : fallback;
}