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

Base Entities, Documents, Events — Практическое Руководство

Назначение​

Документ описывает базовые типы, interfaces, abstract классы и record types, которые используются во всех сервисах Pin: сущности, документы, Kafka envelope, event contracts. Это расширение Base Domain And Contract Models с практическими примерами и правилами применения.

Базовые Entity Contracts​

Иерархия Interfaces​

IEntity<TId>
  └── IAggregateRoot<TId>
        ├── IHasCreatedAt
        ├── IHasUpdatedAt
        ├── IHasConcurrencyToken
        ├── ITenantScoped
        ├── IBrandScoped
        ├── IUserOwned
        ├── IPartnerOwned
        └── ISoftDeletable

Определения Interfaces​

// Talent.Common.Domain

public interface IEntity<TId>
{
    TId Id { get; }
}

public interface IAggregateRoot<TId> : IEntity<TId>
{
    IReadOnlyList<IDomainEvent> DomainEvents { get; }
    IReadOnlyList<IDomainEvent> DequeueDomainEvents();
}

public interface IHasCreatedAt
{
    DateTimeOffset CreatedAt { get; }
}

public interface IHasUpdatedAt
{
    DateTimeOffset UpdatedAt { get; }
}

public interface IHasConcurrencyToken
{
    uint Version { get; }  // EF xmin / rowversion / optimistic token
}

public interface ITenantScoped
{
    TenantId TenantId { get; }
}

public interface IBrandScoped
{
    BrandId BrandId { get; }
}

public interface IUserOwned
{
    UserId UserId { get; }
}

public interface IPartnerOwned
{
    PartnerId PartnerId { get; }
}

public interface ISoftDeletable
{
    DateTimeOffset? DeletedAt { get; }
    ActorId? DeletedByActorId { get; }
    bool IsDeleted => DeletedAt.HasValue;
}

Таблица Применения​

Тип СущностиОбязательные Interfaces
Любая persisted entityIEntity<TId>, IHasCreatedAt
Mutable entity+ IHasUpdatedAt
Aggregate root+ IAggregateRoot<TId>
Конкурентный aggregate+ IHasConcurrencyToken
Tenant-owned+ ITenantScoped
Brand-facing+ IBrandScoped
Данные игрока+ IUserOwned
Данные партнёра+ IPartnerOwned
Soft-deletable dictionary+ ISoftDeletable

Базовые Абстрактные Классы​

// Talent.Common.Domain

public abstract class Entity<TId> : IEntity<TId>, IHasCreatedAt
{
    public TId Id { get; protected set; } = default!;
    public DateTimeOffset CreatedAt { get; protected set; }

    protected Entity() => CreatedAt = DateTimeOffset.UtcNow;
}

public abstract class AggregateRoot<TId> : Entity<TId>, IAggregateRoot<TId>, IHasUpdatedAt, IHasConcurrencyToken
{
    private readonly List<IDomainEvent> _domainEvents = new();
    public IReadOnlyList<IDomainEvent> DomainEvents => _domainEvents.AsReadOnly();
    public DateTimeOffset UpdatedAt { get; protected set; }
    public uint Version { get; private set; }  // mapped to EF xmin

    protected void RaiseDomainEvent(IDomainEvent domainEvent)
        => _domainEvents.Add(domainEvent);

    public IReadOnlyList<IDomainEvent> DequeueDomainEvents()
    {
        var events = _domainEvents.ToList();
        _domainEvents.Clear();
        return events;
    }
}

// Пример: SoftDeletableEntity для admin dictionaries
public abstract class SoftDeletableEntity<TId> : Entity<TId>, ISoftDeletable, IHasUpdatedAt
{
    public DateTimeOffset UpdatedAt { get; protected set; }
    public DateTimeOffset? DeletedAt { get; private set; }
    public ActorId? DeletedByActorId { get; private set; }

    public void SoftDelete(ActorId deletedBy)
    {
        DeletedAt = DateTimeOffset.UtcNow;
        DeletedByActorId = deletedBy;
        UpdatedAt = DateTimeOffset.UtcNow;
    }
}

Status Entity Pattern​

Сущности с жизненным циклом имеют typed status enum и метод перехода:

// Паттерн для stateful aggregate
public abstract class StatusEntity<TId, TStatus> : AggregateRoot<TId>
    where TStatus : struct, Enum
{
    public TStatus Status { get; protected set; }

    protected Result TransitionTo(TStatus newStatus, Func<TStatus, bool> guard)
    {
        if (!guard(Status))
            return Result.Fail(DomainErrors.InvalidStatusTransition(Status.ToString(), newStatus.ToString()));
        Status = newStatus;
        UpdatedAt = DateTimeOffset.UtcNow;
        return Result.Ok();
    }
}

Document Types​

VersionedJsonDocument (PostgreSQL JSONB)​

