Consumer: Identity Companies (pin-chessboard-identity-companies)

Consumer group контекста chessboard, который заводит и поддерживает профиль застройщика (DeveloperEntity) по компании identity c Type = Developer. Дом полей и инвариантов застройщика — ../domain/developer.md (здесь только дельта-маппинг); входные контракты — company-upserted и company-suspended; выход — developer-upserted.

Назначение

Группа pin-chessboard-identity-companies потребляет события жизненного цикла компаний из топика pin.identity.companies.v1 и для компаний Type = Developer материализует профиль застройщика PIN: создаёт DeveloperEntity c Id = CompanyId (INV-DEV-1 IdEqualsCompanyId — единый id снимает lookup «компания ↔ застройщик» и задаёт TenantId для всех дочерних сущностей шахматки, ADR-0052), стартовый Status = Draft, и поддерживает реплика-поля из identity (Name, LegalName). PIN-поля профиля (Description, Highlights, Website, контакты) и Talent-chessboard-реплика-поля (LogoFileId, FoundedYear — INV-DEV-3) этим consumer-ом не трогаются. Блокировка компании в identity (Active → Suspended, а также Archived) каскадно уводит профиль в Status = Hidden (INV-DEV-6 — public-проекции карточки и всех проектов застройщика не отдаются). Каждое значимое изменение переопубликуется собственным DeveloperUpsertedKafkaEvent, чтобы catalog/dev-stats/crm/rewards не подписывались на топик identity напрямую.

Owning-сервис и триггер

Поле	Значение
Сервис/контекст	chessboard (проект Pin.Chessboard.Background)
Consumer group	pin-chessboard-identity-companies
Producer	identity (Pin.Identity): admin-API компаний, синк talent-identity-sync
Топик	pin.identity.companies.v1 — общий для CompanyUpsertedKafkaEvent и CompanySuspendedKafkaEvent (порядок событий одной компании гарантирован общим key = CompanyId)
ContractVersion	ожидается 1; при envelope.Version != ContractVersion — warn-and-proceed (метрика chessboard_identity_contract_mismatch_total)
Порядок	key = CompanyId → все события одной компании в одной партиции, строгий порядок per-aggregate; монотонность применения — по event.Version (см. «Идемпотентность»)
SLA	применение p99 < 1 мин (лаг consumers, README)

Событие	Эффект (событие → что меняется)
CompanyUpsertedKafkaEvent	Type = Developer: создать/обновить DeveloperEntity (Id = CompanyId, при создании → Draft); снапшот Suspended/Archived → Hide; Active из Hidden → Unhide. Type != Developer → skip
CompanySuspendedKafkaEvent	Active → Suspended в identity → Hide профиля (Published → Hidden, INV-DEV-6); профиль не найден/Draft → штатный no-op

Контекст события

• Из метаданных события доступны: MessageId (Guid v7), CreatedAt, Version (= ContractVersion producer-а), AggregateId = CompanyId, AggregateType = "Company". Сквозной trace восстанавливается из метаданных события на span приёма.
• Actor — system-actor консюмера: CreatedByUserId / UpdatedByUserId застройщика штампуются им; работа в platform-scope (запись в не-tenanted таблицу chessboard.developers).
• Tenant-сверка: у застройщика Id = CompanyId = TenantId (INV-DEV-1) — отдельного tenant, с которым можно рассогласоваться, нет по построению; AggregateType != "Company" → fail closed (CHESSBOARD_DEVELOPER_TENANT_MISMATCH).
• Идемпотентность на уровне доставки — по MessageId, ключ identity-companies:{MessageId}; бизнес-монотонность — guard по event.Version (см. ниже).

Входные данные / payload события

Дом контракта — event-contract: company-upserted (payload CompanyUpsertedKafkaEvent + вложенная CompanyRequisitesSnapshot) и company-suspended. Ниже — только поля, которые consumers реально читают, и их маппинг в поля developer.md.

CompanyUpsertedKafkaEvent

