openclaw-voc-skill/services/qiwei-relay/routes.ts

/**
 * Relay API 路由
 */

import { Router, Request, Response } from 'express';
import { getDb, Query, createObject, updateObject, type DbRow } from './lib/db.js';
import {
  generateApiCredentials,
  generateKeyPair,
  generateRelaySecret,
  hashApiSecret,
  encryptWithPublicKey,
  verifyHmacSignature,
  verifyBearerToken,
  sha256Hash,
} from './lib/crypto.js';
import {
  getAdminKey,
  getUpstreamCallbackSecret,
  getEventTtlSeconds,
  getMaxPendingEventsPerTenant,
  getPollBatchSize,
  getPollWaitMs,
  getRelayPublicUrl,
  getRegisterRateLimitSeconds,
  getRegisterRateLimitPerIp,
  getRegisterRateLimitPerUser,
} from './lib/config.js';
import { verifyFmodeToken, type FmodeUser } from './lib/fmode-auth.js';

// ============================================================
// 类型定义
// ============================================================

interface Tenant extends DbRow {
  apiKey: string;
  apiSecretHash: string;
  publicKey: string;
  fmodeUserId: string | null;
  name: string | null;
  description: string | null;
  plan: string;
  status: string;
  maxDevices: number;
  dailyEventLimit: number;
}

interface TenantDevice extends DbRow {
  tenantId: string;
  guid: string;
  relaySecret: string;
  deviceName: string | null;
  lastOnlineAt: string | null;
  status: string;
}

interface WebhookRelayEvent extends DbRow {
  tenantId: string;
  deviceGuid: string;
  eventId: string;
  status: string;
  encryptedPayload: string;
  payloadHash: string;
  retryCount: number;
  deliveredAt: string | null;
  expiresAt: string;
  error: string | null;
}

interface UsageMetricRow extends DbRow {
  metricValue: number;
}

interface AuditLogRow extends DbRow {
  action: string;
  tenantId: string | null;
  fmodeUserId: string | null;
  ip: string | null;
  details: string | null;
}

export const relayRoutes = Router();

// ============================================================
// 中间件
// ============================================================

interface AuthenticatedRequest extends Request {
  tenant?: Tenant;
}

/** 管理员鉴权 */
function adminAuth(req: Request, res: Response, next: () => void): void {
  const key = req.headers['x-admin-key'] as string | undefined;
  if (!key || !getAdminKey() || key !== getAdminKey()) {
    res.status(401).json({ error: 'Unauthorized' });
    return;
  }
  next();
}

/** 租户鉴权：Bearer apiSecret */
async function tenantAuth(req: AuthenticatedRequest, res: Response, next: () => void): Promise<void> {
  const auth = req.headers.authorization;
  if (!auth) {
    res.status(401).json({ error: 'Missing Authorization' });
    return;
  }

  const match = auth.match(/^Bearer\s+(.+)$/i);
  if (!match) {
    res.status(401).json({ error: 'Invalid Authorization format' });
    return;
  }

  const apiSecret = match[1];
  const tenant = await findTenantByApiSecret(apiSecret);
  if (!tenant) {
    res.status(401).json({ error: 'Invalid credentials' });
    return;
  }

  if (tenant.status !== 'active') {
    res.status(403).json({ error: 'Tenant inactive' });
    return;
  }

  req.tenant = tenant;
  next();
}

// ============================================================
// 限流（内存实现，单实例足够；如后续多实例可改 Redis）
// ============================================================

interface RateLimitBucket {
  count: number;
  resetAt: number;
}

const rateLimitMap = new Map<string, RateLimitBucket>();

function checkRateLimit(key: string, windowMs: number, maxCount: number): boolean {
  const now = Date.now();
  const bucket = rateLimitMap.get(key);
  if (!bucket || now >= bucket.resetAt) {
    rateLimitMap.set(key, { count: 1, resetAt: now + windowMs });
    return true;
  }
  if (bucket.count >= maxCount) {
    return false;
  }
  bucket.count += 1;
  return true;
}

