Consumer: Selection Shared (SelectionSharedConsumer)

Один консюмер на входной топик pin.selections.selection-shared.v1, дописывающий ленту событий лида (LeadEventEntity) записями о подборках: агент поделился (SelectionAttached) или клиент открыл (SelectionViewed).

Назначение

Консюмер Selection Shared слушает pin.selections.selection-shared.v1 и дописывает ленту событий лида (LeadEventEntity) двумя типами по дискриминатору Action: SelectionAttached (агент поделился подборкой с клиентом — UC-1 «в лида была добавлена подборка») и SelectionViewed (клиент открыл подборку — UC-1 «лид просмотрел подборку»). Оба типа несут одну payload-модель LeadSelectionPayload (дом — lead.md, раздел «Enum LeadEventType и typed payload»). Меняется только лента лида; сам лид не создаётся и статус не меняется. Идемпотентно по идентификатору сообщения.

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

Поле	Значение
Сервис/контекст	crm
Consumer	Selection Shared
Обрабатываемое событие	pin.selections.selection-shared
Topic	pin.selections.selection-shared.v1
ContractVersion	ожидается 1; при несовпадении версии события — warn-and-proceed
Consumer group	crm-selections
Producer	selections — events/selection-shared.md (активен; share-операция POST /selections/{id}/share)
Эффект (событие → что меняется)	Дописывает в ленту лида LeadEventEntity (SelectionAttached/SelectionViewed)

Статус producer-контракта

Событие pin.selections.selection-shared.v1 публикуется сервисом selections — контракт selections/events/selection-shared.md (share-операция POST /selections/{id}/share, UC-SEL-005). Producer эмитит Action = Shared (агент отправил подборку клиенту). Значение Opened (клиент открыл → SelectionViewed) — зарезервированное в контракте: этим producer-ом пока не эмитится (просмотры по публичной ссылке трекает user-activity, ADR-0054), консюмер мапит его заранее (forward-compat, введение фактической публикации — additive без bump). Карты — crm/README.md, lead.md.

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

• Из метаданных события доступны: идентификатор сообщения MessageId (для дедупликации), время события CreatedAt (= OccurredAt), версия контракта, идентификатор подборки (SelectionId).
• Актор — системный контекст консюмера (не пользовательский); CreatedByUserId дочерней LeadEventEntity = null (лента наполнена консюмером, INV лида: «null — создан консюмером»).
• Tenant — не из события: selections не tenanted (ADR-0052). TenantId ленты берётся из найденного лида (Lead.TenantId = AgencyCompanyId); события пишутся в tenant агентства.
• Корреляция сквозного трейса восстанавливается из метаданных события.
• Бизнес-идемпотентность — по ключу crm:selection-shared:{MessageId}; страховка на уровне хранилища — unique partial ux_lead_events_source_message_id по LeadEventEntity.SourceMessageId (дом — lead.md).

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

Producer-контракт (дом — selections/events/selection-shared.md): событие SelectionShared, topic pin.selections.selection-shared.v1, key = SelectionId. Ниже — поля, которые consumer реально читает, и их маппинг в ленту лида (lead.md).

Поле payload	Тип	Обязательность	Как используется (маппинг)
SelectionId	Guid	Да	→ LeadSelectionPayload.SelectionId; = AggregateId; часть бизнес-ключа лога
SelectionTitle	string	Да	→ LeadSelectionPayload.SelectionTitle (varchar(256); длиннее — truncate + warn, событие не теряется)
PropertyObjectsCount	int	Да	→ LeadSelectionPayload.PropertyObjectsCount (>= 0)
Action	string (enum SelectionShareAction: Shared/Opened)	Да	Дискриминатор: Shared → LeadEventType.SelectionAttached, Opened → LeadEventType.SelectionViewed; неизвестное значение → non-retriable DLQ (CRM_SELECTION_UNKNOWN_ACTION, unknown-enum)
AgentUserId	Guid	Да	Резолв активного лида: Lead.ExecutorAgentUserId (сторона исполнителя пары ux_leads_active_pair)
InvestorUserId	Guid	Да	Резолв активного лида: Lead.PrimaryInvestorUserId
OccurredAt	DateTimeOffset	Да	→ LeadEventEntity.OccurredAt; обновляет Lead.LastEventAt = greatest(...)

Payload не содержит PII (только id, название подборки, счётчик). Состав подборки (PropertyObjectId-список) в событие не входит — при необходимости CRM читает его синхронным GET /internal/selections/{id}.

