WebSocket-стрим уведомлений: /api/v1/fridge/ws/push-stream/{home_group_id}

Документация API: Спецификация асинхронного полнодуплексного канала синхронизации стейта Flutter в реальном времени

Published

July 2, 2026

1 Функциональное назначение

WebSocket-эндпоинт представляет собой постоянное, низкозатратное полнодуплексное сетевое соединение (двусторонний стрим) между мобильным приложением Flutter и бэкенд-шлюзом FastAPI. Этот канал выполняет роль «нервной системы» распределенной архитектуры, заменяя неэффективные и тяжелые HTTP Short Poll-опросы.

Эндпоинт решает две ключевые задачи: 1. Мгновенная реакция UI на фоновые ИИ-события (Reactivity): Когда фоновый асинхронный воркер завершает OCR-распознавание чека (Статья 4) или внешняя торговая сеть Magnum присылает фискальный вебхук (Статья 5), бэкенд мгновенно пушит событие в этот канал. Экран ожидания (лоадер) на смартфоне пользователя мгновенно исчезает, открывая актуальный экран модерации. 2. Синхронизация локального кэша склада (State Management Synchronizer): При возникновении любых транзакционных изменений баланса (утилизация мусора в Waste, списание обеда в Consume, ручное слияние партий в Merge), совершенных одним членом семьи, бэкенд рассылает команду "REFRESH_STOCK" всем остальным участникам группы home_group_id. Стейт-менеджер Flutter (Bloc/Provider) точечно обновляет экран склада без перезапуска приложения.

2 Протокол взаимодействия (WebSocket Контракт)

  • Протокол: WS (при локальной разработке) / WSS (WebSocket Secure — в Live-контуре)
  • Маршрут: /api/v1/fridge/ws/push-stream/{home_group_id}
  • Формат передачи: text/json (Двусторонние текстовые фреймы)

2.1 Спецификация параметров пути (Path Parameters)

Параметр Тип Обязательный Описание Пример значения
home_group_id String Да Идентификатор группы, определяющий скоуп бродкаста уведомлений group-772-alpha

2.2 Спецификация заголовков при установлении соединения (Handshake Headers)

Установление WebSocket-соединения начинается со стандартного HTTP-запроса (Upgrade Handshake), поэтому оно защищено общим протоколом безопасности JWT.

Заголовок Обязательный Описание Пример значения
Upgrade Да Системный заголовок перехода на протокол WebSocket websocket
Connection Да Системный заголовок переключения канала Upgrade
Authorization Да Токен авторизации (Access Token). Бэкенд валидирует подпись перед Upgrade. Bearer eyJhbGciOiJIUzI1Ni...

3 Схема обработки и трансляции событий (Mermaid)

На диаграмме представлена логика работы WebSocket-менеджера. Когда фоновый ИИ-воркер или внешняя шина данных (Magnum) совершают транзакционные изменения, брокер Apache Kafka доставляет событие в шлюз, который через резидентный пул соединений (In-Memory Connection Pool) пушит управляющий фрейм во Flutter.

sequenceDiagram
    autonumber
    actor User as Flutter Клиент (Семья Альфа)
    participant API as Бэкенд-шлюз (FastAPI WS Manager)
    participant KAFKA as Брокер событий (Kafka)
    participant WORKER as Фоновый ИИ-Воркер (OCR / Magnum)

    %% ЭТАП 1: УСТАНОВЛЕНИЕ СОЕДИНЕНИЯ (HANDSHAKE)
    Note over User, API: Этап 1: HTTP Upgrade Handshake & Регистрация сессии
    User->>API: HTTP GET /ws/push-stream/group-772-alpha (Upgrade: websocket + JWT)
    activate API
    note over API: Валидация токена. Upgrade до WSS успешлен.
    API->>API: Добавление соединения в active_connections["group-772-alpha"]
    API-->>User: Подтверждение открытия постоянного стрима (Connection Established)
    
    %% ЭТАП 2: АСИНХРОННЫЙ ПУШ ИЗ КАНАЛА
    Note over WORKER, KAFKA: Этап 2: Асинхронное событие в экосистеме
    WORKER->>KAFKA: Публикация события ("RECEIPT_OCR_READY", payload)
    KAFKA->>API: Потоковое чтение события (Kafka Consumer Loop)
    
    note over API: Шаг 9: Локализация пула для group-772-alpha
    
    loop По всем активным смартфонам внутри группы group-772-alpha
        API->>User: Отправка текстового JSON фрейма (Push Notification Broadcast)
    end
    deactivate API
    note over User: Flutter Bloc ловит эвент и мгновенно перерисовывает экран склада

