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

Consumer: Identity Agent Membership Changed

Назначение​

Consumer слушает событие изменения членства агента в агентстве (ADR-0061: вступление Active / выход Left; ADR-0072: Reason/EffectiveAt) и обновляет в user-centric проекции AgentPerformanceEntity состояние членства строки пары (AgentUserId, CompanyId) — поля MembershipStatus + LeaveReason/LeftAt/EffectiveAt. При мульти-членстве (ADR-0070) у агента несколько строк одновременно: обрабатывается только строка своего CompanyId, прочие членства агента не трогаются. Накопленные (earned) метрики сделки/фиксации/комиссии строки не изменяются: они user-centric и сохраняются при выходе из агентства (ADR-0061 п.5, ADR-0070 п.6) — членство меняет лишь состояние строки, а не историю. Производный display-статус («независимый» / «в агентстве N») вычисляется на чтении из набора строк агента (см. agent-performance.md) — consumer его не хранит.

Consumer перечислен в контракте события identity (agent-membership-changed.md, таблица потребителей). Прекращение оснований комиссий текущего периода при Left (там же) реализуется штатным путём начислений (начисление создаётся по завершённой сделке с иммутабельным снапшотом AgencyCompanyId, не по текущему членству) — отдельной мутации здесь не требуется.

Идемпотентность — по идентификатору сообщения + монотонный guard по Version членства строки: устаревшее/переставленное событие не откатывает актуальное состояние своей строки. Порядок per-agent (key = AgentUserId) сохраняет причинную последовательность join/leave по каждому агентству агента.

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

ПолеЗначение
Сервис/контекстrewards (фоновый обработчик)
Обрабатываемое событиеpin.identity.agent-membership-changed
Topicpin.identity.agent-membership-changed.v1
Partitions / keykey = AgentUserId — строгий порядок join/leave per-agent (3 партиции, ADR-0056)
ContractVersion1; при несовпадении версии — warn-and-proceed (rewards_contract_mismatch_total)
Consumer grouppin-rewards-identity-membership
Produceridentity: accept-invitation / become-agent (join → Active), leave-agency / become-investor (leave → Left)
Эффект (событие → что меняется)состояние членства строки пары (AgentUserId, CompanyId) в проекции AgentPerformanceEntity (MembershipStatus + LeaveReason/LeftAt/EffectiveAt); прочие строки агента не трогаются

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

  • Из метаданных события доступны: идентификатор сообщения (для дедупликации), время события, Version, идентификатор и тип агрегата-источника (AggregateId = MembershipId, AggregateType = "AgentMembership").
  • Корреляция сквозного трейса восстанавливается из метаданных события.
  • Обработка — под системным актором вне tenant-контекста (platform-scope): проекция rewards.agent_performance не привязана к tenant (ADR-0061 п.5, ADR-0070 п.6).
  • Ключ идемпотентности — rewards:identity-membership:{MessageId}; доменный guard — payload.Version > LastMembershipVersion строки (AgentUserId, CompanyId).

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

Контракт — identity/events/agent-membership-changed.md. Ниже — поля, которые consumer реально читает.

Поле payloadТипОбязательностьКак используется (маппинг)
AgentUserIdGuidДаЧасть ключа строки (Upsert по паре (AgentUserId, CompanyId)); = Kafka key
CompanyIdGuidДаЧасть ключа строки — агентство, состояние которого меняем; строка своего CompanyId, прочие не трогаются
Statusstring (enum AgentMembershipStatus: Active/Left)ДаActive → MembershipStatus = Active (+ обнулить LeaveReason/LeftAt/EffectiveAt); Left → MembershipStatus = Left
Reasonstring? (enum LeaveReason: SelfLeave/ExcludedByAgency)НетПри Left → LeaveReason; unknown enum → мягкая деградация (warning, трактовать как состоявшийся выход)
EffectiveAtDateTimeOffset?НетПри Left → EffectiveAt (момент вступления выхода в силу; для отложенного увольнения ≠ LeftAt)
VersionlongДаМонотонный guard строки: применять только если > LastMembershipVersion строки (AgentUserId, CompanyId)
LeftAtDateTimeOffset?НетПри Left → LeftAt; LastActivityAt = max(..., LeftAt)
JoinedAtDateTimeOffsetДаМомент вступления (для LastActivityAt при Active; FirstActivityAt ??= для новой строки)
MembershipIdGuidДаТрассировка (AggregateId); в проекцию не пишется

