# Action Service — Полная карта проекта

Сервис принимает события из Kafka и HTTP, обогащает их (офис, сотрудник, бизнес-правила), формирует `Action` и отправляет в выходной Kafka-топик. Параллельно поддерживает кеши офисов и сотрудников, а также очередь отложенных сообщений.

## 1) Назначение сервиса

`action-service` решает три основные задачи:

- обработка событий движения товаров/действий сотрудников и генерация унифицированных `Action`;
- маршрутизация в Kafka producer после фильтрации и нормализации;
- поддержка инфраструктурных данных (офисы, `wb_user_id -> employee_id`, delayed queue, партиции БД).

## 2) Среда и конфигурация

Источник: `config/config.yml`.

### Основные внешние зависимости

- `PostgreSQL action-db` (схема `action`): действия, delayed queue, `employee_data`, `courier_list`, служебные функции.
- `PostgreSQL office-db`: данные офисов.
- `Redis`: кеш связок `employee_data`.
- `Kafka`: входные consumer topics + выходной producer topic.
- Внешние API:
  - `Hiring API` (актуальные связки сотрудников);
  - `Auth API` (токен для Hiring API);
  - `Admtrf API` (Список офисов из админ-тарификатора).

## 3) Слойная архитектура

- `internal/app`: сборка и запуск приложения (`http`, `kafka-consumer`, `cron`).
- `internal/provider`: wire-up зависимостей, миграции, создание сервисов.
- `internal/facade/httpserver`: HTTP-роуты/хендлеры.
- `internal/facade/kafka-consumer`: чтение батчей Kafka + вызов доменных сервисов.
- `internal/facade/cron`: периодические задачи.
- `internal/service`: бизнес-логика.
- `internal/integration/*`: PostgreSQL/Redis/HTTP/Kafka адаптеры.

## 4) Потоки данных (E2E)

### 4.1 Positions

`Kafka (positions topic)` -> `positionsHandler.ServeMessages` -> `positionsService.ProcessMessages` ->
`filters chain` -> `strategy by status code` -> `produceService.Send` -> `Kafka producer`

Файлы:
- [internal/facade/kafka-consumer/positions.go](/home/pushkin.andrey9/projects/action-service/internal/facade/kafka-consumer/positions.go)
- [internal/service/positions/positions.go](/home/pushkin.andrey9/projects/action-service/internal/service/positions/positions.go)
- [internal/service/positions/filters.go](/home/pushkin.andrey9/projects/action-service/internal/service/positions/filters.go)
- [internal/service/positions/actions.go](/home/pushkin.andrey9/projects/action-service/internal/service/positions/actions.go)
- [internal/service/produce_service.go](/home/pushkin.andrey9/projects/action-service/internal/service/produce_service.go)

Дополнительно:

- если офис не найден: сообщение уходит в delayed queue (`positions_all_queue`);
- cron/HTTP может отдельно запускать обработку delayed queue.

Бизнес-описание:
1. Сервис принимает сырые события изменения статусов по товарам на ПВЗ.
2. Для каждого события проверяет применимость к бизнес-процессу (тип точки, сотрудник, валюта, тип оплаты, исключенные `nm_id`).
3. Подбирает бизнес-стратегию по `status_id` и конвертирует событие в один или несколько `Action` с целевыми кодами (`210`, `230`, `259`, `260`, `270`, `280`, и др.).
4. Если офис временно неизвестен, сообщение не теряется: переносится в delayed-очередь для повторной обработки.
5. Готовые `Action` проходят общий pre-send фильтр и отправляются в `mngtrf-action-out`.

Неочевидные места:
- отсутствие сотрудника в кеше не останавливает поток: используется fallback `employee_id = -wb_user_id`, используется, чтобы не потерять действие при расчетах и мержится в другом сервисе; 
- неизвестный `status_id` скипается без падения всего батча.

### 4.2 Courier Goods

`Kafka (courier topic)` -> `courierGoodsHandler.ServeMessages` -> `courierGoodsService.ProcessMessages` -> `produceService.Send` -> `Kafka producer`

Файлы:
- [internal/facade/kafka-consumer/courier-goods.go](/home/pushkin.andrey9/projects/action-service/internal/facade/kafka-consumer/courier-goods.go)
- [internal/service/courier-goods/courier-goods.go](/home/pushkin.andrey9/projects/action-service/internal/service/courier-goods/courier-goods.go)
- [internal/service/produce_service.go](/home/pushkin.andrey9/projects/action-service/internal/service/produce_service.go)