4 Расшифровка шагов

  • Шаг 1 (User -> API): При старте мобильного приложения Flutter-клиент инициирует постоянное сетевое подключение, отправляя запрос HTTP GET /api/v1/fridge/ws/push-stream/{home_group_id}. В заголовках передаются системные директивы Upgrade: websocket и Connection: Upgrade, а также Access Token.
  • Шаг 2 (API -> API): Бэкенд-шлюз FastAPI перехватывает запрос на этапе Handshake, извлекает и валидирует JWT-токен. Из клеймов токена вычитывается home_group_id. Если токен просрочен или подпись невалидна, шлюз отклоняет соединение со статусом HTTP 401 Unauthorized, не доводя канал до состояния сокета.
  • Шаг 3 (API -> API): После успешной HTTP-валидации происходит процедура Upgrade — стандартное TCP-соединение переключается в режим двустороннего асинхронного WebSocket-канала.
  • Шаг 4 (API -> API): Экземпляр класса ConnectionManager на бэкенде регистрирует сетевой дескриптор сокета и помещает его во внутреннюю оперативную память (In-Memory Shared Dict) под ключом текущей группы: active_connections["group-772-alpha"].append(websocket). Это позволяет масштабировать бродкаст на несколько устройств внутри одной семьи (например, муж и жена увидят обновление склада одновременно).
  • Шаг 5 (API -> User): Сервер отправляет клиенту первый приветственный системный фрейм, подтверждая успешный вход в шину асинхронных уведомлений реального времени. Локальный стейт-менеджер Flutter переводит сокет-репозиторий в статус Connected.
  • Шаг 6 (WORKER -> KAFKA): Независимо от сокет-канала, в фоновом контуре происходят бизнес-процессы. Например, ИИ-воркер завершил тяжелое OCR-распознавание чека (Статья 4) или внешняя шина данных Magnum зачислила продукты через B2B-вебхук (Статья 5). Воркер публикует событие в брокер Apache Kafka.
  • Шаг 7 (KAFKA -> API): На шлюзе FastAPI параллельно запущен фоновый асинхронный демон Kafka Consumer Loop. Он непрерывно слушает топики инфраструктурных изменений и вычитывает пришедший JSON-пакет.
  • Шаг 8 (API -> API): Из тела сообщения Kafka бэкенд извлекает целевой идентификатор получателя home_group_id (например, "group-772-alpha").
  • Шаг 9 (API -> API): ConnectionManager обращается к словарю активных сессий и изолирует пул сокетов, подключенных прямо сейчас от этой семьи. Все остальные 100 000 сокетов других семей полностью игнорируются, обеспечивая жесткую изоляцию данных.
  • Шаг 10 (API -> User): Сервер запускает цикл и транслирует текстовый JSON-фрейм во все активные дескрипторы сокетов этой группы методом await websocket.send_text(json_payload).
  • Шаг 11 (User -> User): Приемник сокета во Flutter-приложении перехватывает входящий эвент. Срабатывает триггер стейт-менеджера (Bloc/Cubit): интерфейс реагирует мгновенно, гасит лоадеры ожидания и выполняет точечную UI-команду (например, перенаправляет пользователя на экран аудита чека или обновляет цифры веса продуктов в холодильнике).

5 Спецификация структуры фреймов (Payload) и Fail-Safe стратегий

5.1 1. Спецификация структуры пакетов сервера (Server-to-Client Frames)