Для хранения сложных typed объектов в JSONB:

// Talent.Common.Infrastructure

public abstract class VersionedJsonDocument<TPayload>
{
    public int SchemaVersion { get; set; }
    public TPayload Data { get; set; } = default!;
    public DateTimeOffset StoredAt { get; set; }

    public abstract int CurrentSchemaVersion { get; }

    // Обязательный compatibility reader при смене версии
    public abstract TPayload MigrateIfNeeded(TPayload data, int fromVersion);
}

Правила:

  • Каждый JSONB-документ имеет SchemaVersion, типизированный TPayload, валидатор и migration reader.
  • При изменении структуры — новая версия + migration метод, а не изменение на месте.
  • Нельзя хранить в JSONB: secrets, raw PII, provider credentials, tokens, unbounded external payloads.
  • Нельзя использовать object, JsonDocument, Dictionary<string, object> в domain/application contracts.

Elasticsearch Document​

// Talent.Common.Domain.Search

public abstract class ElasticsearchDocument
{
    [PropertyName("documentId")]
    public string DocumentId { get; set; } = default!;

    [PropertyName("tenantId")]
    public string TenantId { get; set; } = default!;

    [PropertyName("brandId")]
    public string? BrandId { get; set; }

    [PropertyName("schemaVersion")]
    public int SchemaVersion { get; set; }

    [PropertyName("updatedAt")]
    public DateTimeOffset UpdatedAt { get; set; }

    [PropertyName("indexedAt")]
    public DateTimeOffset IndexedAt { get; set; }
}

Каждый конкретный документ описывается в src/services/<service>/Infrastructure/Search/Documents/.

Канон полнотекстового поиска в Pin — Talent.Common.Domain.Search поверх Elasticsearch (ADR-0008, ADR-0059). Строительные блоки: ElasticsearchDocument (базовый typed-документ), IElasticsearchIndexRepository<TDocument> (индексация/чтение), SearchQuery (типизированный запрос с фильтрами/пагинацией/сортировкой), ISearchRebuildJob (полная переиндексация из source-of-truth PostgreSQL) и alias-index-менеджеры (zero-downtime reindex через переключение alias между версиями индекса). Собственный поисковый слой, прямой клиент Elasticsearch/ElasticSearch в бизнес-коде и сторонние поисковые движки не используются.

ClickHouse Fact​

// Talent.Common.Analytics

public abstract class ClickHouseFact
{
    // Deduplication key (уникален для ClickHouse ReplacingMergeTree)
    public string FactId { get; set; } = default!;

    // Source event id для трассировки
    public string SourceEventId { get; set; } = default!;

    public string TenantId { get; set; } = default!;
    public string? BrandId { get; set; }

    // Бизнес-время события (для group by date)
    public DateTime EventTime { get; set; }

    // Время записи в ClickHouse
    public DateTime IngestionTime { get; set; }
}

Правила:

  • Operational handlers не пишут в ClickHouse синхронно; analytics consumer делает batch ingestion.
  • Каждый fact имеет deduplication key (FactId) и SourceEventId.
  • FINAL запрещён в online dashboard queries без задокументированного исключения.
  • Конкретные facts описываются в docs/06-services/analytics/entities/facts/.

Kafka Event Envelope​

Базовые Типы Envelope​

// Talent.Common.Messaging

/// <summary>
/// Дескриптор Kafka-топика и метаданных события.
/// Декларируется один раз как static readonly на конкретном event record,
/// читается без рефлексии через тип.
/// </summary>
public sealed record KafkaEventDescriptor
{
    /// <summary>Kafka topic: "accounting.payment-events.v1"</summary>
    public required string TopicName { get; init; }

    /// <summary>Уникальный тип события: "accounting.deposit.completed"</summary>
    public required string EventType { get; init; }

    /// <summary>Версия payload schema. Увеличивается при breaking change.</summary>
    public required int EventVersion { get; init; }

    /// <summary>Тип агрегата-источника: "DepositRequest"</summary>
    public required string AggregateType { get; init; }

    /// <summary>Сервис-производитель: "accounting-service"</summary>
    public required string ProducerService { get; init; }

    /// <summary>
    /// Опциональный ключ партиционирования (UserId, TenantId и т.п.).
    /// Null — Kafka выбирает партицию по round-robin.
    /// </summary>
    public string? PartitionKeyField { get; init; }

    /// <summary>True — топик log-compacted (последнее значение на ключ).</summary>
    public bool IsCompacted { get; init; }
}

// C# 11+ static abstract interface members.
// Все метаданные доступны через тип без создания экземпляра и без рефлексии.

/// <summary>
/// Маркер payload integration event. Каждый конкретный event record реализует
/// единственное статическое свойство <see cref="KafkaEvent"/> — вся маршрутизация
/// и envelope-построение выполняются через него без рефлексии.
/// </summary>
public interface IIntegrationEvent
{
    /// <summary>Возвращает неизменяемый дескриптор топика и метаданных события.</summary>
    static abstract KafkaEventDescriptor KafkaEvent { get; }
}