Поле payload	Тип	Обязательность	Как используется (маппинг)
Type	string (enum CompanyType)	Да	Фильтр потока: обрабатывается только Developer; иначе — skip (ack, лог skipped-not-developer); неизвестное значение → skip (не наш тип)
CompanyId	Guid	Да	Адресация профиля застройщика по id; при создании → DeveloperEntity.Id (INV-DEV-1)
Name	string	Да	Реплика-поле → Developer.Name (перезапись, guard INV-DEV-3 при SyncSource = TalentChessboard)
Requisites.LegalName	string?	Нет	Реплика-поле → Developer.LegalName
Status	string (enum CompanyStatus)	Да	Каскад публичности: Pending/Active → создать Draft / Unhide из Hidden; Suspended/Archived → Hide; неизвестное значение → трактуется как Suspended (наиболее ограничивающий, правило company-upserted)
Version	long	Да	Монотонный guard: применяется только при event.Version новее последнего применённого (INV-DEV-4-аналог для identity-линии)
SyncSource	string (enum CompanySyncSource)	Да	Только диагностика/лог (реплика компании из PIN или Talent Identity)

LogoFileId не маппится

LogoFileId компании не маппится: логотип застройщика — Talent-chessboard-реплика-поле (INV-DEV-3), им владеет talent-chessboard-sync, не identity.

CompanySuspendedKafkaEvent

Поле payload	Тип	Обязательность	Как используется
CompanyId	Guid	Да	Адресация профиля застройщика по id; не найден → skip (компания не Developer либо профиль ещё не создан)
SuspendedAt	DateTimeOffset	Да	→ StatusChangedAt перехода Published → Hidden
Reason	string (enum CompanySuspendReason)	Да	Только лог (PinDecision/TalentBlocked/Debt)
Version	long	Да	Монотонный guard (как в A): суспенд с версией не новее применённой → skip

Алгоритм

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

A. CompanyUpsertedKafkaEvent

1. Контекст события. Восстановить метаданные события; Version != ContractVersion → warn-and-proceed; проверить идемпотентность по ключу identity-companies:{MessageId} — дубль → лог skipped-duplicate, ack.
2. Проверка входа. CompanyId задан; Name непустое (≤ 256, при превышении — truncate + warn, событие не теряется); Version >= 1. Фильтр типа: Type != Developer → ack + лог skipped-not-developer (событие штатно игнорируется — профиль застройщика не заводится для агентств/брокеров/партнёров/ПИН). Существование профиля не проверяется — операция upsert, отсутствие допустимо.
3. Что читаем. Профиль застройщика по CompanyId (может отсутствовать — первое событие компании); дополнительные агрегаты не нужны — все guard-ы инкапсулированы в профиле.
4. Бизнес-правила и переходы.
   • Монотонный guard (INV-DEV-4-аналог): профиль существует и event.Version не новее последней применённой версии компании → лог skipped-stale, к шагу 7 (только отметка идемпотентности); защита от out-of-order и replay целого топика.
   • Создание (профиля нет): завести DeveloperEntity из снапшота компании — Id = CompanyId (INV-DEV-1), Status = Draft, дефолтный Highlights, ProjectsPublishedCount = 0, доменное событие DeveloperCreatedEvent. Затем каскад статуса компании (см. ниже).
   • Обновление (профиль есть): перезаписать только identity-реплика-поля Name, LegalName; PIN-поля и Talent-chessboard-реплика-поля не трогать (INV-DEV-3, INV-DEV-5). Guard INV-DEV-3: при SyncSource = TalentChessboard поле Name управляется синком Talent — identity-обновление Name пропускается (владелец имени — Talent), LegalName применяется. Повтор того же состояния → no-op.
   • Каскад публичности (после создания/обновления, по map(payload.Status)): Suspended/Archived → Hide (guard: только из Published; Draft/Hidden — идемпотентный no-op); Active и профиль Hidden → Unhide (Hidden → Published, INV-DEV-6 снят); прочие Pending/Active статус не меняют (переход Draft → Published — только решением застройщика/ПИН, INV-DEV-5). Переход поднимает DeveloperStatusChangedEvent.
