feat: add competitor monitoring and listing review sync

package.json

@@ -18,7 +18,9 @@
     "verify:parse-rest": "tsx scripts/verify-parse-rest.ts",
     "import:workbook:parse-rest": "tsx scripts/import-workbook-parse-rest.ts",
     "sync:competitors:parse-rest": "tsx scripts/sync-parse-rest-competitors.ts",
+    "refresh:competitor-listings": "tsx scripts/refresh-competitor-listings.ts",
     "sync:jd-listings": "tsx scripts/sync-jd-listings.ts",
+    "sync:jd-listing-reviews": "tsx scripts/sync-jd-listing-reviews.ts",
     "enrich:listing-context": "tsx scripts/enrich-listing-context.ts",
     "score:jd-listings": "tsx scripts/score-listings.ts",
     "score:jd-listings:ai": "tsx scripts/score-listings-ai.ts",

scripts/refresh-competitor-listings.ts

@@ -0,0 +1,166 @@
+import 'dotenv/config';
+import { resolve } from 'node:path';
+import { pathToFileURL } from 'node:url';
+import { loadConfig } from '../src/config/env.js';
+import { ParseRestClient } from '../src/db/parse-rest.client.js';
+import { ApiError } from '../src/http/api-error.js';
+import { CompetitorListingMonitorService } from '../src/modules/competitor-listing-monitor/competitor-listing-monitor.service.js';
+import type { CompetitorListingRefreshRun } from '../src/modules/competitor-listing-monitor/domain.js';
+import { ParseRestCompetitorListingMonitorRepository } from '../src/modules/competitor-listing-monitor/repositories/parse-rest-competitor-listing-monitor.repository.js';
+import { FmodeVocEcommerceClient } from '../src/modules/domestic-voc/upstream/fmode-client.js';
+
+export const COMPETITOR_LISTING_REFRESH_EXIT = {
+  completed: 0,
+  failed: 1,
+  partial: 2,
+  alreadyRunning: 3,
+} as const;
+
+type RefreshCliService = Pick<CompetitorListingMonitorService, 'startRefresh' | 'getRun'>;
+
+export interface CompetitorListingRefreshCliOptions {
+  service: RefreshCliService;
+  workspaceId: string;
+  pollMs?: number;
+  timeoutMs?: number;
+  sleep?: (milliseconds: number) => Promise<void>;
+  write?: (line: string) => void;
+  writeError?: (line: string) => void;
+}
+
+export async function runCompetitorListingRefreshCli(
+  options: CompetitorListingRefreshCliOptions,
+): Promise<number> {
+  const pollMs = positiveInteger(options.pollMs ?? 2_000, 'pollMs');
+  const timeoutMs = positiveInteger(options.timeoutMs ?? 6 * 60 * 60 * 1_000, 'timeoutMs');
+  const sleep = options.sleep ?? ((milliseconds) => new Promise((resolve) => setTimeout(resolve, milliseconds)));
+  const write = options.write ?? console.log;
+  const writeError = options.writeError ?? console.error;
+  let queued: CompetitorListingRefreshRun;
+  try {
+    queued = await options.service.startRefresh(options.workspaceId, 'jd', 'scheduled');
+  } catch (error) {
+    if (error instanceof ApiError && error.code === 'competitor_listing_refresh_running') {
+      writeError(JSON.stringify({
+        status: 'already_running',
+        workspaceId: options.workspaceId,
+        error: error.code,
+      }));
+      return COMPETITOR_LISTING_REFRESH_EXIT.alreadyRunning;
+    }
+    writeError(JSON.stringify({
+      status: 'failed',
+      workspaceId: options.workspaceId,
+      error: 'competitor_listing_refresh_start_failed',
+    }));
+    return COMPETITOR_LISTING_REFRESH_EXIT.failed;
+  }
+
+  write(JSON.stringify({
+    status: queued.status,
+    workspaceId: queued.workspaceId,
+    runId: queued.id,
+    total: queued.total,
+    requestedAt: queued.requestedAt,
+  }));
+  const deadline = Date.now() + timeoutMs;
+  for (;;) {
+    let run: CompetitorListingRefreshRun | null;
+    try {
+      run = await options.service.getRun(options.workspaceId, queued.id);
+    } catch {
+      writeError(JSON.stringify({
+        status: 'failed',
+        workspaceId: options.workspaceId,
+        runId: queued.id,
+        error: 'competitor_listing_refresh_status_failed',
+      }));
+      return COMPETITOR_LISTING_REFRESH_EXIT.failed;
+    }
+    if (!run) {
+      writeError(JSON.stringify({
+        status: 'failed',
+        workspaceId: options.workspaceId,
+        runId: queued.id,
+        error: 'competitor_listing_refresh_run_not_found',
+      }));
+      return COMPETITOR_LISTING_REFRESH_EXIT.failed;
+    }
+    if (run.status === 'completed' || run.status === 'partial' || run.status === 'failed') {
+      const output = {
+        status: run.status,
+        workspaceId: run.workspaceId,
+        runId: run.id,
+        total: run.total,
+        completed: run.completed,
+        baseline: run.baseline,
+        unchanged: run.unchanged,
+        changed: run.changed,
+        failed: run.failed,
+        requestedAt: run.requestedAt,
+        startedAt: run.startedAt,
+        completedAt: run.completedAt,
+      };
+      const line = JSON.stringify(output);
+      if (run.status === 'failed') writeError(line);
+      else write(line);
+      return run.status === 'completed'
+        ? COMPETITOR_LISTING_REFRESH_EXIT.completed
+        : run.status === 'partial'
+          ? COMPETITOR_LISTING_REFRESH_EXIT.partial
+          : COMPETITOR_LISTING_REFRESH_EXIT.failed;
+    }
+    if (Date.now() >= deadline) {
+      writeError(JSON.stringify({
+        status: 'failed',
+        workspaceId: options.workspaceId,
+        runId: queued.id,
+        error: 'competitor_listing_refresh_timeout',
+      }));
+      return COMPETITOR_LISTING_REFRESH_EXIT.failed;
+    }
+    await sleep(pollMs);
+  }
+}
+
+export async function main(args = process.argv.slice(2)): Promise<number> {
+  const config = loadConfig();
+  if (config.storageDriver !== 'parse_rest') {
+    console.error(JSON.stringify({ status: 'failed', error: 'parse_rest_storage_required' }));
+    return COMPETITOR_LISTING_REFRESH_EXIT.failed;
+  }
+  const workspaceId = args.find((argument) => !argument.startsWith('--')) || config.auth.defaultWorkspaceId;
+  const pollMs = readIntegerArgument(args, '--poll-ms=', 2_000);
+  const timeoutMs = readIntegerArgument(args, '--timeout-ms=', 6 * 60 * 60 * 1_000);
+  const client = new ParseRestClient({
+    serverUrl: config.parse.serverUrl,
+    appId: config.parse.appId,
+    masterKey: config.parse.masterKey,
+    timeoutMs: config.parse.timeoutMs,
+  });
+  const service = new CompetitorListingMonitorService(
+    new ParseRestCompetitorListingMonitorRepository(client),
+    new FmodeVocEcommerceClient(config.fmode),
+  );
+  return runCompetitorListingRefreshCli({ service, workspaceId, pollMs, timeoutMs });
+}
+
+function readIntegerArgument(args: string[], prefix: string, fallback: number): number {
+  const raw = args.find((argument) => argument.startsWith(prefix))?.slice(prefix.length);
+  return raw === undefined ? fallback : positiveInteger(Number(raw), prefix.slice(2, -1));
+}
+
+function positiveInteger(value: number, name: string): number {
+  if (!Number.isSafeInteger(value) || value <= 0) throw new Error(`${name} must be a positive integer`);
+  return value;
+}
+
+const entry = process.argv[1];
+if (entry && pathToFileURL(resolve(entry)).href === import.meta.url) {
+  main().then((exitCode) => {
+    process.exitCode = exitCode;
+  }).catch(() => {
+    console.error(JSON.stringify({ status: 'failed', error: 'competitor_listing_refresh_cli_failed' }));
+    process.exitCode = COMPETITOR_LISTING_REFRESH_EXIT.failed;
+  });
+}

scripts/sync-jd-listing-reviews.ts

