Marten для C# (.NET): Полное руководство по Document DB и Event Store на PostgreSQL

Marten — это мощная библиотека для .NET, которая превращает PostgreSQL в гибридное хранилище данных, объединяющее возможности Document Database (документной базы данных) и Event Store (хранилища событий). Разработанная командой JasperFx, она позволяет разработчикам работать с PostgreSQL как с NoSQL-подобной системой, сохраняя при этом все преимущества реляционной базы данных.

Зачем нужна Marten? В современной разработке часто требуется гибкость документных БД (например, MongoDB) для хранения сложных, слабоструктурированных данных, и одновременно — надежность и транзакционность реляционных баз. Marten решает эту проблему, предоставляя LINQ-провайдер для работы с JSON-документами, хранящимися в PostgreSQL, и полноценный Event Store для реализации Event Sourcing и CQRS паттернов. Это особенно полезно в микросервисной архитектуре, системах аудита, финансовых приложениях и любых проектах, где требуется сочетание гибкости схемы данных и строгой консистентности.

Библиотека активно поддерживается и используется в production-проектах. Она интегрируется с ASP.NET Core, Entity Framework Core (при необходимости), и предоставляет богатый API для работы с данными. Marten автоматически создает и обновляет схему базы данных, поддерживает миграции, индексы и даже полнотекстовый поиск. Это делает её отличным выбором для .NET-разработчиков, которые хотят использовать мощь PostgreSQL без написания сложного SQL-кода.

Установка

Установка Marten осуществляется через NuGet Package Manager. Выполните одну из следующих команд:

dotnet add package Marten

или через Package Manager Console:

Install-Package Marten

Для работы с Event Store дополнительных пакетов не требуется — всё включено в основной пакет. Рекомендуется также установить Marten.AspNetCore для интеграции с ASP.NET Core:

dotnet add package Marten.AspNetCore

Быстрый старт

Минимальный рабочий пример включает настройку подключения к PostgreSQL, определение документа и выполнение CRUD-операций.

using Marten;
using System;
using System.Threading.Tasks;

// 1. Определяем документ
public class User
{
    public Guid Id { get; set; }
    public string Name { get; set; }
    public int Age { get; set; }
    public string Email { get; set; }
}

class Program
{
    static async Task Main(string[] args)
    {
        // 2. Настраиваем DocumentStore
        var store = DocumentStore.For(options =>
        {
            options.Connection("Host=localhost;Database=marten_db;Username=postgres;Password=password");
            options.AutoCreateSchemaObjects = AutoCreate.All; // Автосоздание таблиц
        });

        // 3. Открываем сессию для работы
        using var session = store.LightweightSession();

        // 4. Создаем документ
        var user = new User
        {
            Id = Guid.NewGuid(),
            Name = "Иван Иванов",
            Age = 30,
            Email = "ivan@example.com"
        };

        session.Store(user);
        await session.SaveChangesAsync();

        // 5. Загружаем документ по ID
        var loadedUser = await session.LoadAsync<User>(user.Id);
        Console.WriteLine($"Загружен пользователь: {loadedUser.Name}, возраст: {loadedUser.Age}");

        // 6. Обновляем документ
        loadedUser.Age = 31;
        session.Store(loadedUser);
        await session.SaveChangesAsync();

        // 7. Удаляем документ
        session.Delete(loadedUser);
        await session.SaveChangesAsync();

        Console.WriteLine("Готово!");
    }
}

Этот пример демонстрирует базовые операции: создание, чтение, обновление и удаление документа. Marten автоматически создаст таблицу mt_doc_user в PostgreSQL и будет сохранять объекты в виде JSONB.

Основные методы и классы

Ниже представлены ключевые классы и методы Marten с подробным описанием и примерами.

1. DocumentStore.For

Сигнатура: public static DocumentStore For(Action<StoreOptions> configure)

Что делает: Создает и настраивает экземпляр DocumentStore — основной класс для работы с Marten. Принимает делегат для конфигурации подключения, схемы, политик и т.д.

Пример:

var store = DocumentStore.For(options =>
{
    options.Connection("Host=localhost;Database=test_db;Username=postgres;Password=pass");
    options.AutoCreateSchemaObjects = AutoCreate.All;
    options.Policies.AllDocumentsAreSoftDeleted(); // Мягкое удаление
});

