Repository Pipeline и Scope Filters — Практическое Руководство

Назначение

Документ объясняет, как реализовать и использовать repository pipeline с автоматическими scope filters в Pin. Это практическое дополнение к Repository Abstractions And Scope Filters с примерами кода и правилами регистрации.

---

Проблема, Которую Решает Pipeline

Без pipeline каждый handler должен вручную передавать TenantId в каждый вызов репозитория:

// ПЛОХО: ручная передача scope в каждый запрос
var accountings = await _repository.GetByUserIdAndTenant(
    userId: _userScopeProvider.GetRequired(),
    tenantId: _tenantScopeProvider.GetRequired(), // ← повтор везде
    brandId: _brandScopeProvider.TryGet(),        // ← повтор везде
    ct);

С pipeline filters TenantId/BrandId/UserId подставляются автоматически:

// ХОРОШО: scope filters применяются автоматически через interceptors
var accountings = await _repository.ListByUser(paging: query.PagedQuery, ct: ct);
// Repository автоматически добавит WHERE tenant_id = ? AND brand_id = ? AND user_id = ?

---

Базовые Repository Interfaces

// Talent.Common.Infrastructure

// Command repository — aggregate mutation
public interface IRepository<TAggregate, TId>
    where TAggregate : IAggregateRoot<TId>
{
    Task<TAggregate?> Find(TId id, CancellationToken ct = default);
    Task<TAggregate>  Get(TId id,  CancellationToken ct = default); // throws if not found
    Task Add(TAggregate aggregate,    CancellationToken ct = default);
    Task Update(TAggregate aggregate, CancellationToken ct = default);
    Task Remove(TAggregate aggregate, CancellationToken ct = default);
}

// Read model / projection repository
public interface IReadRepository<TReadModel, TId>
{
    Task<TReadModel?> Find(TId id, CancellationToken ct = default);
    Task<PagedResult<TReadModel>> List(
        ISpecification<TReadModel> spec,
        PagedQuery paging,
        CancellationToken ct = default);
}

Важно: Repositories не возвращают IQueryable наружу. Вся фильтрация и сортировка — внутри репозитория через specifications и interceptors.

---

Scope Interceptors

Цепочка Interceptors

Каждый repository использует цепочку interceptors:

Query Request
  → TenantScopeInterceptor    ← WHERE tenant_id = {currentTenantId}
  → BrandScopeInterceptor     ← AND brand_id = {currentBrandId} (если entity IBrandScoped)
  → UserScopeInterceptor    ← AND user_id = {currentUserId} (если entity IUserOwned и actor = User)
  → PartnerScopeInterceptor   ← AND partner_id = {currentPartnerId} (если entity IPartnerOwned и actor = Partner)
  → SoftDeleteScopeInterceptor ← AND deleted_at IS NULL (если entity ISoftDeletable)
  → SpecificationInterceptor  ← AND {specification predicate}
  → SortingInterceptor
  → PagingInterceptor
  → Execute
  → AccessAuditWriter (если required)

Реализация TenantScopeInterceptor

// Talent.Common.Infrastructure

public sealed class TenantScopeInterceptor<TEntity> : IRepositoryScopeInterceptor<TEntity>
    where TEntity : class, ITenantScoped
{
    private readonly ITenantScopeProvider _tenantProvider;
    private readonly IScopeBypassPolicy _bypassPolicy;
    private readonly IRepositoryAuditWriter _auditWriter;

    public IQueryable<TEntity> Apply(IQueryable<TEntity> query, RepositoryScopeOptions options)
    {
        switch (options.ScopeMode)
        {
            case ScopeMode.Required:
                var tenantId = _tenantProvider.GetRequired(); // throws MissingScope if none
                return query.Where(e => e.TenantId == tenantId);

            case ScopeMode.Optional:
                var optTenantId = _tenantProvider.TryGet();
                return optTenantId.HasValue
                    ? query.Where(e => e.TenantId == optTenantId.Value)
                    : query; // system/global context — нет фильтра

            case ScopeMode.System:
                return query; // migrator/seed/system jobs

            case ScopeMode.BypassWithPermission:
                if (!_bypassPolicy.IsAllowed(BypassReason.CrossTenantAdmin))
                    throw new ScopeBoundaryException("Cross-tenant bypass denied");
                _auditWriter.WriteBypassAudit(options.BypassReason, "TenantScope");
                if (options.RequiredTenantId.HasValue)
                    return query.Where(e => e.TenantId == options.RequiredTenantId.Value);
                return query; // global admin view

            default:
                throw new UnreachableException();
        }
    }
}