Поле InvitationId consumer-ом не читается. Reason/EffectiveAt — additive-поля (ADR-0072), ContractVersion не меняется; при Status = Active они null.

Алгоритм​

  1. Контекст события. Восстановить метаданные события (идентификатор сообщения, агрегат-источник, время); сверить ContractVersion — при несовпадении warn-and-proceed. Проверить идемпотентность по ключу rewards:identity-membership:{MessageId}: дубль доставки → skipped-duplicate, пропуск (ack).
  2. Проверка входа. AgentUserId != Guid.Empty; CompanyId != Guid.Empty (часть ключа строки — обязателен при любом Status; пусто → REWARDS_MEMBERSHIP_EVENT_INVALID → DLQ); Status ∈ {Active, Left} (unknown enum → warning unknown-membership-status, fail-closed: не считать активным, трактовать строку как Left); при Left — Reason может быть unknown → warning unknown-leave-reason, мягкая деградация (сохранить строку Left, пропустить причину-специфичную логику, ADR-0072). Существование сущностей не проверяется (проекция создаётся лениво).
  3. Что читаем. Строку проекции по паре (AgentUserId, CompanyId) (без tenant-фильтра); её отсутствие допустимо — тогда создаётся новая строка этого агентства с нулевыми метриками. Прочие строки агента не читаются и не меняются.
  4. Бизнес-правила и переходы. Монотонный guard строки: если для существующей строки payload.Version <= LastMembershipVersion — событие устаревшее, skipped-stale-membership, отметка идемпотентности, пропуск. Иначе применить к строке пары: Status = Active → MembershipStatus = Active, LeaveReason = LeftAt = EffectiveAt = null, FirstActivityAt ??= JoinedAt, LastActivityAt = max(LastActivityAt, JoinedAt); Status = Left → MembershipStatus = Left, LeaveReason = payload.Reason (если известен), LeftAt = payload.LeftAt, EffectiveAt = payload.EffectiveAt, LastActivityAt = max(LastActivityAt, LeftAt). LastMembershipVersion = payload.Version; Version += 1, UpdatedAt = now. Earned-счётчики и суммы строки не изменяются; строки других агентств агента не затрагиваются.
  5. Что меняется. Одна строка проекции AgentPerformanceEntity пары (AgentUserId, CompanyId) создаётся или обновляется. Конкурентная доставка в двух инстансах — проигравший перечитывает и повторяет guard (двойного применения нет).
  6. Согласованность. Обновление проекции и отметка идемпотентности фиксируются в одном согласованном изменении; offset подтверждается после фиксации. В skip-ветках фиксируется только отметка идемпотентности.
  7. Результат. Успех → ack; REWARDS_MEMBERSHIP_EVENT_INVALID → DLQ + алерт; инфраструктурные сбои → retry (таблица ниже). Исходящих событий проекция не порождает.

Диаграмма​

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

  • Consumer-level: отметка по идентификатору сообщения (rewards:identity-membership:{MessageId}).
  • Business-level: монотонный guard по LastMembershipVersion строки — повтор/переупорядочивание не откатывает состояние; состояние каждой строки (AgentUserId, CompanyId) сходится к последней (max Version) записи членства этого агентства.
  • Replay: да. Реплей с offset 0 безопасен: применяется только событие с Version больше сохранённого строкой; ключ = AgentUserId гарантирует порядок per-agent (join/leave по каждому агентству в причинном порядке). Полный rebuild проекции — вместе с другими источниками (runbook ../runbooks/rewards-agent-performance-rebuild.md, по конвенции).

Retry и DLQ​

DLQ-топик группы: pin.rewards.identity-membership-dlq.v1.