2. IDocumentSession.Store

Сигнатура: void Store<T>(T document) where T : class

Что делает: Сохраняет (вставляет или обновляет) документ в сессии. Изменения фиксируются только после вызова SaveChangesAsync().

Пример:

using var session = store.LightweightSession();
var order = new Order { Id = Guid.NewGuid(), Product = "Laptop", Quantity = 1 };
session.Store(order);
await session.SaveChangesAsync();

3. IDocumentSession.LoadAsync

Сигнатура: Task<T> LoadAsync<T>(object id, CancellationToken token = default) where T : class

Что делает: Загружает документ по его идентификатору. Возвращает null, если документ не найден.

Пример:

var user = await session.LoadAsync<User>(userId);
if (user != null)
{
    Console.WriteLine($"User: {user.Name}");
}

4. IDocumentSession.Query

Сигнатура: IMartenQueryable<T> Query<T>() where T : class

Что делает: Возвращает IQueryable для выполнения LINQ-запросов к документам. Поддерживает фильтрацию, сортировку, проекцию и пагинацию.

Пример:

var adults = await session.Query<User>()
    .Where(u => u.Age >= 18)
    .OrderBy(u => u.Name)
    .Take(10)
    .ToListAsync();

5. IDocumentSession.Delete

Сигнатура: void Delete<T>(T document) where T : class

Что делает: Помечает документ на удаление. При мягком удалении (soft delete) документ не удаляется физически, а помечается как удаленный.

Пример:

session.Delete(user);
await session.SaveChangesAsync();

6. IDocumentSession.Events.StartStream

Сигнатура: EventStream StartStream<TAggregate>(Guid streamId, params object[] events)

Что делает: Создает новый поток событий (Event Stream) для агрегата. Принимает идентификатор потока и начальные события.

Пример:

var streamId = Guid.NewGuid();
session.Events.StartStream<OrderAggregate>(streamId, 
    new OrderCreated { OrderId = streamId, Product = "Phone" },
    new OrderShipped { OrderId = streamId, TrackingNumber = "TRACK123" });
await session.SaveChangesAsync();

7. IDocumentSession.Events.Append

Сигнатура: void Append(Guid streamId, params object[] events)

Что делает: Добавляет новые события в существующий поток. Поток должен быть уже создан.

Пример:

session.Events.Append(streamId, new OrderDelivered { OrderId = streamId, DeliveryDate = DateTime.UtcNow });
await session.SaveChangesAsync();

8. IDocumentSession.Events.FetchStream

Сигнатура: Task<IReadOnlyList<IEvent>> FetchStream(Guid streamId, int version = 0, DateTime? timestamp = null)

Что делает: Загружает все события из потока. Можно указать версию или временную метку для получения снимка на определенный момент.

Пример:

var events = await session.Events.FetchStream(streamId);
foreach (var @event in events)
{
    Console.WriteLine($"Event: {@event.Data.GetType().Name} at {@event.Timestamp}");
}

9. IDocumentSession.Events.AggregateStream

Сигнатура: Task<TAggregate> AggregateStream<TAggregate>(Guid streamId, int fromVersion = 0, int toVersion = int.MaxValue)

Что делает: Восстанавливает состояние агрегата из потока событий, применяя каждое событие к агрегату. Это ключевой метод для Event Sourcing.

Пример:

var order = await session.Events.AggregateStream<OrderAggregate>(streamId);
Console.WriteLine($"Order status: {order.Status}");

10. IDocumentStore.BulkInsert

Сигнатура: Task BulkInsert<T>(IReadOnlyList<T> documents, BulkInsertMode mode = BulkInsertMode.InsertsOnly, int batchSize = 500)

Что делает: Выполняет массовую вставку документов. Значительно быстрее, чем вставка по одному через сессию. Поддерживает режимы: только вставка, обновление или вставка/обновление.

Пример:

var users = Enumerable.Range(1, 1000).Select(i => new User 
{ 
    Id = Guid.NewGuid(), 
    Name = $"User {i}", 
    Age = Random.Shared.Next(18, 60) 
}).ToList();

await store.BulkInsertAsync(users);

Реальный практический пример: Система управления заказами с Event Sourcing

