Microservice: Сервис сохранения подготовленных событий/сообщений в БД из смежных систем

ingestion-service

WarningОграничение публичной документации

В открытом доступе представлена демонстрационная версия метода. В настоящей публичной документации отображены не все шаги, технические сценарии и приватные эндпоинты для системы цифровых симуляторов бизнес-процессов.

• Полная спецификация метода: Будет доступна только во внутреннем контуре разработки (Confluence / Swagger Enterprise).

1 Общее назначение и архитектурная роль компонента

food-ingestion-service функционирует как горизонтально масштабируемый gRPC-продюсер. Его единственная задача — принять запрос от внешнего клиента (мобильного приложения или симулятора), преобразовать его в событие и отправить в шину данных Kafka.

2 Схема архитектуры компонента (Producer Flow)

graph TD
    %% HIGH-CONTRAST PASTEURIZED UI NODES (Text forced to absolute black)
    classDef core fill:#B2F5EA,stroke:#319795,stroke-width:1.5px,color:#000000,font-weight:bold;
    classDef storage fill:#FEB2B2,stroke:#E53E3E,stroke-width:1.5px,color:#000000,font-weight:bold;

    %% ULTRA-LIGHT CUSTOM PASTEL SUBGRAPH DOMAIN FRAMES
    style Transport_Layer fill:#E6FFFA,stroke:#4FD1C5,stroke-width:1px,color:#234E52,font-weight:bold
    style Core_Layer fill:#E6FFFA,stroke:#4FD1C5,stroke-width:1px,color:#234E52,font-weight:bold
    style Data_Bus fill:#FFF5F5,stroke:#FC8181,stroke-width:1px,color:#9B2C2C,font-weight:bold

    subgraph Transport_Layer ["1. Транспортный слой (Входной контур)"]
        gRPC["gRPC API Handlers"]:::core
    end

    subgraph Core_Layer ["2. Бизнес-логика (Async Core)"]
        AsyncSvc["foodTrackerAsyncService"]:::core
        KP["KafkaPublisher Interface"]:::core
    end

    subgraph Data_Bus ["3. Шина данных (Выходной контур)"]
        T1["topic: purchase-events"]:::storage
        T2["topic: cooking-events"]:::storage
        T3["topic: consumption-events"]:::storage
        T4["topic: waste-events"]:::storage
    end

    %% DATA INTERCHANGE STREAM FLOWS
    gRPC -->|1. Вызов команд| AsyncSvc
    AsyncSvc -->|2. Очистка токенов и маршалинг в JSON| KP

    KP -->|3. Key: group_id| T1
    KP -->|3. Key: group_id| T2
    KP -->|3. Key: group_id| T3
    KP -->|3. Key: group_id| T4

3 Системная спецификация: Ingestion API Producer (Write Path)

• Асинхронный неблокирующий ввод: Входные транспортные хэндлеры полностью изолированы от транзакционной логики базы данных PostgreSQL. Они выполняют быструю валидацию контрактов, очищают метаданные от чувствительной информации (dto.AccessToken = ““), преобразуют структуры домена в JSON-строки и атомарно публикуют их в брокер Apache Kafka. Это гарантирует минимальный и стабильный процент времени ответа (duration_ms) gRPC-клиентам и симуляторам даже при пиковых нагрузках.
• Партиционирование по бизнес-контексту: Публикация всех событий в шину данных выполняется со строгой фиксацией ключа Key = group_id. Брокер Kafka гарантирует, что сообщения с идентичным ключом всегда распределяются в одну и ту же партицию. Это обеспечивает строгую хронологическую последовательность обработки команд консьюмерами в рамках одной домашней группы (исключая состояние гонки, когда команда списания или готовки обгоняет команду закупки).
• Сквозная трассировка (Distributed Tracing): Транспортный слой оснащен перехватчиком, который извлекает маркер сквозного логирования x-request-id из gRPC-метаданных от симулятора/клиента. При отсутствии заголовка генерируется новый UUID v4. Этот маркер принудительно пробрасывается в контексте во все структурированные логи slog и в метаданные сообщений Kafka для сквозного аудита.

4 Спецификация Входных gRPC-интерфейсов (Транспортный контракт)

Транспортный слой обрабатывает данные в рамках спецификации food-ingestion-service. В случае инфраструктурных сбоев (например, сетевой разрыв с кластером брокера) или ошибок сериализации, внутренние исключения Go преобразуются в канонические коды ошибок пакета google.golang.org/grpc/codes.

4.1 Метод TrackPurchase (Регистрация закупки по чеку)

