### Install Monq.Core.ClickHouseBuffer

Source: https://github.com/monqdl/monq.core.clickhousebuffer/blob/master/README.md

Install the library via NuGet package manager.

```powershell
Install-Package Monq.Core.ClickHouseBuffer
```

--------------------------------

### Define Buffer Configuration

Source: https://github.com/monqdl/monq.core.clickhousebuffer/blob/master/README.md

Example JSON configuration for the buffer options.

```json
{
    "bufferOptions": {
      "ConnectionString": "Host=clickhouse<-http-host>;Port=80;Username=<user>;Password=<password>;Database=<database>;",
      "EventsFlushPeriodSec": 2,
      "EventsFlushCount": 10000,
      "DatabaseBatchSize": 100000,
      "DatabaseMaxDegreeOfParallelism": 1
    }
}
```

--------------------------------

### Default ClickHouse Events Writer Implementation

Source: https://github.com/monqdl/monq.core.clickhousebuffer/blob/master/README.md

An example implementation of the `IEventsWriter` interface for writing event batches to ClickHouse using `ClickHouseBulkCopy`. This class handles connection, options, column mapping, and bulk insertion.

```csharp
internal sealed class DefaultClickHouseEventsWriter : IEventsWriter
{
    readonly IClickHouseConnection _connection;
    readonly EngineOptions _options;

    public DefaultClickHouseEventsWriter(IClickHouseConnection connection,
        EngineOptions engineOptions)
    {
        _connection = connection;

        if (engineOptions == null)
            throw new ArgumentNullException(nameof(engineOptions), $"{nameof(engineOptions)} is null.");

        _options = engineOptions;
    }

    /// <inheritdoc />
    public async Task WriteBatch(IEnumerable<EventItem> events, TypeTuple key)
    {
        if (!events.Any())
            return;

        // Get events column names.
        var columns = ClickHouseSchemaConfig.GlobalSettings.GetMappedColumns(key);

        var values = events.Select(x => x.Values);

        using var bulkCopy = new ClickHouseBulkCopy((ClickHouseConnection)_connection)
        {
            MaxDegreeOfParallelism = _options.DatabaseMaxDegreeOfParallelism,
            BatchSize = _options.EventsFlushCount,
            DestinationTableName = _connection.Database + "." + key.TableName,
            ColumnNames = columns
        };

        // Prepares ClickHouseBulkCopy instance by loading target column types
        await bulkCopy.InitAsync().ConfigureAwait(false);

        await bulkCopy.WriteToServerAsync(values).ConfigureAwait(false);
    }
}
```

--------------------------------

### Custom IEventsWriter Implementation for ClickHouse

Source: https://context7.com/monqdl/monq.core.clickhousebuffer/llms.txt

Implement IEventsWriter to customize event writing to ClickHouse, enabling logging, metrics, or routing to different databases. This example shows batch writing using ClickHouseBulkCopy.

```csharp
using ClickHouse.Driver;
using ClickHouse.Driver.ADO;
using ClickHouse.Driver.Copy;
using Monq.Core.ClickHouseBuffer;
using Monq.Core.ClickHouseBuffer.Schemas;
using Microsoft.Extensions.Logging;

public class InstrumentedEventsWriter : IEventsWriter
{
    private readonly IClickHouseConnection _connection;
    private readonly EngineOptions _options;
    private readonly ILogger<InstrumentedEventsWriter> _logger;

    public InstrumentedEventsWriter(
        IClickHouseConnection connection,
        EngineOptions options,
        ILogger<InstrumentedEventsWriter> logger)
    {
        _connection = connection;
        _options = options;
        _logger = logger;
    }

    public async Task WriteBatch(IEnumerable<EventItem> events, TypeTuple key)
    {
        var eventList = events.ToList();
        if (eventList.Count == 0)
            return;

        var stopwatch = System.Diagnostics.Stopwatch.StartNew();

        try
        {
            var columns = ClickHouseSchemaConfig.GlobalSettings.GetMappedColumns(key);
            var values = eventList.Select(x => x.Values);

            using var bulkCopy = new ClickHouseBulkCopy((ClickHouseConnection)_connection)
            {
                MaxDegreeOfParallelism = _options.DatabaseMaxDegreeOfParallelism,
                BatchSize = _options.EventsFlushCount,
                DestinationTableName = $"{_connection.Database}.{key.TableName}",
                ColumnNames = columns
            };

            await bulkCopy.InitAsync();
            await bulkCopy.WriteToServerAsync(values);

            stopwatch.Stop();
            _logger.LogInformation(
                "Wrote {Count} events to {Table} in {ElapsedMs}ms",
                eventList.Count, key.TableName, stopwatch.ElapsedMilliseconds);
        }
        catch (Exception ex)
        {
            _logger.LogError(ex, "Failed to write {Count} events to {Table}", eventList.Count, key.TableName);
            throw;
        }
    }
}

// Register in DI after ConfigureCHBuffer
// builder.Services.AddTransient<IEventsWriter, InstrumentedEventsWriter>();
```

