---
module: vault.import
title: Import
tab: Architect
order: 30
audience: dev
---

How the CIT CSV import feature is built. Single owning service — no cross-service or event fan-out on this path. See [platform.service-ownership].

## Services involved

- **`hiveops-vault` backend** (Spring Boot, port 8094, image `hiveiq-vault`) — owns the entire import path end to end.
- **`hiveops-vault` frontend** (Svelte SPA, port 5185, `vault.bcos.cloud`) — `components/Import/ImportView.svelte` drives upload + polling via `importAPI` in `lib/api.ts`.
- No other service participates. There is **no Kafka on the import path** (vault produces nothing; it only *consumes* `hiveops.device.events` to keep its `device_summary` mirror, unrelated to import). See [platform.kafka].

## Data flow

```
ImportView.svelte
  └─ importAPI.upload(file)  POST /api/vault/imports  (multipart "file")
       └─ ImportController.uploadCsv()
            1. validate: reject empty; reject content-type without "csv"/"text"
            2. save CitImport row  (status=PROCESSING, importedBy=principal.email, format=CSV)
            3. CsvImportService.processAsync(id, file.getInputStream(), institutionKey, email)
            4. return 202 ACCEPTED + ImportResponse(PROCESSING)     ← request ends here
  └─ startPolling(id): every 2s GET /api/vault/imports/{id}
       └─ until status == COMPLETED | FAILED

CsvImportService.processAsync   @Async("importExecutor") @Transactional
  ├─ reload CitImport by id
  ├─ parse(): read header, group data rows by  atm_id|visit_date|vendor|reference
  │     • cassette row (11 cols) → CitCassetteCount builder
  │     • REMOVAL row (8 cols, slot="REMOVAL") → CitRemoval builder
  ├─ per group: save CitVisit (+ nested cassette counts + removals)  via CitVisitRepository
  └─ set status COMPLETED + recordCount(#visits)  |  on any exception → FAILED + errorMessage
```

The upload is fire-and-forget from the client's perspective: `POST` returns `202` immediately and the browser polls the detail endpoint until terminal. This is the **executor → async worker → record-store** pattern; here the "record store" is Postgres directly (no Kafka intermediary — contrast with fleet's executor→Kafka→record-store). See [platform.data-architecture].

## Async execution

- `CsvImportService.processAsync` is annotated `@Async("importExecutor")` (thread pool defined in `AsyncConfig`) and `@Transactional` — the **entire file parse + all visit saves are one transaction**. Any thrown exception (e.g. `Instant.parse` on a bad date) rolls back every visit and marks the import `FAILED`.
- The controller's `CitImport` save commits in its own transaction before the async worker reloads it by id, so the worker always finds the `PROCESSING` row.

## DB tables (all in `hiveiq_vault`, all scoped by `institution_key`)

| Table | Written by import | Notes |
|-------|-------------------|-------|
| `cit_imports` | ✅ insert (PROCESSING) then update (COMPLETED/FAILED, `record_count`, `error_message`) | audit row per upload; `imported_by` = JWT email, `format` = `CSV` |
| `cit_visits` | ✅ one row per `atm_id+visit_date+vendor+reference` group | FK `cit_import` back to `cit_imports`; `created_by` = importer email |
| `cit_cassette_counts` | ✅ one row per cassette CSV line | slot, denomination, `cassette_type`, opening/added/removed/closing |
| `cit_removals` | ✅ one row per REMOVAL CSV line | `removal_type` (CHECK/DEPOSIT/REJECTED_NOTE), count, amount |
| `device_summary` | ❌ read-only mirror (not touched by import) | Kafka-fed from `hiveops.device.events`; only relevant if you cross-check ATM validity |

## Key API endpoints

Base `/api/vault/imports` (`ImportController`). JWT Bearer required; results scoped to `principal.institutionKey()`.

| Method | Path | Purpose |
|--------|------|---------|
| POST | `/api/vault/imports` | multipart `file` → save PROCESSING row, kick async, return **202** |
| GET | `/api/vault/imports` | paginated history, sorted `importedAt` DESC (`?page=&size=`, default size 25) |
| GET | `/api/vault/imports/{id}` | single import status (polled by UI); 404 if not in caller's institution |

## Security & routing

- `SecurityConfig` = JWT filter (`JwtAuthenticationFilter` → `VaultPrincipal`) + `anyRequest().authenticated()`. **No role checks** — authorization is institution-scoping only.
- Spring CORS is disabled; NGINX is the CORS authority and maps `api.bcos.cloud/vault/` → `hiveiq-vault:8094/api/vault/`.

Cross-links: [platform.data-architecture] · [platform.kafka] · [platform.service-ownership] · [technical.vault]