4.1.1 Входящий запрос (Request)

{
  "receipt_no": "MAGNUM-BIG-CHECK-001",
  "total_price": 4100.0,
  "items": [
    {
      "item_id": "8b5da76c-1e82-4140-87e0-0189a367757a",
      "name": "ФИЛЕ КУРИНОЕ ОХЛАЖДЕННОЕ",
      "category": "Мясо",
      "portions": 2.0,
      "weight_grams": 2400.0,
      "price": 3100.0
    },
    {
      "item_id": "c3a8b30e-e289-4b68-9bf9-9fa9d1b9ee56",
      "name": "ТОМАТЫ СЛИВОВИДНЫЕ",
      "category": "Овощи",
      "portions": 4.0,
      "weight_grams": 1600.0,
      "price": 924.0
    }
  ]
}

4.2 Метод TrackCooking (Регистрация приготовления блюда)

4.2.1 Входящий запрос (Request)

{
  "dish_item_id": "99999999-9c0b-4ef8-bb6d-6bb9bd380a11",
  "dish_name": "Куриный суп с томатами",
  "yield_portions": 4.0,
  "yield_weight_grams": 2000.0,
  "ingredients": [
    {
      "source_item_id": "8b5da76c-1e82-4140-87e0-0189a367757a",
      "used_portions": 1.0,
      "used_weight_grams": 1200.0
    },
    {
      "source_item_id": "c3a8b30e-e289-4b68-9bf9-9fa9d1b9ee56",
      "used_portions": 2.0,
      "used_weight_grams": 800.0
    }
  ]
}

4.3 Метод TrackConsumption (Списание остатков)

4.3.1 Входящий запрос (Request)

{
  "items": [
    {
      "item_id": "99999999-9c0b-4ef8-bb6d-6bb9bd380a11",
      "portions": 1.0,
      "weight_grams": 500.0
    },
    {
      "item_id": "11111111-a311-47bb-a8c9-0cc8cd491e77",
      "portions": 0.5,
      "weight_grams": 200.0
    }
  ]
}

4.4 Метод TrackWaste (Списание остатков)

4.4.1 Входящий запрос (Request)

{
  "items": [
    {
      "item_id": "c3a8b30e-e289-4b68-9bf9-9fa9d1b9ee56",
      "portions": 1.0,
      "weight_grams": 400.0,
      "reason": "Покрылись плесенью в ящике"
    },
    {
      "item_id": "11111111-a311-47bb-a8c9-0cc8cd491e77",
      "portions": 0.5,
      "weight_grams": 200.0,
      "reason": "Зачерствел, забыли закрыть пакет"
    }
  ]
}

• При пустом значении item_id или отрицательных значениях объемов (units / weight) отдается статус codes.InvalidArgument.

5 Сценарий взаимодействия при публикации команд (Sequence Diagram)

Ниже представлена диаграмма обработки gRPC-вызова, работы интерцептора логирования, маршалинга структуры в JSON и отправки сообщения в Kafka кластер.

%%{init: {
  'theme': 'base',
  'themeVariables': {
    'actorBkg': '#EBF8FF',
    'actorBorder': '#A3BFFA',
    'actorTextColor': '#000000',
    'rectBkg': '#FAF7F0',
    'rectBorder': '#E3DDCF',
    'noteBkgColor': '#E6F4EA',
    'noteBorderColor': '#34D399',
    'noteTextColor': '#064E3B',
    'signalColor': '#2D3748',
    'signalLineColor': '#4A5568'
  }
}}%%

sequenceDiagram
    autonumber
    actor Client as Внешний Клиент / Симулятор
    participant Interceptor as IngestionLoggingInterceptor
    participant Server as foodTrackerAsyncService
    participant KP as KafkaPublisher (Client)

    Client->>Interceptor: Вызов метода gRPC (например, TrackPurchase) с MD [x-request-id]
    activate Interceptor
    
    note over Interceptor: Шаг 2: Извлечение x-request-id из метаданных.<br/>При отсутствии — генерация uuid.NewV4()
    note over Interceptor: Шаг 3: Инициализация JSON-логгера slog с фиксацией request_id и метода
    
    Interceptor->>Server: Перенаправление контекста и DTO во внутренний хэндлер
    activate Server
    
    note over Server: Шаг 5: Безопасность — dto.AccessToken = ""<br/>Маршалинг структуры DTO в бинарный JSON payload
    
    Server->>KP: Вызов PublishEvent(ctx, "purchase-events", groupID, payload)
    activate KP
    KP-->>Server: Успешный возврат nil (Подтверждение ACK от Kafka зафиксировано)
    deactivate KP

    Server-->>Interceptor: Возврат пустой структуры ответа и статуса ошибки (nil)
    deactivate Server
    
    note over Interceptor: Шаг 9: Расчет duration_ms и запись лога JSON Lines в os.Stdout
    
    Interceptor-->>Client: Ответ EmptyResponse + gRPC Status Code (codes.OK)
    deactivate Interceptor

