50K Messages Per Second: Event Streaming with Noundry Streams

namespace NoundryStreams.Core.Models;

public class StreamConfig
{
    /// <summary>
    /// Unique identifier for this stream. Used as the SQLite database filename.
    /// </summary>
    public string StreamId { get; set; } = string.Empty;

    /// <summary>
    /// Human-readable name shown in admin UI and logs.
    /// </summary>
    public string DisplayName { get; set; } = string.Empty;

    /// <summary>
    /// When true, messages are delivered to relay endpoints in the order
    /// they were ingested. Limits relay to a single worker.
    /// </summary>
    public bool OrderingRequired { get; set; } = false;

    /// <summary>
    /// How long to retain messages in SQLite before cleanup. Default: 168 hours (7 days).
    /// </summary>
    public int RetentionHours { get; set; } = 168;

    /// <summary>
    /// Bounded channel capacity. When full, ingest calls will apply backpressure
    /// (Wait mode) rather than dropping messages.
    /// </summary>
    public int ChannelCapacity { get; set; } = 100_000;

    /// <summary>
    /// Maximum time in milliseconds the batch writer will wait before flushing
    /// a partial batch to SQLite. Lower values reduce latency; higher values
    /// improve throughput via larger batch sizes.
    /// </summary>
    public int BatchIntervalMs { get; set; } = 50;

    /// <summary>
    /// Maximum number of messages per batch insert. The batch writer flushes
    /// when this count is reached OR when BatchIntervalMs elapses, whichever
    /// comes first.
    /// </summary>
    public int BatchMaxSize { get; set; } = 5_000;

    /// <summary>
    /// Number of concurrent relay workers for this stream. More workers means
    /// higher fan-out throughput but more HTTP connections. Forced to 1 when
    /// OrderingRequired is true.
    /// </summary>
    public int RelayWorkerCount { get; set; } = 4;

    /// <summary>
    /// Whether this stream is actively processing messages.
    /// </summary>
    public bool Enabled { get; set; } = true;

    /// <summary>
    /// Relay configurations for this stream. Each config is an HTTP endpoint
    /// that receives copies of every message.
    /// </summary>
    public List<RelayConfig> RelayConfigs { get; set; } = new();
}

namespace NoundryStreams.Core.Models;

public class RelayConfig
{
    /// <summary>
    /// Unique identifier for this relay configuration.
    /// </summary>
    public string ConfigId { get; set; } = Guid.NewGuid().ToString("N");

    /// <summary>
    /// The stream this relay belongs to.
    /// </summary>
    public string StreamId { get; set; } = string.Empty;

    /// <summary>
    /// The HTTP endpoint URL where messages will be POSTed.
    /// </summary>
    public string EndpointUrl { get; set; } = string.Empty;

    /// <summary>
    /// HTTP method to use. Defaults to POST. Some webhooks require PUT.
    /// </summary>
    public string HttpMethod { get; set; } = "POST";

    /// <summary>
    /// Custom headers to include with every relay request. Useful for
    /// API keys, correlation IDs, or content type overrides.
    /// </summary>
    public Dictionary<string, string> Headers { get; set; } = new()
    {
        ["Content-Type"] = "application/json"
    };

    /// <summary>
    /// HTTP request timeout in milliseconds. Keep this short to avoid
    /// blocking relay workers on slow consumers.
    /// </summary>
    public int TimeoutMs { get; set; } = 5_000;

    /// <summary>
    /// Number of retry attempts before a message is moved to the dead letter queue.
    /// </summary>
    public int RetryAttempts { get; set; } = 5;

    /// <summary>
    /// Base delay between retries in milliseconds. Actual delay uses exponential
    /// backoff: RetryBackoffMs * 2^(attemptNumber - 1).
    /// </summary>
    public int RetryBackoffMs { get; set; } = 1_000;

    /// <summary>
    /// Whether this relay configuration is active.
    /// </summary>
    public bool Enabled { get; set; } = true;
}

{
  "Logging": {
    "LogLevel": {
      "Default": "Information",
      "NoundryStreams": "Debug"
    }
  },
  "NoundryStreams": {
    "DataPath": "./data/streams",
    "ApiKey": "your-secure-api-key-here",
    "Defaults": {
      "ChannelCapacity": 100000,
      "BatchIntervalMs": 50,
      "BatchMaxSize": 5000,
      "RetentionHours": 168,
      "RelayWorkerCount": 4,
      "RelayTimeoutMs": 5000,
      "RelayRetryAttempts": 5,
      "RelayRetryBackoffMs": 1000
    },
    "Sqlite": {
      "CacheSizeKb": 65536,
      "MmapSizeMb": 256,
      "WalAutoCheckpoint": 1000
    }
  }
}

