Help / Data Import Pipeline

Data Import Pipeline

How TulaImporter pulls from product DBs into the ERP.

What it does

The TulaImporter Service Fabric app pulls data from three partner product databases on a schedule and writes it into the ERP database. Each source has its own importer; the orchestrator queues one job at a time so two sources never fight for the ERP at once.

Sources

Source Schema in ERP Importer class
BiometricsProd BiometricsProd.* Sources/BiometricsDB/BiometricsDB/BiometricsImport.cs
ConnectProd ConnectProd.* Sources/ConnectDB/ConnectDB/ConnectImport.cs
GatewayProd GatewayProd.* Sources/GatewayDB/GatewayImport.cs

Scheduling and gating

Two layers of feature flags guard imports — both must be true for a source to run on the schedule:

Setting What it controls
ScheduledImports.Enabled Master kill-switch for all scheduled runs
BiometricsProd.Enabled Per-source enable for Biometrics
ConnectProd.Enabled Per-source enable for Connect
GatewayProd.Enabled Per-source enable for Gateway

Manual runs queued through the API are not gated by these flags — they always run.

Job lifecycle

stateDiagram-v2 [*] --> New: scheduler tick<br/>or POST /api/importjobs New --> Processing: dequeued by RunAsync Processing --> Complete: import finished Processing --> Failed: exception thrown Processing --> Failed: stuck > 6h<br/>(TimeoutStaleJobs) Failed --> [*] Complete --> [*]

Job rows live in dbo.ImportJobs (TulaDataImport DB). The Notes column packs the source name and the requester, separated by a pipe — e.g. BiometricsProd|Scheduled or ConnectProd|Manually failed via API.

End-to-end flow

sequenceDiagram autonumber participant Sched as Scheduler<br/>(RunAsync loop) participant DB as TulaDataImport DB<br/>(dbo.ImportJobs) participant Q as Reliable Queue<br/>ImportJobQueue participant Imp as Source Importer<br/>(Biometrics/Connect/Gateway) participant Src as Product DB participant ERP as TulaERP DB Note over Sched: Every 1h Sched->>DB: INSERT ImportJob (status=New) Sched->>Q: enqueue jobId Q-->>Sched: ack loop Process queue Sched->>Q: dequeue jobId Sched->>DB: UPDATE status=Processing Sched->>Imp: run(jobId) Imp->>Src: SELECT rows (paginated) Src-->>Imp: rows Imp->>ERP: UPSERT into [Schema].[Table] Imp->>DB: UPDATE PercentComplete Imp-->>Sched: done Sched->>DB: UPDATE status=Complete end

Triggering a run by hand

From the Data Import > Importer page in the sidebar — or directly via the API:

POST /api/importjobs
Content-Type: application/json

{
  "importSource": "BiometricsProd",
  "requestedBy":  "jeff"
}

Valid importSource values are BiometricsProd, ConnectProd, GatewayProd. The endpoint creates a New job row and returns its id; the stateful service picks it up on the next loop iteration.

Watching progress

  • Data Import > Job History — paginated list of every job, status, counts.
  • Data Import > Scheduled Jobs — the next-run picture.
  • System > System Status — queue depth + service health.

What if a job is wedged?

POST /api/importjobs/timeout-stale?timeoutHours=6

Flips every Processing job older than the threshold to Failed. The default 6h matches the JobTimeoutThreshold in TulaImport.cs.