From bf875f6dea2fe5133995c68d4324a962edbb0c87 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Mathias=20Gla=CC=88ser?= Date: Sun, 6 Sep 2026 15:31:57 +0200 Subject: [PATCH] Automate Shopify reconciliation in daily cron --- .../0018_shopify_delta_sync_state.sql | 16 +++++++ ...-integration.shopify_delta_order_import.md | 7 +-- .../import-integration/shopify-delta-run.php | 45 +++++++++++++++++++ modules/system/naurua-erp-cron.php | 20 +++++++++ 4 files changed, 85 insertions(+), 3 deletions(-) create mode 100644 db/migrations/0018_shopify_delta_sync_state.sql diff --git a/db/migrations/0018_shopify_delta_sync_state.sql b/db/migrations/0018_shopify_delta_sync_state.sql new file mode 100644 index 0000000..d31fdf2 --- /dev/null +++ b/db/migrations/0018_shopify_delta_sync_state.sql @@ -0,0 +1,16 @@ +BEGIN; + +CREATE TABLE shopify_delta_sync_state ( + process_name TEXT PRIMARY KEY, + last_successful_cutoff TIMESTAMP NOT NULL, + updated_at TIMESTAMP NOT NULL DEFAULT NOW() +); + +INSERT INTO shopify_delta_sync_state (process_name, last_successful_cutoff) +SELECT + 'erp.import-integration.shopify_delta_order_import', + COALESCE(MAX(shopify_created_at), NOW()) +FROM sales_order +WHERE shopify_order_gid IS NOT NULL; + +COMMIT; diff --git a/docs/modules/erp/import-integration/processes/main/erp.import-integration.shopify_delta_order_import.md b/docs/modules/erp/import-integration/processes/main/erp.import-integration.shopify_delta_order_import.md index 889207f..b58ff0b 100644 --- a/docs/modules/erp/import-integration/processes/main/erp.import-integration.shopify_delta_order_import.md +++ b/docs/modules/erp/import-integration/processes/main/erp.import-integration.shopify_delta_order_import.md @@ -8,18 +8,19 @@ Read-only-Abgleich von Shopify-Bestellungen nach dem ERP-Stichtag und identifier ## 2. Prozess-Einbettung -Manuell startbarer Initial-Deltalauf und später wiederverwendbarer Reconciliation-Prozess im Submodul `erp/import-integration`. Er ist vom Webhook-Empfang getrennt. +Manuell startbarer Initial-Deltalauf und täglich über den zentralen ERP-Cron gestarteter Reconciliation-Prozess im Submodul `erp/import-integration`. Er ist vom Webhook-Empfang getrennt. ## 3. Input - `cutoff_timestamp` - optional `dry_run` +- beim täglichen Cron der persistierte letzte erfolgreiche Wasserstand sowie der älteste noch nicht erfolgreich verarbeitete Webhook-Eingang Es werden keine vorbereiteten Shopify-Geschäftsdaten an nachgelagerte Module übergeben. ## 4. Exakter Prozessablauf -1. `cutoff_timestamp` und `dry_run` validieren. +1. `cutoff_timestamp` und `dry_run` validieren; beim Cron den kleineren Wert aus Wasserstand und ältestem offenen Webhook-Eingang verwenden. 2. Read-only Shopify-Bestellungen mit `createdAt` nach dem Stichtag in stabiler Reihenfolge laden. 3. Stornierte oder vollständig erstattete Bestellungen als nicht lagerwirksam klassifizieren und nur technisch protokollieren. 4. Für jede qualifizierte Bestellung die Shopify Order-GID idempotent gegen `sales_order.shopify_order_gid` prüfen. @@ -42,7 +43,7 @@ Write targets: ## 5. Batch, Betriebsmodell und Einbettung -Der Initiallauf verarbeitet den festen Zeitraum nach dem bestehenden ERP-Stichtag seriell. Spätere Reconciliation-Läufe nutzen denselben Prozess mit einem definierten Wasserzeichen. Keine Parallelisierung. +Der Initiallauf verarbeitet den festen Zeitraum nach dem bestehenden ERP-Stichtag seriell. Spätere tägliche Reconciliation-Läufe nutzen den persistierten Wasserstand und ziehen offene oder fehlgeschlagene Webhook-Eingänge wieder in den Prüfbereich. Der Wasserstand wird nur nach einem vollständig erfolgreichen Lauf fortgeschrieben. Keine Parallelisierung. ## 6. Output diff --git a/modules/erp/import-integration/shopify-delta-run.php b/modules/erp/import-integration/shopify-delta-run.php index 8329ec9..fde3ec8 100644 --- a/modules/erp/import-integration/shopify-delta-run.php +++ b/modules/erp/import-integration/shopify-delta-run.php @@ -33,6 +33,51 @@ function shopify_delta_start_run(PDO $pdo, string $cutoff, bool $dryRun): int return (int) $stmt->fetchColumn(); } +function shopify_delta_reconciliation_cutoff(PDO $pdo): string +{ + $stateStmt = $pdo->prepare( + 'SELECT last_successful_cutoff + FROM shopify_delta_sync_state + WHERE process_name = :process_name' + ); + $stateStmt->execute([':process_name' => 'erp.import-integration.shopify_delta_order_import']); + $state = $stateStmt->fetchColumn(); + if ($state === false || trim((string) $state) === '') { + throw new RuntimeException('Shopify-Delta-Wasserstand fehlt'); + } + + $pendingStmt = $pdo->query( + "SELECT MIN(received_at) + FROM shopify_webhook_event + WHERE status IN ('received', 'processing', 'failed') + AND shopify_order_gid IS NOT NULL" + ); + $pending = $pendingStmt->fetchColumn(); + if ($pending === false || $pending === null) { + return (string) $state; + } + + $stateDate = new DateTimeImmutable((string) $state); + $pendingDate = new DateTimeImmutable((string) $pending); + return ($pendingDate < $stateDate ? $pendingDate : $stateDate)->format(DATE_ATOM); +} + +function shopify_delta_store_watermark(PDO $pdo, string $cutoff): void +{ + $stmt = $pdo->prepare( + 'UPDATE shopify_delta_sync_state + SET last_successful_cutoff = :cutoff, updated_at = NOW() + WHERE process_name = :process_name' + ); + $stmt->execute([ + ':cutoff' => $cutoff, + ':process_name' => 'erp.import-integration.shopify_delta_order_import', + ]); + if ($stmt->rowCount() !== 1) { + throw new RuntimeException('Shopify-Delta-Wasserstand konnte nicht gespeichert werden'); + } +} + function shopify_delta_finish_run(PDO $pdo, int $runId, string $status, array $result, ?Throwable $error = null): void { $stmt = $pdo->prepare( diff --git a/modules/system/naurua-erp-cron.php b/modules/system/naurua-erp-cron.php index 317f8d2..ca46fbd 100644 --- a/modules/system/naurua-erp-cron.php +++ b/modules/system/naurua-erp-cron.php @@ -3,6 +3,8 @@ declare(strict_types=1); require_once __DIR__ . '/../shared/db.php'; require_once __DIR__ . '/../erp/lager/forecast.php'; +require_once __DIR__ . '/../erp/import-integration/shopify-webhook-worker.php'; +require_once __DIR__ . '/../erp/import-integration/shopify-delta-run.php'; if (PHP_SAPI !== 'cli') { http_response_code(404); @@ -19,7 +21,25 @@ $env = expand_env_values(parse_env_file(dirname(__DIR__, 2) . '/.env')); $pdo = connect_database($env); try { + $reconciliationStartedAt = new DateTimeImmutable('now', new DateTimeZone('Europe/Zurich')); $jobs = [ + 'erp.import-integration.shopify_webhook_worker' => static fn (): array => shopify_webhook_worker_run($pdo, $env, 100), + 'erp.import-integration.shopify_delta_order_import' => static function () use ($pdo, $env, $reconciliationStartedAt): array { + $cutoff = shopify_delta_reconciliation_cutoff($pdo); + $runId = shopify_delta_start_run($pdo, $cutoff, false); + try { + $result = run_shopify_delta($env, $pdo, $cutoff, false); + $result['technical_run_id'] = $runId; + if (($result['status'] ?? '') === 'done') { + shopify_delta_store_watermark($pdo, $reconciliationStartedAt->format(DATE_ATOM)); + } + shopify_delta_finish_run($pdo, $runId, (string) $result['status'], $result); + return $result; + } catch (Throwable $error) { + shopify_delta_finish_run($pdo, $runId, 'failed', ['status' => 'failed', 'technical_run_id' => $runId], $error); + throw $error; + } + }, 'erp.lager.stock_lot_forecast_refresh' => static fn (): array => run_lager_stock_lot_forecast($pdo), ]; $results = [];