5. Что меняется. Профиль создаётся или обновляется; конкуренция с Developers-API/Talent-sync разрешается оптимистической блокировкой (конфликт → немедленный retry с перечитыванием). LastSyncedAt, применённая версия компании и версия профиля продвигаются в переходе.
6. Какие события фиксируются. При создании или значимом изменении (Name/LegalName/Status): DeveloperUpsertedKafkaEvent → топик pin.chessboard.developer-upserted.v1, key = Id, AggregateType = "Developer", ContractVersion = 1 — полный snapshot по контракту. Изменений нет (stale/no-op) → событие не публикуется.
7. Согласованность. Изменение профиля, отметка идемпотентности и исходящее событие фиксируются в одном согласованном изменении; offset подтверждается после успешной фиксации.
8. Результат. Успех → ack. Инфраструктурная ошибка → retry по политике; исчерпание / non-retriable → DLQ + алерт.

B. CompanySuspendedKafkaEvent (Active → Suspended)

Дельта к A (шаги 1, 7, 8 идентичны; ключ идемпотентности identity-companies:suspended:{MessageId}):

2. Проверка входа. CompanyId задан; SuspendedAt не пустой. Тип компании отдельно не проверяется — признак «застройщик» = существование профиля с этим Id (шаг 3).
3. Что читаем. Профиль застройщика по CompanyId; не найден → ack + лог skipped-not-developer (компания не Type = Developer либо профиль ещё не доехал upsert-ом — штатно: последующий CompanyUpsertedKafkaEvent создаст его уже в согласованном статусе).
4. Переход. Монотонный guard по payload.Version (stale → skip, как в A), затем Hide с причиной CompanySuspended (guard Published → Hidden, INV-DEV-6); Draft/Hidden → no-op; поднимается DeveloperStatusChangedEvent.
5. Что меняется. Профиль переходит в Hidden — только при выполненном переходе.
6. Какие события фиксируются. При выполненном Hide — DeveloperUpsertedKafkaEvent (snapshot со Status = Hidden); no-op → событие не публикуется.

Диаграмма

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

• На уровне доставки: по MessageId (ключи identity-companies:{MessageId} и identity-companies:suspended:{MessageId}) — повторная доставка after-commit-crash не мутирует профиль.
• Бизнес-уровень (монотонность): guard по event.Version (версия CompanyEntity) — событие с версией не новее применённой пропускается; маркер применённой версии и LastSyncedAt identity-линии хранятся на профиле (developer.md; identity-аналог ReplicaVersion Talent — источники не смешиваются). Применение конвергентно, повтор → no-op.
• Replay: да. Полный re-read pin.identity.companies.v1 с offset 0 безопасен (upsert + version-guard + конвергентный apply); терминальных откатов нет (Hide/Unhide идемпотентны по guard-ам). Точечный replay DLQ — после ручного resolution (admin-операция chessboard, ADR-0053).

Retry и DLQ

DLQ-топик группы: pin.chessboard.identity-companies-dlq.v1 (envelope исходного события + заголовки x-dlq-reason, x-dlq-error-code, x-source-topic, x-original-offset).

Сбой	Код	Retry	DLQ	Ручное действие
Инфраструктурный (PostgreSQL/Kafka, timeout)	— (exception)	Бесконечный exponential backoff 1s→60s с паузой партиции (событие не теряется)	Нет	Восстановить зависимость (runbook chessboard-overview)
Конфликт версии при гонке с Developers-API/Talent-sync	— (конфликт)	3 попытки немедленно (перечитать профиль; guard-ы по состоянию отсеют применённое)	После исчерпания	Расследование гонки
Недопустимый переход статуса (guard агрегата нарушён неожиданно)	CHESSBOARD_DEVELOPER_INVALID_STATUS_TRANSITION	Нет	Да + алерт	Разбор состояния профиля/событий identity
Tenant/адресация рассогласованы (AggregateType/CompanyId)	CHESSBOARD_DEVELOPER_TENANT_MISMATCH	Нет	Да + алерт	Разбор границ данных identity ↔ chessboard

Skip — не ошибка

Неизвестный тип компании и out-of-order (stale) — не ошибки: штатный skip + ack (метрики chessboard_identity_skipped_total{reason} / _stale_total).