--------------------------------

### Implement Custom IEventsWriter for Multiple ClickHouses

Source: https://github.com/monqdl/monq.core.clickhousebuffer/blob/master/README.md

Implement the IEventsWriter interface to handle writing events to different ClickHouse databases based on table name. Ensure proper dependency injection setup.

```csharp
internal sealed class MultipleClickHouseEventsWriter : IEventsWriter
{
    readonly IServiceProvider _serviceProvider;
    readonly EngineOptions _options;

    public MultiClickHouseEventsWriter(IServiceProvider serviceProvider,
        EngineOptions engineOptions)
    {
        _serviceProvider = serviceProvider;

        if (engineOptions == null)
            throw new ArgumentNullException(nameof(engineOptions), $"{nameof(engineOptions)} is null.");

        _options = engineOptions;
    }

    /// <inheritdoc />
    public async Task WriteBatch(IEnumerable<EventItem> events, TypeTuple key)
    {
        if (!events.Any())
            return;

        // Get events column names.
        var columns = ClickHouseSchemaConfig.GlobalSettings.GetMappedColumns(key);

        var values = events.Select(x => x.Values);

        var connection = GetConnection(key.TableName);

        using var bulkCopy = new ClickHouseBulkCopy(connection)
        {
            MaxDegreeOfParallelism = _options.DatabaseMaxDegreeOfParallelism,
            BatchSize = _options.EventsFlushCount,
            DestinationTableName = _connection.Database + "." + key.TableName,
            ColumnNames = columns
        };

        // Prepares ClickHouseBulkCopy instance by loading target column types
        await bulkCopy.InitAsync().ConfigureAwait(false);

        await bulkCopy.WriteToServerAsync(values).ConfigureAwait(false);
    }
}

ClickHouseConnection GetConnection(string tableName) =>
        _serviceProvider.GetRequiredKeyedService<ClickHouseConnection>(
            tableName == "logs"
                ? "ch1"
                : "ch2");
```

--------------------------------

### Configure Dependency Injection

Source: https://context7.com/monqdl/monq.core.clickhousebuffer/llms.txt

Register the ClickHouse buffer services using various configuration methods including connection strings, configuration sections, or action delegates.

```csharp
using Monq.Core.ClickHouseBuffer.DependencyInjection;

var builder = WebApplication.CreateBuilder(args);

// Option 1: Configure with connection string only
builder.Services.ConfigureCHBuffer("Host=clickhouse-host;Port=8123;Username=default;Password=secret;Database=mydb;");

// Option 2: Configure from appsettings.json section
// appsettings.json:
// {
//     "bufferOptions": {
//         "ConnectionString": "Host=clickhouse-host;Port=8123;Username=default;Password=secret;Database=mydb;",
//         "EventsFlushPeriodSec": 2,
//         "EventsFlushCount": 10000,
//         "DatabaseBatchSize": 100000,
//         "DatabaseMaxDegreeOfParallelism": 1
//     }
// }
builder.Services.ConfigureCHBuffer(builder.Configuration.GetSection("bufferOptions"));

// Option 3: Configure with action delegate
builder.Services.ConfigureCHBuffer(options =>
{
    options.ConnectionString = "Host=clickhouse-host;Port=8123;Username=default;Password=secret;Database=mydb;";
    options.EventsFlushPeriodSec = 5;      // Flush every 5 seconds
    options.EventsFlushCount = 5000;        // Or when 5000 events accumulated
    options.DatabaseBatchSize = 50000;      // Batch size for bulk copy
    options.DatabaseMaxDegreeOfParallelism = 2;  // Parallel insert tasks
});

// Option 4: Combine IConfiguration with action overrides
builder.Services.ConfigureCHBuffer(
    builder.Configuration.GetSection("bufferOptions"),
    options => options.ConnectionString = Environment.GetEnvironmentVariable("CLICKHOUSE_CONN")!
);

var app = builder.Build();
```

--------------------------------

### Await Flush Completion with IEventsBufferEngine

Source: https://context7.com/monqdl/monq.core.clickhousebuffer/llms.txt

Demonstrates using CompleteAsync within a hosted service for graceful shutdown and a batch processor to ensure data persistence.

```csharp
using Monq.Core.ClickHouseBuffer;
using Microsoft.Extensions.Hosting;

public class GracefulShutdownService : IHostedService
{
    private readonly IEventsBufferEngine _engine;

    public GracefulShutdownService(IEventsBufferEngine engine)
    {
        _engine = engine;
    }

    public Task StartAsync(CancellationToken cancellationToken)
    {
        return Task.CompletedTask;
    }

    public async Task StopAsync(CancellationToken cancellationToken)
    {
        // Wait for any pending writes to complete before shutdown
        await _engine.CompleteAsync();
        Console.WriteLine("All buffered events have been written to ClickHouse");
    }
}

public class BatchProcessor
{
    private readonly IEventsBufferEngine _engine;

    public BatchProcessor(IEventsBufferEngine engine)
    {
        _engine = engine;
    }

    public async Task ProcessAndWaitAsync(IEnumerable<LogEvent> events)
    {
        foreach (var evt in events)
        {
            _engine.AddEvent(evt, "logs");
        }

        // Ensure all events in this batch are written before returning
        await _engine.CompleteAsync();
    }
}

public class LogEvent
{
    public long StreamId { get; set; }
    public string Message { get; set; } = string.Empty;
    public DateTime Timestamp { get; set; }
}
```