/// <summary>
/// Generic helper — позволяет читать дескриптор через тип-параметр
/// в infrastructure-коде (producers, serializers, topic resolvers).
/// </summary>
public static class IntegrationEventMeta
{
    public static KafkaEventDescriptor Descriptor<TEvent>()
        where TEvent : IIntegrationEvent
        => TEvent.KafkaEvent;

    public static string TopicName<TEvent>()
        where TEvent : IIntegrationEvent
        => TEvent.KafkaEvent.TopicName;

    public static string EventType<TEvent>()
        where TEvent : IIntegrationEvent
        => TEvent.KafkaEvent.EventType;
}

// Envelope — стандартная обёртка события
public sealed class IntegrationEventEnvelope<TPayload>
    where TPayload : IIntegrationEvent
{
    public Guid EventId { get; init; } = Guid.CreateVersion7();
    public string EventType { get; init; } = default!;
    public int EventVersion { get; init; }
    public string SchemaVersion { get; init; } = "1.0.0";
    public DateTimeOffset OccurredAt { get; init; }
    public string Producer { get; init; } = default!;

    // Scope
    public string? TenantId { get; init; }
    public string? BrandId { get; init; }

    // Aggregate info
    public string AggregateType { get; init; } = default!;
    public string AggregateId { get; init; } = default!;
    public int? AggregateVersion { get; init; }

    // Tracing
    public string CorrelationId { get; init; } = default!;
    public string? CausationId { get; init; }
    public string? TraceId { get; init; }
    public string? IdempotencyKey { get; init; }

    // Actor snapshot (безопасный, без credentials/claims)
    public EventActorSnapshot Actor { get; init; } = default!;

    // Payload — конкретная типизированная модель
    public TPayload Payload { get; init; } = default!;

    // Техническая метаинформация
    public EventTechnicalMetadata? TechnicalMetadata { get; init; }
}

public sealed class EventActorSnapshot
{
    public string ActorId { get; init; } = default!;
    public string ActorType { get; init; } = default!;
    public string? TenantId { get; init; }
    public string? BrandId { get; init; }
    public string? ImpersonationKind { get; init; }
    public string? OnBehalfOfActorId { get; init; }
}

public sealed class EventTechnicalMetadata
{
    public string? SourceOperation { get; init; }
    public int DeliveryAttempt { get; init; }
    public string? ProducerInstanceId { get; init; }
    public string? PartitionKey { get; init; }
    public string? ReplayReason { get; init; }
}

Реализация Event Record​

// Pin.Accounting.Contracts/Events/Payments/DepositCompletedEvent.cs

// Без атрибутов. Дескриптор — единственное статическое свойство KafkaEvent.
public sealed record DepositCompletedEvent : IIntegrationEvent
{
    // --- Дескриптор события ---
    public static KafkaEventDescriptor KafkaEvent { get; } = new()
    {
        TopicName       = "accounting.payment-events.v1",
        EventType       = "accounting.deposit.completed",
        EventVersion    = 1,
        AggregateType   = "DepositRequest",
        ProducerService = "accounting-service",
        PartitionKeyField = "UserId",   // партиционируем по игроку
    };

    // --- Payload ---
    public required DepositSnapshot Deposit { get; init; }
    public required PaymentProviderSnapshot Provider { get; init; }

    public sealed record DepositSnapshot
    {
        public required string DepositId { get; init; }
        public required string UserId { get; init; }
        public required decimal Amount { get; init; }
        public required string CurrencyId { get; init; }
        public required string Status { get; init; }
        public required DateTimeOffset CompletedAt { get; init; }
    }

    public sealed record PaymentProviderSnapshot
    {
        public required string ProviderId { get; init; }
        public required string ProviderName { get; init; }
        public required string ProviderCode { get; init; }
    }
}

Чтение метаданных в infrastructure-коде — через дескриптор без создания экземпляра:

// Прямо через конкретный тип:
var descriptor = DepositCompletedEvent.KafkaEvent;
string topic   = descriptor.TopicName;   // "accounting.payment-events.v1"
string evtType = descriptor.EventType;   // "accounting.deposit.completed"

// Через generic helper в producer/serializer, где тип известен как TEvent:
var descriptor = IntegrationEventMeta.Descriptor<DepositCompletedEvent>();
string topic   = IntegrationEventMeta.TopicName<DepositCompletedEvent>();

// Построение envelope через PinEventPublisher (см. раздел MassTransit Wrappers):
// publisher автоматически читает KafkaEvent и заполняет envelope — вручную не нужно.
await _publisher.Publish(depositEvent, aggregateId: deposit.Id.ToString(), ct);

