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

Consumer: ReferralUserSignedInConsumer (pin.identity.sign-ins.v1)

Назначение​

Consumer ReferralUserSignedInConsumer реализует Триггер T2 — привязку при авторизации инвестора, пришедшего по реферальной ссылке. При ReferrerUrl != null в событии входа исполняется нормативный алгоритм привязки (attach-rules.md) с выбором правила по данным реплики ReferralUserReplica:

  • R2 AuthorizationLazyWindow (IsNameFilled = false ИЛИ now() - RegisteredAt < 4 ч): создаётся ReferralBondEntity (Active, AppliedRule = R2), журнальная запись ReferralRegistrationEntity (Initial, R2) и публикуется ReferralBondCreatedKafkaEvent (events/bond-created.md);
  • R3 AuthorizationSelfRegistered (IsNameFilled = true И >= 4 ч; RegisteredAt = null → окно истекло, fail-closed): связь не создаётся, журнал не пишется — только лид;
  • R5/INV-REF-002 (модификатор): ReferralLeadAttachRequestedKafkaEvent (events/lead-attach-requested.md) публикуется только если владелец ссылки — Agent/DeveloperManager и Status = Active.

Отличие от T1 (user-upserted.md): consumer T2 реплику не наполняет (её дом — consumer user-upserted того же UserId, другой топик/группа), а только читает ReferralUserReplica для условий правил; отсутствие реплики → retry (см. Retry и DLQ). Поле ReferrerUrl в событии входа — additive: identity подставляет его из подтверждения входа существующего пользователя (решение № 4 README.md; до его появления T2 не срабатывает).

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

ПолеЗначение
Сервис/контекстreferral
ConsumerReferralUserSignedInConsumer
Обрабатываемое событиеpin.identity.user-signed-in — контракт user-signed-in.md
Topicpin.identity.sign-ins.v1 (6 partitions, key = UserId — строгий порядок per-user)
ContractVersion1; при несовпадении версии — warn-and-proceed (событие — факт, version-guard не нужен)
Consumer groupreferral-identity-sign-ins — отдельная от группы топика users (referral-identity-users, user-upserted.md): независимый offset/lag; кросс-топик порядок users.v1 → sign-ins.v1 не гарантирован (отсюда retry на отсутствующую реплику)
Produceridentity (подтверждение входа / вход по паролю / вход пин-кодом)
Эффект (событие → что меняется)при R2 — создаётся реферальная связь (Active) + запись журнала (Initial) + публикуется ReferralBondCreatedKafkaEvent; при R2/R3 с владельцем-агентом — публикуется ReferralLeadAttachRequestedKafkaEvent; собственное состояние (реплику) consumer не меняет

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

  • Из метаданных события доступны: идентификатор сообщения (дедупликация), AggregateId = UserId, тип агрегата User, время события OccurredAt (= момент входа; используется как BoundAt/RegisteredAt создаваемой связи по R2).
  • TenantId у identity-событий отсутствует; tenant создаваемых referral-записей — TenantId ссылки (ADR-0052).
  • Actor = system (m2m ACL Kafka на топик pin.identity.sign-ins.v1).
  • Корреляция сквозного трейса восстанавливается из метаданных события.
  • Транспортная идемпотентность — по ключу pin.identity.user-signed-in:{MessageId}; бизнес-идемпотентность привязки — guard G5 + partial unique ux_referral_bonds_referred_active (first-seen не применяется: R3/R4 журнал Initial не пишут, а R2 допускает повторные входы того же пользователя).

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

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

Поле payloadТипОбязательностьКак используется
UserIdGuidДаключ поиска реплики ReferralUserReplica; ReferredUserId связи/журнала; guard G3 (UserId != OwnerUserId ссылки); Kafka key
ReferrerUrlstring?Нет (additive)slug триггера T2: null → процесс привязки не запускается (ack без эффекта); иначе резолв ссылки по Url
OccurredAtDateTimeOffsetДаBoundAt/RegisteredAt создаваемой связи и журнала (момент входа)

Условия правил (Type для G2, Status для G4, IsNameFilled и RegisteredAt для окна R2/R3) в payload события входа отсутствуют — берутся из реплики ReferralUserReplica (attach-rules.md), наполняемой consumer-ом user-upserted. Прочие поля события (SessionId, SignInMethod, DevicePlatform, IsFirstSignIn) consumer referral не читает.

Алгоритм​

