Перейти к основному содержимому

Consumer: identity.users.v1 → AuthorReplica (news-identity-users)

Обработка топика pin.identity.users.v1, наполняющего read-реплику подписи автора AuthorReplica в контексте news. Контракты входных событий — дом user-upserted.md (UserUpsertedKafkaEvent) и user-blocked.md (UserBlockedKafkaEvent). Целевая реплика (дом полей, enum, инвариантов) — identity-replicas.md; здесь описывается только приём событий и маппинг.

Назначение​

Consumer group news-identity-users потребляет топик pin.identity.users.v1 и поддерживает news.author_replicas в актуальном состоянии, чтобы лента и карточка поста рендерили подпись автора (ФИО/тип/компания/статус) одной выборкой без синхронного джойна в identity (README news, паттерн денормализации). Обрабатывает два события общего топика:

  • UserUpsertedKafkaEvent — upsert AuthorReplica (создание при first-seen UserId, иначе обновление display-полей);
  • UserBlockedKafkaEvent — перевод AuthorReplica.Status = Blocked (INV-REP-002; выдача ленты помечает/скрывает посты автора по решению query-слоя).

Собственных Kafka-событий consumer не публикует (реплика — конечная точка read-модели, ADR-0056: без событий без потребителей). PhoneNumber/Email из payload игнорируются — для подписи не нужны, минимизация PII (identity-replicas.md, «Permissions / PII»).

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

ПолеЗначение
Сервис/контекстnews
Consumer groupnews-identity-users
Produceridentity (контракты — upstream source of truth)
Topicpin.identity.users.v1 (key = UserId, порядок per-user)
ContractVersionожидается 1; при несовпадении версии события — warn-and-proceed
Эффект (событие → что меняется)AuthorReplica (news.author_replicas): подпись автора

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

СобытиеЭффект на AuthorReplica
UserUpsertedKafkaEventUpsert всех display-полей
UserBlockedKafkaEventStatus = Blocked

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

  • Из метаданных события доступны: MessageId (дедупликация), CreatedAt, версия контракта, AggregateId = UserId, AggregateType = "User".
  • Actor — system-actor консюмера; реплика не tenanted (платформенные подписи, ADR-0052 п.3).
  • Корреляция сквозного трейса восстанавливается из метаданных события.
  • Ключ бизнес-идемпотентности: news-identity-users:{EventType}:{MessageId}; бизнес-ключ реплики — UserId (INV-REP-004).

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

Ниже — только поля, которые consumer реально читает, и их маппинг в AuthorReplica (дом полей, ограничений и enum — identity-replicas.md; дом контрактов событий — user-upserted.md / user-blocked.md).

UserUpsertedKafkaEvent → AuthorReplica (upsert)​

Поле payloadТипОбяз.Как используется (маппинг)
UserIdGuidДа→ AuthorReplica.Id (PK = бизнес-ключ; прямая адресация без lookup)
NameFullNameSnapshotДа→ DisplayName = FirstName + ' ' + LastName (trim; LastName может быть пустой → только FirstName); обрезка до varchar(384) + warning; MiddleName в подпись не входит
Typestring (enum UserType)Да→ Type : PostAuthorType по маппингу enum ниже
CompanyIdGuid?Нет→ CompanyId (null — инвестор/сотрудник; логическая ссылка, без FK, INV-REP-004)
Statusstring (enum UserStatus)Да→ Status : AuthorReplicaStatus по маппингу ниже
VersionlongДа→ SourceVersion (монотонный guard INV-REP-001)
PhoneNumber, Email, RegisteredVia, ReferrerUrl——игнорируются (PII не нужен для подписи / не относится к реплике)

CreatedAt события → SyncedAt (время исходного события); UpdatedAt = now().

UserBlockedKafkaEvent → AuthorReplica (переход статуса)​

Поле payloadТипОбяз.Как используется (маппинг)
UserIdGuidДаКлюч реплики (AuthorReplica.Id) для перевода в Blocked
VersionlongДа→ SourceVersion (монотонный guard INV-REP-001)
CompanyId, BlockedAt, Reason——не применяются — у AuthorReplica полей причины/времени блокировки нет

