Microservice: Сервис receipt-processor многоканальный асинхронный шлюз

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

Компонент receipt-processor функционирует как многоканальный асинхронный шлюз (Ingestion Transport Gateway & Router) в Core Backend Layer. Его основная задача — изолировать синхронный транспортный gRPC-слой мобильных клиентов и B2B-партнеров от «тяжелых» ИИ-вычислений. Сервис принимает потоковое медиа (картинки, аудио), сохраняет его в объектное хранилище MinIO, регистрирует транзакции в PostgreSQL и асинхронно распределяет задачи по топикам Kafka для последующего анализа ИИ-моделями.

2 Схема архитектуры компонента (Входной контур: Шаг 1)

graph TD
    subgraph Transport_Layer ["1. Edge & Transport Layer (Входные контракты)"]
        App["Мобильный Клиент / Стрим"]
        Partner["B2B Ритейлер / Магнум"]
        gRPC_Srv["grpc.ReceiptServer (Handlers)"]
    end

    subgraph Core_Layer_Step1 ["2. Service Core (Инициация транзакции)"]
        KClient["kafka.Client (franz-go Producer)"]
        S3Client["minio.Client (MinIO SDK)"]
        Repo["postgres.ReceiptRepository (pgxpool)"]
    end

    subgraph Storage_Layer_Step1 ["3. Storage & Persistence (Первичный сброс)"]
        MinIO["MinIO S3 Object Storage"]
        Postgres["PostgreSQL DB"]
        T_Img["topic: receipt-images-topic"]
        T_Audio["topic: receipt-audio-topic"]
        T_Clean["topic: receipt-clean-products-topic"]
    end

    subgraph External_Layer_Step1 ["4. External Dependencies"]
        AuthSvc["auth-service (ValidateToken)"]
    end

    App -->|gRPC клиентский стрим| gRPC_Srv
    Partner -->|gRPC контракт| gRPC_Srv
    
    gRPC_Srv -->|Валидация сессии B2C| AuthSvc
    gRPC_Srv -->|Сохранение чанков медиа-файла| S3Client
    S3Client -->|PutObject| MinIO
    
    gRPC_Srv -->|Фиксация accepted транзакции| Repo
    Repo --> Postgres
    
    gRPC_Srv -->|Инициирующий пуш чека| KClient
    gRPC_Srv -->|Инициирующий пуш аудио| KClient
    KClient -->|ProduceSync| T_Img
    KClient -->|ProduceSync| T_Audio
    
    gRPC_Srv -.->|B2B прямой пуш минуя ИИ| T_Clean

    style gRPC_Srv fill:#2B6CB0,stroke:#fff,stroke-width:2px,color:#fff
    style MinIO fill:#10B981,stroke:#fff,stroke-width:2px,color:#fff
    style Postgres fill:#D69E2E,stroke:#fff,stroke-width:2px,color:#fff
    style T_Img fill:#E53E3E,stroke:#fff,stroke-width:1px,color:#fff
    style T_Audio fill:#E53E3E,stroke:#fff,stroke-width:1px,color:#fff

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

• Изоляция и сборка gRPC-стримов: Сервис реализует неблокирующий прием бинарных данных через механизмы клиентского стриминга (stream.Recv). Кусочки файлов (чанки) собираются в оперативной памяти (bytes.Buffer) и атомарно отгружаются в объектное хранилище MinIO S3 с MIME-типом image/jpeg или дефолтным аудио-форматом. Это разгружает дисковую подсистему и обеспечивает стабильный p99 времени ответа для мобильных приложений.
• Внешняя gRPC-аутентификация (B2C): Перед началом сборки стрима интерцептор извлекает токен access_token из входящего gRPC-контекста метаданных (metadata.FromIncomingContext) и выполняет синхронный RPC-вызов в auth-service. Успешный ответ обогащает контекст идентификатором домашней группы пользователя (GroupId), который пробрасывается дальше по всему пайплайну в структуре model.KafkaMessage.
• Прямой B2B-контур (Паттерн Reply-To): Для крупных ритейл-партнеров (например, торговой сети «Магнум») gRPC-сервер предоставляет выделенный метод RegisterStoreReceipt. Чек, уже распознанный на стороне кассы магазина, минует ресурсоемкие фазы ИИ-анализа (OCR и Whisper) и мгновенно публикуется напрямую в топик receipt-clean-products-topic. При этом в заголовки Kafka-сообщения принудительно внедряются метаданные reply_to="store-callbacks-topic" и уникальный correlation_id для асинхронного трекинга.

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

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

