koc-loop/lib/mcp-operations.ts

import { enrichDistributionAccount } from "./account-enrichment-service";
import {
  collectDistributionMetrics,
  createCollectionRunTasks,
  retryFailedCollections,
  runDueScheduledCollections,
  shanghaiDateFromTimestamp,
} from "./collection-service";
import { issueMcpExportToken } from "./mcp-export-token";
import {
  resolveCollectionMcpConfig,
  type CollectionMcpBindings,
} from "./mcp-collection-client";
import { ensureSchema, getRawDb } from "./mvp-db";
import { extractAnyPublishUrl } from "./publish-url";
import { buildClaimUrl } from "./task-service";

export type McpOperationBindings = CollectionMcpBindings & {
  KOC_PORTAL_URL?: string;
};

type Pagination = { limit?: number; offset?: number };
type ResourceFilters = Pagination & {
  query?: string;
  ipLocation?: string;
  cooperationSource?: string;
  platform?: string;
};

function page(input: Pagination) {
  return {
    limit: Math.max(1, Math.min(200, Math.floor(input.limit ?? 50))),
    offset: Math.max(0, Math.floor(input.offset ?? 0)),
  };
}

function like(value: string) {
  return `%${value.replace(/[\\%_]/g, "\\$&")}%`;
}

function parseJsonArray(value: unknown) {
  try {
    const parsed = JSON.parse(String(value ?? "[]"));
    return Array.isArray(parsed) ? parsed : [];
  } catch {
    return [];
  }
}

async function taskExists(taskId: string) {
  const task = await getRawDb()
    .prepare("SELECT id FROM tasks WHERE id = ?")
    .bind(taskId)
    .first<{ id: string }>();
  if (!task) throw new Error("任务不存在");
}

export async function taskList(
  input: Pagination & { query?: string; status?: string },
  portalUrl: string,
) {
  await ensureSchema();
  const { limit, offset } = page(input);
  const conditions: string[] = [];
  const bindings: unknown[] = [];
  if (input.query?.trim()) {
    conditions.push("(t.name LIKE ? ESCAPE '\\' OR t.brand LIKE ? ESCAPE '\\')");
    const pattern = like(input.query.trim());
    bindings.push(pattern, pattern);
  }
  if (input.status?.trim() && input.status !== "all") {
    conditions.push("t.status = ?");
    bindings.push(input.status.trim());
  }
  const where = conditions.length ? `WHERE ${conditions.join(" AND ")}` : "";
  const db = getRawDb();
  const [rows, count] = await Promise.all([
    db
      .prepare(
        `SELECT t.*,
          (SELECT COUNT(*) FROM contents c WHERE c.task_id = t.id) AS note_count,
          (SELECT COUNT(*) FROM distributions d WHERE d.task_id = t.id) AS claimed_count,
          (SELECT COUNT(*) FROM distributions d WHERE d.task_id = t.id AND COALESCE(d.publish_url, '') != '') AS published_count,
          (SELECT COUNT(*) FROM distributions d WHERE d.task_id = t.id AND d.exposure IS NOT NULL AND d.views IS NOT NULL) AS day7_count
         FROM tasks t ${where}
         ORDER BY t.created_at DESC LIMIT ${limit} OFFSET ${offset}`,
      )
      .bind(...bindings)
      .all<Record<string, unknown>>(),
    db
      .prepare(`SELECT COUNT(*) AS total FROM tasks t ${where}`)
      .bind(...bindings)
      .first<{ total: number }>(),
  ]);
  return {
    total: count?.total ?? 0,
    limit,
    offset,
    tasks: rows.results.map(
      (row): Record<string, unknown> & {
        collection_days: unknown[];
        claim_url: string | null;
      } => ({
        ...row,
        collection_days: parseJsonArray(row.collection_days),
        claim_url:
          portalUrl && row.share_token
            ? buildClaimUrl(portalUrl, String(row.share_token))
            : null,
      }),
    ),
  };
}

