memind/task-unified-service.mjs

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