Реализация UserScopeInterceptor

public sealed class UserScopeInterceptor<TEntity> : IRepositoryScopeInterceptor<TEntity>
    where TEntity : class, IUserOwned
{
    private readonly IIdentityContextProvider _identityProvider;
    private readonly IUserScopeProvider _userProvider;

    public IQueryable<TEntity> Apply(IQueryable<TEntity> query, RepositoryScopeOptions options)
    {
        var identity = _identityProvider.TryGetCurrent();

        // User actor видит только свои данные — всегда
        if (identity?.ActorType == ActorType.User)
        {
            var userId = _userProvider.GetRequired();
            return query.Where(e => e.UserId == userId);
        }

        // Admin с явным ExpectedOwnerActorId — смотрит конкретного игрока
        if (options.ExpectedOwnerActorId.HasValue && identity?.ActorType == ActorType.Admin)
        {
            _auditWriter.WriteUserDataAccessAudit(options.ExpectedOwnerActorId.Value);
            return query.Where(e => e.UserId == new UserId(options.ExpectedOwnerActorId.Value));
        }

        // System/ServiceAccount без ownership — нет фильтра (внутренние operations)
        return query;
    }
}

---

RepositoryScopeOptions

RepositoryScopeOptions передаётся в repository метод явно, когда default scope недостаточен:

// Talent.Common.Infrastructure

public sealed class RepositoryScopeOptions
{
    public static readonly RepositoryScopeOptions Default = new();

    public ScopeMode ScopeMode { get; init; } = ScopeMode.Required;
    public TenantId? RequiredTenantId { get; init; }
    public BrandId? RequiredBrandId { get; init; }
    public Guid? ExpectedOwnerActorId { get; init; }
    public string? BypassReason { get; init; }
    public bool AuditAccess { get; init; }
    public bool IncludeDeleted { get; init; }
}

public enum ScopeMode
{
    Required,            // tenant/brand обязательны, MissingScope если нет
    Optional,            // tenant/brand берётся если есть
    System,              // нет автоматических scope filters (migrator, seed, system job)
    BypassWithPermission // cross-tenant/cross-brand с permission и audit
}

Большинство handler-ов не передают RepositoryScopeOptions вообще — используется Default с ScopeMode.Required.

---

Mutation Pipeline

Для create/update mutation interceptor автоматически проставляет scope fields:

public sealed class MutationScopeInterceptor<TEntity> : IRepositoryMutationInterceptor<TEntity>
{
    private readonly ITenantScopeProvider _tenantProvider;
    private readonly IBrandScopeProvider _brandProvider;

    public void BeforeAdd(TEntity entity)
    {
        if (entity is ITenantScoped tenantScoped && tenantScoped.TenantId == default)
        {
            // Автоматически проставляем TenantId при создании
            var tenantId = _tenantProvider.GetRequired();
            // Используем reflection или explicit interface setter
            SetTenantId(entity, tenantId);
        }

        if (entity is IBrandScoped brandScoped && brandScoped.BrandId == default)
        {
            var brandId = _brandProvider.TryGet();
            if (brandId.HasValue) SetBrandId(entity, brandId.Value);
        }
    }

    public void BeforeUpdate(TEntity entity)
    {
        // Проверяем, что обновляемый entity принадлежит текущему tenant
        if (entity is ITenantScoped tenantScoped)
        {
            var currentTenantId = _tenantProvider.GetRequired();
            if (tenantScoped.TenantId != currentTenantId)
                throw new ScopeBoundaryException("Attempted cross-tenant mutation");
        }
    }
}

---

Создание Сервисного Repository

Конкретный repository в сервисе создаётся следующим образом:

// Pin.Accounting.Infrastructure/Repositories/AccountingRepository.cs