export async function taskGet(taskId: string, portalUrl: string) {
  await ensureSchema();
  const db = getRawDb();
  const task = await db
    .prepare("SELECT * FROM tasks WHERE id = ?")
    .bind(taskId)
    .first<Record<string, unknown>>();
  if (!task) throw new Error("任务不存在");
  const [notes, claims, runs] = await Promise.all([
    db
      .prepare(
        `SELECT c.id AS content_id, c.source_row, c.title, c.body, c.image_assets, c.video_assets, c.status AS content_status,
          d.id AS distribution_id, d.status AS distribution_status, d.publish_url, d.publish_time,
          d.exposure, d.views, d.latest_likes, d.latest_comments, d.latest_collects, d.latest_shares,
          d.collection_status, d.collection_status_description, d.collection_updated_at,
          a.id AS account_id, a.nickname AS account_nickname, a.profile_url, a.followers,
          p.name AS cooperation_source, cl.claimant_name
         FROM contents c
         LEFT JOIN distributions d ON d.content_id = c.id
         LEFT JOIN accounts a ON a.id = d.account_id
         LEFT JOIN partners p ON p.id = d.partner_id
         LEFT JOIN claims cl ON cl.id = d.claim_id
         WHERE c.task_id = ?
         ORDER BY COALESCE(c.source_row, 999999), c.created_at, d.claimed_at`,
      )
      .bind(taskId)
      .all<Record<string, unknown>>(),
    db
      .prepare(
        `SELECT cl.*, p.name AS partner_name,
          (SELECT COUNT(*) FROM distributions d WHERE d.claim_id = cl.id AND COALESCE(d.publish_url, '') != '') AS published_count
         FROM claims cl JOIN partners p ON p.id = cl.partner_id
         WHERE cl.task_id = ? ORDER BY cl.created_at DESC`,
      )
      .bind(taskId)
      .all<Record<string, unknown>>(),
    db
      .prepare(
        `SELECT * FROM collection_runs WHERE task_id = ?
         ORDER BY scheduled_date DESC, created_at DESC LIMIT 500`,
      )
      .bind(taskId)
      .all<Record<string, unknown>>(),
  ]);
  const taskOutput: Record<string, unknown> & {
    collection_days: unknown[];
    claim_url: string | null;
  } = {
      ...task,
      collection_days: parseJsonArray(task.collection_days),
      claim_url:
        portalUrl && task.share_token
          ? buildClaimUrl(portalUrl, String(task.share_token))
          : null,
  };
  return {
    task: taskOutput,
    notes: notes.results.map((row) => ({
      ...row,
      image_assets: parseJsonArray(row.image_assets),
      video_assets: parseJsonArray(row.video_assets),
      total_interactions:
        row.latest_likes == null
          ? null
          : Number(row.latest_likes) +
            Number(row.latest_comments ?? 0) +
            Number(row.latest_collects ?? 0) +
            Number(row.latest_shares ?? 0),
    })),
    claims: claims.results,
    collection_runs: runs.results,
  };
}

