# План работ: доведение `pcp-route-processing-service` / route-passport до production-ready состояния

Дата подготовки: 2026-04-28  
Источник анализа: архив `pcp.zip`, сервис `services/pcp-route-processing-service`, shared library `libs/pcp-types-lib`.

---

## 0. Цель работ

Нужно усилить текущую реализацию обработки паспорта маршрута так, чтобы сервис был готов к production-сценарию:

1. входной Kafka offset не должен коммититься до надежного сохранения результата обработки;
2. принятый маршрут должен сохраняться в БД вместе с геометрией и параметрами;
3. публикация downstream-событий должна идти через outbox, а не напрямую из Kafka listener;
4. ошибки должны быть классифицированы и обрабатываться через bounded retry + DLQ/DLT;
5. ручную GeoJSON → WKT конвертацию нужно заменить на библиотечную;
6. текущая бизнес-семантика `routeStatus = PROCESSED` остается без изменений;
7. flow остается Kafka-only, BPM/Camunda в рамках этой задачи не добавляется.

---

## 1. Текущая реализация, от которой отталкиваемся

### 1.1. Текущий поток

```text
pcp.request.survey-georeference.v1
        ↓
RoutePassportMessageListener.listen(message: String)
        ↓
RoutePassportJsonParserService.parse(message)
        ↓
ParsedRoutePassportMessage(RoutePassportDto + trace/correlation metadata)
        ↓
RoutePassportKafkaPublisherService.publish(parsedMessage)
        ├── pcp.route.in.v1           / ModeStatusChangedEvent
        └── pcp.route.georeference.v1 / RouteGeoRefEvent
```

### 1.2. Ключевые текущие классы

| Файл/класс | Текущая роль |
|---|---|
| `services/pcp-route-processing-service/src/main/kotlin/space/nstart/pcp/routepassportconsumer/listener/RoutePassportMessageListener.kt` | Kafka listener входного topic `pcp.request.survey-georeference.v1`. Сейчас парсит сообщение и сразу вызывает publisher. |
| `services/pcp-route-processing-service/src/main/kotlin/space/nstart/pcp/routepassportconsumer/service/RoutePassportJsonParserService.kt` | Парсит raw envelope в `RoutePassportDto`; вручную конвертирует GeoJSON geometry в WKT. |
| `services/pcp-route-processing-service/src/main/kotlin/space/nstart/pcp/routepassportconsumer/service/RoutePassportKafkaPublisherService.kt` | Публикует один и тот же `RoutePassportDto` в два output topic. Сейчас publish асинхронный и ошибка только логируется. |
| `services/pcp-route-processing-service/src/main/kotlin/space/nstart/pcp/routepassportconsumer/config/KafkaConsumerConfig.kt` | Явно создает `ConsumerFactory` и `ConcurrentKafkaListenerContainerFactory`. |
| `services/pcp-route-processing-service/src/main/kotlin/space/nstart/pcp/routepassportconsumer/config/KafkaProducerConfig.kt` | Явно создает `ProducerFactory` и `KafkaTemplate`. |
| `services/pcp-route-processing-service/src/main/kotlin/space/nstart/pcp/routepassportconsumer/config/KafkaTopicsProperties.kt` | Типизированные имена input/output topic. |
| `libs/pcp-types-lib/src/main/kotlin/space/nstart/pcp/pcp_types_lib/dto/routes/RoutePassportMessageDTO.kt` | Shared DTO: `RoutePassportDto`, `AngleRangeDto`. |
| `libs/pcp-types-lib/src/main/kotlin/space/nstart/pcp/pcp_types_lib/dto/routes/ParsedRoutePassportMessageDTO.kt` | Shared DTO: `ParsedRoutePassportMessage`, metadata. |
| `libs/pcp-types-lib/src/main/kotlin/space/nstart/pcp/pcp_types_lib/dto/routes/RawPassportJsonDTO.kt` | Raw DTO входного envelope/STAC-like паспорта. В рамках этой задачи не переносится. |
| `libs/pcp-types-lib/src/main/kotlin/space/nstart/pcp/pcp_types_lib/message/KafkaMessage.kt` | Общая Kafka envelope-модель. В рамках этой задачи не меняется. |
| `libs/pcp-types-lib/src/main/kotlin/space/nstart/pcp/pcp_types_lib/message/PcpKafkaEvent.kt` | Типы Kafka-событий, включая `RouteGeoRefEvent` и `ModeStatusChangedEvent`. |

### 1.3. Главные текущие проблемы

| Проблема | Почему это риск |
|---|---|
| `enable-auto-commit: true` в `config-repo/pcp-route-processing-service.yaml` | Offset может быть закоммичен до надежной обработки сообщения. |
| Listener сразу вызывает Kafka publisher | Если publish упал, входное сообщение уже могло быть считано/закоммичено. |
| Publisher отправляет два события асинхронно и только логирует ошибку | Возможна частичная публикация: один downstream topic получил событие, второй нет. |
| Нет БД и outbox | Невозможно надежно восстановить публикацию downstream-событий после сбоя Kafka/сервиса. |
| Нет явного DLQ/DLT и bounded retry | Bad input может зависнуть в бесконечных retry или обрабатываться неявным default behavior Spring Kafka. |
| Ручная GeoJSON → WKT конвертация | Риск некорректной геометрии, слабая валидация, ручная поддержка Polygon/MultiPolygon. |
| `routeStatus` всегда `PROCESSED` | Это остается осознанным текущим поведением. Не исправляем в этой задаче. |

---

## 2. Принятые решения по scope

### 2.1. Делаем

| № | Решение |
|---:|---|
| 1 | Перевести Kafka consumer на controlled commit: offset подтверждается только после успешного DB commit. |
| 2 | Настроить bounded retry + DLQ/DLT для входных сообщений. |
| 3 | Добавить БД для `pcp-route-processing-service`. |
| 4 | Сохранять принятый маршрут, геометрию, параметры и статус в БД. |
| 5 | Реализовать outbox через эту БД. |
| 6 | Убрать прямую downstream-публикацию из listener. |
| 7 | Публиковать `ModeStatusChangedEvent` и `RouteGeoRefEvent` из outbox worker. |
| 8 | Добавить producer reliability настройки: `acks=all`, retries, idempotence. |
| 9 | Добавить классификацию ошибок: `SUCCESS`, `RETRYABLE`, `BUSINESS/CONTRACT`, `FATAL`. |
| 10 | Заменить ручной GeoJSON → WKT converter на библиотечный подход. |
| 11 | Оставить flow Kafka-only. BPM/Camunda не добавлять. |
| 12 | Добавить минимальные тесты под измененную логику. |

### 2.2. Не делаем в этой задаче

| Пункт | Решение | Риск, который остается |
|---|---|---|
| Чтение `routeStatus` из входного сообщения | Не делаем. Оставляем `PROCESSED` как сейчас. | Если внешний контур начнет присылать `FAILED`, `CANCELED`, `DROPPED`, сервис это проигнорирует. |
| Замена Kafka key с `traceId` на `routeId`/`correlationId` | Не делаем. | Partitioning/ordering остаются привязанными к traceId, а не к business key. |
| Fallback `traceId`/`correlationId` | Не делаем. | Пустые trace/correlation значения могут пройти дальше. |
| README/fixtures | Не трогаем. | Документация и fixtures могут остаться несогласованными с кодом. |
| E2E serialization test producer → consumers | Не делаем отдельным пунктом. | Контракт producer/consumer не будет полностью проверен end-to-end. |
| `RouteProcessingTestController` | Не трогаем. | Тестовый endpoint остается в main-коде. |
| `OpenApiConfig`/springdoc | Не трогаем. | Лишняя REST/OpenAPI инфраструктура остается. |
| Перенос raw DTO из shared lib | Не делаем. | Shared lib продолжает содержать внутренний raw input contract route-processing-service. |
| Упрощение `KafkaEvent`/ClassGraph | Не делаем. | Существующая абстракция остается как есть. |
| Упрощение Kafka config через `KafkaProperties` | Не делаем. | Но точечно добавляем нужные reliability properties в текущий custom config. |
| Изменение `KafkaMessage` | Не делаем. | `source = "Example"` и mutable поле остаются в shared contract. |
| Полные Kafka transactions | Не делаем. | Достигаем надежности через DB outbox и at-least-once. |
| Camunda/BPM flow | Не делаем. | Сервис остается Kafka-only. |

---

## 3. Целевая архитектура

### 3.1. Целевой поток обработки input message

```text
pcp.request.survey-georeference.v1
        ↓
RoutePassportMessageListener
        ↓
RoutePassportJsonParserService
        ↓
RouteGeometryConverter
        ↓
RoutePassportIngestionService @Transactional
        ├── UPSERT route_passports
        └── INSERT/UPSERT route_passport_outbox x2
                ├── ModeStatusChangedEvent -> pcp.route.in.v1
                └── RouteGeoRefEvent       -> pcp.route.georeference.v1
        ↓
Kafka offset ack/commit only after DB commit
```

### 3.2. Целевой поток публикации outbox