Бизнес-описание:
1. Сервис принимает курьерские операции и маппит их в действия тарификатора.
2. По `ctx` определяет бизнес-тип действия: продажа курьером (`330`) или приемка вещей курьером (`370`).
3. Валидирует валюту, нормализует код валюты (`810 -> 643`), обогащает событие офисом и локальным временем офиса.
4. Пропускает неподходящие точки выдачи и отправляет только валидные действия в общий producer pipeline.

Неочевидные места:
- неизвестные типы `ctx` и пустой `currency_code` тихо скипаются как бизнес-фильтр;
- в текущем коде обрабатывается только `type_point=34`

### 4.3 Motivation / Internship

`Kafka (motivation topic)` -> `motivationHandler.ServeMessages` ->
`motivationService.ProcessMotivations/ProcessInternships` -> `produceService.Send` -> `Kafka producer`

Файлы:
- [internal/facade/kafka-consumer/motivation.go](/home/pushkin.andrey9/projects/action-service/internal/facade/kafka-consumer/motivation.go)
- [internal/service/motivation/motivation.go](/home/pushkin.andrey9/projects/action-service/internal/service/motivation/motivation.go)
- [internal/service/produce_service.go](/home/pushkin.andrey9/projects/action-service/internal/service/produce_service.go)

Бизнес-описание:
1. Поток разделяется на две ветки по ключу сообщения: `motivation` и `internship` - QR-код и стажировка
2. Для обеих веток сервис проверяет, что офис существует и его `type_point` разрешен для мотивационного процесса.
3. Ветка `motivation`:[office.go](internal/service/office.go) нормализует дату в TZ офиса и отправляет действие как мотивационное событие.
4. Ветка `internship`: собирает payload по стажировке, определяет код действия по роли (стажер/наставник) и отправляет.

Неочевидные места:
- некорректный тип сотрудника в стажировке (`EmployeeType`) завершает обработку ошибкой;
- часть полей в мотивационных сообщениях переиспользуется/перезаписывается под формат action (например `SrID` как дата).

### 4.4 WB Track

`Kafka (wb-track topic)` -> `wbTrackHandler.ServeMessages` ->
`wbTrackService.ProcessWbTrackActions` -> `wbTrackService.SendActions` -> `produceService.Send` -> `Kafka producer`

Файлы:
- [internal/facade/kafka-consumer/wb-track.go](/home/pushkin.andrey9/projects/action-service/internal/facade/kafka-consumer/wb-track.go)
- [internal/service/wb-track/wb-track.go](/home/pushkin.andrey9/projects/action-service/internal/service/wb-track/wb-track.go)
- [internal/service/produce_service.go](/home/pushkin.andrey9/projects/action-service/internal/service/produce_service.go)

Бизнес-описание:
1. Сервис принимает wb-track события по упаковке/перепродаже и сначала отбрасывает нерелевантные офисы.
2. Определяет тип итогового действия: упаковка (`510`) или перепродажа (`512`).
3. Для упаковки дополнительно проверяет, что `nm_id` входит в список валидных размеров/пакетов.
4. Пытается резолвить сотрудника по `wb_user_id`, при отсутствии использует fallback.
5. Готовые действия отправляются через общий producer pipeline.

Неочевидные места:
- список валидных package IDs хранится в памяти сервиса и обновляется отдельным HTTP-методом;
- при проблеме с lookup сотрудника поток не падает, а продолжает с отрицательным `employee_id`.

### 4.5 Gift Certificates

`Kafka (gift-certificates topic)` -> `giftCertificatesHandler.ServeMessages` ->
`GiftCertificatesService.ProcessGiftCertificateIssues` -> `giftCertificatesRepo.SaveGiftCertificateIssues`

Файлы:
- [internal/facade/kafka-consumer/gift_certificate.go](/home/pushkin.andrey9/projects/action-service/internal/facade/kafka-consumer/gift_certificate.go)
- [internal/service/gift_certificates.go](/home/pushkin.andrey9/projects/action-service/internal/service/gift_certificates.go)
- [internal/integration/postgres/gift_certificate_db_repo.go](/home/pushkin.andrey9/projects/action-service/internal/integration/postgres/gift_certificate_db_repo.go)

