Procházet zdrojové kódy

feat: harden reusable SaaS operations

gangvy před 2 měsíci
rodič
revize
b781fd1443

+ 5 - 0
.env.example

@@ -1,4 +1,8 @@
 NODE_ENV=development
 NODE_ENV=development
+LOCAL_HOST=127.0.0.1
+LOCAL_PORT=4400
+LOCAL_WORKSPACE_ID=demashi
+LOCAL_CORS_ORIGINS=http://localhost:4200,http://127.0.0.1:4200
 HOST=127.0.0.1
 HOST=127.0.0.1
 PORT=4400
 PORT=4400
 DATABASE_URL=
 DATABASE_URL=
@@ -24,5 +28,6 @@ FMODE_TIMEOUT_MS=30000
 FMODE_RETRIES=2
 FMODE_RETRIES=2
 SYNC_WORKER_ENABLED=true
 SYNC_WORKER_ENABLED=true
 SYNC_WORKER_POLL_MS=2000
 SYNC_WORKER_POLL_MS=2000
+SYNC_JOB_STALE_AFTER_MS=900000
 JD_REVIEW_MAX_PAGES=1
 JD_REVIEW_MAX_PAGES=1
 CORS_ORIGINS=http://127.0.0.1:4300,http://localhost:4200
 CORS_ORIGINS=http://127.0.0.1:4300,http://localhost:4200

+ 10 - 2
README.md

@@ -22,6 +22,9 @@ Completed on 2026-07-23:
 - Parse session authentication for production and a fixed local identity for database-free development.
 - Parse session authentication for production and a fixed local identity for database-free development.
 - Cursor-paginated product, relation, review, import, sync-job, analysis, action, alert, and audit APIs.
 - Cursor-paginated product, relation, review, import, sync-job, analysis, action, alert, and audit APIs.
 - Pending analysis requests, action workflows, alerts, and auditable write operations without fabricated AI output.
 - Pending analysis requests, action workflows, alerts, and auditable write operations without fabricated AI output.
+- Configurable local workspace identity instead of a Demashi-only runtime branch.
+- Manual sync retry/cancel operations plus automatic recovery for stale processing jobs.
+- Owner-only owner changes, last-owner protection, and workspace-member action assignment validation.
 
 
 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 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.
 
 
@@ -76,6 +79,7 @@ npm run start:local
 Local demo `sync` requests create queryable in-memory completion records that validate whether requested products exist in the packaged dataset. They do not collect external data or persist anything. Production collection and persistence remain in `npm run dev`.
 Local demo `sync` requests create queryable in-memory completion records that validate whether requested products exist in the packaged dataset. They do not collect external data or persist anything. Production collection and persistence remain in `npm run dev`.
 
 
 Local mode injects the fixed `local-admin` owner. It is intentionally database-free and must not be internet-facing.
 Local mode injects the fixed `local-admin` owner. It is intentionally database-free and must not be internet-facing.
+Set `LOCAL_WORKSPACE_ID` when using a packaged dataset under a workspace other than `demashi`; omitted API workspace ids then resolve to that configured default.
 
 
 ### Database-backed runtime
 ### Database-backed runtime
 
 
@@ -141,6 +145,8 @@ Read status and frontend data:
 
 
 ```text
 ```text
 GET /api/domestic-voc/jobs/{jobId}
 GET /api/domestic-voc/jobs/{jobId}
+POST /api/domestic-voc/jobs/{jobId}/retry
+POST /api/domestic-voc/jobs/{jobId}/cancel
 GET /api/domestic-voc/snapshot?workspaceId=demashi&platform=jd
 GET /api/domestic-voc/snapshot?workspaceId=demashi&platform=jd
 ```
 ```
 
 
@@ -175,10 +181,12 @@ An empty database returns a valid empty dataset. It does not invent reviews, rat
 - Import and sync writers must use `INSERT ... ON CONFLICT` in batches.
 - Import and sync writers must use `INSERT ... ON CONFLICT` in batches.
 - Foreign-key columns and common workspace/date filters are indexed.
 - Foreign-key columns and common workspace/date filters are indexed.
 - Workers claim jobs with `FOR UPDATE SKIP LOCKED`.
 - Workers claim jobs with `FOR UPDATE SKIP LOCKED`.
+- Worker startup recovers processing jobs older than `SYNC_JOB_STALE_AFTER_MS`; exhausted jobs become failed and eligible jobs return to the queue.
 - External HTTP requests must run outside database transactions.
 - External HTTP requests must run outside database transactions.
 - Workspace membership is checked on every domestic and SaaS business route.
 - Workspace membership is checked on every domestic and SaaS business route.
 - Disabled workspaces cannot pass authorization even when a membership row remains active.
 - Disabled workspaces cannot pass authorization even when a membership row remains active.
-- Owner and admin can manage members and read audit history; analyst can run sync/analysis and manage actions/alerts; viewer is read-only.
+- Owner and admin can manage ordinary members and read audit history; only owners can change owner membership, and the final active owner is protected. Analyst can run sync/analysis and manage actions/alerts; viewer is read-only.
+- Action assignees must be active members of the same workspace.
 - Member, sync, analysis, action, and alert writes append an audit entry.
 - Member, sync, analysis, action, and alert writes append an audit entry.
 - Production deployments must use a pooled, least-privilege runtime role. Migration credentials should be separate from runtime credentials.
 - Production deployments must use a pooled, least-privilege runtime role. Migration credentials should be separate from runtime credentials.
 - The HTTP server never runs DDL automatically; migrations are an explicit deployment step.
 - The HTTP server never runs DDL automatically; migrations are an explicit deployment step.
@@ -209,7 +217,7 @@ npm audit --omit=dev
 Current result:
 Current result:
 
 
 - TypeScript build: passed.
 - TypeScript build: passed.
-- Unit, adapter, worker, local-demo, authentication, RBAC, cursor, workflow, audit, and HTTP contract tests: 25 passed on Node.js 22.13.0.
+- Unit, adapter, worker recovery, local-demo, authentication, RBAC, cursor, workflow, audit, and HTTP contract tests: 29 passed on Node.js 22.13.0.
 - Production dependency audit: 0 critical, 0 high, 13 moderate.
 - Production dependency audit: 0 critical, 0 high, 13 moderate.
 - Company gateway health: HTTP 200 on 2026-07-23.
 - Company gateway health: HTTP 200 on 2026-07-23.
 - Credentialed live JD product-detail request: passed through the company gateway for product `11266507445`; the adapter extracted the product id, title, and brand from the live double-`data` envelope.
 - Credentialed live JD product-detail request: passed through the company gateway for product `11266507445`; the adapter extracted the product id, title, and brand from the live double-`data` envelope.

+ 6 - 1
TASKS.md

@@ -20,6 +20,10 @@ Status date: 2026-07-23
 - [x] Add cursor-paginated product, relation, review, job, import, and platform workflow APIs.
 - [x] Add cursor-paginated product, relation, review, job, import, and platform workflow APIs.
 - [x] Add analysis requests, action workflows, alerts, and audit history without fake AI results.
 - [x] Add analysis requests, action workflows, alerts, and audit history without fake AI results.
 - [x] Cover local platform flows, viewer write denial, Parse session behavior, pagination, and audit writes.
 - [x] Cover local platform flows, viewer write denial, Parse session behavior, pagination, and audit writes.
+- [x] Make the database-free runtime reusable through a configurable local workspace id.
+- [x] Reject partial platform migrations through the health check.
+- [x] Add manual sync retry/cancel, local job events, and stale-worker job recovery.
+- [x] Protect owner membership and validate action assignees inside the workspace boundary.
 - [ ] Run migrations against a newly provisioned PostgreSQL database.
 - [ ] Run migrations against a newly provisioned PostgreSQL database.
 - [x] Smoke-test the local HTTP service under the supported Node 22 runtime.
 - [x] Smoke-test the local HTTP service under the supported Node 22 runtime.
 
 
@@ -53,7 +57,8 @@ Status date: 2026-07-23
 - [ ] Put TLS and same-origin reverse proxying in front of `/parse` and `/api/domestic-voc`.
 - [ ] Put TLS and same-origin reverse proxying in front of `/parse` and `/api/domestic-voc`.
 - [x] Add backend authentication, workspace authorization, and role permission boundaries.
 - [x] Add backend authentication, workspace authorization, and role permission boundaries.
 - [ ] Enable and integrate the frontend `AuthGuard` after Parse credentials and the first user are provisioned.
 - [ ] Enable and integrate the frontend `AuthGuard` after Parse credentials and the first user are provisioned.
-- [ ] Add backup, restore, retention, and failed-job replay procedures.
+- [x] Add failed-job replay/cancel APIs and automatic stale-job recovery.
+- [ ] Add database backup, restore, and retention procedures after infrastructure is provisioned.
 - [ ] Resolve or formally accept remaining moderate Parse transitive advisories.
 - [ ] Resolve or formally accept remaining moderate Parse transitive advisories.
 - [ ] Complete end-to-end acceptance with Demashi product detail and real review evidence.
 - [ ] Complete end-to-end acceptance with Demashi product detail and real review evidence.
 
 

+ 26 - 5
src/app.ts