Алгоритм

1. Контекст события. Восстановить метаданные события (идентификатор сообщения, идентификатор подборки, время, версия контракта); при несовпадении версии — warn-and-proceed (crm_selection_shared_contract_mismatch_total). Проверить идемпотентность по ключу crm:selection-shared:{MessageId}: если событие уже обработано → пропустить как дубль (ack).
2. Проверка входа. SelectionId, AgentUserId, InvestorUserId заданы; PropertyObjectsCount >= 0; SelectionTitle непустая (> 256 — truncate + warn, не ошибка валидации); значение Action распознано как SelectionShareAction — неизвестное → CRM_SELECTION_UNKNOWN_ACTION (non-retriable → DLQ + алерт). Существование лида — шаг 3 (отсутствие допустимо → контролируемая ветка). Допустимость состояния (Status = Active, INV-LEAD-2) — часть условия резолва и guard метода агрегата (шаг 4).
3. Что читаем. Найти единственный активный лид пары (гарантия ux_leads_active_pair) по ExecutorAgentUserId, PrimaryInvestorUserId, Status = Active (резолв по событийным AgentUserId/InvestorUserId). Если лида нет → CRM_SELECTION_LEAD_NOT_FOUND (retriable: out-of-order против создания/продления лида; после исчерпания — DLQ, replay). Отдельного чтения ленты не требуется (append-only, идемпотентность гарантируется unique-индексом при записи).
4. Правила и маппинг поле-в-поле.
   • Тип события: Shared → LeadEventType.SelectionAttached, Opened → LeadEventType.SelectionViewed.
   • Payload LeadSelectionPayload { SelectionId, SelectionTitle, PropertyObjectsCount } (ContractVersion = 1).
   • Дописать в ленту через метод агрегата (INV-LEAD-2: разрешён только при Status = Active, иначе CRM_LEAD_NOT_ACTIVE): создаётся LeadEventEntity (LeadId, TenantId = lead.TenantId, Type, OccurredAt, SourceMessageId = MessageId, Payload, CreatedAt), выставляется Lead.LastEventAt = greatest(LastEventAt, OccurredAt), поднимается доменное событие LeadEventAdded (LeadId, EventId, Type).
5. Что меняется. Появляется строка LeadEventEntity, обновляется LastEventAt/UpdatedAt лида (правка не должна затирать более свежее состояние). Запись события ленты и отметка идемпотентности фиксируются в одном согласованном изменении; unique partial ux_lead_events_source_message_id ловит дубликат при replay после сбоя → трактуется как idempotent no-op. Собственных интеграционных событий консюмер не публикует (LeadEventAdded остаётся доменным; downstream-топика на «событие ленты» нет). Offset подтверждается только после успешной фиксации.
6. Результат. Успех → ack; CRM_SELECTION_LEAD_NOT_FOUND (retriable) → retry → DLQ; CRM_SELECTION_UNKNOWN_ACTION (non-retriable) → DLQ + алерт; нарушение unique SourceMessageId → skipped-duplicate, ack. Метрики/логи — раздел «Наблюдаемость».

Диаграмма

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

• На уровне консюмера: ключ crm:selection-shared:{MessageId} — повторная доставка не дублирует запись ленты.
• На уровне хранилища: LeadEventEntity.SourceMessageId = MessageId + unique partial ux_lead_events_source_message_id — вторая страховка на случай гонки/replay мимо кеша ключей; конфликт вставки → idempotent no-op (ack).
• Replay: да — повторное чтение топика с offset 0 безопасно (обе защиты по MessageId). Точечный replay из DLQ — после появления/продления целевого лида (CRM_SELECTION_LEAD_NOT_FOUND).

Retry и DLQ

Сбой	Код	Retry	DLQ	Ручное действие
Инфраструктурный (PostgreSQL/Kafka недоступны, timeout)	— (exception)	Бесконечный exponential backoff с паузой партиции	Нет	Восстановить зависимость (runbook crm)
Активный лид пары не найден (out-of-order / лид Expired/Archived)	CRM_SELECTION_LEAD_NOT_FOUND	Bounded backoff (30s → 16m)	Да, после исчерпания	Проверить лид пары (agent, investor); replay из DLQ после создания/продления
Неизвестное значение Action	CRM_SELECTION_UNKNOWN_ACTION	Нет (non-retriable)	Да + алерт	Обновить маппинг enum (релиз) → replay
Дубль (SourceMessageId unique / idempotency-ключ)	—	Нет	Нет	Нет — skipped-duplicate, ack
Конкурентная правка лида (гонка с операцией над лидом)	—	3 попытки немедленно (перечитать лид)	После исчерпания	Расследование гонки

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