```text
route_passport_outbox PENDING rows
        ↓
RoutePassportOutboxWorker @Scheduled
        ↓
RoutePassportOutboxService.claimDueBatch()
        ↓
RoutePassportKafkaPublisherService.publish(outboxRow)
        ↓
Kafka output topic
        ↓
mark PUBLISHED or schedule retry / mark FAILED
```

### 3.3. Целевой flow ошибок

```text
Invalid JSON / invalid contract / unsupported geometry
        ↓
RoutePassportContractException
        ↓
DefaultErrorHandler non-retryable
        ↓
DLQ/DLT topic

Temporary DB error during ingest
        ↓
Retry with bounded backoff
        ↓
Success OR DLQ/DLT after max attempts

Kafka output publish error in outbox worker
        ↓
outbox row remains retryable
        ↓
next_attempt_at + attempts
        ↓
PUBLISHED OR FAILED after max attempts
```

---

## 4. Database naming and ownership

### 4.1. Рекомендуемое имя БД

Использовать БД:

```text
pcp_route_processing
```

Причина: БД принадлежит сервису `pcp-route-processing-service`, а не отдельному DTO. Это соответствует подходу service-owned database.

Человеческое описание:

```text
Route Processing DB — БД хранения принятых route passport и outbox событий route-processing-service.
```

### 4.2. Не использовать

```text
pcp-route-passport-bd
```

Причины:

1. дефисы неудобны для PostgreSQL database identifiers;
2. `bd` лучше не использовать, стандартнее `db`, но в имени БД суффикс вообще не нужен;
3. БД должна называться по сервису/домейну, а не по техническому слову “passport DTO”.

---

## 5. Новая структура пакетов

Текущий base package:

```text
space.nstart.pcp.routepassportconsumer
```

Предлагаемая структура:

```text
space.nstart.pcp.routepassportconsumer
├── config
│   ├── KafkaConsumerConfig.kt
│   ├── KafkaProducerConfig.kt
│   ├── KafkaTopicsProperties.kt
│   ├── RoutePassportOutboxProperties.kt
│   └── RoutePassportStorageProperties.kt        # optional, если нужны storage-specific настройки
├── listener
│   └── RoutePassportMessageListener.kt
├── service
│   ├── RoutePassportJsonParserService.kt
│   ├── RouteGeometryConverter.kt
│   ├── RoutePassportIngestionService.kt
│   ├── RoutePassportKafkaPublisherService.kt
│   ├── RoutePassportOutboxService.kt
│   └── RoutePassportOutboxWorker.kt
├── entity
│   ├── RoutePassportEntity.kt
│   └── RoutePassportOutboxEntity.kt
├── repository
│   ├── RoutePassportRepository.kt
│   └── RoutePassportOutboxRepository.kt
├── exception
│   ├── RoutePassportContractException.kt
│   ├── RoutePassportRetryableException.kt        # optional
│   └── RoutePassportFatalException.kt            # optional
└── dto
    └── RoutePassportOutboxPayload.kt             # optional, если нужен внутренний DTO outbox payload
```

Правило: новые классы добавлять только если они реально отделяют ответственность. Не создавать лишние interfaces/factories/strategies.

---

# 6. Детальный план реализации

---

## Шаг 1. Подключить БД, JPA и Flyway к `pcp-route-processing-service`

### 1.1. Шаг реализации

Добавить поддержку PostgreSQL/JPA/Flyway в `pcp-route-processing-service` по аналогии с `pcp-request-service`.

### 1.2. Вход шага

Текущий `pcp-route-processing-service` без БД.

Файлы:

```text
services/pcp-route-processing-service/build.gradle.kts
services/pcp-route-processing-service/src/main/resources/application.yml
config-repo/pcp-route-processing-service.yaml
```

### 1.3. Выход шага

Сервис умеет стартовать с datasource, JPA validation и Flyway migrations.

### 1.4. Какие существующие классы участвуют

На этом шаге классы бизнес-логики не меняются.

### 1.5. Нужно ли вводить новую модель

Нет.

### 1.6. Слой

`adapter / infrastructure`

### 1.7. Изменения в `build.gradle.kts`

Добавить plugin:

```kotlin
kotlin("plugin.jpa")
```

Добавить dependencies:

```kotlin
implementation("org.springframework.boot:spring-boot-starter-data-jpa")
implementation("org.springframework.boot:spring-boot-starter-flyway")
implementation("org.flywaydb:flyway-database-postgresql")
implementation("org.locationtech.jts:jts-core:1.19.0")
runtimeOnly("org.postgresql:postgresql")
```

`org.locationtech.jts:jts-core:1.19.0` уже используется в `pcp-request-service`, поэтому это не новая для проекта библиотека.

### 1.8. Изменения в `config-repo/pcp-route-processing-service.yaml`

Добавить:

```yaml
spring:
  datasource:
    driver-class-name: org.postgresql.Driver
    url: jdbc:postgresql://${pcp.infra.postgres.host}:${pcp.infra.postgres.port}/pcp_route_processing
    username: postgres
    password: password
  jpa:
    hibernate:
      ddl-auto: validate
    properties:
      hibernate:
        format_sql: true
  flyway:
    enabled: true
    baseline-on-migrate: true
    locations: classpath:db/migration
```

### 1.9. Acceptance criteria

1. Gradle module компилируется с JPA/Flyway dependencies.
2. При старте сервис ищет migrations в `classpath:db/migration`.
3. Hibernate не создает schema сам, а валидирует ее через `ddl-auto: validate`.

---

## Шаг 2. Создать таблицу `route_passports`

### 2.1. Шаг реализации

Добавить Flyway migration:

```text
services/pcp-route-processing-service/src/main/resources/db/migration/V1__create_route_passports.sql
```

### 2.2. Вход шага

`RoutePassportDto`:

```text
libs/pcp-types-lib/src/main/kotlin/space/nstart/pcp/pcp_types_lib/dto/routes/RoutePassportMessageDTO.kt
```

### 2.3. Выход шага

Таблица `route_passports` хранит нормализованный паспорт маршрута, геометрию, параметры, metadata входного Kafka record и raw payload для аудита.

### 2.4. Какие существующие классы участвуют

| Класс | Как участвует |
|---|---|
| `RoutePassportDto` | Источник полей для таблицы. |
| `AngleRangeDto` | Источник `roll_angle_min/max`, `visir_angle_min/max`. |
| `SurveyModeStatus` | Источник `route_status`; пока всегда `PROCESSED`. |
| `ParsedRoutePassportMessage` | Источник trace/correlation metadata. |

### 2.5. Нужно ли вводить новую модель

Да:

```text
RoutePassportEntity
RoutePassportRepository
```

### 2.6. Слой

`adapter / storage`

### 2.7. Рекомендуемый SQL

```sql
CREATE TABLE IF NOT EXISTS route_passports (
    route_id UUID PRIMARY KEY,

    route_name_full TEXT NOT NULL,
    route_name_short VARCHAR(255) NOT NULL,

    ka_short VARCHAR(64) NOT NULL,
    ka_full VARCHAR(255) NOT NULL,

    orbit_number BIGINT NOT NULL,
    orbit_state VARCHAR(64) NOT NULL,

    interval_begin TIMESTAMP NOT NULL,
    interval_end TIMESTAMP NOT NULL,

    roll_angle_min NUMERIC NOT NULL,
    roll_angle_max NUMERIC NOT NULL,
    visir_angle_min NUMERIC NOT NULL,
    visir_angle_max NUMERIC NOT NULL,

    resolution_range NUMERIC NOT NULL,
    resolution_azimuth NUMERIC NOT NULL,

    polarisation JSONB NOT NULL,
    processing_level VARCHAR(64) NOT NULL,

    geometry_wkt TEXT NOT NULL,
    route_status VARCHAR(32) NOT NULL,

    trace_id VARCHAR(255) NULL,
    correlation_id VARCHAR(255) NULL,

    source_topic VARCHAR(255) NULL,
    source_partition INTEGER NULL,
    source_offset BIGINT NULL,
    source_key VARCHAR(255) NULL,

    raw_payload JSONB NULL,

    created_at TIMESTAMP NOT NULL,
    updated_at TIMESTAMP NOT NULL,
    version BIGINT NOT NULL DEFAULT 0
);

CREATE INDEX IF NOT EXISTS idx_route_passports_interval
    ON route_passports (interval_begin, interval_end);

CREATE INDEX IF NOT EXISTS idx_route_passports_ka_orbit
    ON route_passports (ka_short, orbit_number);

CREATE INDEX IF NOT EXISTS idx_route_passports_status
    ON route_passports (route_status);

CREATE UNIQUE INDEX IF NOT EXISTS uk_route_passports_source_record
    ON route_passports (source_topic, source_partition, source_offset)
    WHERE source_topic IS NOT NULL
      AND source_partition IS NOT NULL
      AND source_offset IS NOT NULL;
```

### 2.8. Почему `geometry_wkt TEXT`, а не PostGIS geometry

Сейчас downstream `pcp-request-service` получает geometry как WKT string и уже сам выполняет route matching. В `pcp-route-processing-service` пока не планируются spatial queries. Поэтому PostGIS добавлять не нужно.

Если позже появятся запросы вида “найти маршруты, пересекающие область”, тогда можно будет добавить отдельную migration с PostGIS geometry column.

