Consumer: identity.companies.v1 → CompanyReplica

Обработка топика pin.identity.companies.v1, наполняющего read-реплику компании-автора CompanyReplica. Контракты входных событий — дом company-upserted.md и company-suspended.md. Целевая проекция и инварианты — дом identity-replicas.md (CompanyReplica, INV-REP-001..004).

Назначение

Consumer group news-identity-companies потребляет публичный snapshot компаний из identity и поддерживает денормализованную подпись компании-автора поста (CompanyReplica: название, тип, лого, статус) — чтобы лента и карточка поста отдавали автора-компанию одной выборкой без синхронного запроса в identity (паттерн денормализации — identity-replicas.md). CompanyUpsertedKafkaEvent → upsert строки реплики (полный snapshot); CompanySuspendedKafkaEvent → перевод Status = Suspended (INV-REP-003: посты компании исключаются фильтром выдачи из публичных лент, контент не удаляется). Реплика — конечная точка данных: собственных Kafka-событий consumer не публикует (identity-replicas.md, раздел «События»).

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

Поле	Значение
Сервис/контекст	news
Consumer group	news-identity-companies
Producer	identity — upstream source of truth
Topic	pin.identity.companies.v1 (CompanyUpsertedKafkaEvent и CompanySuspendedKafkaEvent — общий топик, key = CompanyId → порядок per-company гарантирован)
ContractVersion	ожидается 1; при несовпадении версии события — warn-and-proceed
Эффект (событие → что меняется)	CompanyReplica (news.company_replicas): подпись компании-автора

Топик несёт два события общего семейства; они обрабатываются как два эффекта над одной репликой:

Событие	Kind (CompanyReplicaChangeKind)	Эффект
CompanyUpsertedKafkaEvent	Upsert	Upsert CompanyReplica (полный snapshot полей)
CompanySuspendedKafkaEvent	Suspend	Status = Suspended (INV-REP-003), прочие поля не трогает

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

• Из метаданных события доступны: идентификатор сообщения (MessageId, для дедупликации), время события (CreatedAt), версия контракта, идентификатор и тип агрегата-источника (AggregateId = CompanyId, AggregateType = "Company"), payload.
• Actor — system-actor консюмера (не пользовательский контекст). CompanyReplica не tenanted — платформенная подпись (ADR-0052 п.3).
• Корреляция сквозного трейса восстанавливается из метаданных события.
• Ключ бизнес-идемпотентности: news-identity-companies:{Kind}:{MessageId}; business key проекции = CompanyId (upsert по Id идемпотентен, INV-REP-004) + монотонный guard SourceVersion (INV-REP-001).

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

Ниже — только поля, которые consumer реально читает, и их маппинг в CompanyReplica. Дом полей и enum — identity-replicas.md; дом значений CompanyType/CompanyStatus — identity/domain/company.md.

CompanyUpsertedKafkaEvent → CompanyReplica (Kind = Upsert)

Поле payload	Тип	Обяз.	Как используется (маппинг)
CompanyId	Guid	Да	→ CompanyReplica.Id (PK, = identity CompanyId; прямая адресация без lookup)
Name	string	Да	→ CompanyReplica.Name (varchar(256); обрезка до 256 + warning)
Type	string (enum CompanyType)	Да	→ CompanyReplica.Type (enum CompanyType, реплика 1:1 additive: Developer/Agency/Broker/Partner/Pin); unknown → сохранить как есть + news_replica_unknown_enum_total
LogoFileId	Guid?	Нет	→ CompanyReplica.LogoFileId (без FK — внешний storage/CDN)
Status	string (enum CompanyStatus)	Да	→ CompanyReplica.Status (CompanyReplicaStatus, маппинг: Pending/Active → Active; Suspended/Archived → Suspended; unknown → Suspended, fail-closed)
Version	long	Да	→ CompanyReplica.SourceVersion (монотонный guard INV-REP-001)
CreatedAt (метаданные)	DateTimeOffset	Да	→ CompanyReplica.SyncedAt (лаг синка)