4.1 Метод UploadReceipt (Клиентский стрим графического чека)

Принимает бинарные чанки фотографии чека, загружает собранный файл в MinIO и отправляет событие-триггер в Kafka.

• Входные параметры (stream pb.UploadReceiptRequest): Поток чанков, содержащий структуру oneof Data (Metadata с указанием расширения файла или Chunk с массивом байт). Если стрим пустой, сервер возвращает статус codes.InvalidArgument.
• Выходные параметры (pb.UploadReceiptResponse): После успешной публикации метаданных в брокер возвращается статус "processing" и сгенерированный уникальный строковый идентификатор receipt_id (с префиксом BILL-).

4.2 Метод UploadAudio (Клиентский стрим голосовой команды)

Принимает чанки аудиозаписи речи пользователя, добавляющего продукты голосом, и отправляет метаданные в топик receipt-audio-topic.

• Входные параметры (stream pb.UploadAudioRequest): Поток данных аудио-файла. При невалидном или отсутствующем токене в контексте метаданных gRPC возвращает статус codes.Unauthenticated.
• Выходные параметры (pb.UploadAudioResponse): Возвращает статус "processing" и сгенерированный идентификатор audio_id (с префиксом AUDIO-).

4.3 Метод RegisterStoreReceipt (B2B Импорт чеков)

Принимает готовый структурированный состав позиций от внешних торговых сетей по токену авторизации.

• Входные параметры (pb.StoreReceiptRequest): Содержит уникальный токен магазина StoreToken, номер чека StoreReceiptNo, массив позиций Items и идентификатор целевой группы TargetGroupId. При невалидном токене возвращает статус codes.PermissionDenied.

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

Ниже представлена диаграмма сквозного жизненного цикла чеков и голосовых сообщений. Схема охватывает потоковую gRPC-загрузку через интерцепторы, неблокирующее хранение в MinIO S3, оркестрацию в ReceiptHandler, многоэтапный ИИ-анализ (Whisper, Llama Vision, Qwen) с фильтрацией аномалий и асинхронный обратный вызов партнеров по паттерну Reply-To.

%%{init: {
  'theme': 'base',
  'themeVariables': {
    'actorBkg': '#E3F2FD',
    'actorBorder': '#546E7A',
    'actorTextColor': '#0D47A1',
    'rectBkg': '#FFF9C4',
    'rectBorder': '#FBC02D',
    'noteBkgColor': '#F3E5F5',
    'noteBorderColor': '#7E57C2',
    'noteTextColor': '#311B92',
    'signalColor': '#2C3E50',
    'signalLineColor': '#2C3E50'
  }
}}%%
sequenceDiagram
    autonumber
    actor Client as Внешний Клиент / Магнум
    participant Server as gRPC ReceiptServer
    participant KC as kafka.Client (franz-go)
    participant H as ReceiptHandler (Router)
    participant S3 as minio.Client (S3 Storage)
    participant AI as ai.Client (Ollama/Whisper)
    participant DB as postgres.ReceiptRepository

    %% ВЕТКА 1: Загрузка файлов (B2C)
    Client->>Server: Стриминг чанков gRPC (UploadReceipt / UploadAudio) с access_token
    activate Server
    note over Server: Шаг 2: Валидация access_token через auth-service.<br/>Извлечение GroupID сессии.
    Server->>S3: UploadFile(bucket, filename, payloadBytes)
    S3-->>Server: Возврат s3Path
    Server->>KC: PublishEvent(topic, jsonBytes) [model.KafkaMessage]
    KC-->>Server: Сетевой ACK подтвержден
    Server-->>Client: Срочный ответ (Status: "processing", ID)
    deactivate Server

    %% ВЕТКА 2: Асинхронный Поллинг очередей (Воркеры)
    activate KC
    KC->>KC: PollFetches(ctx) из бесконечного цикла Start()
    KC->>H: Вызов ProcessMessage(ctx, topic, value)
    activate H

    alt Топик == "receipt-images-topic" ( Фаза 1)
        H->>S3: DownloadFile(bucket, s3Link)
        S3-->>H: imgBytes (image/jpeg)
        note over H: base64.StdEncoding.EncodeToString(imgBytes)
        H->>AI: ExtractEntitiesFromImage(ctx, b64Img) [Llama Vision:11435]
        AI-->>H: rawJsonProducts
        note over H: Шаг 13: Обогащение промежуточной структуры group_id
        H->>KC: PublishEvent("receipt-raw-products-topic", rawJsonProducts)
    
    else Топик == "receipt-audio-topic" (Воркер 3)
        H->>S3: DownloadFile(bucket, s3Link)
        S3-->>H: audioBytes
        H->>AI: RecognizeSpeech(ctx, audioBytes) [Whisper:11436]
        AI-->>H: speechText
        H->>AI: ConvertSpeechToProductsJSON(ctx, speechText) [Qwen:11437]
        AI-->>H: aiJsonBytes (С семантическим флагом validationCheck.Status)
        
        alt ИИ-Статус == "invalid" ( Dead Letter Path)
            H->>KC: PublishEvent("receipt-moderation-topic", aiJsonBytes)
            note over KC: Перенаправление в voice_moderation_queue
        else ИИ-Статус == "valid"
            H->>KC: PublishEvent("receipt-raw-products-topic", aiJsonBytes)
        end
    
    else Топик == "receipt-raw-products-topic" ( Фаза 2)
        H->>AI: FilterProfanityAndSpam(ctx, value) [Qwen:11438]
        AI-->>H: cleanJsonProducts (Мат заменен на '*', спам удален)
        H->>KC: PublishEvent("receipt-clean-products-topic", cleanJsonProducts)
    end
    deactivate H
    deactivate KC

    %% ВЕТКА 3: Финализация и Обратный вызов (B2B Webhooks)
    activate KC
    KC->>KC: PollFetches в StoreCallbackWorker ("store-callbacks-topic")
    KC->>DB: GetWebhookURLByCorrelationID(ctx, correlationID)
    DB-->>KC: WebhookURL ритейлера + Номер чека
    KC->>Client: HTTP POST Webhook (clean payload + status: "processed")
    note over Client: Жесткий таймаут ожидания ответа — 10 секунд
    Client-->>KC: HTTP Status 200 OK
    KC->>DB: UpdateTransactionStatus(ctx, correlationID, "processed", "")
    deactivate KC

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

