koc-loop/lib/account-id-refresh-service.ts

import { runInBackground } from "./background";
import {
  resolveProfileDetailsFromMcp,
  type CollectionMcpConfig,
} from "./mcp-collection-client";
import type { DatabaseClient } from "./database";

export type PublicAccountIdRefreshStatus = {
  status: "running" | "done";
  total: number;
  processed: number;
  updated: number;
  failed: number;
  skipped: number;
  startedAt: string;
  finishedAt: string | null;
};

type RefreshTargetRow = {
  id: string;
  platform: string;
  platform_uid: string;
  profile_url: string;
};

export type ProfileScraper = (
  profileUrl: string,
  platform: "小红书" | "抖音",
) => Promise<{ redId: string | null }>;

const defaultScraper =
  (config: CollectionMcpConfig): ProfileScraper =>
  (profileUrl, platform) =>
    resolveProfileDetailsFromMcp(profileUrl, platform, config);

let currentJob: PublicAccountIdRefreshStatus | null = null;

export function getPublicAccountIdRefreshStatus(): PublicAccountIdRefreshStatus | null {
  return currentJob ? { ...currentJob } : null;
}

export function startPublicAccountIdRefresh(
  db: DatabaseClient,
  config: CollectionMcpConfig,
): PublicAccountIdRefreshStatus {
  if (currentJob?.status === "running") {
    return { ...currentJob };
  }
  const job: PublicAccountIdRefreshStatus = {
    status: "running",
    total: 0,
    processed: 0,
    updated: 0,
    failed: 0,
    skipped: 0,
    startedAt: new Date().toISOString(),
    finishedAt: null,
  };
  currentJob = job;
  runInBackground(
    executePublicAccountIdRefresh(job, db, defaultScraper(config)),
    "public account id refresh",
  );
  return { ...job };
}

async function executePublicAccountIdRefresh(
  job: PublicAccountIdRefreshStatus,
  db: DatabaseClient,
  scraper: ProfileScraper,
) {
  const rows = await db
    .prepare(
      `SELECT id, platform, platform_uid, profile_url
       FROM accounts
       WHERE profile_url IS NOT NULL AND profile_url != ''
         AND platform IN ('小红书', '抖音')`,
    )
    .all<RefreshTargetRow>();
  const targets = rows.results.filter(
    (row) => !row.platform_uid.startsWith("xhs-"),
  );
  job.total = targets.length;

  let cursor = 0;
  const worker = async () => {
    while (cursor < targets.length) {
      const row = targets[cursor];
      cursor += 1;
      try {
        const details = await scraper(
          row.profile_url,
          row.platform === "抖音" ? "抖音" : "小红书",
        );
        const redId = details.redId?.trim();
        if (redId) {
          await db
            .prepare(`UPDATE accounts SET public_account_id = ? WHERE id = ?`)
            .bind(redId, row.id)
            .run();
          job.updated += 1;
        } else {
          job.skipped += 1;
        }
      } catch {
        job.failed += 1;
      }
      job.processed += 1;
    }
  };
  await Promise.all(
    Array.from({ length: Math.min(4, targets.length) }, worker),
  );
  job.status = "done";
  job.finishedAt = new Date().toISOString();
}

export async function runPublicAccountIdRefresh(
  db: DatabaseClient,
  scraper: ProfileScraper,
): Promise<PublicAccountIdRefreshStatus> {
  const job: PublicAccountIdRefreshStatus = {
    status: "running",
    total: 0,
    processed: 0,
    updated: 0,
    failed: 0,
    skipped: 0,
    startedAt: new Date().toISOString(),
    finishedAt: null,
  };
  await executePublicAccountIdRefresh(job, db, scraper);
  return job;
}