Маппинг enum (дом значений и правило — identity-replicas.md)​

  • UserType → PostAuthorType: 1:1 по совпадающему имени (Investor/DeveloperManager/Agent/ PinEmployee); PartnerManager (нет прямого соответствия) → ближайшее AgencyManager + warning + метрика news_replica_unknown_enum_total; событие не теряется (additive-правило источника).
  • UserStatus → AuthorReplicaStatus: Invited/Active → Active; Blocked → Blocked; Archived → Archived; unknown → Blocked (fail-closed).

Алгоритм​

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

  1. Контекст события. Восстановить метаданные (MessageId, CreatedAt, версия, payload); сверить версию с ContractVersion = 1 — при несовпадении warn-and-proceed. Проверить бизнес-идемпотентность по ключу news-identity-users:{EventType}:{MessageId}: дубль → лог skipped-duplicate, пропуск (ack). Определить эффект: upserted → upsert; blocked → переход статуса.
  2. Проверка входа. UserId != Guid.Empty; Version >= 1; для upsert — Name.FirstName непуст. Нарушение контракта payload (обяз. поле отсутствует/пусто) → NEWS_REPLICA_INVALID_PAYLOAD (non-retriable → DLQ). Существование связанной CompanyReplica синхронно не проверяется (порядок событий user/company не гарантирован, без FK, INV-REP-004).
  3. Что читаем. Текущая строка реплики по UserId (может отсутствовать — first-seen допустим). Для события blocked при отсутствии строки: лог skipped-missing-replica, пропуск (ack) — DisplayName из blocked-события не восстановить (порядок per-user гарантирует предшествующий upsert; см. Assumptions).
  4. Бизнес-правила и маппинг. Guard монотонности (INV-REP-001): если строка есть и Version события <= сохранённого SourceVersion → skip + метрика news_replica_stale_total, лог skipped-stale, пропуск (ack). Иначе:
    • upsert: собрать целевые поля (DisplayName, Type, CompanyId, Status, SourceVersion, SyncedAt) по таблицам маппинга (enum-fallback с warning);
    • blocked: Status = Blocked, SourceVersion = Version, SyncedAt = CreatedAt события (INV-REP-002 — эффект на выдачу применяется query-слоем ленты, не здесь).
  5. Что меняется. Upsert строки реплики (insert при first-seen, update иначе). Конкурентный доступ параллельных консюмеров user/company — оптимистическая блокировка (обновление не должно затирать более свежее).
  6. Согласованность. Upsert author_replicas и отметка идемпотентности фиксируются в одном согласованном изменении; offset подтверждается только после успешной фиксации.
  7. Результат. Успех → ack. Контролируемая ошибка → retry/DLQ по политике. No-op-ветки (дубль, stale-версия, отсутствующая реплика на blocked) — тоже ack.

Диаграмма​

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

  • Consumer-level: дедупликация по MessageId (ключ news-identity-users:{EventType}:{MessageId}) — повтор доставки → skipped-duplicate → ack.
  • Business-level: бизнес-ключ — UserId (INV-REP-004): upsert по Id идемпотентен; монотонный guard INV-REP-001 (Version > SourceVersion) гарантирует, что переупорядоченная/повторная доставка не откатывает состояние.
  • Replay: да. Повторный re-read топика (и backfill с earliest на первом деплое greenfield) безопасен — идемпотентный ключ UserId + version-guard дают то же финальное состояние. Runbook — news/runbooks/ (replay DLQ консюмеров identity, README news).

Retry и DLQ​