Бизнес-описание:
1. Сервис принимает события выдачи сертификатов, валидирует DTO и конвертирует в доменную модель.
2. Обогащает запись данными офиса (`type_point`, `country_code`) из кеша офисов.
3. Оставляет только брендированные ПВЗ и сохраняет итоговые записи в БД.

Неочевидные места:
- если офис не найден в кеше, запись будет пропущена;
- если после фильтрации не осталось событий, поток завершится как успешный без записи.

## 5) Kafka карта

### Consumers (из `config/config.yml`)

- `positions-consumer.topic`: `positions_all`
- `courier-consumer.topic`: `courier_goods_movements`
- `motivation-consumer.topic`: `mngtrf_motivation_out`
- `wb-track-consumer.topic`: `pvz_track_firm_offices_rewards`
- `gift-certificates-consumer.topic`: `delivery_gift_certificates`

### Producer

- `action-producer.topic`: `mngtrf-action-out`

### Источники и бизнес-смысл входных топиков

- `positions_all`: события по товарам/статусам на ПВЗ; в обработке используется сценарий обогащения сотрудником и генерации action-кодов.
- `courier_goods_movements`: события по операциям курьеров.
- `mngtrf_motivation_out`: события мотивации по qr-коду и стажировок.
- `pvz_track_firm_offices_rewards`: wb-track события для упаковки/продажи.
- `delivery_gift_certificates`: события выдачи подарочных сертификатов.

Ссылки:
- `positions_all`: `https://gitlab.wildberries.ru/ext-delivery/ext-delivery/syncer/-/blob/master/docs/positions_all.md`
- `courier_goods_movements`: `http://pvz-wiki.wb.ru/ru/kafka/courier_goods_movements`

## 5.1) Модель выходного события

Целевой формат публикуется в `mngtrf-action-out` и содержит поля:

- `action`
- `shk_id`
- `office_id`
- `employee_id`
- `office_dt`
- `srid`
- `nm_id`
- `type_point`
- `payload` (например: `currency_code`, `price`, `arrival_dt`)

Файл с примерами формирования:
- [internal/domain/action.go](/home/pushkin.andrey9/projects/action-service/internal/domain/action.go)
- [internal/service/positions/actions.go](/home/pushkin.andrey9/projects/action-service/internal/service/positions/actions.go)
- [internal/service/courier-goods/courier-goods.go](/home/pushkin.andrey9/projects/action-service/internal/service/courier-goods/courier-goods.go)
- [internal/service/wb-track/wb-track.go](/home/pushkin.andrey9/projects/action-service/internal/service/wb-track/wb-track.go)

## 6) HTTP карта

Реальные роуты регистрируются под префиксом `/api/v1/internal`.

### User

- `POST /api/v1/internal/action/reload-user-cache`
- `POST /api/v1/internal/action/update-employee-data`

### WB Track

- `POST /api/v1/internal/wb-track/update-parcel-ids`

### Cron ручные триггеры

- `POST /api/v1/internal/cron/update-office-cache`
- `POST /api/v1/internal/cron/process-delayed-messages`

### Технические

- `/metrics`
- `/api/version`
- `/api/utc`
- `/debug/pprof/*`
- `/docs/*`

## 7) Cron карта

Из `config/config.yml`:

- `update-employee-data`: обновление `employee_data` из Hiring API.
- `make-employee-data-partitions`: создание партиций `employee_data`.
- `reload-user-cache`: перезагрузка Redis-кеша `wb_user_id -> employee_id`.
- `process-delayed-messages`: обработка очереди отложенных positions-сообщений.
- `update-office-cache`: обновление кеша офисов.
- `delete-old-messages`: удаление старых delayed-сообщений.

## 8) Карта сервисов и методов

## 8.1 `internal/service/user.go`

- `ReloadUserCache`
Цепочки:
`HTTP /action/reload-user-cache -> handler -> service`
`cron.reloadUserCache -> service`
`provider.New -> service`
Суть: читает latest-срез `employee_data` из БД и перезаписывает Redis map.
Шаги внутри метода:
1. Запускает таймер операции и `defer` для метрик `Count/Observe`.
2. Вызывает `employeeRepo.GetAllEmployeeData` (PostgreSQL).
3. Преобразует список в map через `EmployeeByUser()` (`wb_user_id -> employee_id`).
4. Вызывает `employeeCache.SetEmployeeDataMap` (Redis hash `employee_data`).
5. Возвращает ошибку с контекстом конкретного шага.
Неочевидное:
- stale поля в Redis могут оставаться (нет очистки ключей);
- при ошибке `SetEmployeeDataMap` ошибка в метрике может считаться дважды.