• Logs (structured): consumed / skipped-duplicate / attached / viewed / skipped-lead-not-found / dead-lettered с полями MessageId, SelectionId, LeadId, Action, AgentUserId, InvestorUserId (id-поля не PII).
• Metrics: crm_selection_shared_consume_duration_ms, crm_selection_shared_lag_seconds (now − envelope.CreatedAt), crm_selection_shared_lead_not_found_total, crm_selection_shared_retries_total, crm_selection_shared_dlq_total, crm_selection_shared_duplicates_total, crm_selection_shared_contract_mismatch_total.
• Traces: span приёма события → span обработки → spans обращений к хранилищу; корреляция из метаданных события.
• Alerts: лаг > 5 мин (SLO README crm); DLQ > 0; всплеск lead_not_found (сигнал битого producer-контракта или гонки).
• PII / masking: payload PII не содержит (id + SelectionTitle + счётчик); маскирование не требуется.
• Tenant/scope: TenantId не берётся из конверта (selections не tenanted) — устанавливается из найденного лида (AgencyCompanyId); чужой tenant по построению невозможен (адресация по паре agent↔investor активного лида). Пишет только system-actor консюмера; публичных мутаций ленты нет.

Acceptance Criteria

• AC-1 (агент поделился). Дано: активный лид пары (ExecutorAgentUserId, PrimaryInvestorUserId) существует. Когда: событие с Action = Shared. Тогда: в ленту дописывается LeadEventEntity типа SelectionAttached с LeadSelectionPayload, Lead.LastEventAt = greatest(...), ack.
• AC-2 (клиент открыл — forward-compat). Дано: активный лид пары есть. Когда: событие с Action = Opened (зарезервировано в контракте). Тогда: дописывается SelectionViewed, ack (без bump контракта при вводе фактической публикации).
• AC-3 (лид не найден). Дано: активного лида пары нет (out-of-order / лид Expired/Archived). Когда: событие. Тогда: CRM_SELECTION_LEAD_NOT_FOUND — ретраи (30s→16m), после исчерпания DLQ; replay из DLQ после создания/продления лида.
• AC-4 (неизвестный Action). Дано: Action вне enum SelectionShareAction. Когда: событие. Тогда: CRM_SELECTION_UNKNOWN_ACTION (non-retriable) → DLQ + алерт.
• AC-5 (дубль по сообщению). Дано: событие с уже виденным MessageId/SourceMessageId. Когда: повторная доставка/replay. Тогда: skipped-duplicate (либо конфликт ux_lead_events_source_message_id → idempotent no-op), ленты не дублируются, ack.
• AC-6 (лид не активен). Дано: лид перешёл из Active между резолвом и записью (INV-LEAD-2). Когда: запись. Тогда: CRM_LEAD_NOT_ACTIVE, событие ленты не добавляется.

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

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

• API (2): GET /api/agent/crm/leads/{leadId}/events — Лента событий лида, POST /api/agent/crm/deals/{dealId}/selection — Отправить подборку в сделку
• Use Cases (1): UC-SEL-005. Агент отправляет подборку клиенту в лиде/сделке
• События (2): События контекста selections: карта топиков и отклонённые события, Event Contract: pin.selections.selection-shared (SelectionSharedKafkaEvent)
• Консюмеры (1): Consumer: User-Activity Actions (crm-user-activity-actions)
• index.md (1): Selections Service (Pin.Selections)

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

• Целевой агрегат и все поля ленты: domain/lead.md (LeadEventEntity, LeadSelectionPayload, LeadEventType.SelectionAttached/SelectionViewed, INV-LEAD-2, ux_lead_events_source_message_id).
• Карта консюмеров/событий контекста: ../README.md.
• Producer-контракт: selections/events/selection-shared.md (share-операция — UC-SEL-005); карта топиков — selections/events/README.md; соседний контракт-эталон конверта — selection-created.md.
• Шаблон: consumer-template.md. Решения: ADR-0052 (tenancy), ADR-0056 (simplicity/polling-lifecycle лида).