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

Consumer: ReferralUserUpsertedConsumer (pin.identity.users.v1)

Назначение​

Consumer ReferralUserUpsertedConsumer обрабатывает событие pin.identity.user-upserted (регистрация = первое событие агрегата, Version = 1) и выполняет две обязанности:

  1. Реплика — апсерт ReferralUserReplica (referral.users): тип/статус/ФИО/телефон/дата регистрации пользователя для условий правил привязки (INV-REF-002/005, «Имя пустое», окно 4 ч) и admin-проекций.
  2. Триггер T1 — привязка при регистрации: при first-seen UserId с ReferrerUrl != null исполняет нормативный алгоритм привязки (attach-rules.md, правило R1

Привязка событийная: identity публикует ReferrerUrl в снапшоте пользователя, referral реагирует consumer-ом (ADR-0056: без синхронных межсервисных вызовов).

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

ПолеЗначение
Сервис/контекстreferral
ConsumerReferralUserUpsertedConsumer
Обрабатываемое событиеpin.identity.user-upserted — контракт user-upserted.md
Topicpin.identity.users.v1 (6 partitions, key = UserId — строгий порядок per-user)
ContractVersion1; при несовпадении версии — warn-and-proceed
Consumer groupreferral-identity-users (общая для обоих consumer-ов топика)
Produceridentity (регистрация СМС/lazy, обновление профиля, смена телефона, admin-API, talent-identity-sync)
Эффект (событие → что меняется)всегда — апсерт реплики ReferralUserReplica; при first-seen с ReferrerUrl != null — создаётся связь (Active) + запись журнала (Initial) + ReferralBondCreatedKafkaEvent; при владельце-агенте — ReferralLeadAttachRequestedKafkaEvent

Sibling того же топика и consumer group:

ПолеЗначение
ConsumerReferralUserBlockedConsumer
Обрабатываемое событиеpin.identity.user-blocked — контракт user-blocked.md
Эффектобновление ReferralUserReplica.Status = Blocked (guard G4: заблокированный пользователь не привязывается; заблокированный владелец ссылки не получает лидов — edge case attach-rules.md)

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

  • TenantId в identity-событиях отсутствует; tenant создаваемых referral-записей — TenantId ссылки (ADR-0052), реплики — CompanyId из payload.
  • Из метаданных события доступны: идентификатор сообщения (дедупликация), AggregateId = UserId, тип агрегата User, время события; время события с Version = 1 — момент регистрации (RegisteredAt реплики, BoundAt/RegisteredAt привязки).
  • Корреляция сквозного трейса восстанавливается из метаданных события.
  • Транспортная идемпотентность — по ключу pin.identity.user-upserted:{MessageId}; бизнес-идемпотентность привязки — first-seen по журналу Initial (INV-REF-R1), реплики — монотонный ReplicaVersion-guard.

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

Контракт события: user-upserted.md. Ниже — только поля, которые consumer реально читает, и их использование.

Поле payloadТипОбязательностьКак используется
UserIdGuidДаReferralUserReplica.Id (апсерт); ReferredUserId связи/журнала; ключ first-seen; guard G3 (UserId != OwnerUserId ссылки)
Typestring (enum UserType)ДаReferralUserReplica.Type; guard G2 (Investor); для владельца ссылки — INV-REF-002 (unknown → «не агент/застройщик», fail-closed)
CompanyIdGuid?НетReferralUserReplica.CompanyId = TenantId реплики
Statusstring (enum UserStatus)ДаReferralUserReplica.Status; guard G4 (unknown → Blocked, fail-closed)
NameFullNameSnapshotДаReferralUserReplica.Name; вычисление IsNameFilled = Name.FirstName.Trim().Length > 0
PhoneNumberstringДаReferralUserReplica.PhoneNumber (referral — доверенный consumer, m2m ACL топика); только admin-проекции, в логах +7***1234
RegisteredViastring (enum RegistrationSource)Датолько structured log (различение СМС/lazy-регистрации при разборе); в реплику не пишется
ReferrerUrlstring?Нетslug для триггера T1: null → ветка привязки не запускается; иначе резолв ссылки по Url
VersionlongДаReplicaVersion-guard (апсерт только при Version > ReplicaVersion); Version = 1 ⇔ регистрация → RegisteredAt = OccurredAt события

Email в payload отсутствует (PII-минимизация — см. README.md).

Алгоритм​

Нормативный процесс привязки — attach-rules.md; здесь — консюмер-специфика T1. Обработка состоит из двух веток: (A) реплика — на каждое событие, (B) привязка T1 — только при first-seen с ReferrerUrl != null.

  1. Контекст события. Восстановить метаданные (идентификатор сообщения, AggregateId = UserId, время); actor = system (m2m ACL Kafka). Несовпадение версии контракта → warning, продолжить. Транспортная идемпотентность по ключу pin.identity.user-upserted:{MessageId} — дубль → ack без обработки. Бизнес-идемпотентность ветки B — first-seen по журналу Initial (шаг 3).
  2. Проверка входа. UserId != Guid.Empty — иначе битый payload → DLQ. Unknown enum Type → сохранить строку в реплику, для правил трактовать как «не агент/застройщик»; unknown Status → вести себя как Blocked (оба — fail-closed). Для ветки B: ReferrerUrl нормализуется к lowercase и проверяется по ^[a-z0-9][a-z0-9-]{2,63}$ — мусор из cookie (untrusted вход) → warning + пропуск ветки B (ветка A выполняется всегда). Guard-ы G2/G3/G4 в T1 берут данные пользователя (Type, Status, Name) из payload — событие несёт свежий снапшот, порядок per-user гарантирован key-ом.
  3. Что читаем.
    • реплику ReferralUserReplica по UserId — нет → первая репликация (ветка A создаёт), есть → сравнение ReplicaVersion;
    • ветка B (только при ReferrerUrl != null): запись журнала Initial пользователя — если есть, это не first-seen, ветка B завершается no-op (INV-REF-R1: повторная доставка/повторный снапшот не плодит привязок);
    • ссылку по Url = slug, не удалённую — нет → guard G1: no-op ветки B + warning;
    • владельца ссылки из реплики (тип и Status для INV-REF-002); отсутствует (событие владельца ещё не доехало — другой key/partition того же топика) → REFERRAL_OWNER_REPLICA_MISSING → retry с backoff;
    • активную связь пользователя (ReferredUserId = UserId, Status = Active) — guard G5 + сериализация гонки с consumer-ом T2 (user-signed-in).
  4. Бизнес-правила.
    • Ветка A: guard Version > ReplicaVersion (иначе ветка A no-op — out-of-order защита); маппинг поле-в-поле по таблице реплики в attach-rules.md: Id = UserId, Type, CompanyId (= TenantId), Status, Name, PhoneNumber, IsNameFilled = Name.FirstName.Trim() != "", RegisteredAt = OccurredAt при Version == 1 (first-seen с Version > 1 → RegisteredAt = null, окно 4 ч fail-closed), ReplicaVersion = Version.
    • Ветка B: guard-ы G2 (Type == Investor из payload), G3 (UserId != OwnerUserId ссылки), G4 (Status ∉ {Blocked, Archived} из payload), G5 (нет активной связи) → правило R1; маппинг создаваемой связи (attach-rules.md шаг 4): ReferredUserId = UserId, ReferrerUserId = OwnerUserId ссылки, ReferralLinkId, TenantId = TenantId ссылки, BoundAt = OccurredAt, AppliedRule = R1; создание поднимает доменное ReferralBondCreatedEvent (guard INV-REF-B2); решение о лиде — INV-REF-002/R5: leadRequested = владелец Type ∈ {Agent, DeveloperManager} && Status == Active.
  5. Что меняется. Апсерт реплики (создание или обновление полей — с ReplicaVersion-guard). Ветка B: появляется связь и запись журнала (RegistrationType = Initial, AppliedRule = R1, LeadRequested = leadRequested, денормализация LinkUrl, ReferrerUserId, TenantId, RegisteredAt = OccurredAt — INV-REF-R4); связь и её запись журнала связываются друг с другом.
  6. Какие события фиксируются.
    • ReferralBondCreatedKafkaEvent (topic pin.referral.bonds.v1, ContractVersion = 1, key = ReferredUserId) — payload из полей связи + Channel ссылки, контракт events/bond-created.md;
    • при leadRequested — ReferralLeadAttachRequestedKafkaEvent (topic pin.referral.lead-attachments.v1, ContractVersion = 1, key = InvestorUserId = UserId) — payload: AgentUserId = OwnerUserId ссылки, AgentCompanyId = TenantId ссылки, ReferralLinkId, Channel, AppliedRule = R1, BondId, RequestedAt; контракт events/lead-attach-requested.md.
  7. Согласованность. Реплика + связь + журнал + отметка идемпотентности + исходящие события фиксируются в одном согласованном изменении; offset подтверждается после фиксации. Конфликт partial unique ux_referral_bonds_referred_active или ux_referral_registrations_initial (гонка с T2) → идемпотентное завершение: связь/журнал уже созданы конкурентом, событие не дублируется.
  8. Результат. Успех → ack. Structured log: UserId, AppliedRule, guard-исход (G1..G5/first-seen-skip/replica-only), LinkId, BondId — без PII. Контролируемые ошибки (REFERRAL_OWNER_REPLICA_MISSING, транзиентные) → retry; невосстановимые (битый payload) → DLQ.

Диаграмма​

Обработка user-blocked (sibling-consumer)​

ReferralUserBlockedConsumer — тот же топик и consumer group, отдельный тип события. Алгоритм (сокращённая форма — операция однополевая):

  1. Контекст из метаданных (идентификатор сообщения, AggregateId = UserId); идемпотентность — ключ pin.identity.user-blocked:{MessageId}.
  2. Проверка входа: UserId != Guid.Empty; unknown enum Reason игнорируется (в реплику не пишется).
  3. Что читаем: реплику по UserId.
  4. Реплика отсутствует → no-op с warning (снапшот user-upserted придёт и принесёт Status); guard Version > ReplicaVersion (защита от replay).
  5. Что меняется: Status = Blocked, ReplicaVersion = Version.
  6. Событий не публикует (реплика событий не порождает; активные связи блокировка не закрывает — оценка оснований выплат при заблокированном участнике — зона rewards, решение № 7 README.md).
  7. Согласованность: обновление реплики + отметка идемпотентности — в одном изменении.
  8. Успех → ack; лог UserId, Version. Эффект для правил: guard G4 отсекает привязку заблокированного, INV-REF-002 перестаёт создавать лиды заблокированному владельцу ссылок.

Разблокировка отдельного события не имеет — приходит user-upserted со Status = Active (ветка A основного consumer-а).

Промежуточное состояние и downstream​

  • Что видит пользователь. Ветка A (реплика) и ветка B (привязка T1) фиксируются асинхронно от регистрации. Между регистрацией по ссылке и коммитом связи (SLO < 1 мин, README.md) операция get-my-referrer.md возвращает hasReferrer = false, а реплика ReferralUserReplica приглашённого может ещё не наполниться (имя реферера в других выдачах — null). После коммита обе ветки становятся согласованными (read-after-commit).
  • Downstream-порядок. ReferralBondCreatedKafkaEvent (pin.referral.bonds.v1, key = ReferredUserId) потребляют rewards (основание выплат), notifications (СМС рефереру «по вашей ссылке зарегистрировался пользователь»), messenger (контакт «агент ↔ инвестор»); ReferralLeadAttachRequestedKafkaEvent (pin.referral.lead-attachments.v1, key = InvestorUserId) потребляет crm (карточка лида). Ключ = пользователь ⇒ per-user порядок исходов сохраняется.

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

  • Транспортный ключ: pin.identity.user-upserted:{MessageId} / pin.identity.user-blocked:{MessageId} (отметка в одном изменении с результатом). Поведение дубля: ack без обработки.
  • Бизнес-ключи: привязка — first-seen по ux_referral_registrations_initial (ReferredUserId, INV-REF-R1) + partial unique ux_referral_bonds_referred_active (INV-REF-001); реплика — монотонный ReplicaVersion (out-of-order/replay событий → no-op).
  • Replay: да — безопасен (backfill реплики referral.users при внедрении выполняется replay-ем топика с начала: реплики строятся, повторные привязки не создаются first-seen-ом). Runbook — runbooks/referral-consumer-lag.md (реестр в README.md).

Retry и DLQ​

СбойRetryDLQРучное действие
Транзиентная ошибка хранилища / таймаутДа — экспоненциальный backoff (offset не коммитится)Нет—
Реплика владельца ссылки отсутствует (REFERRAL_OWNER_REPLICA_MISSING)Да — backoff до 15 мин (событие владельца в другой partition ещё не применено / идёт backfill)Да, после исчерпанияПроверить lag топика; после backfill-а реплик — redrive из DLQ
Битый payload (UserId = Guid.Empty, десериализация)НетДаРазбор с командой identity (контракт), redrive после фикса producer-а
Конфликт unique-индексов привязки (гонка с T2)Нет — идемпотентное завершение внутри обработки (шаг 7)Нет—
Несовпадение версии контрактаНе сбой: warn-and-proceedНетПлановая миграция контракта
Slug из ReferrerUrl не резолвится (G1)Нет — no-op ветки B, warningНетРост метрики no-op G1 → alert (битые cookie), см. README

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

  • Logs: consumed / skipped-duplicate (MessageId) / replica-upserted (UserId, ReplicaVersion) / attach-outcome (AppliedRule = R1 | guard-код no-op) / failed. PII не логируется: ФИО — нет, телефон — только маскированный +7***1234.
  • Metrics: длительность обработки, retries, DLQ, дубликаты; счётчики срабатываний R1 и no-op по guard-ам G1–G5; consumer lag (alert > 5 мин — SLO привязки < 1 мин от регистрации до ReferralBondCreatedKafkaEvent, откуда СМС рефереру через notifications).
  • Traces: span приёма события → span обработки → spans обращений к хранилищу; correlation из метаданных события; исходящие события получают свой trace.
  • Безопасность: actor — system; доступ к топику pin.identity.users.v1 — m2m ACL Kafka (referral — доверенный consumer, получает PhoneNumber); trusted/untrusted boundary — ReferrerUrl (slug из cookie) валидируется по формату (шаг 2); tenant создаваемых записей — из ссылки (ADR-0052), tenant mismatch невозможен (identity-события не tenant-скоупированы).

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

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

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