Figure 1: Пошаговая последовательность выполнения асинхронной публикации событий в Kafka

5.1 Таблица расшифровки шагов сценария асинхронного логирования команд

Table 1: : Пошаговый сценарий асинхронного логирования команд и публикации событий в Kafka

Шаг	Действие	Параметры / Запросы / DTO	Код ошибки (canonical_code)	HTTP / gRPC статус
1 (Client -> Interceptor)	Симулятор или клиент отправляет DTO команды, передавая маркер трассировки в заголовках.	gRPC Metadata Context: x-request-id: "uuid-v4-trace-id"	INGEST_INVALID_ARGUMENT	codes.InvalidArgument (400)
2-3 (Interceptor -> Interceptor)	Внутренняя логика: Регистрация request_id, инициализация потока логирования и сброс лога о старте операции в stdout.	slog JSON Line Payload: { "msg": "Async request started", "request_id": "uuid", "method": "TrackPurchase" }	Нет	codes.OK (200)
4 (Interceptor -> Server)	Интерцептор пробрасывает модифицированный контекст на уровень асинхронного Go-сервиса.	Внутренний вызов Go: s.TrackPurchase(ctx, userID, groupID, dto)	Нет	codes.OK (200)
5 (Server -> Server)	Внутренняя логика: Стирание токенов из структуры во избежание утечки в логи брокера. Сериализация DTO.	Внутренняя функция: json.Marshal(dto)	JSON_MARSHAL_FAILURE	codes.Internal (500)
6 (Server -> KP)	Сервис вызывает метод клиента Kafka, передавая имя целевого топика и ключ партиционирования.	Аргументы метода: PublishEvent(ctx, "purchase-events", groupID, payload)	KAFKA_PUBLISH_TIMEOUT	codes.Internal (500)
7 (KP -> Server)	Драйвер брокера подтверждает фиксацию сообщения в коммит-логе партиции и возвращает управление.	Kafka Wire Network ACK	KAFKA_CLUSTER_DOWN	codes.Internal (500)
8-9 (Server -> Interceptor)	Проброс успешного статуса выполнения, замер общего времени удержания сетевого gRPC-соединения.	slog JSON Line Payload: { "level": "INFO", "duration_ms": 1.8, "request_id": "trace-id" }	Нет	codes.OK (200)
10 (Interceptor -> Client)	Клиент получает ответ об успешной регистрации команды.	Задача ушла на обработку в очередь.	Нет	codes.OK (200)

6 Справочник Обсервабилити (Структура логов Producer-контура для ClickHouse)

Каждая строка лога, сбрасываемая в os.Stdout, содержит строго типизированные поля для сборщика Vector. При аварийных ситуациях в ClickHouse фиксируются следующие метрики:

Table 2: : Справочная матрица маппинга и обсервабилити логов

Код (canonical_code)	Уровень (log_level)	Статусы (transport_statuses)	Получатель (error_target)	JSON для Фронтенда (ui_payload)	Метрики для ClickHouse (observability_json)
INGEST_INVALID_ARGUMENT	WARN	{"grpc": 3, "http": 400}	EXTERNAL_SYSTEM	{"error": "Неверный формат данных чека или группы"}	{"metric": "bad_request", "labels": {"source": "simulator"}}
JSON_MARSHAL_FAILURE	ERROR	{"grpc": 13, "http": 500}	INTERNAL_SYSTEM	null	{"metric": "serialization_err", "labels": {"service": "ingestion_async"}}
KAFKA_PUBLISH_TIMEOUT	ERROR	{"grpc": 13, "kafka_ack": 0}	INTERNAL_SYSTEM	{"error": "Шина данных временно перегружена"}	{"metric": "kafka_timeout", "labels": {"partition_key": "group_id"}}
KAFKA_CLUSTER_DOWN	CRITICAL	{"grpc": 13, "kafka_ack": -1}	INFRASTRUCTURE	{"error": "Сервис временно недоступен"}	{"metric": "kafka_broker_dead", "labels": {"cluster": "production"}}