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, }; }