using System.Text.Json;
using Microsoft.AspNetCore.Mvc;
using NoundryStreams.Core.Models;
using NoundryStreams.Core.Services;

namespace NoundryStreams.Service.Endpoints;

public static class IngestEndpoints
{
    public static void MapIngestEndpoints(this WebApplication app)
    {
        app.MapPost("/streams/{streamId}/messages", async (
            string streamId,
            [FromHeader(Name = "X-Api-Key")] string apiKey,
            [FromBody] JsonElement body,
            StreamManager streamManager,
            IConfiguration config) =>
        {
            // Validate API key
            var expectedKey = config["NoundryStreams:ApiKey"] ?? "";
            if (string.IsNullOrEmpty(apiKey) || apiKey != expectedKey)
                return Results.Unauthorized();

            // Ensure the stream exists and is active
            var stream = streamManager.GetStream(streamId);
            if (stream is null)
                return Results.NotFound(new { error = $"Stream '{streamId}' not found." });

            if (!stream.Config.Enabled)
                return Results.Problem("Stream is disabled.", statusCode: 503);

            // Build the message
            var message = new StreamMessage
            {
                MessageId = Guid.NewGuid().ToString("N"),
                StreamId = streamId,
                Payload = body.GetRawText(),
                CreatedAt = DateTime.UtcNow,
                SequenceNumber = stream.GetNextSequence()
            };

            // Write to channel — this is the sub-millisecond path.
            // TryWrite is lock-free. If the channel is full, we signal
            // backpressure via 429 rather than blocking or dropping.
            if (!stream.IngestChannel.Writer.TryWrite(message))
            {
                return Results.StatusCode(429); // Too Many Requests — backpressure
            }

            return Results.Ok(new
            {
                messageId = message.MessageId,
                stream = streamId,
                sequence = message.SequenceNumber,
                status = "accepted"
            });
        })
        .WithName("IngestMessage")
        .WithTags("Ingest");

        // Batch ingest endpoint for high-throughput producers
        app.MapPost("/streams/{streamId}/messages/batch", async (
            string streamId,
            [FromHeader(Name = "X-Api-Key")] string apiKey,
            [FromBody] JsonElement[] bodies,
            StreamManager streamManager,
            IConfiguration config) =>
        {
            var expectedKey = config["NoundryStreams:ApiKey"] ?? "";
            if (string.IsNullOrEmpty(apiKey) || apiKey != expectedKey)
                return Results.Unauthorized();

            var stream = streamManager.GetStream(streamId);
            if (stream is null)
                return Results.NotFound(new { error = $"Stream '{streamId}' not found." });

            var accepted = 0;
            var rejected = 0;

            foreach (var body in bodies)
            {
                var message = new StreamMessage
                {
                    MessageId = Guid.NewGuid().ToString("N"),
                    StreamId = streamId,
                    Payload = body.GetRawText(),
                    CreatedAt = DateTime.UtcNow,
                    SequenceNumber = stream.GetNextSequence()
                };

                if (stream.IngestChannel.Writer.TryWrite(message))
                    accepted++;
                else
                    rejected++;
            }

            var statusCode = rejected > 0 ? 207 : 200; // Multi-Status if partial
            return Results.Json(new { accepted, rejected, total = bodies.Length }, statusCode: statusCode);
        })
        .WithName("IngestBatch")
        .WithTags("Ingest");
    }
}

using System.Threading.Channels;
using Microsoft.Data.Sqlite;
using Microsoft.Extensions.Logging;
using NoundryStreams.Core.Models;

namespace NoundryStreams.Core.Services;

public class BatchWriter : IAsyncDisposable
{
    private readonly StreamInstance _stream;
    private readonly ILogger _logger;
    private readonly CancellationTokenSource _cts = new();
    private readonly Task _writerTask;

    public BatchWriter(StreamInstance stream, ILogger<BatchWriter> logger)
    {
        _stream = stream;
        _logger = logger;
        _writerTask = Task.Run(() => WriterLoopAsync(_cts.Token));
    }

