feat: connect jd sync through fmode gateway

.env.example

@@ -10,8 +10,11 @@ PARSE_APP_ID=
 PARSE_MASTER_KEY=
 PARSE_MAINTENANCE_KEY=
 PARSE_SERVER_URL=http://127.0.0.1:4400/parse
-FMODE_BASE_URL=http://127.0.0.1:3000/api/voc-e-commerce
+FMODE_BASE_URL=https://server.fmode.cn/api/voc-e-commerce
 FMODE_API_KEY=
 FMODE_TIMEOUT_MS=30000
 FMODE_RETRIES=2
+SYNC_WORKER_ENABLED=true
+SYNC_WORKER_POLL_MS=2000
+JD_REVIEW_MAX_PAGES=1
 CORS_ORIGINS=http://127.0.0.1:4300,http://localhost:4200

README.md

@@ -1,6 +1,6 @@
 # SaaS VOC Server
 
-Independent backend template for the domestic ecommerce VOC product. The first case workspace is Demashi on JD. This repository does not connect to the cross-border production database and does not contain any Fmode, JustOneAPI, Parse, or PostgreSQL credential.
+Independent backend template for the domestic ecommerce VOC product. The first case workspace is Demashi on JD. This repository does not connect to the cross-border production database and does not contain any Fmode, Parse, PostgreSQL, or supplier credential.
 
 ## Current state
 
@@ -20,7 +20,7 @@ Completed on 2026-07-23:
 
 The current case import resolves to 2,817 operating products, 9,717 daily metrics, 40 relations, and 38 internal `relation_stub` records required to preserve foreign keys. Stub records are excluded from the frontend product list and are never presented as collected product facts.
 
-The worker does not call a JD upstream endpoint yet. A sync request is queued truthfully as `pending`; it is never marked complete until the real JD detail and review contracts are verified.
+The worker calls JD product-detail and review paths only through the company `/api/voc-e-commerce` gateway. It never sends a browser request to a supplier endpoint and never returns raw gateway errors or credentials.
 
 ## Architecture
 
@@ -31,11 +31,19 @@ Saas-voc frontend
       -> dedicated PostgreSQL database (voc schema)
       -> sync queue
           -> existing /api/voc-e-commerce gateway
