Files

T

Дмитрий Соловьев d48ddd2657 Init

2026-05-25 14:23:52 +03:00

79 KiB

Raw Blame History

План работ: доведение `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-сценарию:

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

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

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

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

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

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 ошибок

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. Рекомендуемое имя БД

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

pcp_route_processing

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

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

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

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

pcp-route-passport-bd

Причины:

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

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

Текущий base package:

space.nstart.pcp.routepassportconsumer

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

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 без БД.

Файлы:

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("plugin.jpa")

Добавить dependencies:

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`

Добавить:

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

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

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

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

Добавить Flyway migration:

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

2.2. Вход шага

RoutePassportDto:

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. Нужно ли вводить новую модель

Да:

RoutePassportEntity
RoutePassportRepository

2.6. Слой

adapter / storage

2.7. Рекомендуемый 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

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

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

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

Добавить Flyway migration:

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

3.2. Вход шага

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

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. Нужно ли вводить новую модель

Да:

RoutePassportOutboxEntity
RoutePassportOutboxRepository
RoutePassportOutboxStatus
RoutePassportOutboxService
RoutePassportOutboxWorker

3.6. Слой

adapter / storage, orchestration, adapter / messaging

3.7. Рекомендуемый 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

enum class RoutePassportOutboxStatus {
    PENDING,
    PUBLISHED,
    FAILED
}

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

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

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

Сценарий:

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

На один маршрут создаются максимум два outbox-события: ModeStatusChangedEvent, RouteGeoRefEvent.
Повторная обработка того же routeId не плодит дубли outbox rows.
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. Нужно ли вводить новую модель

Да:

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.

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

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

4.8. `RoutePassportOutboxEntity`

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

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

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

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

4.9. Repository methods

RoutePassportRepository:

interface RoutePassportRepository : JpaRepository<RoutePassportEntity, UUID>

RoutePassportOutboxRepository:

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

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

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

4.10. Acceptance criteria

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

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

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

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

5.2. Вход шага

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

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. Нужно ли вводить новую модель

Да:

RouteGeometryConverter

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

5.6. Слой

algorithm

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

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 смешивает две разные ответственности:

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

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

5.10. Acceptance criteria

В RoutePassportJsonParserService нет ручного toWktCoordinates.
Polygon и MultiPolygon продолжают поддерживаться.
Ошибки geometry классифицируются как non-retryable contract errors.
Есть 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. Нужно ли вводить новую модель

Да:

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

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

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

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

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

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

7.2. Вход шага

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

message value
record key
topic
partition
offset

7.3. Выход шага

DB transaction committed:

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. Нужно ли вводить новую модель

Да:

RoutePassportIngestionService
RoutePassportSourceRecordMetadata

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

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

7.6. Слой

orchestration / application

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

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

@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

if route exists:
    update fields
else:
    insert fields

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

Вариант B — native upsert

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(...):

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 пропущен, поэтому сохраняем текущую семантику:

message_key = parsedMessage.metadata.traceId

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

7.11. Acceptance criteria

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

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

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

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

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

8.2. Вход шага

Текущий listener:

fun listen(message: String)

8.3. Выход шага

Новый listener:

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

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

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:

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

8.8. Acceptance criteria

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

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

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

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

9.2. Вход шага

Текущий config:

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

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

9.8. Kotlin config

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

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

9.9. Acceptance criteria

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

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

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

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

10.2. Вход шага

Ошибки из:

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. Нужно ли вводить новую модель

Да:

RoutePassportContractException

И properties:

RoutePassportDlqProperties / RoutePassportConsumerErrorHandlingProperties

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

10.6. Слой

adapter / infrastructure

10.7. DLQ topic naming

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

pcp.request.survey-georeference.v1.dlt

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

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

pcp.request.survey-georeference-failed.v1

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

pcp.request.survey-georeference-dlq.v1

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

pcp.request.survey-georeference-dlq.v1

Добавить в config:

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:

val inputDlq: String = ""

10.8. Retry policy

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

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 должен сохранять:

original payload;
original topic;
original partition;
original offset;
exception class;
exception message;
service name;
timestamp.

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

10.10. Observability

Логировать:

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

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

10.11. Acceptance criteria

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

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

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

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