export async function recoveryList(
  input: Pagination & {
    taskId?: string;
    stage?: "all" | "published" | "unfilled" | "waiting_day7" | "day7_due";
  },
) {
  await ensureSchema();
  const { limit, offset } = page(input);
  const db = getRawDb();
  const conditions: string[] = [];
  const bindings: unknown[] = [];
  if (input.taskId?.trim()) {
    conditions.push("d.task_id = ?");
    bindings.push(input.taskId.trim());
  }
  const stage = input.stage ?? "all";
  if (stage === "published") conditions.push("COALESCE(d.publish_url, '') != ''");
  if (stage === "unfilled") conditions.push("COALESCE(d.publish_url, '') = ''");
  if (stage === "waiting_day7") {
    conditions.push("COALESCE(d.publish_url, '') != '' AND (d.exposure IS NULL OR d.views IS NULL)");
  }
  if (stage === "day7_due") {
    conditions.push(
      "COALESCE(d.publish_url, '') != '' AND (d.exposure IS NULL OR d.views IS NULL) AND datetime(d.publish_time, '+7 days') <= CURRENT_TIMESTAMP",
    );
  }
  const where = conditions.length ? `WHERE ${conditions.join(" AND ")}` : "";
  const base = `FROM distributions d
    JOIN tasks t ON t.id = d.task_id
    JOIN contents c ON c.id = d.content_id
    LEFT JOIN accounts a ON a.id = d.account_id
    LEFT JOIN partners p ON p.id = d.partner_id
    LEFT JOIN claims cl ON cl.id = d.claim_id ${where}`;
  const [rows, count] = await Promise.all([
    db
      .prepare(
        `SELECT d.*, t.name AS task_name, t.platform AS task_platform,
          t.content_format, c.source_row, c.title,
          a.nickname AS account_nickname, a.profile_url, p.name AS cooperation_source,
          cl.claimant_name ${base}
         ORDER BY COALESCE(d.publish_time, d.claimed_at) DESC LIMIT ${limit} OFFSET ${offset}`,
      )
      .bind(...bindings)
      .all<Record<string, unknown>>(),
    db
      .prepare(`SELECT COUNT(*) AS total ${base}`)
      .bind(...bindings)
      .first<{ total: number }>(),
  ]);
  return {
    total: count?.total ?? 0,
    limit,
    offset,
    items: rows.results.map((row) => ({
      ...row,
      total_interactions:
        row.latest_likes == null
          ? null
          : Number(row.latest_likes) + Number(row.latest_comments ?? 0) + Number(row.latest_collects ?? 0) + Number(row.latest_shares ?? 0),
    })),
  };
}

export async function recoveryExport(taskId: string, origin: string) {
  await ensureSchema();
  await taskExists(taskId);
  const issued = await issueMcpExportToken("recovery", { taskId });
  return {
    task_id: taskId,
    download_url: `${origin}/api/recovery-export?token=${encodeURIComponent(issued.token)}`,
    expires_at: issued.expiresAt,
  };
}

export async function setCollectionPlan(
  taskId: string,
  startDate: string,
  days: number[],
  bindings: McpOperationBindings,
) {
  await ensureSchema();
  await taskExists(taskId);
  const normalizedDays = [...new Set(days)]
    .filter((day) => Number.isInteger(day) && day >= 1 && day <= 7)
    .sort((a, b) => a - b);
  if (!/^\d{4}-\d{2}-\d{2}$/.test(startDate) || Number.isNaN(Date.parse(`${startDate}T00:00:00+08:00`))) {
    throw new Error("开始采集日期无效");
  }
  if (!normalizedDays.length) throw new Error("请至少选择一个采集日");
  const db = getRawDb();
  await db.batch([
    db.prepare(
      `UPDATE tasks SET collection_start_date = ?, collection_days = ?,
       collection_schedule_updated_at = CURRENT_TIMESTAMP WHERE id = ?`,
    ).bind(startDate, JSON.stringify(normalizedDays), taskId),
    db.prepare(
      `UPDATE distributions SET
       collection_status = CASE WHEN latest_likes IS NULL THEN 'scheduled' ELSE collection_status END,
       collection_status_description = CASE WHEN latest_likes IS NULL THEN ? ELSE collection_status_description END,
       updated_at = CURRENT_TIMESTAMP
       WHERE task_id = ? AND COALESCE(publish_url, '') != ''`,
    ).bind(`已安排${normalizedDays.length}个采集日，每日09:00执行`, taskId),
  ]);
  const created = await createCollectionRunTasks(db, taskId, startDate, normalizedDays);
  const catchup = await runDueScheduledCollections(
    db,
    Date.now(),
    resolveCollectionMcpConfig(bindings),
    "catchup",
    taskId,
  );
  return { task_id: taskId, start_date: startDate, days: normalizedDays, created, catchup };
}

export async function runDueCollections(taskId: string | undefined, bindings: McpOperationBindings) {
  await ensureSchema();
  if (taskId) await taskExists(taskId);
  return runDueScheduledCollections(
    getRawDb(),
    Date.now(),
    resolveCollectionMcpConfig(bindings),
    "catchup",
    taskId,
  );
}

