Background Job: talent-initial-import (первичная загрузка / bootstrap из Talent Identity)

Разовый миграционный job контекста identity: наполняет пустую реплику компаниями-застройщиками и их менеджерами из talent.microservices.identity перед тем, как непрерывный consumer talent-identity-sync возьмёт актуализацию на себя. Реализует правило talent-replica.md R9 «Начальная загрузка»; сценарий синка — UC-IDN-007. Дома целевых полей и инвариантов — company.md / user.md (здесь не переописываются, только источник и порядок).

Назначение

Bootstrap первичного наполнения реплики Talent на старте платформы (или полного пересинка одной компании): разово прочитать полный набор Talent-агрегатов (tenant → CompanyEntity, затем менеджеры застройщика → UserEntity типа DeveloperManager) и создать их как реплики PIN (SyncSource = TalentIdentity, ReplicaVersion/LastSyncedAt заполнены), идемпотентно, в порядке tenant→user (правило R4). Job применяет ту же per-item логику применения снапшота tenant/user, что и consumer (маппинг полей и инварианты не дублируются), поэтому результат bootstrap и потоковой синхронизации побитно совпадает. После завершения управление передаётся continuous-consumer-у; дальнейшая актуализация — только событиями talent.identity.{tenants,users}.v1.

Классификация

Поле	Значение
Сервис	identity
Owning team	PIN Platform Core
Статус документа	draft
Область действия	Разовый bootstrap реплики Talent; таблицы identity.companies / identity.users / identity.role_assignments + оперативные identity.talent_import_runs / talent_import_pages / talent_import_failed_items
Тип job	One-off (миграционный / bootstrap; не по расписанию)
Trigger	Ручной запуск администратором платформы (ADR-0053) либо CLI-флаг --run-initial-import
Критичность	high (без реплики нет входа менеджеров застройщиков; запускается контролируемо, вне горячего пути)

Расписание и защита от повторного запуска

Аспект	Правило
Schedule	Нет (one-off, инициируется оператором); типично — один раз при заведении окружения либо точечный пересинк компании (TalentTenantIdFilter)
Единственный запуск	На кластер — один активный запуск: обеспечивается блокировкой Pin:lock:job:identity-talent-initial-import (PostgreSQL advisory lock, без Redis — ADR-0056; переживает рестарт pod-а); если запуск уже идёт — отказ (без ретрая)
Batch size	PageSize (default 500 записей на страницу; см. входную модель)
Параллелизм внутри запуска	Страницы работ раскладываются в очередь talent_import_pages; несколько pod-ов тянут по одной странице (взятая страница блокируется, остальные её пропускают) — параллелизм без двойной обработки и без Redis (ADR-0056); барьер этапа: страницы Users ставятся в очередь только после Done всех страниц Tenants (порядок tenant→user, R4)
Max duration	До 30 мин на полный bootstrap (envelope: тысячи компаний / десятки тысяч менеджеров); аренда страницы 10 мин (протухший InProgress переклаймится)

Входная модель TalentInitialImportRequest

Поле	Тип	Обязательность	Источник	Описание
TenantId	—	Не применяется	—	Job кросс-tenant: пишет под system-actor, CompanyId/TenantId каждой записи выводится из данных Talent (как в consumer); companies — platform-level, users.TenantId = CompanyId
Mode	TalentImportMode	Да	вход администратора	Способ чтения источника (enum ниже)
TalentTenantIdFilter	Guid?	Нет	вход администратора	Ограничить bootstrap одной компанией (ExternalTalentTenantId) — частичный пересинк (Runbook «пересинк компании из Talent»)
PageSize	int	Нет (default 500)	вход администратора	Размер страницы API/replay; 1..2000
DryRun	bool	Нет (default false)	вход администратора	Прогон без записи и без исходящих событий: только валидация и подсчёт
CorrelationId	Guid	Да	генерируется	Сквозной trace всех страниц и исходящих pin.identity.*

Enum TalentImportMode (хранится text)

Значение	Описание
TopicReplay	Топики Talent compacted (по одному последнему snapshot на ключ) → сброс offset consumer-группы pin-identity-talent-sync в 0 и повторное потребление тем же путём; job только оркеструет reset + барьер этапов
ApiPull	Топики не compacted → постраничный HTTP-импорт по API Talent (tenants list, users list — имена assumption, schema registry Talent), подаёт те же снапшоты tenant/user на применение