Поля Slug/Requisites/SyncSource из snapshot не реплицируются — для подписи поста не нужны (минимизация: news показывает только название/лого/тип/статус компании-автора).

CompanySuspendedKafkaEvent → CompanyReplica (Kind = Suspend)

Поле payload	Тип	Обяз.	Как используется (маппинг)
CompanyId	Guid	Да	Резолв строки реплики; адресация перехода
Version	long	Да	→ CompanyReplica.SourceVersion (монотонный guard INV-REP-001)
SuspendedAt	DateTimeOffset	Да	Не хранится (у реплики нет поля; UpdatedAt ставит app)
Reason	string (enum CompanySuspendReason)	Да	Не реплицируется (причина не влияет на подпись; остаётся в аудите identity)

Переход всегда устанавливает CompanyReplica.Status = Suspended (Kind Suspend — сигнальное событие).

Алгоритм

Оба события различаются только шагами 1 и 4 (какой эффект применить); шаги 2–3, 5–8 — общие.

1. Контекст события. Восстановить метаданные (MessageId, CreatedAt, версия, AggregateId, payload); сверить версию с ContractVersion = 1 — при несовпадении warn-and-proceed. Проверить бизнес-идемпотентность по ключу news-identity-companies:{Kind}:{MessageId}: дубль → лог skipped-duplicate, пропуск (ack). Определить эффект: upserted → Kind = Upsert (полный snapshot); suspended → Kind = Suspend.
2. Проверка входа. CompanyId != Guid.Empty; SourceVersion >= 1; для Kind = Upsert — Name непустой, Type присутствует. Нарушение контракта payload (обяз. поле отсутствует) → NEWS_REPLICA_INVALID_PAYLOAD (non-retriable → DLQ).
3. Что читаем. Текущая строка реплики по CompanyId (может отсутствовать при первом Upsert). Для Kind = Suspend при отсутствии строки (аномалия — порядок per-company гарантирован ключом) → лог suspended-missing-replica, пропуск (ack); последующий Upsert восстановит строку.
4. Бизнес-правила и маппинг. Guard монотонности INV-REP-001: если строка есть и SourceVersion события <= сохранённого → устаревшее/повторное событие → лог skipped-stale, метрика news_replica_stale_total, пропуск (ack), мутаций нет. Иначе:
   • Kind = Upsert: собрать целевую строку — Id = CompanyId, Name, Type (CompanyType), LogoFileId, Status (маппинг CompanyStatus → CompanyReplicaStatus), SourceVersion = Version, SyncedAt = CreatedAt события (таблица выше);
   • Kind = Suspend: на существующей строке Status = Suspended (INV-REP-003), SourceVersion = Version, прочие поля (Name/Type/LogoFileId) неизменны.
5. Что меняется. Upsert (Insert/Update по Id) либо обновление статуса существующей строки. Конкурентный доступ параллельных консюмеров — оптимистическая блокировка (обновление не должно затирать более свежее).
6. Согласованность. Мутация company_replicas и отметка идемпотентности фиксируются в одном согласованном изменении; offset подтверждается только после успешной фиксации.
7. Результат. Успех → ack. Контролируемая ошибка → retry/DLQ по политике. No-op-ветки (дубль, skipped-stale, suspended-missing-replica) — тоже ack.

Диаграмма

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

Предусловия. Топик pin.identity.companies.v1 доставляет события с key = CompanyId — порядок событий одной компании гарантирован; identity — upstream source of truth. CompanyReplica может отсутствовать до первого Upsert (и допустимо отсутствует при Suspend до создающего Upsert — аномалия ретеншена → no-op ack).

Постусловия (после обработки события с SourceVersion новее сохранённого):