--------------------------------

### Configure Schema with ITableSchema

Source: https://context7.com/monqdl/monq.core.clickhousebuffer/llms.txt

Implement ITableSchema to define mappings in dedicated classes and use Scan() to register them automatically at startup.

```csharp
using Monq.Core.ClickHouseBuffer.Schemas;

// Define your event model
public class UserActivityEvent
{
    public Guid UserId { get; set; }
    public string Action { get; set; } = string.Empty;
    public DateTime Timestamp { get; set; }
    public string Metadata { get; set; } = string.Empty;
}

// Implement ITableSchema in a dedicated configuration class
public sealed class UserActivitySchemaConfig : ITableSchema
{
    public void Register(ClickHouseSchemaConfig config)
    {
        config.NewConfig<UserActivityEvent>("user_activity")
            .Map("user_id", x => x.UserId)
            .Map("action", x => x.Action)
            .Map("event_time", x => x.Timestamp)
            .Map("metadata", x => x.Metadata);
    }
}

// Another schema configuration
public sealed class OrderEventSchemaConfig : ITableSchema
{
    public void Register(ClickHouseSchemaConfig config)
    {
        config.NewConfig<OrderEvent>("orders")
            .Map("order_id", x => x.OrderId)
            .Map("customer_id", x => x.CustomerId)
            .Map("total_amount", x => x.TotalAmount)
            .Map("created_at", x => x.CreatedAt);
    }
}

public class OrderEvent
{
    public long OrderId { get; set; }
    public long CustomerId { get; set; }
    public decimal TotalAmount { get; set; }
    public DateTime CreatedAt { get; set; }
}

// At startup, scan assemblies to auto-register all ITableSchema implementations
public class Program
{
    public static void Main(string[] args)
    {
        // Scans the assembly and registers all classes implementing ITableSchema
        ClickHouseSchemaConfig.GlobalSettings.Scan(typeof(Program).Assembly);

        var builder = WebApplication.CreateBuilder(args);
        builder.Services.ConfigureCHBuffer("Host=clickhouse;Database=analytics;");

        var app = builder.Build();
        app.Run();
    }
}
```

--------------------------------

### Configure EngineOptions

Source: https://context7.com/monqdl/monq.core.clickhousebuffer/llms.txt

Define buffer engine settings such as connection parameters, flush triggers, and batch processing limits.

```csharp
using Monq.Core.ClickHouseBuffer;

var options = new EngineOptions
{
    // ClickHouse connection string
    ConnectionString = "Host=clickhouse-host;Port=8123;Username=default;Password=secret;Database=analytics;",

    // Time-based flush: write buffered events after this many seconds (default: 2)
    EventsFlushPeriodSec = 2,

    // Count-based flush: write buffered events when count reaches this threshold (default: 10000)
    EventsFlushCount = 10000,

    // Batch size used by ClickHouse bulk inserter (default: 100000)
    DatabaseBatchSize = 100000,

    // Number of parallel tasks for database writes (default: 1)
    DatabaseMaxDegreeOfParallelism = 1
};

// Use in configuration
builder.Services.ConfigureCHBuffer(opts =>
{
    opts.ConnectionString = options.ConnectionString;
    opts.EventsFlushPeriodSec = options.EventsFlushPeriodSec;
    opts.EventsFlushCount = options.EventsFlushCount;
    opts.DatabaseBatchSize = options.DatabaseBatchSize;
    opts.DatabaseMaxDegreeOfParallelism = options.DatabaseMaxDegreeOfParallelism;
});
```

--------------------------------

### Configure ClickHouse Buffer via Dependency Injection

Source: https://github.com/monqdl/monq.core.clickhousebuffer/blob/master/README.md

Register the buffer service in the service collection using a connection string or configuration section.

```csharp
builder.Services.ConfigureCHBuffer(clickHouseConnectionString);
```

```csharp
builder.Services.ConfigureCHBuffer(Configuration.GetSection("bufferOptions"));
```

--------------------------------

### Configure Predefined Schema Mapping

Source: https://context7.com/monqdl/monq.core.clickhousebuffer/llms.txt

Configure schema mappings at application startup for optimal performance. Use NewConfig<T> to define the event model and Map to associate column names with object property expressions. This approach compiles expressions for faster data processing.