public sealed class AccountingRepository : RepositoryBase<AccountingAccount, AccountingAccountId>,
    IAccountingRepository
{
    public AccountingRepository(
        AccountingDbContext dbContext,
        IRepositoryPipeline<AccountingAccount> pipeline) // ← pipeline инжектируется
        : base(dbContext, pipeline)
    { }

    // Кастомный метод — pipeline применяется автоматически через base
    public Task<PagedResult<AccountingAccount>> ListActiveByUser(
        PagedQuery paging,
        SortQuery<AccountingSortField>? sort = null,
        CancellationToken ct = default)
    {
        return List(
            spec: new ActiveAccountingSpecification(),
            paging: paging,
            sort: sort,
            ct: ct);
        // Scope: TenantId + BrandId + UserId применятся автоматически
    }

    // Admin метод с явным bypass для просмотра кошелька конкретного игрока
    public Task<AccountingAccount?> FindByIdForAdmin(
        AccountingAccountId id,
        UserId targetUserId,
        CancellationToken ct = default)
    {
        return Find(
            id: id,
            options: new RepositoryScopeOptions
            {
                ScopeMode = ScopeMode.BypassWithPermission,
                ExpectedOwnerActorId = targetUserId.Value,
                BypassReason = "admin-user-view",
                AuditAccess = true
            },
            ct: ct);
    }
}

---

Elasticsearch И ClickHouse Scope Rules

Scope filters применяются одинаково для всех хранилищ:

// Talent.Common.Domain.Search
public sealed class SearchScopeFilter
{
    public QueryContainer Apply(QueryContainer query, RepositoryScopeOptions options)
    {
        var tenantId = _tenantProvider.GetRequired();

        return Query<ElasticsearchDocument>.Bool(b => b
            .Must(m => m
                .Term(t => t.Field(f => f.TenantId).Value(tenantId.ToString())),
                query));
    }
}

ClickHouse report queries добавляют WHERE tenant_id = {tenantId} до построения агрегаций.

---

Регистрация В DI

// Pin.Accounting.Infrastructure/DependencyInjection/AccountingInfrastructureModule.cs

public static class AccountingInfrastructureModule
{
    public static IPinInfrastructureBuilder AddAccountingInfrastructure(
        this IPinInfrastructureBuilder builder)
    {
        builder
            .AddPostgreSqlDbContext<AccountingDbContext>()
            .AddRepositoryPipeline()          // ← регистрирует все interceptors
            .AddRepository<IAccountingRepository, AccountingRepository>()
            .AddRepository<IWithdrawalRepository, WithdrawalRepository>()
            .AddRedisCache()
            .AddPaymentProviderAdapters();

        return builder;
    }
}

AddRepositoryPipeline() регистрирует:

• TenantScopeInterceptor<T> для всех ITenantScoped entities.
• BrandScopeInterceptor<T> для IBrandScoped.
• UserScopeInterceptor<T> для IUserOwned.
• PartnerScopeInterceptor<T> для IPartnerOwned.
• SoftDeleteScopeInterceptor<T> для ISoftDeletable.
• MutationScopeInterceptor<T> для всех mutable entities.
• IScopeBypassPolicy — default реализация через permission checker.
• IRepositoryAuditWriter — запись bypass audit.

---

Оптимистичный Concurrency

EF Core + PostgreSQL xmin — стандартный механизм для предотвращения lost updates:

// Talent.Common.Infrastructure/EF/Configurations/BaseEntityConfiguration.cs

public static class ConcurrencyTokenExtensions
{
    /// <summary>
    /// Настраивает xmin как concurrency token.
    /// PostgreSQL автоматически инкрементирует xmin при каждом UPDATE.
    /// </summary>
    public static void UseXminConcurrencyToken<TEntity>(
        this EntityTypeBuilder<TEntity> builder)
        where TEntity : class, IHasConcurrencyToken
    {
        builder.Property(e => e.Version)
            .HasColumnName("xmin")
            .HasColumnType("xid")
            .ValueGeneratedOnAddOrUpdate()
            .IsConcurrencyToken();
    }
}

