Indiko.Blocks.DataAccess.Marten 2.5.1

Indiko.Blocks.DataAccess.Marten

Marten implementation of the Indiko data access abstractions, providing PostgreSQL-based document database and event sourcing support.

Overview

This package provides a Marten-based implementation of the Indiko data access layer, combining the power of PostgreSQL with document storage and event sourcing capabilities.

Features

• Marten Integration: Full Marten library integration
• Document Storage: Store objects as JSON documents in PostgreSQL
• Event Sourcing: Built-in event sourcing and event store
• ACID Transactions: PostgreSQL transaction support
• Advanced Querying: LINQ queries compiled to PostgreSQL
• Projections: Real-time and async projections
• Multi-Tenancy: Built-in multi-tenant support
• Optimistic Concurrency: Version-based concurrency control
• Patching: Partial document updates
• Full-Text Search: PostgreSQL full-text search integration

Installation

dotnet add package Indiko.Blocks.DataAccess.Marten

Quick Start

Define Your DbContext

using Indiko.Blocks.DataAccess.Marten;
using Marten;

public class AppDbContext : BaseDbContext
{
    public AppDbContext(IDocumentStore documentStore) : base(documentStore)
    {
    }

    protected override void OnModelCreating(MartenSchemaBuilder schemaBuilder)
    {
        base.OnModelCreating(schemaBuilder);
        
        // Configure document mappings
        schemaBuilder.For<User>()
            .Index(x => x.Email, config => config.IsUnique = true)
            .Index(x => x.LastName);
            
        schemaBuilder.For<Order>()
            .UseOptimisticConcurrency(true)
            .Index(x => x.OrderNumber, config => config.IsUnique = true);
    }
}

Configure Services

public class Startup : WebStartup
{
    public override void ConfigureServices(IServiceCollection services)
    {
        base.ConfigureServices(services);
        
        services.UseMartenDataAccess<AppDbContext>(options =>
        {
            options.ConnectionString = Configuration.GetConnectionString("PostgreSQL");
            options.DatabaseSchemaName = "public";
            options.AutoCreateSchemaObjects = AutoCreate.CreateOrUpdate;
        });
    }
}

Entity Definition

public class User : BaseEntity<Guid>
{
    public string FirstName { get; set; }
    public string LastName { get; set; }
    public string Email { get; set; }
    public Address Address { get; set; }
    public List<string> Roles { get; set; }
}

public class Address
{
    public string Street { get; set; }
    public string City { get; set; }
    public string Country { get; set; }
    public string PostalCode { get; set; }
}

Usage Examples

Basic CRUD

public class UserService
{
    private readonly IRepository<User, Guid> _userRepository;
    private readonly IUnitOfWork _unitOfWork;

    public UserService(IRepository<User, Guid> userRepository, IUnitOfWork unitOfWork)
    {
        _userRepository = userRepository;
        _unitOfWork = unitOfWork;
    }

    public async Task<User> CreateAsync(User user)
    {
        user.Id = Guid.NewGuid();
        await _userRepository.AddAsync(user);
        await _unitOfWork.SaveChangesAsync();
        return user;
    }

    public async Task<User> GetByEmailAsync(string email)
    {
        return await _userRepository.ReadByQueryAsync(u => u.Email == email);
    }
}

Advanced Queries

// Deep property queries
var usersInNewYork = await _userRepository.ReadManyByQueryAsync(u => 
    u.Address.City == "New York");

// Array operations
var admins = await _userRepository.ReadManyByQueryAsync(u => 
    u.Roles.Contains("Admin"));

// Complex LINQ queries
var query = _userRepository.AsQueryable()
    .Where(u => u.LastName.StartsWith("S"))
    .Where(u => u.Address.Country == "USA")
    .OrderBy(u => u.LastName)
    .ThenBy(u => u.FirstName)
    .Select(u => new 
    {
        u.Id,
        FullName = u.FirstName + " " + u.LastName,
        u.Email
    });

var results = await query.ToListAsync();

Patching (Partial Updates)

public async Task UpdateEmailAsync(Guid userId, string newEmail)
{
    var session = _context.Session;
    
    // Patch specific properties without loading the entire document
    session.Patch<User>(userId)
        .Set(u => u.Email, newEmail)
        .Set(u => u.UpdatedAt, DateTime.UtcNow);
    
    await session.SaveChangesAsync();
}

public async Task AddRoleAsync(Guid userId, string role)
{
    var session = _context.Session;
    
    // Append to array
    session.Patch<User>(userId)
        .Append(u => u.Roles, role);
    
    await session.SaveChangesAsync();
}

Optimistic Concurrency

public async Task UpdateWithConcurrencyAsync(User user)
{
    try
    {
        await _userRepository.UpdateAsync(user);
        await _unitOfWork.SaveChangesAsync();
    }
    catch (ConcurrencyException ex)
    {
        // Handle version conflict
        var currentVersion = await _userRepository.ReadByIdAsync(user.Id);
        throw new InvalidOperationException("Document was modified by another transaction", ex);
    }
}

Event Sourcing

Define Events

public class OrderCreated
{
    public Guid OrderId { get; set; }
    public Guid UserId { get; set; }
    public decimal TotalAmount { get; set; }
    public DateTime CreatedAt { get; set; }
}

public class OrderShipped
{
    public Guid OrderId { get; set; }
    public string TrackingNumber { get; set; }
    public DateTime ShippedAt { get; set; }
}