```csharp
using Monq.Core.ClickHouseBuffer.Schemas;

// Define your event model
public class LogEvent
{
    public long StreamId { get; set; }
    public string StreamName { get; set; } = string.Empty;
    public DateTime AggregatedAt { get; set; }
    public int UserspaceId { get; set; }
    public string Value { get; set; } = string.Empty;
}

// Configure mapping at startup (Program.cs or Startup.cs)
public class Program
{
    public static void Main(string[] args)
    {
        // Define schema mapping - column names to property expressions
        ClickHouseSchemaConfig.GlobalSettings.NewConfig<LogEvent>("logs")
            .Map("_streamId", x => x.StreamId)
            .Map("_streamName", x => x.StreamName)
            .Map("_aggregatedAt", x => x.AggregatedAt)
            .Map("_userspaceId", x => x.UserspaceId)
            .Map("_rawJson", x => x.Value);

        // Define another table mapping
        ClickHouseSchemaConfig.GlobalSettings.NewConfig<MetricEvent>("metrics")
            .Map("metric_name", x => x.Name)
            .Map("metric_value", x => x.Value)
            .Map("recorded_at", x => x.RecordedAt)
            .Map("tags", x => x.Tags);

        var builder = WebApplication.CreateBuilder(args);
        builder.Services.ConfigureCHBuffer("Host=clickhouse;Database=analytics;");

        var app = builder.Build();
        app.Run();
    }
}

public class MetricEvent
{
    public string Name { get; set; } = string.Empty;
    public double Value { get; set; }
    public DateTime RecordedAt { get; set; }
    public string Tags { get; set; } = string.Empty;
}
```

--------------------------------

### Register Schema via ITableSchema

Source: https://github.com/monqdl/monq.core.clickhousebuffer/blob/master/README.md

Scan for schema configurations and implement the ITableSchema interface for modular mapping.

```csharp
ClickHouseSchemaConfig.GlobalSettings.Scan(typeof(ClickHouseBench).Assembly);
```

```csharp
using Monq.Core.ClickHouseBuffer.Schemas;

public sealed class MyFullSchemaConfig : ITableSchema
{
    public void Register(ClickHouseSchemaConfig config)
    {
        config.NewConfig<MyEvent>("logs")
            .Map("_streamId", x => x.StreamId)
            .Map("_streamName", x => x.StreamName)
            .Map("_aggregatedAt", x => x.AggregatedAt)
            .Map("_userspaceId", x => x.UserspaceId)
            .Map("_rawJson", x => x.Value);
    }
}
```

--------------------------------

### Add Events to Buffer

Source: https://github.com/monqdl/monq.core.clickhousebuffer/blob/master/README.md

Inject IEventsBufferEngine and use the AddEvent method to queue items for insertion.

```csharp
public class MyService
{
    readonly IEventsBufferEngine _engine;

    public MyService(IEventsBufferEngine engine)
    {
        _engine = engine;
    }

    void Processor() 
    {
        ...
        _engine.AddEvent(item, "table");
    }
}
```

--------------------------------

### ClickHouseSchemaConfig.GlobalSettings.NewConfig - Predefined Schema Mapping

Source: https://context7.com/monqdl/monq.core.clickhousebuffer/llms.txt

Configures predefined schema mappings between object properties and ClickHouse columns at application startup for high performance.

```APIDOC
## ClickHouseSchemaConfig.GlobalSettings.NewConfig - Predefined Schema Mapping

### Description
Defines mappings between object properties and ClickHouse columns at application startup. This approach uses compiled expressions for high performance.

### Method
POST (conceptual, as it's a configuration setup)

### Endpoint
N/A (This is a C# configuration setup, not a direct HTTP endpoint)

### Parameters
#### Path Parameters
None

#### Query Parameters
None

#### Request Body
None (Configuration is done via method chaining)

### Request Example
```csharp
// Define schema mapping for LogEvent
ClickHouseSchemaConfig.GlobalSettings.NewConfig<LogEvent>("logs")
    .Map("_streamId", x => x.StreamId)
    .Map("_streamName", x => x.StreamName)
    .Map("_aggregatedAt", x => x.AggregatedAt)
    .Map("_userspaceId", x => x.UserspaceId)
    .Map("_rawJson", x => x.Value);

// Define schema mapping for MetricEvent
ClickHouseSchemaConfig.GlobalSettings.NewConfig<MetricEvent>("metrics")
    .Map("metric_name", x => x.Name)
    .Map("metric_value", x => x.Value)
    .Map("recorded_at", x => x.RecordedAt)
    .Map("tags", x => x.Tags);
```

### Response
#### Success Response (200)
None (This is a configuration setup, not an HTTP response)

#### Response Example
None
```

--------------------------------

### Implement Multi-Database IEventsWriter in C#

Source: https://context7.com/monqdl/monq.core.clickhousebuffer/llms.txt

Routes events to different ClickHouse instances based on table names. Requires registering keyed ClickHouseConnection services in the dependency injection container.

