Automate Shopify reconciliation in daily cron

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;

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
 

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(

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 = [];