john
a512fecdc5
Memind CI / Test, build, and release guards (push) Successful in 3m32s
MySQL JSON key order differs from JS stringify, which caused duplicate h5_tasks rows on re-run migrate. Co-authored-by: Cursor <cursoragent@cursor.com>
634 lines
22 KiB
JavaScript
634 lines
22 KiB
JavaScript
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}`;
|
||
}
|
||
|
||
/** MySQL JSON column key order is not stable; match by extracted fields. */
|
||
function legacyRefMatchSql(column = 'legacy_ref_json') {
|
||
return `${column} IS NOT NULL
|
||
AND JSON_UNQUOTE(JSON_EXTRACT(${column}, '$.table')) = ?
|
||
AND JSON_UNQUOTE(JSON_EXTRACT(${column}, '$.id')) = ?`;
|
||
}
|
||
|
||
function legacyRefMatchParams(legacyRef) {
|
||
return [legacyRef.table, legacyRef.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 ${legacyRefMatchSql()}
|
||
LIMIT 1`,
|
||
[mapped.userId, ...legacyRefMatchParams(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 ${legacyRefMatchSql()}`,
|
||
[clock.now(), userId, ...legacyRefMatchParams(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 ${legacyRefMatchSql()} LIMIT 1`,
|
||
[userId, ...legacyRefMatchParams(legacyRef)],
|
||
);
|
||
const spec = parseJson(rows?.[0]?.spec_json) ?? {};
|
||
spec.lastError = lastError ?? null;
|
||
sets.push('spec_json = ?');
|
||
params.push(JSON.stringify(spec));
|
||
}
|
||
params.push(userId, ...legacyRefMatchParams(legacyRef));
|
||
const [result] = await pool.query(
|
||
`UPDATE h5_tasks SET ${sets.join(', ')} WHERE user_id = ? AND ${legacyRefMatchSql()}`,
|
||
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 ${legacyRefMatchSql()} LIMIT 1`,
|
||
[candidate.userId, ...legacyRefMatchParams(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 };
|