- `UpdateEmployeeDataInRepo`
Цепочки:
`HTTP /action/update-employee-data -> handler -> service`
`cron.updateEmployeeDataFromAPI -> service`
Суть: тянет сотрудников из Hiring API и вставляет в `employee_data`.
Шаги внутри метода:
1. Запускает таймер и отложенную отправку метрик.
2. Запрашивает список сотрудников через `hiringAPI.GetManagersList`.
3. Передает список в `employeeRepo.InsertEmployeeData`.
4. Возвращает ошибку API или БД с источником проблемы.
Неочевидное:
- кеш не обновляется автоматически, нужен отдельный `ReloadUserCache`.
- по бизнес-смыслу это и есть механизм актуализации `employee_id` для downstream сообщений (описано в README как замена/актуализация сотрудника через Hiring API).
- вставка идет как snapshot за день (`dt=now`), т.е. актуальность lookup зависит от последующего прогрева кеша.

- `CreateEmployeeDataPartitionsForPeriod`
Цепочки:
`app.Run (startup goroutine) -> service`
`cron.makeEmployeeDataPartitions -> service`
Суть: вызывает SQL-функцию создания партиций за период.
Шаги внутри метода:
1. Передает даты периода в `employeeRepo.CreateEmployeeDataPartitionsForPeriod`.
2. При ошибке возвращает fail и отмечает ошибочную метрику.
3. При успехе фиксирует success и latency.
Неочевидное:
- в отличие от других методов здесь latency пишется только в success-ветке;
- на старте приложения вызывается в отдельной goroutine, и ошибка может остановить ранний lifecycle.

- `GetEmployeeID`
Цепочки:
`positionsService.ProcessMessages -> userService.GetEmployeeID`
`wbTrackService.ProcessWbTrackActions -> userService.GetEmployeeID`
Суть: lookup `employee_id` из Redis по `wb_user_id`.
Шаги внутри метода:
1. Читает поле `wb_user_id` из Redis hash `employee_data`.
2. Если ключа нет, возвращает бизнес-ошибку `domain.ErrNotFound`.
3. Если ошибка инфраструктурная (Redis/parse), возвращает техническую ошибку.
4. При успехе возвращает `employee_id`.
Неочевидное:
- `ErrNotFound` обрабатывается как бизнес-кейс, не как hard-fail.

## 8.2 `internal/service/office.go`

- `UpdateOffices`
Цепочки:
`HTTP /cron/update-office-cache -> handler -> service`
`cron.updateOfficeCache -> service`
Суть: загружает офисы из Admtrf API и обновляет office-cache.
Шаги внутри метода:
1. Вызывает `admtrfAPI.GetOfficeInfo`.
2. Конвертирует список в map (`office_id -> Office`) через `ToOfficeMap`.
3. Полностью обновляет in-memory cache `officeCache.SetMap`.
4. Возвращает ошибку при проблемах внешнего API.
Неочевидное:
- `officeRepo` не используется в этом сценарии, источник истины для прогрева — только Admtrf API.

- `GetOffice`
Цепочки:
`positions/motivation/wb-track/courier-goods -> officeService.GetOffice`
Суть: читает офис из cache, иначе `ErrNotFound`.
Шаги внутри метода:
1. Валидирует `officeID > 0`.
2. Ищет офис в кэше.
3. При отсутствии возвращает `domain.ErrNotFound`.
Неочевидное:
- fallback к БД/API внутри метода нет.
- метод является критической точкой для 4 потоков (`positions`, `courier-goods`, `motivation`, `wb-track`).

## 8.3 `internal/service/positions/positions.go`

