Recover pending Shopify orders by stable order ID

This commit is contained in:
gitea_admin committed 2026-09-06 15:36:58 +02:00
1 parent f739106ab9
commit ba796f0a3f
3 files changed
+52 -5

No files matched your search

@@ -20,7 +20,7 @@ Es werden keine vorbereiteten Shopify-Geschäftsdaten an nachgelagerte Module ü
## 4. Exakter Prozessablauf ## 4. Exakter Prozessablauf
1. `cutoff_timestamp` und `dry_run` validieren; beim Cron den kleineren Wert aus Wasserstand und ältestem offenen Webhook-Eingang verwenden. 1. `cutoff_timestamp` und `dry_run` validieren; beim Cron den kleineren Wert aus Wasserstand und ältestem offenen Webhook-Eingang verwenden und offene Webhook-Order-GIDs zusätzlich direkt zur Recovery aufnehmen.
2. Read-only Shopify-Bestellungen mit `createdAt` nach dem Stichtag in stabiler Reihenfolge laden. 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. 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. 4. Für jede qualifizierte Bestellung die Shopify Order-GID idempotent gegen `sales_order.shopify_order_gid` prüfen.
@@ -43,7 +43,7 @@ Write targets:
## 5. Batch, Betriebsmodell und Einbettung ## 5. Batch, Betriebsmodell und Einbettung
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. 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 sowohl über den Prüfbereich als auch über ihre stabile Order-GID wieder in die Verarbeitung. Ein technisches Event wird erst nach erfolgreicher Projektion als verarbeitet markiert. Der Wasserstand wird nur nach einem vollständig erfolgreichen Lauf fortgeschrieben. Keine Parallelisierung.
## 6. Output ## 6. Output
@@ -62,6 +62,32 @@ function shopify_delta_reconciliation_cutoff(PDO $pdo): string
return ($pendingDate < $stateDate ? $pendingDate : $stateDate)->format(DATE_ATOM); return ($pendingDate < $stateDate ? $pendingDate : $stateDate)->format(DATE_ATOM);
} }
function shopify_delta_load_pending_order_gids(PDO $pdo): array
{
$stmt = $pdo->query(
"SELECT DISTINCT shopify_order_gid
FROM shopify_webhook_event
WHERE status IN ('received', 'processing', 'failed')
AND shopify_order_gid IS NOT NULL
ORDER BY shopify_order_gid"
);
return array_values(array_filter(array_map(
static fn (mixed $value): string => trim((string) $value),
$stmt->fetchAll(PDO::FETCH_COLUMN)
), static fn (string $value): bool => $value !== ''));
}
function shopify_delta_mark_order_events_processed(PDO $pdo, string $orderGid): void
{
$stmt = $pdo->prepare(
"UPDATE shopify_webhook_event
SET status = 'processed', processed_at = NOW(), error_code = NULL, error_message = NULL
WHERE shopify_order_gid = :shopify_order_gid
AND status IN ('received', 'processing', 'failed')"
);
$stmt->execute([':shopify_order_gid' => $orderGid]);
}
function shopify_delta_store_watermark(PDO $pdo, string $cutoff): void function shopify_delta_store_watermark(PDO $pdo, string $cutoff): void
{ {
$stmt = $pdo->prepare( $stmt = $pdo->prepare(
@@ -120,7 +146,7 @@ GQL;
return $orders; return $orders;
} }
function run_shopify_delta(array $env, PDO $pdo, string $cutoff, bool $dryRun): array function run_shopify_delta(array $env, PDO $pdo, string $cutoff, bool $dryRun, array $recoveryOrderGids = []): array
{ {
$cutoffDate = new DateTimeImmutable($cutoff); $cutoffDate = new DateTimeImmutable($cutoff);
$searchQuery = 'created_at:>' . $cutoffDate->format('Y-m-d'); $searchQuery = 'created_at:>' . $cutoffDate->format('Y-m-d');
@@ -135,14 +161,31 @@ function run_shopify_delta(array $env, PDO $pdo, string $cutoff, bool $dryRun):
'clarification_order_gids' => [], 'clarification_order_gids' => [],
]; ];
$ordersByGid = [];
foreach (shopify_delta_load_orders($env, $searchQuery) as $summaryOrder) { foreach (shopify_delta_load_orders($env, $searchQuery) as $summaryOrder) {
$orderGid = trim((string) ($summaryOrder['id'] ?? ''));
if ($orderGid !== '') {
$ordersByGid[$orderGid] = $summaryOrder;
}
}
foreach ($recoveryOrderGids as $recoveryOrderGid) {
if (!isset($ordersByGid[$recoveryOrderGid])) {
$ordersByGid[$recoveryOrderGid] = shopify_load_order_by_gid($env, $recoveryOrderGid);
}
}
$recoverySet = array_fill_keys($recoveryOrderGids, true);
foreach ($ordersByGid as $orderGid => $summaryOrder) {
$createdAt = new DateTimeImmutable((string) ($summaryOrder['createdAt'] ?? '')); $createdAt = new DateTimeImmutable((string) ($summaryOrder['createdAt'] ?? ''));
if ($createdAt <= $cutoffDate) { if (!isset($recoverySet[$orderGid]) && $createdAt <= $cutoffDate) {
continue; continue;
} }
$counters['checked_orders']++; $counters['checked_orders']++;
if (($summaryOrder['cancelledAt'] ?? null) !== null || strtoupper((string) ($summaryOrder['displayFinancialStatus'] ?? '')) === 'REFUNDED') { if (($summaryOrder['cancelledAt'] ?? null) !== null || strtoupper((string) ($summaryOrder['displayFinancialStatus'] ?? '')) === 'REFUNDED') {
$counters['skipped_cancelled_or_refunded']++; $counters['skipped_cancelled_or_refunded']++;
if (!$dryRun) {
shopify_delta_mark_order_events_processed($pdo, (string) $summaryOrder['id']);
}
continue; continue;
} }
@@ -168,6 +211,9 @@ function run_shopify_delta(array $env, PDO $pdo, string $cutoff, bool $dryRun):
$counters['already_imported']++; $counters['already_imported']++;
} else { } else {
$counters['handed_off']++; $counters['handed_off']++;
if (!$dryRun) {
shopify_delta_mark_order_events_processed($pdo, (string) $summaryOrder['id']);
}
} }
} catch (Throwable $error) { } catch (Throwable $error) {
$counters['status'] = 'partial_success'; $counters['status'] = 'partial_success';
+2 -1
View File
@@ -26,9 +26,10 @@ try {
'erp.import-integration.shopify_webhook_worker' => static fn (): array => shopify_webhook_worker_run($pdo, $env, 100), '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 { 'erp.import-integration.shopify_delta_order_import' => static function () use ($pdo, $env, $reconciliationStartedAt): array {
$cutoff = shopify_delta_reconciliation_cutoff($pdo); $cutoff = shopify_delta_reconciliation_cutoff($pdo);
$recoveryOrderGids = shopify_delta_load_pending_order_gids($pdo);
$runId = shopify_delta_start_run($pdo, $cutoff, false); $runId = shopify_delta_start_run($pdo, $cutoff, false);
try { try {
$result = run_shopify_delta($env, $pdo, $cutoff, false); $result = run_shopify_delta($env, $pdo, $cutoff, false, $recoveryOrderGids);
$result['technical_run_id'] = $runId; $result['technical_run_id'] = $runId;
if (($result['status'] ?? '') === 'done') { if (($result['status'] ?? '') === 'done') {
shopify_delta_store_watermark($pdo, $reconciliationStartedAt->format(DATE_ATOM)); shopify_delta_store_watermark($pdo, $reconciliationStartedAt->format(DATE_ATOM));