fix(edit): 类目问题修改

apps/worker/src/main.ts

@@ -18,6 +18,7 @@ import {
     syncOzonCatalog,
     type CatalogSyncScope,
 } from './ozon-catalog-sync.js';
+import { scheduleStoreCategoryTreeSync } from './store-category-tree-scheduler.js';
 
 const connection = new Redis(workerConfig.redisUrl, {
     maxRetriesPerRequest: null,
@@ -161,8 +162,8 @@ async function completeVerification(
     rejectedStatus?: 401 | 403,
 ) {
     await database.workspaceTransaction(job.workspaceId, async (tx) => {
-        const current = await tx.query<{ id: string }>(
-            `SELECT s.id FROM atlas.stores s JOIN atlas.store_secrets secret ON secret.id = $2
+        const current = await tx.query<{ id: string; platform_id: string }>(
+            `SELECT s.id, s.platform_id FROM atlas.stores s JOIN atlas.store_secrets secret ON secret.id = $2
              WHERE s.id = $1 AND s.deleted_at IS NULL AND secret.store_id = s.id
                AND secret.revoked_at IS NULL FOR UPDATE`,
             [job.payload.store_id, job.payload.secret_id ?? ''],
@@ -216,38 +217,11 @@ async function completeVerification(
                   ],
         );
         if (!valid) return;
-        const catalogOperation = await tx.query<{
-            id: string;
-        }>(
-            `INSERT INTO atlas.operations
-                (workspace_id, kind, resource_type, resource_id)
-             VALUES (current_setting('app.workspace_id')::uuid,
-                     'store_category_tree_sync', 'store', $1)
-             RETURNING id`,
-            [job.payload.store_id],
-        );
-        const operationId = catalogOperation.rows[0]?.id;
-        if (!operationId) throw new Error('Unable to create store catalog sync operation');
-        await tx.query(
-            `INSERT INTO atlas.outbox_events
-                (workspace_id, aggregate_type, aggregate_id, event_type, event_payload, deduplication_key)
-             VALUES (current_setting('app.workspace_id')::uuid, 'store', $1,
-                     'store_category_tree_sync.requested', $2::jsonb, $3)`,
-            [
-                job.payload.store_id,
-                JSON.stringify({
-                    store_id: job.payload.store_id,
-                    secret_id: job.payload.secret_id,
-                    operation_id: operationId,
-                    platform_id: (await tx.query<{ platform_id: string }>(
-                        `SELECT platform_id FROM atlas.stores WHERE id = $1`,
-                        [job.payload.store_id],
-                    )).rows[0]?.platform_id,
-                    scope: 'tree',
-                }),
-                operationId,
-            ],
-        );
+        await scheduleStoreCategoryTreeSync(tx, {
+            storeId: job.payload.store_id,
+            platformId: current.rows[0].platform_id,
+            secretId: job.payload.secret_id ?? '',
+        });
     });
 }
 

apps/worker/src/store-category-tree-scheduler.ts

@@ -0,0 +1,98 @@
+import type { DatabaseTransaction } from '@atlas/database';
+
+type ScheduleInput = {
+    storeId: string;
+    platformId: string;
+    secretId: string;
+};
+
+export type StoreCategoryTreeScheduleResult =
+    | { scheduled: true; operationId: string }
+    | { scheduled: false; reason: 'cache_available' | 'already_queued' };
+
+/**
+ * Schedule exactly one usable tree refresh for a store.
+ *
+ * The caller must hold the store row with `FOR UPDATE`; credential verification
+ * already does this, which turns the lookup + insert below into a per-store
+ * critical section without introducing a process-local lock.
+ */
+export async function scheduleStoreCategoryTreeSync(
+    transaction: DatabaseTransaction,
+    input: ScheduleInput,
+): Promise<StoreCategoryTreeScheduleResult> {
+    const cache = await transaction.query<{ present: boolean }>(
+        `SELECT EXISTS(
+            SELECT 1 FROM atlas.store_product_categories
+             WHERE store_id = $1 AND synced_at IS NOT NULL
+        ) AS present`,
+        [input.storeId],
+    );
+    if (cache.rows[0]?.present)
+        return { scheduled: false, reason: 'cache_available' };
+
+    const queued = await transaction.query<{
+        id: string;
+        secret_id: string | null;
+    }>(
+        `SELECT operation.id, event.event_payload ->> 'secret_id' AS secret_id
+         FROM atlas.operations operation
+         LEFT JOIN atlas.outbox_events event
+           ON event.event_type = 'store_category_tree_sync.requested'
+          AND event.event_payload ->> 'operation_id' = operation.id::text
+         WHERE operation.kind = 'store_category_tree_sync'
+           AND operation.resource_type = 'store'
+           AND operation.resource_id = $1
+           AND operation.status IN ('queued', 'running')
+         ORDER BY operation.created_at DESC
+         FOR UPDATE OF operation`,
+        [input.storeId],
+    );
+    if (queued.rows.some((operation) => operation.secret_id === input.secretId))
+        return { scheduled: false, reason: 'already_queued' };
+
+    // A task tied to a replaced secret can no longer supply this store's cache.
+    // Mark it cancelled before adding the current credential's task; a worker
+    // that has already picked it up independently re-checks the secret and
+    // leaves this terminal state intact.
+    if (queued.rows.length) {
+        await transaction.query(
+            `UPDATE atlas.operations
+             SET status = 'cancelled', completed_at = now(),
+                 error_code = 'CATALOG_SYNC_SOURCE_REPLACED',
+                 error_message = '门店凭据已更新，旧类目同步已取消'
+             WHERE id = ANY($1::uuid[]) AND status IN ('queued', 'running')`,
+            [queued.rows.map((operation) => operation.id)],
+        );
+    }
+
+    const created = await transaction.query<{ id: string }>(
+        `INSERT INTO atlas.operations
+            (workspace_id, kind, resource_type, resource_id)
+         VALUES (current_setting('app.workspace_id')::uuid,
+                 'store_category_tree_sync', 'store', $1)
+         RETURNING id`,
+        [input.storeId],
+    );
+    const operationId = created.rows[0]?.id;
+    if (!operationId)
+        throw new Error('Unable to create store catalog sync operation');
+    await transaction.query(
+        `INSERT INTO atlas.outbox_events
+            (workspace_id, aggregate_type, aggregate_id, event_type, event_payload, deduplication_key)
+         VALUES (current_setting('app.workspace_id')::uuid, 'store', $1,
+                 'store_category_tree_sync.requested', $2::jsonb, $3)`,
+        [
+            input.storeId,
+            JSON.stringify({
+                store_id: input.storeId,
+                secret_id: input.secretId,
+                operation_id: operationId,
+                platform_id: input.platformId,
+                scope: 'tree',
+            }),
+            operationId,
+        ],
+    );
+    return { scheduled: true, operationId };
+}