Domain Event vs Integration Event​

КритерийDomain EventIntegration Event
ScopeВнутри одной транзакции сервисаCross-service через Kafka
ХранениеВ памяти aggregate (_domainEvents)Outbox → Kafka
ФорматC# record, нет JSON-схемыIntegrationEventEnvelope<TPayload>, версионируется
ПолучателиDomain event handlers в том же сервисеMassTransit consumers в других сервисах
RollbackОткатывается вместе с транзакциейПубликуется только после commit
ИменованиеWithdrawalApprovedEvent (past tense)Payload: WithdrawalApprovedPayload; EventType: "accounting.withdrawal.approved"
Aggregate.Approve()
  → RaiseDomainEvent(WithdrawalApprovedEvent)     ← in-memory, синхронно
  → _repository.Update(aggregate)                  ← persist в той же транзакции
  → DomainEventDispatchBehavior
       → map WithdrawalApprovedEvent → WithdrawalApprovedPayload
       → _publisher.Publish(payload, aggregateId)  ← в MassTransit Outbox (та же транзакция)
  → Commit
  → MassTransit relay service → Kafka
  → Consumer в другом сервисе

Snapshot Payload Rules​

  • After snapshot обязателен для событий создания и изменения.
  • Before snapshot добавляется, когда consumers нужен diff без локального state.
  • Snapshot содержит полезные display fields: имена, статусы, коды.
  • Snapshot не содержит: secrets, raw tokens, provider credentials, raw PII, полные платёжные реквизиты.
  • Большие коллекции — bounded snapshot или summary.

Pagination Models​

// Talent.Common.Application

// Входящий запрос страницы
public sealed record PagedQuery
{
    public int Take { get; init; } = 20;
    public int Offset { get; init; } = 0;

    // Валидация: Take от 1 до 200
}

// Результат с метаинформацией
public sealed class PagedResult<TItem>
{
    public IReadOnlyList<TItem> Items { get; init; } = [];
    public PagedMeta Meta { get; init; } = default!;
}

public sealed class PagedMeta
{
    public int Total { get; init; }
    public int Take { get; init; }
    public int Offset { get; init; }
    public bool HasMore => Offset + Take < Total;
}

// Cursor-based paging для высоконагруженных feeds
public sealed class CursorPagedResult<TItem>
{
    public IReadOnlyList<TItem> Items { get; init; } = [];
    public string? NextCursor { get; init; }
    public bool HasMore { get; init; }
}

// Стандартные response contracts для API
public sealed class PagedResponse<TItem>
{
    public IReadOnlyList<TItem> Items { get; init; } = [];
    public PagedMeta Meta { get; init; } = default!;
    public FreshnessView? Freshness { get; init; }
}

public sealed class FreshnessView
{
    public DateTimeOffset? DataAsOf { get; init; }
    public string? Source { get; init; }   // "postgresql", "elasticsearch", "clickhouse", "redis"
    public int? LagSeconds { get; init; }
    public bool IsStale { get; init; }
}

public sealed class MoneyView
{
    public decimal Amount { get; init; }
    public short CurrencyId { get; init; }                 // ссылка на справочник handbook
    public string CurrencyCode { get; init; } = default!;  // display only (ISO-код из CurrencySnapshot)
    public int Precision { get; init; }
    public string Formatted { get; init; } = default!;
}

Sorting Models​

SortQuery<TField> типизирован по enum — допустимые поля сортировки определяются на уровне конкретного сервиса, а не строкой. PagedSortedQuery не используется: PagedQuery и SortQuery<TField> передаются как два отдельных параметра.

// Talent.Common.Application

// Generic sort — TField задаётся сервисом, не shared-кодом
public sealed record SortQuery<TField>
    where TField : struct, Enum
{
    public TField? SortBy { get; init; }
    public SortDirection Direction { get; init; } = SortDirection.Asc;
}

public enum SortDirection { Asc, Desc }

Enum сортировочных полей объявляется в Application слое каждого сервиса:

// Pin.Accounting.Application/Withdrawals/Models/WithdrawalSortField.cs
public enum WithdrawalSortField { CreatedAt, Amount, Status, ProcessedAt }

// Pin.Accounting.Application/Accounts/Models/AccountingSortField.cs
public enum AccountingSortField { CreatedAt, Balance, CurrencyId }

Сигнатура repository-метода с пагинацией и сортировкой — два отдельных параметра:

// Pin.Accounting.Application/Withdrawals/IWithdrawalRepository.cs

public interface IWithdrawalRepository
{
    // Paging и sort — отдельные параметры; sort опционален
    Task<PagedResult<WithdrawalView>> List(
        WithdrawalFilter filter,
        PagedQuery paging,
        SortQuery<WithdrawalSortField>? sort = null,
        CancellationToken ct = default);