-              -> JustOneAPI
 ```
 
 The browser only calls this service. `FMODE_API_KEY` is used in the server-side `Authorization` header and is never returned to the browser or included in a request URL.
 
+Registered JD gateway contracts:
+
+```text
+GET /api/voc-e-commerce/jd/get-item-detail/v1?itemId={productId}
+GET /api/voc-e-commerce/jd/get-item-comments/v1?itemId={productId}&page={page}
+```
+
+The first worker run defaults to one review page. `JD_REVIEW_MAX_PAGES` can raise the bounded limit after quota and response validation.
+
 ## Runtime
 
 Use Node.js 22.13 or newer within the Node 22 release line. The repository intentionally pins Node 22 because Parse Server publishes explicit supported runtime ranges.
@@ -124,8 +132,8 @@ npm audit --omit=dev
 Current result:
 
 - TypeScript build: passed.
-- Unit and HTTP contract tests: 13 passed.
+- Unit, adapter, worker, and HTTP contract tests: 19 passed.
 - Production dependency audit: 0 critical, 0 high, 13 moderate.
-- Live JD contract and quota-consuming request: intentionally not run yet.
+- Live JD request: attempted through the company gateway, but the public TLS connection was reset before HTTP; no supplier endpoint was contacted.
 
 See `TASKS.md` for the implementation sequence and acceptance boundary.

TASKS.md

@@ -28,13 +28,15 @@ Status date: 2026-07-23
 
 ## Phase 3 - real JD source contract
 
-- [ ] Make one quota-controlled JD search request through the existing Fmode gateway.
-- [ ] Confirm the exact JD product-detail path, parameters, response envelope, and error codes.
-- [ ] Confirm the exact JD review path, pagination fields, and any async task/status flow.
+- [ ] Complete one quota-controlled live JD request through the existing Fmode gateway (currently blocked before HTTP by TLS reset).
+- [x] Confirm the catalog JD product-detail path and required `itemId` parameter.
+- [x] Confirm the catalog JD review path and optional `page` parameter.
+- [ ] Confirm the live response envelope, pagination values, and any async task/status flow.
 - [ ] Save sanitized response fixtures without credentials or personal data.
-- [ ] Implement JD product and review adapters against those fixtures.
-- [ ] Process queued jobs with partial-failure events and bounded retries.
-- [ ] UPSERT source results and expose truthful progress through the job endpoint.
+- [x] Implement JD product and review adapters against company-gateway contract fixtures.
+- [x] Process queued jobs with partial-failure events and bounded retries.
+- [x] Implement idempotent product/review UPSERT and truthful job progress updates.
+- [ ] Verify source UPSERT and snapshot totals against the provisioned PostgreSQL database.
 
 ## Phase 4 - deployable closure
 

src/config/env.ts

@@ -17,6 +17,9 @@ const environmentSchema = z.object({
   FMODE_API_KEY: z.string().min(1, 'FMODE_API_KEY is required'),
   FMODE_TIMEOUT_MS: z.coerce.number().int().min(100).max(120_000).default(30_000),
   FMODE_RETRIES: z.coerce.number().int().min(0).max(5).default(2),
+  SYNC_WORKER_ENABLED: z.enum(['true', 'false']).default('true'),
+  SYNC_WORKER_POLL_MS: z.coerce.number().int().min(500).max(60_000).default(2_000),
+  JD_REVIEW_MAX_PAGES: z.coerce.number().int().min(1).max(10).default(1),
   CORS_ORIGINS: z.string().min(1),
 });
 
@@ -43,6 +46,11 @@ export type AppConfig = {
     timeoutMs: number;
     retries: number;
   };
+  worker: {
+    enabled: boolean;
+    pollMs: number;
+    reviewMaxPages: number;
+  };
   corsOrigins: string[];
 };
 
@@ -87,6 +95,11 @@ export function loadConfig(environment: NodeJS.ProcessEnv = process.env): AppCon
       timeoutMs: value.FMODE_TIMEOUT_MS,
       retries: value.FMODE_RETRIES,
     },
+    worker: {
+      enabled: value.SYNC_WORKER_ENABLED === 'true',
+      pollMs: value.SYNC_WORKER_POLL_MS,
+      reviewMaxPages: value.JD_REVIEW_MAX_PAGES,
+    },
     corsOrigins: value.CORS_ORIGINS.split(',').map((origin) => origin.trim()).filter(Boolean),
   };
 }

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

@@ -0,0 +1,50 @@
+import { makeProductKey } from '../domain/identity.js';
+import {
+  findBestRecord,
+  firstString,
+  type JsonRecord,
+  unwrapGatewayPayload,
+} from './jd-response.js';
+
+export const JD_PRODUCT_DETAIL_PATH = 'jd/get-item-detail/v1';
+
+export interface JdProductRecord {
+  platform: 'jd';
+  productId: string;
+  productKey: string;
+  role: 'own';
+  brand: string;
+  title: string;
+  model: string;
+  category1: string;
+  category2: string;
+  category3: string;
+  source: 'fmode_gateway';
+  rawPayload: JsonRecord | null;
+}
+
+const PRODUCT_KEYS = [
+  'itemId', 'skuId', 'productId', 'itemName', 'skuName', 'title', 'name',
+  'brandName', 'brand', 'model', 'categoryName', 'category1', 'category2', 'category3',
+];
+
+export function adaptJdProductResponse(response: unknown, requestedProductId: string): JdProductRecord {
+  const payload = unwrapGatewayPayload(response);
+  const record = findBestRecord(payload, PRODUCT_KEYS);
+  const productId = firstString(record ?? {}, ['itemId', 'skuId', 'productId', 'id']) || requestedProductId;
+  return {
+    platform: 'jd',
+    productId,
+    productKey: makeProductKey('jd', productId),
+    role: 'own',
+    brand: firstString(record ?? {}, ['brandName', 'brand', 'brand_name']),
+    title: firstString(record ?? {}, ['itemName', 'skuName', 'title', 'name', 'productName']),
+    model: firstString(record ?? {}, ['model', 'modelName', 'skuModel', 'wareModel']),
+    category1: firstString(record ?? {}, ['category1', 'firstCategoryName', 'cid1Name']),
+    category2: firstString(record ?? {}, ['category2', 'secondCategoryName', 'cid2Name']),
+    category3: firstString(record ?? {}, ['category3', 'thirdCategoryName', 'categoryName', 'cid3Name']),
+    source: 'fmode_gateway',
+    rawPayload: record,
+  };
+}
+

src/modules/domestic-voc/adapters/jd-response.ts

@@ -0,0 +1,104 @@
+export type JsonRecord = Record<string, unknown>;
+
+export class GatewayPayloadError extends Error {
+  constructor(message: string, public readonly retryable = false) {
+    super(message);
+    this.name = 'GatewayPayloadError';
+  }
+}
+
+export function isRecord(value: unknown): value is JsonRecord {
+  return Boolean(value) && typeof value === 'object' && !Array.isArray(value);
+}
+
+export function assertGatewayPayloadReady(value: unknown): void {
+  let current = value;
+  for (let depth = 0; depth < 4 && isRecord(current); depth += 1) {
+    const code = Number(current.code);
+    const message = String(current.message ?? current.mess ?? '');
+    if (Number.isFinite(code) && code !== 0 && code !== 200) {
+      const retryable = code === 301 || /collect failed|send request again|pending|retry/i.test(message);
+      throw new GatewayPayloadError(retryable ? 'Gateway data is not ready' : 'Gateway returned an unsuccessful result', retryable);
+    }
+    current = current.data;
+  }
+}
+
+export function unwrapGatewayPayload(value: unknown): unknown {
+  assertGatewayPayloadReady(value);
+  let current = value;
+  for (let depth = 0; depth < 4 && isRecord(current); depth += 1) {
+    const keys = Object.keys(current);
+    const isEnvelope = 'data' in current && keys.every((key) => [
+      'code', 'data', 'message', 'mess', 'success', 'requestId', 'traceId',
+    ].includes(key));
+    if (!isEnvelope) break;
+    current = current.data;
+  }
+  return current;
+}
+
+export function walkRecords(value: unknown, maxDepth = 5): JsonRecord[] {
+  const records: JsonRecord[] = [];
+  const queue: Array<{ value: unknown; depth: number }> = [{ value, depth: 0 }];
+  const seen = new Set<object>();
+  while (queue.length) {
+    const item = queue.shift();
+    if (!item || item.depth > maxDepth || !item.value || typeof item.value !== 'object') continue;
+    if (seen.has(item.value as object)) continue;
+    seen.add(item.value as object);
+    if (Array.isArray(item.value)) {
+      for (const child of item.value) queue.push({ value: child, depth: item.depth + 1 });
+      continue;
+    }
+    const record = item.value as JsonRecord;
+    records.push(record);
+    for (const child of Object.values(record)) queue.push({ value: child, depth: item.depth + 1 });
+  }
+  return records;
+}
+
+export function firstValue(record: JsonRecord, keys: string[]): unknown {
+  for (const key of keys) {
+    const value = record[key];
+    if (value !== undefined && value !== null && value !== '') return value;
+  }
+  return undefined;
+}
+
+export function firstString(record: JsonRecord, keys: string[]): string {
+  const value = firstValue(record, keys);
+  if (typeof value === 'string') return value.trim();
+  if (typeof value === 'number' || typeof value === 'bigint') return String(value);
+  if (isRecord(value)) {
+    const nested = firstValue(value, ['name', 'title', 'text', 'value']);
+    return typeof nested === 'string' ? nested.trim() : '';
+  }
+  return '';
+}
+
+export function firstNumber(record: JsonRecord, keys: string[]): number | null {
+  const value = firstValue(record, keys);
+  const numeric = Number(value);
+  return Number.isFinite(numeric) ? numeric : null;
+}
+
+export function findBestRecord(value: unknown, preferredKeys: string[]): JsonRecord | null {
+  const records = walkRecords(value);
+  let best: { record: JsonRecord; score: number } | null = null;
+  for (const record of records) {
+    const score = preferredKeys.reduce((total, key) => total + (record[key] !== undefined ? 1 : 0), 0);
+    if (score > 0 && (!best || score > best.score)) best = { record, score };
+  }
+  return best?.record ?? (isRecord(value) ? value : null);
+}
+
+export function parseDate(value: unknown): string | null {
+  if (value === undefined || value === null || value === '') return null;
+  const numeric = typeof value === 'number' ? value : Number.NaN;
+  const date = Number.isFinite(numeric)
+    ? new Date(numeric < 10_000_000_000 ? numeric * 1_000 : numeric)
+    : new Date(String(value));
+  return Number.isNaN(date.getTime()) ? null : date.toISOString();
+}
+

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

@@ -0,0 +1,92 @@
+import { makeReviewKey } from '../domain/identity.js';
+import type { DomesticReview } from '../../../types/domestic-dataset.js';
+import {
+  firstNumber,
+  firstString,
+  firstValue,
+  isRecord,
+  parseDate,
+  type JsonRecord,
+  unwrapGatewayPayload,
+  walkRecords,
+} from './jd-response.js';
+
+export const JD_PRODUCT_COMMENTS_PATH = 'jd/get-item-comments/v1';
+
+export interface JdReviewRecord extends DomesticReview {
+  reviewKey: string;
+  rawPayload: JsonRecord;
+}
+
+export interface JdReviewPage {
+  reviews: JdReviewRecord[];
+  hasNextPage: boolean;
+}
+
+const REVIEW_ARRAY_KEYS = ['comments', 'commentList', 'reviews', 'reviewList', 'items', 'list'];
+const REVIEW_CONTENT_KEYS = ['content', 'commentData', 'comment', 'text', 'description', 'reviewContent'];
+
+function findReviewRows(payload: unknown): JsonRecord[] {
+  if (Array.isArray(payload)) return payload.filter(isRecord);
+  for (const record of walkRecords(payload)) {
+    for (const key of REVIEW_ARRAY_KEYS) {
+      const value = record[key];
+      if (!Array.isArray(value)) continue;
+      const rows = value.filter(isRecord);
+      if (!rows.length) return [];
+      if (rows.some((row) => REVIEW_CONTENT_KEYS.some((field) => row[field] !== undefined))) return rows;
+    }
+  }
+  return [];
+}
+
+function normalizeRating(value: number | null): number {
+  if (value === null || value < 0 || value > 5) return 0;
+  return value;
+}
+
+function detectNextPage(payload: unknown, currentPage: number): boolean {
+  for (const record of walkRecords(payload, 3)) {
+    const hasMore = firstValue(record, ['hasMore', 'hasNext', 'hasNextPage']);
+    if (typeof hasMore === 'boolean') return hasMore;
+    const nextPage = firstNumber(record, ['nextPage', 'nextPageNum']);
+    if (nextPage !== null) return nextPage > currentPage;
+    const totalPages = firstNumber(record, ['totalPage', 'totalPages', 'pageCount', 'maxPage']);
+    if (totalPages !== null) return currentPage < totalPages;
+  }
+  return false;
+}
+
+export function adaptJdReviewResponse(
+  response: unknown,
+  productId: string,
+  currentPage: number,
+): JdReviewPage {
+  const payload = unwrapGatewayPayload(response);
+  const reviews = findReviewRows(payload).flatMap((row): JdReviewRecord[] => {
+    const content = firstString(row, REVIEW_CONTENT_KEYS);
+    if (!content) return [];
+    const reviewId = firstString(row, ['commentId', 'reviewId', 'id', 'guid', 'commentIdStr']);
+    const reviewDate = parseDate(firstValue(row, [
+      'creationTime', 'createdAt', 'commentTime', 'referenceTime', 'date', 'time',
+    ]));
+    const reviewKey = makeReviewKey({
+      platform: 'jd',
+      productId,
+      ...(reviewId ? { reviewId } : {}),
+      content,
+      ...(reviewDate ? { reviewDate } : {}),
+    });
+    return [{
+      productId,
+      reviewId: reviewId || reviewKey,
+      rating: normalizeRating(firstNumber(row, ['score', 'rating', 'star', 'starLevel', 'productScore'])),
+      content,
+      ...(reviewDate ? { reviewDate } : {}),
+      reviewKey,
+      rawPayload: row,
+    }];
+  });
+  return { reviews, hasNextPage: detectNextPage(payload, currentPage) };
+}
+

src/modules/domestic-voc/jobs/sync-worker.ts

@@ -1,4 +1,5 @@
 import type { Pool } from 'pg';
+import type { JdSyncService } from '../services/jd-sync.service.js';
 
 export interface ClaimedSyncJob {
   internalId: string;
@@ -8,6 +9,7 @@ export interface ClaimedSyncJob {
   scopes: string[];
   productIds: string[];
   attempts: number;
+  maxAttempts: number;
 }
 
 interface ClaimedSyncJobRow {
@@ -18,6 +20,7 @@ interface ClaimedSyncJobRow {
   scopes: string[];
   product_ids: string[];
   attempts: number;
+  max_attempts: number;
 }
 
 export async function claimNextSyncJob(pool: Pool, workerId: string): Promise<ClaimedSyncJob | null> {
@@ -45,7 +48,8 @@ export async function claimNextSyncJob(pool: Pool, workerId: string): Promise<Cl
       job.platform,
       job.scopes,
       job.product_ids,
-      job.attempts
+      job.attempts,
+      job.max_attempts
   `, [workerId]);
   const row = result.rows[0];
   if (!row) return null;