export async function collectNow(
  distributionId: string,
  scheduleDay: number | undefined,
  bindings: McpOperationBindings,
) {
  await ensureSchema();
  const day = scheduleDay ?? null;
  const metrics = await collectDistributionMetrics(
    getRawDb(),
    distributionId,
    shanghaiDateFromTimestamp(Date.now()),
    day,
    "manual",
    resolveCollectionMcpConfig(bindings),
  );
  return { distribution_id: distributionId, schedule_day: day, ...metrics };
}

export async function retryFailed(taskId: string | undefined, bindings: McpOperationBindings) {
  await ensureSchema();
  const db = getRawDb();
  const ids = taskId
    ? [taskId]
    : (
        await db
          .prepare(
            `SELECT DISTINCT task_id FROM distributions
             WHERE collection_status = 'failed' ORDER BY updated_at DESC LIMIT 20`,
          )
          .all<{ task_id: string }>()
      ).results.map((row) => row.task_id);
  if (taskId) await taskExists(taskId);
  const results = [];
  for (const id of ids) {
    results.push({ task_id: id, ...(await retryFailedCollections(db, id, resolveCollectionMcpConfig(bindings))) });
  }
  return {
    task_count: results.length,
    attempted: results.reduce((sum, item) => sum + item.attempted, 0),
    succeeded: results.reduce((sum, item) => sum + item.succeeded, 0),
    failed: results.reduce((sum, item) => sum + item.failed, 0),
    tasks: results,
  };
}

function resourceWhere(input: ResourceFilters) {
  const conditions: string[] = [];
  const bindings: unknown[] = [];
  if (input.query?.trim()) {
    conditions.push(
      "(a.nickname LIKE ? ESCAPE '\\' OR a.public_account_id LIKE ? ESCAPE '\\' OR a.current_contact LIKE ? ESCAPE '\\' OR a.tags LIKE ? ESCAPE '\\' OR a.bio LIKE ? ESCAPE '\\')",
    );
    const pattern = like(input.query.trim());
    bindings.push(pattern, pattern, pattern, pattern, pattern);
  }
  if (input.ipLocation?.trim()) {
    conditions.push("a.ip_location LIKE ? ESCAPE '\\'");
    bindings.push(like(input.ipLocation.trim()));
  }
  if (input.cooperationSource?.trim()) {
    conditions.push(
      `(a.cooperation_source LIKE ? ESCAPE '\\' OR
        EXISTS (SELECT 1 FROM distributions dx
        JOIN partners px ON px.id = dx.partner_id
        LEFT JOIN claims cx ON cx.id = dx.claim_id
        WHERE dx.account_id = a.id
          AND (cx.claimant_name IS NULL OR px.name != cx.claimant_name)
          AND px.name LIKE ? ESCAPE '\\'))`,
    );
    const pattern = like(input.cooperationSource.trim());
    bindings.push(pattern, pattern);
  }
  if (input.platform?.trim() && input.platform !== "all") {
    conditions.push("a.platform = ?");
    bindings.push(input.platform.trim());
  }
  return { where: conditions.length ? `WHERE ${conditions.join(" AND ")}` : "", bindings };
}

export async function resourceSearch(input: ResourceFilters) {
  await ensureSchema();
  const { limit, offset } = page(input);
  const { where, bindings } = resourceWhere(input);
  const db = getRawDb();
  const select = `SELECT a.*,
    (SELECT COUNT(*) FROM distributions d WHERE d.account_id = a.id) AS cooperation_count,
    (SELECT GROUP_CONCAT(DISTINCT p.name)
     FROM distributions d
     JOIN partners p ON p.id = d.partner_id
     LEFT JOIN claims c ON c.id = d.claim_id
     WHERE d.account_id = a.id
       AND (c.claimant_name IS NULL OR p.name != c.claimant_name)) AS cooperation_sources`;
  const [rows, count] = await Promise.all([
    db.prepare(`${select} FROM accounts a ${where} ORDER BY a.last_seen_at DESC LIMIT ${limit} OFFSET ${offset}`)
      .bind(...bindings).all<Record<string, unknown>>(),
    db.prepare(`SELECT COUNT(*) AS total FROM accounts a ${where}`).bind(...bindings).first<{ total: number }>(),
  ]);
  return {
    total: count?.total ?? 0,
    limit,
    offset,
    accounts: rows.results.map(
      (row): Record<string, unknown> & { cooperation_sources: string[] } => ({
        ...row,
        cooperation_sources: [
          ...new Set(
            `${row.cooperation_sources ?? ""}、${row.cooperation_source ?? ""}`
              .split(/[、,，;；|]/)
              .map((item) => item.trim())
              .filter(Boolean),
          ),
        ],
      }),
    ),
  };
}

