Process Shopify webhooks asynchronously

db/migrations/0012_shopify_webhook_processing.sql

@@ -0,0 +1,13 @@
+BEGIN;
+
+ALTER TABLE shopify_webhook_event
+  DROP CONSTRAINT IF EXISTS chk_shopify_webhook_event_status;
+
+ALTER TABLE shopify_webhook_event
+  ADD CONSTRAINT chk_shopify_webhook_event_status
+  CHECK (status IN ('received', 'processing', 'processed', 'duplicate', 'failed'));
+
+CREATE INDEX IF NOT EXISTS idx_shopify_webhook_event_processing
+  ON shopify_webhook_event(status, received_at, id);
+
+COMMIT;

modules/erp/import-integration/shopify-delta-run.php

@@ -5,6 +5,7 @@ require_once __DIR__ . '/../../../modules/shared/db.php';
 require_once __DIR__ . '/../../../modules/shared/webhook_throttle.php';
 require_once __DIR__ . '/shopify-api.php';
 require_once __DIR__ . '/service.php';
+require_once __DIR__ . '/shopify-source.php';
 require_once __DIR__ . '/../bestellungen/shopify-projection.php';
 
 function shopify_delta_argument(string $name, array $argv, ?string $default = null): ?string
@@ -74,33 +75,6 @@ GQL;
     return $orders;
 }
 