    Task<WithdrawalRequest?> Find(WithdrawalRequestId id, CancellationToken ct = default);
    Task<WithdrawalRequest>  Get(WithdrawalRequestId id,  CancellationToken ct = default);
    Task Add(WithdrawalRequest withdrawal,    CancellationToken ct = default);
    Task Update(WithdrawalRequest withdrawal, CancellationToken ct = default);
}

// Пример вызова из handler:
var result = await _repository.List(
    filter: new WithdrawalFilter { Status = WithdrawalStatus.Requested },
    paging: new PagedQuery { Take = 20, Offset = 0 },
    sort:   new SortQuery<WithdrawalSortField> { SortBy = WithdrawalSortField.CreatedAt, Direction = SortDirection.Desc },
    ct:     ct);

LookupResponse (для dropdown/select)​

// Talent.Common.Application

public sealed class LookupResponse
{
    public IReadOnlyList<ResultEntry> Items { get; init; } = [];
}

// ResultEntry описан в docs/06-services/gateway-shared/models/result-entry.md
public sealed class ResultEntry
{
    public string Id { get; init; } = default!;
    public string Label { get; init; } = default!;
    public string? SecondaryLabel { get; init; }
    public bool Disabled { get; init; }
    public string? Category { get; init; }
}

Дополнительные Value Objects и Примитивы​

Ниже — value objects, которые используются повсеместно в Pin и декларируются в Talent.Common.Domain:

// --- Финансовые примитивы ---

/// <summary>Процентное значение 0..100 с двумя знаками. Используется для wager multiplier, commission rate.</summary>
public readonly record struct PercentageAmount(decimal Value)
{
    public static PercentageAmount Zero => new(0m);
    public static PercentageAmount Of(decimal value)
    {
        if (value < 0 || value > 100)
            throw new DomainException($"Percentage must be 0-100, got {value}");
        return new(Math.Round(value, 2));
    }
    public decimal AsFraction => Value / 100m;
}

/// <summary>Диапазон дат с включительными границами (обе обязательны).</summary>
public readonly record struct DateRange(DateTimeOffset From, DateTimeOffset To)
{
    public static DateRange Of(DateTimeOffset from, DateTimeOffset to)
    {
        if (to < from) throw new DomainException("DateRange: To must be >= From");
        return new(from, to);
    }
    public bool Contains(DateTimeOffset point) => point >= From && point <= To;
    public TimeSpan Duration => To - From;
    public bool Overlaps(DateRange other) => From < other.To && To > other.From;
}

/// <summary>
/// Диапазон значений с необязательными (nullable) включительными границами.
/// Заменяет плоские пары фильтров вида <c>PriceMin/PriceMax</c> и <c>*From/*To</c>
/// одним типизированным полем: <c>price: NullableRange&lt;decimal&gt;</c>.
/// Открытая граница (<c>null</c>) означает «без ограничения с этой стороны».
/// Для диапазона дат с обязательными границами используется <see cref="DateRange"/>.
/// </summary>
/// <typeparam name="T">Сравнимый value type: <c>decimal</c>, <c>int</c>, <c>DateTimeOffset</c> и т. п.</typeparam>
public readonly record struct NullableRange<T>(T? Min, T? Max)
    where T : struct, IComparable<T>
{
    /// <summary>Открытый с обеих сторон диапазон — фильтр не задан.</summary>
    public static NullableRange<T> Any => new(null, null);

    public static NullableRange<T> Of(T? min, T? max)
    {
        if (min.HasValue && max.HasValue && max.Value.CompareTo(min.Value) < 0)
            throw new DomainException("NullableRange: Max must be >= Min");
        return new(min, max);
    }

    /// <summary>Задана хотя бы одна граница.</summary>
    public bool HasBound => Min.HasValue || Max.HasValue;

    public bool Contains(T value)
        => (!Min.HasValue || value.CompareTo(Min.Value) >= 0)
        && (!Max.HasValue || value.CompareTo(Max.Value) <= 0);
}

/// <summary>
/// Типизированный диапазон фильтра с опциональными границами (обе включительные).
/// Один VO вместо плоских пар <c>*Min</c>/<c>*Max</c> в фильтрах каталога:
/// цена (<c>NullableRange&lt;decimal&gt;</c>), площадь, число комнат, этаж
/// (<c>NullableRange&lt;int&gt;</c>). Null-граница означает «без ограничения» с этой стороны.
/// </summary>
public readonly record struct NullableRange<T>(T? From, T? To)
    where T : struct, IComparable<T>
{
    /// <summary>Фабрика диапазона; проверяет корректность порядка непустых границ.</summary>
    public static NullableRange<T> Of(T? from, T? to)
    {
        if (from.HasValue && to.HasValue && to.Value.CompareTo(from.Value) < 0)
            throw new DomainException("NullableRange: To must be >= From");
        return new(from, to);
    }

    /// <summary>Обе границы отсутствуют — фильтр по этому полю не задан.</summary>
    public bool IsEmpty => !From.HasValue && !To.HasValue;

    /// <summary>Хотя бы одна граница открыта (диапазон не ограничен с одной стороны).</summary>
    public bool IsUnbounded => !From.HasValue || !To.HasValue;

    /// <summary>Значение входит в диапазон с учётом открытых границ.</summary>
    public bool Contains(T value)
        => (!From.HasValue || value.CompareTo(From.Value) >= 0)
        && (!To.HasValue || value.CompareTo(To.Value) <= 0);
}

