---
module: transactions.reparse
title: Journal Reparse
tab: Architect
order: 30
audience: dev
---

How **Journal Reparse** is built end to end. The feature spans two services: the **transactions** SPA/proxy (UI + auth pass-through) and **hiveops-journal** (the actual owner of files, transactions, and the reparse action). See [platform.service-ownership].

## Services involved

| Service | Role in this feature |
|---------|----------------------|
| `hiveops-transactions` (frontend + proxy, port 8098) | Serves `JournalReparse.svelte`; proxies `/api/journal/**` to journal. Owns **no** data — stateless BFF. See [platform.service-ownership]. |
| `hiveops-journal` **upload role** (`JOURNAL_ROLE=upload`, port 8087) | Hosts `AdminController` + `JournalApiController`; resets file state; **produces** parse-request messages to Kafka. |
| `hiveops-journal` **worker role** (`JOURNAL_ROLE=worker`) | Consumes per-institution parse-request topics; runs the actual parse; writes transactions; emits `journal.file-parsed`. |
| Kafka | Decouples the reset/enqueue (upload) from the CPU-heavy parse (worker). See [platform.kafka]. |
| PostgreSQL `hiveiq_journal` | Stores `journal_files` and `journal_transactions`. See [platform.data-architecture]. |

## Request path (executor → Kafka → record-store pattern)

```
JournalReparse.svelte
  └─ journalAdminAPI.reparseAll(atmId?, journalType)   [axios, base = transactions proxy /api]
       └─ POST /api/journal/admin/reparse-all?atmId=&journalType=
            └─ transactions proxy (JournalProxyController.forward, JOURNAL_URL)
                 └─ hiveops-journal AdminController.reparseAll   [MSP_ADMIN/BCOS_ADMIN]
                      ├─ select journal_files (by atmId / journalType filter)
                      ├─ FOR EACH file:
                      │    ├─ DELETE journal_transactions WHERE file   (deleteByJournalFile)
                      │    ├─ UPDATE journal_files SET parse_status=PENDING,
                      │    │        transaction_count=NULL, parse_error=NULL, last_parsed_size=NULL
                      │    └─ JournalParsingService.parseJournalFileAsync(file)
                      │         └─ kafkaTemplate.send("journal.parse-requests.{KEY}", atmId, fileId)
                      └─ return { message, queued, total }
```

Then, asynchronously on the **worker**:

```
JournalParseConsumer  @KafkaListener(topics = "#{@parseTopicName}")   // one topic per worker
  ├─ UPDATE journal_files SET parse_status=PARSING
  ├─ JournalParser.parseJournalFile(file)  → INSERT journal_transactions
  ├─ UPDATE journal_files SET parse_status=PARSED (transaction_count, parsed_at, …)
  │     └─ on failure: parse_status=ERROR, parse_error=<message>
  └─ FileParsedEventProducer → kafkaTemplate.send("journal.file-parsed", atmId, event)
        ├─ CassetteInventoryService  @KafkaListener(journal.file-parsed)   // cassette backfill
        └─ JournalGapDetectionService @KafkaListener(journal.file-parsed)  // gap re-check
```

## Institution → topic routing

`JournalParsingService.resolveTopic(institution)` decides the parse-request topic:
1. `jf.getInstitution()` if already set (uppercase key, e.g. `AICU`).
2. else `AtmInstitutionSyncService.resolve(atmId)` — local map synced from the incident DB.
3. else `"unassigned"`.

- Known institution (in `JOURNAL_INSTITUTIONS`) → `journal.parse-requests.{KEY}` (uppercased).
- Unknown/blank → `journal.parse-requests.UNASSIGNED`.

Each worker container listens to exactly one topic via the Spring bean `@parseTopicName` (derived from `JOURNAL_INSTITUTION`). This is how parsing scales horizontally per institution. See [platform.kafka].

## Data flow — DB tables

| Table (DB `hiveiq_journal`) | Read | Written by reparse |
|-----------------------------|------|--------------------|
| `journal_files` (`@Table journal_files`) | file list + status (`GET /files`), reparse selection | status reset (upload), PARSING/PARSED/ERROR (worker) |
| `journal_transactions` (`@Table journal_transactions`) | transaction counts | **deleted** at reparse start, re-inserted by worker |

`GET /api/journal/files` returns a `PageResponse<JournalFileRecord>` (`atmId`, `journalDate`, `parseStatus`, `transactionCount`, `parseError`, …), rendered as status pills + table. See [platform.data-architecture].

## Kafka topics

| Topic | Producer | Consumer | Purpose |
|-------|----------|----------|---------|
| `journal.parse-requests.{KEY}` / `.UNASSIGNED` | upload (`parseJournalFileAsync`) | worker (`JournalParseConsumer`) | one parse job = one file id, keyed by atmId |
| `journal.file-parsed` | worker (`FileParsedEventProducer`) | `CassetteInventoryService`, `JournalGapDetectionService` | fan-out after a file finishes parsing |

Static topics `journal.parse-requests` (base) and `journal.replenishments` are also declared but not on the reparse path. See [platform.kafka].

## Resilience

Because reset (upload) and parse (worker) are decoupled, lost messages are recovered by `JournalParsingService`:
- `reconcileStranded()` — `@EventListener(ApplicationReadyEvent)` on upload startup, and on-demand via `POST /api/journal/admin/reconcile`; re-queues all PENDING/PARSING.
- `recoverStaleParsing()` — `@Scheduled` every 30 min; re-queues files stuck in PARSING > 30 min (worker died mid-parse); a companion sweep re-queues PENDING > 30 min (Kafka send failed silently).

## API endpoints (this feature)

| Method | Path (journal svc) | Controller | Role |
|--------|--------------------|------------|------|
| POST | `/api/journal/admin/reparse-all?atmId=&journalType=` | `AdminController` | `MSP_ADMIN`/`BCOS_ADMIN` |
| GET | `/api/journal/files?atmId=&journalType=&page=&size=` | `JournalApiController` | `isAuthenticated()` |
| POST | `/api/journal/files/{id}/reparse` | `JournalApiController` | `isAuthenticated()` |
| POST | `/api/journal/admin/reconcile` | `AdminController` | `MSP_ADMIN`/`BCOS_ADMIN` |
| POST | `/api/journal/admin/parse-sync?fileId=` | `AdminController` | `MSP_ADMIN`/`BCOS_ADMIN` (diagnostics, synchronous) |

All reach journal through the transactions proxy (`JournalProxyController` → `JOURNAL_URL`, default `http://hiveiq-journal:8087`), which forwards only `Authorization` + `Content-Type`.