    private async Task WriterLoopAsync(CancellationToken ct)
    {
        var config = _stream.Config;
        var batch = new List<StreamMessage>(config.BatchMaxSize);
        var reader = _stream.IngestChannel.Reader;

        _logger.LogInformation(
            "BatchWriter started for stream '{StreamId}' (batch={BatchSize}, interval={IntervalMs}ms)",
            config.StreamId, config.BatchMaxSize, config.BatchIntervalMs);

        while (!ct.IsCancellationRequested)
        {
            try
            {
                batch.Clear();

                // Wait for at least one message (blocks until available)
                if (await reader.WaitToReadAsync(ct))
                {
                    // Drain the channel up to BatchMaxSize or until BatchIntervalMs elapses
                    using var batchCts = CancellationTokenSource
                        .CreateLinkedTokenSource(ct);
                    batchCts.CancelAfter(config.BatchIntervalMs);

                    try
                    {
                        while (batch.Count < config.BatchMaxSize &&
                               reader.TryRead(out var message))
                        {
                            batch.Add(message);
                        }

                        // If we haven't hit max size, keep draining until interval elapses
                        while (batch.Count < config.BatchMaxSize)
                        {
                            var msg = await reader.ReadAsync(batchCts.Token);
                            batch.Add(msg);
                        }
                    }
                    catch (OperationCanceledException)
                    {
                        // Interval elapsed — flush what we have
                    }
                }

                if (batch.Count > 0)
                {
                    await FlushBatchAsync(batch);
                }
            }
            catch (OperationCanceledException) when (ct.IsCancellationRequested)
            {
                break;
            }
            catch (Exception ex)
            {
                _logger.LogError(ex,
                    "BatchWriter error for stream '{StreamId}'. Retrying in 1s.",
                    config.StreamId);
                await Task.Delay(1000, ct);
            }
        }

        // Flush any remaining messages on shutdown
        while (reader.TryRead(out var remaining))
            batch.Add(remaining);

        if (batch.Count > 0)
            await FlushBatchAsync(batch);

        _logger.LogInformation("BatchWriter stopped for stream '{StreamId}'.", config.StreamId);
    }

    private async Task FlushBatchAsync(List<StreamMessage> batch)
    {
        await using var connection = _stream.CreateConnection();
        await connection.OpenAsync();
        await using var transaction = await connection.BeginTransactionAsync();

        // Parameterized bulk insert — safe from injection, fast for SQLite
        await using var cmd = connection.CreateCommand();
        cmd.Transaction = (SqliteTransaction)transaction;

        foreach (var msg in batch)
        {
            cmd.CommandText = @"
                INSERT INTO messages (message_id, stream_id, payload, sequence_number, created_at, status)
                VALUES ($id, $stream, $payload, $seq, $created, 'pending')";

            cmd.Parameters.Clear();
            cmd.Parameters.AddWithValue("$id", msg.MessageId);
            cmd.Parameters.AddWithValue("$stream", msg.StreamId);
            cmd.Parameters.AddWithValue("$payload", msg.Payload);
            cmd.Parameters.AddWithValue("$seq", msg.SequenceNumber);
            cmd.Parameters.AddWithValue("$created", msg.CreatedAt.ToString("O"));

            await cmd.ExecuteNonQueryAsync();
        }

        await transaction.CommitAsync();

        _logger.LogDebug(
            "Flushed {Count} messages to SQLite for stream '{StreamId}'.",
            batch.Count, _stream.Config.StreamId);

        // Push persisted message IDs to the relay channel
        foreach (var msg in batch)
        {
            msg.Status = "persisted";
            await _stream.RelayChannel.Writer.WriteAsync(msg);
        }
    }

    public async ValueTask DisposeAsync()
    {
        _cts.Cancel();
        await _writerTask;
        _cts.Dispose();
    }
}

using System.Net.Http.Headers;
using System.Text;
using Microsoft.Data.Sqlite;
using Microsoft.Extensions.Logging;
using NoundryStreams.Core.Models;

namespace NoundryStreams.Core.Services;

public class RelayWorker : IAsyncDisposable
{
    private readonly StreamInstance _stream;
    private readonly RelayConfig _relayConfig;
    private readonly HttpClient _httpClient;
    private readonly ILogger _logger;
    private readonly CancellationTokenSource _cts = new();
    private readonly Task _workerTask;