@@ -32,14 +32,28 @@ export function createApp(input: {
 
 
   app.get('/health', async (_request, response) => {
   app.get('/health', async (_request, response) => {
     try {
     try {
-      const result = await input.pool.query<{ schema_ready: string | null }>(
-        "SELECT to_regclass('voc.workspace')::text AS schema_ready",
-      );
-      if (!result.rows[0]?.schema_ready) {
+      const result = await input.pool.query<Record<string, string | null>>(`
+        SELECT
+          to_regclass('voc.workspace')::text AS workspace,
+          to_regclass('voc.product')::text AS product,
+          to_regclass('voc.review')::text AS review,
+          to_regclass('voc.sync_job')::text AS sync_job,
+          to_regclass('voc.workspace_member')::text AS workspace_member,
+          to_regclass('voc.analysis_run')::text AS analysis_run,
+          to_regclass('voc.action_item')::text AS action_item,
+          to_regclass('voc.alert')::text AS alert,
+          to_regclass('voc.audit_log')::text AS audit_log
+      `);
+      const readiness = result.rows[0] ?? {};
+      const missingTables = Object.entries(readiness)
+        .filter((entry) => !entry[1])
+        .map((entry) => `voc.${entry[0]}`);
+      if (missingTables.length) {
         response.status(503).json({
         response.status(503).json({
           service: 'saas-voc-server',
           service: 'saas-voc-server',
           status: 'unavailable',
           status: 'unavailable',
           database: 'migration_required',
           database: 'migration_required',
+          missingTables,
           timestamp: new Date().toISOString(),
           timestamp: new Date().toISOString(),
         });
         });
         return;
         return;
@@ -69,7 +83,14 @@ export function createApp(input: {
   const jobs = new SyncJobRepository(input.pool);
   const jobs = new SyncJobRepository(input.pool);
   const sync = new SyncService(jobs);
   const sync = new SyncService(jobs);
   const snapshot = new SnapshotService(input.pool);
   const snapshot = new SnapshotService(input.pool);
-  app.use('/api/domestic-voc', createDomesticVocRouter({ jobs, sync, snapshot, catalog: platform, access }));
+  app.use('/api/domestic-voc', createDomesticVocRouter({
+    jobs,
+    sync,
+    snapshot,
+    catalog: platform,
+    access,
+    defaultWorkspaceId: input.config.auth.defaultWorkspaceId,
+  }));
   app.use('/api/saas', createSaasPlatformRouter({ repository: platform, access }));
   app.use('/api/saas', createSaasPlatformRouter({ repository: platform, access }));
 
 
   app.use((_request, response) => {
   app.use((_request, response) => {

+ 10 - 2
src/config/env.ts

@@ -1,11 +1,16 @@
 import { z } from 'zod';
 import { z } from 'zod';
 
 
+const optionalNonEmptyString = z.preprocess(
+  (value) => value === '' ? undefined : value,
+  z.string().min(1).optional(),
+);
+
 const environmentSchema = z.object({
 const environmentSchema = z.object({
   NODE_ENV: z.enum(['development', 'test', 'production']).default('development'),
   NODE_ENV: z.enum(['development', 'test', 'production']).default('development'),
   HOST: z.string().min(1).default('127.0.0.1'),
   HOST: z.string().min(1).default('127.0.0.1'),
   PORT: z.coerce.number().int().min(1).max(65535).default(4400),
   PORT: z.coerce.number().int().min(1).max(65535).default(4400),
   DATABASE_URL: z.string().min(1, 'DATABASE_URL is required'),
   DATABASE_URL: z.string().min(1, 'DATABASE_URL is required'),
-  MIGRATION_DATABASE_URL: z.string().min(1).optional(),
+  MIGRATION_DATABASE_URL: optionalNonEmptyString,
   DATABASE_POOL_MAX: z.coerce.number().int().min(1).max(50).default(10),
   DATABASE_POOL_MAX: z.coerce.number().int().min(1).max(50).default(10),
   DATABASE_IDLE_TIMEOUT_MS: z.coerce.number().int().min(1000).default(30_000),
   DATABASE_IDLE_TIMEOUT_MS: z.coerce.number().int().min(1000).default(30_000),
   DATABASE_STATEMENT_TIMEOUT_MS: z.coerce.number().int().min(1000).default(30_000),
   DATABASE_STATEMENT_TIMEOUT_MS: z.coerce.number().int().min(1000).default(30_000),
@@ -18,7 +23,7 @@ const environmentSchema = z.object({
   LOCAL_AUTH_USER_EMAIL: z.string().default('local-admin@localhost'),
   LOCAL_AUTH_USER_EMAIL: z.string().default('local-admin@localhost'),
   LOCAL_AUTH_USER_NAME: z.string().min(1).default('Local Admin'),
   LOCAL_AUTH_USER_NAME: z.string().min(1).default('Local Admin'),
   SAAS_DEFAULT_WORKSPACE_ID: z.string().min(1).default('demashi'),
   SAAS_DEFAULT_WORKSPACE_ID: z.string().min(1).default('demashi'),
-  SAAS_BOOTSTRAP_ADMIN_USER_ID: z.string().min(1).optional(),
+  SAAS_BOOTSTRAP_ADMIN_USER_ID: optionalNonEmptyString,
   SAAS_BOOTSTRAP_ADMIN_EMAIL: z.string().default(''),
   SAAS_BOOTSTRAP_ADMIN_EMAIL: z.string().default(''),
   SAAS_BOOTSTRAP_ADMIN_NAME: z.string().default(''),
   SAAS_BOOTSTRAP_ADMIN_NAME: z.string().default(''),
   FMODE_BASE_URL: z.url(),
   FMODE_BASE_URL: z.url(),
@@ -27,6 +32,7 @@ const environmentSchema = z.object({
   FMODE_RETRIES: z.coerce.number().int().min(0).max(5).default(2),
   FMODE_RETRIES: z.coerce.number().int().min(0).max(5).default(2),
   SYNC_WORKER_ENABLED: z.enum(['true', 'false']).default('true'),
   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),
   SYNC_WORKER_POLL_MS: z.coerce.number().int().min(500).max(60_000).default(2_000),
+  SYNC_JOB_STALE_AFTER_MS: z.coerce.number().int().min(60_000).max(86_400_000).default(900_000),
   JD_REVIEW_MAX_PAGES: z.coerce.number().int().min(1).max(10).default(1),
   JD_REVIEW_MAX_PAGES: z.coerce.number().int().min(1).max(10).default(1),
   CORS_ORIGINS: z.string().min(1),
   CORS_ORIGINS: z.string().min(1),
 });
 });
@@ -67,6 +73,7 @@ export type AppConfig = {
   worker: {
   worker: {
     enabled: boolean;
     enabled: boolean;
     pollMs: number;
     pollMs: number;
+    staleAfterMs: number;
     reviewMaxPages: number;
     reviewMaxPages: number;
   };
   };
   corsOrigins: string[];
   corsOrigins: string[];
@@ -131,6 +138,7 @@ export function loadConfig(environment: NodeJS.ProcessEnv = process.env): AppCon
     worker: {
     worker: {
       enabled: value.SYNC_WORKER_ENABLED === 'true',
       enabled: value.SYNC_WORKER_ENABLED === 'true',
       pollMs: value.SYNC_WORKER_POLL_MS,
       pollMs: value.SYNC_WORKER_POLL_MS,
+      staleAfterMs: value.SYNC_JOB_STALE_AFTER_MS,
       reviewMaxPages: value.JD_REVIEW_MAX_PAGES,
       reviewMaxPages: value.JD_REVIEW_MAX_PAGES,
     },
     },
     corsOrigins: value.CORS_ORIGINS.split(',').map((origin) => origin.trim()).filter(Boolean),
     corsOrigins: value.CORS_ORIGINS.split(',').map((origin) => origin.trim()).filter(Boolean),

+ 14 - 4
src/local-app.ts

@@ -14,6 +14,7 @@ import { createSaasPlatformRouter } from './modules/saas-platform/routes.js';
 export function createLocalDemoApp(input: {
 export function createLocalDemoApp(input: {
   dataset: DomesticDataset;
   dataset: DomesticDataset;
   corsOrigins: string[];
   corsOrigins: string[];
+  workspaceId?: string;
 }) {
 }) {
   const app = express();
   const app = express();
   app.disable('x-powered-by');
   app.disable('x-powered-by');
@@ -25,18 +26,19 @@ export function createLocalDemoApp(input: {
   }));
   }));
   app.use(express.json({ limit: '1mb' }));
   app.use(express.json({ limit: '1mb' }));
 
 
-  const jobs = new LocalSyncJobStore(input.dataset);
+  const workspaceId = input.workspaceId ?? 'demashi';
+  const jobs = new LocalSyncJobStore(input.dataset, workspaceId);
   const localPrincipal = {
   const localPrincipal = {
     userId: 'local-admin',
     userId: 'local-admin',
     email: 'local-admin@localhost',
     email: 'local-admin@localhost',
     displayName: 'Local Admin',
     displayName: 'Local Admin',
     authMode: 'disabled' as const,
     authMode: 'disabled' as const,
   };
   };
-  const platform = new LocalPlatformRepository(input.dataset, jobs, localPrincipal);
+  const platform = new LocalPlatformRepository(input.dataset, jobs, localPrincipal, workspaceId);
   const access = new WorkspaceAccessService(platform);
   const access = new WorkspaceAccessService(platform);
   app.use('/api', createAuthenticationMiddleware(new DisabledAuthenticator(localPrincipal)));
   app.use('/api', createAuthenticationMiddleware(new DisabledAuthenticator(localPrincipal)));
   const sync = new SyncService(jobs);
   const sync = new SyncService(jobs);
-  const snapshot = new LocalSnapshotService(input.dataset);
+  const snapshot = new LocalSnapshotService(input.dataset, workspaceId);
 
 
   app.get('/health', (_request, response) => {
   app.get('/health', (_request, response) => {
     response.json({
     response.json({
@@ -44,6 +46,7 @@ export function createLocalDemoApp(input: {
       status: 'ok',
       status: 'ok',
       mode: 'local-demo',
       mode: 'local-demo',
       database: 'deferred',
       database: 'deferred',
+      workspaceId,
       dataset: {
       dataset: {
         caseName: input.dataset.caseName,
         caseName: input.dataset.caseName,
         platform: input.dataset.platform,
         platform: input.dataset.platform,
@@ -54,7 +57,14 @@ export function createLocalDemoApp(input: {
     });
     });
   });
   });
 
 
-  app.use('/api/domestic-voc', createDomesticVocRouter({ jobs, sync, snapshot, catalog: platform, access }));
+  app.use('/api/domestic-voc', createDomesticVocRouter({
+    jobs,
+    sync,
+    snapshot,
+    catalog: platform,
+    access,
+    defaultWorkspaceId: workspaceId,
+  }));
   app.use('/api/saas', createSaasPlatformRouter({ repository: platform, access }));
   app.use('/api/saas', createSaasPlatformRouter({ repository: platform, access }));
 
 
   app.use((_request, response) => {
   app.use((_request, response) => {

+ 7 - 1
src/local-server.ts

@@ -1,3 +1,4 @@
+import 'dotenv/config';
 import { readFile } from 'node:fs/promises';
 import { readFile } from 'node:fs/promises';
 import { createServer } from 'node:http';
 import { createServer } from 'node:http';
 import { resolve } from 'node:path';
 import { resolve } from 'node:path';
@@ -8,7 +9,11 @@ import type { DomesticDataset } from './types/domestic-dataset.js';
 const localEnvironmentSchema = z.object({
 const localEnvironmentSchema = z.object({
   LOCAL_HOST: z.string().min(1).default('127.0.0.1'),
   LOCAL_HOST: z.string().min(1).default('127.0.0.1'),
   LOCAL_PORT: z.coerce.number().int().min(1).max(65535).default(4400),
   LOCAL_PORT: z.coerce.number().int().min(1).max(65535).default(4400),
-  LOCAL_DATASET_PATH: z.string().min(1).optional(),
+  LOCAL_WORKSPACE_ID: z.string().min(1).default('demashi'),
+  LOCAL_DATASET_PATH: z.preprocess(
+    (value) => value === '' ? undefined : value,
+    z.string().min(1).optional(),
+  ),
   LOCAL_CORS_ORIGINS: z.string().min(1).default('http://localhost:4200,http://127.0.0.1:4200'),
   LOCAL_CORS_ORIGINS: z.string().min(1).default('http://localhost:4200,http://127.0.0.1:4200'),
 });
 });
 
 
@@ -41,6 +46,7 @@ async function main(): Promise<void> {
   const dataset = await loadDataset(datasetPath);
   const dataset = await loadDataset(datasetPath);
   const app = createLocalDemoApp({
   const app = createLocalDemoApp({
     dataset,
     dataset,
+    workspaceId: config.LOCAL_WORKSPACE_ID,
     corsOrigins: config.LOCAL_CORS_ORIGINS.split(',').map((origin) => origin.trim()).filter(Boolean),
     corsOrigins: config.LOCAL_CORS_ORIGINS.split(',').map((origin) => origin.trim()).filter(Boolean),
   });
   });
   const server = createServer(app);
   const server = createServer(app);

+ 37 - 0
src/modules/domestic-voc/jobs/sync-worker.ts

@@ -23,6 +23,36 @@ interface ClaimedSyncJobRow {
   max_attempts: number;
   max_attempts: number;
 }
 }
 
 
+export async function recoverStaleSyncJobs(pool: Pool, staleAfterMs: number): Promise<number> {
+  const result = await pool.query<{ recovered_count: number }>(`
+    WITH recovered AS (
+      UPDATE voc.sync_job
+      SET status = CASE WHEN attempts < max_attempts THEN 'pending' ELSE 'failed' END,
+          worker_id = NULL,
+          error_summary = 'Worker interrupted; stale job recovered',
+          completed_at = CASE WHEN attempts < max_attempts THEN NULL ELSE now() END,
+          updated_at = now()
+      WHERE status = 'processing'
+        AND updated_at < now() - ($1::integer * interval '1 millisecond')
+      RETURNING id, status
+    ), recorded AS (
+      INSERT INTO voc.sync_job_event (sync_job_id, level, event_type, message, details)
+      SELECT id,
+             CASE WHEN status = 'pending' THEN 'warning' ELSE 'error' END,
+             'sync_stale_recovered',
+             CASE WHEN status = 'pending'
+               THEN 'Interrupted sync job returned to the queue'
+               ELSE 'Interrupted sync job exhausted retry attempts'
+             END,
+             jsonb_build_object('status', status)
+      FROM recovered
+      RETURNING sync_job_id
+    )
+    SELECT COUNT(*)::integer AS recovered_count FROM recorded
+  `, [staleAfterMs]);
+  return result.rows[0]?.recovered_count ?? 0;
+}
+
 export async function claimNextSyncJob(pool: Pool, workerId: string): Promise<ClaimedSyncJob | null> {
 export async function claimNextSyncJob(pool: Pool, workerId: string): Promise<ClaimedSyncJob | null> {
   const result = await pool.query<ClaimedSyncJobRow>(`
   const result = await pool.query<ClaimedSyncJobRow>(`
     WITH next_job AS (
     WITH next_job AS (
@@ -69,6 +99,7 @@ export function startSyncWorker(input: {
   pool: Pool;
   pool: Pool;
   processor: JdSyncService;
   processor: JdSyncService;
   pollMs: number;
   pollMs: number;
+  staleAfterMs: number;
   workerId?: string;
   workerId?: string;
 }): { stop: () => Promise<void> } {
 }): { stop: () => Promise<void> } {
   const workerId = input.workerId ?? `saas-voc-${process.pid}`;
   const workerId = input.workerId ?? `saas-voc-${process.pid}`;
@@ -76,6 +107,7 @@ export function startSyncWorker(input: {
   let running = false;
   let running = false;
   let timer: NodeJS.Timeout | undefined;
   let timer: NodeJS.Timeout | undefined;
   let activeIteration: Promise<void> | null = null;
   let activeIteration: Promise<void> | null = null;
+  let initialized = false;
 
 
   const schedule = () => {
   const schedule = () => {
     if (stopped) return;
     if (stopped) return;
@@ -87,6 +119,11 @@ export function startSyncWorker(input: {
     running = true;
     running = true;
     activeIteration = (async () => {
     activeIteration = (async () => {
       try {
       try {
+        if (!initialized) {
+          const recovered = await recoverStaleSyncJobs(input.pool, input.staleAfterMs);
+          if (recovered) console.log(`[sync-worker] recovered ${recovered} stale job(s)`);
+          initialized = true;
+        }
         const job = await claimNextSyncJob(input.pool, workerId);
         const job = await claimNextSyncJob(input.pool, workerId);
         if (job) await input.processor.process(job);
         if (job) await input.processor.process(job);
       } catch {
       } catch {

+ 5 - 2
src/modules/domestic-voc/local/local-snapshot.service.ts

@@ -2,10 +2,13 @@ import type { DomesticDataset } from '../../../types/domestic-dataset.js';
 import type { DomesticSnapshotProvider } from '../routes.js';
 import type { DomesticSnapshotProvider } from '../routes.js';
 
 
 export class LocalSnapshotService implements DomesticSnapshotProvider {
 export class LocalSnapshotService implements DomesticSnapshotProvider {
-  constructor(private readonly dataset: DomesticDataset) {}
+  constructor(
+    private readonly dataset: DomesticDataset,
+    private readonly workspaceId = 'demashi',
+  ) {}
 
 
   async getSnapshot(workspaceId: string, platform: string): Promise<DomesticDataset | null> {
   async getSnapshot(workspaceId: string, platform: string): Promise<DomesticDataset | null> {
-    if (workspaceId !== 'demashi' || platform !== this.dataset.platform) return null;
+    if (workspaceId !== this.workspaceId || platform !== this.dataset.platform) return null;
     return this.dataset;
     return this.dataset;
   }
   }
 }
 }

+ 69 - 1
src/modules/domestic-voc/local/local-sync-job.store.ts

@@ -8,17 +8,26 @@ import type { DomesticDataset } from '../../../types/domestic-dataset.js';
 export class LocalSyncJobStore implements SyncJobStore {
 export class LocalSyncJobStore implements SyncJobStore {
   private readonly jobs = new Map<string, SyncJobRecord>();
   private readonly jobs = new Map<string, SyncJobRecord>();
   private readonly idempotency = new Map<string, string>();
   private readonly idempotency = new Map<string, string>();
+  private readonly events = new Map<string, Array<{
+    id: string;
+    level: string;
+    type: string;
+    message: string;
+    details: Record<string, unknown>;
+    createdAt: string;
+  }>>();
   private readonly productIds: Set<string>;
   private readonly productIds: Set<string>;
 
 
   constructor(
   constructor(
     private readonly dataset: DomesticDataset,
     private readonly dataset: DomesticDataset,
+    private readonly workspaceId = 'demashi',
     private readonly now: () => Date = () => new Date(),
     private readonly now: () => Date = () => new Date(),
   ) {
   ) {
     this.productIds = new Set(dataset.products.map((product) => product.productId));
     this.productIds = new Set(dataset.products.map((product) => product.productId));
   }
   }
 
 
   async enqueue(input: EnqueueSyncJobInput): Promise<SyncJobRecord | null> {
   async enqueue(input: EnqueueSyncJobInput): Promise<SyncJobRecord | null> {
-    if (input.workspaceId !== 'demashi' || input.platform !== this.dataset.platform) return null;
+    if (input.workspaceId !== this.workspaceId || input.platform !== this.dataset.platform) return null;
 
 
     const existingId = this.idempotency.get(input.idempotencyKey);
     const existingId = this.idempotency.get(input.idempotencyKey);
     if (existingId) return this.jobs.get(existingId) ?? null;
     if (existingId) return this.jobs.get(existingId) ?? null;
@@ -44,6 +53,12 @@ export class LocalSyncJobStore implements SyncJobStore {
     };
     };
     this.jobs.set(job.id, job);
     this.jobs.set(job.id, job);
     this.idempotency.set(input.idempotencyKey, job.id);
     this.idempotency.set(input.idempotencyKey, job.id);
+    this.addEvent(job.id, 'sync_finished', job.status === 'completed'
+      ? 'Local dataset sync validation completed'
+      : 'Local dataset sync validation found unavailable products', {
+      status: job.status,
+      missingProductCount: missingCount,
+    });
     return job;
     return job;
   }
   }
 
 
@@ -51,9 +66,62 @@ export class LocalSyncJobStore implements SyncJobStore {
     return this.jobs.get(publicId) ?? null;
     return this.jobs.get(publicId) ?? null;
   }
   }
 
 
+  async retry(workspaceId: string, publicId: string): Promise<SyncJobRecord | null> {
+    const current = this.jobs.get(publicId);
+    if (!current || current.workspaceId !== workspaceId || !['partial', 'failed', 'cancelled'].includes(current.status)) return null;
+    const timestamp = this.now().toISOString();
+    const missingCount = current.productIds.filter((productId) => !this.productIds.has(productId)).length;
+    const updated: SyncJobRecord = {
+      ...current,
+      status: missingCount ? 'partial' : 'completed',
+      progress: 100,
+      attempts: 0,
+      errorSummary: missingCount
+        ? `Local demo dataset does not contain ${missingCount} requested product(s)`
+        : null,
+      requestedAt: timestamp,
+      startedAt: timestamp,
+      completedAt: timestamp,
+    };
+    this.jobs.set(publicId, updated);
+    this.addEvent(publicId, 'sync_manually_retried', 'Local sync validation retried', { status: updated.status });
+    return updated;
+  }
+
+  async cancel(workspaceId: string, publicId: string): Promise<SyncJobRecord | null> {
+    const current = this.jobs.get(publicId);
+    if (!current || current.workspaceId !== workspaceId || current.status !== 'pending') return null;
+    const updated: SyncJobRecord = {
+      ...current,
+      status: 'cancelled',
+      completedAt: this.now().toISOString(),
+    };
+    this.jobs.set(publicId, updated);
+    this.addEvent(publicId, 'sync_manually_cancelled', 'Pending local sync job cancelled', {});
+    return updated;
+  }
+
   listByWorkspace(workspaceId: string): SyncJobRecord[] {
   listByWorkspace(workspaceId: string): SyncJobRecord[] {
     return [...this.jobs.values()]
     return [...this.jobs.values()]
       .filter((job) => job.workspaceId === workspaceId)
       .filter((job) => job.workspaceId === workspaceId)
       .sort((left, right) => right.requestedAt.localeCompare(left.requestedAt));
       .sort((left, right) => right.requestedAt.localeCompare(left.requestedAt));
   }
   }
+
+  listEvents(workspaceId: string, publicId: string) {
+    const job = this.jobs.get(publicId);
+    return job?.workspaceId === workspaceId ? [...(this.events.get(publicId) ?? [])] : [];
+  }
+
+  private addEvent(jobId: string, type: string, message: string, details: Record<string, unknown>): void {
+    const items = this.events.get(jobId) ?? [];
+    items.push({
+      id: `${jobId}:${items.length + 1}`,
+      level: 'info',
+      type,
+      message,
+      details,
+      createdAt: this.now().toISOString(),
+    });
+    this.events.set(jobId, items);
+  }
 }
 }

+ 56 - 0
src/modules/domestic-voc/repositories/sync-job.repository.ts

@@ -28,6 +28,8 @@ export interface EnqueueSyncJobInput {
 export interface SyncJobStore {
 export interface SyncJobStore {
   enqueue(input: EnqueueSyncJobInput): Promise<SyncJobRecord | null>;
   enqueue(input: EnqueueSyncJobInput): Promise<SyncJobRecord | null>;
   findByPublicId(publicId: string): Promise<SyncJobRecord | null>;
   findByPublicId(publicId: string): Promise<SyncJobRecord | null>;
+  retry(workspaceId: string, publicId: string): Promise<SyncJobRecord | null>;
+  cancel(workspaceId: string, publicId: string): Promise<SyncJobRecord | null>;
 }
 }
 
 
 interface SyncJobRow {
 interface SyncJobRow {
@@ -122,6 +124,60 @@ export class SyncJobRepository implements SyncJobStore {
     return row ? this.map(row) : null;
     return row ? this.map(row) : null;
   }
   }
 
 
+  async retry(workspaceId: string, publicId: string): Promise<SyncJobRecord | null> {
+    const result = await this.database.query<SyncJobRow>(`
+      WITH updated AS (
+        UPDATE voc.sync_job job
+        SET status = 'pending', progress = 0, attempts = 0, worker_id = NULL,
+            error_summary = NULL, requested_at = now(), started_at = NULL,
+            completed_at = NULL, updated_at = now()
+        FROM voc.workspace workspace
+        WHERE workspace.id = job.workspace_id AND workspace.public_id = $1
+          AND job.public_id = $2 AND job.status IN ('partial', 'failed', 'cancelled')
+        RETURNING job.*
+      ), recorded AS (
+        INSERT INTO voc.sync_job_event (sync_job_id, level, event_type, message, details)
+        SELECT id, 'info', 'sync_manually_retried', 'Sync job queued for manual retry', '{"manual":true}'::jsonb
+        FROM updated
+        RETURNING sync_job_id
+      )
+      SELECT job.public_id, workspace.public_id AS workspace_public_id, job.platform,
+             job.status, job.scopes, job.product_ids, job.progress, job.attempts,
+             job.max_attempts, job.error_summary, job.requested_at, job.started_at, job.completed_at
+      FROM updated job
+      JOIN voc.workspace workspace ON workspace.id = job.workspace_id
+      WHERE EXISTS (SELECT 1 FROM recorded WHERE recorded.sync_job_id = job.id)
+    `, [workspaceId, publicId]);
+    const row = result.rows[0];
+    return row ? this.map(row) : null;
+  }
+
+  async cancel(workspaceId: string, publicId: string): Promise<SyncJobRecord | null> {
+    const result = await this.database.query<SyncJobRow>(`
+      WITH updated AS (
+        UPDATE voc.sync_job job
+        SET status = 'cancelled', worker_id = NULL, completed_at = now(), updated_at = now()
+        FROM voc.workspace workspace
+        WHERE workspace.id = job.workspace_id AND workspace.public_id = $1
+          AND job.public_id = $2 AND job.status = 'pending'
+        RETURNING job.*
+      ), recorded AS (
+        INSERT INTO voc.sync_job_event (sync_job_id, level, event_type, message, details)
+        SELECT id, 'info', 'sync_manually_cancelled', 'Pending sync job cancelled', '{"manual":true}'::jsonb
+        FROM updated
+        RETURNING sync_job_id
+      )
+      SELECT job.public_id, workspace.public_id AS workspace_public_id, job.platform,
+             job.status, job.scopes, job.product_ids, job.progress, job.attempts,
+             job.max_attempts, job.error_summary, job.requested_at, job.started_at, job.completed_at
+      FROM updated job
+      JOIN voc.workspace workspace ON workspace.id = job.workspace_id
+      WHERE EXISTS (SELECT 1 FROM recorded WHERE recorded.sync_job_id = job.id)
+    `, [workspaceId, publicId]);
+    const row = result.rows[0];
+    return row ? this.map(row) : null;
+  }
+
   private map(row: SyncJobRow): SyncJobRecord {
   private map(row: SyncJobRow): SyncJobRecord {
     return {
     return {
       id: row.public_id,
       id: row.public_id,

+ 83 - 19
src/modules/domestic-voc/routes.ts

@@ -12,7 +12,7 @@ export interface DomesticSnapshotProvider {
 }
 }
 
 
 const syncRequestSchema = z.object({
 const syncRequestSchema = z.object({
-  workspaceId: z.string().min(1).default('demashi'),
+  workspaceId: z.string().min(1).optional(),
   platform: z.literal('jd').default('jd'),
   platform: z.literal('jd').default('jd'),
   productIds: z.array(z.string().min(1)).min(1).max(100),
   productIds: z.array(z.string().min(1)).min(1).max(100),
   scopes: z.array(z.enum(['product', 'reviews'])).min(1).default(['product', 'reviews']),
   scopes: z.array(z.enum(['product', 'reviews'])).min(1).default(['product', 'reviews']),
@@ -20,7 +20,7 @@ const syncRequestSchema = z.object({
 });
 });
 
 
 const snapshotQuerySchema = z.object({
 const snapshotQuerySchema = z.object({
-  workspaceId: z.string().min(1).default('demashi'),
+  workspaceId: z.string().min(1).optional(),
   platform: z.literal('jd').default('jd'),
   platform: z.literal('jd').default('jd'),
 });
 });
 
 
@@ -30,18 +30,21 @@ export function createDomesticVocRouter(dependencies: {
   snapshot: DomesticSnapshotProvider;
   snapshot: DomesticSnapshotProvider;
   catalog: PlatformRepository;
   catalog: PlatformRepository;
   access: WorkspaceAccessService;
   access: WorkspaceAccessService;
+  defaultWorkspaceId: string;
 }): Router {
 }): Router {
   const router = Router();
   const router = Router();
 
 
   router.post('/sync', async (request, response, next) => {
   router.post('/sync', async (request, response, next) => {
     try {
     try {
       const input = syncRequestSchema.parse(request.body);
       const input = syncRequestSchema.parse(request.body);
-      await dependencies.access.require(request, input.workspaceId, 'data:sync');
+      const workspaceId = input.workspaceId ?? dependencies.defaultWorkspaceId;
+      await dependencies.access.require(request, workspaceId, 'data:sync');
       const headerIdempotencyKey = request.header('Idempotency-Key')?.trim();
       const headerIdempotencyKey = request.header('Idempotency-Key')?.trim();
       const { idempotencyKey: bodyIdempotencyKey, ...requiredInput } = input;
       const { idempotencyKey: bodyIdempotencyKey, ...requiredInput } = input;
       const idempotencyKey = headerIdempotencyKey || bodyIdempotencyKey;
       const idempotencyKey = headerIdempotencyKey || bodyIdempotencyKey;
       const syncInput = {
       const syncInput = {
         ...requiredInput,
         ...requiredInput,
+        workspaceId,
         ...(idempotencyKey ? { idempotencyKey } : {}),
         ...(idempotencyKey ? { idempotencyKey } : {}),
       };
       };
       const job = await dependencies.sync.enqueue(syncInput);
       const job = await dependencies.sync.enqueue(syncInput);
@@ -50,7 +53,7 @@ export function createDomesticVocRouter(dependencies: {
         return;
         return;
       }
       }
       await dependencies.catalog.appendAudit({
       await dependencies.catalog.appendAudit({
-        workspaceId: input.workspaceId,
+        workspaceId,
         actorUserId: getPrincipal(request).userId,
         actorUserId: getPrincipal(request).userId,
         action: 'sync.requested',
         action: 'sync.requested',
         entityType: 'sync_job',
         entityType: 'sync_job',
@@ -82,11 +85,64 @@ export function createDomesticVocRouter(dependencies: {
     }
     }
   });
   });
 
 
+  router.post('/jobs/:id/retry', async (request, response, next) => {
+    try {
+      const id = z.uuid().parse(request.params.id);
+      const current = await dependencies.jobs.findByPublicId(id);
+      if (!current) {
+        response.status(404).json({ error: 'job_not_found' });
+        return;
+      }
+      await dependencies.access.require(request, current.workspaceId, 'data:sync');
+      const job = await dependencies.jobs.retry(current.workspaceId, id);
+      if (!job) {
+        response.status(409).json({ error: 'sync_job_not_retryable' });
+        return;
+      }
+      await dependencies.catalog.appendAudit({
+        workspaceId: job.workspaceId,
+        actorUserId: getPrincipal(request).userId,
+        action: 'sync.retried',
+        entityType: 'sync_job',
+        entityId: job.id,
+        metadata: { status: job.status },
+      });
+      response.status(202).json({ job });
+    } catch (error) { next(error); }
+  });
+
+  router.post('/jobs/:id/cancel', async (request, response, next) => {
+    try {
+      const id = z.uuid().parse(request.params.id);
+      const current = await dependencies.jobs.findByPublicId(id);
+      if (!current) {
+        response.status(404).json({ error: 'job_not_found' });
+        return;
+      }
+      await dependencies.access.require(request, current.workspaceId, 'data:sync');
+      const job = await dependencies.jobs.cancel(current.workspaceId, id);
+      if (!job) {
+        response.status(409).json({ error: 'sync_job_not_cancellable' });
+        return;
+      }
+      await dependencies.catalog.appendAudit({
+        workspaceId: job.workspaceId,
+        actorUserId: getPrincipal(request).userId,
+        action: 'sync.cancelled',
+        entityType: 'sync_job',
+        entityId: job.id,
+        metadata: { status: job.status },
+      });
+      response.json({ job });
+    } catch (error) { next(error); }
+  });
+
   router.get('/snapshot', async (request, response, next) => {
   router.get('/snapshot', async (request, response, next) => {
     try {
     try {
       const query = snapshotQuerySchema.parse(request.query);
       const query = snapshotQuerySchema.parse(request.query);
-      await dependencies.access.require(request, query.workspaceId, 'workspace:read');
-      const snapshot = await dependencies.snapshot.getSnapshot(query.workspaceId, query.platform);
+      const workspaceId = query.workspaceId ?? dependencies.defaultWorkspaceId;
+      await dependencies.access.require(request, workspaceId, 'workspace:read');
+      const snapshot = await dependencies.snapshot.getSnapshot(workspaceId, query.platform);
       if (!snapshot) {
       if (!snapshot) {
         response.status(404).json({ error: 'workspace_not_found' });
         response.status(404).json({ error: 'workspace_not_found' });
         return;
         return;
@@ -109,9 +165,10 @@ export function createDomesticVocRouter(dependencies: {
         role: z.enum(['own', 'competitor']).optional(),
         role: z.enum(['own', 'competitor']).optional(),
         category: z.string().max(200).default(''),
         category: z.string().max(200).default(''),
       }).parse(request.query);
       }).parse(request.query);
-      await dependencies.access.require(request, query.workspaceId, 'workspace:read');
+      const workspaceId = query.workspaceId ?? dependencies.defaultWorkspaceId;
+      await dependencies.access.require(request, workspaceId, 'workspace:read');
       response.json(await dependencies.catalog.listProducts({
       response.json(await dependencies.catalog.listProducts({
-        workspaceId: query.workspaceId,
+        workspaceId,
         platform: query.platform,
         platform: query.platform,
         limit: query.limit,
         limit: query.limit,
         cursor: query.cursor ?? null,
         cursor: query.cursor ?? null,
@@ -126,9 +183,10 @@ export function createDomesticVocRouter(dependencies: {
     try {
     try {
       const query = catalogQuerySchema.parse(request.query);
       const query = catalogQuerySchema.parse(request.query);
       const productId = z.string().min(1).max(200).parse(request.params.productId);
       const productId = z.string().min(1).max(200).parse(request.params.productId);
-      await dependencies.access.require(request, query.workspaceId, 'workspace:read');
+      const workspaceId = query.workspaceId ?? dependencies.defaultWorkspaceId;
+      await dependencies.access.require(request, workspaceId, 'workspace:read');
       response.json(await dependencies.catalog.listReviews({
       response.json(await dependencies.catalog.listReviews({
-        workspaceId: query.workspaceId, platform: query.platform, productId,
+        workspaceId, platform: query.platform, productId,
         limit: query.limit, cursor: query.cursor ?? null,
         limit: query.limit, cursor: query.cursor ?? null,
       }));
       }));
     } catch (error) { next(error); }
     } catch (error) { next(error); }
@@ -138,8 +196,9 @@ export function createDomesticVocRouter(dependencies: {
     try {
     try {
       const query = snapshotQuerySchema.parse(request.query);
       const query = snapshotQuerySchema.parse(request.query);
       const productId = z.string().min(1).max(200).parse(request.params.productId);
       const productId = z.string().min(1).max(200).parse(request.params.productId);
-      await dependencies.access.require(request, query.workspaceId, 'workspace:read');
-      const product = await dependencies.catalog.getProduct(query.workspaceId, query.platform, productId);
+      const workspaceId = query.workspaceId ?? dependencies.defaultWorkspaceId;
+      await dependencies.access.require(request, workspaceId, 'workspace:read');
+      const product = await dependencies.catalog.getProduct(workspaceId, query.platform, productId);
       if (!product) {
       if (!product) {
         response.status(404).json({ error: 'product_not_found' });
         response.status(404).json({ error: 'product_not_found' });
         return;
         return;
@@ -151,9 +210,10 @@ export function createDomesticVocRouter(dependencies: {
   router.get('/relations', async (request, response, next) => {
   router.get('/relations', async (request, response, next) => {
     try {
     try {
       const query = catalogQuerySchema.parse(request.query);
       const query = catalogQuerySchema.parse(request.query);
-      await dependencies.access.require(request, query.workspaceId, 'workspace:read');
+      const workspaceId = query.workspaceId ?? dependencies.defaultWorkspaceId;
+      await dependencies.access.require(request, workspaceId, 'workspace:read');
       response.json(await dependencies.catalog.listRelations({
       response.json(await dependencies.catalog.listRelations({
-        workspaceId: query.workspaceId, platform: query.platform,
+        workspaceId, platform: query.platform,
         limit: query.limit, cursor: query.cursor ?? null,
         limit: query.limit, cursor: query.cursor ?? null,
       }));
       }));
     } catch (error) { next(error); }
     } catch (error) { next(error); }
@@ -161,10 +221,13 @@ export function createDomesticVocRouter(dependencies: {
 
 
   router.get('/jobs', async (request, response, next) => {
   router.get('/jobs', async (request, response, next) => {
     try {
     try {
-      const query = catalogQuerySchema.extend({ status: z.string().max(40).default('') }).parse(request.query);
-      await dependencies.access.require(request, query.workspaceId, 'workspace:read');
+      const query = catalogQuerySchema.extend({
+        status: z.enum(['pending', 'processing', 'completed', 'partial', 'failed', 'cancelled']).or(z.literal('')).default(''),
+      }).parse(request.query);
+      const workspaceId = query.workspaceId ?? dependencies.defaultWorkspaceId;
+      await dependencies.access.require(request, workspaceId, 'workspace:read');
       response.json(await dependencies.catalog.listJobs({
       response.json(await dependencies.catalog.listJobs({
-        workspaceId: query.workspaceId, limit: query.limit, cursor: query.cursor ?? null, status: query.status,
+        workspaceId, limit: query.limit, cursor: query.cursor ?? null, status: query.status,
       }));
       }));
     } catch (error) { next(error); }
     } catch (error) { next(error); }
   });
   });
@@ -173,8 +236,9 @@ export function createDomesticVocRouter(dependencies: {
     try {
     try {
       const query = snapshotQuerySchema.pick({ workspaceId: true }).parse(request.query);
       const query = snapshotQuerySchema.pick({ workspaceId: true }).parse(request.query);
       const id = z.uuid().parse(request.params.id);
       const id = z.uuid().parse(request.params.id);
-      await dependencies.access.require(request, query.workspaceId, 'workspace:read');
-      response.json({ items: await dependencies.catalog.listJobEvents(query.workspaceId, id) });
+      const workspaceId = query.workspaceId ?? dependencies.defaultWorkspaceId;
+      await dependencies.access.require(request, workspaceId, 'workspace:read');
+      response.json({ items: await dependencies.catalog.listJobEvents(workspaceId, id) });
     } catch (error) { next(error); }
     } catch (error) { next(error); }
   });
   });
 
 

+ 1 - 0
src/modules/saas-platform/domain.ts

@@ -138,6 +138,7 @@ export interface PlatformRepository {
   listWorkspaces(userId: string): Promise<WorkspaceSummary[]>;
   listWorkspaces(userId: string): Promise<WorkspaceSummary[]>;
   getMembership(workspaceId: string, userId: string): Promise<WorkspaceMember | null>;
   getMembership(workspaceId: string, userId: string): Promise<WorkspaceMember | null>;
   listMembers(workspaceId: string): Promise<WorkspaceMember[]>;
   listMembers(workspaceId: string): Promise<WorkspaceMember[]>;
+  countActiveOwners(workspaceId: string): Promise<number>;
   upsertMember(input: {
   upsertMember(input: {
     workspaceId: string;
     workspaceId: string;
     userId: string;
     userId: string;

+ 29 - 14
src/modules/saas-platform/local-platform.repository.ts

@@ -39,12 +39,13 @@ export class LocalPlatformRepository implements PlatformRepository {
     private readonly dataset: DomesticDataset,
     private readonly dataset: DomesticDataset,
     private readonly jobs: LocalSyncJobStore,
     private readonly jobs: LocalSyncJobStore,
     principal: { userId: string; email: string; displayName: string },
     principal: { userId: string; email: string; displayName: string },
+    private readonly workspaceId = 'demashi',
     private readonly now: () => Date = () => new Date(),
     private readonly now: () => Date = () => new Date(),
   ) {
   ) {
     const timestamp = now().toISOString();
     const timestamp = now().toISOString();
     const member: WorkspaceMember = {
     const member: WorkspaceMember = {
       id: 'local-owner',
       id: 'local-owner',
-      workspaceId: 'demashi',
+      workspaceId: this.workspaceId,
       userId: principal.userId,
       userId: principal.userId,
       email: principal.email,
       email: principal.email,
       displayName: principal.displayName,
       displayName: principal.displayName,
@@ -57,13 +58,19 @@ export class LocalPlatformRepository implements PlatformRepository {
   }
   }
 
 
   async getDatasetSnapshot(workspaceId: string, platform: string): Promise<DomesticDataset | null> {
   async getDatasetSnapshot(workspaceId: string, platform: string): Promise<DomesticDataset | null> {
-    return workspaceId === 'demashi' && platform === this.dataset.platform ? this.dataset : null;
+    return workspaceId === this.workspaceId && platform === this.dataset.platform ? this.dataset : null;
   }
   }
 
 
   async listWorkspaces(userId: string): Promise<WorkspaceSummary[]> {
   async listWorkspaces(userId: string): Promise<WorkspaceSummary[]> {
-    const member = this.members.get(this.memberKey('demashi', userId));
+    const member = this.members.get(this.memberKey(this.workspaceId, userId));
     if (!member || member.status !== 'active') return [];
     if (!member || member.status !== 'active') return [];
-    return [{ id: 'demashi', name: 'Demashi JD VOC', caseName: this.dataset.caseName, status: 'active', role: member.role }];
+    return [{
+      id: this.workspaceId,
+      name: `${this.dataset.caseName} ${this.dataset.platform.toUpperCase()} VOC`,
+      caseName: this.dataset.caseName,
+      status: 'active',
+      role: member.role,
+    }];
   }
   }
 
 
   async getMembership(workspaceId: string, userId: string): Promise<WorkspaceMember | null> {
   async getMembership(workspaceId: string, userId: string): Promise<WorkspaceMember | null> {
@@ -74,6 +81,14 @@ export class LocalPlatformRepository implements PlatformRepository {
     return [...this.members.values()].filter((member) => member.workspaceId === workspaceId);
     return [...this.members.values()].filter((member) => member.workspaceId === workspaceId);
   }
   }
 
 
+  async countActiveOwners(workspaceId: string): Promise<number> {
+    return [...this.members.values()].filter((member) => (
+      member.workspaceId === workspaceId
+      && member.role === 'owner'
+      && member.status === 'active'
+    )).length;
+  }
+
   async upsertMember(input: {
   async upsertMember(input: {
     workspaceId: string;
     workspaceId: string;
     userId: string;
     userId: string;
@@ -82,7 +97,7 @@ export class LocalPlatformRepository implements PlatformRepository {
     role: WorkspaceRole;
     role: WorkspaceRole;
     status: WorkspaceMember['status'];
     status: WorkspaceMember['status'];
   }): Promise<WorkspaceMember | null> {
   }): Promise<WorkspaceMember | null> {
-    if (input.workspaceId !== 'demashi') return null;
+    if (input.workspaceId !== this.workspaceId) return null;
     const key = this.memberKey(input.workspaceId, input.userId);
     const key = this.memberKey(input.workspaceId, input.userId);
     const existing = this.members.get(key);
     const existing = this.members.get(key);
     const timestamp = this.now().toISOString();
     const timestamp = this.now().toISOString();
@@ -110,7 +125,7 @@ export class LocalPlatformRepository implements PlatformRepository {
     role?: DomesticProduct['role'];
     role?: DomesticProduct['role'];
     category: string;
     category: string;
   }): Promise<CursorPage<DomesticProduct>> {
   }): Promise<CursorPage<DomesticProduct>> {
-    if (input.workspaceId !== 'demashi' || input.platform !== this.dataset.platform) return { items: [], nextCursor: null };
+    if (input.workspaceId !== this.workspaceId || input.platform !== this.dataset.platform) return { items: [], nextCursor: null };
     const search = input.search.toLocaleLowerCase();
     const search = input.search.toLocaleLowerCase();
     const products = this.dataset.products
     const products = this.dataset.products
       .filter((product) => !input.role || product.role === input.role)
       .filter((product) => !input.role || product.role === input.role)
@@ -122,7 +137,7 @@ export class LocalPlatformRepository implements PlatformRepository {
   }
   }
 
 
   async getProduct(workspaceId: string, platform: string, productId: string): Promise<DomesticProductDetail | null> {
   async getProduct(workspaceId: string, platform: string, productId: string): Promise<DomesticProductDetail | null> {
-    if (workspaceId !== 'demashi' || platform !== this.dataset.platform) return null;
+    if (workspaceId !== this.workspaceId || platform !== this.dataset.platform) return null;
     const product = this.dataset.products.find((item) => item.productId === productId);
     const product = this.dataset.products.find((item) => item.productId === productId);
     if (!product) return null;
     if (!product) return null;
     const reviews = this.dataset.reviews.filter((review) => review.productId === productId);
     const reviews = this.dataset.reviews.filter((review) => review.productId === productId);
@@ -136,7 +151,7 @@ export class LocalPlatformRepository implements PlatformRepository {
   }
   }
 
 
   async listReviews(input: { workspaceId: string; platform: string; productId: string; limit: number; cursor: string | null }) {
   async listReviews(input: { workspaceId: string; platform: string; productId: string; limit: number; cursor: string | null }) {
-    if (input.workspaceId !== 'demashi' || input.platform !== this.dataset.platform) return { items: [], nextCursor: null };
+    if (input.workspaceId !== this.workspaceId || input.platform !== this.dataset.platform) return { items: [], nextCursor: null };
     const reviews = this.dataset.reviews
     const reviews = this.dataset.reviews
       .filter((review) => review.productId === input.productId)
       .filter((review) => review.productId === input.productId)
       .sort((left, right) => left.reviewId.localeCompare(right.reviewId));
       .sort((left, right) => left.reviewId.localeCompare(right.reviewId));
@@ -144,7 +159,7 @@ export class LocalPlatformRepository implements PlatformRepository {
   }
   }
 
 
   async listRelations(input: { workspaceId: string; platform: string; limit: number; cursor: string | null }) {
   async listRelations(input: { workspaceId: string; platform: string; limit: number; cursor: string | null }) {
-    if (input.workspaceId !== 'demashi' || input.platform !== this.dataset.platform) return { items: [], nextCursor: null };
+    if (input.workspaceId !== this.workspaceId || input.platform !== this.dataset.platform) return { items: [], nextCursor: null };
     const relations = [...this.dataset.relations].sort((left, right) => left.relationKey.localeCompare(right.relationKey));
     const relations = [...this.dataset.relations].sort((left, right) => left.relationKey.localeCompare(right.relationKey));
     return paginate(relations, input.limit, input.cursor, (relation) => relation.relationKey);
     return paginate(relations, input.limit, input.cursor, (relation) => relation.relationKey);
   }
   }
@@ -154,14 +169,14 @@ export class LocalPlatformRepository implements PlatformRepository {
     return paginate(jobs, input.limit, input.cursor, (job) => job.id);
     return paginate(jobs, input.limit, input.cursor, (job) => job.id);
   }
   }
 
 
-  async listJobEvents(): Promise<[]> {
-    return [];
+  async listJobEvents(workspaceId: string, jobId: string) {
+    return this.jobs.listEvents(workspaceId, jobId);
   }
   }
 
 
   async listDataSources(workspaceId: string): Promise<DataSourceSummary[]> {
   async listDataSources(workspaceId: string): Promise<DataSourceSummary[]> {
-    if (workspaceId !== 'demashi') return [];
+    if (workspaceId !== this.workspaceId) return [];
     return [{
     return [{
-      id: 'local-jd',
+      id: `local-${this.dataset.platform}`,
       workspaceId,
       workspaceId,
       platform: this.dataset.platform,
       platform: this.dataset.platform,
       kind: 'fmode_gateway',
       kind: 'fmode_gateway',
@@ -172,7 +187,7 @@ export class LocalPlatformRepository implements PlatformRepository {
   }
   }
 
 
   async listImports(input: { workspaceId: string; limit: number; cursor: string | null }): Promise<CursorPage<ImportBatchSummary>> {
   async listImports(input: { workspaceId: string; limit: number; cursor: string | null }): Promise<CursorPage<ImportBatchSummary>> {
-    if (input.workspaceId !== 'demashi') return { items: [], nextCursor: null };
+    if (input.workspaceId !== this.workspaceId) return { items: [], nextCursor: null };
     const item: ImportBatchSummary = {
     const item: ImportBatchSummary = {
       id: 'local-dataset',
       id: 'local-dataset',
       workspaceId: input.workspaceId,
       workspaceId: input.workspaceId,

+ 13 - 1
src/modules/saas-platform/postgres-platform.repository.ts

@@ -168,7 +168,8 @@ export class PostgresPlatformRepository implements PlatformRepository {
   }
   }
 
 
   async bootstrapAdmin(workspaceId: string, principal: { userId: string; email: string; displayName: string }): Promise<void> {
   async bootstrapAdmin(workspaceId: string, principal: { userId: string; email: string; displayName: string }): Promise<void> {
-    await this.upsertMember({ workspaceId, ...principal, role: 'owner', status: 'active' });
+    const member = await this.upsertMember({ workspaceId, ...principal, role: 'owner', status: 'active' });
+    if (!member) throw new Error(`Bootstrap workspace not found or disabled: ${workspaceId}`);
   }
   }
 
 
   async listWorkspaces(userId: string): Promise<WorkspaceSummary[]> {
   async listWorkspaces(userId: string): Promise<WorkspaceSummary[]> {
@@ -214,6 +215,17 @@ export class PostgresPlatformRepository implements PlatformRepository {
     return result.rows.map((row) => this.mapMember(row));
     return result.rows.map((row) => this.mapMember(row));
   }
   }
 
 
+  async countActiveOwners(workspaceId: string): Promise<number> {
+    const result = await this.database.query<{ owner_count: number }>(`
+      SELECT COUNT(*)::integer AS owner_count
+      FROM voc.workspace_member member
+      JOIN voc.workspace workspace ON workspace.id = member.workspace_id
+      WHERE workspace.public_id = $1 AND workspace.status = 'active'
+        AND member.role = 'owner' AND member.status = 'active'
+    `, [workspaceId]);
+    return result.rows[0]?.owner_count ?? 0;
+  }
+
   async upsertMember(input: {
   async upsertMember(input: {
     workspaceId: string; userId: string; email: string; displayName: string; role: WorkspaceRole; status: WorkspaceMember['status'];
     workspaceId: string; userId: string; email: string; displayName: string; role: WorkspaceRole; status: WorkspaceMember['status'];
   }): Promise<WorkspaceMember | null> {
   }): Promise<WorkspaceMember | null> {

+ 53 - 4
src/modules/saas-platform/routes.ts

@@ -8,7 +8,18 @@ import { getPrincipal, WorkspaceAccessService } from './auth.js';
 const pageQuerySchema = z.object({
 const pageQuerySchema = z.object({
   limit: z.coerce.number().int().min(1).max(100).default(25),
   limit: z.coerce.number().int().min(1).max(100).default(25),
   cursor: z.string().max(500).optional(),
   cursor: z.string().max(500).optional(),
-  status: z.string().max(40).default(''),
+});
+
+const analysisPageQuerySchema = pageQuerySchema.extend({
+  status: z.enum(['pending', 'processing', 'completed', 'partial', 'failed', 'cancelled']).or(z.literal('')).default(''),
+});
+
+const actionPageQuerySchema = pageQuerySchema.extend({
+  status: z.enum(['open', 'planned', 'in_progress', 'blocked', 'completed', 'cancelled']).or(z.literal('')).default(''),
+});
+
+const alertPageQuerySchema = pageQuerySchema.extend({
+  status: z.enum(['open', 'acknowledged', 'resolved', 'dismissed']).or(z.literal('')).default(''),
 });
 });
 
 
 const memberSchema = z.object({
 const memberSchema = z.object({
@@ -23,6 +34,14 @@ const analysisSchema = z.object({
   targetKind: z.enum(['workspace', 'category', 'product']).default('workspace'),
   targetKind: z.enum(['workspace', 'category', 'product']).default('workspace'),
   targetKey: z.string().max(300).default(''),
   targetKey: z.string().max(300).default(''),
   input: z.record(z.string(), z.unknown()).default({}),
   input: z.record(z.string(), z.unknown()).default({}),
+}).superRefine((value, context) => {
+  if (value.targetKind !== 'workspace' && !value.targetKey.trim()) {
+    context.addIssue({
+      code: 'custom',
+      path: ['targetKey'],
+      message: 'targetKey is required for category and product analyses',
+    });
+  }
 });
 });
 
 
 const actionCreateSchema = z.object({
 const actionCreateSchema = z.object({
@@ -76,7 +95,9 @@ export function createSaasPlatformRouter(input: {
           workspaceMembers: true,
           workspaceMembers: true,
           cursorCatalog: true,
           cursorCatalog: true,
           syncJobs: true,
           syncJobs: true,
+          syncJobRecovery: true,
           analysisRuns: true,
           analysisRuns: true,
+          analysisExecution: false,
           actionWorkflow: true,
           actionWorkflow: true,
           alerts: true,
           alerts: true,
           auditLog: true,
           auditLog: true,
@@ -105,6 +126,18 @@ export function createSaasPlatformRouter(input: {
       const userId = z.string().min(1).max(200).parse(request.params.userId);
       const userId = z.string().min(1).max(200).parse(request.params.userId);
       const body = memberSchema.parse(request.body);
       const body = memberSchema.parse(request.body);
       await input.access.require(request, workspaceId, 'member:manage');
       await input.access.require(request, workspaceId, 'member:manage');
+      const actor = await input.repository.getMembership(workspaceId, getPrincipal(request).userId);
+      const target = await input.repository.getMembership(workspaceId, userId);
+      const changesOwner = body.role === 'owner' || target?.role === 'owner';
+      if (changesOwner && actor?.role !== 'owner') {
+        throw new ApiError(403, 'workspace_owner_required');
+      }
+      const removesActiveOwner = target?.role === 'owner'
+        && target.status === 'active'
+        && (body.role !== 'owner' || body.status !== 'active');
+      if (removesActiveOwner && await input.repository.countActiveOwners(workspaceId) <= 1) {
+        throw new ApiError(409, 'workspace_requires_active_owner');
+      }
       const member = await input.repository.upsertMember({ workspaceId, userId, ...body });
       const member = await input.repository.upsertMember({ workspaceId, userId, ...body });
       if (!member) throw new ApiError(404, 'workspace_not_found');
       if (!member) throw new ApiError(404, 'workspace_not_found');
       await audit(input.repository, request, workspaceId, 'member.upserted', 'workspace_member', member.id, {
       await audit(input.repository, request, workspaceId, 'member.upserted', 'workspace_member', member.id, {
@@ -134,7 +167,7 @@ export function createSaasPlatformRouter(input: {
   router.get('/workspaces/:workspaceId/analyses', async (request, response, next) => {
   router.get('/workspaces/:workspaceId/analyses', async (request, response, next) => {
     try {
     try {
       const workspaceId = z.string().min(1).parse(request.params.workspaceId);
       const workspaceId = z.string().min(1).parse(request.params.workspaceId);
-      const page = pageQuerySchema.parse(request.query);
+      const page = analysisPageQuerySchema.parse(request.query);
       await input.access.require(request, workspaceId, 'workspace:read');
       await input.access.require(request, workspaceId, 'workspace:read');
       response.json(await input.repository.listAnalyses({ workspaceId, limit: page.limit, cursor: page.cursor ?? null, status: page.status }));
       response.json(await input.repository.listAnalyses({ workspaceId, limit: page.limit, cursor: page.cursor ?? null, status: page.status }));
     } catch (error) { next(error); }
     } catch (error) { next(error); }
@@ -159,7 +192,7 @@ export function createSaasPlatformRouter(input: {
   router.get('/workspaces/:workspaceId/actions', async (request, response, next) => {
   router.get('/workspaces/:workspaceId/actions', async (request, response, next) => {
     try {
     try {
       const workspaceId = z.string().min(1).parse(request.params.workspaceId);
       const workspaceId = z.string().min(1).parse(request.params.workspaceId);
-      const page = pageQuerySchema.parse(request.query);
+      const page = actionPageQuerySchema.parse(request.query);
       await input.access.require(request, workspaceId, 'workspace:read');
       await input.access.require(request, workspaceId, 'workspace:read');
       response.json(await input.repository.listActions({ workspaceId, limit: page.limit, cursor: page.cursor ?? null, status: page.status }));
       response.json(await input.repository.listActions({ workspaceId, limit: page.limit, cursor: page.cursor ?? null, status: page.status }));
     } catch (error) { next(error); }
     } catch (error) { next(error); }
@@ -170,6 +203,7 @@ export function createSaasPlatformRouter(input: {
       const workspaceId = z.string().min(1).parse(request.params.workspaceId);
       const workspaceId = z.string().min(1).parse(request.params.workspaceId);
       const body = actionCreateSchema.parse(request.body);
       const body = actionCreateSchema.parse(request.body);
       await input.access.require(request, workspaceId, 'action:write');
       await input.access.require(request, workspaceId, 'action:write');
+      await requireActiveAssignee(input.repository, workspaceId, body.assigneeUserId);
       const principal = getPrincipal(request);
       const principal = getPrincipal(request);
       const action = await input.repository.createAction({ id: randomUUID(), workspaceId, ...body, createdBy: principal.userId });
       const action = await input.repository.createAction({ id: randomUUID(), workspaceId, ...body, createdBy: principal.userId });
       await audit(input.repository, request, workspaceId, 'action.created', 'action_item', action.id, { priority: action.priority });
       await audit(input.repository, request, workspaceId, 'action.created', 'action_item', action.id, { priority: action.priority });
@@ -183,6 +217,9 @@ export function createSaasPlatformRouter(input: {
       const id = z.uuid().parse(request.params.id);
       const id = z.uuid().parse(request.params.id);
       const body = actionPatchSchema.parse(request.body);
       const body = actionPatchSchema.parse(request.body);
       await input.access.require(request, workspaceId, 'action:write');
       await input.access.require(request, workspaceId, 'action:write');
+      if (Object.hasOwn(body, 'assigneeUserId')) {
+        await requireActiveAssignee(input.repository, workspaceId, body.assigneeUserId ?? null);
+      }
       const definedPatch = Object.fromEntries(
       const definedPatch = Object.fromEntries(
         Object.entries(body).filter((entry) => entry[1] !== undefined),
         Object.entries(body).filter((entry) => entry[1] !== undefined),
       ) as Partial<Pick<ActionItem, 'title' | 'description' | 'priority' | 'status' | 'assigneeUserId' | 'dueAt'>>;
       ) as Partial<Pick<ActionItem, 'title' | 'description' | 'priority' | 'status' | 'assigneeUserId' | 'dueAt'>>;
@@ -196,7 +233,7 @@ export function createSaasPlatformRouter(input: {
   router.get('/workspaces/:workspaceId/alerts', async (request, response, next) => {
   router.get('/workspaces/:workspaceId/alerts', async (request, response, next) => {
     try {
     try {
       const workspaceId = z.string().min(1).parse(request.params.workspaceId);
       const workspaceId = z.string().min(1).parse(request.params.workspaceId);
-      const page = pageQuerySchema.parse(request.query);
+      const page = alertPageQuerySchema.parse(request.query);
       await input.access.require(request, workspaceId, 'workspace:read');
       await input.access.require(request, workspaceId, 'workspace:read');
       response.json(await input.repository.listAlerts({ workspaceId, limit: page.limit, cursor: page.cursor ?? null, status: page.status }));
       response.json(await input.repository.listAlerts({ workspaceId, limit: page.limit, cursor: page.cursor ?? null, status: page.status }));
     } catch (error) { next(error); }
     } catch (error) { next(error); }
@@ -258,3 +295,15 @@ async function audit(
     metadata,
     metadata,
   });
   });
 }
 }
+
+async function requireActiveAssignee(
+  repository: PlatformRepository,
+  workspaceId: string,
+  assigneeUserId: string | null,
+): Promise<void> {
+  if (!assigneeUserId) return;
+  const assignee = await repository.getMembership(workspaceId, assigneeUserId);
+  if (!assignee || assignee.status !== 'active') {
+    throw new ApiError(400, 'assignee_not_workspace_member');
+  }
+}

+ 6 - 1
src/server.ts

@@ -38,7 +38,12 @@ async function main(): Promise<void> {
   const ingestion = new VocIngestionRepository(pool);
   const ingestion = new VocIngestionRepository(pool);
   const processor = new JdSyncService(gateway, ingestion, config.worker.reviewMaxPages);
   const processor = new JdSyncService(gateway, ingestion, config.worker.reviewMaxPages);
   const worker = config.worker.enabled
   const worker = config.worker.enabled
-    ? startSyncWorker({ pool, processor, pollMs: config.worker.pollMs })
+    ? startSyncWorker({
+      pool,
+      processor,
+      pollMs: config.worker.pollMs,
+      staleAfterMs: config.worker.staleAfterMs,
+    })
     : null;
     : null;
 
 
   server.listen(config.port, config.host, () => {
   server.listen(config.port, config.host, () => {

+ 35 - 2
test/app.test.ts

@@ -20,11 +20,26 @@ const config = loadConfig({
   CORS_ORIGINS: 'http://127.0.0.1:4300',
   CORS_ORIGINS: 'http://127.0.0.1:4300',
 });
 });
 
 
-function createMockPool(): Pool {
+function createMockPool(missingTable = ''): Pool {
   return {
   return {
     async query(text: string) {
     async query(text: string) {
       if (text.includes('to_regclass')) {
       if (text.includes('to_regclass')) {
-        return { rows: [{ schema_ready: 'voc.workspace' }], rowCount: 1 };
+        const readiness: Record<string, string | null> = {
+          workspace: 'voc.workspace',
+          product: 'voc.product',
+          review: 'voc.review',
+          sync_job: 'voc.sync_job',
+          workspace_member: 'voc.workspace_member',
+          analysis_run: 'voc.analysis_run',
+          action_item: 'voc.action_item',
+          alert: 'voc.alert',
+          audit_log: 'voc.audit_log',
+        };
+        if (missingTable) readiness[missingTable] = null;
+        return {
+          rows: [readiness],
+          rowCount: 1,
+        };
       }
       }
       if (text.includes('WITH target_workspace')) {
       if (text.includes('WITH target_workspace')) {
         return {
         return {
@@ -99,3 +114,21 @@ test('HTTP surface exposes health, empty snapshot, and sync queue contract', asy
     await new Promise<void>((resolve, reject) => server.close((error) => error ? reject(error) : resolve()));
     await new Promise<void>((resolve, reject) => server.close((error) => error ? reject(error) : resolve()));
   }
   }
 });
 });
+
+test('health reports a required platform migration instead of false readiness', async () => {
+  const app = createApp({ config, pool: createMockPool('action_item') });
+  const server = await new Promise<ReturnType<typeof app.listen>>((resolve) => {
+    const listening = app.listen(0, '127.0.0.1', () => resolve(listening));
+  });
+
+  try {
+    const address = server.address() as AddressInfo;
+    const response = await fetch(`http://127.0.0.1:${address.port}/health`);
+    assert.equal(response.status, 503);
+    const body = await response.json() as { database: string; missingTables: string[] };
+    assert.equal(body.database, 'migration_required');
+    assert.deepEqual(body.missingTables, ['voc.action_item']);
+  } finally {
+    await new Promise<void>((resolve, reject) => server.close((error) => error ? reject(error) : resolve()));
+  }
+});

+ 11 - 0
test/env.test.ts

@@ -48,12 +48,23 @@ 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.fmode.baseUrl, 'http://127.0.0.1:3000/api/voc-e-commerce');
   assert.equal(config.database.migrationUrl, validEnvironment.MIGRATION_DATABASE_URL);
   assert.equal(config.database.migrationUrl, validEnvironment.MIGRATION_DATABASE_URL);
   assert.equal(config.worker.enabled, true);
   assert.equal(config.worker.enabled, true);
+  assert.equal(config.worker.staleAfterMs, 900_000);
   assert.equal(config.worker.reviewMaxPages, 1);
   assert.equal(config.worker.reviewMaxPages, 1);
   assert.equal(config.auth.mode, 'disabled');
   assert.equal(config.auth.mode, 'disabled');
   assert.equal(config.auth.defaultWorkspaceId, 'demashi');
   assert.equal(config.auth.defaultWorkspaceId, 'demashi');
   assert.deepEqual(config.corsOrigins, ['http://127.0.0.1:4300', 'http://localhost:4200']);
   assert.deepEqual(config.corsOrigins, ['http://127.0.0.1:4300', 'http://localhost:4200']);
 });
 });
 
 
+test('loadConfig treats blank optional environment values as unset', () => {
+  const config = loadConfig({
+    ...validEnvironment,
+    MIGRATION_DATABASE_URL: '',
+    SAAS_BOOTSTRAP_ADMIN_USER_ID: '',
+  });
+  assert.equal(config.database.migrationUrl, validEnvironment.DATABASE_URL);
+  assert.equal(config.auth.bootstrapAdminUserId, '');
+});
+
 test('loadConfig rejects disabled authentication in production', () => {
 test('loadConfig rejects disabled authentication in production', () => {
   assert.throws(
   assert.throws(
     () => loadConfig({ ...validEnvironment, NODE_ENV: 'production', API_AUTH_MODE: 'disabled' }),
     () => loadConfig({ ...validEnvironment, NODE_ENV: 'production', API_AUTH_MODE: 'disabled' }),

+ 54 - 0
test/local-app.test.ts

@@ -113,6 +113,60 @@ test('local demo serves a snapshot and queryable completed sync jobs without a d
     const jobResponse = await fetch(`${baseUrl}/api/domestic-voc/jobs/${syncBody.job.id}`);
     const jobResponse = await fetch(`${baseUrl}/api/domestic-voc/jobs/${syncBody.job.id}`);
     assert.equal(jobResponse.status, 200);
     assert.equal(jobResponse.status, 200);
     assert.equal((await jobResponse.json() as { job: { status: string } }).job.status, 'completed');
     assert.equal((await jobResponse.json() as { job: { status: string } }).job.status, 'completed');
+
+    const partialResponse = await fetch(`${baseUrl}/api/domestic-voc/sync`, {
+      method: 'POST',
+      headers: { 'Content-Type': 'application/json', 'Idempotency-Key': 'local-demo-sync-missing-product' },
+      body: JSON.stringify({ productIds: ['missing-product'] }),
+    });
+    assert.equal(partialResponse.status, 202);
+    const partial = await partialResponse.json() as { job: { id: string; status: string } };
+    assert.equal(partial.job.status, 'partial');
+
+    const retryResponse = await fetch(`${baseUrl}/api/domestic-voc/jobs/${partial.job.id}/retry`, { method: 'POST' });
+    assert.equal(retryResponse.status, 202);
+    assert.equal((await retryResponse.json() as { job: { status: string } }).job.status, 'partial');
+
+    const eventsResponse = await fetch(`${baseUrl}/api/domestic-voc/jobs/${partial.job.id}/events`);
+    assert.equal(eventsResponse.status, 200);
+    const eventTypes = (await eventsResponse.json() as { items: Array<{ type: string }> }).items.map((item) => item.type);
+    assert.deepEqual(eventTypes, ['sync_finished', 'sync_manually_retried']);
+
+    const invalidCancel = await fetch(`${baseUrl}/api/domestic-voc/jobs/${syncBody.job.id}/cancel`, { method: 'POST' });
+    assert.equal(invalidCancel.status, 409);
+  } finally {
+    await new Promise<void>((resolve, reject) => server.close((error) => error ? reject(error) : resolve()));
+  }
+});
+
+test('local demo workspace id is configurable and becomes the request default', async () => {
+  const app = createLocalDemoApp({
+    dataset,
+    workspaceId: 'case-two',
+    corsOrigins: ['http://localhost:4200'],
+  });
+  const server = await new Promise<ReturnType<typeof app.listen>>((resolve) => {
+    const listening = app.listen(0, '127.0.0.1', () => resolve(listening));
+  });
+
+  try {
+    const address = server.address() as AddressInfo;
+    const baseUrl = `http://127.0.0.1:${address.port}`;
+    const context = await fetch(`${baseUrl}/api/saas/context`);
+    assert.equal(context.status, 200);
+    assert.equal((await context.json() as { workspaces: Array<{ id: string }> }).workspaces[0]?.id, 'case-two');
+
+    const snapshot = await fetch(`${baseUrl}/api/domestic-voc/snapshot?platform=jd`);
+    assert.equal(snapshot.status, 200);
+    assert.equal((await snapshot.json() as DomesticDataset).caseName, 'Demashi');
+
+    const sync = await fetch(`${baseUrl}/api/domestic-voc/sync`, {
+      method: 'POST',
+      headers: { 'Content-Type': 'application/json', 'Idempotency-Key': 'case-two-default-workspace' },
+      body: JSON.stringify({ productIds: ['11266507445'] }),
+    });
+    assert.equal(sync.status, 202);
+    assert.equal((await sync.json() as { job: { workspaceId: string } }).job.workspaceId, 'case-two');
   } finally {
   } finally {
     await new Promise<void>((resolve, reject) => server.close((error) => error ? reject(error) : resolve()));
     await new Promise<void>((resolve, reject) => server.close((error) => error ? reject(error) : resolve()));
   }
   }

+ 56 - 0
test/saas-platform.test.ts

@@ -149,6 +149,16 @@ test('local SaaS APIs cover context, cursor catalogs, workflows, and audit histo
     assert.equal(invalidCursor.status, 400);
     assert.equal(invalidCursor.status, 400);
     assert.equal((await json(invalidCursor)).error, 'invalid_cursor');
     assert.equal((await json(invalidCursor)).error, 'invalid_cursor');
 
 
+    const invalidActionStatus = await fetch(`${server.baseUrl}/api/saas/workspaces/demashi/actions?status=unknown`);
+    assert.equal(invalidActionStatus.status, 400);
+
+    const missingAnalysisTarget = await fetch(`${server.baseUrl}/api/saas/workspaces/demashi/analyses`, {
+      method: 'POST',
+      headers: { 'Content-Type': 'application/json' },
+      body: JSON.stringify({ analysisType: 'voice', targetKind: 'product' }),
+    });
+    assert.equal(missingAnalysisTarget.status, 400);
+
     const firstRelations = await json(await fetch(`${server.baseUrl}/api/domestic-voc/relations?limit=1`));
     const firstRelations = await json(await fetch(`${server.baseUrl}/api/domestic-voc/relations?limit=1`));
     assert.equal(firstRelations.items.length, 1);
     assert.equal(firstRelations.items.length, 1);
     assert.equal(typeof firstRelations.nextCursor, 'string');
     assert.equal(typeof firstRelations.nextCursor, 'string');
@@ -272,6 +282,7 @@ test('viewer membership can read but cannot call write or member-management rout
   app.use(errorHandler);
   app.use(errorHandler);
   const server = await listen(app);
   const server = await listen(app);
   const viewerHeaders = { 'Content-Type': 'application/json', 'X-Test-User': 'viewer-user' };
   const viewerHeaders = { 'Content-Type': 'application/json', 'X-Test-User': 'viewer-user' };
+  const ownerHeaders = { 'Content-Type': 'application/json', 'X-Test-User': 'local-admin' };
 
 
   try {
   try {
     const readable = await fetch(`${server.baseUrl}/api/saas/workspaces/demashi/actions`, { headers: viewerHeaders });
     const readable = await fetch(`${server.baseUrl}/api/saas/workspaces/demashi/actions`, { headers: viewerHeaders });
@@ -296,6 +307,51 @@ test('viewer membership can read but cannot call write or member-management rout
       assert.equal(response.status, 403);
       assert.equal(response.status, 403);
       assert.equal((await json(response)).error, 'workspace_permission_denied');
       assert.equal((await json(response)).error, 'workspace_permission_denied');
     }
     }
+
+    const lastOwnerRemoval = await fetch(`${server.baseUrl}/api/saas/workspaces/demashi/members/local-admin`, {
+      method: 'PUT',
+      headers: ownerHeaders,
+      body: JSON.stringify({ role: 'analyst', status: 'active' }),
+    });
+    assert.equal(lastOwnerRemoval.status, 409);
+    assert.equal((await json(lastOwnerRemoval)).error, 'workspace_requires_active_owner');
+
+    const invalidAssignee = await fetch(`${server.baseUrl}/api/saas/workspaces/demashi/actions`, {
+      method: 'POST',
+      headers: ownerHeaders,
+      body: JSON.stringify({ title: 'Invalid assignment', assigneeUserId: 'outside-user' }),
+    });
+    assert.equal(invalidAssignee.status, 400);
+    assert.equal((await json(invalidAssignee)).error, 'assignee_not_workspace_member');
+
+    const adminMember = await fetch(`${server.baseUrl}/api/saas/workspaces/demashi/members/admin-user`, {
+      method: 'PUT',
+      headers: ownerHeaders,
+      body: JSON.stringify({ role: 'admin', status: 'active' }),
+    });
+    assert.equal(adminMember.status, 200);
+
+    const adminOwnerChange = await fetch(`${server.baseUrl}/api/saas/workspaces/demashi/members/viewer-user`, {
+      method: 'PUT',
+      headers: { 'Content-Type': 'application/json', 'X-Test-User': 'admin-user' },
+      body: JSON.stringify({ role: 'owner', status: 'active' }),
+    });
+    assert.equal(adminOwnerChange.status, 403);
+    assert.equal((await json(adminOwnerChange)).error, 'workspace_owner_required');
+
+    const secondOwner = await fetch(`${server.baseUrl}/api/saas/workspaces/demashi/members/second-owner`, {
+      method: 'PUT',
+      headers: ownerHeaders,
+      body: JSON.stringify({ role: 'owner', status: 'active' }),
+    });
+    assert.equal(secondOwner.status, 200);
+
+    const allowedOwnerRemoval = await fetch(`${server.baseUrl}/api/saas/workspaces/demashi/members/local-admin`, {
+      method: 'PUT',
+      headers: ownerHeaders,
+      body: JSON.stringify({ role: 'analyst', status: 'active' }),
+    });
+    assert.equal(allowedOwnerRemoval.status, 200);
   } finally {
   } finally {
     await server.close();
     await server.close();
   }
   }

+ 24 - 0
test/sync-worker.test.ts

@@ -0,0 +1,24 @@
+import assert from 'node:assert/strict';
+import test from 'node:test';
+import type { Pool } from 'pg';
+import { recoverStaleSyncJobs } from '../src/modules/domestic-voc/jobs/sync-worker.js';
+
+test('stale processing jobs are recovered with a bounded age threshold', async () => {
+  let sql = '';
+  let values: readonly unknown[] | undefined;
+  const pool = {
+    async query(text: string, input?: readonly unknown[]) {
+      sql = text;
+      values = input;
+      return { rows: [{ recovered_count: 2 }], rowCount: 1 };
+    },
+  } as unknown as Pool;
+
+  const recovered = await recoverStaleSyncJobs(pool, 900_000);
+
+  assert.equal(recovered, 2);
+  assert.deepEqual(values, [900_000]);
+  assert.match(sql, /status = 'processing'/);
+  assert.match(sql, /attempts < max_attempts/);
+  assert.match(sql, /sync_stale_recovered/);
+});