```csharp
using ClickHouse.Driver;
using ClickHouse.Driver.ADO;
using ClickHouse.Driver.Copy;
using Monq.Core.ClickHouseBuffer;
using Monq.Core.ClickHouseBuffer.Schemas;
using Microsoft.Extensions.DependencyInjection;

public class MultiDatabaseEventsWriter : IEventsWriter
{
    private readonly IServiceProvider _serviceProvider;
    private readonly EngineOptions _options;

    public MultiDatabaseEventsWriter(IServiceProvider serviceProvider, EngineOptions options)
    {
        _serviceProvider = serviceProvider;
        _options = options;
    }

    public async Task WriteBatch(IEnumerable<EventItem> events, TypeTuple key)
    {
        var eventList = events.ToList();
        if (eventList.Count == 0)
            return;

        var columns = ClickHouseSchemaConfig.GlobalSettings.GetMappedColumns(key);
        var values = eventList.Select(x => x.Values);

        // Route to appropriate ClickHouse instance based on table
        var connection = GetConnectionForTable(key.TableName);

        using var bulkCopy = new ClickHouseBulkCopy(connection)
        {
            MaxDegreeOfParallelism = _options.DatabaseMaxDegreeOfParallelism,
            BatchSize = _options.EventsFlushCount,
            DestinationTableName = $"{connection.Database}.{key.TableName}",
            ColumnNames = columns
        };

        await bulkCopy.InitAsync();
        await bulkCopy.WriteToServerAsync(values);
    }

    private ClickHouseConnection GetConnectionForTable(string tableName)
    {
        // Route logs to primary, metrics to analytics cluster
        var serviceKey = tableName switch
        {
            "logs" or "audit_log" => "primary",
            "metrics" or "analytics" => "analytics",
            _ => "primary"
        };

        return _serviceProvider.GetRequiredKeyedService<ClickHouseConnection>(serviceKey);
    }
}

// Program.cs configuration
public class Program
{
    public static void Main(string[] args)
    {
        var builder = WebApplication.CreateBuilder(args);

        // Register multiple ClickHouse connections with service keys
        builder.Services.AddClickHouseDataSource(
            "Host=primary-ch;Database=logs;",
            serviceKey: "primary");
        builder.Services.AddClickHouseDataSource(
            "Host=analytics-ch;Database=metrics;",
            serviceKey: "analytics");

        // Configure buffer WITHOUT connection string (handled by custom writer)
        builder.Services.ConfigureCHBuffer(options =>
        {
            options.EventsFlushPeriodSec = 2;
            options.EventsFlushCount = 10000;
        });

        // Register custom multi-database writer
        builder.Services.AddTransient<IEventsWriter, MultiDatabaseEventsWriter>();

        var app = builder.Build();
        app.Run();
    }
}
```

--------------------------------

### Inspect ClickHouse Schemas Programmatically

Source: https://context7.com/monqdl/monq.core.clickhousebuffer/llms.txt

Uses ClickHouseSchemaConfig to verify schema existence and retrieve mapped column names or values for specific types.

```csharp
using Monq.Core.ClickHouseBuffer.Schemas;

public class SchemaInspector
{
    public void InspectSchemas()
    {
        // Check if a schema exists for a type and table
        bool logsSchemaExists = ClickHouseSchemaConfig.GlobalSettings.SchemaExists<LogEvent>("logs");
        Console.WriteLine($"Logs schema exists: {logsSchemaExists}");

        // Get mapped column names for a type/table combination
        var logEvent = new LogEvent
        {
            StreamId = 123,
            StreamName = "main",
            AggregatedAt = DateTime.UtcNow,
            UserspaceId = 1,
            Value = "{\"data\": \"test\"}"
        };

        string[] columns = ClickHouseSchemaConfig.GlobalSettings.GetMappedColumns(logEvent, "logs");
        Console.WriteLine($"Columns: {string.Join(", ", columns)}");
        // Output: Columns: `_streamId`, `_streamName`, `_aggregatedAt`, `_userspaceId`, `_rawJson`

        // Get mapped values from an object
        object?[] values = ClickHouseSchemaConfig.GlobalSettings.GetMappedValues(logEvent, "logs");
        Console.WriteLine($"Values: {string.Join(", ", values)}");
        // Output: Values: 123, main, 2024-01-15 10:30:00, 1, {"data": "test"}

        // Get columns using TypeTuple key
        var key = new TypeTuple(typeof(LogEvent), "logs");
        string[] columnsByKey = ClickHouseSchemaConfig.GlobalSettings.GetMappedColumns(key);
    }
}

public class LogEvent
{
    public long StreamId { get; set; }
    public string StreamName { get; set; } = string.Empty;
    public DateTime AggregatedAt { get; set; }
    public int UserspaceId { get; set; }
    public string Value { get; set; } = string.Empty;
}
```

--------------------------------

### Register Multiple ClickHouse Data Sources

Source: https://github.com/monqdl/monq.core.clickhousebuffer/blob/master/README.md

Register named ClickHouse data sources using the service key for distinct connections. This is required when not passing a single ConnectionString to ConfigureCHBuffer.

```csharp
builder.Services.AddClickHouseDataSource(connectionString1, serviceKey: "ch1");
builder.Services.AddClickHouseDataSource(connectionString2, serviceKey: "ch2");
```