• Upsert: строка CompanyReplica по Id = CompanyId отражает Name/Type/LogoFileId/Status (маппинг CompanyStatus → CompanyReplicaStatus)/SourceVersion/SyncedAt из snapshot; лента и карточка поста показывают свежую подпись компании-автора одной выборкой в пределах лага консюмера (< 1 мин).
• Suspend: Status = Suspended (INV-REP-003), SourceVersion обновлён, прочие поля не тронуты; посты компании исключаются из публичных лент фильтром выдачи query-слоя, контент не удаляется.
• Устаревшее/повторное событие (SourceVersion <= сохранённого) состояние реплики не меняет (INV-REP-001, монотонный guard); offset подтверждается только после фиксации мутации реплики и отметки идемпотентности в одном согласованном изменении.
• Нарушение контракта payload (обяз. поле отсутствует) не мутирует реплику → запись уходит в DLQ (NEWS_REPLICA_INVALID_PAYLOAD), offset не подтверждается как обработанный успешно.
• Консюмер собственных Kafka-событий не публикует — CompanyReplica конечная точка данных.

Acceptance Criteria

• First-seen upsert. Дано: реплики по CompanyId нет. Когда: приходит CompanyUpsertedKafkaEvent (Version = 1, Name непуст, Type задан). Тогда: создаётся строка CompanyReplica с Name/Type/LogoFileId/Status (маппинг), SourceVersion = 1, SyncedAt = event.CreatedAt.
• Монотонный guard (stale). Дано: реплика с SourceVersion = 5. Когда: приходит событие с Version = 4 (переупорядочение/replay). Тогда: строка не меняется, news_replica_stale_total++, лог skipped-stale, ack.
• Дубликат доставки. Дано: событие с MessageId = M уже обработано. Когда: то же M доставлено повторно. Тогда: skipped-duplicate, состояние не меняется, ack (INV-REP-004 — upsert по Id идемпотентен).
• Suspend. Дано: существует CompanyReplica в Active. Когда: приходит CompanySuspendedKafkaEvent со свежей Version. Тогда: Status = Suspended, SourceVersion обновлён, Name/Type/LogoFileId не тронуты; посты компании исключаются из публичных лент query-слоем (INV-REP-003).
• Suspend без реплики. Дано: реплики по CompanyId нет. Когда: приходит CompanySuspendedKafkaEvent. Тогда: suspended-missing-replica, ack (строку восстановит последующий Upsert).
• Reactivate. Дано: реплика в Suspended. Когда: приходит CompanyUpsertedKafkaEvent со Status = Active и свежей Version. Тогда: Status = Active (отдельного события Suspended → Active нет — приходит обычный upsert, INV-REP-001).
• Unknown enum (additive у источника). Дано: Type без прямого соответствия. Когда: обрабатывается upsert. Тогда: Type сохранён как есть + warning + news_replica_unknown_enum_total; Status unknown → Suspended (fail-closed); событие не теряется.
• Битый payload. Дано: в событии отсутствует обязательное поле. Когда: обработка. Тогда: NEWS_REPLICA_INVALID_PAYLOAD → DLQ + алерт, реплика не мутирована.

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

• Consumer-level: дедупликация по MessageId (ключ news-identity-companies:{Kind}:{MessageId}).
• Business-level: business key проекции = CompanyId; upsert по Id идемпотентен (INV-REP-004), повтор доставки не меняет состояние. Монотонный guard SourceVersion (INV-REP-001) отсекает устаревшие/переставленные события.
• Replay: да. Повторный re-read топика безопасен (идемпотентный ключ + upsert-by-Id + version-guard); runbook «replay DLQ консюмеров identity» — README контекста, раздел «Runtime».

Retry и DLQ