public class OrderCompleted
{
    public Guid OrderId { get; set; }
    public DateTime CompletedAt { get; set; }
}

Event Stream

public class OrderService
{
    private readonly IDocumentSession _session;

    public async Task CreateOrderAsync(Guid orderId, Guid userId, decimal amount)
    {
        var @event = new OrderCreated
        {
            OrderId = orderId,
            UserId = userId,
            TotalAmount = amount,
            CreatedAt = DateTime.UtcNow
        };
        
        // Append event to stream
        _session.Events.Append(orderId, @event);
        await _session.SaveChangesAsync();
    }

    public async Task ShipOrderAsync(Guid orderId, string trackingNumber)
    {
        var @event = new OrderShipped
        {
            OrderId = orderId,
            TrackingNumber = trackingNumber,
            ShippedAt = DateTime.UtcNow
        };
        
        _session.Events.Append(orderId, @event);
        await _session.SaveChangesAsync();
    }

    public async Task<IEnumerable<object>> GetOrderEventsAsync(Guid orderId)
    {
        // Fetch all events for an aggregate
        var events = await _session.Events.FetchStreamAsync(orderId);
        return events.Select(e => e.Data);
    }
}

Projections

// Define projection/read model
public class OrderSummary
{
    public Guid Id { get; set; }
    public Guid UserId { get; set; }
    public decimal TotalAmount { get; set; }
    public string Status { get; set; }
    public DateTime CreatedAt { get; set; }
    public DateTime? ShippedAt { get; set; }
    public string TrackingNumber { get; set; }
}

// Projection configuration
public class OrderSummaryProjection : SingleStreamProjection<OrderSummary>
{
    public OrderSummary Create(OrderCreated @event)
    {
        return new OrderSummary
        {
            Id = @event.OrderId,
            UserId = @event.UserId,
            TotalAmount = @event.TotalAmount,
            Status = "Created",
            CreatedAt = @event.CreatedAt
        };
    }

    public void Apply(OrderShipped @event, OrderSummary current)
    {
        current.Status = "Shipped";
        current.ShippedAt = @event.ShippedAt;
        current.TrackingNumber = @event.TrackingNumber;
    }

    public void Apply(OrderCompleted @event, OrderSummary current)
    {
        current.Status = "Completed";
    }
}

// Register projection
schemaBuilder.Events.InlineProjections.Add<OrderSummaryProjection>();

Configuration

appsettings.json

{
  "ConnectionStrings": {
    "PostgreSQL": "Host=localhost;Database=MyAppDb;Username=postgres;Password=password"
  },
  "Marten": {
    "DatabaseSchemaName": "public",
    "AutoCreateSchemaObjects": "CreateOrUpdate",
    "Events": {
      "DatabaseSchemaName": "events",
      "StreamIdentity": "AsString"
    }
  }
}

Advanced Configuration

services.UseMartenDataAccess<AppDbContext>(options =>
{
    options.ConnectionString = Configuration.GetConnectionString("PostgreSQL");
    
    // Schema configuration
    options.AutoCreateSchemaObjects = AutoCreate.CreateOrUpdate;
    options.DatabaseSchemaName = "public";
    
    // Performance tuning
    options.UseDefaultSerialization(enumStorage: EnumStorage.AsString);
    
    // Multi-tenancy
    options.Policies.AllDocumentsAreMultiTenanted();
    
    // Event store configuration
    options.Events.DatabaseSchemaName = "events";
    options.Events.StreamIdentity = StreamIdentity.AsGuid;
});

Marten-Specific Features

Full-Text Search

// Configure full-text index
schemaBuilder.For<Article>()
    .FullTextIndex(x => x.Title)
    .FullTextIndex(x => x.Content);

// Search using full-text
var results = await _articleRepository.AsQueryable()
    .Where(a => a.Search("postgresql database"))
    .ToListAsync();

Compiled Queries

public class UsersByEmail : ICompiledQuery<User, User>
{
    public string Email { get; set; }

    public Expression<Func<IQueryable<User>, User>> QueryIs()
    {
        return q => q.FirstOrDefault(x => x.Email == Email);
    }
}

// Use compiled query
var query = new UsersByEmail { Email = "user@example.com" };
var user = await _session.QueryAsync(query);

Multi-Tenancy

// Configure multi-tenancy
options.Policies.AllDocumentsAreMultiTenanted();

// Use with tenant ID
_session.TenantId = "tenant-123";
var users = await _userRepository.ReadAllAsync();

Soft Delete with Marten

// Configure soft delete
schemaBuilder.For<User>().SoftDeleted();

// Soft delete
await _userRepository.DeleteAsync(userId, useSoftDelete: true);

// Include deleted in queries
var allUsers = _session.Query<User>()
    .Where(x => x.IsDeleted())
    .ToList();

Performance Tips

1. Use Patching: Update specific fields without loading entire documents
2. Compiled Queries: Pre-compile frequently used queries
3. Batch Operations: Use bulk inserts/updates
4. Projections: Use async projections for complex aggregations
5. Indexing: Add indexes on frequently queried fields

Target Framework

• .NET 10

Dependencies

• Indiko.Blocks.DataAccess.Abstractions
• Marten (7.0+)
• Npgsql (8.0+)

License

See LICENSE file in the repository root.

Related Packages

• Indiko.Blocks.DataAccess.Abstractions - Core data access abstractions
• Indiko.Blocks.DataAccess.EntityFramework - EF Core implementation
• Indiko.Blocks.DataAccess.MongoDb - MongoDB implementation