CC Reporter

1. Зачем нужен отдельный сервис

Текущий подход с предварительной загрузкой данных на стороне интерфейса не масштабируется:

1. CSV собирается в памяти браузера, и на длинных периодах это заметно замедляет работу.
2. Если в результате больше 1000 строк, пользователю приходится многократно нажимать preload и fetch.
3. Долгие выборки в момент выгрузки создают лишнюю нагрузку на magista и fistful-magista.
4. Для полугодовых и годовых выгрузок нельзя гарантировать предсказуемое время подготовки в рамках SLA.
5. Требования задач к составу CSV (trx_id, currency, FX, finalized_time, корректный amount) требуют, чтобы отчет формировался в управляемом серверном процессе.

Вывод: длинные транзакционные отчеты нужно формировать не в клиентской части, а в отдельном асинхронном сервисе на стороне сервера. Да, это сложнее в реализации, но именно такой подход соответствует задаче годовых отчетов, изолирует нагрузку и снижает зависимость от внешних поисковых API.

2. Цель и рамки решения

Цель

Сделать сервис, пригодный для промышленной эксплуатации, который:

1. Асинхронно формирует CSV-отчеты по payments и withdrawals.
2. Поддерживает длинные периоды, включая годовые.
3. Работает на собственной модели чтения в PostgreSQL.
4. Выдает готовый файл по временной подписанной ссылке.

Что входит в первую версию

1. Жизненный цикл отчета: Create -> Pending -> Processing -> Created | Failed | TimedOut | Canceled | Expired.
2. Загрузка данных из Kafka в таблицы актуального состояния сущностей по платежам и выплатам.
3. API на Thrift для клиентской (фронт) части.
4. Фоновый обработчик и планировщик для построения отчетов.
5. Публикация CSV в S3-совместимое хранилище.

Что не входит в первую версию

1. Генерация CSV через поисковые API magista и fistful-magista в момент запроса.
2. Хранение полной истории событий по платежам и выплатам в БД CCR.
3. Внутренняя аутентификация сервиса: ее обеспечивает Wachter.
4. Агрегированные аналитические отчеты и произвольные аналитические срезы.

3. Загрузка из Kafka и модель чтения

3.1 Общие правила приема данных

1. CCR читает Kafka пакетами.
2. Каждый пакет обрабатывается в рамках одной транзакции БД.
3. Подтверждение чтения в Kafka выполняется только после успешного commit.
4. В базе хранится только актуальное состояние сущности, а не полный журнал событий.
5. Для каждой сущности используется upsert: запись создается при первом событии, а затем обновляется только если пришло более новое событие (по MachineEvent.eventId). Повторные и более старые события ничего не меняют.
6. Повторное чтение Kafka должно быть безопасным: оно не создает дубликаты и не откатывает состояние назад.

3.2 Почему выбран именно этот подход

1. Для длинных CSV-выгрузок важнее быстро читать актуальное состояние по фильтрам, чем хранить внутри CCR полную историю событий.
2. Модель актуального состояния уменьшает объем хранения и упрощает индексацию под реальные фильтры интерфейса.
3. Такой способ загрузки хорошо сочетается с асинхронной генерацией отчетов: данные сначала приводятся к удобному виду, а затем уже используются при построении файла.

3.3 Актуальное состояние payments

1. Таблица: payment_txn_current.
2. Бизнес-ключ: (invoice_id, payment_id).
3. Порядок событий в домене: MachineEvent.eventId.
4. Контракт upsert:
   • INSERT ... ON CONFLICT (invoice_id, payment_id) DO UPDATE
   • ... WHERE payment_txn_current.domain_event_id < EXCLUDED.domain_event_id
5. finalized_at заполняется только при первом терминальном статусе и не должен затираться последующим нетерминальным обогащением данных.

3.4 Актуальное состояние withdrawals

1. Таблица: withdrawal_txn_current.
2. Бизнес-ключ: withdrawal_id.
3. Порядок событий в домене: MachineEvent.eventId.
4. Контракт upsert:
   • INSERT ... ON CONFLICT (withdrawal_id) DO UPDATE
   • ... WHERE withdrawal_txn_current.domain_event_id < EXCLUDED.domain_event_id
5. Логика обновления такая же: по каждому бизнес-ключу сохраняется только самое новое состояние.

3.5 Повторное чтение и восстановление после сбоев

1. Для повторного чтения достаточно перезапустить процесс чтения с нужной политикой offset или использовать отдельную consumer group.
2. Идемпотентный upsert по бизнес-ключу и доменному event_id делает такое повторное чтение безопасным.

4. Требования

4.1 Бизнес-требования

1. Отчеты должны поддерживать payments и withdrawals.
2. Выгрузка строится на стороне бэкенда, а не в браузере.
3. Поддерживаются длинные периоды, включая 12 месяцев.
4. Должны работать фильтры интерфейса:
   • Provider
   • Shop
   • Wallet
   • Terminal
   • trx_id
   • Currency
   • Status
   • time range с точностью до часов и минут
