memind/wechat-admin.mjs

import { loadWechatMpConfig } from './wechat-mp.mjs';
import { nextDailyRunAt, normalizeTimezone } from './schedule-time.mjs';

const DEFAULT_LIMIT = 50;
const MAX_LIMIT = 200;

function clampLimit(value) {
  const parsed = Number(value);
  if (!Number.isFinite(parsed) || parsed <= 0) return DEFAULT_LIMIT;
  return Math.min(Math.floor(parsed), MAX_LIMIT);
}

function normalizeStatus(value, allowed) {
  const status = typeof value === 'string' ? value.trim() : '';
  return allowed.includes(status) ? status : null;
}

function likePattern(value) {
  const text = typeof value === 'string' ? value.trim() : '';
  return text ? `%${text}%` : null;
}

export function maskWechatId(value) {
  const text = String(value ?? '');
  if (text.length <= 8) return text ? '***' : '';
  return `${text.slice(0, 6)}...${text.slice(-4)}`;
}

function maskAppId(value) {
  const text = String(value ?? '');
  if (text.length <= 10) return text || null;
  return `${text.slice(0, 6)}...${text.slice(-4)}`;
}

function boolFromEnv(value) {
  return value === true || value === '1' || value === 'true';
}

function mapBindingRow(row) {
  return {
    userId: row.user_id,
    username: row.username,
    displayName: row.display_name,
    status: row.status,
    appId: maskAppId(row.app_id),
    openidMasked: maskWechatId(row.openid),
    nickname: row.nickname,
    avatarUrl: row.avatar_url,
    lastLoginAt: Number(row.last_login_at ?? 0),
    boundAt: Number(row.bound_at ?? 0),
    routeId: row.route_id,
    routeStatus: row.route_status,
    agentSessionId: row.agent_session_id,
    routeUpdatedAt: row.route_updated_at ? Number(row.route_updated_at) : null,
  };
}

function mapMessageRow(row) {
  return {
    appId: maskAppId(row.app_id),
    openidMasked: maskWechatId(row.openid),
    msgId: row.msg_id,
    status: row.status,
    agentSessionId: row.agent_session_id,
    createdAt: Number(row.created_at ?? 0),
    updatedAt: Number(row.updated_at ?? 0),
    userId: row.user_id,
    username: row.username,
    displayName: row.display_name,
  };
}

function mapDigestRow(row) {
  return {
    id: row.id,
    userId: row.user_id,
    username: row.username,
    displayName: row.display_name,
    digestType: row.digest_type,
    hour: Number(row.hour),
    minute: Number(row.minute),
    timezone: row.timezone,
    channel: row.channel,
    status: row.status,
    nextRunAt: Number(row.next_run_at ?? 0),
    lastRunAt: row.last_run_at ? Number(row.last_run_at) : null,
    attempts: Number(row.attempts ?? 0),
    lastError: row.last_error,
    sourceText: row.source_text,
    createdAt: Number(row.created_at ?? 0),
    updatedAt: Number(row.updated_at ?? 0),
  };
}

function mapDeliveryRow(row) {
  return {
    id: row.id,
    reminderId: row.reminder_id,
    subscriptionId: row.subscription_id,
    userId: row.user_id,
    username: row.username,
    displayName: row.display_name,
    channel: row.channel,
    status: row.status,
    providerMessageId: row.provider_message_id,
    errorCode: row.error_code,
    errorMessage: row.error_message,
    createdAt: Number(row.created_at ?? 0),
    digestType: row.digest_type,
  };
}

function mapBalanceAlertRow(row) {
  return {
    id: row.id,
    userId: row.user_id,
    username: row.username,
    displayName: row.display_name,
    thresholdCents: Number(row.threshold_cents),
    channel: row.channel,
    status: row.status,
    nextRunAt: Number(row.next_run_at ?? 0),
    lastRunAt: row.last_run_at ? Number(row.last_run_at) : null,
    lastNotifiedBalanceCents:
      row.last_notified_balance_cents == null ? null : Number(row.last_notified_balance_cents),
    attempts: Number(row.attempts ?? 0),
    lastError: row.last_error ?? null,
    sourceText: row.source_text ?? null,
    createdAt: Number(row.created_at ?? 0),
    updatedAt: Number(row.updated_at ?? 0),
  };
}

