fix(scheduled-task): deliver verified WeChat links only after page is ready
Memind CI / Test, build, and release guards (push) Successful in 8m42s

Wait for HTML materialization before releasing delivery contracts, pass
verified MindSpace URLs through proactive WeChat sends, resend on reconcile
when pages land late, and report deferred customer-service delivery as unsent.

Co-authored-by: Cursor <cursoragent@cursor.com>
This commit is contained in:
john
2026-08-17 22:07:41 +08:00
parent 0a7b2df5fe
commit 1ca06e58e7
10 changed files with 374 additions and 63 deletions
+1 -1
View File
@@ -168,7 +168,7 @@ export async function createAdminServices(env = {}) {
}); });
const notificationDispatcher = createNotificationDispatcher({ const notificationDispatcher = createNotificationDispatcher({
sendWechatTextToUser: wechatMpService?.enabled sendWechatTextToUser: wechatMpService?.enabled
? (userId, text) => wechatMpService.sendTextToUser(userId, text) ? (userId, text, options) => wechatMpService.sendTextToUser(userId, text, options)
: null, : null,
}); });
userAuth.setRechargeNotifier(async ({ userId, title, body, dedupeKey }) => { userAuth.setRechargeNotifier(async ({ userId, title, body, dedupeKey }) => {
+12
View File
@@ -1,4 +1,6 @@
import crypto from 'node:crypto'; import crypto from 'node:crypto';
import fs from 'node:fs';
import path from 'node:path';
export function normalizeDeliveryRelativePath(value) { export function normalizeDeliveryRelativePath(value) {
const path = String(value ?? '').replace(/\\/g, '/').replace(/^\/+/, ''); const path = String(value ?? '').replace(/\\/g, '/').replace(/^\/+/, '');
@@ -65,17 +67,27 @@ export async function markPageDeliveryContractFailed({
return Number(result?.affectedRows ?? 0) > 0; return Number(result?.affectedRows ?? 0) > 0;
} }
export function scheduledTaskPublicHtmlExists(publishDir, relativePath) {
const workspaceRelativePath = normalizeDeliveryRelativePath(relativePath);
if (!publishDir || !workspaceRelativePath) return false;
return fs.existsSync(path.join(publishDir, workspaceRelativePath));
}
export async function releaseMaterializedPageDeliveryContracts({ export async function releaseMaterializedPageDeliveryContracts({
pool, pool,
userId, userId,
relativePaths = [], relativePaths = [],
allowPgRequired = false, allowPgRequired = false,
publishDir = null,
} = {}) { } = {}) {
if (!pool || !userId) return []; if (!pool || !userId) return [];
const released = []; const released = [];
for (const rawPath of relativePaths) { for (const rawPath of relativePaths) {
const workspaceRelativePath = normalizeDeliveryRelativePath(rawPath); const workspaceRelativePath = normalizeDeliveryRelativePath(rawPath);
if (!workspaceRelativePath) continue; if (!workspaceRelativePath) continue;
if (publishDir && !scheduledTaskPublicHtmlExists(publishDir, workspaceRelativePath)) {
continue;
}
const contract = await getPageDeliveryContract({ const contract = await getPageDeliveryContract({
pool, pool,
userId, userId,
+13 -5
View File
@@ -1,8 +1,15 @@
function resolveWechatDispatchSent(result) {
if (result && typeof result === 'object') {
return result.sent !== false && !result.deferred && !result.skipped;
}
return result !== false;
}
export function createNotificationDispatcher({ sendWechatTextToUser, logger = console } = {}) { export function createNotificationDispatcher({ sendWechatTextToUser, logger = console } = {}) {
const sendWechat = async (userId, text) => { const sendWechat = async (userId, text, options = {}) => {
if (typeof sendWechatTextToUser !== 'function') return false; if (typeof sendWechatTextToUser !== 'function') return false;
await sendWechatTextToUser(userId, text); const result = await sendWechatTextToUser(userId, text, options);
return true; return resolveWechatDispatchSent(result);
}; };
return { return {
@@ -16,13 +23,14 @@ export function createNotificationDispatcher({ sendWechatTextToUser, logger = co
}); });
return sent; return sent;
}, },
async sendScheduleNotification({ userId, text }) { async sendScheduleNotification({ userId, text, verifiedHtmlUrls = [] }) {
const sent = await sendWechat(userId, text); const sent = await sendWechat(userId, text, { verifiedHtmlUrls });
logger.info?.('Notification dispatch:', { logger.info?.('Notification dispatch:', {
type: 'schedule_notification', type: 'schedule_notification',
userId, userId,
dedupeKey: null, dedupeKey: null,
sent, sent,
verifiedHtmlCount: verifiedHtmlUrls.length,
}); });
return sent; return sent;
}, },
+20
View File
@@ -73,11 +73,30 @@ test('notification dispatcher forwards schedule notification text unchanged', as
userId: 'user-2', userId: 'user-2',
dedupeKey: null, dedupeKey: null,
sent: true, sent: true,
verifiedHtmlCount: 0,
}, },
}, },
]); ]);
}); });
test('notification dispatcher returns false when wechat sender defers delivery', async () => {
const dispatcher = createNotificationDispatcher({
async sendWechatTextToUser() {
return { sent: false, deferred: true, errcode: 45015 };
},
logger: { info() {} },
});
assert.equal(
await dispatcher.sendScheduleNotification({
userId: 'user-4',
text: '定时任务完成',
verifiedHtmlUrls: ['https://m.tkmind.cn/MindSpace/user-4/public/news.html'],
}),
false,
);
});
test('notification dispatcher returns false when wechat sender is unavailable', async () => { test('notification dispatcher returns false when wechat sender is unavailable', async () => {
const logs = []; const logs = [];
const dispatcher = createNotificationDispatcher({ const dispatcher = createNotificationDispatcher({
@@ -120,6 +139,7 @@ test('notification dispatcher returns false when wechat sender is unavailable',
userId: 'user-3', userId: 'user-3',
dedupeKey: null, dedupeKey: null,
sent: false, sent: false,
verifiedHtmlCount: 0,
}, },
}, },
]); ]);
+168 -40
View File
@@ -78,6 +78,23 @@ export function formatScheduledTaskDeliveryMessage(task, deliveryText) {
return `${header}\n\n${body}`.trim(); return `${header}\n\n${body}`.trim();
} }
export function resolveScheduledTaskPublicBaseUrl(env = process.env) {
return String(env.H5_PUBLIC_BASE_URL ?? 'https://m.tkmind.cn').replace(/\/+$/, '');
}
export function buildScheduledTaskVerifiedHtmlUrls(userId, readyPaths = [], {
publicBaseUrl = resolveScheduledTaskPublicBaseUrl(),
} = {}) {
const normalizedUserId = String(userId ?? '').trim();
if (!normalizedUserId) return [];
return [...new Set(
(Array.isArray(readyPaths) ? readyPaths : [])
.map((relativePath) => normalizeDeliveryRelativePath(relativePath))
.filter(Boolean)
.map((relativePath) => `${publicBaseUrl}/MindSpace/${normalizedUserId}/${relativePath}`),
)];
}
const SCHEDULED_TASK_CLARIFICATION_PATTERNS = [ const SCHEDULED_TASK_CLARIFICATION_PATTERNS = [
/需确认/u, /需确认/u,
/请确认/u, /请确认/u,
@@ -90,30 +107,24 @@ const SCHEDULED_TASK_CLARIFICATION_PATTERNS = [
/缺(?:少|失)/u, /缺(?:少|失)/u,
]; ];
const DEFAULT_SCHEDULED_TASK_DELIVERY_RETRY_DELAYS_MS = [ const DEFAULT_SCHEDULED_TASK_DELIVERY_POLL_INTERVAL_MS = 5_000;
250,
1_000,
3_000,
5_000,
10_000,
30_000,
60_000,
];
export function resolveScheduledTaskDeliveryRetryDelaysMs( export function resolveScheduledTaskDeliveryPollIntervalMs(
env = process.env, env = process.env,
) { ) {
const raw = String( const parsed = Number(
env.H5_SCHEDULED_TASK_DELIVERY_RETRY_DELAYS_MS ?? '', env.H5_SCHEDULED_TASK_DELIVERY_POLL_INTERVAL_MS
).trim(); ?? DEFAULT_SCHEDULED_TASK_DELIVERY_POLL_INTERVAL_MS,
if (!raw) return [...DEFAULT_SCHEDULED_TASK_DELIVERY_RETRY_DELAYS_MS]; );
const parsed = raw return Number.isFinite(parsed) && parsed > 0
.split(',')
.map((value) => Number(value.trim()))
.filter((value) => Number.isFinite(value) && value >= 0);
return parsed.length > 0
? parsed ? parsed
: [...DEFAULT_SCHEDULED_TASK_DELIVERY_RETRY_DELAYS_MS]; : DEFAULT_SCHEDULED_TASK_DELIVERY_POLL_INTERVAL_MS;
}
/** @deprecated use resolveScheduledTaskDeliveryPollIntervalMs */
export function resolveScheduledTaskDeliveryRetryDelaysMs(env = process.env) {
const interval = resolveScheduledTaskDeliveryPollIntervalMs(env);
return [0, interval, interval, interval];
} }
export function extractPublicHtmlPathsFromText(text) { export function extractPublicHtmlPathsFromText(text) {
@@ -260,6 +271,7 @@ export async function finalizeScheduledTaskPageDelivery({
userId, userId,
relativePaths: [...pathsToRelease], relativePaths: [...pathsToRelease],
allowPgRequired: true, allowPgRequired: true,
publishDir,
}).catch((error) => { }).catch((error) => {
logger.warn?.('[ScheduledTask] release delivery contracts failed:', error); logger.warn?.('[ScheduledTask] release delivery contracts failed:', error);
return []; return [];
@@ -281,57 +293,72 @@ export async function awaitScheduledTaskPageDelivery({
messages, messages,
publishDir, publishDir,
deliveryText = '', deliveryText = '',
task = null,
tkmindProxy = null, tkmindProxy = null,
sessionSnapshotService = null, sessionSnapshotService = null,
retryDelaysMs = resolveScheduledTaskDeliveryRetryDelaysMs(), timeoutMs = 15 * 60 * 1000,
pollIntervalMs = resolveScheduledTaskDeliveryPollIntervalMs(),
sleepFn = (delayMs) => new Promise((resolve) => { sleepFn = (delayMs) => new Promise((resolve) => {
setTimeout(resolve, delayMs); setTimeout(resolve, delayMs);
}), }),
logger = console, logger = console,
} = {}) { } = {}) {
let currentMessages = Array.isArray(messages) ? messages : []; let currentMessages = Array.isArray(messages) ? messages : [];
let currentDeliveryText = String(deliveryText ?? '');
let readyPaths = []; let readyPaths = [];
const attempts = [0, ...retryDelaysMs]; const deadline = Date.now() + Math.max(Number(pollIntervalMs) || 0, Number(timeoutMs) || 0);
let attempt = 0;
for (let attempt = 0; attempt < attempts.length; attempt += 1) { while (true) {
if (attempt > 0) { attempt += 1;
await sleepFn(attempts[attempt]);
currentMessages = await refreshScheduledTaskMessages({
userId,
sessionId,
tkmindProxy,
sessionSnapshotService,
});
}
readyPaths = await finalizeScheduledTaskPageDelivery({ readyPaths = await finalizeScheduledTaskPageDelivery({
pool, pool,
userId, userId,
sessionId, sessionId,
messages: currentMessages, messages: currentMessages,
publishDir, publishDir,
deliveryText, deliveryText: currentDeliveryText,
logger, logger,
}).catch((error) => { }).catch((error) => {
logger.warn?.('[ScheduledTask] finalize page delivery failed:', error); logger.warn?.('[ScheduledTask] finalize page delivery failed:', error);
return []; return [];
}); });
const promisesHtml = deliveryTextPromisesPublicHtml(deliveryText) const promisesHtml = deliveryTextPromisesPublicHtml(currentDeliveryText)
|| collectScheduledTaskPageRelativePaths({ || collectScheduledTaskPageRelativePaths({
messages: currentMessages, messages: currentMessages,
publishDir, publishDir,
userId, userId,
deliveryText, deliveryText: currentDeliveryText,
}).relativePaths.length > 0; }).relativePaths.length > 0;
if (!promisesHtml || readyPaths.length > 0) { if (!promisesHtml || readyPaths.length > 0 || Date.now() >= deadline) {
if (promisesHtml && readyPaths.length === 0 && Date.now() >= deadline) {
logger.warn?.('[ScheduledTask] page delivery timed out while preparing', {
userId,
sessionId,
attempt,
timeoutMs,
});
}
break; break;
} }
logger.warn?.('[ScheduledTask] page delivery still preparing', { logger.warn?.('[ScheduledTask] page delivery still preparing', {
userId, userId,
sessionId, sessionId,
attempt: attempt + 1, attempt,
maxAttempts: attempts.length, nextPollMs: pollIntervalMs,
}); });
await sleepFn(pollIntervalMs);
currentMessages = await refreshScheduledTaskMessages({
userId,
sessionId,
tkmindProxy,
sessionSnapshotService,
});
if (task) {
currentDeliveryText = extractScheduledTaskDeliveryText(currentMessages, task);
}
} }
return { return {
@@ -390,7 +417,7 @@ export async function reconcileStuckStaticPageDeliveryContracts({
relativePath, relativePath,
}) })
) { ) {
released.push(relativePath); released.push({ userId: row.user_id, relativePath });
logger.info?.('[ScheduledTask] reconciled static delivery contract', { logger.info?.('[ScheduledTask] reconciled static delivery contract', {
userId: row.user_id, userId: row.user_id,
relativePath, relativePath,
@@ -400,6 +427,105 @@ export async function reconcileStuckStaticPageDeliveryContracts({
return released; return released;
} }
export async function resendScheduledTaskWechatForReadyPage({
pool,
userId,
relativePath,
notificationDispatcher = null,
publicBaseUrl = resolveScheduledTaskPublicBaseUrl(),
lookbackMs = 24 * 60 * 60 * 1000,
now = Date.now(),
logger = console,
} = {}) {
const normalizedPath = normalizeDeliveryRelativePath(relativePath);
const normalizedUserId = String(userId ?? '').trim();
if (
!pool
|| !normalizedUserId
|| !normalizedPath
|| typeof notificationDispatcher?.sendScheduleNotification !== 'function'
) {
return false;
}
const [rows] = await pool.query(
`SELECT id, title, notify_channel, last_result_json, last_run_at
FROM h5_scheduled_tasks
WHERE user_id = ? AND last_run_at IS NOT NULL AND last_run_at >= ?
ORDER BY last_run_at DESC
LIMIT 20`,
[normalizedUserId, now - Math.max(lookbackMs, 60_000)],
);
const verifiedUrl = `${publicBaseUrl}/MindSpace/${normalizedUserId}/${normalizedPath}`;
for (const row of rows ?? []) {
const channel = row.notify_channel ?? 'both';
if (channel !== 'wechat' && channel !== 'both') continue;
const lastResult = row.last_result_json && typeof row.last_result_json === 'object'
? row.last_result_json
: null;
const deliveryText = String(lastResult?.deliveryText ?? '');
if (
!deliveryText.includes(normalizedPath)
&& !deliveryText.includes(verifiedUrl)
) {
continue;
}
const priorDelivery = lastResult?.wechatDelivery;
if (
priorDelivery?.sentAt
&& Array.isArray(priorDelivery.relativePaths)
&& priorDelivery.relativePaths.includes(normalizedPath)
) {
continue;
}
const text = formatScheduledTaskDeliveryMessage(
{ title: row.title },
deliveryText,
);
const verifiedHtmlUrls = buildScheduledTaskVerifiedHtmlUrls(
normalizedUserId,
[normalizedPath],
{ publicBaseUrl },
);
const sent = await notificationDispatcher.sendScheduleNotification({
userId: normalizedUserId,
text,
verifiedHtmlUrls,
}).catch((error) => {
logger.warn?.('[ScheduledTask] reconcile wechat resend failed:', error);
return false;
});
if (!sent) continue;
const nextResult = {
...(lastResult ?? {}),
wechatDelivery: {
sentAt: now,
relativePaths: [normalizedPath],
source: 'reconcile',
},
};
await pool.query(
`UPDATE h5_scheduled_tasks
SET last_result_json = ?, updated_at = ?
WHERE id = ? AND user_id = ?`,
[JSON.stringify(nextResult), now, row.id, normalizedUserId],
);
logger.info?.('[ScheduledTask] reconciled wechat delivery resent', {
userId: normalizedUserId,
taskId: row.id,
relativePath: normalizedPath,
});
return true;
}
return false;
}
export async function executeScheduledTask(task, { export async function executeScheduledTask(task, {
userAuth, userAuth,
tkmindProxy, tkmindProxy,
@@ -470,8 +596,10 @@ export async function executeScheduledTask(task, {
messages, messages,
publishDir, publishDir,
deliveryText, deliveryText,
task,
tkmindProxy, tkmindProxy,
sessionSnapshotService, sessionSnapshotService,
timeoutMs,
logger, logger,
}) })
: { messages, readyPaths: [] }; : { messages, readyPaths: [] };
@@ -485,7 +613,7 @@ export async function executeScheduledTask(task, {
&& task.userId && task.userId
) { ) {
const links = readyPaths.map( const links = readyPaths.map(
(relativePath) => `https://m.tkmind.cn/MindSpace/${task.userId}/${relativePath}`, (relativePath) => `${resolveScheduledTaskPublicBaseUrl()}/MindSpace/${task.userId}/${relativePath}`,
); );
deliveryText = `${deliveryText}\n\n页面链接:\n${links.join('\n')}`.trim(); deliveryText = `${deliveryText}\n\n页面链接:\n${links.join('\n')}`.trim();
} }
+61 -7
View File
@@ -5,6 +5,7 @@ import path from 'node:path';
import test from 'node:test'; import test from 'node:test';
import { import {
buildScheduledTaskExecutionPrompt, buildScheduledTaskExecutionPrompt,
buildScheduledTaskVerifiedHtmlUrls,
deliveryTextPromisesPublicHtml, deliveryTextPromisesPublicHtml,
extractPublicHtmlPathsFromText, extractPublicHtmlPathsFromText,
extractScheduledTaskDeliveryText, extractScheduledTaskDeliveryText,
@@ -12,6 +13,7 @@ import {
formatScheduledTaskDeliveryMessage, formatScheduledTaskDeliveryMessage,
looksLikeScheduledTaskNonDelivery, looksLikeScheduledTaskNonDelivery,
reconcileStuckStaticPageDeliveryContracts, reconcileStuckStaticPageDeliveryContracts,
resolveScheduledTaskDeliveryPollIntervalMs,
resolveScheduledTaskDeliveryRetryDelaysMs, resolveScheduledTaskDeliveryRetryDelaysMs,
} from './scheduled-task-executor.mjs'; } from './scheduled-task-executor.mjs';
@@ -75,22 +77,45 @@ test('deliveryTextPromisesPublicHtml detects page delivery replies', () => {
); );
}); });
test('resolveScheduledTaskDeliveryRetryDelaysMs reads env override', () => { test('resolveScheduledTaskDeliveryPollIntervalMs reads env override', () => {
assert.equal(
resolveScheduledTaskDeliveryPollIntervalMs({
H5_SCHEDULED_TASK_DELIVERY_POLL_INTERVAL_MS: '7500',
}),
7500,
);
});
test('resolveScheduledTaskDeliveryRetryDelaysMs keeps backward-compatible shape', () => {
assert.deepEqual( assert.deepEqual(
resolveScheduledTaskDeliveryRetryDelaysMs({ resolveScheduledTaskDeliveryRetryDelaysMs({
H5_SCHEDULED_TASK_DELIVERY_RETRY_DELAYS_MS: '0,100,250', H5_SCHEDULED_TASK_DELIVERY_POLL_INTERVAL_MS: '100',
}), }),
[0, 100, 250], [0, 100, 100, 100],
);
});
test('buildScheduledTaskVerifiedHtmlUrls maps ready paths to public urls', () => {
assert.deepEqual(
buildScheduledTaskVerifiedHtmlUrls('user-1', ['public/news.html'], {
publicBaseUrl: 'https://m.tkmind.cn',
}),
['https://m.tkmind.cn/MindSpace/user-1/public/news.html'],
); );
}); });
test('finalizeScheduledTaskPageDelivery prepares and releases static contracts', async () => { test('finalizeScheduledTaskPageDelivery prepares and releases static contracts', async () => {
const dir = await fs.mkdtemp(path.join(os.tmpdir(), 'scheduled-task-finalize-'));
const publishDir = path.join(dir, 'MindSpace', 'user-1');
const relativePath = 'public/news.html';
await fs.mkdir(path.dirname(path.join(publishDir, relativePath)), { recursive: true });
await fs.writeFile(path.join(publishDir, relativePath), '<html></html>', 'utf8');
const calls = []; const calls = [];
const pool = { const pool = {
async query(sql, params) { async query(sql, params) {
calls.push({ sql, params }); calls.push({ sql, params });
if (sql.includes('FROM h5_page_delivery_contracts')) { if (sql.includes('FROM h5_page_delivery_contracts')) {
return [[{ workspace_relative_path: 'public/news.html' }]]; return [[{ workspace_relative_path: relativePath }]];
} }
if (sql.includes('SELECT id, data_mode, status')) { if (sql.includes('SELECT id, data_mode, status')) {
return [[{ id: 'c1', data_mode: 'static', status: 'preparing' }]]; return [[{ id: 'c1', data_mode: 'static', status: 'preparing' }]];
@@ -106,14 +131,43 @@ test('finalizeScheduledTaskPageDelivery prepares and releases static contracts',
userId: 'user-1', userId: 'user-1',
sessionId: 'session-1', sessionId: 'session-1',
messages: [], messages: [],
publishDir: '/tmp/publish', publishDir,
deliveryText: 'public/news.html 已生成', deliveryText: 'public/news.html 已生成',
logger: { warn() {}, info() {} }, logger: { warn() {}, info() {} },
}); });
assert.deepEqual(readyPaths, ['public/news.html']); assert.deepEqual(readyPaths, [relativePath]);
assert.ok(calls.some((call) => call.sql.includes('INSERT INTO h5_page_delivery_contracts'))); assert.ok(calls.some((call) => call.sql.includes('INSERT INTO h5_page_delivery_contracts')));
}); });
test('finalizeScheduledTaskPageDelivery skips release when html file is missing', async () => {
const publishDir = path.join(await fs.mkdtemp(path.join(os.tmpdir(), 'scheduled-task-finalize-missing-')), 'MindSpace', 'user-1');
await fs.mkdir(publishDir, { recursive: true });
const pool = {
async query(sql) {
if (sql.includes('FROM h5_page_delivery_contracts')) {
return [[{ workspace_relative_path: 'public/news.html' }]];
}
if (sql.includes('SELECT id, data_mode, status')) {
return [[{ id: 'c1', data_mode: 'static', status: 'preparing' }]];
}
if (sql.includes("SET status = 'ready'")) {
throw new Error('should not mark ready without html file');
}
return [[]];
},
};
const readyPaths = await finalizeScheduledTaskPageDelivery({
pool,
userId: 'user-1',
sessionId: 'session-1',
messages: [],
publishDir,
deliveryText: 'public/news.html 已生成',
logger: { warn() {}, info() {} },
});
assert.deepEqual(readyPaths, []);
});
test('reconcileStuckStaticPageDeliveryContracts releases materialized static pages', async () => { test('reconcileStuckStaticPageDeliveryContracts releases materialized static pages', async () => {
const dir = await fs.mkdtemp(path.join(os.tmpdir(), 'scheduled-task-')); const dir = await fs.mkdtemp(path.join(os.tmpdir(), 'scheduled-task-'));
const userId = 'user-1'; const userId = 'user-1';
@@ -137,7 +191,7 @@ test('reconcileStuckStaticPageDeliveryContracts releases materialized static pag
h5Root: dir, h5Root: dir,
logger: { info() {} }, logger: { info() {} },
}); });
assert.deepEqual(released, [relativePath]); assert.deepEqual(released, [{ userId, relativePath }]);
}); });
test('reconcileStuckStaticPageDeliveryContracts fails orphan contracts without html', async () => { test('reconcileStuckStaticPageDeliveryContracts fails orphan contracts without html', async () => {
+49 -6
View File
@@ -1,8 +1,11 @@
import { import {
buildScheduledTaskVerifiedHtmlUrls,
deliveryTextPromisesPublicHtml,
executeScheduledTask, executeScheduledTask,
formatScheduledTaskDeliveryMessage, formatScheduledTaskDeliveryMessage,
looksLikeScheduledTaskNonDelivery, looksLikeScheduledTaskNonDelivery,
reconcileStuckStaticPageDeliveryContracts, reconcileStuckStaticPageDeliveryContracts,
resendScheduledTaskWechatForReadyPage,
} from './scheduled-task-executor.mjs'; } from './scheduled-task-executor.mjs';
export function startScheduledTaskWorker({ export function startScheduledTaskWorker({
@@ -36,9 +39,12 @@ export function startScheduledTaskWorker({
let stopped = false; let stopped = false;
let running = false; let running = false;
const deliverTaskResult = async (task, deliveryText) => { const deliverTaskResult = async (task, deliveryText, { readyPaths = [] } = {}) => {
const text = formatScheduledTaskDeliveryMessage(task, deliveryText); const text = formatScheduledTaskDeliveryMessage(task, deliveryText);
const notifyChannel = task.notifyChannel ?? 'both'; const notifyChannel = task.notifyChannel ?? 'both';
const verifiedHtmlUrls = buildScheduledTaskVerifiedHtmlUrls(task.userId, readyPaths);
const promisesHtml = deliveryTextPromisesPublicHtml(deliveryText);
let wechatDelivery = null;
if (scheduleService?.createUserNotification && (notifyChannel === 'web' || notifyChannel === 'both')) { if (scheduleService?.createUserNotification && (notifyChannel === 'web' || notifyChannel === 'both')) {
await scheduleService.createUserNotification({ await scheduleService.createUserNotification({
userId: task.userId, userId: task.userId,
@@ -58,10 +64,30 @@ export function startScheduledTaskWorker({
(notifyChannel === 'wechat' || notifyChannel === 'both') (notifyChannel === 'wechat' || notifyChannel === 'both')
&& typeof sendScheduleNotification === 'function' && typeof sendScheduleNotification === 'function'
) { ) {
await sendScheduleNotification({ userId: task.userId, text }).catch((err) => { if (promisesHtml && verifiedHtmlUrls.length === 0) {
logger.warn?.('Scheduled task wechat notification failed:', err); logger.warn?.('[ScheduledTask] skip wechat until public html is ready', {
}); taskId: task.id,
userId: task.userId,
});
} else {
const sent = await sendScheduleNotification({
userId: task.userId,
text,
verifiedHtmlUrls,
}).catch((err) => {
logger.warn?.('Scheduled task wechat notification failed:', err);
return false;
});
if (sent && verifiedHtmlUrls.length > 0) {
wechatDelivery = {
sentAt: Date.now(),
relativePaths: readyPaths.map((value) => String(value ?? '').trim()).filter(Boolean),
source: 'scheduled_task_worker',
};
}
}
} }
return { wechatDelivery };
}; };
const runOnce = async () => { const runOnce = async () => {
@@ -69,13 +95,25 @@ export function startScheduledTaskWorker({
running = true; running = true;
try { try {
if (pool && h5Root) { if (pool && h5Root) {
await reconcileStuckStaticPageDeliveryContracts({ const reconciled = await reconcileStuckStaticPageDeliveryContracts({
pool, pool,
h5Root, h5Root,
logger, logger,
}).catch((err) => { }).catch((err) => {
logger.warn?.('Scheduled task delivery reconcile failed:', err); logger.warn?.('Scheduled task delivery reconcile failed:', err);
return [];
}); });
for (const item of reconciled) {
await resendScheduledTaskWechatForReadyPage({
pool,
userId: item.userId,
relativePath: item.relativePath,
notificationDispatcher,
logger,
}).catch((err) => {
logger.warn?.('Scheduled task reconcile wechat resend failed:', err);
});
}
} }
const dueTasks = await scheduledTaskService.listDueTasks({ limit: 10 }); const dueTasks = await scheduledTaskService.listDueTasks({ limit: 10 });
for (const candidate of dueTasks) { for (const candidate of dueTasks) {
@@ -99,10 +137,15 @@ export function startScheduledTaskWorker({
err.code = 'SCHEDULED_TASK_NON_DELIVERY'; err.code = 'SCHEDULED_TASK_NON_DELIVERY';
throw err; throw err;
} }
await deliverTaskResult(task, result.deliveryText); const deliveryMeta = await deliverTaskResult(task, result.deliveryText, {
readyPaths: result.readyPaths,
});
await scheduledTaskService.markTaskSucceeded(task, { await scheduledTaskService.markTaskSucceeded(task, {
result: { result: {
deliveryText: result.deliveryText, deliveryText: result.deliveryText,
...(deliveryMeta.wechatDelivery
? { wechatDelivery: deliveryMeta.wechatDelivery }
: {}),
}, },
deliveryText: result.deliveryText, deliveryText: result.deliveryText,
sessionId: result.sessionId, sessionId: result.sessionId,
@@ -236,10 +236,11 @@ export async function bootstrapPortalIntegrationServices({
const notificationDispatcher = const notificationDispatcher =
createNotificationDispatcherFn({ createNotificationDispatcherFn({
sendWechatTextToUser: wechatMpService?.enabled sendWechatTextToUser: wechatMpService?.enabled
? (userId, text) => ? (userId, text, options) =>
wechatMpService.sendTextToUser( wechatMpService.sendTextToUser(
userId, userId,
text, text,
options,
) )
: null, : null,
}); });
+14 -3
View File
@@ -2017,14 +2017,25 @@ export function createWechatMpService({
}); });
}; };
const sendTextToUser = async (userId, content) => { const sendTextToUser = async (userId, content, { verifiedHtmlUrls = [] } = {}) => {
const openid = await userAuth.getWechatOpenidForUser(userId, config.appId); const openid = await userAuth.getWechatOpenidForUser(userId, config.appId);
if (!openid) { if (!openid) {
throw new Error('用户尚未绑定服务号,无法推送提醒'); throw new Error('用户尚未绑定服务号,无法推送提醒');
} }
return sendCustomerServiceText(openid, content, null, { const normalizedVerified = verifiedHtmlUrls
.map((url) => String(url ?? '').trim())
.filter(Boolean);
const sendOptions = {
deliveryPriority: 'formal_reply', deliveryPriority: 'formal_reply',
}); verifiedHtmlUrls: normalizedVerified,
};
if (normalizedVerified.length > 0) {
sendOptions.linkExistsForRequest = createPreparedPublicHtmlLinkExists({
userId,
prepared: { validReplyUrls: normalizedVerified },
});
}
return sendCustomerServiceText(openid, content, null, sendOptions);
}; };
const enforceFreshPageThumbnailDelivery = async ({ const enforceFreshPageThumbnailDelivery = async ({
+34
View File
@@ -1333,6 +1333,40 @@ test('wechat mp service splits long agent replies into multiple customer message
assert.equal(combined, longReply); assert.equal(combined, longReply);
}); });
test('wechat mp sendTextToUser preserves verified MindSpace public links', async () => {
const sentBodies = [];
const service = createBoundWechatService({
userAuth: {
async getWechatOpenidForUser() {
return 'openid-1';
},
},
wechatFetch: async (url, init) => {
if (String(url).includes('/cgi-bin/stable_token')) {
return new Response(JSON.stringify({ access_token: 'access-1', expires_in: 7200 }), {
status: 200,
headers: { 'Content-Type': 'application/json' },
});
}
sentBodies.push(JSON.parse(String(init?.body ?? '{}')).text.content);
return new Response(JSON.stringify({ errcode: 0, errmsg: 'ok' }), {
status: 200,
headers: { 'Content-Type': 'application/json' },
});
},
});
const url = 'https://m.tkmind.cn/MindSpace/user-1/public/daily-news-0817.html';
const result = await service.sendTextToUser(
'user-1',
`页面已生成:${url}`,
{ verifiedHtmlUrls: [url] },
);
assert.equal(result.sent, true);
assert.match(sentBodies[0], /daily-news-0817\.html/);
assert.doesNotMatch(sentBodies[0], /页面生成未完成/);
});
test('wechat mp service defers 45047 customer-service delivery instead of throwing', async () => { test('wechat mp service defers 45047 customer-service delivery instead of throwing', async () => {
const service = createBoundWechatService({ const service = createBoundWechatService({
userAuth: { userAuth: {