function getClientIp(req: Request): string {
  const forwarded = req.headers['x-forwarded-for'];
  if (typeof forwarded === 'string' && forwarded) {
    return forwarded.split(',')[0].trim();
  }
  if (typeof forwarded === 'object' && forwarded.length > 0) {
    return forwarded[0].trim();
  }
  return req.socket.remoteAddress || 'unknown';
}

// ============================================================
// 管理员接口
// ============================================================

/** POST /api/admin/tenant — 创建租户 */
relayRoutes.post('/admin/tenant', adminAuth, (req: Request, res: Response) => {
  try {
    const { name, plan } = req.body;
    const { apiKey, apiSecret } = generateApiCredentials();
    const keyPair = generateKeyPair();

    const tenant = createObject('Tenant', {
      apiKey,
      apiSecretHash: hashApiSecret(apiSecret),
      publicKey: keyPair.publicKey,
      name: name || null,
      plan: plan || 'free',
      status: 'active',
      metadata: JSON.stringify({ privateKey: keyPair.privateKey }),
    });

    console.log(`[Relay] 创建租户: ${tenant.objectId}, apiKey=${apiKey}`);

    res.json({
      success: true,
      tenantId: tenant.objectId,
      apiKey,
      apiSecret,
      privateKey: keyPair.privateKey,
      publicKey: keyPair.publicKey,
      name: tenant.name,
      plan: tenant.plan,
    });
  } catch (err: unknown) {
    const msg = err instanceof Error ? err.message : String(err);
    console.error('[Relay] 创建租户失败:', msg);
    res.status(500).json({ error: `创建租户失败: ${msg}` });
  }
});

// ============================================================
// 租户自助注册接口
// ============================================================

/** POST /api/tenant/register — 使用 Fmode token 自助注册 */
relayRoutes.post('/tenant/register', async (req: Request, res: Response) => {
  try {
    const auth = req.headers.authorization || '';
    const token = auth.replace(/^Bearer\s+/i, '');
    if (!token) {
      res.status(401).json({
        success: false,
        error: '缺少 Fmode token',
        code: 'MISSING_TOKEN',
      });
      return;
    }

    const ip = getClientIp(req);
    const rateWindowMs = getRegisterRateLimitSeconds() * 1000;
    if (!checkRateLimit(`register:ip:${ip}`, rateWindowMs, getRegisterRateLimitPerIp())) {
      res.status(429).json({
        success: false,
        error: '该 IP 注册过于频繁，请稍后再试',
        code: 'RATE_LIMITED',
      });
      return;
    }

    const { description, deviceGuid } = req.body || {};
    const descriptionStr = typeof description === 'string' ? description.trim() : '';

    // 1. 校验 Fmode token
    const fmodeUser: FmodeUser = await verifyFmodeToken(token, deviceGuid);

    // 2. 风控：已注册则直接返回已有租户
    const existing = await findTenantByFmodeUserId(fmodeUser.userId);
    if (existing) {
      res.status(409).json({
        success: false,
        error: '该 Fmode 账号已开通 Relay',
        code: 'ALREADY_REGISTERED',
        tenantId: existing.objectId,
      });
      return;
    }

    // 3. 单个 Fmode 用户限流
    if (!checkRateLimit(`register:user:${fmodeUser.userId}`, rateWindowMs, getRegisterRateLimitPerUser())) {
      res.status(429).json({
        success: false,
        error: '该 Fmode 账号注册过于频繁',
        code: 'RATE_LIMITED',
      });
      return;
    }

    // 3. 生成租户凭证
    const { apiKey, apiSecret } = generateApiCredentials();
    const keyPair = generateKeyPair();

    const tenant = createObject('Tenant', {
      apiKey,
      apiSecretHash: hashApiSecret(apiSecret),
      publicKey: keyPair.publicKey,
      fmodeUserId: fmodeUser.userId,
      name: descriptionStr.slice(0, 100) || null,
      description: descriptionStr.slice(0, 255) || null,
      plan: 'free',
      status: 'active',
      maxDevices: 10,
      dailyEventLimit: 100000,
      metadata: JSON.stringify({
        registeredFrom: 'self-service',
        fmodeNickname: fmodeUser.nickname,
        fmodeMobile: fmodeUser.mobile,
      }),
    });

    // 4. 可选：预注册设备
    let registeredDevice: { guid: string; relaySecret: string } | null = null;
    if (deviceGuid && typeof deviceGuid === 'string') {
      const device = createObject('TenantDevice', {
        tenantId: tenant.objectId,
        guid: deviceGuid.trim(),
        relaySecret: generateRelaySecret(),
        deviceName: descriptionStr.slice(0, 100) || null,
        lastOnlineAt: new Date().toISOString(),
        status: 'active',
      }) as TenantDevice;
      registeredDevice = { guid: device.guid, relaySecret: device.relaySecret };
    }

    // 5. 审计日志
    createObject('AuditLog', {
      action: 'tenant.register',
      tenantId: tenant.objectId,
      fmodeUserId: fmodeUser.userId,
      ip,
      details: JSON.stringify({
        description: descriptionStr,
        deviceGuid: registeredDevice?.guid || null,
      }),
    });

    console.log(`[Relay] 自助注册租户: ${tenant.objectId}, fmodeUser=${fmodeUser.userId}, ip=${ip}`);

    res.json({
      success: true,
      relayBaseUrl: getRelayPublicUrl(),
      tenantId: tenant.objectId,
      apiKey,
      apiSecret,
      privateKey: keyPair.privateKey,
      publicKey: keyPair.publicKey,
      createdAt: tenant.createdAt,
      ...(registeredDevice
        ? { deviceGuid: registeredDevice.guid, relaySecret: registeredDevice.relaySecret }
        : {}),
    });
  } catch (err: unknown) {
    const msg = err instanceof Error ? err.message : String(err);
    console.error('[Relay] 租户自助注册失败:', msg);
    const code = /Fmode token 无效|Missing Fmode token/i.test(msg)
      ? 'INVALID_TOKEN'
      : 'REGISTER_FAILED';
    const status = code === 'INVALID_TOKEN' ? 401 : 500;
    res.status(status).json({ success: false, error: msg, code });
  }
});