export async function resourceGet(accountId: string) {
  await ensureSchema();
  const db = getRawDb();
  const account = await db.prepare("SELECT * FROM accounts WHERE id = ?").bind(accountId).first<Record<string, unknown>>();
  if (!account) throw new Error("账号不存在");
  const history = await db.prepare(
    `SELECT d.id AS distribution_id, d.task_id, t.name AS task_name, c.title,
      d.publish_url, d.publish_time, d.latest_likes, d.latest_comments, d.latest_collects, d.latest_shares,
      d.exposure, d.views, p.name AS cooperation_source, cl.claimant_name
     FROM distributions d JOIN tasks t ON t.id = d.task_id
     JOIN contents c ON c.id = d.content_id JOIN partners p ON p.id = d.partner_id
     LEFT JOIN claims cl ON cl.id = d.claim_id
     WHERE d.account_id = ? ORDER BY COALESCE(d.publish_time, d.claimed_at) DESC`,
  ).bind(accountId).all<Record<string, unknown>>();
  return { account, cooperation_history: history.results };
}

export async function backfillResourceProfile(
  input: { distributionId?: string; publishUrl?: string },
  bindings: McpOperationBindings,
) {
  await ensureSchema();
  const db = getRawDb();
  const publishUrl = input.publishUrl ? extractAnyPublishUrl(input.publishUrl) : "";
  const row = input.distributionId
    ? await db.prepare(
        `SELECT d.id, d.publish_url, COALESCE(a.nickname, '') AS nickname
         FROM distributions d LEFT JOIN accounts a ON a.id = d.account_id WHERE d.id = ?`,
      ).bind(input.distributionId).first<{ id: string; publish_url: string | null; nickname: string }>()
    : publishUrl
      ? await db.prepare(
          `SELECT d.id, d.publish_url, COALESCE(a.nickname, '') AS nickname
           FROM distributions d LEFT JOIN accounts a ON a.id = d.account_id
           WHERE d.publish_url = ? ORDER BY d.updated_at DESC LIMIT 1`,
        ).bind(publishUrl).first<{ id: string; publish_url: string | null; nickname: string }>()
      : null;
  if (!row?.publish_url) throw new Error("没有找到可补全的发布记录");
  const result = await enrichDistributionAccount(
    db,
    row.id,
    row.publish_url,
    row.nickname,
    resolveCollectionMcpConfig(bindings),
  );
  const account = result.updated && result.accountId
    ? await db.prepare("SELECT * FROM accounts WHERE id = ?").bind(result.accountId).first<Record<string, unknown>>()
    : null;
  return { distribution_id: row.id, ...result, account };
}

export async function resourceExport(input: ResourceFilters, origin: string) {
  const result = await resourceSearch({ ...input, limit: 200, offset: 0 });
  const allIds: string[] = result.accounts.map((account) => String(account.id));
  if (result.total > result.accounts.length) {
    const { where, bindings } = resourceWhere(input);
    const rows = await getRawDb().prepare(`SELECT a.id FROM accounts a ${where} ORDER BY a.last_seen_at DESC LIMIT 5000`)
      .bind(...bindings).all<{ id: string }>();
    allIds.splice(0, allIds.length, ...rows.results.map((row) => row.id));
  }
  if (!allIds.length) throw new Error("当前筛选结果为空");
  const issued = await issueMcpExportToken("resources", { accountIds: allIds });
  return {
    account_count: allIds.length,
    download_url: `${origin}/api/resources-export?token=${encodeURIComponent(issued.token)}`,
    expires_at: issued.expiresAt,
  };
}