Нормативный процесс привязки и матрица — attach-rules.md; здесь — консюмер-специфика T2. Ветка привязки исполняется только при ReferrerUrl != null.

  1. Контекст события. Восстановить метаданные (идентификатор сообщения, AggregateId = UserId, время); actor = system. Несовпадение версии контракта → warning, продолжить. Проверить транспортную идемпотентность по ключу pin.identity.user-signed-in:{MessageId} — дубль → ack без обработки.
  2. Проверка входа. UserId != Guid.Empty — иначе битый payload → DLQ (retry бессмыслен). ReferrerUrl == null → процесс не запускается: ack (в отличие от T1, у consumer-а входа нет обязанности по реплике — делать нечего). ReferrerUrl нормализуется к lowercase и проверяется по ^[a-z0-9][a-z0-9-]{2,63}$ — мусор из cookie (untrusted вход) → warning + no-op ack.
  3. Что читаем. По реплике и хранилищу referral:
    • реплику пользователя ReferralUserReplica по UserId — источник условий правил (Type, Status, IsNameFilled, RegisteredAt); отсутствует → REFERRAL_USER_REPLICA_MISSING → retry с backoff (событие user-upserted того же UserId из другого топика/группы ещё не применено или идёт backfill);
    • ссылку по Url = slug, не удалённую — нет → guard G1: no-op ветки привязки + warning;
    • владельца ссылки из реплики (тип и Status для INV-REF-002); отсутствует → REFERRAL_OWNER_REPLICA_MISSING → retry;
    • активную связь пользователя (ReferredUserId = UserId, Status = Active) — guard G5 + сериализация гонки с consumer-ом T1 (user-upserted), выполняющимся параллельно в своей группе.
  4. Бизнес-правила.
    • Guard-ы (из реплики): G2 (Type = Investor), G3 (UserId != OwnerUserId ссылки), G4 (Status ∉ {Blocked, Archived}) — нарушение → no-op с warning (INV-REF-005/004).
    • Выбор правила (матрица): окно lazy-регистрации = IsNameFilled == false ИЛИ (RegisteredAt != null И now() - RegisteredAt < Referral:LazyRegistrationWindow, default 04:00:00); RegisteredAt == null → окно истекло (fail-closed). Истинно → R2; иначе → R3.
    • leadRequested = владелец Type ∈ {Agent, DeveloperManager} && Status == Active (R5/INV-REF-002) — общий для R2 и R3.
    • R2 при отсутствии активной связи (G5): создаётся связь (guard INV-REF-B2) — ReferredUserId = UserId, ReferrerUserId = OwnerUserId ссылки, ReferralLinkId, TenantId = TenantId ссылки, BoundAt = OccurredAt, AppliedRule = R2; поднимается доменное ReferralBondCreatedEvent.
    • R3, либо R2 при существующей активной связи: связь не создаётся (INV-REF-003 / G5), журнал не пишется — далее только лид по leadRequested.
  5. Что меняется. Только в ветке «R2 + нет активной связи»: появляется связь и запись журнала (RegistrationType = Initial, AppliedRule = R2, LeadRequested = leadRequested, денормализация LinkUrl, ReferrerUserId, TenantId, RegisteredAt = OccurredAt — INV-REF-R4). R3/G5-ветка — без изменений состояния (реплику consumer T2 не трогает).
  6. Какие события фиксируются.
    • при созданной связи (R2) — 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 (R2/R3), BondId (создана в этом проходе) либо null (R3, или R2 при существующей активной связи), RequestedAt; контракт events/lead-attach-requested.md.
  7. Согласованность. Связь + журнал + отметка идемпотентности + исходящие события фиксируются в одном согласованном изменении; offset подтверждается после фиксации. Конфликт partial unique ux_referral_bonds_referred_active (гонка с T1) → идемпотентное завершение: связь уже создана конкурентом (G5-исход), ReferralBondCreatedKafkaEvent не дублируется; лид — по INV-REF-002.
  8. Результат. Успех → ack. Structured log: UserId, AppliedRule (R2/R3), guard-исход (G1..G5/window), LinkId, BondId — без PII. Контролируемые ошибки (REFERRAL_USER_REPLICA_MISSING, REFERRAL_OWNER_REPLICA_MISSING, транзиентные) → retry; невосстановимые (битый payload) → DLQ.

Диаграмма​

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

  • Что видит пользователь. Привязка по T2 фиксируется асинхронно: между входом по ссылке и коммитом связи (SLO < 1 мин, README.md) операция get-my-referrer.md возвращает hasReferrer = false, а партнёрская статистика владельца ещё не учитывает новую связь. После коммита обе операции становятся согласованными (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-signed-in:{MessageId} (отметка в одном изменении с результатом). Поведение дубля: ack без обработки.
  • Бизнес-ключ: guard G5 + partial unique ux_referral_bonds_referred_active (ReferredUserId, INV-REF-001). First-seen по журналу Initial не применяется: R3/R4 журнал не пишут, а R2 допускает повторные входы — защита от двойной связи целиком на G5 + unique-индексе (проигравший гонку → идемпотентный no-op). Повтор R2/R3 при уже существующей активной связи: связь не создаётся, но лид отправляется идемпотентно (дедуп CRM по паре InvestorUserId+AgentUserId).
  • Replay: да — безопасен (G5 + unique-индекс не плодят связи; лиды идемпотентны на стороне CRM). Runbook — runbooks/referral-consumer-lag.md (реестр в README.md).

Retry и DLQ​

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

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

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

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

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

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