--------------------------------

### Register Custom IEventsWriter Implementation

Source: https://github.com/monqdl/monq.core.clickhousebuffer/blob/master/README.md

Register your custom IEventsWriter implementation with dependency injection. This should be done after configuring the ClickHouse buffer.

```csharp
// after builder.Services.ConfigureCHBuffer();
builder.Services.AddTransient<IEventsWriter, MultipleClickHouseEventsWriter>();
```

--------------------------------

### Map ClickHouse Columns with Reflection Attribute

Source: https://github.com/monqdl/monq.core.clickhousebuffer/blob/master/README.md

Use the `[ClickHouseColumn]` attribute to map class properties and fields to ClickHouse columns when a predefined schema is not used. Only columns with this attribute will be pushed to the database. This method is slower than using a predefined schema.

```csharp
class TestObject
{
    [ClickHouseColumn("publicProp")]
    public string PublicProp { get; set; } = Guid.NewGuid().ToString();

    [ClickHouseColumn("privateProp")]
    private string PrivateProp { get; set; } = Guid.NewGuid().ToString();

    public string GetPrivateProp() => PrivateProp;

    [ClickHouseColumn("publicField")]
    public string PublicField = Guid.NewGuid().ToString();

    [ClickHouseColumn("privateField")]
    private string _privateField = Guid.NewGuid().ToString();

    public string GetPrivateField() => _privateField;

    public string IgnoredProp { get; set; } = Guid.NewGuid().ToString();
}
```

--------------------------------

### IEventsBufferEngine.Add - Adding Raw EventItem

Source: https://context7.com/monqdl/monq.core.clickhousebuffer/llms.txt

Adds a pre-constructed EventItem directly to the buffer, giving you full control over the values array. This is useful when you've already extracted property values.

```APIDOC
## IEventsBufferEngine.Add - Adding Raw EventItem

### Description
Adds a pre-constructed `EventItem` directly to the buffer, giving you full control over the values array. This is useful when you've already extracted property values.

### Method
POST (conceptual, as it's a method call)

### Endpoint
N/A (This is a C# method call, not a direct HTTP endpoint)

### Parameters
#### Path Parameters
None

#### Query Parameters
None

#### Request Body
- **eventItem** (EventItem) - Required - The pre-constructed `EventItem` to add to the buffer.

### Request Example
```csharp
var values = new object?[]
{
    metricName,
    value,
    timestamp
};

var eventItem = new EventItem(
    tableName: "metrics",
    eventType: typeof(MetricRecord),
    values: values
);

_engine.Add(eventItem);
```

### Response
#### Success Response (200)
None (This is a method call, not an HTTP response)

#### Response Example
None
```

--------------------------------

### Define Global Schema Mapping

Source: https://github.com/monqdl/monq.core.clickhousebuffer/blob/master/README.md

Map model properties to table columns using global settings.

```csharp
ClickHouseSchemaConfig.GlobalSettings.NewConfig<MyEvent>("logs")
    .Map("_streamId", x => x.StreamId)
    .Map("_streamName", x => x.StreamName)
    .Map("_aggregatedAt", x => x.AggregatedAt)
    .Map("_userspaceId", x => x.UserspaceId)
    .Map("_rawJson", x => x.Value);
```

--------------------------------

### Add Raw EventItem to Buffer

Source: https://context7.com/monqdl/monq.core.clickhousebuffer/llms.txt

Use the Add method to insert a pre-constructed EventItem directly into the buffer. This provides full control over the values array, useful when data is already extracted. Ensure the values array matches the expected schema order for the target table.

```csharp
using Monq.Core.ClickHouseBuffer;

public class MetricsCollector
{
    private readonly IEventsBufferEngine _engine;

    public MetricsCollector(IEventsBufferEngine engine)
    {
        _engine = engine;
    }

    public void RecordMetric(string metricName, double value, DateTime timestamp)
    {
        // Manually construct values array matching your schema order
        var values = new object?[]
        {
            metricName,
            value,
            timestamp
        };

        var eventItem = new EventItem(
            tableName: "metrics",
            eventType: typeof(MetricRecord),
            values: values
        );

        _engine.Add(eventItem);
    }
}

public class MetricRecord
{
    public string Name { get; set; } = string.Empty;
    public double Value { get; set; }
    public DateTime Timestamp { get; set; }
}
```

--------------------------------

### Register Custom Events Handler for ClickHouse Buffer

Source: https://github.com/monqdl/monq.core.clickhousebuffer/blob/master/README.md

Implement the `IEventsHandler` interface to define custom methods like `OnAfterWriteEvents` or `OnWriteErrors`. Register your custom implementation in the dependency injection container.

```csharp
builder.Services.AddTransient<IEventsHandler, CustomEventsHandler>();
```

--------------------------------

### Register Custom Events Writer for ClickHouse Buffer

Source: https://github.com/monqdl/monq.core.clickhousebuffer/blob/master/README.md