    public RelayWorker(
        StreamInstance stream,
        RelayConfig relayConfig,
        IHttpClientFactory httpClientFactory,
        ILogger<RelayWorker> logger)
    {
        _stream = stream;
        _relayConfig = relayConfig;
        _httpClient = httpClientFactory.CreateClient("NoundryStreamsRelay");
        _httpClient.Timeout = TimeSpan.FromMilliseconds(relayConfig.TimeoutMs);
        _logger = logger;
        _workerTask = Task.Run(() => RelayLoopAsync(_cts.Token));
    }

    private async Task RelayLoopAsync(CancellationToken ct)
    {
        var reader = _stream.RelayChannel.Reader;

        _logger.LogInformation(
            "RelayWorker started for stream '{StreamId}' -> {Endpoint}",
            _stream.Config.StreamId, _relayConfig.EndpointUrl);

        await foreach (var message in reader.ReadAllAsync(ct))
        {
            if (!_relayConfig.Enabled) continue;

            var delivered = false;
            Exception? lastException = null;

            for (var attempt = 1; attempt <= _relayConfig.RetryAttempts; attempt++)
            {
                try
                {
                    await DeliverMessageAsync(message, ct);
                    delivered = true;

                    // Mark as delivered in SQLite
                    await UpdateMessageStatusAsync(message.MessageId, "delivered");

                    _logger.LogDebug(
                        "Delivered message {MessageId} to {Endpoint} (attempt {Attempt})",
                        message.MessageId, _relayConfig.EndpointUrl, attempt);
                    break;
                }
                catch (Exception ex)
                {
                    lastException = ex;
                    _logger.LogWarning(
                        "Relay attempt {Attempt}/{Max} failed for message {MessageId}: {Error}",
                        attempt, _relayConfig.RetryAttempts, message.MessageId, ex.Message);

                    if (attempt < _relayConfig.RetryAttempts)
                    {
                        // Exponential backoff: base * 2^(attempt-1)
                        var delay = _relayConfig.RetryBackoffMs * (int)Math.Pow(2, attempt - 1);
                        await Task.Delay(delay, ct);
                    }
                }
            }

            if (!delivered)
            {
                _logger.LogError(
                    "Message {MessageId} exhausted all {Max} retry attempts. Moving to dead letter queue.",
                    message.MessageId, _relayConfig.RetryAttempts);

                await MoveToDeadLetterAsync(message, lastException);
            }
        }
    }

    private async Task DeliverMessageAsync(StreamMessage message, CancellationToken ct)
    {
        var request = new HttpRequestMessage(
            new HttpMethod(_relayConfig.HttpMethod),
            _relayConfig.EndpointUrl);

        request.Content = new StringContent(message.Payload, Encoding.UTF8, "application/json");

        // Apply custom headers from relay config
        foreach (var header in _relayConfig.Headers)
        {
            if (header.Key.StartsWith("Content-", StringComparison.OrdinalIgnoreCase))
                continue; // Content headers are set on the content object
            request.Headers.TryAddWithoutValidation(header.Key, header.Value);
        }

        // Add tracing headers
        request.Headers.TryAddWithoutValidation("X-Stream-Id", message.StreamId);
        request.Headers.TryAddWithoutValidation("X-Message-Id", message.MessageId);
        request.Headers.TryAddWithoutValidation("X-Sequence-Number", message.SequenceNumber.ToString());

        var response = await _httpClient.SendAsync(request, ct);
        response.EnsureSuccessStatusCode();
    }

    private async Task UpdateMessageStatusAsync(string messageId, string status)
    {
        await using var connection = _stream.CreateConnection();
        await connection.OpenAsync();
        await using var cmd = connection.CreateCommand();
        cmd.CommandText = "UPDATE messages SET status = $status WHERE message_id = $id";
        cmd.Parameters.AddWithValue("$status", status);
        cmd.Parameters.AddWithValue("$id", messageId);
        await cmd.ExecuteNonQueryAsync();
    }