### 2.9. Почему `polarisation JSONB`

`polarisation` — список строк. Хранить как JSONB проще, чем вводить отдельную таблицу, потому что по нему пока не видно сложных запросов/джоинов.

В entity можно хранить это поле как `String` с `@Column(columnDefinition = "jsonb")`, чтобы не тащить дополнительные Hibernate JSON types.

### 2.10. Acceptance criteria

1. Таблица хранит все поля `RoutePassportDto`.
2. `route_id` является primary key и idempotency key.
3. Есть исходная Kafka metadata для аудита.
4. Есть `raw_payload`, чтобы можно было восстановить, что реально пришло во входной topic.

---

## Шаг 3. Создать таблицу `route_passport_outbox`

### 3.1. Шаг реализации

Добавить Flyway migration:

```text
services/pcp-route-processing-service/src/main/resources/db/migration/V2__create_route_passport_outbox.sql
```

### 3.2. Вход шага

Текущие downstream-публикации из `RoutePassportKafkaPublisherService`:

```text
pcp.route.in.v1           / PcpKafkaEvent.ModeStatusChangedEvent
pcp.route.georeference.v1 / PcpKafkaEvent.RouteGeoRefEvent
```

### 3.3. Выход шага

Таблица outbox хранит pending Kafka-события для последующей надежной публикации.

### 3.4. Какие существующие классы участвуют

| Класс | Как участвует |
|---|---|
| `RoutePassportKafkaPublisherService` | Будет использоваться outbox worker, а не listener. |
| `KafkaTopicsProperties` | Источник имен output topics. |
| `PcpKafkaEvent` | Тип события в outbox. |
| `KafkaMessage` | Envelope payload, сохраняемый в outbox. |
| `RoutePassportDto` | Data внутри KafkaMessage. |

### 3.5. Нужно ли вводить новую модель

Да:

```text
RoutePassportOutboxEntity
RoutePassportOutboxRepository
RoutePassportOutboxStatus
RoutePassportOutboxService
RoutePassportOutboxWorker
```

### 3.6. Слой

`adapter / storage`, `orchestration`, `adapter / messaging`

### 3.7. Рекомендуемый SQL

```sql
CREATE TABLE IF NOT EXISTS route_passport_outbox (
    id UUID PRIMARY KEY,
    route_id UUID NOT NULL,

    event_type VARCHAR(128) NOT NULL,
    topic VARCHAR(255) NOT NULL,
    message_key VARCHAR(255) NOT NULL,

    payload JSONB NOT NULL,
    headers JSONB NULL,

    status VARCHAR(32) NOT NULL,
    attempts INTEGER NOT NULL DEFAULT 0,
    max_attempts INTEGER NOT NULL DEFAULT 10,

    next_attempt_at TIMESTAMP NOT NULL,
    locked_until TIMESTAMP NULL,

    last_error TEXT NULL,

    created_at TIMESTAMP NOT NULL,
    updated_at TIMESTAMP NOT NULL,
    published_at TIMESTAMP NULL,

    CONSTRAINT fk_route_passport_outbox_route
        FOREIGN KEY (route_id)
        REFERENCES route_passports(route_id)
        ON DELETE CASCADE,

    CONSTRAINT uk_route_passport_outbox_event
        UNIQUE (route_id, event_type)
);

CREATE INDEX IF NOT EXISTS idx_route_passport_outbox_pending
    ON route_passport_outbox (status, next_attempt_at, created_at);

CREATE INDEX IF NOT EXISTS idx_route_passport_outbox_locked
    ON route_passport_outbox (locked_until);
```

### 3.8. Outbox statuses

```kotlin
enum class RoutePassportOutboxStatus {
    PENDING,
    PUBLISHED,
    FAILED
}
```

Не вводить `PUBLISHING` на первом этапе. Для защиты от конкурентных publisher’ов использовать `locked_until`.

### 3.9. Почему `UNIQUE(route_id, event_type)`

Это нужно для идемпотентности.

Сценарий:

```text
1. Listener сохранил route_passports + route_passport_outbox.
2. DB transaction успешно закоммитилась.
3. Сервис упал до Kafka offset ack.
4. Kafka отдаст тот же input message повторно.
5. Повторная обработка не должна создать дубль outbox-событий.
```

`route_id + event_type` защищает от повторного создания тех же downstream-событий.

Ограничение: если в будущем один и тот же `route_id` может прийти как новая версия маршрута, понадобится добавить revision/version/source_event_id в idempotency key. В текущем контракте явной версии маршрута нет, поэтому `route_id` считаем business idempotency key.

### 3.10. Acceptance criteria

1. На один маршрут создаются максимум два outbox-события: `ModeStatusChangedEvent`, `RouteGeoRefEvent`.
2. Повторная обработка того же routeId не плодит дубли outbox rows.
3. Outbox rows можно безопасно читать несколькими instance’ами сервиса.

---

## Шаг 4. Добавить JPA entity и repositories

### 4.1. Шаг реализации

Добавить storage-модели и repositories.

### 4.2. Вход шага

Миграции из шагов 2–3.

### 4.3. Выход шага

Код может сохранять маршрут и outbox-события в одной DB transaction.

### 4.4. Какие существующие классы участвуют

Прямо существующие storage-классы в route-processing-service отсутствуют.

### 4.5. Нужно ли вводить новую модель

Да:

```text
services/pcp-route-processing-service/src/main/kotlin/space/nstart/pcp/routepassportconsumer/entity/RoutePassportEntity.kt
services/pcp-route-processing-service/src/main/kotlin/space/nstart/pcp/routepassportconsumer/entity/RoutePassportOutboxEntity.kt
services/pcp-route-processing-service/src/main/kotlin/space/nstart/pcp/routepassportconsumer/repository/RoutePassportRepository.kt
services/pcp-route-processing-service/src/main/kotlin/space/nstart/pcp/routepassportconsumer/repository/RoutePassportOutboxRepository.kt
```

### 4.6. Слой

`adapter / storage`

### 4.7. `RoutePassportEntity`

Поля entity должны 1-в-1 соответствовать таблице `route_passports`.

Важные правила:

1. `routeId` — `@Id`.
2. `version` — `@Version`, если нужна optimistic locking.
3. `createdAt` и `updatedAt` заполнять в коде, без магии БД.
4. `polarisation` и `rawPayload` можно хранить как JSON string с `columnDefinition = "jsonb"`.
5. `geometryWkt` хранить как `TEXT`.

### 4.8. `RoutePassportOutboxEntity`

Поля entity должны соответствовать таблице `route_passport_outbox`.

Важные правила:

1. `id` — generated UUID в коде.
2. `routeId` можно хранить как UUID без JPA relation, чтобы не усложнять entity graph.
3. `payload` хранить как JSON string с `columnDefinition = "jsonb"`.
4. `headers` хранить как JSON string или nullable JSON string.
5. `status` — enum string.

Не нужно делать сложные bidirectional JPA relations. Они не нужны для этого flow.

### 4.9. Repository methods

`RoutePassportRepository`:

```kotlin
interface RoutePassportRepository : JpaRepository<RoutePassportEntity, UUID>
```

`RoutePassportOutboxRepository`:

```kotlin
interface RoutePassportOutboxRepository : JpaRepository<RoutePassportOutboxEntity, UUID> {
    // native query with FOR UPDATE SKIP LOCKED for claim operation
}
```

Для outbox лучше сделать не один “find pending”, а сервисный метод `claimDueBatch`, который в transaction:

1. выбирает due rows;
2. ставит `locked_until`;
3. возвращает claimed rows.

### 4.10. Acceptance criteria

1. Entity валидируются Hibernate against Flyway schema.
2. Нет лишних JPA relations, которые усложняют загрузку.
3. Есть repository method для claim pending outbox rows.

---

## Шаг 5. Добавить `RouteGeometryConverter` и заменить ручную GeoJSON → WKT конвертацию

### 5.1. Шаг реализации

Вынести конвертацию geometry из `RoutePassportJsonParserService` в отдельный сервис и использовать библиотечную конвертацию.

### 5.2. Вход шага

Текущий код в `RoutePassportJsonParserService`:

```kotlin
private fun RawGeoJsonGeometryDto.toWkt(): String
private fun tools.jackson.databind.JsonNode.toWktCoordinates(depth: Int): String
```

### 5.3. Выход шага

Ручная рекурсивная сборка WKT удалена. Parser делегирует конвертацию в `RouteGeometryConverter`.

### 5.4. Какие существующие классы участвуют

| Класс | Как участвует |
|---|---|
| `RoutePassportJsonParserService` | Перестает собирать WKT вручную. |
| `RawGeoJsonGeometryDto` | Вход converter’а. |
| `RoutePassportDto` | Получает готовый `geometry: String`. |

### 5.5. Нужно ли вводить новую модель

Да:

```text
RouteGeometryConverter
```

Отдельные DTO не нужны.

### 5.6. Слой

`algorithm`

### 5.7. Предлагаемая ответственность converter