- `ProcessMessages`
Цепочки:
`Kafka positions -> positionsHandler -> positionsService.ProcessMessages`
`ProcessDelayedMessages -> ProcessMessages`
Суть:
- достает офис;
- прогоняет фильтры;
- резолвит сотрудника;
- выбирает стратегию по `StatusCode`;
- отправляет actions;
- сохраняет delayed сообщения.
Детально по шагам:
1. Для каждого сообщения запрашивает офис через `officeService.GetOffice`.
2. Если офис не найден (`ErrNotFound`), кладет сообщение в delayed-массив и продолжает цикл.
3. Прогоняет цепочку фильтров (`office -> employee -> currency -> nm_id -> payment_type`).
4. Резолвит `employee_id` через `userService.GetEmployeeID`; при `ErrNotFound` ставит `-wb_user_id`.
5. Выбирает стратегию по `status_id` и строит 0..N actions.
6. После цикла отправляет сформированные actions в producer.
7. Сохраняет отложенные сообщения в queue repo.
Неочевидное:
- при `userService.ErrNotFound` используется fallback `employeeID = -wb_user_id`;
- неизвестные `StatusCode` скипаются, батч не падает.
- ошибка отправки actions останавливает обработку до сохранения delayed сообщений.
- если стратегия вернула пустой набор actions, это считается штатным сценарием.

- `ProcessDelayedMessages`
Цепочки:
`HTTP /cron/process-delayed-messages -> handler -> service`
`cron.processDelayedMessages -> service`
Суть: читает delayed queue, повторно обрабатывает, удаляет старые line id.
Шаги внутри метода:
1. Читает набор отложенных сообщений `queueRepo.GetMessages`.
2. Преобразует `map[lineID]message` в два списка: `lineIDs` и `messages`.
3. Повторно запускает `ProcessMessages` для полученных сообщений.
4. После успешной обработки удаляет исходные строки `DeleteMessagesByIDs`.
Неочевидное:
- если часть сообщений снова откладывается, они попадают как новые записи с новыми id.

- `DeleteOldMessages`
Цепочка:
`cron.deleteOldMessages -> positionsService.DeleteOldMessages`
Суть: чистит старые delayed-записи.
Шаги внутри метода:
1. Передает пороговую дату в `queueRepo.DeleteOldMessages`.
2. Фиксирует метрики успеха/ошибки и длительности.

## 8.4 `internal/service/positions/filters.go`

Порядок в `provider.New`:

1. `filterByOffice`
2. `filterByEmployee`
3. `filterByCurrency`
4. `filterByNmID`
5. `filterByPaymentType`

Каждый фильтр при нарушении условий возвращает `ErrFilterSkipMessage`.
Детализация фильтров:
- `filterByOffice`: пропускает только разрешенные `TypePoint`.
- `filterByEmployee`: отбрасывает сообщения с `EmployeeID <= 0`.
- `filterByCurrency`: отбрасывает сообщения с `CurrencyID == 0`.
- `filterByNmID`: отбрасывает `VendorCode`, попавшие в список package/excluded.
- `filterByPaymentType`: отбрасывает пустые и запрещенные payment types.
Неочевидное:
- фильтры работают как short-circuit chain: первая ошибка останавливает дальнейшие проверки для сообщения.

## 8.5 `internal/service/positions/actions.go`

Стратегии по `StatusCode`:

- `28` -> `210`/`210_ret`
- `16` -> `530` или `230`/`230_ret`
- `615` -> `259`
- `7` -> `260`
- `8` -> `270` (+ `280` при новом стикере)
- `2` -> ветки `540` / `275` / `285` / `280`
- `140` -> `212`
- `710` -> `280` при выполнении условий

Неочевидное:
- в нескольких стратегиях ошибка `arrival_dt` только логируется, action все равно отправляется.
Детализация ключевых веток:
- `strategy28`: формирует `210` или `210_ret`, добавляет `arrival_dt` в payload при наличии.
- `strategy16`: для `sm_id=101` формирует `530`, иначе `230`/`230_ret`.
- `strategy8`: формирует `270` при `box_code>0`, и дополнительно `280` при `new_sticker_code>0`.
- `strategy2`: поддерживает развилки `540` (утилизация), `275`, `285` (auto cancel), `280`.
- `strategy140`: формирует `212`, опционально добавляет `arrival_dt`.
- `strategy710`: формирует `280` только если одновременно есть `new_sticker_code` и `box_code`.

## 8.6 `internal/service/produce_service.go`