Implement the `IEventsWriter` interface to provide a custom mechanism for saving buffered items to ClickHouse. Register your custom implementation in the dependency injection container after configuring the ClickHouse buffer.

```csharp
builder.Services.AddTransient<IEventsWriter, CustomEventsWriter>();
```

--------------------------------

### Implement IEventsHandler for Event Notification

Source: https://context7.com/monqdl/monq.core.clickhousebuffer/llms.txt

Provides a concrete implementation of IEventsHandler to log successful writes and handle errors by routing failed events to a dead-letter queue.

```csharp
using Monq.Core.ClickHouseBuffer;
using Microsoft.Extensions.Logging;

public class EventNotificationHandler : IEventsHandler
{
    private readonly ILogger<EventNotificationHandler> _logger;

    public EventNotificationHandler(ILogger<EventNotificationHandler> logger)
    {
        _logger = logger;
    }

    public async Task OnAfterWriteEvents(IEnumerable<EventItem> events)
    {
        var eventList = events.ToList();
        if (eventList.Count == 0)
            return;

        // Group by table for reporting
        var byTable = eventList.GroupBy(e => e.Key.TableName);
        foreach (var group in byTable)
        {
            _logger.LogInformation(
                "Successfully wrote {Count} events to table {Table}",
                group.Count(), group.Key);
        }

        // Additional post-processing: send metrics, trigger webhooks, etc.
        await SendMetricsAsync(eventList.Count);
    }

    public async Task OnWriteErrors(IEnumerable<EventItem> events)
    {
        var erroredEvents = events.ToList();

        _logger.LogError(
            "Failed to write {Count} events. Initiating error handling...",
            erroredEvents.Count);

        // Error handling: retry logic, dead letter queue, alerting, etc.
        foreach (var evt in erroredEvents)
        {
            _logger.LogWarning(
                "Failed event for table {Table}, type {Type}",
                evt.Key.TableName, evt.Key.Source.Name);
        }

        await SaveToDeadLetterQueueAsync(erroredEvents);
    }

    private Task SendMetricsAsync(int count)
    {
        // Send to metrics system
        return Task.CompletedTask;
    }

    private Task SaveToDeadLetterQueueAsync(List<EventItem> events)
    {
        // Save failed events for later retry
        return Task.CompletedTask;
    }
}

// Register in DI
// builder.Services.AddTransient<IEventsHandler, EventNotificationHandler>();
```

--------------------------------

### IEventsBufferEngine.AddEvent - Adding Events to Buffer

Source: https://context7.com/monqdl/monq.core.clickhousebuffer/llms.txt

Adds an object to the buffer for eventual writing to ClickHouse. Values and column names are calculated based on predefined schema or attribute mappings.

```APIDOC
## IEventsBufferEngine.AddEvent - Adding Events to Buffer

### Description
Adds an object to the buffer for eventual writing to ClickHouse. Values and column names are calculated based on predefined schema or attribute mappings.

### Method
POST (conceptual, as it's a method call)

### Endpoint
N/A (This is a C# method call, not a direct HTTP endpoint)

### Parameters
#### Path Parameters
None

#### Query Parameters
None

#### Request Body
- **eventItem** (object) - Required - The object to add to the buffer. The type `T` determines how values are extracted.
- **tableName** (string) - Required - The name of the ClickHouse table to which the event will be written.

### Request Example
```csharp
var logEvent = new LogEvent
{
    StreamId = streamId,
    StreamName = streamName,
    AggregatedAt = DateTime.UtcNow,
    UserspaceId = userspaceId,
    RawJson = logData
};

_engine.AddEvent(logEvent, "logs");
```

### Response
#### Success Response (200)
None (This is a method call, not an HTTP response)

#### Response Example
None
```

--------------------------------

### Add Typed Event Object to Buffer

Source: https://context7.com/monqdl/monq.core.clickhousebuffer/llms.txt

Use AddEvent<T> to add an object to the buffer. The engine maps object properties to ClickHouse columns based on schema or attribute mappings. Ensure the LogEvent class is defined with appropriate properties.

```csharp
using Monq.Core.ClickHouseBuffer;

public class LogEvent
{
    public long StreamId { get; set; }
    public string StreamName { get; set; } = string.Empty;
    public DateTime AggregatedAt { get; set; }
    public int UserspaceId { get; set; }
    public string RawJson { get; set; } = string.Empty;
}

public class LogProcessingService
{
    private readonly IEventsBufferEngine _engine;

    public LogProcessingService(IEventsBufferEngine engine)
    {
        _engine = engine;
    }

    public void ProcessLog(string logData, long streamId, string streamName, int userspaceId)
    {
        var logEvent = new LogEvent
        {
            StreamId = streamId,
            StreamName = streamName,
            AggregatedAt = DateTime.UtcNow,
            UserspaceId = userspaceId,
            RawJson = logData
        };

        // Add event to buffer - will be written when flush triggers
        _engine.AddEvent(logEvent, "logs");
    }

    public async Task ProcessBatchAsync(IEnumerable<string> logs, long streamId, string streamName, int userspaceId)
    {
        foreach (var log in logs)
        {
            ProcessLog(log, streamId, streamName, userspaceId);
        }

        // Optionally wait for current flush to complete
        await _engine.CompleteAsync();
    }
}
```