@@ -0,0 +1,351 @@
+import 'dotenv/config';
+import { createHash } from 'node:crypto';
+import { mkdir, readFile, writeFile } from 'node:fs/promises';
+import { dirname, resolve } from 'node:path';
+import { z } from 'zod';
+import { loadConfig } from '../src/config/env.js';
+import { ParseRestClient, parseDate, type ParseObject } from '../src/db/parse-rest.client.js';
+import { VOC_PARSE_CLASSES } from '../src/db/parse-rest.schema.js';
+import { makeReviewKey } from '../src/modules/domestic-voc/domain/identity.js';
+import type { ListingSourceSnapshot } from '../src/modules/listing-ai/domain.js';
+
+type JsonRecord = Record<string, unknown>;
+interface StoredSource extends ParseObject { productId?: string; payload: ListingSourceSnapshot }
+interface StoredReview extends ParseObject { naturalKey: string; productId: string; reviewDate?: unknown }
+interface Checkpoint {
+  workspaceId: string;
+  cohort: string;
+  nextPage: number;
+  totalItems: number;
+  totalPages: number;
+  pagesProcessed: number;
+  commentsScanned: number;
+  commentsMatched: number;
+  commentsCreated: number;
+  commentsUpdated: number;
+  commentsDuplicate: number;
+  commentsUnmatched: number;
+  matchedProducts: string[];
+  startedAt: string;
+  updatedAt: string;
+  status: 'running' | 'completed';
+  phase?: 'windowed';
+  historyStart?: string;
+  historyEnd?: string;
+  windowStart?: string;
+  windowEnd?: string;
+  windowsProcessed?: number;
+}
+
+const ACTIVE_COHORT = 'listing-jd-v3-formal-625';
+const METHOD = 'jingdong.pop.PopCommentJsfService.getVenderCommentsForJos';
+const PAGE_SIZE = 50;
+const DEFAULT_HISTORY_START = '2026-06-01 00:00:00';
+const DEFAULT_HISTORY_END = '2026-08-29 00:00:00';
+const args = new Map(process.argv.slice(2).map((arg) => {
+  const [key, ...rest] = arg.split('=');
+  return [key!, rest.join('=') || 'true'];
+}));
+const workspaceId = args.get('--workspace') ?? process.env.SAAS_DEFAULT_WORKSPACE_ID ?? 'demashi';
+const checkpointPath = resolve(args.get('--checkpoint') ?? 'logs/jd-listing-review-sync-2026-06-01-to-2026-08-28.json');
+const resume = args.get('--resume') === 'true';
+const delayMs = Math.max(100, Number(args.get('--delay-ms') ?? 200));
+const maxPages = Math.max(1, Number(args.get('--max-pages') ?? Number.POSITIVE_INFINITY));
+const historyStartArg = args.get('--history-start') ?? DEFAULT_HISTORY_START;
+const historyEndArg = args.get('--history-end') ?? DEFAULT_HISTORY_END;
+const windowDays = Math.max(1, Number(args.get('--window-days') ?? 7));
+const pruneOutsideRange = args.get('--prune-outside-range') !== 'false';
+const sourceSchema = z.object({
+  JD_SOURCE_PARSE_URL: z.url(),
+  JD_SOURCE_PARSE_APP_ID: z.string().min(1),
+  JD_SOURCE_PARSE_MASTER_KEY: z.string().min(1),
+  JD_APP_KEY: z.string().min(1),
+  JD_APP_SECRET: z.string().min(1),
+});
+
+async function main(): Promise<void> {
+  const sourceConfig = sourceSchema.parse(process.env);
+  const config = loadConfig();
+  const target = new ParseRestClient({
+    serverUrl: config.parse.serverUrl,
+    appId: config.parse.appId,
+    masterKey: config.parse.masterKey,
+    timeoutMs: config.parse.timeoutMs,
+  });
+  const authSource = new ParseRestClient({
+    serverUrl: sourceConfig.JD_SOURCE_PARSE_URL,
+    appId: sourceConfig.JD_SOURCE_PARSE_APP_ID,
+    masterKey: sourceConfig.JD_SOURCE_PARSE_MASTER_KEY,
+    timeoutMs: config.jdListing.timeoutMs,
+  });
+  const authRow = (await authSource.find<JsonRecord>('EcomAuth', {
+    where: { platform: 'jd', type: 'access_token' }, order: '-createdAt', limit: 1,
+  })).results[0];
+  const authData = record(authRow?.['data']);
+  const accessToken = text(authData['access_token']);
+  if (!accessToken) throw new Error('jd_authorization_missing');
+
+  const currentSources = await target.findAll<StoredSource>(VOC_PARSE_CLASSES.listingSourceSnapshot, {
+    workspaceId, platform: 'jd', isCurrent: true, catalogIncluded: true, catalogCohort: ACTIVE_COHORT,
+  });
+  const cohortProductIds = new Set(currentSources.map((row) => row.payload.productId));
+  if (cohortProductIds.size !== 625) throw new Error(`listing_review_cohort_count_mismatch:${cohortProductIds.size}:625`);
+
+  const allSources = await target.findAll<StoredSource>(VOC_PARSE_CLASSES.listingSourceSnapshot, { workspaceId, platform: 'jd' });
+  const skuToProduct = buildSkuMap(allSources, cohortProductIds);
+  if (!skuToProduct.size) throw new Error('listing_review_sku_map_empty');
+  const conflicts = findSkuConflicts(allSources, cohortProductIds);
+  if (conflicts.length) throw new Error(`listing_review_sku_conflicts:${conflicts.slice(0, 5).join(',')}`);
+
+  let existingReviews = await target.findAll<StoredReview>(VOC_PARSE_CLASSES.review, { workspaceId, platform: 'jd' });
+  if (pruneOutsideRange) {
+    const outsideRange = existingReviews.filter((review) => cohortProductIds.has(review.productId)
+      && !isWithinJdRange(review.reviewDate, historyStartArg, historyEndArg));
+    const deleteRequests = outsideRange.map((review) => ({
+      method: 'DELETE' as const,
+      path: `/classes/${VOC_PARSE_CLASSES.review}/${review.objectId}`,
+    }));
+    for (let index = 0; index < deleteRequests.length; index += 50) {
+      await writeFmodeBatch(target, deleteRequests.slice(index, index + 50));
+    }
+    existingReviews = existingReviews.filter((review) => !outsideRange.includes(review));
+    console.log(JSON.stringify({ event: 'listing_review_prune', deleted: outsideRange.length, historyStart: historyStartArg, historyEndExclusive: historyEndArg }));
+  }
+  const existingByNaturalKey = new Map(existingReviews.map((row) => [row.naturalKey, row]));
+  const seen = new Set(existingByNaturalKey.keys());
+  let checkpoint = resume ? await readCheckpoint(checkpointPath) : null;
+  if (checkpoint?.status === 'completed') {
+    console.log(JSON.stringify({ mode: 'already_completed', checkpointPath, ...checkpoint }, null, 2));
+    return;
+  }
+  const now = new Date().toISOString();
+  checkpoint ??= {
+    workspaceId, cohort: ACTIVE_COHORT, nextPage: 1, totalItems: 0, totalPages: 0,
+    pagesProcessed: 0, commentsScanned: 0, commentsMatched: 0, commentsCreated: 0,
+    commentsUpdated: 0, commentsDuplicate: 0, commentsUnmatched: 0, matchedProducts: [],
+    startedAt: now, updatedAt: now, status: 'running',
+  };
+  if (checkpoint.workspaceId !== workspaceId || checkpoint.cohort !== ACTIVE_COHORT) {
+    throw new Error('listing_review_checkpoint_scope_mismatch');
+  }
+  if (checkpoint.phase === 'windowed'
+    && (checkpoint.historyStart !== normalizeJdDateTime(historyStartArg)
+      || checkpoint.historyEnd !== normalizeJdDateTime(historyEndArg))) {
+    throw new Error(`listing_review_checkpoint_range_mismatch:${checkpoint.historyStart}:${checkpoint.historyEnd}`);
+  }
+  const matchedProducts = new Set(checkpoint.matchedProducts);
+  const client = new JdJosReviewClient(sourceConfig.JD_APP_KEY, sourceConfig.JD_APP_SECRET, accessToken);
+  if (!checkpoint.totalItems) {
+    const metadata = await client.page(1, PAGE_SIZE);
+    checkpoint.totalItems = metadata.totalItem;
+    checkpoint.totalPages = Math.ceil(metadata.totalItem / PAGE_SIZE);
+  }
+  if (checkpoint.phase !== 'windowed') {
+    const oldestStoredReview = existingReviews
+      .filter((review) => cohortProductIds.has(review.productId))
+      .map((review) => storedDateIso(review.reviewDate))
+      .filter((value): value is string => Boolean(value))
+      .sort()[0];
+    const historyEnd = normalizeJdDateTime(historyEndArg || oldestStoredReview || jdTomorrow());
+    const historyStart = normalizeJdDateTime(historyStartArg);
+    checkpoint.phase = 'windowed';
+    checkpoint.historyStart = historyStart;
+    checkpoint.historyEnd = historyEnd;
+    checkpoint.windowStart = historyStart;
+    checkpoint.windowEnd = minJdDateTime(addJdDays(historyStart, windowDays), historyEnd);
+    checkpoint.windowsProcessed = 0;
+    checkpoint.nextPage = 1;
+    checkpoint.updatedAt = new Date().toISOString();
+    await persistCheckpoint(checkpointPath, checkpoint);
+    console.log(JSON.stringify({ event: 'listing_review_window_migration', historyStart, historyEnd, retainedRecentPages: checkpoint.pagesProcessed }));
+  }
+  let pagesThisRun = 0;
+
+  while (checkpoint.windowStart && checkpoint.windowEnd && checkpoint.historyEnd
+    && jdEpoch(checkpoint.windowStart) < jdEpoch(checkpoint.historyEnd)
+    && pagesThisRun < maxPages) {
+    const page = checkpoint.nextPage;
+    const response = await client.page(page, PAGE_SIZE, {
+      beginTime: checkpoint.windowStart,
+      endTime: checkpoint.windowEnd,
+    });
+    const windowPages = Math.ceil(response.totalItem / PAGE_SIZE);
+    if (windowPages > 200) throw new Error(`listing_review_window_too_large:${checkpoint.windowStart}:${checkpoint.windowEnd}:${response.totalItem}`);
+    if (!response.comments.length && page <= windowPages) throw new Error(`listing_review_empty_page:${page}/${windowPages}`);
+    const requests: Array<{ method: 'POST' | 'PUT'; path: string; body: unknown }> = [];
+    for (const comment of response.comments) {
+      checkpoint.commentsScanned += 1;
+      const skuId = text(comment['skuid'] ?? comment['skuId']);
+      const productId = skuToProduct.get(skuId);
+      if (!productId) { checkpoint.commentsUnmatched += 1; continue; }
+      const content = text(comment['content']);
+      if (!content) { checkpoint.commentsUnmatched += 1; continue; }
+      const reviewId = text(comment['commentId']);
+      const reviewDate = dateIso(comment['creationTime']);
+      const reviewKey = makeReviewKey({ platform: 'jd', productId, reviewId, content, reviewDate });
+      const naturalKey = [workspaceId, 'jd', reviewKey].map(encodeURIComponent).join('|');
+      const existing = existingByNaturalKey.get(naturalKey);
+      if (!existing && seen.has(naturalKey)) { checkpoint.commentsDuplicate += 1; continue; }
+      const body = {
+        naturalKey, workspaceId, platform: 'jd', productId, sourceReviewId: reviewId || null,
+        reviewKey, rating: rating(comment['score']), content,
+        reviewDate: reviewDate ? parseDate(reviewDate) : null,
+        rawPayload: sanitizeComment(comment),
+      };
+      if (existing) {
+        requests.push({ method: 'PUT', path: `/classes/${VOC_PARSE_CLASSES.review}/${existing.objectId}`, body });
+        checkpoint.commentsUpdated += 1;
+      } else {
+        requests.push({ method: 'POST', path: `/classes/${VOC_PARSE_CLASSES.review}`, body });
+        checkpoint.commentsCreated += 1;
+      }
+      seen.add(naturalKey);
+      matchedProducts.add(productId);
+      checkpoint.commentsMatched += 1;
+    }
+    for (let index = 0; index < requests.length; index += 50) {
+      await writeFmodeBatch(target, requests.slice(index, index + 50));
+    }
+    checkpoint.pagesProcessed += 1;
+    pagesThisRun += 1;
+    const windowFinished = page >= Math.max(1, windowPages);
+    if (windowFinished) {
+      checkpoint.windowsProcessed = (checkpoint.windowsProcessed ?? 0) + 1;
+      checkpoint.windowStart = checkpoint.windowEnd;
+      checkpoint.windowEnd = minJdDateTime(addJdDays(checkpoint.windowStart, windowDays), checkpoint.historyEnd);
+      checkpoint.nextPage = 1;
+    } else {
+      checkpoint.nextPage = page + 1;
+    }
+    checkpoint.matchedProducts = [...matchedProducts].sort();
+    checkpoint.updatedAt = new Date().toISOString();
+    await persistCheckpoint(checkpointPath, checkpoint);
+    if (checkpoint.pagesProcessed % 25 === 0 || windowFinished && (checkpoint.windowsProcessed ?? 0) % 25 === 0) {
+      console.log(JSON.stringify({ event: 'listing_review_progress', window: checkpoint.windowsProcessed, windowStart: checkpoint.windowStart, page, windowPages, scanned: checkpoint.commentsScanned, matched: checkpoint.commentsMatched, created: checkpoint.commentsCreated, updated: checkpoint.commentsUpdated, matchedProducts: matchedProducts.size }));
+    }
+    await wait(delayMs);
+  }
+  if (checkpoint.windowStart && checkpoint.historyEnd && jdEpoch(checkpoint.windowStart) >= jdEpoch(checkpoint.historyEnd)) checkpoint.status = 'completed';
+  checkpoint.updatedAt = new Date().toISOString();
+  checkpoint.matchedProducts = [...matchedProducts].sort();
+  await persistCheckpoint(checkpointPath, checkpoint);
+  console.log(JSON.stringify({
+    mode: checkpoint.status, workspaceId, cohortProducts: cohortProductIds.size, mappedSkus: skuToProduct.size,
+    totalItems: checkpoint.totalItems, totalPages: checkpoint.totalPages, pagesProcessed: checkpoint.pagesProcessed,
+    commentsScanned: checkpoint.commentsScanned, commentsMatched: checkpoint.commentsMatched,
+    commentsCreated: checkpoint.commentsCreated, commentsUpdated: checkpoint.commentsUpdated,
+    commentsDuplicate: checkpoint.commentsDuplicate, commentsUnmatched: checkpoint.commentsUnmatched,
+    matchedProducts: matchedProducts.size, checkpointPath,
+  }, null, 2));
+}
+
+class JdJosReviewClient {
+  constructor(private readonly appKey: string, private readonly appSecret: string, private readonly accessToken: string) {}
+  async page(page: number, pageSize: number, filters: JsonRecord = {}): Promise<{ totalItem: number; comments: JsonRecord[] }> {
+    for (let attempt = 0; attempt < 6; attempt += 1) {
+      try {
+        const params: Record<string, string> = {
+          method: METHOD, access_token: this.accessToken, app_key: this.appKey,
+          timestamp: jdTime(), v: '2.0', sign_method: 'md5',
+          '360buy_param_json': JSON.stringify({ page, pageSize, ...filters }),
+        };
+        const plain = Object.keys(params).sort().map((key) => `${key}${params[key] ?? ''}`).join('');
+        params['sign'] = createHash('md5').update(`${this.appSecret}${plain}${this.appSecret}`).digest('hex').toUpperCase();
+        const response = await fetch(`https://api.jd.com/routerjson?${new URLSearchParams(params)}`, { signal: AbortSignal.timeout(30_000) });
+        const body = await response.json() as JsonRecord;
+        if (!response.ok) throw new Error(`jd_review_http_${response.status}`);
+        const root = record(body[Object.keys(body)[0] ?? '']);
+        if (text(root['code']) !== '0' || text(root['resultCode']) !== '200') {
+          const detail = text(root['resultMessage'] ?? root['resultMsg'] ?? root['message'] ?? root['msg'] ?? root['errorMessage']);
+          const diagnostic = JSON.stringify(Object.fromEntries(Object.entries(root).filter(([key]) => key !== 'comments'))).slice(0, 1_000);
+          throw new Error(`jd_review_api_${text(root['code'])}_${text(root['resultCode'])}${detail ? `:${detail}` : ''}:${diagnostic}`);
+        }
+        return { totalItem: Math.max(0, Number(root['totalItem']) || 0), comments: list(root['comments']).map(record) };
+      } catch (error) {
+        if (attempt >= 5) throw error;
+        await wait(Math.min(8_000, 500 * 2 ** attempt));
+      }
+    }
+    throw new Error('jd_review_retry_exhausted');
+  }
+}
+
+function buildSkuMap(rows: StoredSource[], cohortProductIds: Set<string>): Map<string, string> {
+  const output = new Map<string, string>();
+  for (const row of rows) {
+    const source = row.payload;
+    if (!cohortProductIds.has(source.productId)) continue;
+    output.set(source.productId, source.productId);
+    for (const sku of source.skus ?? []) if (sku.skuId) output.set(String(sku.skuId), source.productId);
+  }
+  return output;
+}
+
+function findSkuConflicts(rows: StoredSource[], cohortProductIds: Set<string>): string[] {
+  const owners = new Map<string, string>(); const conflicts = new Set<string>();
+  for (const row of rows) {
+    const source = row.payload; if (!cohortProductIds.has(source.productId)) continue;
+    for (const sku of source.skus ?? []) {
+      const skuId = String(sku.skuId || ''); if (!skuId) continue;
+      const previous = owners.get(skuId); if (previous && previous !== source.productId) conflicts.add(skuId); else owners.set(skuId, source.productId);
+    }
+  }
+  return [...conflicts];
+}
+
+function sanitizeComment(comment: JsonRecord): JsonRecord {
+  const keys = ['commentId', 'creationTime', 'content', 'skuName', 'score', 'skuid', 'images', 'isVenderReply', 'replyCount', 'usefulCount', 'skuImage', 'status', 'videos'];
+  return Object.fromEntries(keys.filter((key) => comment[key] !== undefined).map((key) => [key, comment[key]]));
+}
+async function writeFmodeBatch(
+  client: ParseRestClient,
+  requests: Array<{ method: 'POST' | 'PUT' | 'DELETE'; path: string; body?: unknown }>,
+): Promise<void> {
+  // Fmode exposes Parse at /backend/{appId}/data, but its batch router expects
+  // nested request paths to start at /data rather than repeat the full mount.
+  const results = await client.request<Array<{ error?: { code?: number; error?: string } }>>('/batch', {
+    method: 'POST',
+    body: {
+      requests: requests.map((request) => ({ ...request, path: `/data${request.path}` })),
+    },
+  });
+  const failure = results.find((result) => result.error)?.error;
+  if (failure) throw new Error(`listing_review_batch_${failure.code ?? 'unknown'}:${failure.error ?? 'failed'}`);
+}
+function storedDateIso(value: unknown): string | null {
+  if (typeof value === 'string') return dateIso(value);
+  const iso = text(record(value)['iso']);
+  return iso ? dateIso(iso) : null;
+}
+function isWithinJdRange(value: unknown, start: string, endExclusive: string): boolean {
+  const iso = storedDateIso(value);
+  if (!iso) return false;
+  const epoch = new Date(iso).valueOf();
+  return epoch >= jdEpoch(normalizeJdDateTime(start)) && epoch < jdEpoch(normalizeJdDateTime(endExclusive));
+}
+function normalizeJdDateTime(value: string): string {
+  if (/^\d{4}-\d{2}-\d{2} \d{2}:\d{2}:\d{2}$/.test(value)) return value;
+  const date = new Date(value);
+  if (Number.isNaN(date.valueOf())) throw new Error(`listing_review_invalid_date:${value}`);
+  return formatJdDateTime(date);
+}
+function jdEpoch(value: string): number {
+  const epoch = new Date(`${value.replace(' ', 'T')}+08:00`).valueOf();
+  if (Number.isNaN(epoch)) throw new Error(`listing_review_invalid_jd_date:${value}`);
+  return epoch;
+}
+function addJdDays(value: string, days: number): string { return formatJdDateTime(new Date(jdEpoch(value) + days * 86_400_000)); }
+function minJdDateTime(left: string, right: string): string { return jdEpoch(left) <= jdEpoch(right) ? left : right; }
+function jdTomorrow(): string { return `${formatJdDateTime(new Date(Date.now() + 86_400_000)).slice(0, 10)} 00:00:00`; }
+function rating(value: unknown): number { const parsed = Number(value); return Number.isFinite(parsed) && parsed >= 0 && parsed <= 5 ? parsed : 0; }
+function dateIso(value: unknown): string | null { const parsed = Number(value); const date = Number.isFinite(parsed) ? new Date(parsed < 10_000_000_000 ? parsed * 1_000 : parsed) : new Date(String(value ?? '')); return Number.isNaN(date.valueOf()) ? null : date.toISOString(); }
+function record(value: unknown): JsonRecord { return value && typeof value === 'object' && !Array.isArray(value) ? value as JsonRecord : {}; }
+function list(value: unknown): unknown[] { return Array.isArray(value) ? value : []; }
+function text(value: unknown): string { return value === null || value === undefined ? '' : String(value).trim(); }
+function wait(milliseconds: number): Promise<void> { return new Promise((resolve) => setTimeout(resolve, milliseconds)); }
+function formatJdDateTime(date: Date): string { const parts = new Intl.DateTimeFormat('sv-SE', { timeZone: 'Asia/Shanghai', year: 'numeric', month: '2-digit', day: '2-digit', hour: '2-digit', minute: '2-digit', second: '2-digit', hourCycle: 'h23' }).formatToParts(date); const get = (type: Intl.DateTimeFormatPartTypes) => parts.find((part) => part.type === type)?.value ?? ''; return `${get('year')}-${get('month')}-${get('day')} ${get('hour')}:${get('minute')}:${get('second')}`; }
+function jdTime(): string { return formatJdDateTime(new Date()); }
+async function readCheckpoint(path: string): Promise<Checkpoint | null> { try { return JSON.parse(await readFile(path, 'utf8')) as Checkpoint; } catch { return null; } }
+async function persistCheckpoint(path: string, checkpoint: Checkpoint): Promise<void> { await mkdir(dirname(path), { recursive: true }); await writeFile(path, JSON.stringify(checkpoint, null, 2)); }
+
+main().catch((error) => { console.error(`[sync-jd-listing-reviews] ${error instanceof Error ? error.message : error}`); process.exitCode = 1; });

src/app.ts

@@ -21,6 +21,8 @@ import { InMemoryListingAiRepository } from './modules/listing-ai/repositories/i
 import { FmodeListingAiScoringProvider, ListingAiService } from './modules/listing-ai/listing-ai.service.js';
 import { createListingAiRouter } from './modules/listing-ai/routes.js';
 import type { ListingAiRepository } from './modules/listing-ai/domain.js';