```text
RouteGeometryConverter
├── принимает RawGeoJsonGeometryDto
├── проверяет supported geometry type: Polygon / MultiPolygon
├── преобразует raw geometry в GeoJSON string/object
├── читает geometry через JTS-compatible reader
├── пишет WKT через WKT writer
└── выбрасывает RoutePassportContractException при invalid geometry
```

### 5.8. Что считать ошибкой контракта

| Ситуация | Исключение | Retry? |
|---|---|---|
| `geometry.type` не `Polygon` и не `MultiPolygon` | `RoutePassportContractException` | Нет |
| coordinates отсутствуют или не массив | `RoutePassportContractException` | Нет |
| geometry не парсится библиотекой | `RoutePassportContractException` | Нет |
| geometry пустая | `RoutePassportContractException` | Нет |

### 5.9. Почему отдельный converter — не лишняя прослойка

Это не абстракция “на будущее”. Сейчас parser смешивает две разные ответственности:

1. разбор route passport fields;
2. геометрический алгоритм GeoJSON → WKT.

После замены ручной конвертации на библиотеку converter будет изолировать геометрическую логику и упростит тестирование.

### 5.10. Acceptance criteria

1. В `RoutePassportJsonParserService` нет ручного `toWktCoordinates`.
2. `Polygon` и `MultiPolygon` продолжают поддерживаться.
3. Ошибки geometry классифицируются как non-retryable contract errors.
4. Есть unit tests на converter.

---

## Шаг 6. Добавить exception taxonomy

### 6.1. Шаг реализации

Ввести явные exception-классы для классификации ошибок.

### 6.2. Вход шага

Сейчас parser использует `require(...)` и `error(...)`, что приводит к generic `IllegalArgumentException`/`IllegalStateException`.

### 6.3. Выход шага

Kafka error handler может различать retryable и non-retryable ошибки.

### 6.4. Какие существующие классы участвуют

| Класс | Как участвует |
|---|---|
| `RoutePassportJsonParserService` | Бросает contract exceptions вместо generic error. |
| `RouteGeometryConverter` | Бросает contract exceptions при invalid geometry. |
| `KafkaConsumerConfig` | Регистрирует non-retryable exceptions. |
| `RoutePassportMessageListener` | Не проглатывает ошибки. |

### 6.5. Нужно ли вводить новую модель

Да:

```kotlin
class RoutePassportContractException(message: String, cause: Throwable? = null) : RuntimeException(message, cause)
class RoutePassportProcessingException(message: String, cause: Throwable? = null) : RuntimeException(message, cause)
```

`RoutePassportRetryableException` можно не вводить, если retryable определяются по типам инфраструктурных исключений.

### 6.6. Слой

`domain / algorithm / adapter`

### 6.7. Error classification

| Категория | Примеры | Поведение |
|---|---|---|
| `SUCCESS` | parse ok, route saved, outbox created, DB commit ok | Kafka offset ack |
| `BUSINESS/CONTRACT` | invalid JSON shape, missing required field, invalid angle, unsupported geometry | no retry или минимальный retry, потом DLQ/DLT |
| `RETRYABLE` | transient DB error, temporary DB connection issue | bounded retry, потом DLQ/DLT |
| `FATAL` | unexpected bug, incompatible serialization, unknown runtime failure | bounded retry, потом DLQ/DLT + alert |

### 6.8. Acceptance criteria

1. Parser/converter не используют `error(...)` для contract validation.
2. Bad input предсказуемо классифицируется как non-retryable.
3. Infrastructure failures остаются retryable.

---

## Шаг 7. Добавить `RoutePassportIngestionService`

### 7.1. Шаг реализации

Ввести application service, который в одной DB transaction:

1. сохраняет/обновляет `route_passports`;
2. создает две outbox-записи;
3. не публикует Kafka напрямую.

### 7.2. Вход шага

`ParsedRoutePassportMessage` + исходная Kafka metadata:

```text
message value
record key
topic
partition
offset
```

### 7.3. Выход шага

DB transaction committed:

```text
route_passports row
route_passport_outbox row: ModeStatusChangedEvent
route_passport_outbox row: RouteGeoRefEvent
```

### 7.4. Какие существующие классы участвуют

| Класс | Как участвует |
|---|---|
| `RoutePassportMessageListener` | Будет вызывать ingestion вместо publisher. |
| `RoutePassportJsonParserService` | Возвращает `ParsedRoutePassportMessage`. |
| `RoutePassportKafkaPublisherService` | Убирается из listener path. |
| `KafkaTopicsProperties` | Нужен для создания outbox rows. |
| `KafkaMessage` | Payload сохраняется в outbox. |
| `PcpKafkaEvent` | Типы outbox events. |

### 7.5. Нужно ли вводить новую модель

Да:

```text
RoutePassportIngestionService
RoutePassportSourceRecordMetadata
```

`RoutePassportSourceRecordMetadata` может быть простой internal data class:

```kotlin
data class RoutePassportSourceRecordMetadata(
    val topic: String,
    val partition: Int,
    val offset: Long,
    val key: String?,
)
```

### 7.6. Слой

`orchestration / application`

### 7.7. Транзакционная логика

Псевдологика:

```text
@Transactional
fun ingest(parsedMessage, sourceRecord, rawPayload) {
    val route = parsedMessage.routePassportDto

    upsert route_passports by route.routeId

    create outbox event if absent:
      routeId + ModeStatusChangedEvent

    create outbox event if absent:
      routeId + RouteGeoRefEvent
}
```

### 7.8. Upsert route passport

Варианты:

#### Вариант A — простой JPA save

```text
if route exists:
    update fields
else:
    insert fields
```

Плюс: понятно и просто.  
Минус: два запроса.

#### Вариант B — native upsert

```sql
INSERT ... ON CONFLICT(route_id) DO UPDATE ...
```

Плюс: атомарнее.  
Минус: больше SQL, сложнее тестировать.

Рекомендация: начать с Варианта A. Он достаточно надежен при текущей нагрузке и проще для команды.

### 7.9. Outbox payload generation

Сохранять в outbox уже готовый JSON payload, а не DTO.

Причина: outbox row должен быть самодостаточным. Если код DTO изменится после сохранения outbox row, старое pending event все равно должно быть опубликовано как было сформировано в момент ingest.

Payload должен повторять текущий формат `RoutePassportKafkaPublisherService.buildPayload(...)`:

```kotlin
KafkaMessage(
    type = eventType,
    data = parsedMessage.routePassportDto,
    traceId = parsedMessage.metadata.traceId,
    correlationId = parsedMessage.metadata.correlationId,
    group = "pcp",
).apply {
    source = "route-processing"
}
```

### 7.10. Kafka message key

Пункт смены key пропущен, поэтому сохраняем текущую семантику:

```text
message_key = parsedMessage.metadata.traceId
```

Это не лучший business key, но это осознанно вне scope.

### 7.11. Acceptance criteria

1. Listener больше не публикует Kafka напрямую.
2. При успешной обработке input message в БД появляется route row и две outbox rows.
3. Повторная обработка того же routeId не создает дублей outbox rows.
4. `route_status` сохраняется как `PROCESSED`, как и сейчас.

---

## Шаг 8. Переделать `RoutePassportMessageListener` на controlled commit

### 8.1. Шаг реализации

Переписать listener так, чтобы он:

1. принимал `ConsumerRecord<String, String>`;
2. парсил value;
3. вызывал `RoutePassportIngestionService`;
4. подтверждал offset только после успешного завершения ingest;
5. не вызывал `RoutePassportKafkaPublisherService` напрямую.

### 8.2. Вход шага

Текущий listener:

```kotlin
fun listen(message: String)
```

### 8.3. Выход шага

Новый listener:

```kotlin
fun listen(record: ConsumerRecord<String, String>, acknowledgment: Acknowledgment)
```

### 8.4. Какие существующие классы участвуют

| Класс | Изменение |
|---|---|
| `RoutePassportMessageListener` | Основное изменение. |
| `RoutePassportJsonParserService` | Остается. |
| `RoutePassportKafkaPublisherService` | Убирается из constructor injection listener’а. |
| `RoutePassportIngestionService` | Добавляется в constructor injection listener’а. |
| `KafkaTopicsProperties` | Остается. |

### 8.5. Нужно ли вводить новую модель

Да, если используется `RoutePassportSourceRecordMetadata`.

### 8.6. Слой

`adapter / messaging`, `orchestration`

### 8.7. Listener behavior

Псевдологика:

```text
listen(record, ack):
    log received topic/partition/offset/payloadSize

    parsedMessage = parser.parse(record.value())

    ingestionService.ingest(
        parsedMessage = parsedMessage,
        rawPayload = record.value(),
        sourceRecord = metadata(record),
    )

    ack.acknowledge()
```

Если parser/ingestion бросили exception:

```text
exception bubbles to Spring Kafka DefaultErrorHandler
ack is not called
handler applies retry/DLQ policy
```

### 8.8. Acceptance criteria

1. В listener нет `try/catch`, который ломает error handler, либо catch только логирует и rethrow.
2. `ack.acknowledge()` вызывается строго после успешного `ingestionService.ingest(...)`.
3. Offset не подтверждается при parse/DB ошибках.

---

## Шаг 9. Настроить Kafka consumer на manual ack и disable auto commit