Шаг	Действие	Параметры / Запросы / DTO	Код ошибки (canonical_code)	HTTP / gRPC статус
1-2	Клиент инициирует gRPC-стрим, передавая токен в заголовках метаданных.	metadata.MD: {"access_token": ["jwt"]}	METADATA_MISSING	codes.Unauthenticated (401)
3-4	Потоковая сборка файла в буфер и неблокирующая отправка в объектное хранилище.	minio.PutObject(ctx, "receipts", fileName, reader)	MINIO_UPLOAD_FAILURE	codes.Internal (500)
5	Публикация инициирующего сообщения с метаданными во входные очереди.	model.KafkaMessage JSON-строка	KAFKA_PRODUCE_SYNC_ERR	codes.Internal (500)
8-9	Воркер вычитывает картинку и запрашивает бинарный объект из S3 без удержания блокировок БД.	s3Client.DownloadFile(ctx, bucket, path)	MINIO_DOWNLOAD_FAILURE	Внутренний перезапуск
11	Вызов мультимодального OCR-слоя инференса LLM на выделенном Ubuntu-сервере.	PromptRequest: {model: "llama3.2-vision:11b", images: [b64]}	AI_INFERENCE_TIMEOUT	Внутренний перезапуск
15-16	STT-транскрибация голоса и последующий семантический парсинг полноты данных.	RecognizeSpeech(ctx, audioBytes) [Whisper]	WHISPER_STT_CRASH	Внутренний перезапуск
18	Маршрутизация аномалий: ИИ выявил нехватку метрик (нет веса/количества). Пуш в DLQ топик.	validationCheck.Status == "invalid"	VOICE_METRIC_INVALID	DLQ_ROUTING (Лог)
21-22	Вызов SLM Qwen для санитизации контента, удаления спама и посимвольного маскирования мата.	FilterProfanityAndSpam(ctx, rawJson) [Qwen]	AI_MODEL_UNAVAILABLE	Внутренний перезапуск
26-28	Поллинг результатов, поиск партнера по correlation_id в БД и отправка уведомления.	http.MethodPost к WebhookURL с таймаутом 10с.	STORE_WEBHOOK_TIMEOUT	HTTP 504 / 500 Error
29	Фиксация успешного завершения бизнес-транзакции в локальной СУБД.	UpdateTransactionStatus(ctx, corrID, "processed", "")	POSTGRES_EXEC_FAILURE	ClickHouse Alerting

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

Каждая строка лога, сбрасываемая компонентами ReceiptServer, ReceiptHandler и StoreCallbackWorker в поток os.Stdout, представляет собой структурированную запись в формате JSON Lines (Structured Logging через slog).