Оперативное состояние (checkpoint, job-owned)

Не доменные агрегаты — операционные таблицы прогресса (resume + параллелизм).

Сущность	Ключевые поля	Назначение
TalentImportRunEntity	Id, Mode, Status (Running/Completed/Failed/Cancelled), StartedAt, FinishedAt, CorrelationId, счётчики по этапам	Один запуск; защита от двойного старта вместе с блокировкой
TalentImportPageEntity	Id, RunId, Stage (TalentImportStage: Tenants / Users), PageNo, Cursor (opaque токен API), Status (Pending/InProgress/Done/Failed), LeaseUntil, ItemsApplied, ItemsFailed	Единица работы; берётся под блокировкой; идемпотентный resume по Status
TalentImportFailedItemEntity	Id, RunId, Stage, ExternalId, ErrorCode, Payload (snapshot), CreatedAt	Карантин poison-item-ов (не валит страницу); ручной разбор

Алгоритм

Оркестратор планирует два этапа с барьером; тело — цикл обработки одной страницы (повторяемая единица; параллелится между pod-ами).

1. Контекст. Актор = система (тот же system-actor, что у consumer); захватить блокировку запуска (занята → отказ, без ретрая); создать либо загрузить TalentImportRunEntity (existing Running → resume того же запуска, идемпотентность рестарта). CorrelationId из входа. Идемпотентность per-item — не здесь, а через version-guard шага 4.
2. Валидация входа. Mode — валидный enum; PageSize в 1..2000; доступность источника — health-probe Talent API (ApiPull) либо наличие топиков и права reset offset (TopicReplay) → иначе отказ IDENTITY_IMPORT_SOURCE_UNAVAILABLE. Существование сущностей не проверяется — операция upsert.
3. Планирование. Два этапа в порядке Tenants → Users: для текущего этапа — постранично перечислить источник (ApiPull: list(cursor, PageSize, filter); TopicReplay: чтение партиций с offset 0), записать TalentImportPageEntity(Status = Pending). Этап Users планируется только после того, как все страницы Tenants — Done (барьер R4: компания уже в реплике до её пользователя, исключает IDENTITY_SYNC_COMPANY_NOT_FOUND). Взятие страницы воркером: первая Pending-страница текущего этапа берётся под блокировкой (параллельные pod-ы пропускают взятые), помечается InProgress, LeaseUntil = now + 10 мин.
4. Применение per-item страницы. Для каждого элемента собрать снапшот tenant/user и применить той же логикой применения снапшота, что у consumer (маппинг поле-в-поле — talent-replica.md, не дублируется):
   • Tenant: поиск по ExternalTalentTenantId; version-guard payload.Version > ReplicaVersion (INV-6 — идемпотентный skip при повторе); при отсутствии — создание CompanyEntity из снапшота Talent (Status = Pending) → активация (Pending → Active, применён первый snapshot); неизвестный CompanyType → карантин + алерт (unknown enum, INV-4 ImmutableType).
   • User: поиск по ExternalTalentUserId; резолв CompanyId по ExternalTalentTenantId (компания гарантированно существует — барьер шага 3); version-guard (INV-10); при отсутствии — создание UserEntity из снапшота (Type = DeveloperManager, Status = Invited, ContactsManagedBy = TalentSync) + дефолтное назначение RoleAssignment(role = developer-manager, scope = Organization(CompanyId)); конфликт телефона (R6) → карантин (авто-merge запрещён).