    private async Task MoveToDeadLetterAsync(StreamMessage message, Exception? exception)
    {
        await using var connection = _stream.CreateConnection();
        await connection.OpenAsync();
        await using var cmd = connection.CreateCommand();
        cmd.CommandText = @"
            INSERT INTO dead_letter_messages
                (message_id, stream_id, relay_config_id, payload, error_message,
                 attempt_count, original_created_at, dead_lettered_at)
            VALUES ($msgId, $stream, $configId, $payload, $error,
                    $attempts, $created, $deadLettered)";

        cmd.Parameters.AddWithValue("$msgId", message.MessageId);
        cmd.Parameters.AddWithValue("$stream", message.StreamId);
        cmd.Parameters.AddWithValue("$configId", _relayConfig.ConfigId);
        cmd.Parameters.AddWithValue("$payload", message.Payload);
        cmd.Parameters.AddWithValue("$error", exception?.Message ?? "Unknown error");
        cmd.Parameters.AddWithValue("$attempts", _relayConfig.RetryAttempts);
        cmd.Parameters.AddWithValue("$created", message.CreatedAt.ToString("O"));
        cmd.Parameters.AddWithValue("$deadLettered", DateTime.UtcNow.ToString("O"));

        await cmd.ExecuteNonQueryAsync();

        // Update original message status
        await UpdateMessageStatusAsync(message.MessageId, "dead_lettered");
    }

    public async ValueTask DisposeAsync()
    {
        _cts.Cancel();
        await _workerTask;
        _httpClient.Dispose();
        _cts.Dispose();
    }
}

namespace NoundryStreams.Core.Models;

/// <summary>
/// Represents a message that failed all relay attempts and has been
/// moved to the dead letter queue for manual inspection or replay.
/// </summary>
public record DeadLetterMessage
{
    public string MessageId { get; init; } = string.Empty;
    public string StreamId { get; init; } = string.Empty;
    public string RelayConfigId { get; init; } = string.Empty;
    public string Payload { get; init; } = string.Empty;
    public string ErrorMessage { get; init; } = string.Empty;
    public int AttemptCount { get; init; }
    public DateTime OriginalCreatedAt { get; init; }
    public DateTime DeadLetteredAt { get; init; }
}

using Microsoft.AspNetCore.Mvc;
using NoundryStreams.Core.Models;
using NoundryStreams.Core.Services;

namespace NoundryStreams.Service.Endpoints;

public static class AdminEndpoints
{
    public static void MapAdminEndpoints(this WebApplication app)
    {
        var admin = app.MapGroup("/streams/admin")
            .WithTags("Admin");

        // List all streams with their current stats
        admin.MapGet("/", (StreamManager manager) =>
        {
            var streams = manager.GetAllStreams().Select(s => new
            {
                s.Config.StreamId,
                s.Config.DisplayName,
                s.Config.Enabled,
                IngestChannelCount = s.IngestChannel.Reader.Count,
                RelayChannelCount = s.RelayChannel.Reader.Count,
                RelayConfigs = s.Config.RelayConfigs.Count,
                s.Config.RelayWorkerCount
            });
            return Results.Ok(streams);
        });

        // Get dead letter messages for a specific stream
        admin.MapGet("/{streamId}/dead-letters", async (
            string streamId,
            StreamManager manager,
            [FromQuery] int limit = 50,
            [FromQuery] int offset = 0) =>
        {
            var stream = manager.GetStream(streamId);
            if (stream is null)
                return Results.NotFound(new { error = $"Stream '{streamId}' not found." });

            await using var connection = stream.CreateConnection();
            await connection.OpenAsync();
            await using var cmd = connection.CreateCommand();
            cmd.CommandText = @"
                SELECT message_id, stream_id, relay_config_id, payload,
                       error_message, attempt_count, original_created_at, dead_lettered_at
                FROM dead_letter_messages
                ORDER BY dead_lettered_at DESC
                LIMIT $limit OFFSET $offset";
            cmd.Parameters.AddWithValue("$limit", limit);
            cmd.Parameters.AddWithValue("$offset", offset);

            var deadLetters = new List<DeadLetterMessage>();
            await using var reader = await cmd.ExecuteReaderAsync();
            while (await reader.ReadAsync())
            {
                deadLetters.Add(new DeadLetterMessage
                {
                    MessageId = reader.GetString(0),
                    StreamId = reader.GetString(1),
                    RelayConfigId = reader.GetString(2),
                    Payload = reader.GetString(3),
                    ErrorMessage = reader.GetString(4),
                    AttemptCount = reader.GetInt32(5),
                    OriginalCreatedAt = DateTime.Parse(reader.GetString(6)),
                    DeadLetteredAt = DateTime.Parse(reader.GetString(7))
                });
            }

            return Results.Ok(new { streamId, count = deadLetters.Count, deadLetters });
        });

        // Replay dead letter messages back into the relay pipeline
        admin.MapPost("/{streamId}/dead-letters/replay", async (
            string streamId,
            [FromBody] ReplayRequest request,
            StreamManager manager) =>
        {
            var stream = manager.GetStream(streamId);
            if (stream is null)
                return Results.NotFound(new { error = $"Stream '{streamId}' not found." });

            var replayed = 0;

            await using var connection = stream.CreateConnection();
            await connection.OpenAsync();

            foreach (var messageId in request.MessageIds)
            {
                await using var cmd = connection.CreateCommand();
                cmd.CommandText = "SELECT payload, stream_id FROM dead_letter_messages WHERE message_id = $id";
                cmd.Parameters.AddWithValue("$id", messageId);

                await using var reader = await cmd.ExecuteReaderAsync();
                if (await reader.ReadAsync())
                {
                    var message = new StreamMessage
                    {
                        MessageId = messageId,
                        StreamId = reader.GetString(1),
                        Payload = reader.GetString(0),
                        CreatedAt = DateTime.UtcNow,
                        SequenceNumber = stream.GetNextSequence(),
                        Status = "replayed"
                    };

                    await stream.RelayChannel.Writer.WriteAsync(message);
                    replayed++;

                    // Remove from dead letter queue
                    await using var deleteCmd = connection.CreateCommand();
                    deleteCmd.CommandText = "DELETE FROM dead_letter_messages WHERE message_id = $id";
                    deleteCmd.Parameters.AddWithValue("$id", messageId);
                    await deleteCmd.ExecuteNonQueryAsync();
                }
            }

            return Results.Ok(new { replayed, requested = request.MessageIds.Count });
        });

        // Purge all dead letter messages for a stream
        admin.MapDelete("/{streamId}/dead-letters", async (
            string streamId,
            StreamManager manager) =>
        {
            var stream = manager.GetStream(streamId);
            if (stream is null)
                return Results.NotFound(new { error = $"Stream '{streamId}' not found." });

            await using var connection = stream.CreateConnection();
            await connection.OpenAsync();
            await using var cmd = connection.CreateCommand();
            cmd.CommandText = "DELETE FROM dead_letter_messages";
            var deleted = await cmd.ExecuteNonQueryAsync();

            return Results.Ok(new { streamId, deleted });
        });
    }
}