function mapWebNotificationRow(row) {
  return {
    id: row.id,
    userId: row.user_id,
    username: row.username,
    displayName: row.display_name,
    channel: row.channel,
    notificationType: row.notification_type,
    title: row.title,
    body: row.body,
    status: row.status,
    readAt: row.read_at ? Number(row.read_at) : null,
    createdAt: Number(row.created_at ?? 0),
    updatedAt: Number(row.updated_at ?? 0),
  };
}

export function createWechatAdminService(
  pool,
  {
    config = loadWechatMpConfig(),
    scheduleEnabled = boolFromEnv(process.env.H5_SCHEDULE_ENABLED),
    reminderWorkerEnabled = boolFromEnv(process.env.H5_REMINDER_WORKER_ENABLED),
    sendWechatTextToUser = null,
  } = {},
) {
  async function getSummary() {
    const now = Date.now();
    const recentSince = now - 24 * 60 * 60 * 1000;
    const [[identityRows], [routeRows], [messageRows], [digestRows], [deliveryRows]] =
      await Promise.all([
        pool.query('SELECT COUNT(*) AS count FROM h5_user_wechat_identities'),
        pool.query(
          `SELECT
             COUNT(*) AS total,
             SUM(CASE WHEN status = 'active' THEN 1 ELSE 0 END) AS active
           FROM h5_wechat_agent_routes`,
        ),
        pool.query(
          `SELECT status, COUNT(*) AS count
           FROM h5_wechat_mp_messages
           WHERE updated_at >= ?
           GROUP BY status`,
          [recentSince],
        ),
        pool.query(
          `SELECT status, COUNT(*) AS count
           FROM h5_schedule_digest_subscriptions
           GROUP BY status`,
        ),
        pool.query(
          `SELECT status, COUNT(*) AS count
           FROM h5_schedule_delivery_logs
           WHERE created_at >= ?
           GROUP BY status`,
          [recentSince],
        ),
      ]);

    return {
      config: {
        mpEnabled: Boolean(config?.enabled),
        scheduleEnabled,
        reminderWorkerEnabled,
        appId: maskAppId(config?.appId),
        publicBaseUrl: config?.publicBaseUrl ?? null,
        bindPath: config?.bindPath ?? null,
        tokenEndpointConfigured: Boolean(config?.tokenUrl),
        customerServiceEndpointConfigured: Boolean(config?.customerServiceUrl),
      },
      counts: {
        boundUsers: Number(identityRows[0]?.count ?? 0),
        routes: {
          total: Number(routeRows[0]?.total ?? 0),
          active: Number(routeRows[0]?.active ?? 0),
        },
        recentMessages: Object.fromEntries(
          messageRows.map((row) => [row.status, Number(row.count ?? 0)]),
        ),
        digests: Object.fromEntries(digestRows.map((row) => [row.status, Number(row.count ?? 0)])),
        recentDeliveries: Object.fromEntries(
          deliveryRows.map((row) => [row.status, Number(row.count ?? 0)]),
        ),
      },
    };
  }

  async function listBindings({ search, limit } = {}) {
    const values = [];
    const where = [];
    const pattern = likePattern(search);
    if (pattern) {
      where.push(
        `(u.username LIKE ? OR u.display_name LIKE ? OR wi.nickname LIKE ? OR wi.openid LIKE ?)`,
      );
      values.push(pattern, pattern, pattern, pattern);
    }
    values.push(clampLimit(limit));
    const [rows] = await pool.query(
      `SELECT
         wi.user_id, wi.app_id, wi.openid, wi.nickname, wi.avatar_url,
         wi.last_login_at, wi.created_at AS bound_at,
         u.username, u.display_name, u.status,
         r.id AS route_id, r.status AS route_status, r.agent_session_id,
         r.updated_at AS route_updated_at
       FROM h5_user_wechat_identities wi
       JOIN h5_users u ON u.id = wi.user_id
       LEFT JOIN h5_wechat_agent_routes r ON r.app_id = wi.app_id AND r.openid = wi.openid
       ${where.length ? `WHERE ${where.join(' AND ')}` : ''}
       ORDER BY wi.updated_at DESC
       LIMIT ?`,
      values,
    );
    return { bindings: rows.map(mapBindingRow) };
  }

  async function listMessages({ status, limit } = {}) {
    const values = [];
    const where = [];
    const normalizedStatus = normalizeStatus(status, ['processing', 'done', 'failed']);
    if (normalizedStatus) {
      where.push('m.status = ?');
      values.push(normalizedStatus);
    }
    values.push(clampLimit(limit));
    const [rows] = await pool.query(
      `SELECT
         m.app_id, m.openid, m.msg_id, m.status, m.agent_session_id,
         m.created_at, m.updated_at,
         wi.user_id, u.username, u.display_name
       FROM h5_wechat_mp_messages m
       LEFT JOIN h5_user_wechat_identities wi ON wi.app_id = m.app_id AND wi.openid = m.openid
       LEFT JOIN h5_users u ON u.id = wi.user_id
       ${where.length ? `WHERE ${where.join(' AND ')}` : ''}
       ORDER BY m.updated_at DESC
       LIMIT ?`,
      values,
    );
    return { messages: rows.map(mapMessageRow) };
  }

  async function listDigests({ status, limit } = {}) {
    const values = [];
    const where = [];
    const normalizedStatus = normalizeStatus(status, ['active', 'locked', 'failed', 'cancelled']);
    if (normalizedStatus) {
      where.push('s.status = ?');
      values.push(normalizedStatus);
    }
    values.push(clampLimit(limit));
    const [rows] = await pool.query(
      `SELECT
         s.*, u.username, u.display_name
       FROM h5_schedule_digest_subscriptions s
       JOIN h5_users u ON u.id = s.user_id
       ${where.length ? `WHERE ${where.join(' AND ')}` : ''}
       ORDER BY s.updated_at DESC
       LIMIT ?`,
      values,
    );
    return { digests: rows.map(mapDigestRow) };
  }

  async function listDeliveries({ status, limit } = {}) {
    const values = [];
    const where = [];
    const normalizedStatus = normalizeStatus(status, ['success', 'failed']);
    if (normalizedStatus) {
      where.push('d.status = ?');
      values.push(normalizedStatus);
    }
    values.push(clampLimit(limit));
    const [rows] = await pool.query(
      `SELECT
         d.*, u.username, u.display_name, s.digest_type
       FROM h5_schedule_delivery_logs d
       JOIN h5_users u ON u.id = d.user_id
       LEFT JOIN h5_schedule_digest_subscriptions s ON s.id = d.subscription_id
       ${where.length ? `WHERE ${where.join(' AND ')}` : ''}
       ORDER BY d.created_at DESC
       LIMIT ?`,
      values,
    );
    return { deliveries: rows.map(mapDeliveryRow) };
  }

  async function listBalanceAlerts({ status, limit } = {}) {
    const values = [];
    const where = [];
    const normalizedStatus = normalizeStatus(status, ['active', 'locked', 'failed', 'cancelled']);
    if (normalizedStatus) {
      where.push('s.status = ?');
      values.push(normalizedStatus);
    }
    values.push(clampLimit(limit));
    const [rows] = await pool.query(
      `SELECT
         s.*, u.username, u.display_name
       FROM h5_balance_alert_subscriptions s
       JOIN h5_users u ON u.id = s.user_id
       ${where.length ? `WHERE ${where.join(' AND ')}` : ''}
       ORDER BY s.updated_at DESC
       LIMIT ?`,
      values,
    );
    return { balanceAlerts: rows.map(mapBalanceAlertRow) };
  }

  async function createBalanceAlert(input = {}) {
    const userId = typeof input.userId === 'string' ? input.userId : null;
    const thresholdCents = Number(input.thresholdCents);
    if (!userId) return { ok: false, message: '缺少用户' };
    if (!Number.isFinite(thresholdCents) || thresholdCents < 0) {
      return { ok: false, message: '余额阈值无效' };
    }
    const result = await pool.query(
      `INSERT INTO h5_balance_alert_subscriptions
       (id, user_id, threshold_cents, channel, status, next_run_at, last_run_at,
        last_notified_balance_cents, attempts, locked_until, last_error,
        source_channel, source_session_id, source_message_id, source_text, created_at, updated_at)
       VALUES (UUID(), ?, ?, 'wechat', 'active', ?, NULL, NULL, 0, NULL, NULL, 'api', NULL, NULL, ?, ?, ?)`,
      [userId, thresholdCents, Date.now(), input.sourceText ?? null, Date.now(), Date.now()],
    );
    return { ok: Number(result[0]?.affectedRows ?? 0) > 0 };
  }

  async function cancelBalanceAlert(id) {
    const now = Date.now();
    const [result] = await pool.query(
      `UPDATE h5_balance_alert_subscriptions
       SET status = 'cancelled', locked_until = NULL, updated_at = ?
       WHERE id = ? AND status <> 'cancelled'`,
      [now, id],
    );
    return { ok: Number(result.affectedRows ?? 0) > 0 };
  }

  async function resumeBalanceAlert(id) {
    const [rows] = await pool.query(
      `SELECT id, threshold_cents
       FROM h5_balance_alert_subscriptions
       WHERE id = ?
       LIMIT 1`,
      [id],
    );
    const alert = rows[0];
    if (!alert) return { ok: false, message: '订阅不存在' };
    await pool.query(
      `UPDATE h5_balance_alert_subscriptions
       SET status = 'active', next_run_at = ?, attempts = 0, locked_until = NULL,
           last_error = NULL, updated_at = ?
       WHERE id = ?`,
      [Date.now(), Date.now(), id],
    );
    return { ok: true, nextRunAt: Date.now() };
  }

  async function listWebNotifications({ status, limit } = {}) {
    const values = ['web'];
    const where = ['n.channel = ?'];
    const normalizedStatus = normalizeStatus(status, ['unread', 'read']);
    if (normalizedStatus) {
      where.push('n.status = ?');
      values.push(normalizedStatus);
    }
    values.push(clampLimit(limit));
    const [rows] = await pool.query(
      `SELECT
         n.*, u.username, u.display_name
       FROM h5_user_notifications n
       JOIN h5_users u ON u.id = n.user_id
       WHERE ${where.join(' AND ')}
       ORDER BY n.created_at DESC
       LIMIT ?`,
      values,
    );
    return { notifications: rows.map(mapWebNotificationRow) };
  }

  async function resolveNotificationTargets(input = {}) {
    const userId = typeof input.userId === 'string' ? input.userId.trim() : '';
    const userIds = Array.isArray(input.userIds)
      ? [...new Set(input.userIds.map((value) => String(value).trim()).filter(Boolean))]
      : [];
    if (userIds.length > 0) return userIds;
    if (userId) return [userId];
    const audience = typeof input.audience === 'string' ? input.audience.trim() : '';
    if (input.allUsers === true || audience === 'all') {
      const [rows] = await pool.query(
        `SELECT id
         FROM h5_users
         WHERE status <> 'disabled'
         ORDER BY created_at ASC`,
      );
      return rows.map((row) => row.id).filter(Boolean);
    }
    return [];
  }

  async function createWebNotification(input = {}) {
    const title = String(input.title ?? '').trim();
    const body = String(input.body ?? '').trim();
    const notificationType = String(input.notificationType ?? 'manual').trim() || 'manual';
    const channels = Array.isArray(input.channels)
      ? [...new Set(input.channels.map((value) => String(value).trim()).filter(Boolean))]
      : typeof input.channel === 'string' && input.channel.trim()
        ? [input.channel.trim()]
        : ['web'];
    const includeWeb = channels.includes('web');
    const includeWechat = channels.includes('wechat');
    const targets = await resolveNotificationTargets(input);
    if (targets.length === 0) return { ok: false, message: '缺少用户' };
    if (!title) return { ok: false, message: '缺少标题' };
    if (!body) return { ok: false, message: '缺少内容' };
    if (!includeWeb && !includeWechat) return { ok: false, message: '至少选择一种通知渠道' };
    if (includeWechat && typeof sendWechatTextToUser !== 'function' && !includeWeb) {
      return { ok: false, message: '公众号推送未启用' };
    }
    const now = Date.now();
    let created = 0;
    let wechatSent = 0;
    const wechatFailures = [];
    for (const targetUserId of targets) {
      if (includeWeb) {
        const [result] = await pool.query(
          `INSERT INTO h5_user_notifications
           (id, user_id, channel, notification_type, title, body, data_json, status, read_at, created_at, updated_at)
           VALUES (UUID(), ?, 'web', ?, ?, ?, NULL, 'unread', NULL, ?, ?)`,
          [targetUserId, notificationType, title, body, now, now],
        );
        created += Number(result?.affectedRows ?? 0);
      }
      if (includeWechat && typeof sendWechatTextToUser === 'function') {
        try {
          await sendWechatTextToUser(targetUserId, `${title}\n${body}`.trim());
          wechatSent += 1;
        } catch (err) {
          wechatFailures.push({
            userId: targetUserId,
            message: err instanceof Error ? err.message : String(err),
          });
        }
      }
    }
    return {
      ok: created > 0 || wechatSent > 0,
      created,
      wechatSent,
      wechatFailures,
      targets: targets.length,
    };
  }

  async function clearRouteForUser(userId) {
    const [rows] = await pool.query(
      `SELECT app_id, openid
       FROM h5_user_wechat_identities
       WHERE user_id = ?
       ORDER BY updated_at DESC
       LIMIT 1`,
      [userId],
    );
    const identity = rows[0];
    if (!identity) return { ok: false, message: '用户未绑定服务号' };
    const [result] = await pool.query(
      `DELETE FROM h5_wechat_agent_routes
       WHERE app_id = ? AND openid = ?`,
      [identity.app_id, identity.openid],
    );
    return {
      ok: true,
      deleted: Number(result.affectedRows ?? 0),
      openidMasked: maskWechatId(identity.openid),
    };
  }

  async function cancelDigest(id) {
    const now = Date.now();
    const [result] = await pool.query(
      `UPDATE h5_schedule_digest_subscriptions
       SET status = 'cancelled', locked_until = NULL, updated_at = ?
       WHERE id = ? AND status <> 'cancelled'`,
      [now, id],
    );
    return { ok: Number(result.affectedRows ?? 0) > 0 };
  }

  async function resumeDigest(id) {
    const [rows] = await pool.query(
      `SELECT id, hour, minute, timezone
       FROM h5_schedule_digest_subscriptions
       WHERE id = ?
       LIMIT 1`,
      [id],
    );
    const digest = rows[0];
    if (!digest) return { ok: false, message: '订阅不存在' };
    const timezone = normalizeTimezone(digest.timezone);
    const now = Date.now();
    const nextRun = nextDailyRunAt({
      hour: Number(digest.hour),
      minute: Number(digest.minute),
      timezone,
      now,
    });
    await pool.query(
      `UPDATE h5_schedule_digest_subscriptions
       SET status = 'active', next_run_at = ?, attempts = 0, locked_until = NULL,
           last_error = NULL, timezone = ?, updated_at = ?
       WHERE id = ?`,
      [nextRun, timezone, now, id],
    );
    return { ok: true, nextRunAt: nextRun };
  }

  return {
    getSummary,
    listBindings,
    listMessages,
    listDigests,
    listDeliveries,
    listBalanceAlerts,
    createBalanceAlert,
    cancelBalanceAlert,
    resumeBalanceAlert,
    listWebNotifications,
    createWebNotification,
    clearRouteForUser,
    cancelDigest,
    resumeDigest,
  };
}