СбойКодRetryDLQРучное действие
Инфраструктурный (PostgreSQL/Kafka недоступны)— (exception)Exponential backoff 1s→60s с паузой партицииНетRunbook: восстановить зависимость
Нарушение контракта (AgentUserId пуст; CompanyId пуст)REWARDS_MEMBERSHIP_EVENT_INVALIDНет (non-retriable)Да + алертРазбор с командой identity; после фикса — replay из DLQ
Unknown enum Status— (не ошибка)—НетWarning, fail-closed Left (строка не активна); мониторинг rewards_membership_unknown_status_total
Unknown enum Reason— (не ошибка)—НетWarning unknown-leave-reason; строка Left без причины-специфичной логики (ADR-0072); мониторинг rewards_membership_unknown_reason_total
Устаревшее событие (guard)— (skip)—НетНе требуется

Acceptance-критерии​

  • Вступление (Active). Дано: событие Status = Active с CompanyId, Version > LastMembershipVersion строки. Когда: обработка. Тогда: строка (AgentUserId, CompanyId) — MembershipStatus = Active, LeaveReason/LeftAt/EffectiveAt = null, FirstActivityAt ??= JoinedAt, LastActivityAt = max(..., JoinedAt), LastMembershipVersion = Version; earned-метрики не меняются; ack.
  • Выход (Left) с Reason. Дано: событие Status = Left, Reason = SelfLeave (или ExcludedByAgency), Version новее сохранённой строкой. Когда: обработка. Тогда: строка своего CompanyId — MembershipStatus = Left, LeaveReason = из события, LeftAt/EffectiveAt заполнены; earned-метрики строки сохранены (ADR-0061 п.5, ADR-0070 п.6).
  • Мульти-членство: прочие членства не теряются. Дано: агент состоит в агентствах A и B (две строки Active); приходит Left по A. Когда: обработка. Тогда: строка A → Left; строка B остаётся Active с прежними метриками; агрегат display-статуса — по-прежнему InAgency (есть активная B).
  • Несколько Active подряд. Дано: агент получает Active по A, затем Active по B (разные CompanyId). Когда: обработка. Тогда: две строки Active одновременно (INV-AP5); ни одна не затирает другую.
  • Ленивое создание. Дано: строки пары ещё нет. Когда: обработка. Тогда: создаётся новая строка (AgentUserId, CompanyId) с нулевыми метриками и состоянием членства из события.
  • Устаревшее/переставленное событие (guard). Дано: Version <= LastMembershipVersion строки. Когда: обработка. Тогда: skipped-stale-membership, состояние строки не откатывается, ack.
  • Дубль доставки. Дано: то же MessageId. Когда: обработка. Тогда: skipped-duplicate, ack.
  • Unknown Status (fail-closed). Дано: значение Status вне {Active, Left}. Когда: обработка. Тогда: warning unknown-membership-status, строка трактуется как Left (не активна), метрика rewards_membership_unknown_status_total.
  • Unknown Reason (мягкая деградация). Дано: Status = Left с неизвестным Reason. Когда: обработка. Тогда: warning unknown-leave-reason, строка Left (сам выход состоялся), причина-специфичная логика пропущена, метрика rewards_membership_unknown_reason_total.
  • Нарушение контракта. Дано: пустой AgentUserId или пустой CompanyId. Когда: обработка. Тогда: REWARDS_MEMBERSHIP_EVENT_INVALID → DLQ + алерт.
  • Replay. Дано: re-read с offset 0. Когда: обработка. Тогда: для каждой строки применяется только событие с max Version; состояние членства каждого агентства сходится к последней записи.

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

  • Logs (structured): consumed / skipped-duplicate / skipped-stale-membership / membership-applied / dead-lettered с полями MessageId, AgentUserId, CompanyId, Status, Reason, Version. PII/суммы отсутствуют (событие — только id + метки времени, класс Internal).
  • Metrics: rewards_membership_consume_duration_ms, rewards_membership_applied_total{status}, rewards_membership_skipped_stale_total, rewards_membership_unknown_status_total, rewards_membership_unknown_reason_total, rewards_identity_membership_lag_seconds, rewards_identity_membership_dlq_total.
  • Traces: span приёма события → span обработки → span обращения к хранилищу; корреляция из метаданных события.
  • Alerts: lag consumer-group > 5 мин; DLQ > 0.
  • Безопасность: system-actor, platform-scope (проекция вне tenant); tenant mismatch неприменим (нет tenant-поля); payload — доверенный внутренний топик identity, без untrusted-ввода.

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

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

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