public record ReplayRequest(List<string> MessageIds);

using Microsoft.Data.Sqlite;
using Microsoft.Extensions.Logging;
using NoundryStreams.Core.Models;

namespace NoundryStreams.Core.Services;

public static class DatabaseInitializer
{
    /// <summary>
    /// Initializes a SQLite database for a stream with performance-tuned pragmas
    /// and the required schema. Each stream gets its own database file.
    /// </summary>
    public static async Task InitializeAsync(
        string dbPath,
        StreamsConfig config,
        ILogger logger)
    {
        var connectionString = new SqliteConnectionStringBuilder
        {
            DataSource = dbPath,
            Mode = SqliteOpenMode.ReadWriteCreate,
            Cache = SqliteCacheMode.Shared
        }.ToString();

        await using var connection = new SqliteConnection(connectionString);
        await connection.OpenAsync();

        // ── Performance Pragmas ──────────────────────────────────────
        var pragmas = new[]
        {
            // WAL mode: readers never block writers, writers never block readers.
            // Critical for concurrent BatchWriter inserts + RelayWorker reads.
            "PRAGMA journal_mode = WAL;",

            // NORMAL sync: fsync at commit boundaries only. Trades a tiny crash
            // window (last transaction may be lost) for 5-10x write throughput.
            // Acceptable for a messaging system where the producer can retry.
            "PRAGMA synchronous = NORMAL;",

            // 64MB page cache (negative value = KB). Keeps hot message pages in
            // memory so the BatchWriter's INSERT and RelayWorker's SELECT don't
            // hit disk for recently-written messages.
            $"PRAGMA cache_size = -{config.Sqlite.CacheSizeKb};",

            // Memory-mapped I/O up to 256MB. The OS maps the database file into
            // virtual memory, eliminating read() syscalls and letting the kernel's
            // page cache handle eviction.
            $"PRAGMA mmap_size = {config.Sqlite.MmapSizeMb * 1024L * 1024L};",

            // Keep temp tables in RAM. Sorting and grouping for admin queries
            // stay in memory rather than spilling to temp files.
            "PRAGMA temp_store = MEMORY;",

            // Checkpoint WAL every 1000 pages (~4MB). Prevents unbounded WAL
            // growth during sustained high-throughput ingestion.
            $"PRAGMA wal_autocheckpoint = {config.Sqlite.WalAutoCheckpoint};",

            // Wait up to 5 seconds when the database is locked. Prevents
            // SQLITE_BUSY errors during checkpoint operations.
            "PRAGMA busy_timeout = 5000;",

            // 4KB page size matches the OS page size for aligned I/O.
            "PRAGMA page_size = 4096;"
        };

        foreach (var pragma in pragmas)
        {
            await using var cmd = connection.CreateCommand();
            cmd.CommandText = pragma;
            await cmd.ExecuteNonQueryAsync();
        }

        logger.LogInformation(
            "SQLite pragmas applied for database at {DbPath}", dbPath);

        // ── Schema ───────────────────────────────────────────────────
        await using var schemaCmd = connection.CreateCommand();
        schemaCmd.CommandText = Schema;
        await schemaCmd.ExecuteNonQueryAsync();

        logger.LogInformation(
            "Schema initialized for database at {DbPath}", dbPath);
    }