- `Send`
Цепочки:
`positions/motivation/courier-goods/wb-track -> produceService.Send`
Суть: общий шлюз отправки перед Kafka producer.
Шаги внутри метода:
1. Если входной список пустой, возвращает успех без отправки.
2. Пропускает список через `filterOldActions` (старше 14 дней не отправляются).
3. Пропускает список через `filterPresents` (исключение по `NmID=521556303`).
4. Отправляет остаток в `actionProducer.Send`.

Встроенные фильтры:

- `filterOldActions`: отбрасывает actions старше 14 дней.
- `filterPresents`: отбрасывает `NmID=521556303`.

Неочевидное:
- это глобальная бизнес-фильтрация для нескольких доменов сразу.
- дата сравнивается в timezone конкретного `OfficeDT`, что важно для пограничных суток.

## 8.7 `internal/service/courier-goods/courier-goods.go`

- `ProcessMessages`
Цепочка:
`Kafka courier -> courierGoodsHandler -> courierGoodsService.ProcessMessages`
Суть: формирует actions (`330`/`370`), валидирует валюту, проверяет офис, пропускает только `type_point=34`, отправляет.
Шаги внутри метода:
1. Для каждого DTO заполняет базовые поля `Action` (`shk_id`, `office_id`, `srid`, опционально `employee_id`, `nm_id`).
2. По `ctx` выбирает код действия (`330`/`370`) или скипает событие.
3. Проверяет и нормализует валюту (`810 -> 643`), пишет в payload.
4. Подтягивает офис, вычисляет `office_dt` в локальной TZ, проставляет `type_point`.
5. Пропускает все точки, кроме `34`.
6. Отправляет накопленные действия через producer.
Неочевидное:
- неизвестный `ctx` просто skip;
- `810` нормализуется в `643`.
- ошибка по отдельному сообщению чаще приводит к skip, а не к прерыванию всего батча.

## 8.8 `internal/service/motivation/motivation.go`

- `ProcessMotivations`
Цепочка:
`Kafka motivation -> motivationHandler -> ProcessMotivations`
Суть: фильтр по офису/типу точки, TZ-нормализация, отправка.
Шаги внутри метода:
1. Для каждого сообщения находит офис.
2. Проверяет, что `TypePoint` входит в `allowedTypePoints`.
3. Нормализует `OfficeDT` в TZ офиса.
4. Обновляет отдельные поля action (например, `SrID` как дата офиса).
5. Отправляет набор в producer.
Неочевидное:
- `op` в метриках назван нестандартно (`qrAccountActivationService...`), что усложняет навигацию в мониторинге.

- `ProcessInternships`
Цепочка:
`Kafka motivation -> motivationHandler -> ProcessInternships`
Суть: формирует actions по стажировкам (`trainee`/`mentor`).
Шаги внутри метода:
1. Валидирует офис и разрешенный `TypePoint`.
2. Формирует payload со структурой договора и датами выплат/окончания.
3. Определяет action type по роли сотрудника (`TraineePayment`/`MentorPayment`).
4. Формирует итоговый `Action` и отправляет в producer.
Неочевидное:
- неизвестный `EmployeeType` валит обработку с ошибкой.

## 8.9 `internal/service/wb-track/wb-track.go`

- `FilterInvalidWbTrackMessages`
Цепочка:
`ProcessWbTrackActions -> FilterInvalidWbTrackMessages`
Суть: отсекает офисы вне разрешенных type-point.
Шаги внутри метода:
1. Для каждого сообщения находит офис.
2. Пропускает только офисы из `allowedTypePoints`.
3. Возвращает отфильтрованный список для следующего этапа.

- `ProcessWbTrackActions`
Цепочка:
`Kafka wb-track -> handler -> ProcessWbTrackActions`
Суть: определяет тип action (`510`/`512`), проверяет валидность пакета/пользователя, резолвит сотрудника, собирает actions.
Шаги внутри метода:
1. Применяет первичную фильтрацию `FilterInvalidWbTrackMessages`.
2. Для каждого сообщения определяет код действия:
`Packaging -> 510`, `Resale -> 512`, иначе skip.
3. Для `Packaging` проверяет наличие `nm_id` в `validPackageIDs`.
4. Валидирует `WBUserID > 0`.
5. Резолвит сотрудника через `userService.GetEmployeeID`, при ошибке применяет fallback.
6. Формирует итоговый `Action` и возвращает список.
Неочевидное:
- при ошибке user lookup fallback на отрицательный `employeeID`.
- внутри метода офис читается повторно после фильтра, что добавляет второй lookup на сообщение.