11.2. Вход шага

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

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

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 нужно расширить:

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 нужен, чтобы событие не потерялось, если:

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

11.11. Acceptance criteria

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

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

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

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

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

12.2. Вход шага

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

12.3. Выход шага

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

RoutePassportKafkaPublisherService.publish(outboxEvent)

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

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

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

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

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

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

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 должен дождаться результата отправки.

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

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

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

12.8. Acceptance criteria

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

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

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

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

13.2. Вход шага

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

13.3. Выход шага

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

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. Нужно ли вводить новую модель

Да:

RoutePassportOutboxWorker
RoutePassportOutboxService
RoutePassportOutboxProperties

13.6. Слой

orchestration / adapter

13.7. Properties

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

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:

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:

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:

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

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

13.11. At-least-once behavior

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

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

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

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

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

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

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

14.2. Вход шага

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

RoutePassportIngestionService
RoutePassportOutboxWorker
RoutePassportKafkaPublisherService

14.3. Выход шага

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

сколько маршрутов принято;
сколько outbox events pending;
сколько published;
сколько failed;
какие ошибки уходят в 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

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

14.8. Метрики

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

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

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

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

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

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

15.2. Вход шага

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

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

15.3. Выход шага

Добавить:

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

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

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

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

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

16.2. Вход шага

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

RoutePassportJsonParserServiceTest.kt
RoutePassportKafkaPublisherServiceTest.kt
RoutePassportMessageListenerTest.kt
RouteProcessingTestControllerTest.kt

16.3. Выход шага

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

parser;
geometry converter;
ingestion service;
outbox worker;
listener ack/error behavior;
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.

Варианты:

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

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

16.9. Acceptance criteria

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

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

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

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

17.2. Вход шага

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

17.3. Выход шага

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

datasource;
flyway;
kafka consumer manual ack settings;
producer reliability;
DLQ topic;
outbox properties.

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

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

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

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

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

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

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

17.6. Слой

adapter / config

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

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

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

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

7.1. Migrations

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

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

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

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

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

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. Итоговый список изменяемых файлов

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

Состав:

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

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

MR 2. Parser cleanup and geometry converter

Состав:

RouteGeometryConverter;
убрать ручной WKT converter из parser;
добавить RoutePassportContractException;
обновить parser tests;
добавить converter tests.

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

MR 3. Ingestion service + outbox writing

Состав:

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

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

MR 4. Listener controlled commit + DLQ

Состав:

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

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

MR 5. Outbox publisher worker

Состав:

RoutePassportOutboxService;
RoutePassportOutboxWorker;
publish one outbox row;
mark published/retry/failed;
producer reliability config;
tests на success/retry/failed.

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

MR 6. Observability and hardening

Состав:

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

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

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

10.1. Happy path

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

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

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

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

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

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 вручную

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

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

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

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

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. Критерии готовности всей задачи

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

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

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

Доработать `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.

79 KiB Raw Blame History Unescape Escape

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

0. Цель работ

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

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

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

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

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

2.1. Делаем

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

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

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

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

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

4. Database naming and ownership

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

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

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

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

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

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

1.2. Вход шага

1.3. Выход шага

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

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

1.6. Слой

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

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

1.9. Acceptance criteria

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

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

2.2. Вход шага

2.3. Выход шага

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

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

2.6. Слой

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

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

2.9. Почему polarisation JSONB

2.10. Acceptance criteria

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

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

3.2. Вход шага

3.3. Выход шага

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

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

3.6. Слой

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

3.8. Outbox statuses

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

3.10. Acceptance criteria

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

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

4.2. Вход шага

4.3. Выход шага

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

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

4.6. Слой

4.7. RoutePassportEntity

4.8. RoutePassportOutboxEntity

4.9. Repository methods

4.10. Acceptance criteria

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

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

5.2. Вход шага

5.3. Выход шага

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

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

5.6. Слой

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

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

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

5.10. Acceptance criteria

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

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

6.2. Вход шага

6.3. Выход шага

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

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

6.6. Слой

79 KiB

Raw Blame History

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

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

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

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

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

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

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

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

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

4.7. `RoutePassportEntity`

4.8. `RoutePassportOutboxEntity`

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

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

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

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

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