    private const string Schema = @"
        -- Stream metadata
        CREATE TABLE IF NOT EXISTS streams (
            stream_id       TEXT PRIMARY KEY,
            display_name    TEXT NOT NULL,
            config_json     TEXT NOT NULL,
            created_at      TEXT NOT NULL DEFAULT (datetime('now')),
            updated_at      TEXT
        );

        -- Ingested messages
        CREATE TABLE IF NOT EXISTS messages (
            message_id      TEXT PRIMARY KEY,
            stream_id       TEXT NOT NULL,
            payload         TEXT NOT NULL,
            sequence_number INTEGER NOT NULL,
            created_at      TEXT NOT NULL,
            status          TEXT NOT NULL DEFAULT 'pending',
            delivered_at    TEXT
        );

        CREATE INDEX IF NOT EXISTS idx_messages_stream_status
            ON messages (stream_id, status);

        CREATE INDEX IF NOT EXISTS idx_messages_stream_sequence
            ON messages (stream_id, sequence_number);

        CREATE INDEX IF NOT EXISTS idx_messages_created_at
            ON messages (created_at);

        -- Relay endpoint configurations
        CREATE TABLE IF NOT EXISTS relay_configs (
            config_id       TEXT PRIMARY KEY,
            stream_id       TEXT NOT NULL,
            endpoint_url    TEXT NOT NULL,
            http_method     TEXT NOT NULL DEFAULT 'POST',
            headers_json    TEXT,
            timeout_ms      INTEGER NOT NULL DEFAULT 5000,
            retry_attempts  INTEGER NOT NULL DEFAULT 5,
            retry_backoff_ms INTEGER NOT NULL DEFAULT 1000,
            enabled         INTEGER NOT NULL DEFAULT 1,
            created_at      TEXT NOT NULL DEFAULT (datetime('now'))
        );

        CREATE INDEX IF NOT EXISTS idx_relay_configs_stream
            ON relay_configs (stream_id);

        -- Relay attempt log (for diagnostics)
        CREATE TABLE IF NOT EXISTS relay_attempts (
            id              INTEGER PRIMARY KEY AUTOINCREMENT,
            message_id      TEXT NOT NULL,
            config_id       TEXT NOT NULL,
            attempt_number  INTEGER NOT NULL,
            status_code     INTEGER,
            error_message   TEXT,
            duration_ms     INTEGER,
            attempted_at    TEXT NOT NULL DEFAULT (datetime('now'))
        );

        CREATE INDEX IF NOT EXISTS idx_relay_attempts_message
            ON relay_attempts (message_id);

        -- Dead letter queue
        CREATE TABLE IF NOT EXISTS dead_letter_messages (
            message_id          TEXT PRIMARY KEY,
            stream_id           TEXT NOT NULL,
            relay_config_id     TEXT NOT NULL,
            payload             TEXT NOT NULL,
            error_message       TEXT,
            attempt_count       INTEGER NOT NULL,
            original_created_at TEXT NOT NULL,
            dead_lettered_at    TEXT NOT NULL DEFAULT (datetime('now'))
        );

        CREATE INDEX IF NOT EXISTS idx_dead_letter_stream
            ON dead_letter_messages (stream_id);

        CREATE INDEX IF NOT EXISTS idx_dead_letter_date
            ON dead_letter_messages (dead_lettered_at);
    ";
}