СбойКодRetryDLQРучное действие
Инфраструктурный (PostgreSQL/Kafka недоступны, timeout)—Exponential backoff с паузой партицииНетRunbook: восстановить зависимость
Нарушение контракта payload (обяз. поле отсутствует/пусто)NEWS_REPLICA_INVALID_PAYLOADНет (non-retriable)Да + алертИсправить продьюсер identity → replay
Систематическая ошибка маппинга (нераспознаваемая структура Name/enum вне additive-правила)NEWS_REPLICA_MAPPING_FAILEDНетДа + алертСогласовать контракт с identity, добавить маппинг, replay
Unknown UserType/UserStatus (additive у продьюсера)—НетНетWarn + fallback (AgencyManager/Blocked) + метрика; при необходимости добавить значение (релиз)
Гонка параллельных консюмеров (user/company на одну строку)—Оптимистичный ретрай (перечитать)НетАвтоматически

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

  • Logs: consumed / skipped-duplicate / skipped-stale / skipped-missing-replica / upserted / blocked / dead-lettered с полями MessageId, UserId (= AggregateId), EventType, SourceVersion. PII: в логи попадает только DisplayName (ФИО автора) — маскируется по правилу producer/consumer; PhoneNumber/Email не реплицируются и в логах отсутствуют.
  • Metrics: news_replica_consume_duration_ms, news_replica_upserted_total, news_replica_blocked_total, news_replica_skipped_total{reason} (stale/duplicate/missing), news_replica_unknown_enum_total{field}, news_replica_dlq_total, news_replica_lag_seconds (now − время события).
  • Traces: span приёма события → span обработки → span обращения к хранилищу; correlation из метаданных.
  • Alerts: лаг консюмеров identity > 1 мин (SLO README news); DLQ > 0; всплеск news_replica_unknown_enum_total (drift enum источника).
  • Безопасность: consumer под system-actor; реплика не tenanted (платформенные подписи, ADR-0052 п.3); топик pin.identity.users.v1 — доступ по m2m ACL Kafka (доверенные consumer-ы). Кросс-контекстной записи нет: consumer пишет только news.author_replicas по UserId.

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

  • Порядок per-user гарантирован Kafka key = UserId (оба события общего топика упорядочены между собой, user-upserted.md/user-blocked.md), поэтому UserBlocked приходит после регистрационного UserUpserted (Version = 1) — реплика к моменту blocked существует. Defensive skipped-missing-replica оставлен на случай старта группы не с earliest.
  • PartnerManager → AgencyManager — «ближайший» маппинг (identity публикует PartnerManager, news знает AgencyManager); фиксируется как assumption, требует сверки при появлении новых значений UserType.
  • Свежесть подписи — eventual (лаг консюмера); INV-REP-002 (эффект Blocked на выдачу) и INV-REP-003 (компании) применяются query-слоем, а не этим консюмером.
  • AvatarFileId в текущем контракте user-upserted отсутствует — реплика получает null до появления поля в источнике (additive-совместимо).

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

  • Предусловие: топик pin.identity.users.v1 доступен; consumer group news-identity-users активна; identity — единственный писатель топика (m2m ACL). Реплика может как существовать, так и отсутствовать (first-seen допустим).
  • Постусловие (успех upsert): строка AuthorReplica(UserId) содержит display-поля события, SourceVersion = event.Version, SyncedAt = event.CreatedAt; ключ идемпотентности {EventType}:{MessageId} зафиксирован; offset подтверждён. Никакие поля, кроме реплики подписи, не менялись — consumer пишет только news.author_replicas.
  • Постусловие (no-op ветки): при дубле/stale/отсутствии строки на blocked состояние реплики не меняется, offset подтверждён (ack), событие не попадает в DLQ.

Acceptance Criteria​

  • First-seen upsert. Дано реплики по UserId нет; когда приходит UserUpsertedKafkaEvent (Version = 1, Name.FirstName непуст); тогда создаётся строка AuthorReplica с DisplayName, Type, CompanyId, 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.
  • Блокировка. Дано существует AuthorReplica в Active; когда приходит UserBlockedKafkaEvent со свежей Version; тогда Status = Blocked, SourceVersion обновлён (эффект на выдачу — query-слой, INV-REP-002).
  • Blocked без реплики. Дано реплики по UserId нет; когда приходит UserBlockedKafkaEvent; тогда skipped-missing-replica, ack (подпись из blocked-события не восстановить).
  • Unknown enum (additive у источника). Дано UserType = PartnerManager (нет прямого соответствия); когда обрабатывается upsert; тогда Type = AgencyManager (ближайшее) + warning + news_replica_unknown_enum_total; событие не теряется. UserStatus unknown → Blocked (fail-closed).
  • Нарушение контракта payload. Дано обязательное поле отсутствует/пусто (напр. пустой FirstName в upsert); когда обрабатывается событие; тогда NEWS_REPLICA_INVALID_PAYLOAD, без retry, в DLQ + алерт.
  • Идемпотентность цепочки. Дано полный replay топика с earliest; когда все события переиграны; тогда финальное состояние реплик идентично (upsert по UserId + version-guard детерминированы).

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

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

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