[codex] Fix workflow queue dedupe

apps/medusa-be/src/api/store/reviews/helpers.ts

@@ -1,4 +1,5 @@
 import type { MedusaRequest } from "@medusajs/framework/http"
+import type { Query } from "@medusajs/framework/types"
 import {
   ContainerRegistrationKeys,
   MedusaError,
@@ -203,3 +204,29 @@ export const ensureProductExists = async (
     )
   }
 }
+
+export async function ensureCustomerPurchasedProduct(
+  req: MedusaRequest,
+  customerId: string,
+  productId: string
+) {
+  const query = req.scope.resolve<Query>(ContainerRegistrationKeys.QUERY)
+  const { data } = await query.graph({
+    entity: "order",
+    fields: ["id"],
+    filters: {
+      customer_id: customerId,
+      items: {
+        product_id: productId,
+      },
+      payment_status: ["captured", "completed"],
+    },
+  })
+
+  if (!Array.isArray(data) || data.length === 0) {
+    throw new MedusaError(
+      MedusaError.Types.NOT_ALLOWED,
+      "You can only review products you have purchased."
+    )
+  }
+}

apps/medusa-be/src/api/store/reviews/route.ts

@@ -1,6 +1,7 @@
 import type { MedusaRequest, MedusaResponse } from "@medusajs/framework/http"
 import { createReviewWorkflow } from "../../../workflows/product-review/workflows/create-review"
 import {
+  ensureCustomerPurchasedProduct,
   ensureProductExists,
   ensureReviewDoesNotExist,
   getCustomerId,
@@ -24,6 +25,11 @@ export async function POST(
     : getCustomerId(req)
 
   await ensureProductExists(req, product_id)
+
+  if (!tokenRecord) {
+    await ensureCustomerPurchasedProduct(req, customerId, product_id)
+  }
+
   await ensureReviewDoesNotExist({
     customerId,
     productId: product_id,

apps/medusa-be/src/modules/workflow-queue/migrations/.snapshot-workflow-queue.json

@@ -1,7 +1,5 @@
 {
-  "namespaces": [
-    "public"
-  ],
+  "namespaces": ["public"],
   "name": "public",
   "tables": [
     {
@@ -54,37 +52,37 @@
           "enumItems": [],
           "mappedType": "text"
         },
-        "arguments": {
-          "name": "arguments",
-          "type": "jsonb",
+        "dedupe_key": {
+          "name": "dedupe_key",
+          "type": "text",
           "unsigned": false,
           "autoincrement": false,
           "primary": false,
-          "nullable": false,
+          "nullable": true,
           "unique": false,
           "length": null,
           "precision": null,
           "scale": null,
           "default": null,
           "comment": null,
           "enumItems": [],
-          "mappedType": "json"
+          "mappedType": "text"
         },
-        "order_id": {
-          "name": "order_id",
-          "type": "text",
+        "arguments": {
+          "name": "arguments",
+          "type": "jsonb",
           "unsigned": false,
           "autoincrement": false,
           "primary": false,
-          "nullable": true,
+          "nullable": false,
           "unique": false,
           "length": null,
           "precision": null,
           "scale": null,
           "default": null,
           "comment": null,
           "enumItems": [],
-          "mappedType": "text"
+          "mappedType": "json"
         },
         "created_at": {
           "name": "created_at",
@@ -166,19 +164,17 @@
           "expression": "CREATE INDEX IF NOT EXISTS \"IDX_workflow_queue_item_workflow\" ON \"workflow_queue_item\" (\"workflow\") WHERE deleted_at IS NULL"
         },
         {
-          "keyName": "IDX_workflow_queue_item_workflow_order_id",
+          "keyName": "IDX_workflow_queue_item_workflow_dedupe_key_unique",
           "columnNames": [],
           "composite": false,
           "constraint": false,
           "primary": false,
           "unique": false,
-          "expression": "CREATE INDEX IF NOT EXISTS \"IDX_workflow_queue_item_workflow_order_id\" ON \"workflow_queue_item\" (\"workflow\", \"order_id\") WHERE deleted_at IS NULL"
+          "expression": "CREATE UNIQUE INDEX IF NOT EXISTS \"IDX_workflow_queue_item_workflow_dedupe_key_unique\" ON \"workflow_queue_item\" (\"workflow\", \"dedupe_key\") WHERE deleted_at IS NULL AND dedupe_key IS NOT NULL"
         },
         {
           "keyName": "workflow_queue_item_pkey",
-          "columnNames": [
-            "id"
-          ],
+          "columnNames": ["id"],
           "composite": false,
           "constraint": true,
           "primary": true,

apps/medusa-be/src/modules/workflow-queue/migrations/Migration20260610110000.ts

@@ -0,0 +1,24 @@
+import { Migration } from "@medusajs/framework/mikro-orm/migrations"
+
+export class Migration20260610110000 extends Migration {
+  override async up(): Promise<void> {
+    this.addSql(
+      `DROP INDEX IF EXISTS "IDX_workflow_queue_item_workflow_order_id";`
+    )
+    this.addSql(
+      `alter table if exists "workflow_queue_item" drop column if exists "order_id";`
+    )
+  }
+
+  override async down(): Promise<void> {
+    this.addSql(
+      `alter table if exists "workflow_queue_item" add column if not exists "order_id" text null;`
+    )
+    this.addSql(
+      `update "workflow_queue_item" set "order_id" = "arguments"->>'order_id' where "order_id" is null and "arguments" ? 'order_id';`
+    )
+    this.addSql(
+      `CREATE INDEX IF NOT EXISTS "IDX_workflow_queue_item_workflow_order_id" ON "workflow_queue_item" ("workflow", "order_id") WHERE deleted_at IS NULL;`
+    )
+  }
+}

apps/medusa-be/src/modules/workflow-queue/migrations/Migration20260610120000.ts

@@ -0,0 +1,27 @@
+import { Migration } from "@medusajs/framework/mikro-orm/migrations"
+
+export class Migration20260610120000 extends Migration {
+  override async up(): Promise<void> {
+    this.addSql(
+      `alter table if exists "workflow_queue_item" add column if not exists "dedupe_key" text null;`
+    )
+    this.addSql(
+      `update "workflow_queue_item" set "dedupe_key" = 'send-review-reminder-' || ("arguments"->>'order_id') where "workflow" = 'send-product-review-request' and "dedupe_key" is null and "arguments" ? 'order_id';`
+    )
+    this.addSql(
+      `delete from "workflow_queue_item" a using (select ctid, row_number() over (partition by "workflow", "dedupe_key" order by "created_at" asc, "run_at" asc, "id" asc) as rn from "workflow_queue_item" where "deleted_at" is null and "dedupe_key" is not null) ranked where a.ctid = ranked.ctid and ranked.rn > 1;`
+    )
+    this.addSql(
+      `CREATE UNIQUE INDEX IF NOT EXISTS "IDX_workflow_queue_item_workflow_dedupe_key_unique" ON "workflow_queue_item" ("workflow", "dedupe_key") WHERE deleted_at IS NULL AND dedupe_key IS NOT NULL;`
+    )
+  }
+
+  override async down(): Promise<void> {
+    this.addSql(
+      `DROP INDEX IF EXISTS "IDX_workflow_queue_item_workflow_dedupe_key_unique";`
+    )
+    this.addSql(
+      `alter table if exists "workflow_queue_item" drop column if exists "dedupe_key";`
+    )
+  }
+}

apps/medusa-be/src/modules/workflow-queue/migrations/Migration20260617104357.ts

@@ -0,0 +1,20 @@
+import { Migration } from "@medusajs/framework/mikro-orm/migrations";
+
+export class Migration20260617104357 extends Migration {
+
+  override async up(): Promise<void> {
+    this.addSql(`alter table if exists "workflow_queue_item" drop constraint if exists "workflow_queue_item_workflow_dedupe_key_unique";`);
+    this.addSql(`drop index if exists "IDX_workflow_queue_item_workflow_order_id";`);
+
+    this.addSql(`alter table if exists "workflow_queue_item" rename column "order_id" to "dedupe_key";`);
+    this.addSql(`CREATE UNIQUE INDEX IF NOT EXISTS "IDX_workflow_queue_item_workflow_dedupe_key_unique" ON "workflow_queue_item" ("workflow", "dedupe_key") WHERE deleted_at IS NULL AND dedupe_key IS NOT NULL;`);
+  }
+
+  override async down(): Promise<void> {
+    this.addSql(`drop index if exists "IDX_workflow_queue_item_workflow_dedupe_key_unique";`);
+
+    this.addSql(`alter table if exists "workflow_queue_item" rename column "dedupe_key" to "order_id";`);
+    this.addSql(`CREATE INDEX IF NOT EXISTS "IDX_workflow_queue_item_workflow_order_id" ON "workflow_queue_item" ("workflow", "order_id") WHERE deleted_at IS NULL;`);
+  }
+
+}

apps/medusa-be/src/modules/workflow-queue/models/workflow-queue-item.ts

@@ -5,8 +5,8 @@ const WorkflowQueueItem = model
     id: model.id().primaryKey(),
     run_at: model.dateTime(),
     workflow: model.text(),
+    dedupe_key: model.text().nullable(),
     arguments: model.json(),
-    order_id: model.text().nullable(),
   })
   .indexes([
     {
@@ -20,9 +20,10 @@ const WorkflowQueueItem = model
       where: "deleted_at IS NULL",
     },
     {
-      name: "IDX_workflow_queue_item_workflow_order_id",
-      on: ["workflow", "order_id"],
-      where: "deleted_at IS NULL",
+      name: "IDX_workflow_queue_item_workflow_dedupe_key_unique",
+      on: ["workflow", "dedupe_key"],
+      unique: true,
+      where: "deleted_at IS NULL AND dedupe_key IS NOT NULL",
     },
   ])
 

apps/medusa-be/src/utils/product-review-request-queue.ts

@@ -15,6 +15,7 @@ import {
 import { workflowQueueNames } from "./workflow-queue-registry"
 
 const PRODUCT_REVIEW_REQUEST_TEMPLATE = "product-review-request"
+const PRODUCT_REVIEW_REQUEST_DEDUPE_KEY_PREFIX = "send-review-reminder"
 
 const ORDER_FIELDS = [
   "id",
@@ -43,18 +44,26 @@ type EmailLogService = EmailLogModuleService & {
 
 type WorkflowQueueItemDTO = {
   arguments: Record<string, unknown> | null
+  dedupe_key: string | null
   id: string
-  order_id?: null | string
   workflow: string
 }
 
 type WorkflowQueueService = WorkflowQueueModuleService & {
   createWorkflowQueueItems: (data: {
     arguments: Record<string, unknown>
-    order_id?: string
+    dedupe_key?: string
     run_at: Date
     workflow: string
   }) => Promise<WorkflowQueueItemDTO>
+  listWorkflowQueueItems: (
+    filters?: Record<string, unknown>,
+    config?: Record<string, unknown>
+  ) => Promise<WorkflowQueueItemDTO[]>
+}
+
+function getProductReviewRequestDedupeKey(orderId: string) {
+  return `${PRODUCT_REVIEW_REQUEST_DEDUPE_KEY_PREFIX}-${orderId}`
 }
 
 function isReviewRequestOrder(value: unknown): value is ReviewRequestOrder {
@@ -106,22 +115,23 @@ async function hasReviewRequestEmailLog(
 
 async function hasQueuedReviewRequest(
   container: MedusaContainer,
-  orderId: string
+  dedupeKey: string
 ) {
-  const query = container.resolve<Query>(ContainerRegistrationKeys.QUERY)
-  const { data } = await query.graph({
-    entity: "workflow_queue_item",
-    fields: ["id"],
-    filters: {
-      order_id: orderId,
+  const workflowQueueService = container.resolve<WorkflowQueueService>(
+    WORKFLOW_QUEUE_MODULE
+  )
+  const items = await workflowQueueService.listWorkflowQueueItems(
+    {
+      dedupe_key: dedupeKey,
       workflow: workflowQueueNames.SEND_PRODUCT_REVIEW_REQUEST,
     },
-    pagination: {
+    {
+      select: ["id"],
       take: 1,
-    },
-  })
+    }
+  )
 
-  return Array.isArray(data) && data.length > 0
+  return items.length > 0
 }
 
 export async function scheduleProductReviewRequestForOrder({
@@ -149,9 +159,10 @@ export async function scheduleProductReviewRequestForOrder({
     return null
   }
 
+  const dedupeKey = getProductReviewRequestDedupeKey(order.id)
   const [alreadySent, alreadyQueued] = await Promise.all([
     hasReviewRequestEmailLog(container, order.id),
-    hasQueuedReviewRequest(container, order.id),
+    hasQueuedReviewRequest(container, dedupeKey),
   ])
 
   if (alreadySent || alreadyQueued) {
@@ -177,7 +188,7 @@ export async function scheduleProductReviewRequestForOrder({
 
   const queueItem = await workflowQueueService.createWorkflowQueueItems({
     workflow: workflowQueueNames.SEND_PRODUCT_REVIEW_REQUEST,
-    order_id: order.id,
+    dedupe_key: dedupeKey,
     run_at: runAt,
     arguments: {
       order_id: order.id,