// ============================================================
// 租户接口
// ============================================================

/** GET /api/tenant/status — 租户状态 */
relayRoutes.get('/tenant/status', tenantAuth, async (req: AuthenticatedRequest, res: Response) => {
  try {
    const tenantId = req.tenant!.objectId;

    const devices = await new Query<TenantDevice>('TenantDevice')
      .equalTo('tenantId', tenantId)
      .equalTo('status', 'active')
      .find();

    const pendingCount = await new Query('WebhookRelayEvent')
      .equalTo('tenantId', tenantId)
      .equalTo('status', 'pending')
      .count();

    const deliveredToday = await countMetric(tenantId, 'webhook_event', new Date().toISOString().slice(0, 10));

    res.json({
      success: true,
      tenantId,
      createdAt: req.tenant!.createdAt,
      name: req.tenant!.name,
      plan: req.tenant!.plan,
      status: req.tenant!.status,
      deviceCount: devices.length,
      devices: devices.map((d) => ({
        guid: d.guid,
        deviceName: d.deviceName,
        lastOnlineAt: d.lastOnlineAt,
      })),
      pendingEvents: pendingCount,
      pendingEventCount: pendingCount,
      webhookEventsToday: deliveredToday,
      eventCount24h: deliveredToday,
    });
  } catch (err: unknown) {
    const msg = err instanceof Error ? err.message : String(err);
    res.status(500).json({ success: false, error: msg });
  }
});