/// <summary>Wager multiplier — множитель отыгрыша бонуса. Например, x35.</summary>
public readonly record struct WagerMultiplier(decimal Value)
{
    public static WagerMultiplier Of(decimal value)
    {
        if (value <= 0) throw new DomainException($"WagerMultiplier must be positive, got {value}");
        return new(Math.Round(value, 1));
    }
}

// --- Идентификаторы ---

/// <summary>Внешний игровой идентификатор у провайдера. Формат зависит от провайдера.</summary>
public readonly record struct ProviderGameId(string Value)
{
    public static ProviderGameId Of(string value)
        => string.IsNullOrWhiteSpace(value)
            ? throw new DomainException("ProviderGameId must not be empty")
            : new(value);
}

/// <summary>Идентификатор игровой сессии.</summary>
public readonly record struct GameSessionId(Guid Value)
{
    public static GameSessionId NewId() => new(Guid.CreateVersion7());
    public override string ToString() => Value.ToString();
}

/// <summary>Код бонуса, вводимый игроком. Нормализуется к UPPER_CASE.</summary>
public readonly record struct IncentiveCode(string Value)
{
    public static IncentiveCode Of(string raw)
    {
        var normalized = raw.Trim().ToUpperInvariant();
        if (normalized.Length < 3 || normalized.Length > 64)
            throw new DomainException($"IncentiveCode length must be 3-64 chars");
        return new(normalized);
    }
}

/// <summary>Промокод партнёра — аналогично IncentiveCode, но в пространстве партнёра.</summary>
public readonly record struct PromoCode(string Value)
{
    public static PromoCode Of(string raw)
        => new(raw.Trim().ToUpperInvariant());
}

// --- Геолокация и сеть ---

/// <summary>IP-адрес игрока (IPv4 или IPv6). Хранится как string для простоты поиска.</summary>
public readonly record struct IpAddress(string Value)
{
    public static IpAddress Of(string ip)
    {
        if (!System.Net.IPAddress.TryParse(ip, out _))
            throw new DomainException($"Invalid IP address: {ip}");
        return new(ip);
    }
    public bool IsLoopback => Value is "127.0.0.1" or "::1";
}

/// <summary>
/// Гео-локация из MaxMind или аналога — adapter-модель IP-geo-провайдера.
/// Не хранится как PII — только страна и регион.
/// Страна — ISO-2-код провайдера, НЕ ссылка <c>CountryId</c> справочника handbook.
/// </summary>
public sealed record GeoLocation
{
    public required string CountryIsoCode2 { get; init; }   // ISO 3166-1 alpha-2 (код провайдера)
    public string? RegionCode { get; init; }
    public string? City { get; init; }
    public bool IsVpnOrProxy { get; init; }
}

/// <summary>
/// Device fingerprint snapshot — для fraud detection и session binding.
/// Хранится зашифрованным; RAW значение не логируется.
/// </summary>
public sealed record DeviceFingerprint
{
    public required string FingerprintHash { get; init; }   // SHA-256 fingerprint
    public string? UserAgent { get; init; }
    public string? Platform { get; init; }   // "web", "ios", "android"
    public string? DeviceType { get; init; } // "mobile", "desktop", "tablet"
}

// --- User-specific ---

/// <summary>Сегмент игрока для персонализации и маркетинга.</summary>
public enum UserSegment
{
    Unknown,
    NewUser,
    Casual,
    Regular,
    PrioritySegment,
    AtRisk,
    Churned,
}

/// <summary>Тег игрока (label) для операционной классификации (не PII).</summary>
public readonly record struct UserTag(string Value)
{
    public static UserTag Of(string value)
    {
        var normalized = value.Trim().ToLowerInvariant();
        if (normalized.Length > 50) throw new DomainException("UserTag max 50 chars");
        return new(normalized);
    }
}

MassTransit Wrappers — Producer и Consumer​

Прямое использование IPublishEndpoint и IConsumer<T> запрещено в бизнес-коде — обёртки берут на себя построение envelope, восстановление context и структурированное логирование.

IPinEventPublisher — Producer​

// Talent.Common.Messaging/Producers/IPinEventPublisher.cs