--------------------------------

### Map Properties with ClickHouseColumnAttribute

Source: https://context7.com/monqdl/monq.core.clickhousebuffer/llms.txt

Annotate properties or fields with [ClickHouseColumn] to include them in persistence. Members without the attribute are ignored by the reflection-based engine.

```csharp
using Monq.Core.ClickHouseBuffer.Attributes;
using Monq.Core.ClickHouseBuffer;

public class AuditEvent
{
    [ClickHouseColumn("event_id")]
    public Guid EventId { get; set; } = Guid.NewGuid();

    [ClickHouseColumn("user_id")]
    public long UserId { get; set; }

    [ClickHouseColumn("action_type")]
    public string ActionType { get; set; } = string.Empty;

    [ClickHouseColumn("resource_path")]
    public string ResourcePath { get; set; } = string.Empty;

    [ClickHouseColumn("timestamp")]
    public DateTime Timestamp { get; set; }

    [ClickHouseColumn("ip_address")]
    private string _ipAddress = string.Empty;

    public string GetIpAddress() => _ipAddress;
    public void SetIpAddress(string ip) => _ipAddress = ip;

    // This property will NOT be written to ClickHouse (no attribute)
    public string InternalNote { get; set; } = string.Empty;
}

public class AuditService
{
    private readonly IEventsBufferEngine _engine;

    public AuditService(IEventsBufferEngine engine)
    {
        _engine = engine;
    }

    public void LogAuditEvent(long userId, string action, string resource, string ipAddress)
    {
        var audit = new AuditEvent
        {
            UserId = userId,
            ActionType = action,
            ResourcePath = resource,
            Timestamp = DateTime.UtcNow,
            InternalNote = "This won't be saved" // Ignored - no attribute
        };
        audit.SetIpAddress(ipAddress);

        // Uses reflection to extract values from annotated properties/fields
        _engine.AddEvent(audit, "audit_log");
    }
}
```

--------------------------------

### Handle Custom EventItem Properties in WriteBatch

Source: https://github.com/monqdl/monq.core.clickhousebuffer/blob/master/README.md

When using custom EventItem models, cast the item to EventItemWithSourceObject within the WriteBatch method to access and process the 'Source' property.

```csharp
    public async Task WriteBatch(IEnumerable<EventItem> events, TypeTuple key)
    {
        ...
        foreach (EventItem item in events)
        {
            if (item is EventItemWithSourceObject eventWithSource)
            {
                // Working с item.Source
            }
        }
        ...
    }
```

--------------------------------

### AddEventWithSourceObject - Preserve Source Object Reference

Source: https://context7.com/monqdl/monq.core.clickhousebuffer/llms.txt

Use AddEventWithSourceObject to add events while keeping a reference to the original source object. This is useful in custom IEventsWriter implementations for accessing original data.

```csharp
using Monq.Core.ClickHouseBuffer;
using Monq.Core.ClickHouseBuffer.Extensions;
using Monq.Core.ClickHouseBuffer.Schemas;

public class EnrichedLogEvent
{
    public long StreamId { get; set; }
    public string Message { get; set; } = string.Empty;
    public DateTime Timestamp { get; set; }

    // Additional data not written to ClickHouse but needed for post-processing
    public string CorrelationId { get; set; } = string.Empty;
    public Dictionary<string, object> Context { get; set; } = new();
}

public class EnrichedLogService
{
    private readonly IEventsBufferEngine _engine;

    public EnrichedLogService(IEventsBufferEngine engine)
    {
        _engine = engine;
    }

    public void LogWithContext(string message, long streamId, string correlationId, Dictionary<string, object> context)
    {
        var logEvent = new EnrichedLogEvent
        {
            StreamId = streamId,
            Message = message,
            Timestamp = DateTime.UtcNow,
            CorrelationId = correlationId,
            Context = context
        };

        // Preserves reference to source object for use in custom IEventsWriter
        _engine.AddEventWithSourceObject(logEvent, "logs");
    }
}

// Custom writer that accesses the source object
public class CustomEventsWriter : IEventsWriter
{
    public async Task WriteBatch(IEnumerable<EventItem> events, TypeTuple key)
    {
        foreach (var item in events)
        {
            if (item is EventItemWithSourceObject eventWithSource)
            {
                var source = (EnrichedLogEvent)eventWithSource.Source;
                // Access original object properties like CorrelationId, Context
                Console.WriteLine($"Processing event with correlation: {source.CorrelationId}");
            }
        }

        // Perform actual write to ClickHouse...
        await Task.CompletedTask;
    }
}
```

=== COMPLETE CONTENT === This response contains all available snippets from this library. No additional content exists. Do not make further requests.