Сборщик Vector в реальном времени перехватывает данный поток, выполняет парсинг и осуществляет атомарную отгрузку пакетов в аналитическую СУБД ClickHouse DBMS. Эти данные служат базовым источником (Event Logs) для восстановления сквозных цепочек поставок и анализа узких мест бизнес-процессов с помощью аналитического движка PM4Py (Process Mining Engine).

При возникновении штатных событий или аварийных инфраструктурных инцидентов в ClickHouse фиксируются следующие типизированные метрики и метаданные:

1. Код (canonical_code)	2. Уровень (log_level)	3. Статусы (transport_statuses)	4. Получатель (error_target)	5. JSON для Фронтенда (ui_payload)	6. Метрики для ClickHouse (observability_json)
METADATA_MISSING	WARN	{"grpc": 16, "http": 401}	EXTERNAL_SYSTEM	{"error": "Авторизационные метаданные gRPC отсутствуют"}	{"metric": "grpc_metadata_err", "labels": {"layer": "transport"}}
AUTH_SERVICE_UNREACHABLE	ERROR	{"grpc": 13, "auth_alive": 0}	INTERNAL_SYSTEM	{"error": "Не удалось связаться с сервисом авторизации"}	{"metric": "auth_rpc_fail", "labels": {"client": "auth_service"}}
UNAUTHORIZED_STORE_TOKEN	WARN	{"grpc": 7, "http": 403}	EXTERNAL_SYSTEM	{"error": "Отказано в доступе: невалидный токен ритейлера"}	{"metric": "b2b_auth_fail", "labels": {"handler": "store_grpc"}}
POSTGRES_EXEC_FAILURE	ERROR	{"grpc": 13, "db_connected": 1}	INTERNAL_SYSTEM	{"error": "Ошибка сохранения данных транзакции в СУБД"}	{"metric": "db_write_err", "labels": {"table": "store_transactions"}}
KAFKA_POLL_FETCH_FAILURE	ERROR	{"kafka_connected": 0}	INFRASTRUCTURE	null	{"metric": "kafka_fetch_fail", "labels": {"client": "franz_go"}}
MINIO_DOWNLOAD_FAILURE	ERROR	{"s3_status": 404}	INFRASTRUCTURE	null	{"metric": "s3_not_found", "labels": {"bucket": "receipts"}}
WHISPER_STT_CRASH	WARN	{"grpc": 14, "gpu_util": 0}	INTERNAL_SYSTEM	null	{"metric": "ai_inference_fail", "labels": {"engine": "whisper"}}
OLLAMA_VISION_TIMEOUT	WARN	{"grpc": 14, "llama_status": 0, "duration_ms": 45000}	INTERNAL_SYSTEM	null	{"metric": "ai_timeout", "labels": {"model": "llama3.2_vision"}}
VOICE_METRIC_INVALID	INFO	{"kafka_produced": 1, "dlq": 1}	EXTERNAL_SYSTEM	{"warn": "Недостаточно данных в голосе, команда отправлена на модерацию"}	{"metric": "dlq_routing", "labels": {"reason": "missing_weight"}}
STORE_WEBHOOK_TIMEOUT	ERROR	{"http_status": 504, "webhook_ack": 0}	EXTERNAL_SYSTEM	{"error": "Превышено время ожидания ответа сервера ритейлера"}	{"metric": "partner_webhook_timeout", "labels": {"tx": "outbound"}}
PARTNER_WEBHOOK_BAD_STATUS	ERROR	{"http_status": 500, "webhook_ack": 0}	EXTERNAL_SYSTEM	{"error": "Сервер партнера вернул внутреннюю ошибку"}	{"metric": "webhook_http_err", "labels": {"partner": "magnum"}}

6.1 Регламент разбора инцидентов и аномалий ИИ-слоя

1. Многоканальный аудит по ключам трассировки: Поле observability_json принудительно обогащается сквозными идентификаторами request_id (для B2C gRPC-стримов) или correlation_id (для B2B партнерских вебхуков). При возникновении аномалии VOICE_METRIC_INVALID аналитик выполняет поиск в ClickHouse по audio_id, восстанавливая исходный текст (speechText) и причину отбраковки моделью Qwen (moderation_reason).
2. Сегрегация зон ответственности (error_target):
   • Инциденты с маркером INFRASTRUCTURE сигнализируют о деградации внутренних кластеров (Kafka, MinIO) и требуют немедленного реагирования DevOps-команды.
   • Ошибки EXTERNAL_SYSTEM фиксируют недоступность или некорректное поведение интеграционных шлюзов третьих сторон (серверы «Магнум», клиентские gRPC-таймауты) и переводятся на уровень клиентской поддержки.