/** POST /api/tenant/device — 注册/更新设备 */
relayRoutes.post('/tenant/device', tenantAuth, async (req: AuthenticatedRequest, res: Response) => {
  try {
    const tenantId = req.tenant!.objectId;
    const { guid, deviceName } = req.body;

    if (!guid || typeof guid !== 'string') {
      res.status(400).json({ error: 'Missing guid' });
      return;
    }

    let device: TenantDevice | null = await findDevice(tenantId, guid);

    if (!device) {
      const fmodeToken = getHeader(req.headers as Record<string, string | string[] | undefined>, 'x-fmode-token');
      const tenant = req.tenant!;
      const assignedDevices = await findDevicesByGuid(guid);
      const foreignDevices = assignedDevices.filter((item) => item.tenantId !== tenantId);

      // 已绑定 Fmode 身份的租户注册新设备时必须重新证明账号身份。
      // 历史管理租户仍可注册未被占用的设备，以保持向后兼容。
      if ((tenant.fmodeUserId || foreignDevices.length > 0) && !fmodeToken) {
        res.status(401).json({ error: 'Fmode token is required to register or transfer this device' });
        return;
      }

      if (fmodeToken) {
        const fmodeUser = await verifyFmodeToken(fmodeToken, guid);
        if (tenant.fmodeUserId && tenant.fmodeUserId !== fmodeUser.userId) {
          res.status(403).json({ error: 'Fmode account does not match this tenant' });
          return;
        }

        for (const assigned of foreignDevices) {
          const owner = await findTenantById(assigned.tenantId);
          if (!owner?.fmodeUserId || owner.fmodeUserId !== fmodeUser.userId) {
            res.status(409).json({ error: 'Device is already assigned to another Fmode account' });
            return;
          }
        }

        if (!tenant.fmodeUserId) {
          updateObject('Tenant', tenantId, { fmodeUserId: fmodeUser.userId });
        }
        for (const assigned of foreignDevices) {
          updateObject('TenantDevice', assigned.objectId, { status: 'inactive' });
          console.log(`[Relay] 同账号设备迁移: guid=${guid}, from=${assigned.tenantId}, to=${tenantId}`);
        }
      }
    }

    const now = new Date().toISOString();
    if (!device) {
      device = createObject('TenantDevice', {
        tenantId,
        guid,
        relaySecret: generateRelaySecret(),
        deviceName: deviceName || null,
        lastOnlineAt: now,
        status: 'active',
      }) as TenantDevice;
      console.log(`[Relay] 注册设备: tenant=${tenantId}, guid=${guid}`);
    } else {
      updateObject('TenantDevice', device.objectId, {
        lastOnlineAt: now,
        deviceName: deviceName || device.deviceName,
        status: 'active',
      });
    }

    res.json({
      success: true,
      tenantId,
      guid,
      relaySecret: device.relaySecret,
      deviceName: device.deviceName,
      lastOnlineAt: device.lastOnlineAt,
    });
  } catch (err: unknown) {
    const msg = err instanceof Error ? err.message : String(err);
    res.status(500).json({ error: msg });
  }
});

// ============================================================
// Webhook 接收接口
// ============================================================

/**
 * The upstream callback is token-wide, so this endpoint routes each payload by
 * its real device guid before encrypting it for the owning tenant.
 */