- `SendActions`
Цепочка:
`handler -> SendActions -> produceService.Send`
Суть: отправка в общий producer pipeline.
Шаги внутри метода:
1. Принимает уже сформированные действия.
2. Делегирует отправку в `actionProducer.Send` (дальше включается общий pre-send filter в producer service).
3. Возвращает ошибку отправки как техническую ошибку потока.

- `UpdateValidParcelIDs`
Цепочка:
`HTTP /wb-track/update-parcel-ids -> handler -> service`
Суть: сохраняет валидные package IDs в БД и обновляет in-memory список.
Шаги внутри метода:
1. Сохраняет входной набор ID в репозиторий (`SaveValidPackageIDs`).
2. Сразу перечитывает полный список (`GetValidPackageIDs`).
3. Обновляет in-memory `validPackageIDs` для онлайн-обработки новых сообщений.
Неочевидное:
- обновление списка без синхронизации; возможна гонка при высокой конкуренции.

## 8.10 `internal/service/gift_certificates.go`

- `ProcessGiftCertificateIssues`
Цепочка:
`Kafka gift-certificates -> handler -> ProcessGiftCertificateIssues`
Суть: фильтрация + обогащение через office-cache, затем запись в БД.
Шаги внутри метода:
1. Делегирует фильтрацию и enrichment в `filterAndEnrichBrandedPickpoints`.
2. Если после фильтрации массив пустой, возвращает success без записи.
3. Иначе сохраняет issues в репозиторий `SaveGiftCertificateIssues`.

- `filterAndEnrichBrandedPickpoints`
Цепочка:
`ProcessGiftCertificateIssues -> filterAndEnrichBrandedPickpoints`
Суть: пропускает только брендированные ПВЗ и проставляет `TypePoint/CountryCode`.
Шаги внутри метода:
1. Ищет офис по `office_id` в office-cache.
2. Отбрасывает записи без офиса.
3. Отбрасывает небрендированные точки (`!office.IsBrandedPickpoint()`).
4. Проставляет в issue поля `TypePoint` и `CountryCode`.
5. Возвращает итоговый список для сохранения.
Неочевидное:
- если office-cache не прогрет, валидные события будут скипаться.

## 8.11 `internal/service/courier.go`

- `StartUpdatingCouriers`
Цепочка: прямых внешних вызовов в текущем runtime не обнаружено.
Суть: поднимает фоновую goroutine, которая раз в 5 минут проверяет давность последнего апдейта и при необходимости инициирует `UpdateCouriers`.

- `UpdateCouriers`
Цепочка: `StartUpdatingCouriers -> UpdateCouriers` (внутренний вызов).
Суть:
1. Пробует получить данные курьеров из внешнего источника.
2. Если источник недоступен, берет последний срез из БД.
3. При успешном внешнем источнике пробует синхронизировать БД.
4. Обновляет in-memory cache и `updatedAt`.
Неочевидное:
- ошибка обновления БД после успешного внешнего запроса не останавливает обновление cache;
- в текущем `provider.New` сервис не подключен, поэтому код фактически «спящий».

- `IsCourier`
Цепочка: внешних вызовов в активном runtime не обнаружено.
Суть: быстрый lookup `employee_id` в курьерском кеше.

## 9) Startup-последовательность

`app.New` -> `provider.New`:

1. Подключение `action-db`, создание схемы, запуск миграций.
2. Инициализация Kafka producer.
3. Инициализация `office-db` + office cache/service.
4. Инициализация Redis action cache.
5. Создание `userService`, `positionsService`, `motivationService`, `wbTrackService`, `courierGoodsService`, `giftCertificatesService`.
6. Прогрев `userService.ReloadUserCache` (если ошибка — startup fail).

`app.Run`:

- параллельно запускает HTTP, Kafka consumer, Cron;
- отдельно в goroutine создает партиции `employee_data` на `now-1 .. now+10`.

## 10) Ключевые хранилища и объекты

- `action.employee_data`: исторические связки `employee_id`/`wb_user_id` с датой `dt`.
- `Redis hash employee_data`: текущий lookup `wb_user_id -> employee_id`.
- `action.positions_all_queue`: delayed queue для positions.
- `valid package IDs` для wb-track: БД + in-memory кэш в сервисе.