Рассмотрим полноценный пример реализации системы управления заказами с использованием Event Sourcing и CQRS. Мы создадим агрегат заказа, определим события, реализуем команды и запросы.

using Marten;
using Marten.Events;
using Marten.Events.Aggregation;
using Marten.Events.Projections;
using System;
using System.Collections.Generic;
using System.Linq;
using System.Threading.Tasks;

// ========== События ==========
public record OrderCreated(Guid OrderId, string CustomerName, List<OrderItem> Items, DateTime CreatedAt);
public record OrderItemAdded(Guid OrderId, OrderItem Item);
public record OrderItemRemoved(Guid OrderId, Guid ItemId);
public record OrderShipped(Guid OrderId, DateTime ShippedAt);
public record OrderDelivered(Guid OrderId, DateTime DeliveredAt);
public record OrderCancelled(Guid OrderId, string Reason);

public record OrderItem(Guid ItemId, string ProductName, decimal Price, int Quantity);

// ========== Агрегат ==========
public class OrderAggregate
{
    public Guid Id { get; set; }
    public string CustomerName { get; set; }
    public List<OrderItem> Items { get; set; } = new();
    public decimal TotalAmount => Items.Sum(i => i.Price * i.Quantity);
    public OrderStatus Status { get; set; }
    public DateTime CreatedAt { get; set; }
    public DateTime? ShippedAt { get; set; }
    public DateTime? DeliveredAt { get; set; }
    public string CancellationReason { get; set; }
    public int Version { get; set; }

    // Методы применения событий
    public void Apply(OrderCreated @event)
    {
        Id = @event.OrderId;
        CustomerName = @event.CustomerName;
        Items = @event.Items;
        Status = OrderStatus.Created;
        CreatedAt = @event.CreatedAt;
    }

    public void Apply(OrderItemAdded @event)
    {
        Items.Add(@event.Item);
    }

    public void Apply(OrderItemRemoved @event)
    {
        Items.RemoveAll(i => i.ItemId == @event.ItemId);
    }

    public void Apply(OrderShipped @event)
    {
        Status = OrderStatus.Shipped;
        ShippedAt = @event.ShippedAt;
    }

    public void Apply(OrderDelivered @event)
    {
        Status = OrderStatus.Delivered;
        DeliveredAt = @event.DeliveredAt;
    }

    public void Apply(OrderCancelled @event)
    {
        Status = OrderStatus.Cancelled;
        CancellationReason = @event.Reason;
    }
}

public enum OrderStatus
{
    Created,
    Shipped,
    Delivered,
    Cancelled
}

// ========== Проекция для чтения ==========
public class OrderSummary
{
    public Guid Id { get; set; }
    public string CustomerName { get; set; }
    public int ItemCount { get; set; }
    public decimal TotalAmount { get; set; }
    public OrderStatus Status { get; set; }
    public DateTime CreatedAt { get; set; }
}

public class OrderSummaryProjection : EventProjection
{
    public OrderSummaryProjection()
    {
        // Подписываемся на события
        ProjectionName = "OrderSummary";
    }

    public OrderSummary Create(OrderCreated @event)
    {
        return new OrderSummary
        {
            Id = @event.OrderId,
            CustomerName = @event.CustomerName,
            ItemCount = @event.Items.Count,
            TotalAmount = @event.Items.Sum(i => i.Price * i.Quantity),
            Status = OrderStatus.Created,
            CreatedAt = @event.CreatedAt
        };
    }

    public OrderSummary Apply(OrderItemAdded @event, OrderSummary current)
    {
        current.ItemCount++;
        // TotalAmount будет пересчитан при необходимости
        return current;
    }

    public OrderSummary Apply(OrderItemRemoved @event, OrderSummary current)
    {
        current.ItemCount--;
        return current;
    }

    public OrderSummary Apply(OrderShipped @event, OrderSummary current)
    {
        current.Status = OrderStatus.Shipped;
        return current;
    }

    public OrderSummary Apply(OrderDelivered @event, OrderSummary current)
    {
        current.Status = OrderStatus.Delivered;
        return current;
    }

    public OrderSummary Apply(OrderCancelled @event, OrderSummary current)
    {
        current.Status = OrderStatus.Cancelled;
        return current;
    }
}

// ========== Сервис приложения ==========
public class OrderService
{
    private readonly IDocumentStore _store;

    public OrderService