relayRoutes.post('/webhook/ingest', async (req: Request, res: Response) => {
  const startTime = Date.now();
  const rawBody: string | undefined = (req as any).rawBody;
  const callbackSecret = getUpstreamCallbackSecret();

  try {
    if (!callbackSecret) {
      res.status(503).json({ error: 'Global callback is not configured' });
      return;
    }

    const signature = extractSignature(req.headers as Record<string, string | string[] | undefined>);
    if (!signature || !rawBody || !verifyHmacSignature(callbackSecret, rawBody, signature)) {
      const auth = req.headers.authorization as string | undefined;
      if (!auth || !verifyBearerToken(callbackSecret, auth)) {
        console.warn('[Relay] 全局回调签名验证失败');
        res.status(401).json({ error: 'Invalid signature' });
        return;
      }
    }

    if (isCallbackSetupProbe(req.body)) {
      res.json({ success: true, probe: true, queued: 0 });
      return;
    }

    const groups = splitGlobalPayload(req.body);
    if (groups.length === 0) {
      console.warn('[Relay] 全局回调缺少可路由的 guid');
      res.status(400).json({ error: 'Missing device guid in callback payload' });
      return;
    }

    const resolved: Array<{
      tenant: Tenant;
      device: TenantDevice;
      body: unknown;
      rawBody: string;
    }> = [];
    for (const group of groups) {
      const device = await findDeviceByGuid(group.guid);
      if (!device) {
        console.warn(`[Relay] 全局回调设备未注册: guid=${group.guid}`);
        res.status(409).json({ error: 'Device is not registered' });
        return;
      }
      const tenant = await findTenantById(device.tenantId);
      if (!tenant || tenant.status !== 'active') {
        res.status(409).json({ error: 'Tenant is not active' });
        return;
      }
      resolved.push({ tenant, device, body: group.body, rawBody: group.rawBody });
    }

    let queued = 0;
    let duplicates = 0;
    for (const item of resolved) {
      const result = await storeWebhookEvent(
        item.tenant,
        item.device,
        item.rawBody,
        item.body,
      );
      if (result.duplicate) duplicates += 1;
      else queued += 1;
    }

    console.log(
      `[Relay] 全局回调完成: devices=${resolved.length}, queued=${queued}, duplicates=${duplicates}, elapsed=${Date.now() - startTime}ms`,
    );
    res.status(200).json({ code: 0, msg: 'received', queued, duplicates });
  } catch (err: unknown) {
    const msg = err instanceof Error ? err.message : String(err);
    console.error('[Relay] 全局回调接收失败:', msg);
    res.status(500).json({ error: 'Global ingest failed' });
  }
});

/** POST /api/webhook/ingest/:tenantId/:deviceGuid */
relayRoutes.post('/webhook/ingest/:tenantId/:deviceGuid', async (req: Request, res: Response) => {
  const startTime = Date.now();
  const { tenantId, deviceGuid } = req.params;
  const rawBody: string | undefined = (req as any).rawBody;

  try {
    // 1. 查找设备与租户
    const device = await findDevice(tenantId, deviceGuid);

    if (!device) {
      res.status(404).json({ error: 'Device not found' });
      return;
    }

    const tenant = await findTenantById(tenantId);
    if (!tenant || tenant.status !== 'active') {
      res.status(404).json({ error: 'Tenant not found or inactive' });
      return;
    }

    // 2. 验证 HMAC 签名（来自 qiweapi 或本地测试）
    const signature = extractSignature(req.headers as Record<string, string | string[] | undefined>);
    const relaySecret = device.relaySecret as string;

    if (!signature || !rawBody || !verifyHmacSignature(relaySecret, rawBody, signature)) {
      // 兼容直接 Bearer secret 的测试方式
      const auth = req.headers.authorization as string | undefined;
      if (!auth || !verifyBearerToken(relaySecret, auth)) {
        console.warn(`[Relay] 签名验证失败: tenant=${tenantId}, guid=${deviceGuid}`);
        res.status(401).json({ error: 'Invalid signature' });
        return;
      }
    }

    if (!rawBody) {
      res.status(400).json({ error: 'Missing raw body' });
      return;
    }
    const result = await storeWebhookEvent(tenant, device, rawBody, req.body);

    const elapsed = Date.now() - startTime;
    console.log(`[Relay] 接收事件: tenant=${tenantId}, guid=${deviceGuid}, eventId=${result.eventId}, elapsed=${elapsed}ms`);

    res.status(200).json({
      code: 0,
      msg: result.duplicate ? 'duplicate event' : 'received',
      eventId: result.eventId,
    });
  } catch (err: unknown) {
    const msg = err instanceof Error ? err.message : String(err);
    console.error('[Relay] 接收事件失败:', msg);
    res.status(500).json({ error: `Ingest failed: ${msg}` });
  }
});

// ============================================================
// Relay 拉取接口
// ============================================================