Acceptance Criteria

• Given профиля застройщика ещё нет, When приходит CompanyUpsertedKafkaEvent c Type = Developer, Status ∈ {Pending, Active}, Then создан DeveloperEntity (Id = CompanyId, Status = Draft, реплика-поля Name/LegalName), опубликован DeveloperUpsertedKafkaEvent, ack.
• Given событие c Type != Developer (агентство/брокер/партнёр/ПИН), When consumed, Then ack + лог skipped-not-developer, профиль не заводится.
• Given профиль существует и event.Version не новее применённой, When приходит upsert/suspend, Then skipped-stale (только отметка идемпотентности), состояние не меняется, событие не публикуется.
• Given профиль Published, When CompanySuspendedKafkaEvent (или upsert со Status ∈ {Suspended, Archived}), Then Hide (Published → Hidden, INV-DEV-6), опубликован DeveloperUpsertedKafkaEvent(Status = Hidden), ack; профиль Draft/Hidden — идемпотентный no-op без события.
• Given профиль Hidden из-за suspend, When приходит upsert со Status = Active, Then Unhide (Hidden → Published), событие публикуется.
• Given SyncSource = TalentChessboard (имя во владении синка Talent, INV-DEV-3), When identity-upsert меняет Name, Then Name не перезаписывается (владелец — Talent), LegalName применяется.
• Given CompanySuspendedKafkaEvent пришёл раньше первого upsert-а (профиля нет), When consumed, Then ack + skipped-not-developer (последующий upsert создаст профиль уже в согласованном статусе).
• Given гонка с Developers-API/Talent-sync, When конфликт версии профиля, Then до 3 немедленных ретраев с перечитыванием (guard-ы по состоянию отсеют применённое); исчерпание — расследование.

Наблюдаемость и безопасность

• Logs (structured, по событию): consumed / skipped-duplicate / skipped-not-developer / skipped-stale / created / updated / status-changed / dead-lettered с полями MessageId, CompanyId, payload.Version, payload.Status. PII нет (реквизиты компании — юрданные, не PII физлица; в логи попадает только CompanyId).
• Metrics: chessboard_identity_lag_seconds (now − envelope.CreatedAt), chessboard_identity_consume_duration_ms, chessboard_identity_created_total, chessboard_identity_status_changed_total{to}, chessboard_identity_skipped_total{reason}, chessboard_identity_stale_total, chessboard_identity_retries_total, chessboard_identity_dlq_total, chessboard_identity_contract_mismatch_total.
• Traces: span приёма события → span обработки → span обращения к хранилищу; correlation из метаданных события — в исходящее pin.chessboard.developer-upserted.v1.
• Alerts: лаг > 5 мин; DLQ > 0; CHESSBOARD_DEVELOPER_TENANT_MISMATCH > 0.
• Dashboard / Runbook: chessboard-overview; runbook «replay DLQ identity-companies», «ручное заведение застройщика» (admin-путь при паузе авто-потока).
• Безопасность: system-actor (не пользовательский контекст); доступ к топику identity — m2m ACL Kafka; смена публичности из identity ограничена каскадом Hide/Unhide (полноценная публикация — только Developers/Admin API, README chessboard); tenant mismatch → fail closed; секретов/токенов событие не несёт.

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

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

• Домен (CQRS) (1): Entity / Aggregate: DeveloperEntity
• События (3): Event Contract: pin.chessboard.developer-upserted (DeveloperUpsertedKafkaEvent), Event Contract: pin.identity.company-suspended (CompanySuspendedKafkaEvent), Event Contract: pin.identity.company-upserted (CompanyUpsertedKafkaEvent)

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

• Целевой агрегат, инварианты, state-машина: ../domain/developer.md (INV-DEV-1/3/5/6, stateDiagram); реестр — ../README.md.
• Входные контракты: company-upserted, company-suspended; сущность-источник company.md. Производимое событие: developer-upserted.
• Соседний источник тех же полей: talent-chessboard-sync (владение полями по INV-DEV-3). ADR-0052 (tenancy), ADR-0053.