QIWEI-skill/mcp/src/core/webhook-store.js

const fs = require('fs');
const path = require('path');
const crypto = require('crypto');
const { categoryDir, createRunDir, writeRunManifest } = require('./output-paths');

const PROCESSED_EVENT_IDS_FILE = path.join(categoryDir('webhook'), 'processed-event-ids.json');
const MAX_PROCESSED_IDS = 5000;

function safeReadJson(filePath, fallback = null) {
  try {
    if (!filePath || !fs.existsSync(filePath)) return fallback;
    return JSON.parse(fs.readFileSync(filePath, 'utf8'));
  } catch {
    return fallback;
  }
}

function atomicWriteJson(filePath, data) {
  const dir = path.dirname(filePath);
  fs.mkdirSync(dir, { recursive: true });
  const tmpPath = `${filePath}.tmp.${Date.now()}.${Math.random().toString(36).slice(2, 8)}`;
  try {
    fs.writeFileSync(tmpPath, JSON.stringify(data, null, 2), 'utf8');
    fs.renameSync(tmpPath, filePath);
  } catch (err) {
    try { fs.unlinkSync(tmpPath); } catch {}
    throw err;
  }
  return filePath;
}

function sanitizePayload(payload) {
  if (!payload || typeof payload !== 'object') return payload;
  const output = {};
  for (const [key, value] of Object.entries(payload)) {
    if (/^(token|tokenId|secret|signature|privateKey|authorization)$/i.test(key)) {
      output[key] = '••••••••';
      continue;
    }
    if (value && typeof value === 'object') {
      output[key] = sanitizePayload(value);
    } else {
      output[key] = value;
    }
  }
  return output;
}

function readProcessedEventIds() {
  const data = safeReadJson(PROCESSED_EVENT_IDS_FILE, { ids: [], count: 0 });
  const ids = Array.isArray(data.ids) ? data.ids : [];
  return new Set(ids);
}

function writeProcessedEventIds(set) {
  let ids = Array.from(set);
  if (ids.length > MAX_PROCESSED_IDS) {
    ids = ids.slice(ids.length - MAX_PROCESSED_IDS);
  }
  atomicWriteJson(PROCESSED_EVENT_IDS_FILE, {
    ids,
    count: ids.length,
    updatedAt: new Date().toISOString()
  });
  return ids;
}

function eventScopeKey(eventId, account = {}) {
  const userId = String(account.userId || '').trim();
  return userId ? `${userId}:${eventId}` : String(eventId || '');
}

function markEventProcessed(eventId, account = {}) {
  const set = readProcessedEventIds();
  set.add(eventScopeKey(eventId, account));
  writeProcessedEventIds(set);
  return true;
}

function isDuplicateEvent(eventId, account = {}) {
  const set = readProcessedEventIds();
  return set.has(eventScopeKey(eventId, account));
}

function saveWebhookEventStructured(event, source = 'callback', rawBody = null) {
  const runDir = createRunDir('webhook', source);
  const eventId = event.eventId || `evt-${Date.now()}-${Math.random().toString(36).slice(2, 8)}`;
  const fileName = `event-${eventId}.json`;
  const filePath = path.join(runDir, fileName);

  const record = {
    receivedAt: new Date().toISOString(),
    source,
    eventId,
    parsedType: event.parsedType,
    cmd: event.cmd,
    msgType: event.msgType,
    guid: event.guid,
    externalUserId: event.externalUserId || undefined,
    status: 'PENDING',
    event: sanitizePayload(event.raw),
    rawBody: rawBody ? sanitizePayload(typeof rawBody === 'string' ? JSON.parse(rawBody) : rawBody) : undefined
  };

  atomicWriteJson(filePath, record);

  writeRunManifest(runDir, {
    tool: 'webhook-receiver',
    summary: { eventId, parsedType: event.parsedType, source },
    files: [fileName]
  });

  return filePath;
}

function saveRawEventFailed(body, source, reason, rawBody = null) {
  const runDir = createRunDir('webhook', `${source}-failed`);
  const hash = crypto.createHash('sha256').update(JSON.stringify(body || {})).digest('hex').slice(0, 16);
  const fileName = `event-failed-${hash}.json`;
  const filePath = path.join(runDir, fileName);

  const record = {
    receivedAt: new Date().toISOString(),
    source,
    reason,
    status: 'FAILED',
    body: sanitizePayload(body),
    rawBody: rawBody ? sanitizePayload(typeof rawBody === 'string' ? JSON.parse(rawBody) : rawBody) : undefined
  };

  atomicWriteJson(filePath, record);

  writeRunManifest(runDir, {
    tool: 'webhook-receiver',
    summary: { reason, source },
    files: [fileName]
  });

  return filePath;
}

function updateWebhookEventStatus(eventFilePath, status, result = null) {
  if (!eventFilePath || !fs.existsSync(eventFilePath)) return null;
  try {
    const record = safeReadJson(eventFilePath) || {};
    record.status = status;
    record.processedAt = new Date().toISOString();
    if (result !== null) record.result = result;
    atomicWriteJson(eventFilePath, record);
    return record;
  } catch (err) {
    console.warn('[WebhookStore] 更新事件状态失败:', err.message);
    return null;
  }
}

module.exports = {
  PROCESSED_EVENT_IDS_FILE,
  readProcessedEventIds,
  writeProcessedEventIds,
  markEventProcessed,
  isDuplicateEvent,
  saveWebhookEventStructured,
  saveRawEventFailed,
  updateWebhookEventStatus,
  sanitizePayload
};