/** POST /api/poll 与 /api/relay/poll（别名） */
async function handlePoll(req: AuthenticatedRequest, res: Response): Promise<void> {
  const tenantId = req.tenant!.objectId;
  const { guid, batchSize = getPollBatchSize(), waitMs = getPollWaitMs() } = req.body;

  if (!guid || typeof guid !== 'string') {
    res.status(400).json({ error: 'Missing guid' });
    return;
  }

  // 校验 guid 属于当前租户
  const device = await findDevice(tenantId, guid);

  if (!device) {
    res.status(404).json({ error: 'Device not registered for this tenant' });
    return;
  }

  // 更新最后在线时间
  updateObject('TenantDevice', device.objectId, {
    lastOnlineAt: new Date().toISOString(),
  });

  // 长轮询：最多等待 waitMs，有新事件或超时返回
  const deadline = Date.now() + Math.min(waitMs, getPollWaitMs());
  const pollInterval = 500;

  while (Date.now() < deadline) {
    const events = await fetchPendingEvents(tenantId, guid, Math.min(batchSize, getPollBatchSize()));
    if (events.length > 0) {
      res.json({ success: true, events });
      return;
    }
    await sleep(pollInterval);
  }

  // 超时返回空
  res.json({ success: true, events: [] });
}

relayRoutes.post('/poll', tenantAuth, handlePoll);
relayRoutes.post('/relay/poll', tenantAuth, handlePoll);

/** POST /api/ack 与 /api/relay/ack（别名） */
async function handleAck(req: AuthenticatedRequest, res: Response): Promise<void> {
  const tenantId = req.tenant!.objectId;
  const { eventIds, guid } = req.body;

  if (!Array.isArray(eventIds) || eventIds.length === 0) {
    res.status(400).json({ error: 'eventIds must be a non-empty array' });
    return;
  }

  const db = getDb();
  const stmt = db.prepare(
    `UPDATE "WebhookRelayEvent" SET status = 'delivered', deliveredAt = ? WHERE tenantId = ? AND eventId IN (${eventIds.map(() => '?').join(',')})`
  );
  stmt.run(new Date().toISOString(), tenantId, ...eventIds);

  // 可选：校验 guid 归属
  console.log(`[Relay] ACK: tenant=${tenantId}, guid=${guid}, count=${eventIds.length}`);

  res.json({ success: true, ackedCount: eventIds.length });
}

relayRoutes.post('/ack', tenantAuth, handleAck);
relayRoutes.post('/relay/ack', tenantAuth, handleAck);

// ============================================================
// 健康检查
// ============================================================

relayRoutes.get('/health', (_req: Request, res: Response) => {
  res.json({ ok: true, timestamp: new Date().toISOString() });
});

// ============================================================
// 辅助函数
// ============================================================

async function findTenantByApiSecret(apiSecret: string): Promise<Tenant | null> {
  return new Query<Tenant>('Tenant').equalTo('apiSecretHash', hashApiSecret(apiSecret)).first();
}

async function findTenantById(tenantId: string): Promise<Tenant | null> {
  return new Query<Tenant>('Tenant').equalTo('objectId', tenantId).first();
}

async function findTenantByFmodeUserId(fmodeUserId: string): Promise<Tenant | null> {
  return new Query<Tenant>('Tenant').equalTo('fmodeUserId', fmodeUserId).first();
}

async function findDevice(tenantId: string, guid: string): Promise<TenantDevice | null> {
  return new Query<TenantDevice>('TenantDevice')
    .equalTo('tenantId', tenantId)
    .equalTo('guid', guid)
    .equalTo('status', 'active')
    .first();
}

async function findDeviceByGuid(guid: string): Promise<TenantDevice | null> {
  const devices = await findDevicesByGuid(guid);
  if (devices.length > 1) {
    const tenants = await Promise.all(devices.map((device) => findTenantById(device.tenantId)));
    const owners = new Set(tenants.filter(Boolean).map((tenant) => tenant!.fmodeUserId).filter(Boolean));
    if (tenants.some((tenant) => !tenant) || owners.size !== 1) {
      throw new Error(`Device guid is assigned to multiple tenants: ${guid}`);
    }
    devices.sort((left, right) => String(right.lastOnlineAt || right.updatedAt || '').localeCompare(String(left.lastOnlineAt || left.updatedAt || '')));
    console.warn(`[Relay] 收敛同一 Fmode 用户的重复设备路由: guid=${guid}, selectedTenant=${devices[0].tenantId}`);
  }
  return devices[0] || null;
}