+import type { CompetitorListingMonitorService } from './modules/competitor-listing-monitor/competitor-listing-monitor.service.js';
+import { createCompetitorListingMonitorRouter } from './modules/competitor-listing-monitor/routes.js';
 
 export function createApp(input: {
   config: AppConfig;
@@ -33,6 +35,7 @@ export function createApp(input: {
   aiPromptConfigs?: AiPromptConfigStore;
   productKnowledge?: ProductKnowledgeStore;
   listingAiRepository?: ListingAiRepository;
+  competitorListingMonitor?: CompetitorListingMonitorService;
 }) {
   const app = express();
   app.disable('x-powered-by');
@@ -164,6 +167,13 @@ export function createApp(input: {
       defaultWorkspaceId: input.config.auth.defaultWorkspaceId,
     }));
   }
+  if (input.competitorListingMonitor) {
+    app.use('/api/competitor-listings', createCompetitorListingMonitorRouter({
+      service: input.competitorListingMonitor,
+      access,
+      defaultWorkspaceId: input.config.auth.defaultWorkspaceId,
+    }));
+  }
 
   app.use((_request, response) => {
     response.status(404).json({ error: 'not_found' });

src/db/parse-rest.schema.ts

@@ -34,6 +34,9 @@ export const VOC_PARSE_CLASSES = {
   listingScoreItem: 'VocListingScoreItem',
   listingCurrentScore: 'VocListingCurrentScore',
   listingVersion: 'VocListingVersion',
+  competitorListingSnapshot: 'VocCompetitorListingSnapshot',
+  competitorListingChange: 'VocCompetitorListingChange',
+  competitorListingRefreshRun: 'VocCompetitorListingRefreshRun',
 } as const;
 
 export const VOC_PARSE_SCHEMAS: ParseClassSchema[] = [
@@ -292,6 +295,62 @@ export const VOC_PARSE_SCHEMAS: ParseClassSchema[] = [
     },
     indexes: indexes('voc_listing_version', 'publicId', 'naturalKey', 'workspaceId', 'productId', 'status', 'versionCreatedAt'),
   },
+  {
+    className: VOC_PARSE_CLASSES.competitorListingSnapshot,
+    fields: {
+      publicId: string(true), naturalKey: string(true), workspaceId: string(true), platform: string(true),
+      productId: string(true), previousSnapshotId: string(), contentHash: string(true), observedAt: date(true),
+      title: string(), priceCents: number(), currency: string(true), availability: string(true), mainImageUrl: string(),
+      keySpecifications: object(true), observedFields: array(true), collectionStatus: string(true),
+    },
+    indexes: indexes(
+      'voc_competitor_listing_snapshot',
+      'publicId',
+      'naturalKey',
+      'workspaceId',
+      'platform',
+      'productId',
+      'previousSnapshotId',
+      'contentHash',
+      'observedAt',
+    ),
+  },
+  {
+    className: VOC_PARSE_CLASSES.competitorListingChange,
+    fields: {
+      publicId: string(true), naturalKey: string(true), workspaceId: string(true), platform: string(true),
+      productId: string(true), previousSnapshotId: string(true), currentSnapshotId: string(true),
+      detectedAt: date(true), changeTypes: array(true), changes: array(true),
+    },
+    indexes: indexes(
+      'voc_competitor_listing_change',
+      'publicId',
+      'naturalKey',
+      'workspaceId',
+      'platform',
+      'productId',
+      'previousSnapshotId',
+      'currentSnapshotId',
+      'detectedAt',
+    ),
+  },
+  {
+    className: VOC_PARSE_CLASSES.competitorListingRefreshRun,
+    fields: {
+      publicId: string(true), workspaceId: string(true), platform: string(true), trigger: string(true),
+      status: string(true), total: number(true), completed: number(true), baseline: number(true),
+      unchanged: number(true), changed: number(true), failed: number(true), itemResults: array(true),
+      requestedAt: date(true), startedAt: date(), completedAt: date(),
+    },
+    indexes: indexes(
+      'voc_competitor_listing_refresh_run',
+      'publicId',
+      'workspaceId',
+      'platform',
+      'status',
+      'requestedAt',
+    ),
+  },
 ];
 
 export interface ParseSchemaSyncResult {

src/modules/competitor-listing-monitor/competitor-listing-monitor.service.ts

@@ -0,0 +1,475 @@
+import { randomUUID } from 'node:crypto';
+import { ApiError } from '../../http/api-error.js';
+import { adaptJdProductProfile, adaptJdProductResponse, JD_PRODUCT_DETAIL_PATH } from '../domestic-voc/adapters/jd-product.adapter.js';
+import { adaptJdPriceResponse, JD_PRODUCT_PRICE_PATH } from '../domestic-voc/adapters/jd-price.adapter.js';
+import { adaptJdSearchResponse, JD_PRODUCT_SEARCH_PATH } from '../domestic-voc/adapters/jd-search.adapter.js';
+import type {
+  CompetitorListingChange,
+  CompetitorListingCollectionResult,
+  CompetitorListingKeySpecifications,
+  CompetitorListingObservation,
+  CompetitorListingOverview,
+  CompetitorListingOverviewRow,
+  CompetitorListingPlatform,
+  CompetitorListingRefreshItemResult,
+  CompetitorListingRefreshRun,
+  CompetitorListingRelatedProduct,
+  CompetitorListingSnapshot,
+} from './domain.js';
+import { decideCompetitorListingSnapshot } from './snapshot-diff.js';
+
+export interface CompetitorListingMonitorTarget {
+  productId: string;
+  brand: string | null;
+  title: string | null;
+  mainImageUrl: string | null;
+  keySpecifications: CompetitorListingKeySpecifications;
+  category: string | null;
+  relatedProducts: CompetitorListingRelatedProduct[];
+}
+
+export interface CompetitorListingMonitorRepository {
+  listTargets(workspaceId: string, platform: CompetitorListingPlatform): Promise<CompetitorListingMonitorTarget[]>;
+  getLatestSnapshot(workspaceId: string, platform: CompetitorListingPlatform, productId: string): Promise<CompetitorListingSnapshot | null>;
+  listSnapshots(workspaceId: string, platform: CompetitorListingPlatform): Promise<CompetitorListingSnapshot[]>;
+  saveSnapshot(snapshot: CompetitorListingSnapshot): Promise<CompetitorListingSnapshot>;
+  listChanges(workspaceId: string, platform: CompetitorListingPlatform): Promise<CompetitorListingChange[]>;
+  saveChange(change: CompetitorListingChange): Promise<CompetitorListingChange>;
+  findActiveRun(workspaceId: string, platform: CompetitorListingPlatform): Promise<CompetitorListingRefreshRun | null>;
+  createRun(run: CompetitorListingRefreshRun): Promise<CompetitorListingRefreshRun>;
+  updateRun(run: CompetitorListingRefreshRun): Promise<CompetitorListingRefreshRun>;
+  getRun(workspaceId: string, runId: string): Promise<CompetitorListingRefreshRun | null>;
+  listRuns(workspaceId: string, platform: CompetitorListingPlatform): Promise<CompetitorListingRefreshRun[]>;
+}
+
+export interface CompetitorListingGateway {
+  request<T>(
+    path: string,
+    init?: { method?: 'GET' | 'POST'; params?: Record<string, unknown>; refresh?: boolean },
+  ): Promise<T>;
+}
+
+interface CollectedItem {
+  collection: CompetitorListingCollectionResult;
+  partialErrorCode?: string;
+}
+
+const EMPTY_SPECIFICATIONS: CompetitorListingKeySpecifications = {
+  model: null,
+  color: null,
+  specification: null,
+  origin: null,
+  weightKg: null,
+  lengthMm: null,
+  widthMm: null,
+  heightMm: null,
+};
+
+export class CompetitorListingMonitorService {
+  private readonly activeWorkspaceKeys = new Set<string>();
+
+  constructor(
+    readonly repository: CompetitorListingMonitorRepository,
+    private readonly gateway: CompetitorListingGateway,
+    private readonly now: () => Date = () => new Date(),
+    private readonly concurrency = 3,
+  ) {}
+
+  async overview(workspaceId: string, platform: CompetitorListingPlatform): Promise<CompetitorListingOverview> {
+    const [targets, snapshots, changes, runs] = await Promise.all([
+      this.repository.listTargets(workspaceId, platform),
+      this.repository.listSnapshots(workspaceId, platform),
+      this.repository.listChanges(workspaceId, platform),
+      this.repository.listRuns(workspaceId, platform),
+    ]);
+    const snapshotsByProduct = groupByProduct(snapshots);
+    const changesByProduct = groupByProduct(changes);
+    const latestRun = runs.toSorted((left, right) => right.requestedAt.localeCompare(left.requestedAt))[0] ?? null;
+    const latestSuccessfulRun = runs
+      .filter((run) => run.status === 'completed' || run.status === 'partial')
+      .toSorted((left, right) => right.requestedAt.localeCompare(left.requestedAt))[0] ?? null;
+    const latestResultByProduct = new Map(
+      (latestRun?.itemResults ?? []).map((result) => [result.productId, result]),
+    );
+    const sevenDaysAgo = this.now().getTime() - 7 * 24 * 60 * 60 * 1_000;
+    const recentChanges = changes.filter((change) => new Date(change.detectedAt).getTime() >= sevenDaysAgo);
+    const comparedTotal = targets.filter((target) => (snapshotsByProduct.get(target.productId)?.length ?? 0) >= 2).length;
+    const baselineTotal = targets.filter((target) => (snapshotsByProduct.get(target.productId)?.length ?? 0) >= 1).length;
+    const historyStatus = baselineTotal === 0 ? 'not_initialized' : comparedTotal === 0 ? 'baseline_only' : 'comparable';
+
+    const items = targets.map<CompetitorListingOverviewRow>((target) => {
+      const productSnapshots = snapshotsByProduct.get(target.productId) ?? [];
+      const currentSnapshot = productSnapshots.toSorted(compareObservedAtDescending)[0] ?? null;
+      const latestChange = (changesByProduct.get(target.productId) ?? [])
+        .toSorted((left, right) => right.detectedAt.localeCompare(left.detectedAt))[0] ?? null;
+      const latestCollectionResult = latestResultByProduct.get(target.productId) ?? null;
+      const monitorStatus = latestCollectionResult?.status === 'failed'
+        ? 'failed'
+        : !currentSnapshot
+          ? 'not_initialized'
+          : productSnapshots.length < 2
+            ? 'awaiting_comparison'
+            : latestChange?.currentSnapshotId === currentSnapshot.id
+              ? 'changed'
+              : 'unchanged';
+      return {
+        productId: target.productId,
+        platform,
+        title: currentSnapshot?.title ?? target.title,
+        brand: target.brand,
+        mainImageUrl: currentSnapshot?.mainImageUrl ?? target.mainImageUrl,
+        category: target.category,
+        relatedProducts: target.relatedProducts,
+        currentSnapshot,
+        latestChange,
+        latestCollectionResult,
+        monitorStatus,
+      };
+    });
+
+    return {
+      summary: {
+        targetTotal: targets.length,
+        baselineTotal,
+        comparedTotal,
+        changedIn7d: historyStatus === 'comparable' ? new Set(recentChanges.map((change) => change.productId)).size : null,
+        priceChangedIn7d: historyStatus === 'comparable'
+          ? recentChanges.filter((change) => change.changeTypes.includes('price')).length
+          : null,
+        contentChangedIn7d: historyStatus === 'comparable'
+          ? recentChanges.filter((change) => change.changeTypes.some((type) => type !== 'price')).length
+          : null,
+        latestFailedTotal: latestRun?.failed ?? 0,
+        lastRefreshAt: latestRun ? latestRun.completedAt ?? latestRun.startedAt ?? latestRun.requestedAt : null,
+        lastSuccessfulRefreshAt: latestSuccessfulRun
+          ? latestSuccessfulRun.completedAt ?? latestSuccessfulRun.startedAt ?? latestSuccessfulRun.requestedAt
+          : null,
+        historyStatus,
+      },
+      items,
+      facets: {
+        brands: facet(targets.map((target) => target.brand)),
+        categories: facet(targets.map((target) => target.category)),
+      },
+    };
+  }
+
+  async startRefresh(
+    workspaceId: string,
+    platform: CompetitorListingPlatform,
+    trigger: CompetitorListingRefreshRun['trigger'] = 'manual',
+  ): Promise<CompetitorListingRefreshRun> {
+    const workspaceKey = `${workspaceId}:${platform}`;
+    if (this.activeWorkspaceKeys.has(workspaceKey)) throw new ApiError(409, 'competitor_listing_refresh_running');
+    this.activeWorkspaceKeys.add(workspaceKey);
+    try {
+      const persistedActiveRun = await this.repository.findActiveRun(workspaceId, platform);
+      if (persistedActiveRun) throw new ApiError(409, 'competitor_listing_refresh_running');
+      const targets = await this.repository.listTargets(workspaceId, platform);
+      const requestedAt = this.now().toISOString();
+      const run = await this.repository.createRun({
+        id: randomUUID(),
+        workspaceId,
+        platform,
+        trigger,
+        status: 'queued',
+        total: targets.length,
+        completed: 0,
+        baseline: 0,
+        unchanged: 0,
+        changed: 0,
+        failed: 0,
+        itemResults: [],
+        requestedAt,
+        startedAt: null,
+        completedAt: null,
+      });
+      queueMicrotask(() => {
+        void this.processRefresh(run, targets, workspaceKey);
+      });
+      return run;
+    } catch (error) {
+      this.activeWorkspaceKeys.delete(workspaceKey);
+      throw error;
+    }
+  }
+
+  async getRun(workspaceId: string, runId: string): Promise<CompetitorListingRefreshRun | null> {
+    return this.repository.getRun(workspaceId, runId);
+  }
+
+  private async processRefresh(
+    initialRun: CompetitorListingRefreshRun,
+    targets: CompetitorListingMonitorTarget[],
+    workspaceKey: string,
+  ): Promise<void> {
+    let run: CompetitorListingRefreshRun = {
+      ...initialRun,
+      status: 'running',
+      startedAt: this.now().toISOString(),
+    };
+    let partialCollections = 0;
+    try {
+      run = await this.repository.updateRun(run);
+      let cursor = 0;
+      let progressWrite = Promise.resolve();
+      const worker = async (): Promise<void> => {
+        for (;;) {
+          const target = targets[cursor];
+          cursor += 1;
+          if (!target) return;
+          const processed = await this.processTarget(initialRun.workspaceId, initialRun.platform, target);
+          if (processed.partial) partialCollections += 1;
+          run = applyItemResult(run, processed.result);
+          const progress = run;
+          progressWrite = progressWrite.then(async () => {
+            await this.repository.updateRun(progress);
+          });
+          await progressWrite;
+        }
+      };
+      const workerCount = Math.min(targets.length, Math.max(1, Math.floor(this.concurrency)));
+      await Promise.all(Array.from({ length: workerCount }, () => worker()));
+      run = {
+        ...run,
+        status: run.failed === run.total && run.total > 0
+          ? 'failed'
+          : run.failed > 0 || partialCollections > 0
+            ? 'partial'
+            : 'completed',
+        completedAt: this.now().toISOString(),
+      };
+      await this.repository.updateRun(run);
+    } catch {
+      const failedRun: CompetitorListingRefreshRun = {
+        ...run,
+        status: 'failed',
+        completedAt: this.now().toISOString(),
+      };
+      await this.persistTerminalFailure(failedRun);
+    } finally {
+      this.activeWorkspaceKeys.delete(workspaceKey);
+    }
+  }
+
+  private async persistTerminalFailure(run: CompetitorListingRefreshRun): Promise<void> {
+    for (let attempt = 0; attempt < 3; attempt += 1) {
+      try {
+        await this.repository.updateRun(run);
+        return;
+      } catch {
+        if (attempt < 2) await new Promise((resolve) => setTimeout(resolve, 25 * (attempt + 1)));
+      }
+    }
+  }
+
+  private async processTarget(
+    workspaceId: string,
+    platform: CompetitorListingPlatform,
+    target: CompetitorListingMonitorTarget,
+  ): Promise<{ result: CompetitorListingRefreshItemResult; partial: boolean }> {
+    const observedAt = this.now().toISOString();
+    const collected = await this.collectTarget(workspaceId, platform, target, observedAt);
+    const previous = await this.repository.getLatestSnapshot(workspaceId, platform, target.productId);
+    const decision = decideCompetitorListingSnapshot(previous, collected.collection, {
+      snapshotId: randomUUID(),
+      changeId: randomUUID(),
+    });
+    if (decision.status === 'failed') {
+      return {
+        result: {
+          productId: target.productId,
+          status: 'failed',
+          errorCode: decision.errorCode ?? 'jd_detail_failed',
+          observedAt,
+        },
+        partial: false,
+      };
+    }
+
+    let savedSnapshot = decision.snapshot;
+    if (savedSnapshot) savedSnapshot = await this.repository.saveSnapshot(savedSnapshot);
+    if (decision.status === 'changed') {
+      if (!savedSnapshot) throw new Error('A changed listing decision must include a snapshot');
+      const change = savedSnapshot.id === decision.change.currentSnapshotId
+        ? decision.change
+        : {
+            ...decision.change,
+            naturalKey: `${decision.change.previousSnapshotId}:${savedSnapshot.id}`,
+            currentSnapshotId: savedSnapshot.id,
+          };
+      await this.repository.saveChange(change);
+    }
+    const status = decision.status === 'baseline' ? 'baseline' : decision.status;
+    const result: CompetitorListingRefreshItemResult = {
+      productId: target.productId,
+      status,
+      observedAt,
+      ...(collected.partialErrorCode ? { errorCode: collected.partialErrorCode } : {}),
+    };
+    return { result, partial: collected.collection.collectionStatus === 'partial' };
+  }
+
+  private async collectTarget(
+    workspaceId: string,
+    platform: CompetitorListingPlatform,
+    target: CompetitorListingMonitorTarget,
+    observedAt: string,
+  ): Promise<CollectedItem> {
+    let product;
+    try {
+      const detailResponse = await this.gateway.request<unknown>(JD_PRODUCT_DETAIL_PATH, {
+        params: { itemId: target.productId },
+        refresh: true,
+      });
+      product = adaptJdProductResponse(detailResponse, target.productId, 'competitor');
+      const returnedProductId = detailPayloadProductId(product.rawPayload);
+      if (!product.rawPayload || returnedProductId !== target.productId || product.productId !== target.productId) {
+        return {
+          collection: failure(workspaceId, platform, target.productId, observedAt, 'jd_detail_product_mismatch'),
+        };
+      }
+    } catch {
+      return {
+        collection: failure(workspaceId, platform, target.productId, observedAt, 'jd_detail_failed'),
+      };
+    }
+
+    const profile = adaptJdProductProfile(product.rawPayload, observedAt);
+    let priceCents: number | null = null;
+    let partialErrorCode: string | undefined;
+    try {
+      const priceResponse = await this.gateway.request<unknown>(JD_PRODUCT_PRICE_PATH, {
+        params: { itemId: target.productId },
+        refresh: true,
+      });
+      priceCents = adaptJdPriceResponse(priceResponse, target.productId);
+    } catch {
+      // The search contract remains a compatibility fallback while the dedicated
+      // endpoint is rolled out across all gateway accounts.
+    }
+    if (priceCents === null) {
+      try {
+        const searchResponse = await this.gateway.request<unknown>(JD_PRODUCT_SEARCH_PATH, {
+          params: { keyword: target.productId, page: 1 },
+          refresh: true,
+        });
+        const exactProduct = adaptJdSearchResponse(searchResponse, {
+          brand: product.brand || target.brand || '',
+          category: product.category3 || target.category || '',
+          keyword: target.productId,
+          collectedAt: observedAt,
+        }).find((candidate) => candidate.productId === target.productId);
+        const price = exactProduct?.market?.currentPrice;
+        if (price !== undefined && Number.isFinite(price) && price > 0) {
+          priceCents = Math.round(price * 100);
+        } else {
+          partialErrorCode = 'jd_price_not_found';
+        }
+      } catch {
+        partialErrorCode = 'jd_price_failed';
+      }
+    }
+
+    const availability = explicitAvailability(profile?.skuStatus ?? '');
+    const keySpecifications: CompetitorListingKeySpecifications = {
+      ...EMPTY_SPECIFICATIONS,
+      model: product.model || null,
+      color: profile?.color || null,
+      specification: profile?.specification || null,
+      origin: profile?.origin || null,
+      weightKg: profile?.weightKg || null,
+      lengthMm: profile?.dimensionsMm.length || null,
+      widthMm: profile?.dimensionsMm.width || null,
+      heightMm: profile?.dimensionsMm.height || null,
+    };
+    const title = product.title || null;
+    const mainImageUrl = profile?.imageUrl || null;
+    const observedFields: CompetitorListingObservation['observedFields'] = [];
+    if (title) observedFields.push('title');
+    if (priceCents !== null) observedFields.push('price');
+    if (availability !== 'unknown') observedFields.push('availability');
+    if (mainImageUrl) observedFields.push('main_image');
+    if (Object.values(keySpecifications).some((value) => value !== null)) observedFields.push('key_specifications');
+    return {
+      collection: {
+        workspaceId,
+        platform,
+        productId: target.productId,
+        observedAt,
+        title,
+        priceCents,
+        availability,
+        mainImageUrl,
+        keySpecifications,
+        observedFields,
+        collectionStatus: partialErrorCode ? 'partial' : 'succeeded',
+      },
+      ...(partialErrorCode ? { partialErrorCode } : {}),
+    };
+  }
+}
+
+function failure(
+  workspaceId: string,
+  platform: CompetitorListingPlatform,
+  productId: string,
+  observedAt: string,
+  errorCode: string,
+): CompetitorListingCollectionResult {
+  return { workspaceId, platform, productId, observedAt, collectionStatus: 'failed', errorCode };
+}
+
+function explicitAvailability(value: string): CompetitorListingObservation['availability'] {
+  const normalized = value.normalize('NFKC').trim().toLowerCase();
+  if (['1', 'on', 'on_shelf', 'online', 'available', 'active', '在售', '上架'].includes(normalized)) return 'online';
+  if (['0', 'off', 'off_shelf', 'offline', 'unavailable', 'inactive', '下架'].includes(normalized)) return 'offline';
+  return 'unknown';
+}
+
+function detailPayloadProductId(payload: Record<string, unknown> | null): string {
+  if (!payload) return '';
+  for (const key of ['itemId', 'skuId', 'productId', 'id']) {
+    const value = payload[key];
+    if (typeof value === 'string' || typeof value === 'number') {
+      const normalized = String(value).trim();
+      if (normalized) return normalized;
+    }
+  }
+  return '';
+}
+
+function applyItemResult(
+  run: CompetitorListingRefreshRun,
+  item: CompetitorListingRefreshItemResult,
+): CompetitorListingRefreshRun {
+  return {
+    ...run,
+    completed: run.completed + 1,
+    baseline: run.baseline + Number(item.status === 'baseline'),
+    unchanged: run.unchanged + Number(item.status === 'unchanged'),
+    changed: run.changed + Number(item.status === 'changed'),
+    failed: run.failed + Number(item.status === 'failed'),
+    itemResults: [...run.itemResults, item],
+  };
+}
+
+function groupByProduct<T extends { productId: string }>(items: T[]): Map<string, T[]> {
+  const output = new Map<string, T[]>();
+  for (const item of items) output.set(item.productId, [...(output.get(item.productId) ?? []), item]);
+  return output;
+}
+
+function compareObservedAtDescending(left: CompetitorListingSnapshot, right: CompetitorListingSnapshot): number {
+  return right.observedAt.localeCompare(left.observedAt) || right.id.localeCompare(left.id);
+}
+
+function facet(values: Array<string | null>): Array<{ value: string; count: number }> {
+  const counts = new Map<string, number>();
+  for (const value of values) {
+    if (value) counts.set(value, (counts.get(value) ?? 0) + 1);
+  }
+  return [...counts].map(([value, count]) => ({ value, count })).toSorted((left, right) => (
+    right.count - left.count || left.value.localeCompare(right.value, 'zh-CN')
+  ));
+}

src/modules/competitor-listing-monitor/domain.ts

@@ -0,0 +1,165 @@
+export const COMPETITOR_LISTING_OBSERVED_FIELDS = [
+  'title',
+  'price',
+  'availability',
+  'main_image',
+  'key_specifications',
+] as const;
+
+export type CompetitorListingPlatform = 'jd';
+export type CompetitorListingAvailability = 'online' | 'offline' | 'unknown';
+export type CompetitorListingObservedField = typeof COMPETITOR_LISTING_OBSERVED_FIELDS[number];
+export type CompetitorListingChangeType = CompetitorListingObservedField;
+export type CompetitorListingCollectionStatus = 'succeeded' | 'partial';
+
+export interface CompetitorListingKeySpecifications {
+  model: string | null;
+  color: string | null;
+  specification: string | null;
+  origin: string | null;
+  weightKg: string | null;
+  lengthMm: string | null;
+  widthMm: string | null;
+  heightMm: string | null;
+}
+
+export interface CompetitorListingSnapshot {
+  id: string;
+  naturalKey: string;
+  workspaceId: string;
+  platform: CompetitorListingPlatform;
+  productId: string;
+  previousSnapshotId: string | null;
+  contentHash: string;
+  observedAt: string;
+  title: string | null;
+  priceCents: number | null;
+  currency: 'CNY';
+  availability: CompetitorListingAvailability;
+  mainImageUrl: string | null;
+  keySpecifications: CompetitorListingKeySpecifications;
+  observedFields: CompetitorListingObservedField[];
+  collectionStatus: CompetitorListingCollectionStatus;
+}
+
+export interface CompetitorListingFieldChange {
+  field: string;
+  before: unknown;
+  after: unknown;
+}
+
+export interface CompetitorListingChange {
+  id: string;
+  naturalKey: string;
+  workspaceId: string;
+  platform: CompetitorListingPlatform;
+  productId: string;
+  previousSnapshotId: string;
+  currentSnapshotId: string;
+  detectedAt: string;
+  changeTypes: CompetitorListingChangeType[];
+  changes: CompetitorListingFieldChange[];
+}
+
+export type CompetitorListingRefreshRunStatus = 'queued' | 'running' | 'completed' | 'partial' | 'failed';
+export type CompetitorListingRefreshItemStatus = 'baseline' | 'unchanged' | 'changed' | 'failed';
+
+export interface CompetitorListingRefreshItemResult {
+  productId: string;
+  status: CompetitorListingRefreshItemStatus;
+  errorCode?: string;
+  observedAt: string;
+}
+
+export interface CompetitorListingRefreshRun {
+  id: string;
+  workspaceId: string;
+  platform: CompetitorListingPlatform;
+  trigger: 'manual' | 'scheduled';
+  status: CompetitorListingRefreshRunStatus;
+  total: number;
+  completed: number;
+  baseline: number;
+  unchanged: number;
+  changed: number;
+  failed: number;
+  itemResults: CompetitorListingRefreshItemResult[];
+  requestedAt: string;
+  startedAt: string | null;
+  completedAt: string | null;
+}
+
+export type CompetitorListingHistoryStatus = 'not_initialized' | 'baseline_only' | 'comparable';
+export type CompetitorListingMonitorStatus = 'not_initialized' | 'awaiting_comparison' | 'changed' | 'unchanged' | 'failed';
+
+export interface CompetitorListingOverviewSummary {
+  targetTotal: number;
+  baselineTotal: number;
+  comparedTotal: number;
+  changedIn7d: number | null;
+  priceChangedIn7d: number | null;
+  contentChangedIn7d: number | null;
+  latestFailedTotal: number;
+  lastRefreshAt: string | null;
+  lastSuccessfulRefreshAt: string | null;
+  historyStatus: CompetitorListingHistoryStatus;
+}
+
+export interface CompetitorListingRelatedProduct {
+  productId: string;
+  title: string | null;
+}
+
+export interface CompetitorListingOverviewRow {
+  productId: string;
+  platform: CompetitorListingPlatform;
+  title: string | null;
+  brand: string | null;
+  mainImageUrl: string | null;
+  category: string | null;
+  relatedProducts: CompetitorListingRelatedProduct[];
+  currentSnapshot: CompetitorListingSnapshot | null;
+  latestChange: CompetitorListingChange | null;
+  latestCollectionResult: CompetitorListingRefreshItemResult | null;
+  monitorStatus: CompetitorListingMonitorStatus;
+}
+
+export interface CompetitorListingOverview {
+  summary: CompetitorListingOverviewSummary;
+  items: CompetitorListingOverviewRow[];
+  facets: {
+    brands: Array<{ value: string; count: number }>;
+    categories: Array<{ value: string; count: number }>;
+  };
+}
+
+export interface CompetitorListingObservation {
+  workspaceId: string;
+  platform: CompetitorListingPlatform;
+  productId: string;
+  observedAt: string;
+  title: string | null;
+  priceCents: number | null;
+  availability: CompetitorListingAvailability;
+  mainImageUrl: string | null;
+  keySpecifications: CompetitorListingKeySpecifications;
+  observedFields: CompetitorListingObservedField[];
+  collectionStatus: CompetitorListingCollectionStatus;
+}
+
+export interface CompetitorListingCollectionFailure {
+  workspaceId: string;
+  platform: CompetitorListingPlatform;
+  productId: string;
+  observedAt: string;
+  collectionStatus: 'failed';
+  errorCode?: string;
+}
+
+export type CompetitorListingCollectionResult = CompetitorListingObservation | CompetitorListingCollectionFailure;
+
+export type CompetitorListingSnapshotDecision =
+  | { status: 'failed'; snapshot: null; change: null; errorCode?: string }
+  | { status: 'baseline'; snapshot: CompetitorListingSnapshot; change: null }
+  | { status: 'unchanged'; snapshot: CompetitorListingSnapshot | null; change: null }
+  | { status: 'changed'; snapshot: CompetitorListingSnapshot; change: CompetitorListingChange };

src/modules/competitor-listing-monitor/repositories/parse-rest-competitor-listing-monitor.repository.ts

@@ -0,0 +1,308 @@
+import { parseDate, parseDateIso, type ParseObject, ParseRestClient } from '../../../db/parse-rest.client.js';
+import { VOC_PARSE_CLASSES } from '../../../db/parse-rest.schema.js';
+import { adaptJdProductProfile } from '../../domestic-voc/adapters/jd-product.adapter.js';
+import type {
+  CompetitorListingChange,
+  CompetitorListingKeySpecifications,
+  CompetitorListingPlatform,
+  CompetitorListingRefreshRun,
+  CompetitorListingSnapshot,
+} from '../domain.js';
+import type {
+  CompetitorListingMonitorRepository,
+  CompetitorListingMonitorTarget,
+} from '../competitor-listing-monitor.service.js';
+
+interface RelationObject {
+  competitorProductId: string;
+  competitorBrand?: string;
+  category?: string;
+  ownProductId: string;
+}
+
+interface ProductObject {
+  productId: string;
+  role: 'own' | 'competitor';
+  brand?: string;
+  title?: string;
+  model?: string;
+  rawPayload?: Record<string, unknown> | null;
+  updatedAt?: string;
+}
+
+type StoredSnapshot = Omit<CompetitorListingSnapshot, 'id' | 'observedAt'> & {
+  publicId: string;
+  observedAt: unknown;
+};
+
+type StoredChange = Omit<CompetitorListingChange, 'id' | 'detectedAt'> & {
+  publicId: string;
+  detectedAt: unknown;
+};
+
+type StoredRun = Omit<CompetitorListingRefreshRun, 'id' | 'requestedAt' | 'startedAt' | 'completedAt'> & {
+  publicId: string;
+  requestedAt: unknown;
+  startedAt?: unknown;
+  completedAt?: unknown;
+};
+
+const EMPTY_SPECIFICATIONS: CompetitorListingKeySpecifications = {
+  model: null,
+  color: null,
+  specification: null,
+  origin: null,
+  weightKg: null,
+  lengthMm: null,
+  widthMm: null,
+  heightMm: null,
+};
+
+export class ParseRestCompetitorListingMonitorRepository implements CompetitorListingMonitorRepository {
+  constructor(private readonly client: ParseRestClient) {}
+
+  async listTargets(
+    workspaceId: string,
+    platform: CompetitorListingPlatform,
+  ): Promise<CompetitorListingMonitorTarget[]> {
+    const [relations, products] = await Promise.all([
+      this.client.findAll<RelationObject>(VOC_PARSE_CLASSES.productRelation, { workspaceId, platform }),
+      this.client.findAll<ProductObject>(VOC_PARSE_CLASSES.product, { workspaceId, platform }),
+    ]);
+    const productById = new Map(products.map((product) => [product.productId, product]));
+    const relationsByCompetitor = new Map<string, RelationObject[]>();
+    for (const relation of relations) {
+      const productId = relation.competitorProductId?.trim();
+      if (!productId) continue;
+      relationsByCompetitor.set(productId, [...(relationsByCompetitor.get(productId) ?? []), relation]);
+    }
+    return [...relationsByCompetitor].map(([productId, productRelations]) => {
+      const product = productById.get(productId);
+      const profile = adaptJdProductProfile(product?.rawPayload, product?.updatedAt ?? '');
+      const firstRelation = productRelations[0];
+      const relatedProducts = [...new Set(productRelations.map((relation) => relation.ownProductId).filter(Boolean))]
+        .map((ownProductId) => ({
+          productId: ownProductId,
+          title: productById.get(ownProductId)?.title?.trim() || null,
+        }));
+      return {
+        productId,
+        brand: product?.brand?.trim() || firstRelation?.competitorBrand?.trim() || null,
+        title: product?.title?.trim() || null,
+        mainImageUrl: profile?.imageUrl || null,
+        keySpecifications: {
+          ...EMPTY_SPECIFICATIONS,
+          model: product?.model?.trim() || null,
+          color: profile?.color || null,
+          specification: profile?.specification || null,
+          origin: profile?.origin || null,
+          weightKg: profile?.weightKg || null,
+          lengthMm: profile?.dimensionsMm.length || null,
+          widthMm: profile?.dimensionsMm.width || null,
+          heightMm: profile?.dimensionsMm.height || null,
+        },
+        category: productRelations.map((relation) => relation.category?.trim()).find(Boolean) || null,
+        relatedProducts,
+      };
+    }).toSorted((left, right) => left.productId.localeCompare(right.productId));
+  }
+
+  async getLatestSnapshot(
+    workspaceId: string,
+    platform: CompetitorListingPlatform,
+    productId: string,
+  ): Promise<CompetitorListingSnapshot | null> {
+    const response = await this.client.find<StoredSnapshot>(VOC_PARSE_CLASSES.competitorListingSnapshot, {
+      where: { workspaceId, platform, productId },
+      order: '-observedAt,-createdAt',
+      limit: 1,
+    });
+    const row = response.results[0];
+    return row ? mapSnapshot(row) : null;
+  }
+
+  async listSnapshots(workspaceId: string, platform: CompetitorListingPlatform): Promise<CompetitorListingSnapshot[]> {
+    return (await this.client.findAll<StoredSnapshot>(
+      VOC_PARSE_CLASSES.competitorListingSnapshot,
+      { workspaceId, platform },
+    )).map(mapSnapshot);
+  }
+
+  async saveSnapshot(snapshot: CompetitorListingSnapshot): Promise<CompetitorListingSnapshot> {
+    const existing = await this.client.findOne<StoredSnapshot>(VOC_PARSE_CLASSES.competitorListingSnapshot, {
+      naturalKey: snapshot.naturalKey,
+    });
+    if (existing) return mapSnapshot(existing);
+    await this.client.create(VOC_PARSE_CLASSES.competitorListingSnapshot, {
+      publicId: snapshot.id,
+      naturalKey: snapshot.naturalKey,
+      workspaceId: snapshot.workspaceId,
+      platform: snapshot.platform,
+      productId: snapshot.productId,
+      previousSnapshotId: snapshot.previousSnapshotId,
+      contentHash: snapshot.contentHash,
+      observedAt: parseDate(snapshot.observedAt),
+      title: snapshot.title,
+      priceCents: snapshot.priceCents,
+      currency: snapshot.currency,
+      availability: snapshot.availability,
+      mainImageUrl: snapshot.mainImageUrl,
+      keySpecifications: snapshot.keySpecifications,
+      observedFields: snapshot.observedFields,
+      collectionStatus: snapshot.collectionStatus,
+    });
+    return snapshot;
+  }
+
+  async listChanges(workspaceId: string, platform: CompetitorListingPlatform): Promise<CompetitorListingChange[]> {
+    return (await this.client.findAll<StoredChange>(
+      VOC_PARSE_CLASSES.competitorListingChange,
+      { workspaceId, platform },
+    )).map(mapChange);
+  }
+
+  async saveChange(change: CompetitorListingChange): Promise<CompetitorListingChange> {
+    const existing = await this.client.findOne<StoredChange>(VOC_PARSE_CLASSES.competitorListingChange, {
+      naturalKey: change.naturalKey,
+    });
+    if (existing) return mapChange(existing);
+    await this.client.create(VOC_PARSE_CLASSES.competitorListingChange, {
+      publicId: change.id,
+      naturalKey: change.naturalKey,
+      workspaceId: change.workspaceId,
+      platform: change.platform,
+      productId: change.productId,
+      previousSnapshotId: change.previousSnapshotId,
+      currentSnapshotId: change.currentSnapshotId,
+      detectedAt: parseDate(change.detectedAt),
+      changeTypes: change.changeTypes,
+      changes: change.changes,
+    });
+    return change;
+  }
+
+  async findActiveRun(
+    workspaceId: string,
+    platform: CompetitorListingPlatform,
+  ): Promise<CompetitorListingRefreshRun | null> {
+    const response = await this.client.find<StoredRun>(VOC_PARSE_CLASSES.competitorListingRefreshRun, {
+      where: { workspaceId, platform, status: { $in: ['queued', 'running'] } },
+      order: '-requestedAt',
+      limit: 1,
+    });
+    const row = response.results[0];
+    return row ? mapRun(row) : null;
+  }
+
+  async createRun(run: CompetitorListingRefreshRun): Promise<CompetitorListingRefreshRun> {
+    await this.client.create(VOC_PARSE_CLASSES.competitorListingRefreshRun, runBody(run));
+    return run;
+  }
+
+  async updateRun(run: CompetitorListingRefreshRun): Promise<CompetitorListingRefreshRun> {
+    const existing = await this.client.findOne<StoredRun>(VOC_PARSE_CLASSES.competitorListingRefreshRun, {
+      workspaceId: run.workspaceId,
+      publicId: run.id,
+    });
+    if (!existing) throw new Error('Competitor listing refresh run was not found');
+    await this.client.update(VOC_PARSE_CLASSES.competitorListingRefreshRun, existing.objectId, runBody(run));
+    return run;
+  }
+
+  async getRun(workspaceId: string, runId: string): Promise<CompetitorListingRefreshRun | null> {
+    const row = await this.client.findOne<StoredRun>(VOC_PARSE_CLASSES.competitorListingRefreshRun, {
+      workspaceId,
+      publicId: runId,
+    });
+    return row ? mapRun(row) : null;
+  }
+
+  async listRuns(workspaceId: string, platform: CompetitorListingPlatform): Promise<CompetitorListingRefreshRun[]> {
+    const response = await this.client.find<StoredRun>(VOC_PARSE_CLASSES.competitorListingRefreshRun, {
+      where: { workspaceId, platform },
+      order: '-requestedAt',
+      limit: 100,
+    });
+    return response.results.map(mapRun);
+  }
+}
+
+function mapSnapshot(row: StoredSnapshot & ParseObject): CompetitorListingSnapshot {
+  return {
+    id: row.publicId || row.objectId,
+    naturalKey: row.naturalKey,
+    workspaceId: row.workspaceId,
+    platform: row.platform,
+    productId: row.productId,
+    previousSnapshotId: row.previousSnapshotId ?? null,
+    contentHash: row.contentHash,
+    observedAt: dateIso(row.observedAt, row.createdAt),
+    title: row.title ?? null,
+    priceCents: row.priceCents ?? null,
+    currency: 'CNY',
+    availability: row.availability,
+    mainImageUrl: row.mainImageUrl ?? null,
+    keySpecifications: row.keySpecifications,
+    observedFields: row.observedFields,
+    collectionStatus: row.collectionStatus,
+  };
+}
+
+function mapChange(row: StoredChange & ParseObject): CompetitorListingChange {
+  return {
+    id: row.publicId || row.objectId,
+    naturalKey: row.naturalKey,
+    workspaceId: row.workspaceId,
+    platform: row.platform,
+    productId: row.productId,
+    previousSnapshotId: row.previousSnapshotId,
+    currentSnapshotId: row.currentSnapshotId,
+    detectedAt: dateIso(row.detectedAt, row.createdAt),
+    changeTypes: row.changeTypes,
+    changes: row.changes,
+  };
+}
+
+function mapRun(row: StoredRun & ParseObject): CompetitorListingRefreshRun {
+  return {
+    id: row.publicId || row.objectId,
+    workspaceId: row.workspaceId,
+    platform: row.platform,
+    trigger: row.trigger,
+    status: row.status,
+    total: row.total,
+    completed: row.completed,
+    baseline: row.baseline,
+    unchanged: row.unchanged,
+    changed: row.changed,
+    failed: row.failed,
+    itemResults: row.itemResults,
+    requestedAt: dateIso(row.requestedAt, row.createdAt),
+    startedAt: row.startedAt ? dateIso(row.startedAt, row.createdAt) : null,
+    completedAt: row.completedAt ? dateIso(row.completedAt, row.updatedAt) : null,
+  };
+}
+
+function runBody(run: CompetitorListingRefreshRun): Record<string, unknown> {
+  return {
+    publicId: run.id,
+    workspaceId: run.workspaceId,
+    platform: run.platform,
+    trigger: run.trigger,
+    status: run.status,
+    total: run.total,
+    completed: run.completed,
+    baseline: run.baseline,
+    unchanged: run.unchanged,
+    changed: run.changed,
+    failed: run.failed,
+    itemResults: run.itemResults,
+    requestedAt: parseDate(run.requestedAt),
+    startedAt: run.startedAt ? parseDate(run.startedAt) : null,
+    completedAt: run.completedAt ? parseDate(run.completedAt) : null,
+  };
+}
+
+function dateIso(value: unknown, fallback: string): string {
+  return parseDateIso(value) ?? new Date(fallback).toISOString();
+}

src/modules/competitor-listing-monitor/routes.ts

@@ -0,0 +1,63 @@
+import { Router } from 'express';
+import { ApiError } from '../../http/api-error.js';
+import type { WorkspaceAccessService } from '../saas-platform/auth.js';
+import type { CompetitorListingMonitorService } from './competitor-listing-monitor.service.js';
+import {
+  competitorListingOverviewResponseSchema,
+  competitorListingRefreshRequestSchema,
+  competitorListingRefreshResponseSchema,
+  competitorListingRunIdSchema,
+  competitorListingRunResponseSchema,
+  competitorListingWorkspaceSchema,
+} from './schemas.js';
+
+export function createCompetitorListingMonitorRouter(input: {
+  service: CompetitorListingMonitorService;
+  access: WorkspaceAccessService;
+  defaultWorkspaceId: string;
+}): Router {
+  const router = Router();
+
+  router.get('/overview', async (request, response, next) => {
+    try {
+      const query = competitorListingWorkspaceSchema.parse(request.query);
+      const workspaceId = query.workspaceId ?? input.defaultWorkspaceId;
+      await input.access.require(request, workspaceId, 'workspace:read');
+      const overview = await input.service.overview(workspaceId, query.platform);
+      response.json(competitorListingOverviewResponseSchema.parse(overview));
+    } catch (error) { next(error); }
+  });
+
+  router.post('/refresh', async (request, response, next) => {
+    try {
+      const body = competitorListingRefreshRequestSchema.parse(request.body);
+      const workspaceId = body.workspaceId ?? input.defaultWorkspaceId;
+      await input.access.require(request, workspaceId, 'data:sync');
+      const run = await input.service.startRefresh(workspaceId, body.platform, 'manual');
+      response.status(202).json(competitorListingRefreshResponseSchema.parse({
+        run: {
+          id: run.id,
+          status: run.status,
+          total: run.total,
+          requestedAt: run.requestedAt,
+        },
+      }));
+    } catch (error) { next(error); }
+  });
+
+  router.get('/runs/:runId', async (request, response, next) => {
+    try {
+      const query = competitorListingWorkspaceSchema.parse(request.query);
+      const workspaceId = query.workspaceId ?? input.defaultWorkspaceId;
+      const runId = competitorListingRunIdSchema.parse(request.params.runId);
+      await input.access.require(request, workspaceId, 'workspace:read');
+      const run = await input.service.getRun(workspaceId, runId);
+      if (!run || run.platform !== query.platform) {
+        throw new ApiError(404, 'competitor_listing_refresh_run_not_found');
+      }
+      response.json(competitorListingRunResponseSchema.parse({ run }));
+    } catch (error) { next(error); }
+  });
+
+  return router;
+}

src/modules/competitor-listing-monitor/schemas.ts

@@ -0,0 +1,129 @@
+import { z } from 'zod';
+
+export const competitorListingWorkspaceSchema = z.object({
+  workspaceId: z.string().trim().min(1).max(200).optional(),
+  platform: z.enum(['jd']).default('jd'),
+});
+
+export const competitorListingRefreshRequestSchema = competitorListingWorkspaceSchema;
+
+export const competitorListingRunIdSchema = z.uuid();
+
+const keySpecificationsSchema = z.object({
+  model: z.string().nullable(),
+  color: z.string().nullable(),
+  specification: z.string().nullable(),
+  origin: z.string().nullable(),
+  weightKg: z.string().nullable(),
+  lengthMm: z.string().nullable(),
+  widthMm: z.string().nullable(),
+  heightMm: z.string().nullable(),
+});
+
+const observedFieldSchema = z.enum(['title', 'price', 'availability', 'main_image', 'key_specifications']);
+
+export const competitorListingSnapshotSchema = z.object({
+  id: z.string().min(1),
+  naturalKey: z.string().min(1),
+  workspaceId: z.string().min(1),
+  platform: z.literal('jd'),
+  productId: z.string().min(1),
+  previousSnapshotId: z.string().nullable(),
+  contentHash: z.string().length(64),
+  observedAt: z.iso.datetime(),
+  title: z.string().nullable(),
+  priceCents: z.number().int().nonnegative().nullable(),
+  currency: z.literal('CNY'),
+  availability: z.enum(['online', 'offline', 'unknown']),
+  mainImageUrl: z.string().nullable(),
+  keySpecifications: keySpecificationsSchema,
+  observedFields: z.array(observedFieldSchema),
+  collectionStatus: z.enum(['succeeded', 'partial']),
+});
+
+export const competitorListingChangeSchema = z.object({
+  id: z.string().min(1),
+  naturalKey: z.string().min(1),
+  workspaceId: z.string().min(1),
+  platform: z.literal('jd'),
+  productId: z.string().min(1),
+  previousSnapshotId: z.string().min(1),
+  currentSnapshotId: z.string().min(1),
+  detectedAt: z.iso.datetime(),
+  changeTypes: z.array(observedFieldSchema),
+  changes: z.array(z.object({
+    field: z.string().min(1),
+    before: z.unknown(),
+    after: z.unknown(),
+  })),
+});
+
+export const competitorListingRefreshItemResultSchema = z.object({
+  productId: z.string().min(1),
+  status: z.enum(['baseline', 'unchanged', 'changed', 'failed']),
+  errorCode: z.string().min(1).max(100).optional(),
+  observedAt: z.iso.datetime(),
+});
+
+export const competitorListingRefreshRunSchema = z.object({
+  id: z.uuid(),
+  workspaceId: z.string().min(1),
+  platform: z.literal('jd'),
+  trigger: z.enum(['manual', 'scheduled']),
+  status: z.enum(['queued', 'running', 'completed', 'partial', 'failed']),
+  total: z.number().int().nonnegative(),
+  completed: z.number().int().nonnegative(),
+  baseline: z.number().int().nonnegative(),
+  unchanged: z.number().int().nonnegative(),
+  changed: z.number().int().nonnegative(),
+  failed: z.number().int().nonnegative(),
+  itemResults: z.array(competitorListingRefreshItemResultSchema),
+  requestedAt: z.iso.datetime(),
+  startedAt: z.iso.datetime().nullable(),
+  completedAt: z.iso.datetime().nullable(),
+});
+
+export const competitorListingOverviewResponseSchema = z.object({
+  summary: z.object({
+    targetTotal: z.number().int().nonnegative(),
+    baselineTotal: z.number().int().nonnegative(),
+    comparedTotal: z.number().int().nonnegative(),
+    changedIn7d: z.number().int().nonnegative().nullable(),
+    priceChangedIn7d: z.number().int().nonnegative().nullable(),
+    contentChangedIn7d: z.number().int().nonnegative().nullable(),
+    latestFailedTotal: z.number().int().nonnegative(),
+    lastRefreshAt: z.iso.datetime().nullable(),
+    lastSuccessfulRefreshAt: z.iso.datetime().nullable(),
+    historyStatus: z.enum(['not_initialized', 'baseline_only', 'comparable']),
+  }),
+  items: z.array(z.object({
+    productId: z.string().min(1),
+    platform: z.literal('jd'),
+    title: z.string().nullable(),
+    brand: z.string().nullable(),
+    mainImageUrl: z.string().nullable(),
+    category: z.string().nullable(),
+    relatedProducts: z.array(z.object({ productId: z.string().min(1), title: z.string().nullable() })),
+    currentSnapshot: competitorListingSnapshotSchema.nullable(),
+    latestChange: competitorListingChangeSchema.nullable(),
+    latestCollectionResult: competitorListingRefreshItemResultSchema.nullable(),
+    monitorStatus: z.enum(['not_initialized', 'awaiting_comparison', 'changed', 'unchanged', 'failed']),
+  })),
+  facets: z.object({
+    brands: z.array(z.object({ value: z.string().min(1), count: z.number().int().positive() })),
+    categories: z.array(z.object({ value: z.string().min(1), count: z.number().int().positive() })),
+  }),
+});
+
+export const competitorListingRefreshResponseSchema = z.object({
+  run: z.object({
+    id: z.uuid(),
+    status: z.literal('queued'),
+    total: z.number().int().nonnegative(),
+    requestedAt: z.iso.datetime(),
+  }),
+});
+
+export const competitorListingRunResponseSchema = z.object({
+  run: competitorListingRefreshRunSchema,
+});

src/modules/competitor-listing-monitor/snapshot-diff.ts

@@ -0,0 +1,255 @@
+import { createHash } from 'node:crypto';
+import {
+  COMPETITOR_LISTING_OBSERVED_FIELDS,
+  type CompetitorListingAvailability,
+  type CompetitorListingChange,
+  type CompetitorListingChangeType,
+  type CompetitorListingCollectionResult,
+  type CompetitorListingFieldChange,
+  type CompetitorListingKeySpecifications,
+  type CompetitorListingObservation,
+  type CompetitorListingObservedField,
+  type CompetitorListingSnapshot,
+  type CompetitorListingSnapshotDecision,
+} from './domain.js';
+
+const SPECIFICATION_FIELDS = [
+  'model',
+  'color',
+  'specification',
+  'origin',
+  'weightKg',
+  'lengthMm',
+  'widthMm',
+  'heightMm',
+] as const satisfies readonly (keyof CompetitorListingKeySpecifications)[];
+
+const IGNORED_IMAGE_QUERY_PARAMETERS = new Set([
+  't', 'ts', 'timestamp', 'v', 'version', 'w', 'width', 'h', 'height', 'q', 'quality',
+  'x-oss-process', 'imageview2',
+]);
+
+const EMPTY_SPECIFICATIONS: CompetitorListingKeySpecifications = {
+  model: null,
+  color: null,
+  specification: null,
+  origin: null,
+  weightKg: null,
+  lengthMm: null,
+  widthMm: null,
+  heightMm: null,
+};
+
+export function normalizeCompetitorListingText(value: string | null | undefined): string | null {
+  const normalized = value?.normalize('NFKC').replace(/\s+/gu, ' ').trim() ?? '';
+  return normalized.length > 0 ? normalized : null;
+}
+
+export function normalizeCompetitorListingImageUrl(value: string | null | undefined): string | null {
+  const normalized = normalizeCompetitorListingText(value);
+  if (!normalized) return null;
+  const withProtocol = normalized.startsWith('//') ? `https:${normalized}` : normalized;
+  try {
+    const url = new URL(withProtocol);
+    if (url.protocol === 'http:') url.protocol = 'https:';
+    url.hash = '';
+    for (const name of [...url.searchParams.keys()]) {
+      const lowerName = name.toLowerCase();
+      if (lowerName.startsWith('utm_') || IGNORED_IMAGE_QUERY_PARAMETERS.has(lowerName)) {
+        url.searchParams.delete(name);
+      }
+    }
+    url.searchParams.sort();
+    return url.toString();
+  } catch {
+    return withProtocol;
+  }
+}
+
+export function normalizeCompetitorListingSpecifications(
+  value: Partial<CompetitorListingKeySpecifications> | null | undefined,
+): CompetitorListingKeySpecifications {
+  return Object.fromEntries(SPECIFICATION_FIELDS.map((field) => [
+    field,
+    normalizeCompetitorListingText(value?.[field]),
+  ])) as unknown as CompetitorListingKeySpecifications;
+}
+
+export function normalizeCompetitorListingObservation(
+  observation: CompetitorListingObservation,
+): CompetitorListingObservation {
+  const title = normalizeCompetitorListingText(observation.title);
+  const mainImageUrl = normalizeCompetitorListingImageUrl(observation.mainImageUrl);
+  const priceCents = Number.isFinite(observation.priceCents) && observation.priceCents !== null && observation.priceCents >= 0
+    ? Math.round(observation.priceCents)
+    : null;
+  const availability: CompetitorListingAvailability = observation.availability === 'online' || observation.availability === 'offline'
+    ? observation.availability
+    : 'unknown';
+  const keySpecifications = normalizeCompetitorListingSpecifications(observation.keySpecifications);
+  const requestedFields = new Set<CompetitorListingObservedField>(observation.observedFields);
+  const observedFields = COMPETITOR_LISTING_OBSERVED_FIELDS.filter((field) => {
+    if (!requestedFields.has(field)) return false;
+    if (field === 'title') return title !== null;
+    if (field === 'price') return priceCents !== null;
+    if (field === 'availability') return availability !== 'unknown';
+    if (field === 'main_image') return mainImageUrl !== null;
+    return SPECIFICATION_FIELDS.some((specificationField) => keySpecifications[specificationField] !== null);
+  });
+
+  return {
+    ...observation,
+    title: observedFields.includes('title') ? title : null,
+    priceCents: observedFields.includes('price') ? priceCents : null,
+    availability: observedFields.includes('availability') ? availability : 'unknown',
+    mainImageUrl: observedFields.includes('main_image') ? mainImageUrl : null,
+    keySpecifications: observedFields.includes('key_specifications') ? keySpecifications : { ...EMPTY_SPECIFICATIONS },
+    observedFields,
+  };
+}
+
+export function competitorListingContentHash(
+  value: CompetitorListingObservation | CompetitorListingSnapshot,
+): string {
+  const normalized = normalizeHashInput(value);
+  return createHash('sha256').update(JSON.stringify(normalized)).digest('hex');
+}
+
+export function decideCompetitorListingSnapshot(
+  previous: CompetitorListingSnapshot | null,
+  collection: CompetitorListingCollectionResult,
+  ids: { snapshotId: string; changeId?: string },
+): CompetitorListingSnapshotDecision {
+  if (collection.collectionStatus === 'failed') {
+    return collection.errorCode
+      ? { status: 'failed', snapshot: null, change: null, errorCode: collection.errorCode }
+      : { status: 'failed', snapshot: null, change: null };
+  }
+
+  const observation = normalizeCompetitorListingObservation(collection);
+  const contentHash = competitorListingContentHash(observation);
+  const previousSnapshotId = previous?.id ?? null;
+  const naturalKey = [
+    observation.workspaceId,
+    observation.platform,
+    observation.productId,
+    previousSnapshotId ?? 'baseline',
+    contentHash,
+  ].join(':');
+  const snapshot: CompetitorListingSnapshot = {
+    id: ids.snapshotId,
+    naturalKey,
+    workspaceId: observation.workspaceId,
+    platform: observation.platform,
+    productId: observation.productId,
+    previousSnapshotId,
+    contentHash,
+    observedAt: observation.observedAt,
+    title: observation.title,
+    priceCents: observation.priceCents,
+    currency: 'CNY',
+    availability: observation.availability,
+    mainImageUrl: observation.mainImageUrl,
+    keySpecifications: observation.keySpecifications,
+    observedFields: observation.observedFields,
+    collectionStatus: observation.collectionStatus,
+  };
+
+  if (!previous) return { status: 'baseline', snapshot, change: null };
+  if (previous.contentHash === contentHash) return { status: 'unchanged', snapshot: null, change: null };
+
+  const differences = diffCompetitorListingSnapshots(previous, snapshot);
+  if (differences.changes.length === 0) {
+    // Coverage may change on a partial collection. Keep that immutable observation,
+    // but do not turn an unknown or missing field into a business change.
+    return { status: 'unchanged', snapshot, change: null };
+  }
+
+  const changeNaturalKey = `${previous.id}:${snapshot.id}`;
+  const change: CompetitorListingChange = {
+    id: ids.changeId ?? changeNaturalKey,
+    naturalKey: changeNaturalKey,
+    workspaceId: snapshot.workspaceId,
+    platform: snapshot.platform,
+    productId: snapshot.productId,
+    previousSnapshotId: previous.id,
+    currentSnapshotId: snapshot.id,
+    detectedAt: snapshot.observedAt,
+    changeTypes: differences.changeTypes,
+    changes: differences.changes,
+  };
+  return { status: 'changed', snapshot, change };
+}
+
+export function diffCompetitorListingSnapshots(
+  previous: CompetitorListingSnapshot,
+  current: CompetitorListingSnapshot,
+): { changeTypes: CompetitorListingChangeType[]; changes: CompetitorListingFieldChange[] } {
+  const previousObserved = new Set(previous.observedFields);
+  const currentObserved = new Set(current.observedFields);
+  const changes: CompetitorListingFieldChange[] = [];
+  const changeTypes: CompetitorListingChangeType[] = [];
+  const comparable = (field: CompetitorListingObservedField): boolean => (
+    previousObserved.has(field) && currentObserved.has(field)
+  );
+  const record = (type: CompetitorListingChangeType, field: string, before: unknown, after: unknown): void => {
+    if (before === after) return;
+    if (!changeTypes.includes(type)) changeTypes.push(type);
+    changes.push({ field, before, after });
+  };
+
+  if (comparable('price')) record('price', 'priceCents', previous.priceCents, current.priceCents);
+  if (comparable('availability') && previous.availability !== 'unknown' && current.availability !== 'unknown') {
+    record('availability', 'availability', previous.availability, current.availability);
+  }
+  if (comparable('title')) {
+    record('title', 'title', normalizeCompetitorListingText(previous.title), normalizeCompetitorListingText(current.title));
+  }
+  if (comparable('main_image')) {
+    record(
+      'main_image',
+      'mainImageUrl',
+      normalizeCompetitorListingImageUrl(previous.mainImageUrl),
+      normalizeCompetitorListingImageUrl(current.mainImageUrl),
+    );
+  }
+  if (comparable('key_specifications')) {
+    const before = normalizeCompetitorListingSpecifications(previous.keySpecifications);
+    const after = normalizeCompetitorListingSpecifications(current.keySpecifications);
+    for (const field of SPECIFICATION_FIELDS) {
+      // A null specification is treated as unobserved at sub-field level. This is
+      // deliberately conservative: incomplete upstream data must not look deleted.
+      if (before[field] === null || after[field] === null) continue;
+      record('key_specifications', `keySpecifications.${field}`, before[field], after[field]);
+    }
+  }
+
+  return { changeTypes, changes };
+}
+
+function normalizeHashInput(value: CompetitorListingObservation | CompetitorListingSnapshot): Record<string, unknown> {
+  const normalized = normalizeCompetitorListingObservation({
+    workspaceId: value.workspaceId,
+    platform: value.platform,
+    productId: value.productId,
+    observedAt: value.observedAt,
+    title: value.title,
+    priceCents: value.priceCents,
+    availability: value.availability,
+    mainImageUrl: value.mainImageUrl,
+    keySpecifications: value.keySpecifications,
+    observedFields: value.observedFields,
+    collectionStatus: value.collectionStatus,
+  });
+  return {
+    title: normalized.observedFields.includes('title') ? normalized.title : null,
+    priceCents: normalized.observedFields.includes('price') ? normalized.priceCents : null,
+    currency: 'CNY',
+    availability: normalized.observedFields.includes('availability') ? normalized.availability : 'unknown',
+    mainImageUrl: normalized.observedFields.includes('main_image') ? normalized.mainImageUrl : null,
+    keySpecifications: normalized.observedFields.includes('key_specifications')
+      ? Object.fromEntries(SPECIFICATION_FIELDS.map((field) => [field, normalized.keySpecifications[field]]))
+      : { ...EMPTY_SPECIFICATIONS },
+    observedFields: normalized.observedFields,
+  };
+}

src/modules/domestic-voc/adapters/jd-price.adapter.ts

@@ -0,0 +1,19 @@
+import { firstNumber, firstString, unwrapGatewayPayload, walkRecords } from './jd-response.js';
+
+export const JD_PRODUCT_PRICE_PATH = 'jd/get-item-price/v1';
+
+/**
+ * The JD price endpoint returns integer CNY minor units (for example 229900 = CNY 2,299.00).
+ * Only accept a row whose product identity exactly matches the requested product.
+ */
+export function adaptJdPriceResponse(response: unknown, requestedProductId: string): number | null {
+  const payload = unwrapGatewayPayload(response);
+  for (const record of walkRecords(payload, 6)) {
+    const productId = firstString(record, ['good_id', 'itemId', 'skuId', 'productId', 'id']);
+    if (!productId || productId !== requestedProductId) continue;
+    const price = firstNumber(record, ['price', 'jdPrice', 'p']);
+    if (price === null || !Number.isFinite(price) || price <= 0) continue;
+    return Math.round(price);
+  }
+  return null;
+}

src/modules/listing-ai/presentation/listing-score.presenter.ts

@@ -1,4 +1,4 @@
-import type { ListingDimension, ListingProductSummary, ListingRuleEvidence, ListingScoreJob, ListingScoreJobItem, ListingScoreResult, ListingSourceSnapshot } from '../domain.js';
+import type { ListingComplianceStatus, ListingDimension, ListingProductSummary, ListingRuleEvidence, ListingScoreJob, ListingScoreJobItem, ListingScoreResult, ListingSourceSnapshot } from '../domain.js';
 
 const DIMENSIONS: Record<ListingDimension, string> = {
   title: '商品标题', selling_points: '核心卖点', images: '图片资产', description: '商品详情', specifications: '规格与履约',
@@ -21,7 +21,12 @@ export interface ListingScorePresentation {
   methodLabel: string;
   standardLabel: string;
   dataUpdatedAtText: string;
+  scoredAtText: string;
   scopeNote: string;
+  aiConfidence: number | null;
+  suggestions: string[];
+  complianceStatus: ListingComplianceStatus | null;
+  complianceFindings: Array<{ severity: string; message: string; evidence: string[] }>;
   dimensions: Array<{ key: ListingDimension; name: string; score: number | null; maxScore: number; scoreText: string; conclusion: string; items: Array<{ title: string; resultLabel: string; reason: string; action: string | null; impactText: string; sources: string[] }> }>;
 }
 
@@ -35,7 +40,12 @@ export function presentListingScore(score: ListingScoreResult | null, source: Li
     methodLabel: simulated ? '模拟评分' : score.scoreKind === 'hybrid_ai' ? '智能评分' : '自动检查',
     standardLabel: '京东五维评分 V7',
     dataUpdatedAtText: source.syncedAt,
+    scoredAtText: score.createdAt,
     scopeNote: simulated ? '当前为展示用模拟评分，依据现有商品资料生成，不代表真实智能模型结论。' : '当前评分不包含图片审美与构图、详情图片文字识别、用户评价和竞品差异。上述能力不会形成商品扣分。',
+    aiConfidence: score.aiConfidence ?? null,
+    suggestions: score.aiSuggestions.map((item) => item.trim()).filter(Boolean),
+    complianceStatus: score.compliance?.status ?? null,
+    complianceFindings: (score.compliance?.findings ?? []).map((finding) => ({ severity: finding.severity, message: finding.message, evidence: finding.evidence })),
     dimensions: score.dimensions.map((dimension) => ({
       key: dimension.dimension, name: DIMENSIONS[dimension.dimension], score: dimension.score, maxScore: dimension.maxScore,
       scoreText: dimension.score === null ? '等待智能评分' : `${dimension.score} / ${dimension.maxScore}`,

src/modules/listing-ai/schemas.ts

@@ -102,7 +102,10 @@ const listingPresentationItemSchema = z.object({
 });
 export const listingScorePresentationSchema = z.object({
   score: z.number().min(0).max(100).nullable(), scoreText: z.string(), statusLabel: z.string(), methodLabel: z.string(), standardLabel: z.string(),
-  dataUpdatedAtText: z.string(), scopeNote: z.string(), dimensions: z.array(z.object({
+  dataUpdatedAtText: z.string(), scoredAtText: z.string(), scopeNote: z.string(), aiConfidence: z.number().min(0).max(1).nullable(), suggestions: z.array(z.string()),
+  complianceStatus: z.enum(['normal', 'warning', 'needs_review', 'blocked']).nullable(),
+  complianceFindings: z.array(z.object({ severity: z.enum(['low', 'medium', 'high', 'critical']), message: z.string(), evidence: z.array(z.string()) })),
+  dimensions: z.array(z.object({
     key: z.enum(['title', 'selling_points', 'images', 'description', 'specifications']), name: z.string(), score: z.number().nullable(), maxScore: z.number(),
     scoreText: z.string(), conclusion: z.string(), items: z.array(listingPresentationItemSchema),
   })),

src/server.ts

@@ -19,6 +19,8 @@ import { PostgresPlatformRepository } from './modules/saas-platform/postgres-pla
 import { ParseRestProductKnowledgeStore } from './modules/product-knowledge/product-knowledge.store.js';
 import { ParseRestListingAiRepository } from './modules/listing-ai/repositories/parse-rest-listing-ai.repository.js';
 import { PostgresListingAiRepository } from './modules/listing-ai/repositories/postgres-listing-ai.repository.js';
+import { CompetitorListingMonitorService } from './modules/competitor-listing-monitor/competitor-listing-monitor.service.js';
+import { ParseRestCompetitorListingMonitorRepository } from './modules/competitor-listing-monitor/repositories/parse-rest-competitor-listing-monitor.repository.js';
 
 async function main(): Promise<void> {
   const config = loadConfig();
@@ -50,6 +52,10 @@ async function main(): Promise<void> {
     const repository = new ParseRestVocRepository(client);
     const promptConfigs = new ParseRestAiPromptConfigStore(client, config.auth.defaultWorkspaceId);
     const productKnowledge = new ParseRestProductKnowledgeStore(client);
+    const competitorListingMonitor = new CompetitorListingMonitorService(
+      new ParseRestCompetitorListingMonitorRepository(client),
+      gateway,
+    );
     try { if (!skipParseStartupReconciliation) await promptConfigs.ensureDefaults(DEFAULT_DOMESTIC_AI_PROMPT_CONFIGS); }
     catch (error) { if (!(error instanceof ParseRestError && error.status === 404)) throw error; console.warn('[server] skipped prompt reconciliation because Parse gateway is unavailable'); }
     if (bootstrapUserId && !skipParseStartupReconciliation) {
@@ -73,6 +79,7 @@ async function main(): Promise<void> {
       aiPromptConfigs: promptConfigs,
       productKnowledge,
       listingAiRepository: new ParseRestListingAiRepository(client),
+      competitorListingMonitor,
     });
     const processor = new JdSyncService(gateway, repository, config.worker.reviewMaxPages);
     worker = config.worker.enabled

test/competitor-listing-monitor.routes.test.ts

@@ -0,0 +1,156 @@
+import assert from 'node:assert/strict';
+import type { AddressInfo } from 'node:net';
+import test from 'node:test';
+import express, { type ErrorRequestHandler } from 'express';
+import { ZodError } from 'zod';
+import { ApiError } from '../src/http/api-error.js';
+import type { CompetitorListingMonitorService } from '../src/modules/competitor-listing-monitor/competitor-listing-monitor.service.js';
+import type { CompetitorListingRefreshRun } from '../src/modules/competitor-listing-monitor/domain.js';
+import { createCompetitorListingMonitorRouter } from '../src/modules/competitor-listing-monitor/routes.js';
+import {
+  createAuthenticationMiddleware,
+  DisabledAuthenticator,
+  WorkspaceAccessService,
+} from '../src/modules/saas-platform/auth.js';
+import type { PlatformRepository, WorkspaceRole } from '../src/modules/saas-platform/domain.js';
+
+const runId = '11111111-1111-4111-8111-111111111111';
+const requestedAt = '2026-08-26T00:00:00.000Z';
+
+function queuedRun(): CompetitorListingRefreshRun {
+  return {
+    id: runId,
+    workspaceId: 'demashi',
+    platform: 'jd',
+    trigger: 'manual',
+    status: 'queued',
+    total: 37,
+    completed: 0,
+    baseline: 0,
+    unchanged: 0,
+    changed: 0,
+    failed: 0,
+    itemResults: [],
+    requestedAt,
+    startedAt: null,
+    completedAt: null,
+  };
+}
+
+test('competitor listing routes enforce validation, permissions, 202, 409 and run lookup contracts', async () => {
+  let role: WorkspaceRole = 'viewer';
+  let refreshCalls = 0;
+  const platform = {
+    async getMembership(workspaceId: string, userId: string) {
+      return {
+        id: 'membership-1', workspaceId, userId, email: 'user@test.local', displayName: 'Test User',
+        role, status: 'active' as const, createdAt: requestedAt, updatedAt: requestedAt,
+      };
+    },
+  } as unknown as PlatformRepository;
+  const service = {
+    async overview() {
+      return {
+        summary: {
+          targetTotal: 37, baselineTotal: 0, comparedTotal: 0,
+          changedIn7d: null, priceChangedIn7d: null, contentChangedIn7d: null,
+          latestFailedTotal: 0, lastRefreshAt: null, lastSuccessfulRefreshAt: null,
+          historyStatus: 'not_initialized' as const,
+        },
+        items: [],
+        facets: { brands: [], categories: [] },
+      };
+    },
+    async startRefresh() {
+      refreshCalls += 1;
+      if (refreshCalls > 1) throw new ApiError(409, 'competitor_listing_refresh_running');
+      return queuedRun();
+    },
+    async getRun(workspaceId: string, id: string) {
+      return workspaceId === 'demashi' && id === runId ? queuedRun() : null;
+    },
+  } as unknown as CompetitorListingMonitorService;
+
+  const app = express();
+  app.use(express.json());
+  app.use(createAuthenticationMiddleware(new DisabledAuthenticator({
+    userId: 'local-user', email: 'user@test.local', displayName: 'Test User', authMode: 'disabled',
+  })));
+  app.use('/api/competitor-listings', createCompetitorListingMonitorRouter({
+    service,
+    access: new WorkspaceAccessService(platform),
+    defaultWorkspaceId: 'demashi',
+  }));
+  const errorHandler: ErrorRequestHandler = (error, _request, response, _next) => {
+    if (error instanceof ApiError) {
+      response.status(error.status).json({ error: error.code });
+      return;
+    }
+    if (error instanceof ZodError) {
+      response.status(400).json({ error: 'invalid_request' });
+      return;
+    }
+    response.status(500).json({ error: 'internal_error' });
+  };
+  app.use(errorHandler);
+  const server = await new Promise<ReturnType<typeof app.listen>>((resolve) => {
+    const listener = app.listen(0, '127.0.0.1', () => resolve(listener));
+  });
+
+  try {
+    const base = `http://127.0.0.1:${(server.address() as AddressInfo).port}/api/competitor-listings`;
+
+    const overview = await fetch(`${base}/overview?workspaceId=demashi&platform=jd`);
+    assert.equal(overview.status, 200);
+    const overviewBody = await overview.json() as { summary: { targetTotal: number; historyStatus: string } };
+    assert.deepEqual(overviewBody.summary, {
+      targetTotal: 37,
+      baselineTotal: 0,
+      comparedTotal: 0,
+      changedIn7d: null,
+      priceChangedIn7d: null,
+      contentChangedIn7d: null,
+      latestFailedTotal: 0,
+      lastRefreshAt: null,
+      lastSuccessfulRefreshAt: null,
+      historyStatus: 'not_initialized',
+    });
+
+    const invalidPlatform = await fetch(`${base}/overview?platform=tmall`);
+    assert.equal(invalidPlatform.status, 400);
+    assert.equal((await invalidPlatform.json() as { error: string }).error, 'invalid_request');
+
+    const forbidden = await fetch(`${base}/refresh`, {
+      method: 'POST', headers: { 'Content-Type': 'application/json' }, body: '{}',
+    });
+    assert.equal(forbidden.status, 403);
+    assert.equal((await forbidden.json() as { error: string }).error, 'workspace_permission_denied');
+
+    role = 'admin';
+    const accepted = await fetch(`${base}/refresh`, {
+      method: 'POST', headers: { 'Content-Type': 'application/json' }, body: JSON.stringify({ platform: 'jd' }),
+    });
+    assert.equal(accepted.status, 202);
+    assert.deepEqual(await accepted.json(), {
+      run: { id: runId, status: 'queued', total: 37, requestedAt },
+    });
+
+    const conflict = await fetch(`${base}/refresh`, {
+      method: 'POST', headers: { 'Content-Type': 'application/json' }, body: '{}',
+    });
+    assert.equal(conflict.status, 409);
+    assert.equal((await conflict.json() as { error: string }).error, 'competitor_listing_refresh_running');
+
+    const run = await fetch(`${base}/runs/${runId}?workspaceId=demashi`);
+    assert.equal(run.status, 200);
+    assert.deepEqual((await run.json() as { run: CompetitorListingRefreshRun }).run, queuedRun());
+
+    const invalidRun = await fetch(`${base}/runs/not-a-uuid`);
+    assert.equal(invalidRun.status, 400);
+    const missingRun = await fetch(`${base}/runs/${runId}?workspaceId=another-workspace`);
+    assert.equal(missingRun.status, 404);
+    assert.equal((await missingRun.json() as { error: string }).error, 'competitor_listing_refresh_run_not_found');
+  } finally {
+    await new Promise<void>((resolve, reject) => server.close((error) => error ? reject(error) : resolve()));
+  }
+});

test/competitor-listing-monitor.service.test.ts

@@ -0,0 +1,389 @@
+import assert from 'node:assert/strict';
+import test from 'node:test';
+import type { ParseRestClient } from '../src/db/parse-rest.client.js';
+import { VOC_PARSE_CLASSES } from '../src/db/parse-rest.schema.js';
+import {
+  CompetitorListingMonitorService,
+  type CompetitorListingGateway,
+  type CompetitorListingMonitorRepository,
+  type CompetitorListingMonitorTarget,
+} from '../src/modules/competitor-listing-monitor/competitor-listing-monitor.service.js';
+import type {
+  CompetitorListingChange,
+  CompetitorListingRefreshRun,
+  CompetitorListingSnapshot,
+} from '../src/modules/competitor-listing-monitor/domain.js';
+import { ParseRestCompetitorListingMonitorRepository } from '../src/modules/competitor-listing-monitor/repositories/parse-rest-competitor-listing-monitor.repository.js';
+
+const EMPTY_SPECIFICATIONS = {
+  model: null,
+  color: null,
+  specification: null,
+  origin: null,
+  weightKg: null,
+  lengthMm: null,
+  widthMm: null,
+  heightMm: null,
+};
+
+function target(productId: string): CompetitorListingMonitorTarget {
+  return {
+    productId,
+    brand: '测试品牌',
+    title: `档案商品 ${productId}`,
+    mainImageUrl: `https://img.test/profile-${productId}.jpg`,
+    keySpecifications: { ...EMPTY_SPECIFICATIONS, model: `MODEL-${productId}` },
+    category: '商用设备',
+    relatedProducts: [{ productId: `own-${productId}`, title: `本品 ${productId}` }],
+  };
+}
+
+class FakeRepository implements CompetitorListingMonitorRepository {
+  readonly snapshots: CompetitorListingSnapshot[] = [];
+  readonly changes: CompetitorListingChange[] = [];
+  readonly runs: CompetitorListingRefreshRun[] = [];
+  runUpdateFailures = 0;
+
+  constructor(readonly targets: CompetitorListingMonitorTarget[]) {}
+
+  async listTargets(): Promise<CompetitorListingMonitorTarget[]> { return this.targets; }
+
+  async getLatestSnapshot(
+    _workspaceId: string,
+    _platform: 'jd',
+    productId: string,
+  ): Promise<CompetitorListingSnapshot | null> {
+    return this.snapshots.filter((snapshot) => snapshot.productId === productId).at(-1) ?? null;
+  }
+
+  async listSnapshots(): Promise<CompetitorListingSnapshot[]> { return [...this.snapshots]; }
+
+  async saveSnapshot(snapshot: CompetitorListingSnapshot): Promise<CompetitorListingSnapshot> {
+    const existing = this.snapshots.find((item) => item.naturalKey === snapshot.naturalKey);
+    if (existing) return existing;
+    this.snapshots.push(structuredClone(snapshot));
+    return snapshot;
+  }
+
+  async listChanges(): Promise<CompetitorListingChange[]> { return [...this.changes]; }
+
+  async saveChange(change: CompetitorListingChange): Promise<CompetitorListingChange> {
+    const existing = this.changes.find((item) => item.naturalKey === change.naturalKey);
+    if (existing) return existing;
+    this.changes.push(structuredClone(change));
+    return change;
+  }
+
+  async findActiveRun(workspaceId: string): Promise<CompetitorListingRefreshRun | null> {
+    return this.runs.find((run) => run.workspaceId === workspaceId && ['queued', 'running'].includes(run.status)) ?? null;
+  }
+
+  async createRun(run: CompetitorListingRefreshRun): Promise<CompetitorListingRefreshRun> {
+    this.runs.push(structuredClone(run));
+    return run;
+  }
+
+  async updateRun(run: CompetitorListingRefreshRun): Promise<CompetitorListingRefreshRun> {
+    if (this.runUpdateFailures > 0) {
+      this.runUpdateFailures -= 1;
+      throw new Error('fake run update failure');
+    }
+    const index = this.runs.findIndex((candidate) => candidate.id === run.id);
+    assert.notEqual(index, -1);
+    this.runs[index] = structuredClone(run);
+    return run;
+  }
+
+  async getRun(workspaceId: string, runId: string): Promise<CompetitorListingRefreshRun | null> {
+    return this.runs.find((run) => run.workspaceId === workspaceId && run.id === runId) ?? null;
+  }
+
+  async listRuns(): Promise<CompetitorListingRefreshRun[]> { return this.runs.map((run) => structuredClone(run)); }
+}
+
+interface GatewayProduct {
+  title: string;
+  price: number;
+  skuStatus: string;
+}
+
+class FakeGateway implements CompetitorListingGateway {
+  readonly products = new Map<string, GatewayProduct>();
+  readonly detailFailures = new Set<string>();
+  readonly priceFailures = new Set<string>();
+  readonly mismatchedSearches = new Set<string>();
+  gate: Promise<void> | null = null;
+
+  async request<T>(path: string, init: { params?: Record<string, unknown> } = {}): Promise<T> {
+    await this.gate;
+    if (path === 'jd/get-item-detail/v1') {
+      const productId = String(init.params?.['itemId'] ?? '');
+      if (this.detailFailures.has(productId)) throw new Error('fake detail failure');
+      const product = this.products.get(productId);
+      if (!product) throw new Error('missing fake product');
+      return {
+        data: {
+          item: {
+            itemId: productId,
+            itemName: product.title,
+            brandName: '测试品牌',
+            model: `MODEL-${productId}`,
+            categoryName: '商用设备',
+            mainImages: [`https://img.test/${productId}.jpg`],
+            skuStatus: product.skuStatus,
+            color: '银色',
+            specName: '100 L',
+          },
+        },
+      } as T;
+    }
+    if (path === 'jd/search-item-list/v1') {
+      const productId = String(init.params?.['keyword'] ?? '');
+      if (this.priceFailures.has(productId)) throw new Error('fake price failure');
+      const product = this.products.get(productId);
+      const returnedId = this.mismatchedSearches.has(productId) ? `${productId}-other` : productId;
+      return {
+        data: { data: { products: [{ id: returnedId, title: product?.title ?? '商品', price: product?.price ?? 0 }] } },
+      } as T;
+    }
+    if (path === 'jd/get-item-price/v1') {
+      const productId = String(init.params?.['itemId'] ?? '');
+      if (this.priceFailures.has(productId)) throw new Error('fake price failure');
+      const product = this.products.get(productId);
+      const returnedId = this.mismatchedSearches.has(productId) ? `${productId}-other` : productId;
+      return {
+        data: { data: { data: [{ good_id: returnedId, price: Math.round((product?.price ?? 0) * 100) }] } },
+      } as T;
+    }
+    throw new Error(`Unexpected path: ${path}`);
+  }
+}
+
+function clock(): () => Date {
+  let tick = 0;
+  return () => new Date(Date.UTC(2026, 7, 26, 0, 0, tick++));
+}
+
+async function waitForTerminal(repository: FakeRepository, runId: string): Promise<CompetitorListingRefreshRun> {
+  for (let attempt = 0; attempt < 100; attempt += 1) {
+    const run = repository.runs.find((candidate) => candidate.id === runId);
+    if (run && ['completed', 'partial', 'failed'].includes(run.status)) return run;
+    await new Promise((resolve) => setTimeout(resolve, 2));
+  }
+  throw new Error('refresh did not reach a terminal state');
+}
+
+test('Parse repository deduplicates relation targets and uses VocProduct profile fallbacks', async () => {
+  const parse = {
+    async findAll(className: string) {
+      if (className === VOC_PARSE_CLASSES.productRelation) {
+        return [
+          { objectId: 'r1', createdAt: '', updatedAt: '', competitorProductId: 'c1', competitorBrand: '关系品牌', category: '类目 A', ownProductId: 'o1' },
+          { objectId: 'r2', createdAt: '', updatedAt: '', competitorProductId: 'c1', competitorBrand: '关系品牌', category: '类目 A', ownProductId: 'o2' },
+          { objectId: 'r3', createdAt: '', updatedAt: '', competitorProductId: 'c2', competitorBrand: '品牌 B', category: '类目 B', ownProductId: 'o1' },
+        ];
+      }
+      return [
+        { objectId: 'p1', createdAt: '', updatedAt: '2026-08-26T00:00:00.000Z', productId: 'c1', role: 'competitor', brand: '档案品牌', title: '档案竞品', model: 'M1', rawPayload: { mainImages: ['https://img.test/c1.jpg'], color: '银色' } },
+        { objectId: 'p2', createdAt: '', updatedAt: '', productId: 'o1', role: 'own', title: '本品一' },
+        { objectId: 'p3', createdAt: '', updatedAt: '', productId: 'o2', role: 'own', title: '本品二' },
+      ];
+    },
+  } as unknown as ParseRestClient;
+  const repository = new ParseRestCompetitorListingMonitorRepository(parse);
+
+  const targets = await repository.listTargets('demashi', 'jd');
+
+  assert.equal(targets.length, 2);
+  assert.deepEqual(targets.map((item) => item.productId), ['c1', 'c2']);
+  assert.equal(targets[0]?.title, '档案竞品');
+  assert.equal(targets[0]?.brand, '档案品牌');
+  assert.equal(targets[0]?.mainImageUrl, 'https://img.test/c1.jpg');
+  assert.deepEqual(targets[0]?.relatedProducts, [
+    { productId: 'o1', title: '本品一' },
+    { productId: 'o2', title: '本品二' },
+  ]);
+  assert.equal(targets[1]?.brand, '品牌 B');
+});
+
+test('refresh establishes baselines and identical data does not duplicate snapshots or changes', async () => {
+  const repository = new FakeRepository([target('c1'), target('c2'), target('c3')]);
+  const gateway = new FakeGateway();
+  for (const productId of ['c1', 'c2', 'c3']) {
+    gateway.products.set(productId, { title: `商品 ${productId}`, price: 1000, skuStatus: '1' });
+  }
+  const service = new CompetitorListingMonitorService(repository, gateway, clock(), 3);
+
+  const first = await service.startRefresh('demashi', 'jd');
+  const firstCompleted = await waitForTerminal(repository, first.id);
+  assert.equal(firstCompleted.status, 'completed');
+  assert.deepEqual(
+    { baseline: firstCompleted.baseline, unchanged: firstCompleted.unchanged, changed: firstCompleted.changed, failed: firstCompleted.failed },
+    { baseline: 3, unchanged: 0, changed: 0, failed: 0 },
+  );
+  assert.equal(repository.snapshots.length, 3);
+  assert.equal(repository.changes.length, 0);
+
+  const second = await service.startRefresh('demashi', 'jd');
+  const secondCompleted = await waitForTerminal(repository, second.id);
+  assert.equal(secondCompleted.status, 'completed');
+  assert.equal(secondCompleted.unchanged, 3);
+  assert.equal(repository.snapshots.length, 3);
+  assert.equal(repository.changes.length, 0);
+
+  const overview = await service.overview('demashi', 'jd');
+  assert.equal(overview.summary.targetTotal, 3);
+  assert.equal(overview.summary.baselineTotal, 3);
+  assert.equal(overview.summary.comparedTotal, 0);
+  assert.equal(overview.summary.historyStatus, 'baseline_only');
+  assert.equal(overview.items[0]?.relatedProducts.length, 1);
+});
+
+test('a scheduled run processes all 37 mapped targets with bounded concurrency and complete statistics', async () => {
+  const targets = Array.from({ length: 37 }, (_, index) => target(`c${index + 1}`));
+  const repository = new FakeRepository(targets);
+  const innerGateway = new FakeGateway();
+  for (const item of targets) {
+    innerGateway.products.set(item.productId, { title: `商品 ${item.productId}`, price: 1000, skuStatus: '1' });
+  }
+  let active = 0;
+  let maxActive = 0;
+  const gateway: CompetitorListingGateway = {
+    async request<T>(
+      path: string,
+      init?: { method?: 'GET' | 'POST'; params?: Record<string, unknown>; refresh?: boolean },
+    ) {
+      active += 1;
+      maxActive = Math.max(maxActive, active);
+      try {
+        await new Promise((resolve) => setTimeout(resolve, 1));
+        return await innerGateway.request<T>(path, init);
+      } finally {
+        active -= 1;
+      }
+    },
+  };
+  const service = new CompetitorListingMonitorService(repository, gateway, clock(), 3);
+
+  const refresh = await service.startRefresh('demashi', 'jd', 'scheduled');
+  const completed = await waitForTerminal(repository, refresh.id);
+
+  assert.equal(completed.status, 'completed');
+  assert.equal(completed.total, 37);
+  assert.equal(completed.completed, 37);
+  assert.equal(completed.baseline, 37);
+  assert.equal(completed.failed, 0);
+  assert.equal(repository.snapshots.length, 37);
+  assert.ok(maxActive > 1);
+  assert.ok(maxActive <= 3);
+});
+
+test('partial product failures are isolated and detail or price failures preserve prior valid evidence', async () => {
+  const repository = new FakeRepository([target('c1'), target('c2'), target('c3')]);
+  const gateway = new FakeGateway();
+  for (const productId of ['c1', 'c2', 'c3']) {
+    gateway.products.set(productId, { title: `商品 ${productId}`, price: 1000, skuStatus: '1' });
+  }
+  const service = new CompetitorListingMonitorService(repository, gateway, clock(), 3);
+  const baselineRun = await service.startRefresh('demashi', 'jd');
+  await waitForTerminal(repository, baselineRun.id);
+  const c2BaselineId = (await repository.getLatestSnapshot('demashi', 'jd', 'c2'))?.id;
+  const c3Baseline = await repository.getLatestSnapshot('demashi', 'jd', 'c3');
+  assert.ok(c3Baseline?.observedFields.includes('price'));
+
+  gateway.products.set('c1', { title: '商品 c1 新标题', price: 899, skuStatus: '0' });
+  gateway.detailFailures.add('c2');
+  gateway.priceFailures.add('c3');
+  const refresh = await service.startRefresh('demashi', 'jd');
+  const completed = await waitForTerminal(repository, refresh.id);
+
+  assert.equal(completed.status, 'partial');
+  assert.deepEqual(
+    { completed: completed.completed, changed: completed.changed, unchanged: completed.unchanged, failed: completed.failed },
+    { completed: 3, changed: 1, unchanged: 1, failed: 1 },
+  );
+  assert.equal((await repository.getLatestSnapshot('demashi', 'jd', 'c2'))?.id, c2BaselineId);
+  assert.equal(repository.snapshots.filter((snapshot) => snapshot.productId === 'c2').length, 1);
+  assert.ok(repository.snapshots.some((snapshot) => snapshot.id === c3Baseline?.id && snapshot.priceCents === 100_000));
+  const c3Latest = await repository.getLatestSnapshot('demashi', 'jd', 'c3');
+  assert.equal(c3Latest?.priceCents, null);
+  assert.equal(repository.changes.some((change) => change.productId === 'c3' && change.changeTypes.includes('price')), false);
+  const c1Change = repository.changes.find((change) => change.productId === 'c1');
+  assert.deepEqual(c1Change?.changeTypes, ['price', 'availability', 'title']);
+});
+
+test('an exact productId mismatch in search is treated as partial instead of borrowing another product price', async () => {
+  const repository = new FakeRepository([target('c1')]);
+  const gateway = new FakeGateway();
+  gateway.products.set('c1', { title: '商品 c1', price: 1000, skuStatus: '1' });
+  gateway.mismatchedSearches.add('c1');
+  const service = new CompetitorListingMonitorService(repository, gateway, clock());
+
+  const refresh = await service.startRefresh('demashi', 'jd');
+  const completed = await waitForTerminal(repository, refresh.id);
+
+  assert.equal(completed.status, 'partial');
+  assert.equal(completed.baseline, 1);
+  assert.equal(completed.itemResults[0]?.errorCode, 'jd_price_not_found');
+  assert.equal(repository.snapshots[0]?.priceCents, null);
+  assert.equal(repository.snapshots[0]?.observedFields.includes('price'), false);
+});
+
+test('all item failures produce a failed terminal run without creating snapshots', async () => {
+  const repository = new FakeRepository([target('c1'), target('c2')]);
+  const gateway = new FakeGateway();
+  gateway.detailFailures.add('c1');
+  gateway.detailFailures.add('c2');
+  const service = new CompetitorListingMonitorService(repository, gateway, clock(), 2);
+
+  const refresh = await service.startRefresh('demashi', 'jd', 'scheduled');
+  const completed = await waitForTerminal(repository, refresh.id);
+
+  assert.equal(completed.trigger, 'scheduled');
+  assert.equal(completed.status, 'failed');
+  assert.equal(completed.completed, 2);
+  assert.equal(completed.failed, 2);
+  assert.ok(completed.completedAt);
+  assert.equal(repository.snapshots.length, 0);
+  assert.equal(repository.changes.length, 0);
+
+  const overview = await service.overview('demashi', 'jd');
+  assert.equal(overview.summary.baselineTotal, 0);
+  assert.equal(overview.summary.latestFailedTotal, 2);
+  assert.deepEqual(overview.items.map((item) => item.monitorStatus), ['failed', 'failed']);
+  assert.deepEqual(overview.items.map((item) => item.title), ['档案商品 c1', '档案商品 c2']);
+});
+
+test('a transient background persistence exception is forced into a failed terminal run', async () => {
+  const repository = new FakeRepository([target('c1')]);
+  repository.runUpdateFailures = 1;
+  const gateway = new FakeGateway();
+  gateway.products.set('c1', { title: '商品 c1', price: 1000, skuStatus: '1' });
+  const service = new CompetitorListingMonitorService(repository, gateway, clock());
+
+  const refresh = await service.startRefresh('demashi', 'jd', 'scheduled');
+  const completed = await waitForTerminal(repository, refresh.id);
+
+  assert.equal(completed.status, 'failed');
+  assert.ok(completed.completedAt);
+  assert.notEqual(completed.status, 'queued');
+  assert.notEqual(completed.status, 'running');
+});
+
+test('manual API and scheduled CLI service instances share the persisted workspace mutex', async () => {
+  const repository = new FakeRepository([target('c1')]);
+  const gateway = new FakeGateway();
+  gateway.products.set('c1', { title: '商品 c1', price: 1000, skuStatus: '1' });
+  let release: () => void = () => {};
+  gateway.gate = new Promise<void>((resolve) => { release = resolve; });
+  const apiService = new CompetitorListingMonitorService(repository, gateway, clock());
+  const cliService = new CompetitorListingMonitorService(repository, gateway, clock());
+
+  const first = await apiService.startRefresh('demashi', 'jd', 'manual');
+  await assert.rejects(
+    cliService.startRefresh('demashi', 'jd', 'scheduled'),
+    (error: unknown) => error instanceof Error && 'code' in error && error.code === 'competitor_listing_refresh_running',
+  );
+  release();
+  await waitForTerminal(repository, first.id);
+});

test/competitor-listing-monitor.test.ts

@@ -0,0 +1,169 @@
+import assert from 'node:assert/strict';
+import test from 'node:test';
+import { VOC_PARSE_CLASSES, VOC_PARSE_SCHEMAS } from '../src/db/parse-rest.schema.js';
+import type {
+  CompetitorListingCollectionResult,
+  CompetitorListingObservation,
+  CompetitorListingSnapshot,
+} from '../src/modules/competitor-listing-monitor/domain.js';
+import {
+  decideCompetitorListingSnapshot,
+  normalizeCompetitorListingImageUrl,
+  normalizeCompetitorListingText,
+} from '../src/modules/competitor-listing-monitor/snapshot-diff.js';
+
+function observation(overrides: Partial<CompetitorListingObservation> = {}): CompetitorListingObservation {
+  return {
+    workspaceId: 'demashi',
+    platform: 'jd',
+    productId: '1001',
+    observedAt: '2026-08-26T01:00:00.000Z',
+    title: '德玛仕 商用冰箱',
+    priceCents: 199_900,
+    availability: 'online',
+    mainImageUrl: 'https://img.example.test/product.jpg',
+    keySpecifications: {
+      model: 'DM-100',
+      color: '银色',
+      specification: '100 L',
+      origin: '广东',
+      weightKg: '40',
+      lengthMm: '600',
+      widthMm: '650',
+      heightMm: '1800',
+    },
+    observedFields: ['title', 'price', 'availability', 'main_image', 'key_specifications'],
+    collectionStatus: 'succeeded',
+    ...overrides,
+  };
+}
+
+function baseline(overrides: Partial<CompetitorListingObservation> = {}): CompetitorListingSnapshot {
+  const decision = decideCompetitorListingSnapshot(null, observation(overrides), { snapshotId: 'snapshot-1' });
+  assert.equal(decision.status, 'baseline');
+  return decision.snapshot;
+}
+
+test('first successful collection creates only an immutable baseline', () => {
+  const result = decideCompetitorListingSnapshot(null, observation(), { snapshotId: 'snapshot-1' });
+  assert.equal(result.status, 'baseline');
+  assert.equal(result.snapshot.previousSnapshotId, null);
+  assert.equal(result.snapshot.collectionStatus, 'succeeded');
+  assert.equal(result.change, null);
+});
+
+test('equivalent normalized content creates neither a duplicate snapshot nor an event', () => {
+  const previous = baseline();
+  const result = decideCompetitorListingSnapshot(previous, observation({
+    observedAt: '2026-08-27T01:00:00.000Z',
+    title: '  德玛仕\u3000商用冰箱  ',
+    mainImageUrl: 'http://img.example.test/product.jpg?width=800&utm_source=test',
+    observedFields: ['key_specifications', 'main_image', 'availability', 'price', 'title'],
+  }), { snapshotId: 'snapshot-2' });
+  assert.equal(result.status, 'unchanged');
+  assert.equal(result.snapshot, null);
+  assert.equal(result.change, null);
+  assert.equal(normalizeCompetitorListingText('ＡＢＣ\u3000  商品'), 'ABC 商品');
+  assert.equal(normalizeCompetitorListingImageUrl('//img.example.test/product.jpg?t=123'), 'https://img.example.test/product.jpg');
+});
+
+test('price, title, main image, availability and individual specifications are detected', () => {
+  const previous = baseline();
+  const result = decideCompetitorListingSnapshot(previous, observation({
+    observedAt: '2026-08-27T01:00:00.000Z',
+    title: '德玛仕 商用冷柜',
+    priceCents: 189_900,
+    availability: 'offline',
+    mainImageUrl: 'https://img.example.test/new-product.jpg',
+    keySpecifications: {
+      ...previous.keySpecifications,
+      model: 'DM-200',
+      widthMm: '700',
+    },
+  }), { snapshotId: 'snapshot-2', changeId: 'change-1' });
+
+  assert.equal(result.status, 'changed');
+  assert.deepEqual(result.change.changeTypes, [
+    'price', 'availability', 'title', 'main_image', 'key_specifications',
+  ]);
+  assert.deepEqual(result.change.changes.map((item) => item.field), [
+    'priceCents',
+    'availability',
+    'title',
+    'mainImageUrl',
+    'keySpecifications.model',
+    'keySpecifications.widthMm',
+  ]);
+  assert.equal(result.snapshot.previousSnapshotId, 'snapshot-1');
+  assert.equal(result.change.naturalKey, 'snapshot-1:snapshot-2');
+});
+
+test('missing and unknown fields change coverage without producing false changes', () => {
+  const previous = baseline();
+  const result = decideCompetitorListingSnapshot(previous, observation({
+    observedAt: '2026-08-27T01:00:00.000Z',
+    title: null,
+    priceCents: null,
+    availability: 'unknown',
+    mainImageUrl: null,
+    keySpecifications: {
+      ...previous.keySpecifications,
+      model: null,
+    },
+    collectionStatus: 'partial',
+  }), { snapshotId: 'snapshot-2' });
+
+  assert.equal(result.status, 'unchanged');
+  assert.ok(result.snapshot, 'a changed partial observation remains available for persistence');
+  assert.deepEqual(result.snapshot.observedFields, ['key_specifications']);
+  assert.equal(result.change, null);
+});
+
+test('unmonitored upstream fields do not participate in hashing or comparison', () => {
+  const previous = baseline();
+  const withUnknownFields = {
+    ...observation({ observedAt: '2026-08-27T01:00:00.000Z' }),
+    sellerName: 'a newly returned upstream field',
+    observedFields: [
+      'title', 'price', 'availability', 'main_image', 'key_specifications', 'seller_name',
+    ],
+  } as unknown as CompetitorListingObservation;
+  const result = decideCompetitorListingSnapshot(previous, withUnknownFields, { snapshotId: 'snapshot-2' });
+  assert.equal(result.status, 'unchanged');
+  assert.equal(result.snapshot, null);
+  assert.equal(result.change, null);
+});
+
+test('an unknown specification sub-field is not interpreted as a deletion', () => {
+  const previous = baseline();
+  const result = decideCompetitorListingSnapshot(previous, observation({
+    keySpecifications: { ...previous.keySpecifications, model: null, color: '黑色' },
+  }), { snapshotId: 'snapshot-2' });
+  assert.equal(result.status, 'changed');
+  assert.deepEqual(result.change.changeTypes, ['key_specifications']);
+  assert.deepEqual(result.change.changes, [{
+    field: 'keySpecifications.color', before: '银色', after: '黑色',
+  }]);
+});
+
+test('collection failure never creates an empty snapshot or an offline event', () => {
+  const previous = baseline();
+  const failure: CompetitorListingCollectionResult = {
+    workspaceId: 'demashi', platform: 'jd', productId: '1001',
+    observedAt: '2026-08-27T01:00:00.000Z', collectionStatus: 'failed', errorCode: 'jd_detail_failed',
+  };
+  const result = decideCompetitorListingSnapshot(previous, failure, { snapshotId: 'snapshot-2' });
+  assert.deepEqual(result, {
+    status: 'failed', snapshot: null, change: null, errorCode: 'jd_detail_failed',
+  });
+});
+
+test('the three Parse classes required by phase 1 are registered', () => {
+  const classNames = new Set(VOC_PARSE_SCHEMAS.map((schema) => schema.className));
+  assert.equal(VOC_PARSE_CLASSES.competitorListingSnapshot, 'VocCompetitorListingSnapshot');
+  assert.equal(VOC_PARSE_CLASSES.competitorListingChange, 'VocCompetitorListingChange');
+  assert.equal(VOC_PARSE_CLASSES.competitorListingRefreshRun, 'VocCompetitorListingRefreshRun');
+  assert.equal(classNames.has(VOC_PARSE_CLASSES.competitorListingSnapshot), true);
+  assert.equal(classNames.has(VOC_PARSE_CLASSES.competitorListingChange), true);
+  assert.equal(classNames.has(VOC_PARSE_CLASSES.competitorListingRefreshRun), true);
+});

test/competitor-listing-refresh-cli.test.ts

@@ -0,0 +1,95 @@
+import assert from 'node:assert/strict';
+import test from 'node:test';
+import {
+  COMPETITOR_LISTING_REFRESH_EXIT,
+  runCompetitorListingRefreshCli,
+} from '../scripts/refresh-competitor-listings.js';
+import { ApiError } from '../src/http/api-error.js';
+import type { CompetitorListingRefreshRun } from '../src/modules/competitor-listing-monitor/domain.js';
+
+const runId = '11111111-1111-4111-8111-111111111111';
+const requestedAt = '2026-08-26T00:00:00.000Z';
+
+function run(status: CompetitorListingRefreshRun['status']): CompetitorListingRefreshRun {
+  const terminal = ['completed', 'partial', 'failed'].includes(status);
+  return {
+    id: runId,
+    workspaceId: 'demashi',
+    platform: 'jd',
+    trigger: 'scheduled',
+    status,
+    total: 3,
+    completed: terminal ? 3 : 0,
+    baseline: status === 'completed' ? 3 : 0,
+    unchanged: 0,
+    changed: 0,
+    failed: status === 'partial' ? 1 : status === 'failed' ? 3 : 0,
+    itemResults: [],
+    requestedAt,
+    startedAt: status === 'queued' ? null : '2026-08-26T00:00:01.000Z',
+    completedAt: terminal ? '2026-08-26T00:00:02.000Z' : null,
+  };
+}
+
+test('scheduled CLI waits for terminal status and maps completed, partial and failed exit codes', async (context) => {
+  for (const scenario of [
+    { status: 'completed' as const, exitCode: COMPETITOR_LISTING_REFRESH_EXIT.completed },
+    { status: 'partial' as const, exitCode: COMPETITOR_LISTING_REFRESH_EXIT.partial },
+    { status: 'failed' as const, exitCode: COMPETITOR_LISTING_REFRESH_EXIT.failed },
+  ]) {
+    await context.test(scenario.status, async () => {
+      const outputs: string[] = [];
+      const errors: string[] = [];
+      let trigger = '';
+      let polls = 0;
+      const service = {
+        async startRefresh(_workspaceId: string, _platform: 'jd', value: 'manual' | 'scheduled') {
+          trigger = value;
+          return run('queued');
+        },
+        async getRun() {
+          polls += 1;
+          return polls === 1 ? run('running') : run(scenario.status);
+        },
+      };
+
+      const exitCode = await runCompetitorListingRefreshCli({
+        service,
+        workspaceId: 'demashi',
+        pollMs: 1,
+        timeoutMs: 1_000,
+        sleep: async () => undefined,
+        write: (line) => outputs.push(line),
+        writeError: (line) => errors.push(line),
+      });
+
+      assert.equal(trigger, 'scheduled');
+      assert.equal(exitCode, scenario.exitCode);
+      assert.equal(polls, 2);
+      const terminalLine = scenario.status === 'failed' ? errors.at(-1) : outputs.at(-1);
+      assert.equal((JSON.parse(terminalLine ?? '{}') as { status?: string }).status, scenario.status);
+    });
+  }
+});
+
+test('scheduled CLI reports a distinct exit code when a manual refresh already owns the workspace lock', async () => {
+  const errors: string[] = [];
+  const service = {
+    async startRefresh(): Promise<CompetitorListingRefreshRun> {
+      throw new ApiError(409, 'competitor_listing_refresh_running');
+    },
+    async getRun(): Promise<CompetitorListingRefreshRun | null> {
+      throw new Error('must not poll after a lock conflict');
+    },
+  };
+
+  const exitCode = await runCompetitorListingRefreshCli({
+    service,
+    workspaceId: 'demashi',
+    write: () => undefined,
+    writeError: (line) => errors.push(line),
+  });
+
+  assert.equal(exitCode, COMPETITOR_LISTING_REFRESH_EXIT.alreadyRunning);
+  assert.equal((JSON.parse(errors[0] ?? '{}') as { error?: string }).error, 'competitor_listing_refresh_running');
+});

test/jd-adapters.test.ts

@@ -1,6 +1,7 @@
 import assert from 'node:assert/strict';
 import test from 'node:test';
 import { adaptJdProductProfile, adaptJdProductResponse } from '../src/modules/domestic-voc/adapters/jd-product.adapter.js';
+import { adaptJdPriceResponse } from '../src/modules/domestic-voc/adapters/jd-price.adapter.js';
 import { GatewayPayloadError } from '../src/modules/domestic-voc/adapters/jd-response.js';
 import { adaptJdReviewResponse } from '../src/modules/domestic-voc/adapters/jd-review.adapter.js';
 
@@ -56,6 +57,29 @@ test('JD product profile exposes normalized catalog fields without the raw respo
   assert.equal(profile?.collectedAt, '2026-07-24T00:00:00.000Z');
 });
 
+test('JD price adapter maps exact product price minor units', () => {
+  const priceCents = adaptJdPriceResponse({
+    code: 200,
+    data: {
+      code: 0,
+      data: {
+        data: [
+          { good_id: 'other-product', price: 999_900 },
+          { good_id: '100119862001', price: 229_900 },
+        ],
+      },
+    },
+  }, '100119862001');
+
+  assert.equal(priceCents, 229_900);
+});
+
+test('JD price adapter rejects a mismatched product identity', () => {
+  assert.equal(adaptJdPriceResponse({
+    data: { data: { data: [{ good_id: 'different-product', price: 229_900 }] } },
+  }, '100119862001'), null);
+});
+
 test('JD review adapter maps evidence and pagination without reviewer identity', () => {
   const page = adaptJdReviewResponse({
     code: 200,