@@ -57,6 +61,51 @@ export async function claimNextSyncJob(pool: Pool, workerId: string): Promise<Cl
     scopes: row.scopes,
     productIds: row.product_ids,
     attempts: row.attempts,
+    maxAttempts: row.max_attempts,
   };
 }
 
+export function startSyncWorker(input: {
+  pool: Pool;
+  processor: JdSyncService;
+  pollMs: number;
+  workerId?: string;
+}): { stop: () => Promise<void> } {
+  const workerId = input.workerId ?? `saas-voc-${process.pid}`;
+  let stopped = false;
+  let running = false;
+  let timer: NodeJS.Timeout | undefined;
+  let activeIteration: Promise<void> | null = null;
+
+  const schedule = () => {
+    if (stopped) return;
+    timer = setTimeout(() => void tick(), input.pollMs);
+    timer.unref();
+  };
+  const tick = async () => {
+    if (running || stopped) return;
+    running = true;
+    activeIteration = (async () => {
+      try {
+        const job = await claimNextSyncJob(input.pool, workerId);
+        if (job) await input.processor.process(job);
+      } catch {
+        console.error('[sync-worker] worker iteration failed');
+      } finally {
+        running = false;
+        activeIteration = null;
+        schedule();
+      }
+    })();
+    await activeIteration;
+  };
+
+  void tick();
+  return {
+    async stop() {
+      stopped = true;
+      if (timer) clearTimeout(timer);
+      if (activeIteration) await activeIteration;
+    },
+  };
+}