async function findDevicesByGuid(guid: string): Promise<TenantDevice[]> {
  return new Query<TenantDevice>('TenantDevice')
    .equalTo('guid', guid)
    .equalTo('status', 'active')
    .find();
}

function extractGuid(value: unknown): string {
  if (!value || typeof value !== 'object' || Array.isArray(value)) return '';
  const item = value as Record<string, unknown>;
  const direct = item.guid || item.deviceGuid || item.device_guid;
  if (typeof direct === 'string' && /^[A-Za-z0-9._:-]{1,128}$/.test(direct.trim())) {
    return direct.trim();
  }
  if (item.meta && typeof item.meta === 'object') return extractGuid(item.meta);
  return '';
}

function isCallbackSetupProbe(body: unknown): boolean {
  if (!body || typeof body !== 'object' || Array.isArray(body)) return false;
  const value = body as Record<string, unknown>;
  return typeof value.callBackUrl === 'string'
    && typeof value.msg === 'string'
    && value.msg.includes('设置订阅成功');
}

function splitGlobalPayload(body: unknown): Array<{ guid: string; body: unknown; rawBody: string }> {
  if (Array.isArray(body)) {
    const groups = new Map<string, unknown[]>();
    for (const item of body) {
      const guid = extractGuid(item);
      if (!guid) return [];
      const values = groups.get(guid) || [];
      values.push(item);
      groups.set(guid, values);
    }
    return [...groups.entries()].map(([guid, values]) => ({
      guid,
      body: values,
      rawBody: JSON.stringify(values),
    }));
  }

  if (!body || typeof body !== 'object') return [];
  const wrapper = body as Record<string, unknown>;
  const wrapperGuid = extractGuid(wrapper);
  if (!Array.isArray(wrapper.data)) {
    return wrapperGuid ? [{ guid: wrapperGuid, body, rawBody: JSON.stringify(body) }] : [];
  }

  const groups = new Map<string, unknown[]>();
  for (const item of wrapper.data) {
    const guid = extractGuid(item) || wrapperGuid;
    if (!guid) return [];
    const values = groups.get(guid) || [];
    values.push(item);
    groups.set(guid, values);
  }
  return [...groups.entries()].map(([guid, values]) => {
    const groupedBody = { ...wrapper, data: values };
    return { guid, body: groupedBody, rawBody: JSON.stringify(groupedBody) };
  });
}

function extractSignature(headers: Record<string, string | string[] | undefined>): string | null {
  const auth = getHeader(headers, 'authorization');
  if (auth) {
    const match = auth.match(/^Bearer\s+(.+)$/i);
    if (match) return match[1].trim();
    return auth.trim();
  }
  const sigHeaders = ['x-qiwe-signature', 'x-qiweapi-signature', 'x-signature'];
  for (const name of sigHeaders) {
    const val = getHeader(headers, name);
    if (val) return val;
  }
  return null;
}

function getHeader(headers: Record<string, string | string[] | undefined>, name: string): string | undefined {
  const lower = name.toLowerCase();
  for (const [key, value] of Object.entries(headers)) {
    if (key.toLowerCase() === lower) {
      return Array.isArray(value) ? value[0] : value;
    }
  }
  return undefined;
}