5. Что меняется. Создание/обновление реплик по External*Id; порядок FK обеспечен барьером (companies этапа Tenants раньше users этапа Users), внутри страницы Users — role_assignments добавляется вместе с пользователем. Ошибка одного элемента → TalentImportFailedItemEntity (карантин), страница продолжается (ItemsFailed++). DryRun = true → запись и шаг 6 пропускаются (только счётчики).
6. События. В том же согласованном изменении — те же pin.identity.*, что у consumer: CompanyUpsertedKafkaEvent → pin.identity.companies.v1 (key = Id); UserUpsertedKafkaEvent → pin.identity.users.v1; при создании пользователя с дефолтной ролью — дополнительно UserAccessChangedKafkaEvent → pin.identity.access.v1 (инвалидация кэшей прав). Payload собирается из применённого агрегата (snapshot-семантика R3).
7. Согласованность. На страницу — одно согласованное изменение: применённые агрегаты + RoleAssignment + TalentImportPageEntity(Status = Done, ItemsApplied/Failed) + отметки идемпотентности + исходящие события — атомарно. Падение до фиксации → страница остаётся InProgress, аренда протухает, переклаймится и переигрывается (upsert + version-guard делают повтор безопасным).
8. Результат. По завершении обоих этапов — TalentImportRunEntity(Status = Completed, FinishedAt), освобождение блокировки; результат TalentInitialImportResult (по агрегатам companies/users: Planned/Applied/Skipped/Failed, длительность, RunId). Оператор передаёт актуализацию continuous-consumer-у.

Согласованность изменения

Элемент	Правило
Граница	Одно согласованное изменение на страницу; не на весь запуск (иначе долгая блокировка и невозможность resume)
Сохраняемое состояние	Агрегаты-реплики (companies/users) + RoleAssignment + TalentImportPageEntity(Done) + отметки идемпотентности + исходящие события — атомарно; TalentImportFailedItemEntity — в том же изменении
Внешние вызовы	Чтение Talent (API/replay) — до изменения страницы (network вне фиксации); публикация событий — после фиксации (at-least-once)
Частичная ошибка	Poison-item → карантин, страница продолжается; сбой страницы → откат, переклайм по аренде; сбой этапа не трогает Done-страницы (resume)

Идемпотентность и retry

Сценарий	Правило
Job restart	Resume: TalentImportRunEntity(Running) подхватывается; Done-страницы пропускаются, Pending/протухший InProgress переигрываются
Duplicate item / повторный запуск	поиск по ExternalTalentTenantId / ExternalTalentUserId + монотонный ReplicaVersion-guard (INV-6 / INV-10) → повторное применение — no-op, состояние не откатывается (тот же контракт, что replay топика у consumer)
Дефолтная роль	RoleAssignment(developer-manager, Organization) создаётся только при первом создании пользователя; повторный upsert существующего его не дублирует
Повторяемая ошибка (источник/БД недоступны)	Страница InProgress не помечается Done; аренда протухает → переклайм; infra-ошибки — exponential backoff
Poison item (unknown CompanyType, конфликт телефона R6)	TalentImportFailedItemEntity + identity_import_failed_items_total + алерт; страница/запуск не падают; ручной разбор (Runbook), затем точечный replay
Гонка с continuous-consumer	Безопасна: ReplicaVersion-guard дедуплицирует двойное применение; рекомендуемая последовательность — bootstrap, затем поток; одновременная работа корректна по идемпотентности

Наблюдаемость

Тип	Правило
Logs	import-started / stage-planned / page-claimed / page-applied / item-quarantined / stage-completed / import-completed/import-failed с RunId, Stage, PageNo, ExternalId, ItemsApplied/Failed, CorrelationId; PII (телефон/ФИО) — маскируется (+7***1234)
Metrics	identity_import_running (gauge 0/1), identity_import_pages_total{stage,status}, identity_import_items_total{aggregate,result}, identity_import_page_duration_ms, identity_import_failed_items_total; переиспользуются talent_sync_stale_events_total, talent_sync_conflicts_total (R6)
Traces	CorrelationId запуска → span этапа → span страницы → spans записи/событий; проброс в исходящие pin.identity.*
SLO цикла	Полный bootstrap реплики (envelope ADR-0056: тысячи компаний / десятки тысяч менеджеров) < 30 мин; страница p99 < 3 с; после хендовера лаг consumer-а < 5 мин (SLO README)
Alerts	identity_import_running = 1 без роста pages_total{status=Done} > 10 мин (stuck); identity_import_failed_items_total > 0; длительность запуска > SLO; рост конфликтов телефона
Dashboard	identity-overview (панель bootstrap: прогресс по этапам, throughput, карантин)

Пред/постусловия и хендовер