src/modules/domestic-voc/repositories/voc-ingestion.repository.ts

@@ -0,0 +1,198 @@
+import type { Queryable } from '../../../db/types.js';
+import type { JdProductRecord } from '../adapters/jd-product.adapter.js';
+import type { JdReviewRecord } from '../adapters/jd-review.adapter.js';
+
+export type SyncJobFinalStatus = 'completed' | 'partial' | 'failed';
+
+export interface SyncPersistence {
+  upsertProduct(workspaceId: string, product: JdProductRecord): Promise<void>;
+  ensureProductStub(workspaceId: string, platform: string, productId: string): Promise<void>;
+  upsertReviews(workspaceId: string, platform: string, productId: string, reviews: JdReviewRecord[]): Promise<number>;
+  setJobProgress(jobInternalId: string, progress: number): Promise<void>;
+  finishJob(jobInternalId: string, status: SyncJobFinalStatus, errorSummary?: string): Promise<void>;
+  requeueJob(jobInternalId: string, errorSummary: string): Promise<void>;
+  addJobEvent(input: {
+    jobInternalId: string;
+    level: 'info' | 'warning' | 'error';
+    eventType: string;
+    message: string;
+    details?: Record<string, unknown>;
+  }): Promise<void>;
+}
+
+function chunks<T>(values: T[], size = 100): T[][] {
+  const output: T[][] = [];
+  for (let index = 0; index < values.length; index += size) output.push(values.slice(index, index + size));
+  return output;
+}
+
+function valuesClause(rows: unknown[][]): { sql: string; values: unknown[] } {
+  const values: unknown[] = [];
+  const sql = rows.map((row) => {
+    const placeholders = row.map((value) => {
+      values.push(value);
+      return `$${values.length}`;
+    });
+    return `(${placeholders.join(', ')})`;
+  }).join(', ');
+  return { sql, values };
+}
+
+export class VocIngestionRepository implements SyncPersistence {
+  constructor(private readonly database: Queryable) {}
+
+  async upsertProduct(workspacePublicId: string, product: JdProductRecord): Promise<void> {
+    const result = await this.database.query(`
+      INSERT INTO voc.product (
+        workspace_id, platform, product_id, product_key, role, brand, title, model,
+        category_1, category_2, category_3, source, raw_payload
+      )
+      SELECT id, $2, $3, $4, $5, $6, $7, $8, $9, $10, $11, $12, $13::jsonb
+      FROM voc.workspace
+      WHERE public_id = $1 AND status = 'active'
+      ON CONFLICT (workspace_id, platform, product_id) DO UPDATE SET
+        product_key = EXCLUDED.product_key,
+        role = EXCLUDED.role,
+        brand = COALESCE(NULLIF(EXCLUDED.brand, ''), voc.product.brand),
+        title = COALESCE(NULLIF(EXCLUDED.title, ''), voc.product.title),
+        model = COALESCE(NULLIF(EXCLUDED.model, ''), voc.product.model),
+        category_1 = COALESCE(NULLIF(EXCLUDED.category_1, ''), voc.product.category_1),
+        category_2 = COALESCE(NULLIF(EXCLUDED.category_2, ''), voc.product.category_2),
+        category_3 = COALESCE(NULLIF(EXCLUDED.category_3, ''), voc.product.category_3),
+        source = EXCLUDED.source,
+        raw_payload = EXCLUDED.raw_payload,
+        updated_at = now()
+      RETURNING id
+    `, [
+      workspacePublicId,
+      product.platform,
+      product.productId,
+      product.productKey,
+      product.role,
+      product.brand,
+      product.title,
+      product.model,
+      product.category1,
+      product.category2,
+      product.category3,
+      product.source,
+      JSON.stringify(product.rawPayload),
+    ]);
+    if (!result.rowCount) throw new Error(`Workspace not found: ${workspacePublicId}`);
+  }
+
+  async ensureProductStub(workspacePublicId: string, platform: string, productId: string): Promise<void> {
+    const result = await this.database.query(`
+      INSERT INTO voc.product (workspace_id, platform, product_id, product_key, role, source)
+      SELECT id, $2, $3, $4, 'own', 'sync_stub'
+      FROM voc.workspace
+      WHERE public_id = $1 AND status = 'active'
+      ON CONFLICT (workspace_id, platform, product_id) DO NOTHING
+      RETURNING id
+    `, [workspacePublicId, platform, productId, `${platform}:${productId}`]);
+    if (!result.rowCount) {
+      const workspace = await this.database.query(
+        "SELECT id FROM voc.workspace WHERE public_id = $1 AND status = 'active'",
+        [workspacePublicId],
+      );
+      if (!workspace.rowCount) throw new Error(`Workspace not found: ${workspacePublicId}`);
+    }
+  }
+
+  async upsertReviews(
+    workspacePublicId: string,
+    platform: string,
+    naturalProductId: string,
+    reviews: JdReviewRecord[],
+  ): Promise<number> {
+    if (!reviews.length) return 0;
+    const product = await this.database.query<{ workspace_id: string; product_internal_id: string }>(`
+      SELECT product.workspace_id, product.id AS product_internal_id
+      FROM voc.product product
+      JOIN voc.workspace workspace ON workspace.id = product.workspace_id
+      WHERE workspace.public_id = $1 AND product.platform = $2 AND product.product_id = $3
+    `, [workspacePublicId, platform, naturalProductId]);
+    const target = product.rows[0];
+    if (!target) throw new Error(`Product not found: ${platform}:${naturalProductId}`);
+
+    let affected = 0;
+    for (const batch of chunks(reviews)) {
+      const values = valuesClause(batch.map((review) => [
+        target.workspace_id,
+        target.product_internal_id,
+        platform,
+        review.reviewId || null,
+        review.reviewKey,
+        review.rating,
+        review.content,
+        review.reviewDate || null,
+        JSON.stringify(review.rawPayload),
+      ]));
+      const result = await this.database.query(`
+        INSERT INTO voc.review (
+          workspace_id, product_id, platform, source_review_id, review_key,
+          rating, content, review_date, raw_payload
+        ) VALUES ${values.sql}
+        ON CONFLICT (workspace_id, platform, review_key) DO UPDATE SET
+          rating = EXCLUDED.rating,
+          content = EXCLUDED.content,
+          review_date = EXCLUDED.review_date,
+          raw_payload = EXCLUDED.raw_payload,
+          updated_at = now()
+      `, values.values);
+      affected += result.rowCount ?? batch.length;
+    }
+    return affected;
+  }
+
+  async setJobProgress(jobInternalId: string, progress: number): Promise<void> {
+    await this.database.query(`
+      UPDATE voc.sync_job
+      SET progress = $2, updated_at = now()
+      WHERE id = $1 AND status = 'processing'
+    `, [jobInternalId, Math.max(0, Math.min(100, Math.round(progress)))]);
+  }
+
+  async finishJob(jobInternalId: string, status: SyncJobFinalStatus, errorSummary = ''): Promise<void> {
+    await this.database.query(`
+      UPDATE voc.sync_job
+      SET status = $2,
+          progress = 100,
+          error_summary = NULLIF($3, ''),
+          completed_at = now(),
+          updated_at = now()
+      WHERE id = $1
+    `, [jobInternalId, status, errorSummary.slice(0, 1_000)]);
+  }
+
+  async requeueJob(jobInternalId: string, errorSummary: string): Promise<void> {
+    await this.database.query(`
+      UPDATE voc.sync_job
+      SET status = 'pending',
+          progress = 0,
+          worker_id = NULL,
+          error_summary = $2,
+          updated_at = now()
+      WHERE id = $1 AND status = 'processing'
+    `, [jobInternalId, errorSummary.slice(0, 1_000)]);
+  }
+
+  async addJobEvent(input: {
+    jobInternalId: string;
+    level: 'info' | 'warning' | 'error';
+    eventType: string;
+    message: string;
+    details?: Record<string, unknown>;
+  }): Promise<void> {
+    await this.database.query(`
+      INSERT INTO voc.sync_job_event (sync_job_id, level, event_type, message, details)
+      VALUES ($1, $2, $3, $4, $5::jsonb)
+    `, [
+      input.jobInternalId,
+      input.level,
+      input.eventType,
+      input.message.slice(0, 1_000),
+      JSON.stringify(input.details ?? {}),
+    ]);
+  }
+}