function extractEventId(body: unknown, guid: string): string {
  if (!body || typeof body !== 'object') return `${guid}_${Date.now()}_${Math.random().toString(36).slice(2)}`;
  const b = body as Record<string, unknown>;

  // v1: data[].msgUniqueIdentifier
  if (Array.isArray(b.data)) {
    const items = b.data as Array<Record<string, unknown>>;
    const first = items[0];
    if (first?.msgUniqueIdentifier) return `${guid}_${first.msgUniqueIdentifier}`;
  }

  // v2: meta.seq / meta.token_id
  if (b.version === '2.0' && b.meta && typeof b.meta === 'object') {
    const meta = b.meta as Record<string, unknown>;
    return `${guid}_${meta.token_id || ''}_${meta.seq || Date.now()}`;
  }

  return `${guid}_${Date.now()}_${Math.random().toString(36).slice(2)}`;
}

async function storeWebhookEvent(
  tenant: Tenant,
  device: TenantDevice,
  rawBody: string,
  body: unknown,
): Promise<{ eventId: string; duplicate: boolean }> {
  const eventId = extractEventId(body, device.guid);
  const existing = await new Query('WebhookRelayEvent')
    .equalTo('tenantId', tenant.objectId)
    .equalTo('eventId', eventId)
    .first();
  if (existing) return { eventId, duplicate: true };

  const encryptedPayload = encryptWithPublicKey(tenant.publicKey, rawBody);
  const payloadHash = sha256Hash(rawBody);
  await enforcePendingEventLimit(tenant.objectId);
  const expiresAt = new Date(Date.now() + getEventTtlSeconds() * 1000).toISOString();
  createObject('WebhookRelayEvent', {
    tenantId: tenant.objectId,
    deviceGuid: device.guid,
    eventId,
    status: 'pending',
    encryptedPayload,
    payloadHash,
    retryCount: 0,
    expiresAt,
  });
  recordMetric(tenant.objectId, device.guid, 'webhook_event');
  return { eventId, duplicate: false };
}

async function enforcePendingEventLimit(tenantId: string): Promise<void> {
  const max = getMaxPendingEventsPerTenant();
  const db = getDb();

  const countRow = db
    .prepare('SELECT COUNT(*) as count FROM "WebhookRelayEvent" WHERE tenantId = ? AND status = ?')
    .get(tenantId, 'pending') as { count: number };

  if (countRow.count >= max) {
    // 删除最旧的 pending 事件
    const toDrop = countRow.count - max + 1;
    db.prepare(
      `DELETE FROM "WebhookRelayEvent" WHERE objectId IN (
        SELECT objectId FROM "WebhookRelayEvent"
        WHERE tenantId = ? AND status = ?
        ORDER BY createdAt ASC
        LIMIT ?
      )`
    ).run(tenantId, 'pending', toDrop);
    console.warn(`[Relay] 租户 ${tenantId} pending 事件超过上限，已丢弃 ${toDrop} 条最旧事件`);
  }
}

async function fetchPendingEvents(tenantId: string, guid: string, limit: number): Promise<WebhookRelayEvent[]> {
  return new Query<WebhookRelayEvent>('WebhookRelayEvent')
    .equalTo('tenantId', tenantId)
    .equalTo('deviceGuid', guid)
    .equalTo('status', 'pending')
    .ascending('createdAt')
    .limit(limit)
    .find();
}

function recordMetric(tenantId: string, deviceGuid: string, metricName: string, value = 1): void {
  try {
    const metricDate = new Date().toISOString().slice(0, 10);
    createObject('UsageMetric', {
      tenantId,
      deviceGuid,
      metricName,
      metricValue: value,
      metricDate,
    });
  } catch (err: unknown) {
    console.warn('[Relay] 用量埋点失败:', err instanceof Error ? err.message : String(err));
  }
}

async function countMetric(tenantId: string, metricName: string, metricDate: string): Promise<number> {
  try {
    const rows = await new Query<UsageMetricRow>('UsageMetric')
      .equalTo('tenantId', tenantId)
      .equalTo('metricName', metricName)
      .equalTo('metricDate', metricDate)
      .find();
    return rows.reduce((sum, row) => sum + (row.metricValue || 0), 0);
  } catch {
    return 0;
  }
}

function sleep(ms: number): Promise<void> {
  return new Promise((resolve) => setTimeout(resolve, ms));
}