• До запуска. Реплика Talent пуста либо частично наполнена (для TalentTenantIdFilter — пересинк одной компании); блокировка запуска свободна; источник доступен (ApiPull — health-probe API Talent; TopicReplay — топики compacted и есть право reset offset). Continuous-consumer может быть остановлен или работать параллельно (гонка безопасна — version-guard).
• После Completed. Для набора, охваченного фильтром: каждая Talent-компания представлена CompanyEntity (SyncSource = TalentIdentity, ReplicaVersion/LastSyncedAt заполнены); каждый менеджер — UserEntity(DeveloperManager, Invited) с ровно одним дефолтным RoleAssignment(developer-manager, Organization); по применённым агрегатам ушли pin.identity.companies.v1 / users.v1 [+ access.v1]; poison-item-ы в карантине (TalentImportFailedItemEntity), не заблокировав прогон; блокировка освобождена.
• Хендовер. Дальнейшая актуализация — только continuous-consumer talent-identity-sync; bootstrap и потоковый синк дают идентичное состояние реплики (parity), повторный запуск идемпотентен (version-guard). DryRun = true не меняет состояние и не публикует события — только счётчики.

Связанный runbook

Сценарий	Runbook
Первичный запуск окружения	«bootstrap реплики Talent»: запустить job (Mode по свойству топиков), дождаться Completed, передать поток consumer-у
Пересинк одной компании	«пересинк компании из Talent»: запуск с TalentTenantIdFilter (после ручного resolution конфликтов)
Карантин poison-item-ов	Разбор TalentImportFailedItemEntity: unknown-enum — согласовать тип; конфликт телефона (R6) — ручной merge учёток → точечный replay
Job завис (протухшая блокировка/аренда)	identity-overview: проверить держателя блокировки, дождаться/сбросить аренду страниц, resume

Тесты

Тест	Ожидание
Empty source	Пустой Talent → Completed, 0 применённых, идемпотентно
Порядок tenant→user	Барьер этапов: пользователи применяются только после компаний (нет IDENTITY_SYNC_COMPANY_NOT_FOUND)
Duplicate run	Повторный запуск на наполненной БД — no-op (version-guard); состояние не меняется, роль не дублируется
Partial failure / resume	Падение на середине этапа → рестарт продолжает с Pending-страниц, Done не переигрываются
Poison item	Конфликт телефона (R6) / unknown CompanyType → карантин + метрика, страница и запуск завершаются успешно
Дефолтная роль	При создании DeveloperManager заводится ровно один RoleAssignment(developer-manager, Organization) + access-changed
Parity с consumer	Bootstrap и потоковый синк одной компании дают идентичное состояние реплики и одинаковые pin.identity.*
DryRun	Записи и исходящих событий нет; TalentInitialImportResult содержит корректные счётчики
Cancellation	Остановка после текущей страницы; Run = Cancelled, Done-страницы сохранены

Известные ограничения и assumptions

Assumption

Имена топиков/HTTP-эндпоинтов Talent (talent.identity.{tenants,users}.v1, tenants/users list) — зафиксированное assumption до подтверждения schema registry Talent (talent-replica.md «Связанные документы»).

• TopicReplay применим только если топики Talent compacted; иначе — ApiPull.
• Секреты входа (PasswordHash/PinCodeHash/IsTwoFactorEnabled) не реплицируются: менеджер входит в PIN обычным СМС-flow по реплицированному номеру; пользователь создаётся Invited → Active при первом входе (user.md).
• Конфликт телефона Talent-менеджера с существующим инвестором (R6) — карантин + ручной merge учёток (авто-merge запрещён — риск отдать чужой аккаунт); физическое удаление PII — регламент администрирования (ADR-0053).

Обратные ссылки

Автоссылки: где используется этот документ.

• Домен (CQRS) (1): Реплика Talent Identity: правила синка + Entity TalentSyncState
• Фоновые задачи (1): Background Job: talent-initial-import (bootstrap и reconciliation справочников из Talent Handbook)

Связанные документы

• Consumer (тот же маппинг/инварианты, хендовер): ../consumers/talent-identity-sync.md; правило bootstrap: ../domain/talent-replica.md (R9); сценарий: UC-IDN-007; сервис: ../README.md.
• Целевые сущности: company (INV-4/5/6) · user (INV-10), дефолтная роль — role; события: company-upserted, user-upserted, user-access-changed.
• Решения: ADR-0056 simplicity-first, ADR-0052, ADR-0053.