Каждое сообщение, передаваемое сервером в WebSocket-канал, имеет строго типизированную структуру. Поле event_type определяет бизнес-контекст, а ui_action — директиву для стейт-менеджера Flutter (какой метод вызвать или на какой экран переключить пользователя).

5.1.1 Сценарий А: Сигнал о готовности OCR-распознавания чека (Контур Статьи 4)

Отправляется фоновым ИИ-воркером, когда тяжелое распознавание фотографии чека завершено. Flutter обязан убрать экран ожидания и перенаправить пользователя на аудит.

{
  "event_type": "RECEIPT_OCR_READY",
  "timestamp": "2026-07-02T02:05:00Z",
  "data": {
    "draft_id": "9b1deb4d-3b7d-4bad-9bdd-2b0d7b3dcb6d",
    "ui_action": "REDIRECT_TO_RECEIPT_AUDIT_SCREEN",
    "fetch_url": "/api/v1/fridge/receipt-draft/9b1deb4d-3b7d-4bad-9bdd-2b0d7b3dcb6d/items"
  }
}

5.1.2 Сценарий Б: Мгновенное обновление остатков (Контуры Потребления, Готовки, Слияния или Magnum)

Отправляется, когда баланс продуктов в холодильнике изменился (например, муж дома съел йогурт, у жены на смартфоне склад должен обновиться без перезапуска приложения).

{
  "event_type": "REFRESH_STOCK",
  "timestamp": "2026-07-02T02:05:12Z",
  "data": {
    "reason": "TRANSACTIONAL_MUTATION_CONFIRMED",
    "ui_action": "TRIGGER_BLOC_EVENT_FETCH_INVENTORY",
    "fetch_url": "/api/v1/fridge/inventory"
  }
}

5.2 2. Спецификация структуры Pydantic-модели фрейма (FastAPI)

На стороне бэкенда каждый WebSocket-фрейм перед отправкой проходит строгую типизацию с помощью базовой схемы Pydantic.

from pydantic import BaseModel, Field
from datetime import datetime
from typing import Dict, Any, Optional

class WebSocketServerFrame(BaseModel):
    event_type: str = Field(..., description="Тип бизнес-события экосистемы")
    timestamp: datetime = Field(default_factory=datetime.utcnow, description="Время генерации кадра")
    data: Dict[str, Any] = Field(..., description="Контекстный payload и UI-директивы для Flutter")

5.3 3. Fail-Safe стратегии: Обработка обрывов связи и Heartbeat (Ping/Pong)

Поскольку мобильные устройства постоянно меняют точки доступа (переключение с Wi-Fi на LTE в лифте или метро), WebSocket-канал подвержен частым обрывам связи (Dead Connections). Для обеспечения отказоустойчивости закладываются два механизма:

5.3.1 1. Механизм Heartbeat (Пинг/Понг на Шаге 5)

Для предотвращения скрытого закрытия сокета сетевыми провайдерами (таймаут неактивности NAT-маршрутизаторов) шлюз FastAPI каждые 30 секунд отправляет клиенту пустой системный фрейм PING. * Если мобильное приложение Flutter не возвращает текстовый ответ PONG в течение 5 секунд, шлюз считает соединение мертвым, принудительно вызывает метод websocket.close(), вычищает дескриптор из оперативной памяти active_connections и освобождает ресурсы сервера.

5.3.2 2. Стратегия Автоматического Переподключения (Flutter Auto-Reconnect)

При фиксации обрыва связи сокет-репозиторий Flutter переходит в статус Disconnected. Включается алгоритм Экспоненциального бэкграунд-ретрая (Exponential Backoff): * Приложение пытается восстановить соединение (выполнить HTTP Upgrade Handshake) через 1 секунду. * Если сервер недоступен (ошибка 503) — следующая попытка через 2 секунды, затем через 4, 8, 16, вплоть до максимального интервала в 32 секунды. * Как только связь восстанавливается, Flutter-клиент отправляет запрос, бэкенд заново регистрирует сессию в пуле, и асинхронный стрим уведомлений возобновляется без потери стейта интерфейса.