5. Поиск по части имени и/или ID должен работать без учета регистра.
6. В CSV обязательно должны быть:
   • trx_id
   • currency
   • отдельные колонки date и time
   • finalized_time
   • блок валютного пересчета (original_amount, original_currency, converted_amount, exchange_rate_internal, provider_amount)
7. Форматирование amount должно корректно учитывать exponent валюты.
8. Тип отчета должен состоять из бизнес-типа (payments/withdrawals) и типа файла (csv).
9. Сейчас одному отчету соответствует ровно один итоговый файл.

4.2 Технические требования

1. Источник данных: Kafka, по тому же принципу, что и в текущих доменных сервисах.
2. Процесс чтения подтверждает offset только после успешного commit транзакции БД по всему пакету.
3. Повторное чтение topic должно быть безопасным и не приводить к дубликатам или откату актуального состояния.
4. CreateReport должен поддерживать идемпотентность по (created_by, idempotency_key). (created_by это идентификатор субьъекта из jwt токена который приходит в Wachter, idempotency_key uid с фронт энда)
5. Нужен управляемый механизм восстановления для повторных попыток и зависших заданий.
6. Нужны индексы под реальные фильтры интерфейса и длинные диапазоны.
7. При ошибке генерации не должно оставаться поврежденных локальных артефактов.

5. Архитектура и жизненный цикл отчета

5.1 Компоненты

1. Control Center Frontend
2. Wachter (аутентификация, авторизация и маршрутизация по JWT)
3. CC Reporter API (Thrift)
4. CC Reporter Schedulator: обработчик и планировщик
5. CC Reporter Kakfa Listener: процессы чтения Kafka
6. PostgreSQL (актуальное состояние и жизненный цикл отчетов)
7. Minio: S3-совместимое хранилище

5.2 Сквозной сценарий

1. Пользователь задает фильтры и нажимает Download report.
2. Клиентская часть вызывает CreateReport.
3. API проверяет соответствие пары report_type + file_type и ветки query, сохраняет report_job(status = pending) и возвращает report_id.
4. Обработчик забирает задание со статусом pending, переводит его в processing и увеличивает attempt.
5. Обработчик открывает отдельную транзакцию READ ONLY REPEATABLE READ для чтения данных отчета.
6. Сразу после открытия транзакции он фиксирует data_window_fixed_at = transaction_timestamp().
7. started_at отражает момент старта обработки задания worker-ом, а data_window_fixed_at отражает момент фиксации MVCC-снимка данных для всего отчета.
8. Внутри этой транзакции обработчик потоково читает актуальное состояние через серверный курсор, порциями.
9. CSV записывается во временный артефакт:
   • локальный временный файл или
   • временный ключ объекта в хранилище, который не публикуется во внешнем API
10. После полной записи файл хешируется, загружается в итоговый ключ объекта, затем создается report_file (одна запись на один report_job).
11. Только после успешной публикации файла задание завершается в статусе created.
12. Если возникает ошибка, обработчик удаляет временный файл или объект и не создает report_file.
13. При временной ошибке задание возвращается в pending с новым next_attempt_at.
14. Отдельный процесс контроля таймаутов переводит зависшие задания в timed_out.
15. Клиентская часть получает статусы через GetReports и GetReport.
16. Для скачивания клиентская часть вызывает GeneratePresignedUrl с file_id и получает ссылку, для которой TTL принудительно ограничивается на стороне сервиса.

5.3 Согласованность данных при генерации

CCR гарантирует два уровня фиксации:

1. Логическое окно данных задается фильтрами пользователя (requested_time_from, requested_time_to).
2. Физическая согласованность чтения обеспечивается транзакцией READ ONLY REPEATABLE READ на все время построения файла.

Это означает:

1. Во время одной генерации параллельные обновления актуального состояния не попадут в тот же отчет частично.
2. Отчет представляет собой согласованный снимок на момент начала processing, а не на момент вызова CreateReport.

6. Связанные спецификации

1. SQL DDL
2. Thrift API
3. Формат CSV

7. Сценарий в админке

1. На страницах payments и withdrawals пользователь задает фильтры и нажимает Download report.
2. Интерфейс показывает встроенный статус или всплывающее уведомление: отчет поставлен в очередь.
3. Во вкладке Reports отображается таблица:
   • type
   • period
   • created_at
   • status
   • rows_count
   • actions
4. Для period интерфейс использует query.time_range, который возвращается в Report.
5. Поддерживаются статусы:
   • pending
   • processing
   • created
   • failed
   • timed_out (таймаут так чисто показать определенность ошибки зависшие таски (строились настолько долго что вышли за рамки sla ожиданий) , это пробрасывается дофронта, там выводится таймаут и кнопка retry)
   • canceled
   • expired
6. Для created доступна кнопка Download.
7. Для pending и processing доступна Cancel.
8. Для failed и timed_out доступна Retry через повторный CreateReport с тем же query.
9. Список отчетов загружается постранично через continuation_token.

8. Риски и меры снижения

8.1 Рост объема хранения

Риск:

1. Со временем сильно вырастут и таблицы актуального состояния, и архив файлов.

Меры снижения:

1. Настроить сроки хранения для report_job и report_file.
2. Настроить политику жизненного цикла объектов в S3 с ограничением по TTL.
3. Регулярно выполнять VACUUM/ANALYZE.
4. Контролировать разрастание таблиц и смотреть EXPLAIN по самым тяжелым запросам.
5. Ограничить количество одновременно запущенных больших отчетов на одного пользователя.

8.2 Долгая генерация больших отчетов

Риск:

1. Годовые отчеты могут строиться несколько минут и задерживать всю очередь.

Меры снижения:

1. Использовать потоковую запись без накопления всего файла в памяти.
2. Читать данные через серверный курсор, порциями.
3. Выделить отдельный пул обработчиков для тяжелых заданий.
4. Настроить политику повторных попыток через next_attempt_at.
5. Отдельно отслеживать и переводить в таймаут зависшие задания со статусом processing.

8.3 Позднее дозаполнение (trx_id, FX, данные провайдера)

Риск:

1. Часть полей может приходить не в первом событии.

Меры снижения:

1. Модель актуального состояния допускает частичное заполнение с последующим объединением данных.
2. Допустимость nullable-полей должна быть явно зафиксирована в контракте.
3. Отчет всегда строится по последнему актуальному состоянию на момент data_window_fixed_at.

8.4 Долгие транзакции REPEATABLE READ

Риск:

1. Долгая транзакция чтения может слишком долго удерживать MVCC-снимок и увеличивать нагрузку на autovacuum.

Меры снижения:

1. Ограничить число одновременно строящихся длинных отчетов.
2. Использовать только потоковое чтение, без помещения всего набора результатов в память.
3. Ввести отдельный таймаут на построение отчета.
4. Проводить нагрузочное тестирование именно для длительных сценариев с REPEATABLE READ.

8.5 Ограничение модели актуального состояния

Риск:

1. Если задание долго ждет в очереди, отчет отражает согласованное состояние на момент начала processing, а не на момент нажатия кнопки.

Меры снижения:

1. Минимизировать задержку до старта обработчика.
2. Не держать очередь длинной без необходимости.
3. Если бизнесу критично строгое состояние именно на момент запроса, вынести это в отдельное следующее требование.

9. Неуточненные вопросы в текущем пакете

1. Источник и обязательность полей shop_name, wallet_name, provider_name, terminal_name пока окончательно не зафиксированы. На текущем этапе их нужно рассматривать как возможные поля для денормализованного поиска.
2. provider_currency пока не является окончательно подтвержденным требованием. Сейчас это рабочий вариант, который снимает неоднозначность provider_amount, если сумма провайдера может быть в валюте, отличной от currency.
3. Для trx_id, FX-полей и полей, связанных с провайдером, может понадобиться объединение данных из нескольких типов Kafka-событий. Окончательное соответствие между событиями и полями нужно подтвердить до реализации загрузки.
4. В трех модельных документах эти поля могут встречаться как текущий проектный вариант. На ревью их нужно воспринимать как предмет согласования, а не как уже утвержденную часть решения.

10. Безопасность

1. Аутентификация и авторизация внутри CCR не реализуются: это зона ответственности Wachter.
2. Доступ к файлам дается только через presigned URL с ограниченным TTL.
3. После выдачи presigned URL сервис уже не контролирует дальнейшее распространение ссылки.

Вопросы для согласования правил доступа:

1. Какой максимальный TTL допустим по внутренней политике.
2. Нужен ли режим однократного скачивания.
3. Нужен ли доп аудит помимо предложенного аудита скачиваний report_audit_event (см sql).

Ответы: 10-15 минут кажется будет достаточно чтобы сотрудник выгрузил себе файл можно включить. не помешает даже при минимальном ttl достаточно логировать генерацию и дальнейшие все обращения по ссылке. (userid, ttl, ip, время при генерации. ip, время, ua при обращении к ссылке. это надо на s3 вероятно настроить, может с помощью девопс команды)

11. План реализации

1. Этап 1: процессы чтения Kafka, таблицы актуального состояния и идемпотентный upsert по domain_event_id.
2. Этап 2: API на Thrift и таблицы report_job, report_file, report_audit_event.
3. Этап 3: обработчик и планировщик, повторные попытки, обработка таймаутов и атомарная публикация файла.
4. Этап 4: интеграция с клиентской частью (Reports, статусы, повторный запуск, пагинация).
5. Этап 5: нагрузочные тесты для длинных периодов, включая длительное чтение в REPEATABLE READ.

12. Критерии готовности

1. Клиентская часть больше не делает предварительную загрузку по 1000 строк для CSV.
2. payments и withdrawals формируются через асинхронный серверный жизненный цикл.
3. Повторное чтение Kafka не создает дубликаты и не откатывает актуальное состояние назад.
4. Во время генерации параллельные обновления не смешиваются в одном отчете.
5. Поврежденные временные артефакты не публикуются наружу.
6. CSV соответствует обязательным полям по требованиям.