// При конфликте EF бросает DbUpdateConcurrencyException.
// Repository pipeline перехватывает и возвращает typed error:
public sealed class ConcurrencyInterceptor<TEntity> : IRepositoryMutationInterceptor<TEntity>
{
    public async Task<Result> ExecuteUpdate(
        TEntity entity,
        Func<Task> next)
    {
        try
        {
            await next();
            return Result.Ok();
        }
        catch (DbUpdateConcurrencyException ex)
        {
            _logger.LogWarning(
                "Concurrency conflict on {Entity} id={Id}",
                typeof(TEntity).Name, GetId(entity));
            return Result.Fail(InfrastructureErrors.ConcurrencyConflict);
        }
    }
}

Правила:

• Aggregate root с IHasConcurrencyToken всегда использует xmin.
• Handler при ConcurrencyConflict → retry 1 раз (через IdempotencyBehavior или explicit retry в handler).
• Distributed lock нужен только когда retry недопустим (см. 11-distributed-locking.md).

---

Правило Fail Closed

Если scope обязателен, но provider не вернул значение — repository бросает MissingScopeException до запроса в БД:

// Сценарий: user endpoint без аутентификации
// TenantId есть (из domain routing), но UserId нет (нет JWT)
// → UserScopeInterceptor бросает MissingScopeException
// → TransactionBehavior откатывает (если есть)
// → MediatR pipeline возвращает Failure("missing-scope")
// → ResultMappingFilter маппит в 401/403
// БД запрос не выполняется

---

Тестирование Scope Filters

Integration Test Шаблон

[Fact]
public async Task UserCannotSeeAnotherUserAccountings()
{
    // Arrange: два игрока, у каждого свои кошельки
    var user1 = await CreateTestUser(TenantId);
    var user2 = await CreateTestUser(TenantId);
    await CreateAccounting(user1.UserId);
    await CreateAccounting(user2.UserId);

    // Act: запрос от имени user1
    var response = await _factory
        .WithActorContext(ctx => ctx.AsUser(user1.UserId, TenantId, BrandId))
        .GetAsync("/api/accounting/accounts");

    // Assert: видит только свои кошельки
    response.EnsureSuccessStatusCode();
    var accountings = await response.ReadAs<PagedResponse<AccountingView>>();
    accountings.Items.Should().AllSatisfy(w => w.UserId.Should().Be(user1.UserId.ToString()));
    accountings.Items.Should().NotContain(w => w.UserId == user2.UserId.ToString());
}

[Fact]
public async Task MissingTenantScopeFailsClosed()
{
    // Act: запрос без tenant context (system context без tenant)
    var response = await _factory
        .WithActorContext(ctx => ctx.AsSystemAccount())
        .GetAsync("/api/accounting/accounts");

    // Assert: 403 или typed failure — нет данных без tenant scope
    response.StatusCode.Should().BeOneOf(HttpStatusCode.Forbidden, HttpStatusCode.Unauthorized);
}

Обязательные Scope Tests Для Каждого Сервиса

• User не видит чужие данные (accounting, incentive, saved-items, history).
• Partner не видит данные другого partner.
• Tenant admin не видит данные другого tenant.
• Global admin видит cross-tenant только с permission и bypass audit.
• Отсутствующий tenant scope → fail closed (нет запроса в БД).
• Repository filter работает для PostgreSQL, Elasticsearch и ClickHouse read paths.

---

Ссылки

• Repository Abstractions And Scope Filters
• ADR-0014: Repository Pipeline And Scope Filters
• 04 — Actors, Providers, Middleware
• 08 — Тестирование

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

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

• Фоновые задачи (2): Background Job: BookingExpirationJob, Background Job: LeadExpirationJob
• adr (1): ADR-0059: Поиск во всех поверхностях PIN — Elasticsearch-проекции через Talent.Common.Domain.Search
• index.md (1): Backend SDK — Руководство Разработчика
• 04-actors-providers-middleware.md (1): Actors, Providers, Middleware — Практическое Руководство
• 08-testing-patterns.md (1): Тестирование — Практическое Руководство
• 11-distributed-locking.md (1): Distributed Locking — Архитектура и Практическое Руководство
• 14-security-patterns.md (1): Security Patterns — Authorization, PII, OWASP
• backend-shared-standards-index.md (1): Backend Shared Standards Index