src/modules/domestic-voc/services/jd-sync.service.ts

@@ -0,0 +1,169 @@
+import { adaptJdProductResponse, JD_PRODUCT_DETAIL_PATH } from '../adapters/jd-product.adapter.js';
+import { GatewayPayloadError } from '../adapters/jd-response.js';
+import {
+  adaptJdReviewResponse,
+  JD_PRODUCT_COMMENTS_PATH,
+  type JdReviewRecord,
+} from '../adapters/jd-review.adapter.js';
+import type { ClaimedSyncJob } from '../jobs/sync-worker.js';
+import type { SyncPersistence } from '../repositories/voc-ingestion.repository.js';
+import { FmodeRequestError } from '../upstream/fmode-client.js';
+
+export interface GatewayRequestClient {
+  request<T>(
+    path: string,
+    init?: { method?: 'GET' | 'POST'; params?: Record<string, unknown>; refresh?: boolean },
+  ): Promise<T>;
+}
+
+interface SafeFailure {
+  message: string;
+  retryable: boolean;
+}
+
+export function classifySyncFailure(error: unknown): SafeFailure {
+  if (error instanceof GatewayPayloadError) {
+    return { message: error.message, retryable: error.retryable };
+  }
+  if (error instanceof FmodeRequestError) {
+    if (error.status === 401) return { message: 'Data gateway authentication failed', retryable: false };
+    if (error.status === 402) return { message: 'Data gateway quota is unavailable', retryable: false };
+    if (error.status === 403) return { message: 'Data gateway permission is unavailable', retryable: false };
+    if (error.status === 408 || error.status === 429 || error.retryable) {
+      return { message: 'Data gateway is temporarily unavailable', retryable: true };
+    }
+    return { message: 'Data gateway request was rejected', retryable: false };
+  }
+  return { message: 'Domestic VOC sync failed', retryable: false };
+}
+
+export class JdSyncService {
+  constructor(
+    private readonly gateway: GatewayRequestClient,
+    private readonly persistence: SyncPersistence,
+    private readonly reviewMaxPages = 1,
+  ) {}
+
+  async process(job: ClaimedSyncJob): Promise<void> {
+    if (job.platform !== 'jd') {
+      await this.persistence.finishJob(job.internalId, 'failed', 'Unsupported domestic ecommerce platform');
+      return;
+    }
+
+    const scopes = new Set(job.scopes);
+    const operationsPerProduct = Number(scopes.has('product')) + Number(scopes.has('reviews'));
+    const totalOperations = Math.max(1, job.productIds.length * operationsPerProduct);
+    let processedOperations = 0;
+    let successfulOperations = 0;
+    const failures: SafeFailure[] = [];
+
+    await this.persistence.addJobEvent({
+      jobInternalId: job.internalId,
+      level: 'info',
+      eventType: 'sync_started',
+      message: 'JD sync started through the company data gateway',
+      details: { productCount: job.productIds.length, scopes: [...scopes] },
+    });
+
+    for (const productId of job.productIds) {
+      if (scopes.has('product')) {
+        try {
+          const response = await this.gateway.request<unknown>(JD_PRODUCT_DETAIL_PATH, {
+            params: { itemId: productId },
+          });
+          const product = adaptJdProductResponse(response, productId);
+          await this.persistence.upsertProduct(job.workspaceId, product);
+          successfulOperations += 1;
+          await this.persistence.addJobEvent({
+            jobInternalId: job.internalId,
+            level: 'info',
+            eventType: 'product_synced',
+            message: 'Product detail synchronized',
+            details: { productId },
+          });
+        } catch (error) {
+          await this.recordFailure(job, productId, 'product', error, failures);
+        } finally {
+          processedOperations += 1;
+          await this.persistence.setJobProgress(job.internalId, processedOperations / totalOperations * 100);
+        }
+      }
+
+      if (scopes.has('reviews')) {
+        try {
+          await this.persistence.ensureProductStub(job.workspaceId, job.platform, productId);
+          const reviews: JdReviewRecord[] = [];
+          let pagesFetched = 0;
+          for (let page = 1; page <= this.reviewMaxPages; page += 1) {
+            const response = await this.gateway.request<unknown>(JD_PRODUCT_COMMENTS_PATH, {
+              params: { itemId: productId, page },
+            });
+            const adapted = adaptJdReviewResponse(response, productId, page);
+            reviews.push(...adapted.reviews);
+            pagesFetched = page;
+            if (!adapted.hasNextPage) break;
+          }
+          const written = await this.persistence.upsertReviews(job.workspaceId, job.platform, productId, reviews);
+          successfulOperations += 1;
+          await this.persistence.addJobEvent({
+            jobInternalId: job.internalId,
+            level: 'info',
+            eventType: 'reviews_synced',
+            message: 'Product reviews synchronized',
+            details: { productId, pagesFetched, reviewCount: written },
+          });
+        } catch (error) {
+          await this.recordFailure(job, productId, 'reviews', error, failures);
+        } finally {
+          processedOperations += 1;
+          await this.persistence.setJobProgress(job.internalId, processedOperations / totalOperations * 100);
+        }
+      }
+    }
+
+    const errorSummary = [...new Set(failures.map((failure) => failure.message))].join('; ');
+    if (failures.length && failures.every((failure) => failure.retryable) && job.attempts < job.maxAttempts) {
+      await this.persistence.requeueJob(job.internalId, errorSummary);
+      await this.persistence.addJobEvent({
+        jobInternalId: job.internalId,
+        level: 'warning',
+        eventType: 'sync_requeued',
+        message: 'Sync will retry after a temporary gateway failure',
+        details: { attempt: job.attempts, maxAttempts: job.maxAttempts },
+      });
+      return;
+    }
+
+    const status = failures.length === 0
+      ? 'completed'
+      : successfulOperations > 0
+        ? 'partial'
+        : 'failed';
+    await this.persistence.finishJob(job.internalId, status, errorSummary);
+    await this.persistence.addJobEvent({
+      jobInternalId: job.internalId,
+      level: failures.length ? 'warning' : 'info',
+      eventType: 'sync_finished',
+      message: status === 'completed' ? 'JD sync completed' : 'JD sync finished with unavailable data',
+      details: { status, successfulOperations, failedOperations: failures.length },
+    });
+  }
+
+  private async recordFailure(
+    job: ClaimedSyncJob,
+    productId: string,
+    scope: 'product' | 'reviews',
+    error: unknown,
+    failures: SafeFailure[],
+  ): Promise<void> {
+    const failure = classifySyncFailure(error);
+    failures.push(failure);
+    await this.persistence.addJobEvent({
+      jobInternalId: job.internalId,
+      level: failure.retryable ? 'warning' : 'error',
+      eventType: `${scope}_sync_failed`,
+      message: failure.message,
+      details: { productId, scope, retryable: failure.retryable },
+    });
+  }
+}

