From 8c4b9a7f8998e1f0e1a5f3a8936a7a95b6f6c181 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Mathias=20Gla=CC=88ser?= Date: Wed, 12 Aug 2026 18:15:24 +0200 Subject: [PATCH] Process Shopify webhooks asynchronously --- .../0012_shopify_webhook_processing.sql | 13 ++ .../import-integration/shopify-delta-run.php | 30 +--- .../erp/import-integration/shopify-source.php | 29 ++++ .../shopify-webhook-worker.php | 150 ++++++++++++++++++ .../import-integration/shopify-webhook.php | 2 +- 5 files changed, 195 insertions(+), 29 deletions(-) create mode 100644 db/migrations/0012_shopify_webhook_processing.sql create mode 100644 modules/erp/import-integration/shopify-source.php create mode 100644 modules/erp/import-integration/shopify-webhook-worker.php diff --git a/db/migrations/0012_shopify_webhook_processing.sql b/db/migrations/0012_shopify_webhook_processing.sql new file mode 100644 index 0000000..315add1 --- /dev/null +++ b/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; diff --git a/modules/erp/import-integration/shopify-delta-run.php b/modules/erp/import-integration/shopify-delta-run.php index 9352d44..8329ec9 100644 --- a/modules/erp/import-integration/shopify-delta-run.php +++ b/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') { diff --git a/modules/erp/import-integration/shopify-source.php b/modules/erp/import-integration/shopify-source.php new file mode 100644 index 0000000..c792c2e --- /dev/null +++ b/modules/erp/import-integration/shopify-source.php @@ -0,0 +1,29 @@ + $orderGid]); + $order = $data['order'] ?? null; + if (!is_array($order)) { + throw new RuntimeException("Shopify-Bestellung nicht gefunden: {$orderGid}"); + } + return $order; +} diff --git a/modules/erp/import-integration/shopify-webhook-worker.php b/modules/erp/import-integration/shopify-webhook-worker.php new file mode 100644 index 0000000..70127c6 --- /dev/null +++ b/modules/erp/import-integration/shopify-webhook-worker.php @@ -0,0 +1,150 @@ +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); +} diff --git a/modules/erp/import-integration/shopify-webhook.php b/modules/erp/import-integration/shopify-webhook.php index efae453..5136f00 100644 --- a/modules/erp/import-integration/shopify-webhook.php +++ b/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' );