/// <summary>
/// Публикует integration event, автоматически оборачивая payload в
/// <see cref="IntegrationEventEnvelope{TPayload}"/> с корреляцией, actor snapshot
/// и метаданными из <see cref="IIntegrationEvent.KafkaEvent"/>.
/// </summary>
public interface IPinEventPublisher
{
    Task Publish<TEvent>(
        TEvent payload,
        string aggregateId,
        int? aggregateVersion = null,
        CancellationToken ct = default)
        where TEvent : class, IIntegrationEvent;
}

// Talent.Common.Messaging/Producers/PinEventPublisher.cs
public sealed class PinEventPublisher : IPinEventPublisher
{
    private readonly IPublishEndpoint _publishEndpoint;
    private readonly IIdentityContextProvider _identityProvider;
    private readonly ICorrelationContext _correlationContext;
    private readonly IClock _clock;
    private readonly ILogger<PinEventPublisher> _logger;

    public async Task Publish<TEvent>(
        TEvent payload,
        string aggregateId,
        int? aggregateVersion = null,
        CancellationToken ct = default)
        where TEvent : class, IIntegrationEvent
    {
        var descriptor  = TEvent.KafkaEvent;
        var actor       = _identityProvider.TryGetCurrent();
        var correlation = _correlationContext.CorrelationId;

        var envelope = new IntegrationEventEnvelope<TEvent>
        {
            EventId          = Guid.CreateVersion7(),
            EventType        = descriptor.EventType,
            EventVersion     = descriptor.EventVersion,
            AggregateType    = descriptor.AggregateType,
            AggregateId      = aggregateId,
            AggregateVersion = aggregateVersion,
            Producer         = descriptor.ProducerService,
            OccurredAt       = _clock.UtcNow,
            CorrelationId    = correlation,
            TenantId         = actor?.TenantId?.ToString(),
            BrandId          = actor?.BrandId?.ToString(),
            Actor            = actor?.ToEventSnapshot(),
            Payload          = payload,
        };

        _logger.LogDebug(
            "Publishing {EventType} v{Version} aggregateId={AggregateId} correlationId={CorrelationId}",
            descriptor.EventType, descriptor.EventVersion, aggregateId, correlation);

        await _publishEndpoint.Publish(envelope, ctx =>
        {
            // Проставляем Kafka partition key из дескриптора
            if (descriptor.PartitionKeyField is not null)
                ctx.Headers.Set("partition-key", aggregateId);
        }, ct);
    }
}

PinConsumerBase — Consumer​

// Talent.Common.Messaging/Consumers/PinConsumerBase.cs

/// <summary>
/// Базовый consumer. Автоматически:
/// - Восстанавливает CorrelationId, TenantId, BrandId из envelope headers.
/// - Логирует начало/конец/ошибку потребления с полным контекстом.
/// - Проверяет версию payload: при неизвестной версии — dead-letter.
/// - Предоставляет методы CheckIdempotency / MarkProcessed для business-level идемпотентности.
/// </summary>
public abstract class PinConsumerBase<TEvent>
    : IConsumer<IntegrationEventEnvelope<TEvent>>
    where TEvent : class, IIntegrationEvent
{
    private readonly IMessageContextRestorer _contextRestorer;
    private readonly ILogger _logger;
    private readonly IBusinessIdempotencyStore _idempotencyStore;

    protected PinConsumerBase(
        IMessageContextRestorer contextRestorer,
        IBusinessIdempotencyStore idempotencyStore,
        ILogger logger)
    {
        _contextRestorer  = contextRestorer;
        _idempotencyStore = idempotencyStore;
        _logger           = logger;
    }

    public async Task Consume(ConsumeContext<IntegrationEventEnvelope<TEvent>> context)
    {
        var envelope    = context.Message;
        var descriptor  = TEvent.KafkaEvent;

        // 1. Восстанавливаем correlation/tenant context из envelope
        using var scope = _contextRestorer.Restore(envelope);

        using var logScope = _logger.BeginScope(new Dictionary<string, object>
        {
            ["EventType"]     = envelope.EventType,
            ["EventVersion"]  = envelope.EventVersion,
            ["EventId"]       = envelope.EventId,
            ["CorrelationId"] = envelope.CorrelationId,
            ["AggregateId"]   = envelope.AggregateId,
        });

        // 2. Версия payload — dead-letter при неизвестной версии
        if (envelope.EventVersion > descriptor.KafkaEvent.EventVersion)
        {
            _logger.LogWarning(
                "Unknown EventVersion {Received} for {EventType}, expected max {Expected}. Dead-lettering.",
                envelope.EventVersion, envelope.EventType, descriptor.KafkaEvent.EventVersion);
            throw new UnknownEventVersionException(envelope.EventType, envelope.EventVersion);
        }

        // 3. Business-level idempotency check
        var idempotencyKey = $"{descriptor.EventType}:{envelope.EventId}";
        if (await _idempotencyStore.IsProcessed(idempotencyKey, context.CancellationToken))
        {
            _logger.LogInformation(
                "Skipping duplicate event {EventType} eventId={EventId}", 
                envelope.EventType, envelope.EventId);
            return;
        }

        _logger.LogInformation(
            "Consuming {EventType} v{Version} eventId={EventId}",
            envelope.EventType, envelope.EventVersion, envelope.EventId);

        try
        {
            await Handle(envelope.Payload, envelope, context.CancellationToken);
            await _idempotencyStore.MarkProcessed(idempotencyKey, context.CancellationToken);

            _logger.LogInformation(
                "Consumed {EventType} eventId={EventId}", 
                envelope.EventType, envelope.EventId);
        }
        catch (Exception ex)
        {
            _logger.LogError(ex,
                "Error consuming {EventType} eventId={EventId}",
                envelope.EventType, envelope.EventId);
            throw; // MassTransit retry / dead-letter policy
        }
    }

    /// <summary>
    /// Бизнес-логика обработки. Вызывается после восстановления context и проверки идемпотентности.
    /// </summary>
    protected abstract Task Handle(
        TEvent payload,
        IntegrationEventEnvelope<TEvent> envelope,
        CancellationToken ct);
}