Сбой	Код	Retry	DLQ	Ручное действие
Инфраструктурный (PostgreSQL/Kafka недоступны, timeout)	—	Бесконечный exponential backoff 1s→60s с паузой партиции	Нет	Runbook: восстановить зависимость
Нарушение контракта payload (обяз. поле отсутствует)	NEWS_REPLICA_INVALID_PAYLOAD	Нет (non-retriable)	Да + алерт	Исправить продьюсер identity → replay
Устаревшее событие (SourceVersion <= сохранённого)	—	Нет	Нет	No-op ack (INV-REP-001, ожидаемо при replay/переупорядочивании)
Unknown enum Type/Status (additive у продьюсера)	—	Нет	Нет	Warn + fallback (Type как есть; Status → Suspended); при необходимости добавить маппинг (релиз)
Suspend для отсутствующей строки	—	Нет	Нет	No-op ack (suspended-missing-replica); строку восстановит Upsert
Гонка параллельных консюмеров (конфликт оптимистической блокировки)	—	Да (короткий backoff, перечитать)	Нет	Автоповтор; при исчерпании — как инфраструктурный

Политика retry/DLQ единообразна с консюмером пользователей identity-users.md (тот же топик-семейство identity, тот же паттерн реплики).

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

• Logs: consumed / skipped-duplicate / skipped-stale / suspended-applied / suspended-missing-replica / dead-lettered с полями MessageId, CompanyId, Kind, SourceVersion. PII нет — реплицируются только публичные display-поля (Name/Type/LogoFileId); Requisites/Slug/SyncSource не реплицируются.
• Metrics: news_identity_companies_consume_duration_ms, news_replica_upserted_total{kind}, news_replica_suspended_total, news_replica_stale_total (INV-REP-001 skip), news_replica_unknown_enum_total, news_replica_retries_total, news_replica_dlq_total, news_replica_lag_seconds (now − время события).
• Traces: span приёма события → span обработки → span обращения к хранилищу; correlation из метаданных.
• Alerts: лаг > 1 мин (SLO README «лаг консюмеров identity < 1 мин»); DLQ > 0; всплеск news_replica_unknown_enum_total (drift enum источника).
• Безопасность: consumer под system-actor; CompanyReplica не tenanted (платформенная подпись, ADR-0052 п.3), кросс-tenant-записи нет по построению. Топик pin.identity.companies.v1 — платформенный, доступ — m2m ACL Kafka; identity публикует display-поля доверенным consumer-ам (company-upserted.md, раздел «Безопасность»).

Assumptions и ограничения

• Порядок событий одной компании гарантирован общим Kafka key = CompanyId (оба типа — один топик); поэтому Suspend ожидается после создающего Upsert. Отсутствие строки на Suspend трактуется как аномалия ретеншена → no-op ack (см. Retry).
• Обратный переход Suspended → Active отдельного события не имеет — приходит обычный CompanyUpsertedKafkaEvent со Status = Active (INV-REP-001 переводит реплику обратно в Active).
• INV-REP-003 (исключение постов компании из публичных лент) — правило выдачи query-слоя по CompanyReplica.Status = Suspended; consumer только выставляет статус, посты физически не трогает.
• Sibling-консюмер identity-users.md наполняет AuthorReplica из pin.identity.users.v1 — отдельный документ (разные топик/проекция).

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

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

• Домен (CQRS) (1): Entity / Aggregate: AuthorReplica + CompanyReplica
• API (1): GET /api/v1/posts — Глобальная лента + поиск
• События (3): События сервиса news: карта топиков и отклонённые события/консюмеры, Event Contract: pin.identity.company-suspended (CompanySuspendedKafkaEvent), Event Contract: pin.identity.company-upserted (CompanyUpsertedKafkaEvent)
• Консюмеры (1): Consumer: identity.users.v1 → AuthorReplica (news-identity-users)

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

• Целевая проекция и инварианты: identity-replicas.md (CompanyReplica, INV-REP-001..004, маппинг CompanyStatus → CompanyReplicaStatus); потребитель полей — post.md (AuthorCompanyId).
• Входные контракты (дом полей): company-upserted.md, company-suspended.md; дом enum — identity/domain/company.md (CompanyType/CompanyStatus).
• Sibling-консюмер реплик: identity-users.md; карта событий/консюмеров — README контекста.
• ADR: 0052 tenancy, 0056 simplicity-first.