How the asynchronous ingestion pipeline is wired for engineers: the entry point and every step, with the services it uses and the concrete resources it touches — database tables, SQL stored procedures, blob & PVC storage, Redis keys, SignalR, Hangfire and domain events.
The pipeline is asynchronous and choreographed: each step is a Hangfire background command that runs in its own transaction, records its outcome, pushes live progress over SignalR, and enqueues the next command. There is no central orchestrator — the chain advances itself, and a business failure simply stops it.
Every step inherits InboundDeliveryIngestionStepHandler<TCommand>, which runs the same lifecycle around the step's own ExecuteAsync:
GetWithStepsAsync (with steps + creator).HaltIfStopRequestedAsync reads the Redis stop flag; persists Stopped if set.Success / Skipped / BusinessFailure / NoOp.RecordStep* + SaveChangesAsync (dispatches domain events).Each command runs under TransactionResultCommandBehavior (ReadCommitted, 300 s). Transient exceptions
propagate and Hangfire retries (10×); a business failure records the step as failed and stops the chain (no next command).
Two views of the same machine: the control flow of commands and events, and the status the aggregate moves through.
→ command enqueued (Hangfire) · ⏱ delayed schedule (grace) · ⇢ domain / integration event. Each step is its own background transaction.
CreateInboundDeliveryIngestionCommand
→create
ingestion→ queued
⇢post-commit
…QueuedIntegrationEvent
→enqueue
StartPipelineWhenIngestionQueuedEventHandler
AwaitGracePeriodStepHandler
⏱schedule
CreateInboundDeliveryStepCommandcarries grace token
CreateInboundDeliveryStepCommand
→handler
InboundDelivery.Create()
InboundDeliveryCreatedInternalEvent
→
SupplyInboundOrdersEventHandler
→if overflow
inbound ordergenerated
InboundDeliveryCompletedInternalEvent
→
PickTower loadPT_Load_InboundDelivery
SupplyInboundOrders in the same save) and records it inline (→ IBO generated) — no follow-up command, no Succeeded→IboGenerated flash.InboundOrderCompletedIntegrationEvent
→
SucceedIngestion…Handler
→
ingestion→ succeeded
The aggregate's status and what drives each transition. Dashed blue = user re-entry.
start resumes the same attempt · restart begins a new attempt · resume re-arms the grace window. Failed and Stopped are deletable. Internally the converge sets Succeeded first, then capture flips it to IBO generated when an order was produced.
Everything that can move an ingestion forward.
POST /DeliveriesManagement/Inbounds/IngestionsCreateInboundDeliveryIngestionCommand creates the aggregate in Queued.InboundDeliveryIngestionQueuedIntegrationEvent → StartPipelineWhenIngestionQueuedEventHandlerParseRawPayloadStepCommand). This is what actually starts the pipeline — nothing runs inside the HTTP request.EnqueueNext<Name>StepCommand via Hangfire. The chain is self-advancing.IBackgroundService.Schedule(CreateInboundDeliveryStepCommand, grace)GraceToken. Stop/Resume re-arm or invalidate it.POST {id}/stop · /resume · /start · /restart · DELETE {id}InboundOrderCompletedIntegrationEvent → SucceedIngestionWhenGeneratedOrderCompletedEventHandlerIBO generated → Succeeded.PurgeStaleIngestionRawPayloadsCronJob (daily 03:00)Succeeded / IboGenerated ingestions older than the retention window.Referenced by every step — listed once here rather than repeated below.
IInboundDeliveryIngestionStorage at AppSettings.InboundDeliveryStoragePrefix.{prefix}/ingestions/{id}/attempt-{n}/ → {n}.raw, parsed.json, preparedkeys.json.{prefix}.…:ibd-ingestion:{id} — progress snapshot (24 h).…:ibd-ingestion-stop:{id} — stop flag (6 h).…:ibd-key-reservation:{hash} + …-owned:{id} — key reservations (Lua, all-or-nothing, 10 min).lock:ibd-ingestion-upload:{id} — upload lock (30 s).IIngestionProgressService writes the Redis snapshot and broadcasts InboundDeliveryIngestionUpdated on WholeSaleHub (/notifications) via INotificationService.BroadcastToAll.PublishStepProgressAsync for per-item counts.IBackgroundService.Enqueue (next step) / Schedule (delayed converge), routed by MediatRHangfireBridge over Redis.HangfireFailedJobNotification email.What a running step touches. The handler is a Hangfire worker; everything else is a backing system it reads from or writes to.
Per step: what it does, the services it depends on, the concrete things it calls, the artifacts it reads/writes, the events it raises, and how it can fail.
Stop / resume / start / restart / delete — and how each touches state vs. cache.
While running, sets the Redis stop flag only — no aggregate write, zero concurrency risk; the next step's start check catches it and persists Stopped. During the grace window (no step running) it persists Stopped synchronously via PauseGracePeriod (sole writer).
Clears the stop flag, calls ResumeGracePeriod (fresh GraceToken, same attempt, prepared keys reused), and Schedules a new converge. The old scheduled converge no-ops on the token guard.
Clears the stop flag and calls Start() (→ Queued, same attempt) — the pipeline resumes from where it paused; enqueued via the queued event. Hidden for grace-paused (use Resume).
Clears the stop flag and calls RestartFromBeginning — bumps AttemptNumber and re-runs from step 1; the prior attempt is kept as history.
Removes the aggregate, clears the cache + broadcasts InboundDeliveryIngestionRemoved, releases any Redis key reservations, and sweeps the PVC attempt folder.
The converge records any generated IBO inline (id read back from the scoped IGeneratedInboundOrderSink in the same transaction) and moves the ingestion to IBO generated — no follow-up command. When the user completes that order, InboundOrderCompletedIntegrationEvent → SucceedIngestionWhenGeneratedOrderCompletedEventHandler → Succeeded.