Пример конкретного Consumer​

// Pin.GameHost.Application/Consumers/AccountingDepositCompletedConsumer.cs

public sealed class AccountingDepositCompletedConsumer
    : PinConsumerBase<DepositCompletedEvent>
{
    private readonly IUserBalanceProjectionRepository _projectionRepo;

    public AccountingDepositCompletedConsumer(
        IMessageContextRestorer contextRestorer,
        IBusinessIdempotencyStore idempotencyStore,
        ILogger<AccountingDepositCompletedConsumer> logger,
        IUserBalanceProjectionRepository projectionRepo)
        : base(contextRestorer, idempotencyStore, logger)
    {
        _projectionRepo = projectionRepo;
    }

    protected override async Task Handle(
        DepositCompletedEvent payload,
        IntegrationEventEnvelope<DepositCompletedEvent> envelope,
        CancellationToken ct)
    {
        // --- Уже в правильном tenant/correlation context ---
        await _projectionRepo.UpdateBalance(
            userId:  new UserId(Guid.Parse(payload.Deposit.UserId)),
            amount:    payload.Deposit.Amount,
            currency:  new CurrencyId(payload.Deposit.CurrencyId),
            ct:        ct);
    }
}

// Регистрация:
services.AddMassTransit(x =>
{
    x.AddConsumer<AccountingDepositCompletedConsumer>();
    x.UsingKafka((ctx, k) =>
    {
        k.TopicEndpoint<IntegrationEventEnvelope<DepositCompletedEvent>>(
            topicName:     DepositCompletedEvent.KafkaEvent.TopicName,
            groupId:       "game-host-accounting-deposit",
            configure:     e => e.ConfigureConsumer<AccountingDepositCompletedConsumer>(ctx));
    });
});

Outbox и Inbox Entity Models​

// Talent.Common.Messaging

// Outbox record — создаётся в той же транзакции что и aggregate mutation
public sealed class OutboxMessage
{
    public Guid   Id { get; init; } = Guid.CreateVersion7();
    public string EventType { get; init; } = default!;
    public string TopicName { get; init; } = default!;
    public int    EventVersion { get; init; }
    public string AggregateId { get; init; } = default!;
    public string AggregateType { get; init; } = default!;
    public string CorrelationId { get; init; } = default!;
    public string? TenantId { get; init; }
    public string? BrandId { get; init; }
    public string PayloadJson { get; init; } = default!;       // сериализованный envelope
    public DateTimeOffset CreatedAt { get; init; }
    public DateTimeOffset? ProcessedAt { get; set; }
    public int RetryCount { get; set; }
    public string? Error { get; set; }
    public OutboxMessageStatus Status { get; set; } = OutboxMessageStatus.Pending;
}

public enum OutboxMessageStatus { Pending, Processed, Failed, DeadLettered }

// Inbox record — предотвращает двойную обработку входящих событий
public sealed class InboxMessage
{
    public Guid   Id { get; init; }           // = envelope.EventId
    public string EventType { get; init; } = default!;
    public string ConsumerGroup { get; init; } = default!;
    public DateTimeOffset ReceivedAt { get; init; }
    public DateTimeOffset? ProcessedAt { get; set; }
    public InboxMessageStatus Status { get; set; } = InboxMessageStatus.Received;
    public string? Error { get; set; }
}

public enum InboxMessageStatus { Received, Processed, Failed }

MassTransit Transactional Outbox управляется через AddEntityFrameworkOutbox<TDbContext>() — ручная работа с OutboxMessage не нужна. Модель описана для понимания контракта и для custom audit/reporting.

Ссылки​

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

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