feat(wechat): add Intent Transaction Layer with unified task schema
Introduce Draft → Confirm → Commit flow for WeChat schedule intents behind feature flags, plus h5_tasks dual-write/read aggregation and rollout scripts so reminders and automations get explicit user confirmation before persisting. Co-authored-by: Cursor <cursoragent@cursor.com>
This commit is contained in:
john
@@ -0,0 +1,622 @@
|
||||
import crypto from 'node:crypto';
|
||||
|
||||
const DEFAULT_TIMEZONE = 'Asia/Shanghai';
|
||||
|
||||
function parseJson(value) {
|
||||
if (value == null || value === '') return null;
|
||||
if (typeof value === 'object') return value;
|
||||
try { return JSON.parse(value); } catch { return null; }
|
||||
}
|
||||
|
||||
function rowToUnifiedTask(row) {
|
||||
if (!row) return null;
|
||||
return {
|
||||
id: row.id,
|
||||
userId: row.user_id,
|
||||
type: row.type,
|
||||
title: row.title,
|
||||
spec: parseJson(row.spec_json) ?? {},
|
||||
trigger: parseJson(row.trigger_json) ?? {},
|
||||
action: parseJson(row.action_json) ?? {},
|
||||
actionLevel: Number(row.action_level ?? 1),
|
||||
notifyChannel: row.notify_channel ?? 'both',
|
||||
status: row.status,
|
||||
nextRunAt: row.next_run_at == null ? null : Number(row.next_run_at),
|
||||
lastRunAt: row.last_run_at == null ? null : Number(row.last_run_at),
|
||||
legacyRef: parseJson(row.legacy_ref_json),
|
||||
sourceChannel: row.source_channel ?? null,
|
||||
sourceMessageId: row.source_message_id ?? null,
|
||||
sourceText: row.source_text ?? null,
|
||||
createdAt: Number(row.created_at),
|
||||
updatedAt: Number(row.updated_at),
|
||||
};
|
||||
}
|
||||
|
||||
function legacyKey(ref) {
|
||||
if (!ref?.table || !ref?.id) return null;
|
||||
return `${ref.table}:${ref.id}`;
|
||||
}
|
||||
|
||||
function formatTriggerLabel(task, timezone = DEFAULT_TIMEZONE) {
|
||||
const trigger = task.trigger ?? {};
|
||||
if (trigger.at) return trigger.at;
|
||||
if (trigger.runAtLocal) return trigger.runAtLocal;
|
||||
if (trigger.hour != null) {
|
||||
return `${String(trigger.hour).padStart(2, '0')}:${String(trigger.minute ?? 0).padStart(2, '0')}`;
|
||||
}
|
||||
if (task.nextRunAt) {
|
||||
return new Date(task.nextRunAt).toLocaleString('zh-CN', {
|
||||
timeZone: trigger.timezone || timezone,
|
||||
hour12: false,
|
||||
});
|
||||
}
|
||||
return '待确认';
|
||||
}
|
||||
|
||||
export function formatUnifiedTaskListReply(tasks = [], { timezone = DEFAULT_TIMEZONE } = {}) {
|
||||
if (!tasks.length) return '你当前没有进行中的提醒或定时任务。';
|
||||
const lines = ['你的任务一览：'];
|
||||
for (const task of tasks.slice(0, 10)) {
|
||||
const when = formatTriggerLabel(task, timezone);
|
||||
const repeat = task.trigger?.repeat ? ` · ${task.trigger.repeat}` : '';
|
||||
const typeLabel = {
|
||||
reminder: '提醒',
|
||||
todo: '待办',
|
||||
digest: '待办摘要',
|
||||
automation: '自动任务',
|
||||
condition: '条件提醒',
|
||||
}[task.type] ?? task.type;
|
||||
lines.push(`- [${typeLabel}] ${task.title}（${when}${repeat}）`);
|
||||
}
|
||||
return lines.join('\n');
|
||||
}
|
||||
|
||||
function mapCommitToUnifiedTask({ userId, kind, committed, timezone = DEFAULT_TIMEZONE }) {
|
||||
const now = Date.now();
|
||||
const base = {
|
||||
userId,
|
||||
actionLevel: kind === 'scheduled_task' ? 2 : 1,
|
||||
notifyChannel: 'wechat',
|
||||
status: 'active',
|
||||
sourceChannel: committed?.item?.sourceChannel
|
||||
?? committed?.task?.sourceChannel
|
||||
?? committed?.subscription?.sourceChannel
|
||||
?? 'wechat',
|
||||
sourceMessageId: committed?.item?.sourceMessageId
|
||||
?? committed?.task?.sourceMessageId
|
||||
?? committed?.subscription?.sourceMessageId
|
||||
?? null,
|
||||
sourceText: committed?.item?.sourceText
|
||||
?? committed?.task?.sourceText
|
||||
?? committed?.subscription?.sourceText
|
||||
?? null,
|
||||
createdAt: now,
|
||||
updatedAt: now,
|
||||
};
|
||||
|
||||
if (kind === 'timed_reminder') {
|
||||
const item = committed.item ?? {};
|
||||
const reminder = committed.reminder ?? {};
|
||||
return {
|
||||
...base,
|
||||
type: 'reminder',
|
||||
title: item.title ?? '提醒',
|
||||
spec: { itemId: item.id, reminderId: reminder.id },
|
||||
trigger: {
|
||||
kind: 'at',
|
||||
at: item.startAt ?? reminder.remindAt,
|
||||
timezone: item.timezone ?? timezone,
|
||||
},
|
||||
action: { kind: 'notify', channel: reminder.channel ?? 'wechat' },
|
||||
nextRunAt: Number(reminder.remindAt ?? item.startAt ?? null) || null,
|
||||
legacyRef: { table: 'h5_schedule_reminders', id: reminder.id },
|
||||
};
|
||||
}
|
||||
|
||||
if (kind === 'create_todo') {
|
||||
const item = committed.item ?? {};
|
||||
return {
|
||||
...base,
|
||||
type: 'todo',
|
||||
title: item.title ?? '待办',
|
||||
spec: { itemId: item.id },
|
||||
trigger: { kind: 'none', timezone: item.timezone ?? timezone },
|
||||
action: { kind: 'none' },
|
||||
nextRunAt: null,
|
||||
legacyRef: { table: 'h5_schedule_items', id: item.id },
|
||||
};
|
||||
}
|
||||
|
||||
if (kind === 'create_daily_todo_digest') {
|
||||
const subscription = committed.subscription ?? {};
|
||||
return {
|
||||
...base,
|
||||
type: 'digest',
|
||||
title: '每日待办摘要',
|
||||
spec: { subscriptionId: subscription.id },
|
||||
trigger: {
|
||||
kind: 'cron',
|
||||
repeat: 'daily',
|
||||
hour: subscription.hour,
|
||||
minute: subscription.minute ?? 0,
|
||||
timezone: subscription.timezone ?? timezone,
|
||||
},
|
||||
action: { kind: 'notify', channel: subscription.channel ?? 'wechat' },
|
||||
nextRunAt: subscription.nextRunAt ?? null,
|
||||
legacyRef: { table: 'h5_schedule_digest_subscriptions', id: subscription.id },
|
||||
};
|
||||
}
|
||||
|
||||
if (kind === 'create_balance_alert') {
|
||||
const subscription = committed.subscription ?? {};
|
||||
return {
|
||||
...base,
|
||||
type: 'condition',
|
||||
title: '余额低提醒',
|
||||
spec: { subscriptionId: subscription.id, thresholdCents: subscription.thresholdCents },
|
||||
trigger: {
|
||||
kind: 'condition',
|
||||
watch: 'balance',
|
||||
op: 'lt',
|
||||
value: subscription.thresholdCents,
|
||||
},
|
||||
action: { kind: 'notify', channel: subscription.channel ?? 'wechat' },
|
||||
nextRunAt: subscription.nextRunAt ?? null,
|
||||
actionLevel: 2,
|
||||
legacyRef: { table: 'h5_balance_alert_subscriptions', id: subscription.id },
|
||||
};
|
||||
}
|
||||
|
||||
if (kind === 'scheduled_task') {
|
||||
const task = committed.task ?? {};
|
||||
return {
|
||||
...base,
|
||||
type: 'automation',
|
||||
title: task.title ?? '定时自动任务',
|
||||
spec: { taskSpec: task.taskSpec, scheduledTaskId: task.id },
|
||||
trigger: {
|
||||
kind: 'cron',
|
||||
repeat: task.recurrence,
|
||||
hour: task.hour,
|
||||
minute: task.minute ?? 0,
|
||||
weekday: task.weekday,
|
||||
timezone: task.timezone ?? timezone,
|
||||
},
|
||||
action: { kind: 'agent_run', notifyChannel: task.notifyChannel ?? 'both' },
|
||||
nextRunAt: task.nextRunAt ?? null,
|
||||
actionLevel: 2,
|
||||
legacyRef: { table: 'h5_scheduled_tasks', id: task.id },
|
||||
};
|
||||
}
|
||||
|
||||
return null;
|
||||
}
|
||||
|
||||
export function mapLegacyScheduledTaskRow(row, timezone = DEFAULT_TIMEZONE) {
|
||||
return {
|
||||
userId: row.user_id,
|
||||
type: 'automation',
|
||||
title: row.title,
|
||||
spec: { taskSpec: row.task_spec, scheduledTaskId: row.id },
|
||||
trigger: {
|
||||
kind: 'cron',
|
||||
repeat: row.recurrence,
|
||||
hour: row.hour == null ? null : Number(row.hour),
|
||||
minute: Number(row.minute ?? 0),
|
||||
weekday: row.weekday == null ? null : Number(row.weekday),
|
||||
timezone: row.timezone || timezone,
|
||||
},
|
||||
action: { kind: 'agent_run' },
|
||||
actionLevel: 2,
|
||||
notifyChannel: row.notify_channel ?? 'both',
|
||||
status: row.status ?? 'active',
|
||||
nextRunAt: row.next_run_at == null ? null : Number(row.next_run_at),
|
||||
lastRunAt: row.last_run_at == null ? null : Number(row.last_run_at),
|
||||
legacyRef: { table: 'h5_scheduled_tasks', id: row.id },
|
||||
sourceChannel: row.source_channel ?? null,
|
||||
sourceMessageId: row.source_message_id ?? null,
|
||||
sourceText: row.source_text ?? null,
|
||||
};
|
||||
}
|
||||
|
||||
export function mapLegacyReminderRow(row, timezone = DEFAULT_TIMEZONE) {
|
||||
return {
|
||||
userId: row.user_id ?? row.userId,
|
||||
type: 'reminder',
|
||||
title: row.title,
|
||||
spec: { itemId: row.item_id, reminderId: row.reminder_id ?? row.id },
|
||||
trigger: {
|
||||
kind: 'at',
|
||||
at: Number(row.remind_at),
|
||||
timezone: row.timezone || timezone,
|
||||
},
|
||||
action: { kind: 'notify', channel: row.channel ?? 'wechat' },
|
||||
actionLevel: 1,
|
||||
notifyChannel: row.channel ?? 'wechat',
|
||||
status: row.status ?? 'pending',
|
||||
nextRunAt: Number(row.remind_at),
|
||||
legacyRef: { table: 'h5_schedule_reminders', id: row.reminder_id ?? row.id },
|
||||
sourceChannel: row.source_channel ?? null,
|
||||
sourceMessageId: row.source_message_id ?? null,
|
||||
sourceText: row.source_text ?? null,
|
||||
};
|
||||
}
|
||||
|
||||
export function mapLegacyDigestRow(row, timezone = DEFAULT_TIMEZONE) {
|
||||
return {
|
||||
userId: row.user_id,
|
||||
type: 'digest',
|
||||
title: '每日待办摘要',
|
||||
spec: { subscriptionId: row.id },
|
||||
trigger: {
|
||||
kind: 'cron',
|
||||
repeat: 'daily',
|
||||
hour: Number(row.hour),
|
||||
minute: Number(row.minute ?? 0),
|
||||
timezone: row.timezone || timezone,
|
||||
},
|
||||
action: { kind: 'notify', channel: row.channel ?? 'wechat' },
|
||||
actionLevel: 2,
|
||||
notifyChannel: row.channel ?? 'wechat',
|
||||
status: row.status ?? 'active',
|
||||
nextRunAt: row.next_run_at == null ? null : Number(row.next_run_at),
|
||||
legacyRef: { table: 'h5_schedule_digest_subscriptions', id: row.id },
|
||||
sourceChannel: row.source_channel ?? null,
|
||||
sourceMessageId: row.source_message_id ?? null,
|
||||
sourceText: row.source_text ?? null,
|
||||
};
|
||||
}
|
||||
|
||||
export function mapLegacyBalanceAlertRow(row) {
|
||||
return {
|
||||
userId: row.user_id,
|
||||
type: 'condition',
|
||||
title: '余额低提醒',
|
||||
spec: { subscriptionId: row.id, thresholdCents: Number(row.threshold_cents) },
|
||||
trigger: {
|
||||
kind: 'condition',
|
||||
watch: 'balance',
|
||||
op: 'lt',
|
||||
value: Number(row.threshold_cents),
|
||||
},
|
||||
action: { kind: 'notify', channel: row.channel ?? 'wechat' },
|
||||
actionLevel: 2,
|
||||
notifyChannel: row.channel ?? 'wechat',
|
||||
status: row.status ?? 'active',
|
||||
nextRunAt: row.next_run_at == null ? null : Number(row.next_run_at),
|
||||
legacyRef: { table: 'h5_balance_alert_subscriptions', id: row.id },
|
||||
sourceChannel: row.source_channel ?? null,
|
||||
sourceMessageId: row.source_message_id ?? null,
|
||||
sourceText: row.source_text ?? null,
|
||||
};
|
||||
}
|
||||
|
||||
export function mapLegacyTodoItemRow(row, timezone = DEFAULT_TIMEZONE) {
|
||||
return {
|
||||
userId: row.user_id,
|
||||
type: 'todo',
|
||||
title: row.title,
|
||||
spec: { itemId: row.id },
|
||||
trigger: { kind: 'none', timezone: row.timezone || timezone },
|
||||
action: { kind: 'none' },
|
||||
actionLevel: 1,
|
||||
notifyChannel: 'both',
|
||||
status: row.status ?? 'active',
|
||||
nextRunAt: null,
|
||||
legacyRef: { table: 'h5_schedule_items', id: row.id },
|
||||
sourceChannel: row.source_channel ?? null,
|
||||
sourceMessageId: row.source_message_id ?? null,
|
||||
sourceText: row.source_text ?? null,
|
||||
};
|
||||
}
|
||||
|
||||
export function createTaskUnifiedService(pool, { clock = { now: () => Date.now() } } = {}) {
|
||||
if (!pool) throw new Error('缺少数据库连接');
|
||||
|
||||
const insertUnifiedTask = async (task) => {
|
||||
const id = crypto.randomUUID();
|
||||
const now = clock.now();
|
||||
await pool.query(
|
||||
`INSERT INTO h5_tasks
|
||||
(id, user_id, type, title, spec_json, trigger_json, action_json, action_level,
|
||||
notify_channel, status, next_run_at, last_run_at, legacy_ref_json,
|
||||
source_channel, source_message_id, source_text, created_at, updated_at)
|
||||
VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)`,
|
||||
[
|
||||
id,
|
||||
task.userId,
|
||||
task.type,
|
||||
task.title,
|
||||
JSON.stringify(task.spec ?? {}),
|
||||
JSON.stringify(task.trigger ?? {}),
|
||||
JSON.stringify(task.action ?? {}),
|
||||
task.actionLevel ?? 1,
|
||||
task.notifyChannel ?? 'both',
|
||||
task.status ?? 'active',
|
||||
task.nextRunAt ?? null,
|
||||
task.lastRunAt ?? null,
|
||||
task.legacyRef ? JSON.stringify(task.legacyRef) : null,
|
||||
task.sourceChannel ?? null,
|
||||
task.sourceMessageId ?? null,
|
||||
task.sourceText ?? null,
|
||||
now,
|
||||
now,
|
||||
],
|
||||
);
|
||||
const [rows] = await pool.query(`SELECT * FROM h5_tasks WHERE id = ? LIMIT 1`, [id]);
|
||||
return rowToUnifiedTask(rows?.[0]);
|
||||
};
|
||||
|
||||
const upsertUnifiedTask = async (mapped) => {
|
||||
if (!mapped?.userId || !mapped?.type || !mapped?.title) return null;
|
||||
if (mapped.legacyRef?.table && mapped.legacyRef?.id) {
|
||||
const [existing] = await pool.query(
|
||||
`SELECT * FROM h5_tasks
|
||||
WHERE user_id = ? AND legacy_ref_json = ?
|
||||
LIMIT 1`,
|
||||
[mapped.userId, JSON.stringify(mapped.legacyRef)],
|
||||
);
|
||||
if (existing?.[0]) return rowToUnifiedTask(existing[0]);
|
||||
}
|
||||
return insertUnifiedTask(mapped);
|
||||
};
|
||||
|
||||
const syncFromCommit = async ({ userId, kind, committed }) => {
|
||||
const mapped = mapCommitToUnifiedTask({ userId, kind, committed });
|
||||
return upsertUnifiedTask(mapped);
|
||||
};
|
||||
|
||||
const syncFromLegacyMapped = async (mapped) => upsertUnifiedTask(mapped);
|
||||
|
||||
const markLegacyCancelled = async ({ userId, legacyRef }) => {
|
||||
if (!userId || !legacyRef?.table || !legacyRef?.id) return 0;
|
||||
const [result] = await pool.query(
|
||||
`UPDATE h5_tasks
|
||||
SET status = 'cancelled', updated_at = ?
|
||||
WHERE user_id = ? AND legacy_ref_json = ?`,
|
||||
[clock.now(), userId, JSON.stringify(legacyRef)],
|
||||
);
|
||||
return Number(result?.affectedRows ?? 0);
|
||||
};
|
||||
|
||||
const syncLegacyLifecycle = async ({
|
||||
userId,
|
||||
legacyRef,
|
||||
status,
|
||||
nextRunAt = undefined,
|
||||
lastRunAt = undefined,
|
||||
lastError = undefined,
|
||||
}) => {
|
||||
if (!userId || !legacyRef?.table || !legacyRef?.id || !status) return 0;
|
||||
const sets = ['status = ?', 'updated_at = ?'];
|
||||
const params = [status, clock.now()];
|
||||
if (nextRunAt !== undefined) {
|
||||
sets.push('next_run_at = ?');
|
||||
params.push(nextRunAt);
|
||||
}
|
||||
if (lastRunAt !== undefined) {
|
||||
sets.push('last_run_at = ?');
|
||||
params.push(lastRunAt);
|
||||
}
|
||||
if (lastError !== undefined) {
|
||||
const [rows] = await pool.query(
|
||||
`SELECT spec_json FROM h5_tasks WHERE user_id = ? AND legacy_ref_json = ? LIMIT 1`,
|
||||
[userId, JSON.stringify(legacyRef)],
|
||||
);
|
||||
const spec = parseJson(rows?.[0]?.spec_json) ?? {};
|
||||
spec.lastError = lastError ?? null;
|
||||
sets.push('spec_json = ?');
|
||||
params.push(JSON.stringify(spec));
|
||||
}
|
||||
params.push(userId, JSON.stringify(legacyRef));
|
||||
const [result] = await pool.query(
|
||||
`UPDATE h5_tasks SET ${sets.join(', ')} WHERE user_id = ? AND legacy_ref_json = ?`,
|
||||
params,
|
||||
);
|
||||
return Number(result?.affectedRows ?? 0);
|
||||
};
|
||||
|
||||
const listStoredTasks = async ({ userId, status = 'active', limit = 20 } = {}) => {
|
||||
const [rows] = await pool.query(
|
||||
`SELECT * FROM h5_tasks
|
||||
WHERE user_id = ? AND status = ?
|
||||
ORDER BY COALESCE(next_run_at, 9223372036854775807), created_at DESC
|
||||
LIMIT ?`,
|
||||
[userId, status, Math.max(1, Math.min(100, Number(limit) || 20))],
|
||||
);
|
||||
return rows.map(rowToUnifiedTask);
|
||||
};
|
||||
|
||||
const listLegacyTasks = async ({ userId, limit = 20, timezone = DEFAULT_TIMEZONE } = {}) => {
|
||||
const tasks = [];
|
||||
const cap = Math.max(1, Math.min(100, Number(limit) || 20));
|
||||
|
||||
const [automations] = await pool.query(
|
||||
`SELECT id, user_id, title, task_spec, recurrence, hour, minute, weekday, timezone,
|
||||
next_run_at, status, source_channel, source_message_id, source_text, created_at, updated_at
|
||||
FROM h5_scheduled_tasks
|
||||
WHERE user_id = ? AND status = 'active'
|
||||
ORDER BY next_run_at ASC
|
||||
LIMIT ?`,
|
||||
[userId, cap],
|
||||
);
|
||||
for (const row of automations) {
|
||||
tasks.push({
|
||||
id: `legacy:scheduled:${row.id}`,
|
||||
...mapLegacyScheduledTaskRow(row, timezone),
|
||||
createdAt: Number(row.created_at),
|
||||
updatedAt: Number(row.updated_at),
|
||||
});
|
||||
}
|
||||
|
||||
const [reminders] = await pool.query(
|
||||
`SELECT r.id AS reminder_id, r.user_id, r.remind_at, r.channel, r.status,
|
||||
i.id AS item_id, i.title, i.timezone, i.source_channel, i.source_message_id, i.source_text,
|
||||
i.created_at, i.updated_at
|
||||
FROM h5_schedule_reminders r
|
||||
JOIN h5_schedule_items i ON i.id = r.item_id
|
||||
WHERE r.user_id = ? AND r.status IN ('pending', 'locked') AND i.deleted_at IS NULL
|
||||
ORDER BY r.remind_at ASC
|
||||
LIMIT ?`,
|
||||
[userId, cap],
|
||||
);
|
||||
for (const row of reminders) {
|
||||
tasks.push({
|
||||
id: `legacy:reminder:${row.reminder_id}`,
|
||||
...mapLegacyReminderRow(row, timezone),
|
||||
createdAt: Number(row.created_at),
|
||||
updatedAt: Number(row.updated_at),
|
||||
});
|
||||
}
|
||||
|
||||
const [digests] = await pool.query(
|
||||
`SELECT id, hour, minute, timezone, channel, status, next_run_at, source_channel,
|
||||
source_message_id, source_text, created_at, updated_at
|
||||
FROM h5_schedule_digest_subscriptions
|
||||
WHERE user_id = ? AND status = 'active'
|
||||
LIMIT ?`,
|
||||
[userId, cap],
|
||||
);
|
||||
for (const row of digests) {
|
||||
tasks.push({
|
||||
id: `legacy:digest:${row.id}`,
|
||||
...mapLegacyDigestRow(row, timezone),
|
||||
createdAt: Number(row.created_at),
|
||||
updatedAt: Number(row.updated_at),
|
||||
});
|
||||
}
|
||||
|
||||
const [alerts] = await pool.query(
|
||||
`SELECT id, user_id, threshold_cents, channel, status, next_run_at, source_channel,
|
||||
source_message_id, source_text, created_at, updated_at
|
||||
FROM h5_balance_alert_subscriptions
|
||||
WHERE user_id = ? AND status = 'active'
|
||||
LIMIT ?`,
|
||||
[userId, cap],
|
||||
);
|
||||
for (const row of alerts) {
|
||||
tasks.push({
|
||||
id: `legacy:condition:${row.id}`,
|
||||
...mapLegacyBalanceAlertRow(row),
|
||||
createdAt: Number(row.created_at),
|
||||
updatedAt: Number(row.updated_at),
|
||||
});
|
||||
}
|
||||
|
||||
const [todos] = await pool.query(
|
||||
`SELECT id, user_id, title, timezone, status, source_channel, source_message_id, source_text,
|
||||
created_at, updated_at
|
||||
FROM h5_schedule_items
|
||||
WHERE user_id = ? AND kind = 'task' AND status = 'active' AND deleted_at IS NULL
|
||||
ORDER BY created_at DESC
|
||||
LIMIT ?`,
|
||||
[userId, cap],
|
||||
);
|
||||
for (const row of todos) {
|
||||
tasks.push({
|
||||
id: `legacy:todo:${row.id}`,
|
||||
...mapLegacyTodoItemRow(row, timezone),
|
||||
createdAt: Number(row.created_at),
|
||||
updatedAt: Number(row.updated_at),
|
||||
});
|
||||
}
|
||||
|
||||
return tasks.slice(0, cap);
|
||||
};
|
||||
|
||||
const collectLegacyMigrationCandidates = async ({ userId = null } = {}) => {
|
||||
const mapped = [];
|
||||
|
||||
const [automations] = userId
|
||||
? await pool.query(`SELECT * FROM h5_scheduled_tasks WHERE user_id = ? AND status = 'active'`, [userId])
|
||||
: await pool.query(`SELECT * FROM h5_scheduled_tasks WHERE status = 'active'`);
|
||||
mapped.push(...automations.map((row) => mapLegacyScheduledTaskRow(row)));
|
||||
|
||||
const reminderSql = `SELECT r.id AS reminder_id, r.user_id, r.remind_at, r.channel, r.status,
|
||||
i.id AS item_id, i.title, i.timezone, i.source_channel, i.source_message_id, i.source_text
|
||||
FROM h5_schedule_reminders r
|
||||
JOIN h5_schedule_items i ON i.id = r.item_id
|
||||
WHERE ${userId ? 'r.user_id = ? AND' : ''} r.status IN ('pending', 'locked') AND i.deleted_at IS NULL`;
|
||||
const [reminders] = userId
|
||||
? await pool.query(reminderSql, [userId])
|
||||
: await pool.query(reminderSql);
|
||||
mapped.push(...reminders.map((row) => mapLegacyReminderRow(row)));
|
||||
|
||||
const [digests] = userId
|
||||
? await pool.query(`SELECT * FROM h5_schedule_digest_subscriptions WHERE user_id = ? AND status = 'active'`, [userId])
|
||||
: await pool.query(`SELECT * FROM h5_schedule_digest_subscriptions WHERE status = 'active'`);
|
||||
mapped.push(...digests.map((row) => mapLegacyDigestRow(row)));
|
||||
|
||||
const [alerts] = userId
|
||||
? await pool.query(`SELECT * FROM h5_balance_alert_subscriptions WHERE user_id = ? AND status = 'active'`, [userId])
|
||||
: await pool.query(`SELECT * FROM h5_balance_alert_subscriptions WHERE status = 'active'`);
|
||||
mapped.push(...alerts.map((row) => mapLegacyBalanceAlertRow(row)));
|
||||
|
||||
const [todos] = userId
|
||||
? await pool.query(`SELECT * FROM h5_schedule_items WHERE user_id = ? AND kind = 'task' AND status = 'active' AND deleted_at IS NULL`, [userId])
|
||||
: await pool.query(`SELECT * FROM h5_schedule_items WHERE kind = 'task' AND status = 'active' AND deleted_at IS NULL`);
|
||||
mapped.push(...todos.map((row) => mapLegacyTodoItemRow(row)));
|
||||
|
||||
return mapped;
|
||||
};
|
||||
|
||||
const migrateLegacyTasks = async ({ userId = null, dryRun = true } = {}) => {
|
||||
const candidates = await collectLegacyMigrationCandidates({ userId });
|
||||
const stats = { scanned: candidates.length, inserted: 0, skipped: 0, errors: 0 };
|
||||
if (dryRun) return { ...stats, dryRun: true, sample: candidates.slice(0, 5) };
|
||||
|
||||
for (const candidate of candidates) {
|
||||
try {
|
||||
const before = await pool.query(
|
||||
`SELECT id FROM h5_tasks WHERE user_id = ? AND legacy_ref_json = ? LIMIT 1`,
|
||||
[candidate.userId, JSON.stringify(candidate.legacyRef)],
|
||||
);
|
||||
if (before?.[0]?.[0]) {
|
||||
stats.skipped += 1;
|
||||
continue;
|
||||
}
|
||||
await upsertUnifiedTask(candidate);
|
||||
stats.inserted += 1;
|
||||
} catch {
|
||||
stats.errors += 1;
|
||||
}
|
||||
}
|
||||
return { ...stats, dryRun: false };
|
||||
};
|
||||
|
||||
const listUserTasks = async ({ userId, status = 'active', limit = 20, timezone = DEFAULT_TIMEZONE } = {}) => {
|
||||
if (!userId) throw new Error('缺少用户');
|
||||
const stored = await listStoredTasks({ userId, status, limit });
|
||||
const mirrored = new Set(
|
||||
stored.map((task) => legacyKey(task.legacyRef)).filter(Boolean),
|
||||
);
|
||||
const legacy = await listLegacyTasks({ userId, limit, timezone });
|
||||
const merged = [...stored];
|
||||
for (const task of legacy) {
|
||||
const key = legacyKey(task.legacyRef);
|
||||
if (key && mirrored.has(key)) continue;
|
||||
merged.push(task);
|
||||
}
|
||||
merged.sort((a, b) => {
|
||||
const aRun = a.nextRunAt ?? Number.MAX_SAFE_INTEGER;
|
||||
const bRun = b.nextRunAt ?? Number.MAX_SAFE_INTEGER;
|
||||
if (aRun !== bRun) return aRun - bRun;
|
||||
return b.createdAt - a.createdAt;
|
||||
});
|
||||
return merged.slice(0, Math.max(1, Math.min(100, Number(limit) || 20)));
|
||||
};
|
||||
|
||||
return {
|
||||
syncFromCommit,
|
||||
syncFromLegacyMapped,
|
||||
markLegacyCancelled,
|
||||
syncLegacyLifecycle,
|
||||
migrateLegacyTasks,
|
||||
listUserTasks,
|
||||
listStoredTasks,
|
||||
listLegacyTasks,
|
||||
formatUnifiedTaskListReply,
|
||||
};
|
||||
}
|
||||
|
||||
export { formatTriggerLabel, mapCommitToUnifiedTask };
|
||||
Reference in New Issue
Block a user