### 9.1. Шаг реализации

Изменить Kafka consumer config.

### 9.2. Вход шага

Текущий config:

```yaml
spring:
  kafka:
    consumer:
      enable-auto-commit: true
```

Текущий `KafkaConsumerConfig` читает `enable-auto-commit`, но container factory не задает ack mode и error handler.

### 9.3. Выход шага

Consumer не коммитит offset автоматически. Ack происходит вручную после DB commit.

### 9.4. Какие существующие классы участвуют

| Класс/файл | Изменение |
|---|---|
| `config-repo/pcp-route-processing-service.yaml` | `enable-auto-commit: false`, `listener.ack-mode`. |
| `KafkaConsumerConfig.kt` | Установить `AckMode.MANUAL_IMMEDIATE`, error handler. |

### 9.5. Нужно ли вводить новую модель

Нет.

### 9.6. Слой

`adapter / infrastructure`

### 9.7. YAML

```yaml
spring:
  kafka:
    consumer:
      enable-auto-commit: false
    listener:
      ack-mode: manual_immediate
```

### 9.8. Kotlin config

В `KafkaConsumerConfig.kafkaListenerContainerFactory()` добавить:

```kotlin
containerProperties.ackMode = ContainerProperties.AckMode.MANUAL_IMMEDIATE
setCommonErrorHandler(defaultErrorHandler)
```

### 9.9. Acceptance criteria

1. Offset commit невозможен до явного ack.
2. При exception ack не вызывается.
3. Error handler получает ошибку и применяет retry/DLQ policy.

---

## Шаг 10. Настроить bounded retry + DLQ/DLT для input topic

### 10.1. Шаг реализации

Добавить `DefaultErrorHandler` и `DeadLetterPublishingRecoverer` в `KafkaConsumerConfig`.

### 10.2. Вход шага

Ошибки из:

```text
RoutePassportJsonParserService
RouteGeometryConverter
RoutePassportIngestionService
JPA/database layer
```

### 10.3. Выход шага

Ошибочные input messages не теряются и не создают бесконечный retry.

### 10.4. Какие существующие классы участвуют

| Класс | Изменение |
|---|---|
| `KafkaConsumerConfig` | Добавить beans для error handler/recoverer. |
| `KafkaProducerConfig` | KafkaTemplate используется recoverer’ом для DLT publish. |
| `RoutePassportContractException` | Non-retryable exception. |
| `RoutePassportMessageListener` | Не проглатывает исключения. |

### 10.5. Нужно ли вводить новую модель

Да:

```text
RoutePassportContractException
```

И properties:

```text
RoutePassportDlqProperties / RoutePassportConsumerErrorHandlingProperties
```

Можно обойтись `@Value`, но typed properties лучше для читаемости.

### 10.6. Слой

`adapter / infrastructure`

### 10.7. DLQ topic naming

Не использовать topic вида:

```text
pcp.request.survey-georeference.v1.dlt
```

Он нарушает правило, что major version должен быть в конце.

Рекомендуемый DLQ/DLT topic:

```text
pcp.request.survey-georeference-failed.v1
```

или, если команда предпочитает явно `dlq`:

```text
pcp.request.survey-georeference-dlq.v1
```

В плане используем:

```text
pcp.request.survey-georeference-dlq.v1
```

Добавить в config:

```yaml
app:
  kafka:
    topics:
      input: pcp.request.survey-georeference.v1
      processed-route: pcp.route.in.v1
      route-georeference: pcp.route.georeference.v1
      input-dlq: pcp.request.survey-georeference-dlq.v1
```

Соответственно расширить `KafkaTopicsProperties`:

```kotlin
val inputDlq: String = ""
```

### 10.8. Retry policy

Рекомендуемый стартовый вариант:

```yaml
app:
  route-passport:
    consumer:
      retry:
        max-attempts: 3
        backoff-ms: 1000
        max-backoff-ms: 10000
```

Поведение:

| Exception | Retry | Destination |
|---|---:|---|
| `RoutePassportContractException` | 0 или 1 | DLQ сразу |
| JSON parse/mapping exception | 0 или 1 | DLQ сразу |
| Invalid geometry | 0 или 1 | DLQ сразу |
| DB connection/transient exceptions | bounded retry | DLQ after attempts |
| Unexpected runtime exceptions | bounded retry | DLQ after attempts |

### 10.9. DLT payload/headers

`DeadLetterPublishingRecoverer` должен сохранять:

1. original payload;
2. original topic;
3. original partition;
4. original offset;
5. exception class;
6. exception message;
7. service name;
8. timestamp.

Spring Kafka добавляет часть headers сам. Если стандартных headers недостаточно, добавить custom headers в recoverer.

### 10.10. Observability

Логировать:

```text
routeId if parsed
topic/partition/offset
traceId
correlationId
exception category
DLQ topic
```

Не логировать полный payload. Только preview/размер.

### 10.11. Acceptance criteria

1. Bad contract message уходит в DLQ и не блокирует consumer group бесконечно.
2. Retry ограничен по количеству попыток.
3. DLQ topic имеет version suffix `.v1`.
4. В logs понятно, почему сообщение ушло в DLQ.

---

## Шаг 11. Настроить producer reliability

### 11.1. Шаг реализации

Добавить надежные настройки Kafka producer.

### 11.2. Вход шага

Текущий `KafkaProducerConfig` задает только:

```text
bootstrap.servers
key.serializer
value.serializer
```

Даже если добавить настройки в YAML, текущий custom config их не подхватит автоматически.

### 11.3. Выход шага

Kafka producer получает reliability properties явно.

### 11.4. Какие существующие классы участвуют

| Класс/файл | Изменение |
|---|---|
| `KafkaProducerConfig.kt` | Добавить producer properties в map. |
| `config-repo/pcp-route-processing-service.yaml` | Добавить `acks`, `retries`, `enable.idempotence`. |

### 11.5. Нужно ли вводить новую модель

Нет.

### 11.6. Слой

`adapter / infrastructure`

### 11.7. YAML

```yaml
spring:
  kafka:
    producer:
      acks: all
      retries: 10
      properties:
        enable.idempotence: true
        max.in.flight.requests.per.connection: 5
```

### 11.8. Kotlin config

Не делаем полный refactor на `KafkaProperties`, потому что это вне scope. Но текущую map нужно расширить:

```kotlin
ProducerConfig.ACKS_CONFIG to acks
ProducerConfig.RETRIES_CONFIG to retries
ProducerConfig.ENABLE_IDEMPOTENCE_CONFIG to enableIdempotence
ProducerConfig.MAX_IN_FLIGHT_REQUESTS_PER_CONNECTION to maxInFlightRequests
```

### 11.9. Что это дает

| Настройка | Что дает |
|---|---|
| `acks=all` | Broker подтверждает запись только после записи leader + in-sync replicas. |
| `retries` | Producer повторяет отправку при временных ошибках. |
| `enable.idempotence=true` | Защита от дублей, возникающих из-за producer retries. |
| `max.in.flight.requests.per.connection=5` | Совместимо с idempotent producer. |

### 11.10. Чего это не дает

Producer reliability не заменяет outbox.

Outbox нужен, чтобы событие не потерялось, если:

1. Kafka недоступна;
2. сервис упал;
3. publish прошел частично;
4. producer не успел отправить после обработки input.

### 11.11. Acceptance criteria

1. Outbox worker использует producer с `acks=all` и idempotence.
2. Ошибка producer publish не теряет событие, а переводит outbox row в retry/failed state.

---

## Шаг 12. Переделать `RoutePassportKafkaPublisherService` под outbox

### 12.1. Шаг реализации

Разделить две ответственности:

1. формирование outbox payload — в `RoutePassportIngestionService` или отдельном small helper;
2. физическая отправка Kafka record — в `RoutePassportKafkaPublisherService`.

### 12.2. Вход шага

Текущий `RoutePassportKafkaPublisherService.publish(parsedMessage)`.

### 12.3. Выход шага

Новый publisher публикует конкретную outbox row:

```text
RoutePassportKafkaPublisherService.publish(outboxEvent)
```

И возвращает результат/бросает exception.

### 12.4. Какие существующие классы участвуют

| Класс | Изменение |
|---|---|
| `RoutePassportKafkaPublisherService` | Больше не строит два события из `ParsedRoutePassportMessage`. Публикует одну outbox row. |
| `RoutePassportIngestionService` | Создает две outbox rows. |
| `RoutePassportOutboxWorker` | Вызывает publisher. |

### 12.5. Нужно ли вводить новую модель

Можно использовать `RoutePassportOutboxEntity` как вход publisher’а.

Если не хочется давать service слой доступ к JPA entity, можно добавить internal DTO:

```kotlin
data class RoutePassportOutboxMessage(
    val id: UUID,
    val topic: String,
    val messageKey: String,
    val eventType: String,
    val payload: String,
    val headers: Map<String, String>,
)
```

Но для первого варианта допустимо использовать entity, чтобы не плодить лишние классы.

### 12.6. Слой

`adapter / messaging`

### 12.7. Publish behavior

Важно: publisher должен дождаться результата отправки.

Псевдологика:

```text
record = ProducerRecord(topic, messageKey, payload)
record.headers.add("type", eventType)
future = kafkaTemplate.send(record)
future.get(timeout)
```

Если publish упал — бросить exception. Не просто логировать.

### 12.8. Acceptance criteria

1. Ошибка Kafka publish видна outbox worker’у как exception.
2. Publisher не знает ничего про два downstream topic. Он публикует одну outbox row.
3. Заголовок `type` сохраняется как сейчас.

---

## Шаг 13. Добавить `RoutePassportOutboxWorker`

### 13.1. Шаг реализации

Добавить scheduled worker, который публикует pending outbox events.

### 13.2. Вход шага

Rows в `route_passport_outbox` со статусом `PENDING` и `next_attempt_at <= now()`.

### 13.3. Выход шага

Rows становятся:

```text
PUBLISHED — если Kafka publish успешен
PENDING   — если ошибка retryable и attempts < max_attempts
FAILED    — если attempts >= max_attempts
```

### 13.4. Какие существующие классы участвуют

| Класс | Как участвует |
|---|---|
| `RoutePassportKafkaPublisherService` | Физически отправляет Kafka record. |
| `RoutePassportOutboxRepository` | Читает/обновляет outbox rows. |
| `KafkaTopicsProperties` | Не нужен worker’у напрямую, topic уже записан в outbox. |

### 13.5. Нужно ли вводить новую модель

Да:

```text
RoutePassportOutboxWorker
RoutePassportOutboxService
RoutePassportOutboxProperties
```

### 13.6. Слой

`orchestration / adapter`

### 13.7. Properties

```yaml
app:
  route-passport:
    outbox:
      enabled: true
      batch-size: 50
      fixed-delay-ms: 1000
      lock-timeout-ms: 60000
      publish-timeout-ms: 10000
      max-attempts: 10
      initial-backoff-ms: 1000
      max-backoff-ms: 60000
```

### 13.8. Worker algorithm

```text
on schedule:
    claimedEvents = outboxService.claimDueBatch(batchSize)

    for event in claimedEvents:
        try:
            publisher.publish(event)
            outboxService.markPublished(event.id)
        catch exception:
            outboxService.markRetryOrFailed(event.id, exception)
```

### 13.9. Claim algorithm

В отдельной transaction:

```text
1. select due rows with FOR UPDATE SKIP LOCKED
2. set locked_until = now + lockTimeout
3. update updated_at
4. return rows
```

Native query example:

```sql
SELECT *
FROM route_passport_outbox
WHERE status = 'PENDING'
  AND next_attempt_at <= now()
  AND (locked_until IS NULL OR locked_until < now())
ORDER BY created_at
LIMIT :limit
FOR UPDATE SKIP LOCKED;
```

После select обновить `locked_until` у выбранных rows.

### 13.10. Backoff formula

Простая exponential backoff с cap:

```text
nextDelay = min(initialBackoffMs * 2^(attempts - 1), maxBackoffMs)
```

Для первого варианта можно без jitter. Если будут синхронные retry spikes — добавить jitter позже.

### 13.11. At-least-once behavior

Outbox worker дает at-least-once delivery.

Возможен дубль в сценарии:

```text
1. Kafka publish succeeded.
2. Service crashed before marking row as PUBLISHED.
3. locked_until expired.
4. Worker retries same event.
```

Это нормально для outbox. Downstream consumers обязаны быть идемпотентными.

Текущий downstream status:

| Consumer | Идемпотентность |
|---|---|
| `pcp-request-service` / `RouteMatchingService` | Частично есть: `request_route_matches` unique `(request_id, route_id)` защищает от дублей matching. |
| `pcp-mission-planing-service` | Требует отдельной проверки: повторный `ModeStatusChangedEvent` может повторно обновить режим и породить побочные события. |

### 13.12. Acceptance criteria

1. Pending outbox rows публикуются в Kafka.
2. Ошибки publish не теряют rows.
3. После max attempts row становится `FAILED`.
4. Несколько instances сервиса не публикуют один row одновременно.

---

## Шаг 14. Настроить storage и outbox observability

### 14.1. Шаг реализации

Добавить логи и базовые метрики для ingestion/outbox.

### 14.2. Вход шага

Новые сервисы:

```text
RoutePassportIngestionService
RoutePassportOutboxWorker
RoutePassportKafkaPublisherService
```

### 14.3. Выход шага

По logs/metrics можно понять:

1. сколько маршрутов принято;
2. сколько outbox events pending;
3. сколько published;
4. сколько failed;
5. какие ошибки уходят в DLQ.

### 14.4. Какие существующие классы участвуют

| Класс | Изменение |
|---|---|
| `RoutePassportMessageListener` | Логи input topic/partition/offset. |
| `RoutePassportIngestionService` | Логи routeId, status, created/updated, outbox inserted/skipped. |
| `RoutePassportOutboxWorker` | Логи batch size, publish success/failure. |
| `RoutePassportKafkaPublisherService` | Логи topic/partition/offset после publish. |

### 14.5. Нужно ли вводить новую модель

Нет.

### 14.6. Слой

`observability`

### 14.7. Рекомендуемые log fields

```text
routeId
routeNameFull
traceId
correlationId
sourceTopic
sourcePartition
sourceOffset
outboxEventId
eventType
targetTopic
attempts
status
```

### 14.8. Метрики

Если в проекте уже есть Micrometer через Actuator, добавить counters/gauges:

```text
route_passport_ingest_success_total
route_passport_ingest_failure_total
route_passport_outbox_published_total
route_passport_outbox_failed_total
route_passport_outbox_retry_total
route_passport_outbox_pending_count
route_passport_outbox_failed_count
```

Gauge `pending_count` и `failed_count` можно добавить позже, если не хочется усложнять первый MR. Counters на publish/retry/fail — минимально полезны.

### 14.9. Acceptance criteria

1. В логах можно отследить путь routeId от input до outbox publish.
2. Failed outbox rows не теряются незаметно.
3. DLQ события видны в логах с причиной.

---

## Шаг 15. Обновить конфигурацию topic properties

### 15.1. Шаг реализации

Расширить `KafkaTopicsProperties` новым DLQ topic.

### 15.2. Вход шага

Текущий класс:

```kotlin
@ConfigurationProperties(prefix = "app.kafka.topics")
data class KafkaTopicsProperties(
    val input: String = "",
    val processedRoute: String = "",
    val routeGeoreference: String = "",
)
```

### 15.3. Выход шага

Добавить:

```kotlin
val inputDlq: String = ""
```

### 15.4. Какие существующие классы участвуют

| Класс | Изменение |
|---|---|
| `KafkaTopicsProperties` | Добавить `inputDlq`. |
| `KafkaConsumerConfig` | Использовать `inputDlq` в `DeadLetterPublishingRecoverer`. |
| `config-repo/pcp-route-processing-service.yaml` | Добавить topic name. |

### 15.5. Нужно ли вводить новую модель

Нет.

### 15.6. Слой

`adapter / config`

### 15.7. Acceptance criteria

1. Сервис не стартует без DLQ topic name.
2. DLQ topic задается централизованно через config server.

---

## Шаг 16. Минимальные тесты

### 16.1. Шаг реализации

Добавить/обновить тесты только под измененную логику.

### 16.2. Вход шага

Текущие тесты:

```text
RoutePassportJsonParserServiceTest.kt
RoutePassportKafkaPublisherServiceTest.kt
RoutePassportMessageListenerTest.kt
RouteProcessingTestControllerTest.kt
```

### 16.3. Выход шага

Тесты покрывают:

1. parser;
2. geometry converter;
3. ingestion service;
4. outbox worker;
5. listener ack/error behavior;
6. error classification.

### 16.4. Какие существующие классы участвуют

| Тест | Что изменить |
|---|---|
| `RoutePassportJsonParserServiceTest` | Убрать ожидания ручной WKT конвертации, использовать `RouteGeometryConverter`. |
| `RoutePassportKafkaPublisherServiceTest` | Проверять publish одной outbox row и exception on failure. |
| `RoutePassportMessageListenerTest` | Проверять, что listener вызывает ingestion и ack после успеха; не ack при exception. |

### 16.5. Нужно ли вводить новую модель

Только test fixtures/builders, если они упрощают тесты.

### 16.6. Слой

`tests`

### 16.7. Unit tests

#### `RouteGeometryConverterTest`

Проверки:

| Case | Ожидание |
|---|---|
| valid Polygon | возвращается WKT `POLYGON ...` |
| valid MultiPolygon | возвращается WKT `MULTIPOLYGON ...` |
| unsupported geometry type | `RoutePassportContractException` |
| invalid coordinates | `RoutePassportContractException` |
| empty geometry | `RoutePassportContractException` |

#### `RoutePassportJsonParserServiceTest`

Проверки:

| Case | Ожидание |
|---|---|
| valid envelope | возвращается `ParsedRoutePassportMessage` |
| missing routeId | routeId генерируется из `id`, как сейчас |
| rollAngle list | min/max вычисляются корректно |
| visirAngle list | min/max вычисляются корректно |
| invalid interval | contract exception |
| routeStatus | остается default `PROCESSED` |