-function shopify_delta_load_order(array $env, string $orderGid): array
-{
-    $query = <<<'GQL'
-query ($id: ID!) {
-  order(id: $id) {
-    id name createdAt updatedAt cancelledAt displayFinancialStatus displayFulfillmentStatus
-    currentSubtotalPriceSet { shopMoney { amount currencyCode } }
-    totalShippingPriceSet { shopMoney { amount currencyCode } }
-    totalTaxSet { shopMoney { amount currencyCode } }
-    totalPriceSet { shopMoney { amount currencyCode } }
-    customer { id firstName lastName email phone }
-    billingAddress { firstName lastName company address1 address2 zip city provinceCode country countryCodeV2 phone }
-    shippingAddress { firstName lastName company address1 address2 zip city provinceCode country countryCodeV2 phone }
-    lineItems(first: 50) {
-      nodes { id sku name quantity originalUnitPriceSet { shopMoney { amount currencyCode } } variant { id sku product { id } } }
-    }
-  }
-}
-GQL;
-    $data = shopify_graphql_query($env, $query, ['id' => $orderGid]);
-    $order = $data['order'] ?? null;
-    if (!is_array($order)) {
-        throw new RuntimeException("Shopify-Bestellung nicht gefunden: {$orderGid}");
-    }
-    return $order;
-}
-
 function run_shopify_delta(array $env, PDO $pdo, string $cutoff, bool $dryRun): array
 {
     $cutoffDate = new DateTimeImmutable($cutoff);
@@ -132,7 +106,7 @@ function run_shopify_delta(array $env, PDO $pdo, string $cutoff, bool $dryRun):
             $result = project_shopify_order(
                 $pdo,
                 $orderGid,
-                static fn (string $gid): array => shopify_delta_load_order($env, $gid),
+                static fn (string $gid): array => shopify_load_order_by_gid($env, $gid),
                 $dryRun
             );
             if (!$dryRun && ($result['status'] ?? '') === 'imported') {

modules/erp/import-integration/shopify-source.php

@@ -0,0 +1,29 @@
+<?php
+declare(strict_types=1);
+
+function shopify_load_order_by_gid(array $env, string $orderGid): array
+{
+    $query = <<<'GQL'
+query ($id: ID!) {
+  order(id: $id) {
+    id name createdAt updatedAt cancelledAt displayFinancialStatus displayFulfillmentStatus
+    currentSubtotalPriceSet { shopMoney { amount currencyCode } }
+    totalShippingPriceSet { shopMoney { amount currencyCode } }
+    totalTaxSet { shopMoney { amount currencyCode } }
+    totalPriceSet { shopMoney { amount currencyCode } }
+    customer { id firstName lastName email phone }
+    billingAddress { firstName lastName company address1 address2 zip city provinceCode country countryCodeV2 phone }
+    shippingAddress { firstName lastName company address1 address2 zip city provinceCode country countryCodeV2 phone }
+    lineItems(first: 50) {
+      nodes { id sku name quantity originalUnitPriceSet { shopMoney { amount currencyCode } } variant { id sku product { id } } }
+    }
+  }
+}
+GQL;
+    $data = shopify_graphql_query($env, $query, ['id' => $orderGid]);
+    $order = $data['order'] ?? null;
+    if (!is_array($order)) {
+        throw new RuntimeException("Shopify-Bestellung nicht gefunden: {$orderGid}");
+    }
+    return $order;
+}

modules/erp/import-integration/shopify-webhook-worker.php

@@ -0,0 +1,150 @@
+<?php
+declare(strict_types=1);
+
+require_once __DIR__ . '/../../../modules/shared/db.php';
+require_once __DIR__ . '/../../../modules/shared/webhook_throttle.php';
+require_once __DIR__ . '/shopify-api.php';
+require_once __DIR__ . '/shopify-source.php';
+require_once __DIR__ . '/service.php';
+require_once __DIR__ . '/../bestellungen/shopify-projection.php';
+
+function shopify_webhook_worker_start(PDO $pdo, int $limit): int
+{
+    $stmt = $pdo->prepare(
+        "INSERT INTO process_runs (process_name, status, scope_json, requested_by)
+         VALUES ('erp.import-integration.shopify_webhook_worker', 'running', :scope_json::jsonb, 'cron')
+         RETURNING id"
+    );
+    $stmt->execute([':scope_json' => json_encode(['limit' => $limit], JSON_THROW_ON_ERROR)]);
+    return (int) $stmt->fetchColumn();
+}
+
+function shopify_webhook_worker_finish(PDO $pdo, int $runId, string $status, array $result, ?Throwable $error = null): void
+{
+    $stmt = $pdo->prepare(
+        'UPDATE process_runs
+         SET status = :status, result_json = :result_json::jsonb, finished_at = NOW(),
+             error_code = :error_code, error_message = :error_message
+         WHERE id = :id'
+    );
+    $stmt->execute([
+        ':status' => $status,
+        ':result_json' => json_encode($result, JSON_THROW_ON_ERROR),
+        ':error_code' => $error === null ? null : 'shopify_webhook_worker_failed',
+        ':error_message' => $error?->getMessage(),
+        ':id' => $runId,
+    ]);
+}
+
+function shopify_webhook_worker_claim(PDO $pdo): ?array
+{
+    $pdo->beginTransaction();
+    $stmt = $pdo->query(
+        "SELECT id, topic, shopify_order_gid
+         FROM shopify_webhook_event
+         WHERE status = 'received'
+         ORDER BY received_at, id
+         LIMIT 1
+         FOR UPDATE SKIP LOCKED"
+    );
+    $event = $stmt->fetch(PDO::FETCH_ASSOC);
+    if (!is_array($event)) {
+        $pdo->commit();
+        return null;
+    }
+
+    $update = $pdo->prepare("UPDATE shopify_webhook_event SET status = 'processing' WHERE id = :id");
+    $update->execute([':id' => (int) $event['id']]);
+    $pdo->commit();
+    return $event;
+}
+
+function shopify_webhook_worker_run(PDO $pdo, array $env, int $limit): array
+{
+    $runId = shopify_webhook_worker_start($pdo, $limit);
+    $result = [
+        'status' => 'done',
+        'processed' => 0,
+        'skipped' => 0,
+        'failed' => 0,
+        'n8n_triggered' => 0,
+        'failed_event_ids' => [],
+        'technical_run_id' => $runId,
+    ];
+
+    try {
+        for ($i = 0; $i < $limit; $i++) {
+            $event = shopify_webhook_worker_claim($pdo);
+            if ($event === null) {
+                break;
+            }
+
+            $eventId = (int) $event['id'];
+            try {
+                $topic = (string) $event['topic'];
+                $orderGid = trim((string) ($event['shopify_order_gid'] ?? ''));
+                if ($topic !== 'orders/create' || $orderGid === '') {
+                    $result['skipped']++;
+                    $status = 'processed';
+                } else {
+                    $order = shopify_load_order_by_gid($env, $orderGid);
+                    if (($order['cancelledAt'] ?? null) !== null || strtoupper((string) ($order['displayFinancialStatus'] ?? '')) === 'REFUNDED') {
+                        $result['skipped']++;
+                        $status = 'processed';
+                    } else {
+                        $projection = project_shopify_order(
+                            $pdo,
+                            $orderGid,
+                            static fn (string $gid): array => $order,
+                            false
+                        );
+                        if (($projection['status'] ?? '') === 'imported') {
+                            $label = trigger_shipping_label_flow(
+                                build_legacy_label_order_payload($pdo, (int) $projection['sales_order_id']),
+                                $env
+                            );
+                            $excel = trigger_excel_webhook((string) ($order['name'] ?? $orderGid), $env);
+                            if (!$label['ok'] || !$excel['ok']) {
+                                throw new RuntimeException('Mindestens ein n8n-Webhook ist fehlgeschlagen');
+                            }
+                            $result['n8n_triggered']++;
+                        }
+                        $result['processed']++;
+                        $status = 'processed';
+                    }
+                }
+
+                $stmt = $pdo->prepare("UPDATE shopify_webhook_event SET status = :status, processed_at = NOW(), error_code = NULL, error_message = NULL WHERE id = :id");
+                $stmt->execute([':status' => $status, ':id' => $eventId]);
+            } catch (Throwable $error) {
+                $result['status'] = 'partial_success';
+                $result['failed']++;
+                $result['failed_event_ids'][] = $eventId;
+                $stmt = $pdo->prepare("UPDATE shopify_webhook_event SET status = 'failed', error_code = :error_code, error_message = :error_message WHERE id = :id");
+                $stmt->execute([':error_code' => 'shopify_webhook_event_processing_failed', ':error_message' => $error->getMessage(), ':id' => $eventId]);
+            }
+        }
+        shopify_webhook_worker_finish($pdo, $runId, $result['status'], $result);
+        return $result;
+    } catch (Throwable $error) {
+        shopify_webhook_worker_finish($pdo, $runId, 'failed', $result, $error);
+        throw $error;
+    }
+}
+
+$argvLimit = 20;
+foreach ($argv as $argument) {
+    if (str_starts_with($argument, '--limit=')) {
+        $argvLimit = max(1, min(100, (int) substr($argument, 8)));
+    }
+}
+
+try {
+    $env = expand_env_values(parse_env_file(__DIR__ . '/../../../.env'));
+    $pdo = connect_database($env);
+    echo json_encode(shopify_webhook_worker_run($pdo, $env, $argvLimit), JSON_UNESCAPED_SLASHES) . PHP_EOL;
+    exit(0);
+} catch (Throwable $error) {
+    fwrite(STDERR, $error->getMessage() . PHP_EOL);
+    exit(1);
+}

modules/erp/import-integration/shopify-webhook.php

@@ -112,7 +112,7 @@ function handle_shopify_webhook(PDO $pdo, array $env, string $rawPayload, array
         $stmt = $pdo->prepare(
             'INSERT INTO public.shopify_webhook_event
                 (webhook_id, topic, shop_domain, payload_sha256, payload, shopify_order_gid, status)
-             VALUES (:webhook_id, :topic, :shop_domain, :payload_sha256, :payload::jsonb, :shopify_order_gid, \'processed\')
+             VALUES (:webhook_id, :topic, :shop_domain, :payload_sha256, :payload::jsonb, :shopify_order_gid, \'received\')
              ON CONFLICT (webhook_id) DO NOTHING
              RETURNING id'
         );