src/modules/domestic-voc/services/snapshot.service.ts

@@ -161,7 +161,7 @@ export class SnapshotService {
         SELECT id, platform, product_id, product_key, role, brand, title, model,
                category_1, category_2, category_3, source
         FROM voc.product
-        WHERE workspace_id = $1 AND platform = $2 AND source <> 'relation_stub'
+        WHERE workspace_id = $1 AND platform = $2 AND source NOT IN ('relation_stub', 'sync_stub')
         ORDER BY id
       `, [workspace.id, platform]),
       this.database.query<MetricRow>(`

src/server.ts

@@ -5,6 +5,10 @@ import { createApp } from './app.js';
 import { loadConfig } from './config/env.js';
 import { createParseServer } from './config/parse.js';
 import { createDatabasePool } from './db/pool.js';
+import { startSyncWorker } from './modules/domestic-voc/jobs/sync-worker.js';
+import { VocIngestionRepository } from './modules/domestic-voc/repositories/voc-ingestion.repository.js';
+import { JdSyncService } from './modules/domestic-voc/services/jd-sync.service.js';
+import { FmodeVocEcommerceClient } from './modules/domestic-voc/upstream/fmode-client.js';
 
 async function main(): Promise<void> {
   const config = loadConfig();
@@ -16,6 +20,12 @@ async function main(): Promise<void> {
     parseApp: parseServer.app as unknown as RequestHandler,
   });
   const server = createServer(app);
+  const gateway = new FmodeVocEcommerceClient(config.fmode);
+  const ingestion = new VocIngestionRepository(pool);
+  const processor = new JdSyncService(gateway, ingestion, config.worker.reviewMaxPages);
+  const worker = config.worker.enabled
+    ? startSyncWorker({ pool, processor, pollMs: config.worker.pollMs })
+    : null;
 
   server.listen(config.port, config.host, () => {
     console.log(`[server] listening on http://${config.host}:${config.port}`);
@@ -23,6 +33,7 @@ async function main(): Promise<void> {
 
   const shutdown = async (signal: string) => {
     console.log(`[server] received ${signal}; shutting down`);
+    await worker?.stop();
     server.close(async () => {
       await parseServer.handleShutdown();
       await pool.end();

test/env.test.ts

@@ -46,5 +46,7 @@ test('loadConfig normalizes URLs and CORS origins', () => {
 
   assert.equal(config.fmode.baseUrl, 'http://127.0.0.1:3000/api/voc-e-commerce');
   assert.equal(config.database.migrationUrl, validEnvironment.MIGRATION_DATABASE_URL);
+  assert.equal(config.worker.enabled, true);
+  assert.equal(config.worker.reviewMaxPages, 1);
   assert.deepEqual(config.corsOrigins, ['http://127.0.0.1:4300', 'http://localhost:4200']);
 });

test/jd-adapters.test.ts

@@ -0,0 +1,71 @@
+import assert from 'node:assert/strict';
+import test from 'node:test';
+import { adaptJdProductResponse } from '../src/modules/domestic-voc/adapters/jd-product.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';
+
+test('JD product adapter unwraps the company gateway envelope', () => {
+  const product = adaptJdProductResponse({
+    code: 200,
+    data: {
+      code: 200,
+      data: {
+        item: {
+          itemId: '11266507445',
+          itemName: 'Demashi commercial cooker',
+          brandName: 'Demashi',
+          model: '35P6-CM1',
+          firstCategoryName: 'Commercial appliances',
+          thirdCategoryName: 'Commercial cooker',
+        },
+      },
+    },
+  }, '11266507445');
+
+  assert.equal(product.productId, '11266507445');
+  assert.equal(product.productKey, 'jd:11266507445');
+  assert.equal(product.title, 'Demashi commercial cooker');
+  assert.equal(product.brand, 'Demashi');
+  assert.equal(product.model, '35P6-CM1');
+  assert.equal(product.source, 'fmode_gateway');
+});
+
+test('JD review adapter maps evidence and pagination without reviewer identity', () => {
+  const page = adaptJdReviewResponse({
+    code: 200,
+    data: {
+      code: 200,
+      data: {
+        comments: [{
+          commentId: 'comment-1',
+          score: 5,
+          content: 'Easy to clean and heats quickly.',
+          creationTime: '2026-07-20T10:00:00+08:00',
+          nickname: 'not-exposed-by-snapshot',
+        }],
+        pageCount: 2,
+      },
+    },
+  }, '11266507445', 1);
+
+  assert.equal(page.reviews.length, 1);
+  assert.equal(page.reviews[0]?.reviewId, 'comment-1');
+  assert.equal(page.reviews[0]?.rating, 5);
+  assert.equal(page.reviews[0]?.reviewDate, '2026-07-20T02:00:00.000Z');
+  assert.match(page.reviews[0]?.reviewKey ?? '', /^jd:11266507445:/);
+  assert.equal(page.hasNextPage, true);
+});
+
+test('JD adapters classify pending collection as retryable without raw body leakage', () => {
+  assert.throws(
+    () => adaptJdProductResponse({ code: 301, message: 'COLLECT FAILED, SEND REQUEST AGAIN', data: null }, '1'),
+    (error: unknown) => {
+      assert.ok(error instanceof GatewayPayloadError);
+      assert.equal(error.retryable, true);
+      assert.equal(error.message, 'Gateway data is not ready');
+      assert.doesNotMatch(error.message, /COLLECT FAILED/);
+      return true;
+    },
+  );
+});
+

test/jd-sync.service.test.ts

@@ -0,0 +1,110 @@
+import assert from 'node:assert/strict';
+import test from 'node:test';
+import type { ClaimedSyncJob } from '../src/modules/domestic-voc/jobs/sync-worker.js';
+import type {
+  SyncJobFinalStatus,
+  SyncPersistence,
+} from '../src/modules/domestic-voc/repositories/voc-ingestion.repository.js';
+import { JdSyncService, type GatewayRequestClient } from '../src/modules/domestic-voc/services/jd-sync.service.js';
+import { FmodeRequestError } from '../src/modules/domestic-voc/upstream/fmode-client.js';
+
+class FakePersistence implements SyncPersistence {
+  products = 0;
+  reviews = 0;
+  progress: number[] = [];
+  finished: Array<{ status: SyncJobFinalStatus; errorSummary?: string }> = [];
+  requeued: string[] = [];
+  events: Array<{ eventType: string; message: string }> = [];
+
+  async upsertProduct() { this.products += 1; }
+  async ensureProductStub() {}
+  async upsertReviews(_workspaceId: string, _platform: string, _productId: string, reviews: unknown[]) {
+    this.reviews += reviews.length;
+    return reviews.length;
+  }
+  async setJobProgress(_jobInternalId: string, progress: number) { this.progress.push(progress); }
+  async finishJob(_jobInternalId: string, status: SyncJobFinalStatus, errorSummary?: string) {
+    this.finished.push({ status, ...(errorSummary ? { errorSummary } : {}) });
+  }
+  async requeueJob(_jobInternalId: string, errorSummary: string) { this.requeued.push(errorSummary); }
+  async addJobEvent(input: { eventType: string; message: string }) {
+    this.events.push({ eventType: input.eventType, message: input.message });
+  }
+}
+
+function job(overrides: Partial<ClaimedSyncJob> = {}): ClaimedSyncJob {
+  return {
+    internalId: '1',
+    publicId: '11111111-1111-4111-8111-111111111111',
+    workspaceId: 'demashi',
+    platform: 'jd',
+    scopes: ['product', 'reviews'],
+    productIds: ['11266507445'],
+    attempts: 1,
+    maxAttempts: 3,
+    ...overrides,
+  };
+}
+
+test('JD sync calls only registered company gateway paths and completes', async () => {
+  const calls: Array<{ path: string; params: Record<string, unknown> | undefined }> = [];
+  const gateway: GatewayRequestClient = {
+    async request(path, init) {
+      calls.push({ path, params: init?.params });
+      if (path === 'jd/get-item-detail/v1') {
+        return { code: 200, data: { code: 200, data: { item: { itemId: '11266507445', itemName: 'Product' } } } } as never;
+      }
+      return {
+        code: 200,
+        data: { code: 200, data: { comments: [{ commentId: 'c1', score: 5, content: 'Useful evidence.' }] } },
+      } as never;
+    },
+  };
+  const persistence = new FakePersistence();
+
+  await new JdSyncService(gateway, persistence, 1).process(job());
+
+  assert.deepEqual(calls, [
+    { path: 'jd/get-item-detail/v1', params: { itemId: '11266507445' } },
+    { path: 'jd/get-item-comments/v1', params: { itemId: '11266507445', page: 1 } },
+  ]);
+  assert.equal(persistence.products, 1);
+  assert.equal(persistence.reviews, 1);
+  assert.deepEqual(persistence.finished, [{ status: 'completed' }]);
+  assert.equal(persistence.progress.at(-1), 100);
+});
+
+test('JD sync returns a safe partial status when reviews are not permitted', async () => {
+  const gateway: GatewayRequestClient = {
+    async request(path) {
+      if (path === 'jd/get-item-detail/v1') {
+        return { code: 200, data: { item: { itemId: '11266507445', itemName: 'Product' } } } as never;
+      }
+      throw new FmodeRequestError('Bearer secret-value raw 403 response', 403, false);
+    },
+  };
+  const persistence = new FakePersistence();
+
+  await new JdSyncService(gateway, persistence).process(job());
+
+  assert.deepEqual(persistence.finished, [{
+    status: 'partial',
+    errorSummary: 'Data gateway permission is unavailable',
+  }]);
+  assert.doesNotMatch(JSON.stringify(persistence.events), /secret-value|Bearer|raw 403/);
+});
+
+test('JD sync requeues temporary gateway failures while attempts remain', async () => {
+  const gateway: GatewayRequestClient = {
+    async request() {
+      throw new FmodeRequestError('network internals', undefined, true);
+    },
+  };
+  const persistence = new FakePersistence();
+
+  await new JdSyncService(gateway, persistence).process(job({ scopes: ['product'] }));
+
+  assert.deepEqual(persistence.finished, []);
+  assert.deepEqual(persistence.requeued, ['Data gateway is temporarily unavailable']);
+});
+