#### `RoutePassportIngestionServiceTest`

Проверки:

| Case | Ожидание |
|---|---|
| first route ingest | created route + 2 outbox rows |
| duplicate same routeId | no duplicate outbox rows |
| updated same routeId | route row updated, outbox unique preserved |
| raw payload | saved to `raw_payload` |
| source metadata | topic/partition/offset saved |

#### `RoutePassportOutboxWorkerTest`

Проверки:

| Case | Ожидание |
|---|---|
| pending event publish success | status `PUBLISHED`, `published_at` set |
| publish failure attempts < max | status `PENDING`, attempts incremented, `next_attempt_at` moved |
| publish failure attempts >= max | status `FAILED`, `last_error` set |
| locked row | not claimed until lock expires |

#### `RoutePassportMessageListenerTest`

Проверки:

| Case | Ожидание |
|---|---|
| success | parser + ingestion called, ack called |
| parser exception | ack not called, exception rethrown |
| ingestion exception | ack not called, exception rethrown |

### 16.8. Integration tests

Минимально желательно добавить JPA test для repositories и migration validation.

Варианты:

1. если Testcontainers уже используется в проекте — использовать PostgreSQL Testcontainer;
2. если нет — на первом этапе ограничиться `@DataJpaTest` + embedded database нельзя, потому что используются JSONB и PostgreSQL-specific SQL;
3. pragmatic вариант — unit tests + локальный manual check через real PostgreSQL в dev environment.

Рекомендация: добавить Testcontainers PostgreSQL, если это приемлемо для проекта. Если нет — явно прописать manual verification checklist.

### 16.9. Acceptance criteria

1. Измененная логика покрыта unit tests.
2. Outbox retry/fail behavior проверен.
3. Listener ack behavior проверен.
4. Parser сохраняет прежнюю бизнес-семантику routeStatus = PROCESSED.

---

## Шаг 17. Обновить runtime configuration

### 17.1. Шаг реализации

Свести все новые настройки в config server file.

### 17.2. Вход шага

```text
config-repo/pcp-route-processing-service.yaml
```

### 17.3. Выход шага

Один конфиг содержит:

1. datasource;
2. flyway;
3. kafka consumer manual ack settings;
4. producer reliability;
5. DLQ topic;
6. outbox properties.

### 17.4. Какие существующие классы участвуют

| Класс | Как участвует |
|---|---|
| `KafkaTopicsProperties` | topic settings. |
| `RoutePassportOutboxProperties` | outbox worker settings. |
| `KafkaConsumerConfig` | retry/ack settings. |
| `KafkaProducerConfig` | producer reliability settings. |

### 17.5. Нужно ли вводить новую модель

Да, желательно:

```kotlin
@ConfigurationProperties(prefix = "app.route-passport.outbox")
data class RoutePassportOutboxProperties(...)
```

Для retry тоже можно:

```kotlin
@ConfigurationProperties(prefix = "app.route-passport.consumer.retry")
data class RoutePassportConsumerRetryProperties(...)
```

### 17.6. Слой

`adapter / config`

### 17.7. Итоговый YAML fragment

```yaml
spring:
  datasource:
    driver-class-name: org.postgresql.Driver
    url: jdbc:postgresql://${pcp.infra.postgres.host}:${pcp.infra.postgres.port}/pcp_route_processing
    username: postgres
    password: password
  jpa:
    hibernate:
      ddl-auto: validate
    properties:
      hibernate:
        format_sql: true
  flyway:
    enabled: true
    baseline-on-migrate: true
    locations: classpath:db/migration
  kafka:
    bootstrap-servers: ${pcp.infra.kafka.host}:${pcp.infra.kafka.port}
    consumer:
      group-id: pcp-route-processing-service
      auto-offset-reset: earliest
      key-deserializer: org.apache.kafka.common.serialization.StringDeserializer
      value-deserializer: org.apache.kafka.common.serialization.StringDeserializer
      enable-auto-commit: false
    producer:
      key-serializer: org.apache.kafka.common.serialization.StringSerializer
      value-serializer: org.apache.kafka.common.serialization.StringSerializer
      acks: all
      retries: 10
      properties:
        enable.idempotence: true
        max.in.flight.requests.per.connection: 5
    listener:
      missing-topics-fatal: false
      ack-mode: manual_immediate

app:
  kafka:
    topics:
      input: pcp.request.survey-georeference.v1
      input-dlq: pcp.request.survey-georeference-dlq.v1
      processed-route: pcp.route.in.v1
      route-georeference: pcp.route.georeference.v1
  route-passport:
    consumer:
      retry:
        max-attempts: 3
        backoff-ms: 1000
        max-backoff-ms: 10000
    outbox:
      enabled: true
      batch-size: 50
      fixed-delay-ms: 1000
      lock-timeout-ms: 60000
      publish-timeout-ms: 10000
      max-attempts: 10
      initial-backoff-ms: 1000
      max-backoff-ms: 60000
```

### 17.8. Acceptance criteria

1. Все новые параметры находятся в config-repo.
2. Defaults безопасны для dev/local.
3. Production может менять retry/outbox параметры без пересборки сервиса.

---

# 7. Итоговый список новых файлов

## 7.1. Migrations

```text
services/pcp-route-processing-service/src/main/resources/db/migration/V1__create_route_passports.sql
services/pcp-route-processing-service/src/main/resources/db/migration/V2__create_route_passport_outbox.sql
```

## 7.2. Entities

```text
services/pcp-route-processing-service/src/main/kotlin/space/nstart/pcp/routepassportconsumer/entity/RoutePassportEntity.kt
services/pcp-route-processing-service/src/main/kotlin/space/nstart/pcp/routepassportconsumer/entity/RoutePassportOutboxEntity.kt
```

## 7.3. Repositories

```text
services/pcp-route-processing-service/src/main/kotlin/space/nstart/pcp/routepassportconsumer/repository/RoutePassportRepository.kt
services/pcp-route-processing-service/src/main/kotlin/space/nstart/pcp/routepassportconsumer/repository/RoutePassportOutboxRepository.kt
```

## 7.4. Services

```text
services/pcp-route-processing-service/src/main/kotlin/space/nstart/pcp/routepassportconsumer/service/RouteGeometryConverter.kt
services/pcp-route-processing-service/src/main/kotlin/space/nstart/pcp/routepassportconsumer/service/RoutePassportIngestionService.kt
services/pcp-route-processing-service/src/main/kotlin/space/nstart/pcp/routepassportconsumer/service/RoutePassportOutboxService.kt
services/pcp-route-processing-service/src/main/kotlin/space/nstart/pcp/routepassportconsumer/service/RoutePassportOutboxWorker.kt
```

## 7.5. Exceptions

```text
services/pcp-route-processing-service/src/main/kotlin/space/nstart/pcp/routepassportconsumer/exception/RoutePassportContractException.kt
services/pcp-route-processing-service/src/main/kotlin/space/nstart/pcp/routepassportconsumer/exception/RoutePassportProcessingException.kt
```

`RoutePassportProcessingException` optional. Добавлять только если реально нужен для отделения unexpected processing failures от generic exceptions.

## 7.6. Config properties

```text
services/pcp-route-processing-service/src/main/kotlin/space/nstart/pcp/routepassportconsumer/config/RoutePassportOutboxProperties.kt
services/pcp-route-processing-service/src/main/kotlin/space/nstart/pcp/routepassportconsumer/config/RoutePassportConsumerRetryProperties.kt
```

---

# 8. Итоговый список изменяемых файлов

```text
services/pcp-route-processing-service/build.gradle.kts
config-repo/pcp-route-processing-service.yaml
services/pcp-route-processing-service/src/main/kotlin/space/nstart/pcp/routepassportconsumer/config/KafkaConsumerConfig.kt
services/pcp-route-processing-service/src/main/kotlin/space/nstart/pcp/routepassportconsumer/config/KafkaProducerConfig.kt
services/pcp-route-processing-service/src/main/kotlin/space/nstart/pcp/routepassportconsumer/config/KafkaTopicsProperties.kt
services/pcp-route-processing-service/src/main/kotlin/space/nstart/pcp/routepassportconsumer/listener/RoutePassportMessageListener.kt
services/pcp-route-processing-service/src/main/kotlin/space/nstart/pcp/routepassportconsumer/service/RoutePassportJsonParserService.kt
services/pcp-route-processing-service/src/main/kotlin/space/nstart/pcp/routepassportconsumer/service/RoutePassportKafkaPublisherService.kt
services/pcp-route-processing-service/src/test/kotlin/space/nstart/pcp/routepassportconsumer/listener/RoutePassportMessageListenerTest.kt
services/pcp-route-processing-service/src/test/kotlin/space/nstart/pcp/routepassportconsumer/service/RoutePassportJsonParserServiceTest.kt
services/pcp-route-processing-service/src/test/kotlin/space/nstart/pcp/routepassportconsumer/service/RoutePassportKafkaPublisherServiceTest.kt
```

---

# 9. Порядок реализации по MR/commit-ам

## MR 1. DB foundation

Состав:

1. build.gradle: JPA/Flyway/PostgreSQL/JTS dependencies;
2. datasource/flyway config;
3. migrations `route_passports`, `route_passport_outbox`;
4. entities/repositories;
5. basic repository tests или compile validation.

Риск: низкий/средний. Код обработки Kafka еще можно не переключать.

---

## MR 2. Parser cleanup and geometry converter

Состав:

1. `RouteGeometryConverter`;
2. убрать ручной WKT converter из parser;
3. добавить `RoutePassportContractException`;
4. обновить parser tests;
5. добавить converter tests.

Риск: средний, потому что меняется geometry output. Нужно проверить WKT formatting compatibility с `pcp-request-service`.

---

## MR 3. Ingestion service + outbox writing

Состав:

1. `RoutePassportIngestionService`;
2. создание route row;
3. создание двух outbox rows;
4. idempotency by `route_id + event_type`;
5. tests на duplicate processing.

Риск: средний. Kafka listener еще можно временно не переключать, если сделать service isolated.

---

## MR 4. Listener controlled commit + DLQ

Состав:

1. listener принимает `ConsumerRecord` + `Acknowledgment`;
2. listener вызывает ingestion;
3. manual ack после успешного ingest;
4. `enable-auto-commit=false`;
5. `DefaultErrorHandler` + `DeadLetterPublishingRecoverer`;
6. DLQ topic config;
7. tests на ack/not ack.

Риск: высокий, потому что меняется delivery semantics входного Kafka topic.

---

## MR 5. Outbox publisher worker

Состав:

1. `RoutePassportOutboxService`;
2. `RoutePassportOutboxWorker`;
3. publish one outbox row;
4. mark published/retry/failed;
5. producer reliability config;
6. tests на success/retry/failed.

Риск: высокий, потому что это основной механизм downstream publication.

---

## MR 6. Observability and hardening

Состав:

1. structured logs;
2. metrics/counters;
3. failed outbox visibility;
4. manual replay notes/runbook draft;
5. финальный cleanup.

Риск: низкий.

---

# 10. Поведение после внедрения

## 10.1. Happy path

```text
1. Kafka input message received.
2. Parser reads envelope and creates RoutePassportDto.
3. Geometry converter converts GeoJSON geometry to WKT.
4. Ingestion service stores/updates route_passports.
5. Ingestion service creates outbox events:
   - ModeStatusChangedEvent
   - RouteGeoRefEvent
6. DB transaction commits.
7. Listener acknowledges Kafka offset.
8. Outbox worker publishes events.
9. Outbox rows become PUBLISHED.
```

## 10.2. Bad input

```text
1. Kafka input message received.
2. Parser/converter detects contract error.
3. Listener throws RoutePassportContractException.
4. Error handler sends original input to DLQ.
5. Offset is committed after recoverer succeeds.
6. No route_passports row is created.
7. No outbox rows are created.
```

## 10.3. DB down during ingest

```text
1. Kafka input message received.
2. Parser succeeds.
3. Ingestion fails due to DB error.
4. Listener throws exception.
5. Error handler retries with bounded backoff.
6. If DB recovers, ingest succeeds and offset is acked.
7. If DB does not recover after max attempts, message goes to DLQ.
```

## 10.4. Kafka output down during outbox publish

```text
1. Input message already safely stored in DB.
2. Outbox worker tries to publish.
3. Kafka publish fails.
4. Outbox row remains retryable.
5. next_attempt_at is moved by backoff.
6. Worker retries later.
7. After max attempts row becomes FAILED.
```

---

# 11. Production readiness checklist

## 11.1. Correctness

- [ ] `route_passports` stores all normalized route passport fields.
- [ ] `route_status` remains `PROCESSED` intentionally.
- [ ] `route_id` is primary idempotency key.
- [ ] Duplicate input message does not create duplicate outbox events.
- [ ] Geometry conversion supports Polygon and MultiPolygon.
- [ ] Invalid geometry goes to DLQ, not infinite retry.

## 11.2. Reliability

- [ ] `enable-auto-commit=false`.
- [ ] Manual ack happens only after DB commit.
- [ ] Listener does not publish Kafka output directly.
- [ ] Outbox worker retries failed publishes.
- [ ] Outbox worker marks permanently failed events.
- [ ] Kafka producer uses `acks=all`.
- [ ] Kafka producer idempotence is enabled.
- [ ] DLQ topic configured.
- [ ] Retry is bounded.

## 11.3. Observability

- [ ] Logs include `routeId`, `traceId`, `correlationId`, source topic/partition/offset.
- [ ] Outbox failed events are visible.
- [ ] DLQ sends are logged with exception category.
- [ ] Metrics exist for ingest success/failure and outbox publish/fail/retry.

## 11.4. Operational

- [ ] DB `pcp_route_processing` exists in target environment.
- [ ] Flyway migrations run successfully.
- [ ] DLQ topic exists or auto-created by Kafka infra policy.
- [ ] Output topics exist:
  - [ ] `pcp.route.in.v1`
  - [ ] `pcp.route.georeference.v1`
- [ ] Runbook exists for failed outbox rows.
- [ ] Runbook exists for DLQ replay.

---

# 12. Runbook drafts

## 12.1. Проверить зависшие outbox events

```sql
SELECT status, COUNT(*)
FROM route_passport_outbox
GROUP BY status;
```

```sql
SELECT id, route_id, event_type, topic, attempts, next_attempt_at, locked_until, last_error
FROM route_passport_outbox
WHERE status IN ('PENDING', 'FAILED')
ORDER BY updated_at DESC
LIMIT 100;
```

## 12.2. Переотправить FAILED outbox event вручную

Только после проверки причины ошибки.

```sql
UPDATE route_passport_outbox
SET status = 'PENDING',
    attempts = 0,
    next_attempt_at = now(),
    locked_until = NULL,
    last_error = NULL,
    updated_at = now()
WHERE id = '<outbox-event-id>'
  AND status = 'FAILED';
```

## 12.3. Найти route passport по routeId

```sql
SELECT *
FROM route_passports
WHERE route_id = '<route-id>';
```

## 12.4. Проверить события по routeId

```sql
SELECT id, event_type, topic, status, attempts, published_at, last_error
FROM route_passport_outbox
WHERE route_id = '<route-id>'
ORDER BY created_at;
```

---

# 13. Риски, которые останутся после выполнения плана

| Риск | Почему остается | Что делать позже |
|---|---|---|
| Kafka key остается `traceId` | Пункт смены key исключен из scope. | Отдельно решить: `routeId` или `correlationId` как business key. |
| Нет fallback trace/correlation | Пункт исключен из scope. | Добавить генерацию fallback и явно логировать missing metadata. |
| `routeStatus` всегда `PROCESSED` | Осознанное решение оставить как есть. | Когда появится контракт статусов, читать статус из input и валидировать enum. |
| Test REST controller остается в main-коде | Пункт исключен из scope. | Удалить или закрыть через `@Profile("local", "dev")`. |
| README/fixtures могут быть устаревшими | Пункт исключен из scope. | Обновить документацию контракта route passport. |
| Downstream idempotency не полностью проверена | План меняет producer delivery на at-least-once. | Проверить `pcp-mission-planing-service` на повторный `ModeStatusChangedEvent`. |
| Нет Kafka transactions | Выбран outbox вместо Kafka transactions. | Это нормально, если downstream consumers идемпотентны. |

---

# 14. Критерии готовности всей задачи

Задачу можно считать выполненной, если:

1. `pcp-route-processing-service` имеет собственную БД `pcp_route_processing`.
2. Принятый route passport сохраняется в `route_passports`.
3. Downstream-события создаются в `route_passport_outbox`, а не публикуются напрямую из listener.
4. Kafka offset input message подтверждается только после успешного DB commit.
5. Outbox worker публикует pending events в:
   - `pcp.route.in.v1`;
   - `pcp.route.georeference.v1`.
6. Ошибки publish не теряют события: outbox row остается retryable или становится `FAILED` после max attempts.
7. Bad input уходит в DLQ/DLT topic.
8. Retry bounded, бесконечного retry нет.
9. GeoJSON → WKT конвертация выполняется библиотечно.
10. `routeStatus` остается `PROCESSED`, и это явно зафиксировано в коде/тестах как текущая бизнес-семантика.
11. Минимальные тесты покрывают parser, converter, ingestion, outbox worker, listener ack/error behavior.

---

# 15. Короткая итоговая формулировка для задачи в трекере

```md
Доработать `pcp-route-processing-service` route-passport flow до production-ready обработки:

- добавить service-owned PostgreSQL DB `pcp_route_processing`;
- сохранять принятые route passports в `route_passports`;
- реализовать DB outbox `route_passport_outbox` для downstream Kafka events;
- убрать прямую публикацию Kafka events из input listener;
- подтверждать input Kafka offset только после DB commit;
- добавить bounded retry + DLQ для input topic;
- добавить outbox worker для публикации `ModeStatusChangedEvent` и `RouteGeoRefEvent`;
- настроить Kafka producer reliability (`acks=all`, retries, idempotence);
- заменить ручной GeoJSON → WKT converter на библиотечный;
- оставить текущую семантику `routeStatus = PROCESSED`;
- не добавлять BPM/Camunda, flow остается Kafka-only.
```