diff --git a/db/migrations/0003_selfhost_runtime.sql b/db/migrations/0003_selfhost_runtime.sql new file mode 100644 index 0000000..b00fbd1 --- /dev/null +++ b/db/migrations/0003_selfhost_runtime.sql @@ -0,0 +1,112 @@ +-- Runtime tables for the self-host backend (Backend:Mode=Postgres, D53): queues, delivery state, rollups, event +-- archive, idempotency keys and monthly counters. The AWS backend uses SQS, DynamoDB and S3 instead and ignores these. + +-- A durable queue: messages become visible at visible_at, are leased by pushing visible_at forward, and move to +-- "-dlq" after too many receives (the same contract as SQS standard queues). +CREATE TABLE webhook.queue_messages ( + id bigserial PRIMARY KEY, + queue text NOT NULL, + body text NOT NULL, + enqueued_at timestamptz NOT NULL DEFAULT now(), + visible_at timestamptz NOT NULL DEFAULT now(), + receive_count integer NOT NULL DEFAULT 0, + receipt uuid +); +CREATE INDEX queue_messages_ready_idx ON webhook.queue_messages (queue, visible_at, id); + +-- Non-success deliveries only (docs/data-plane.md section 4). +CREATE TABLE webhook.delivery_state ( + tenant_id text NOT NULL, + delivery_id text NOT NULL, + endpoint_id text NOT NULL, + event_id text NOT NULL, + event_type text NOT NULL, + occurred_at timestamptz NOT NULL, + status text NOT NULL, + attempt integer NOT NULL, + tier text NOT NULL, + next_attempt_at timestamptz, + last_status_code integer, + last_error text, + payload text, + payload_ref text, + created_at timestamptz NOT NULL, + updated_at timestamptz NOT NULL, + expires_at timestamptz NOT NULL, + PRIMARY KEY (tenant_id, delivery_id) +); +CREATE INDEX delivery_state_due_idx ON webhook.delivery_state (next_attempt_at) WHERE status = 'Scheduled'; +CREATE INDEX delivery_state_tenant_status_idx ON webhook.delivery_state (tenant_id, status, updated_at DESC); +CREATE INDEX delivery_state_endpoint_idx ON webhook.delivery_state (tenant_id, endpoint_id, updated_at DESC); +CREATE INDEX delivery_state_expiry_idx ON webhook.delivery_state (expires_at); + +-- Per tenant/endpoint and minute counters and latency histograms, added to with ON CONFLICT (docs/data-plane.md). +CREATE TABLE webhook.rollups ( + pk text NOT NULL, + sk text NOT NULL, + expires_at timestamptz NOT NULL, + created bigint NOT NULL DEFAULT 0, + attempts bigint NOT NULL DEFAULT 0, + succeeded bigint NOT NULL DEFAULT 0, + retried bigint NOT NULL DEFAULT 0, + failed bigint NOT NULL DEFAULT 0, + exhausted bigint NOT NULL DEFAULT 0, + timeouts bigint NOT NULL DEFAULT 0, + connection_ms bigint NOT NULL DEFAULT 0, + rt_sum bigint NOT NULL DEFAULT 0, + qw_sum bigint NOT NULL DEFAULT 0, + rt0 bigint NOT NULL DEFAULT 0, + rt1 bigint NOT NULL DEFAULT 0, + rt2 bigint NOT NULL DEFAULT 0, + rt3 bigint NOT NULL DEFAULT 0, + rt4 bigint NOT NULL DEFAULT 0, + rt5 bigint NOT NULL DEFAULT 0, + rt6 bigint NOT NULL DEFAULT 0, + rt7 bigint NOT NULL DEFAULT 0, + rt8 bigint NOT NULL DEFAULT 0, + rt9 bigint NOT NULL DEFAULT 0, + rt10 bigint NOT NULL DEFAULT 0, + rt11 bigint NOT NULL DEFAULT 0, + rt12 bigint NOT NULL DEFAULT 0, + qw0 bigint NOT NULL DEFAULT 0, + qw1 bigint NOT NULL DEFAULT 0, + qw2 bigint NOT NULL DEFAULT 0, + qw3 bigint NOT NULL DEFAULT 0, + qw4 bigint NOT NULL DEFAULT 0, + qw5 bigint NOT NULL DEFAULT 0, + qw6 bigint NOT NULL DEFAULT 0, + qw7 bigint NOT NULL DEFAULT 0, + qw8 bigint NOT NULL DEFAULT 0, + qw9 bigint NOT NULL DEFAULT 0, + qw10 bigint NOT NULL DEFAULT 0, + qw11 bigint NOT NULL DEFAULT 0, + qw12 bigint NOT NULL DEFAULT 0, + PRIMARY KEY (pk, sk) +); +CREATE INDEX rollups_expiry_idx ON webhook.rollups (expires_at); + +-- Event archive: one gzip JSON Lines batch per row (the replay source and home of large payloads). +CREATE TABLE webhook.archive_batches ( + key text PRIMARY KEY, + body bytea NOT NULL, + event_count integer NOT NULL, + created_at timestamptz NOT NULL DEFAULT now(), + expires_at timestamptz NOT NULL +); +CREATE INDEX archive_batches_expiry_idx ON webhook.archive_batches (expires_at); + +CREATE TABLE webhook.idempotency_keys ( + tenant_id text NOT NULL, + key text NOT NULL, + event_id text NOT NULL, + expires_at timestamptz NOT NULL, + PRIMARY KEY (tenant_id, key) +); +CREATE INDEX idempotency_keys_expiry_idx ON webhook.idempotency_keys (expires_at); + +CREATE TABLE webhook.usage_counters ( + tenant_id text NOT NULL, + month text NOT NULL, + accepted bigint NOT NULL DEFAULT 0, + PRIMARY KEY (tenant_id, month) +); diff --git a/docs/decisions.md b/docs/decisions.md index 04895c5..5387261 100644 --- a/docs/decisions.md +++ b/docs/decisions.md @@ -55,3 +55,4 @@ Format: ID, date, decision, why. Add new entries at the bottom; do not rewrite h - **D51 (2026-10-21) Validate demand before production spend; the product category is real but the independent market is small, so we need a wedge.** Owner asked whether this can be a business (web research 2026-10-21). Findings: the category exists with paying customers (Svix lists Brex, Lob, Lightspeed and Clerk; Hookdeck has a public pricing ladder from a free 10,000 events to $39 and $499 plans), but the independent vendors are small: Svix has raised roughly $2.6M to $13M depending on the source with an estimated 5 to 11 staff and revenue estimates around $1M, Hookdeck about $2.7M raised and 1 to 10 staff (all from low-quality directories, no company-reported revenue). No credible independent market size exists (one report-selling site claims $1.8B growing to $10.2B, not corroborated). The space has free tiers (Hookdeck), open-source and self-hosted options (Svix, Hookdeck Outpost, Hook0), and many companies simply build it themselves or use AWS services. Conclusion: viable as a small, profitable niche business if cost and focus are right, unlikely to be a large market, and not proven for us until a real customer pays. Decisions: (1) Phase L gets demand-validation items L-L12 and L-L13; (2) keep spending near zero (local stack and a staging environment only) until at least one design partner commits, because the production floor is about $200-250 per month (cost-model.md, unverified); (3) pick a wedge from customer conversations; our structural advantage is cost per event, not features. Sources are in the PR description. - **D52 (2026-10-21) The project becomes open source (Apache-2.0) and self-hostable with Docker Compose; the hosted paid service is postponed.** Owner changed course after the market review (D51): publish the repo so anyone can use it, goal adoption and reputation, public after a clean-up and a working quick start. Why Apache-2.0: it is the license companies are most comfortable adopting (permissive, patent grant), which matters more for adoption than protecting against cloud vendors; the trade-off is that someone else may sell it as a service, which the owner accepted by choosing adoption over a protected hosted business (AGPL was the alternative). Consequences: Phase O is now the main track; Phase L (company registration, AWS account, payments, legal pages for a paid service) and S-9/S-10 (checkout, self-service signup) are postponed; the developer docs and sample code (S-1..S-5, S-7) move into the public launch; costs drop to zero for the owner. Honest note recorded: open source lowers the barrier to try the product but does not remove the need for a reason to choose it, so the cost per event and one-command simplicity are the wedge (D51). - **D53 (2026-10-21) Self-host mode uses PostgreSQL for queues, delivery state, rollups and the event archive; Valkey is optional; AWS adapters stay as an alternative backend.** Why: a self-hoster wants one `docker compose up` with as few moving parts as possible. Today's local stack uses Moto, an AWS test emulator that is not durable or meant for traffic, so it is demo-only. Postgres is already required for configuration (D21), supports a durable queue with `FOR UPDATE SKIP LOCKED` at a few thousand messages per second on one instance, and gives backup/restore for everything in one place. Valkey stays optional for shared rate limits (without it Ingest uses in-process limits, which is exact for one Ingest instance). The event archive defaults to Postgres (compressed batches with a retention job) and can use any S3-compatible store; MinIO is not bundled because of its license and maintenance status. KMS is replaced by a master key from an environment variable or file (envelope encryption, rotation documented); KMS stays available. Trade-offs accepted: a Postgres queue will not reach the 30,000 events/s Stage 2 design (that remains the AWS backend's job, `architecture.md` section 10), and the self-host guide states the supported throughput honestly. Port interfaces in `src/Abstractions` (D37) make this an additional adapter project (`src/Postgres` grows the queue/state/archive adapters), not a rewrite. +- **D54 (2026-10-21) Self-host backend implemented: `Backend:Mode=Postgres` (migration `0003_selfhost_runtime.sql`, `src/Postgres/Backend`).** Queue: one table `queue_messages` with SQS-like behavior (delay, visibility timeout, five receives then `-dlq`), receiving with `UPDATE ... WHERE id IN (SELECT ... FOR UPDATE SKIP LOCKED)` so workers never block or double-receive; waiting is a 200 ms poll up to the long-poll time (simple and reliable; `LISTEN/NOTIFY` is a later optimization if latency matters). Why a lease with a receipt UUID: a worker that finished late cannot delete a message that already reappeared and was leased by someone else. State, rollups and archive mirror the DynamoDB/S3 layouts (same column names, same gzip JSONL format, pointer `pg://archive/#`) so documentation and the Athena-style tooling concepts carry over. Idempotency keys and the monthly count live in PostgreSQL by default in this mode (durable across restarts and exact with several Ingest instances) instead of memory; Valkey is therefore optional and only speeds up the rate limit, which stays per-process without it (exact for a single Ingest instance). A maintenance job deletes expired rows hourly (state, archive, rollup minutes, idempotency) and dead-letter messages after 14 days. Retention values are options (`Backend:Postgres:*`). Known limits, stated honestly in the self-host docs: one PostgreSQL instance bounds throughput (target a few thousand events per second; the AWS backend is the path to tens of thousands), the archive grows the database (30-day default retention), and queue polling adds up to ~200 ms latency when idle. diff --git a/docs/flow-walkthrough.md b/docs/flow-walkthrough.md index dea94b9..a0571a0 100644 --- a/docs/flow-walkthrough.md +++ b/docs/flow-walkthrough.md @@ -4,6 +4,10 @@ Status: **living document**, kept in step with the code (rule in `CLAUDE.md`, it Last verified against the code: **2026-10-21** (after the ControlPlane API). The whole path is proven by one test, `tests/IntegrationTests/Aws/PipelineEndToEndTests.cs`: tenant, key and endpoint in PostgreSQL, event posted to Ingest, SQS, Fan-out with S3 archive and routing from the config database, delivery queue, Dispatcher loading the endpoint and its decrypted secret. It uses the local stack (`./scripts/dev-up.sh`); the final HTTP call is a recording sender because loopback targets are blocked by design. +## Backends + +The same services run on three backends, selected by `Backend:Mode` (`docs/configuration.md`): `InMemory` (tests, nothing durable), **`Postgres`** (self-host: queues, delivery state, rollups, archive, idempotency and the monthly count all in PostgreSQL, D53/D54; no AWS) and `Aws` (SQS, DynamoDB, S3, KMS; the scale path). The step table below names the AWS adapter files; the PostgreSQL equivalents are in `src/Postgres/Backend`. + ## The path of one event ``` diff --git a/docs/roadmap.md b/docs/roadmap.md index c51e17a..4571b36 100644 --- a/docs/roadmap.md +++ b/docs/roadmap.md @@ -62,9 +62,9 @@ Owner decision: make the repo public (Apache-2.0), goal adoption and reputation, | ID | Work item | Size | Notes | Depends | |---|---|---|---|---| | O-1 | **DONE 2026-10-21** (history scanned: no secrets; remaining owner choices: public contact email, GitHub private vulnerability reporting on) **Public-repo hygiene:** `LICENSE` (Apache-2.0), `NOTICE`, scan the full git history for secrets/personal data, check dependency licenses, `README.md` (what it is, features, architecture picture, quick start, status), `CONTRIBUTING.md`, `CODE_OF_CONDUCT.md`, `SECURITY.md` (how to report), issue and PR templates, repo description and topics | M | do first; blocks going public | | -| O-2 | **PostgreSQL queue adapters** (`Backend:Mode=Postgres`): durable queue table with visibility timeout, delay, receive count and dead-letter, `SELECT ... FOR UPDATE SKIP LOCKED`; implements the event and delivery publishers/sources and the retry publisher | L | replaces SQS for self-host; target a few thousand messages per second on one Postgres | | -| O-3 | **PostgreSQL state adapters:** delivery state + due index, rollups (upsert add), idempotency/quota when Valkey is absent | M | replaces DynamoDB | O-2 | -| O-4 | **Archive adapter without S3:** Postgres (compressed batches) by default with a retention job, S3-compatible storage (any endpoint) optional | M | replaces S3 for self-host; the existing S3 adapter works with any S3-compatible service | | +| O-2 | **DONE 2026-10-21 (D54)** **PostgreSQL queue adapters** (`Backend:Mode=Postgres`): durable queue table with visibility timeout, delay, receive count and dead-letter, `SELECT ... FOR UPDATE SKIP LOCKED`; implements the event and delivery publishers/sources and the retry publisher | L | replaces SQS for self-host; target a few thousand messages per second on one Postgres | | +| O-3 | **DONE 2026-10-21 (D54)** **PostgreSQL state adapters:** delivery state + due index, rollups (upsert add), idempotency/quota when Valkey is absent | M | replaces DynamoDB | O-2 | +| O-4 | **DONE for Postgres archive 2026-10-21 (D54); S3-compatible option exists via the S3 adapter** **Archive adapter without S3:** Postgres (compressed batches) by default with a retention job, S3-compatible storage (any endpoint) optional | M | replaces S3 for self-host; the existing S3 adapter works with any S3-compatible service | | | O-5 | **Secrets without KMS:** a production-grade key provider from an environment variable or mounted file (envelope encryption with a master key), documented key rotation; KMS stays optional | S | D45 keeps KMS for AWS | | | O-6 | **One-command stack:** `docker compose up` brings up migrator, Ingest, Fan-out, Dispatcher, ControlPlane, PostgreSQL (+ optional Valkey) with generated secrets and healthchecks; prebuilt images published to GitHub Container Registry from CI | M | | O-2..O-5 | | O-7 | **Quick start that works in 5 minutes:** create a tenant and key, register an endpoint, send an event, see it arrive (a bundled demo receiver container that verifies the signature and prints events), plus a smoke test script run in CI | M | | O-6 | diff --git a/docs/worklog.md b/docs/worklog.md index 9f93436..a1b5035 100644 --- a/docs/worklog.md +++ b/docs/worklog.md @@ -36,6 +36,11 @@ Resume point for agents after a sudden stop, crash, or new session. **Read this - **Outcome:** `architecture.md` v4 approved; decisions D20-D28 logged; `contracts.md` v2; `database.md` rewritten config-only (still **in review**); `docs/roadmap.md` lists the work items; plan numbers still provisional. - **Resume from:** see the next entry (roadmap execution). +### 2026-10-21 Postgres self-host backend (O-2..O-4) +- **Status:** PR open +- **Outcome:** `Backend:Mode=Postgres` with queue, delivery state + due index, rollups, archive, idempotency, quota, maintenance job; hosts wired; tests on real PostgreSQL including the full pipeline without AWS. D54. +- **Next:** O-5 master key from env/file, network option for private-address delivery (self-hosters call internal services), compose stack + demo receiver + quickstart script (O-6, O-7), README full guide + sample code (O-8, O-11). + ### 2026-10-21 Open source and self-host (Phase O) - current main track - **Status:** in progress - **Why:** owner decided to publish the repo (Apache-2.0) for adoption and reputation, running on Docker Compose; hosted business postponed (D52, D53). Owner answers: goal adoption, Apache-2.0, real self-host backends, public after clean-up + quick start. diff --git a/src/Dispatcher/Program.cs b/src/Dispatcher/Program.cs index 628ffd0..0bd0dc7 100644 --- a/src/Dispatcher/Program.cs +++ b/src/Dispatcher/Program.cs @@ -8,6 +8,7 @@ using WebhookGateway.Dispatcher.Source; using WebhookGateway.Hosting; using WebhookGateway.Postgres; +using WebhookGateway.Postgres.Backend; var builder = Host.CreateApplicationBuilder(args); builder.Logging.AddStructuredJsonConsole("Dispatcher"); @@ -36,12 +37,20 @@ // The consumer needs the queues, the archive and delivery_state, so it only runs with Backend:Mode=Aws // (Moto locally, see dev/README.md). The retry scheduler and the rollup flusher run next to it. -if (builder.Configuration.UseAws()) +if (builder.Configuration.UseAws() || builder.Configuration.UsePostgresBackend()) { builder.Services.Configure(builder.Configuration.GetSection(DeliveryOptions.SectionName)); builder.Services.Configure(builder.Configuration.GetSection(DeliveryConsumerOptions.SectionName)); - builder.Services.AddAwsDispatcher(builder.Configuration); - builder.Services.AddAwsRollups(builder.Configuration); + if (builder.Configuration.UseAws()) + { + builder.Services.AddAwsDispatcher(builder.Configuration); + builder.Services.AddAwsRollups(builder.Configuration); + } + else + { + builder.Services.AddPostgresDispatcher(builder.Configuration); + } + builder.Services.Configure(builder.Configuration.GetSection(RollupFlusherOptions.SectionName)); builder.Services.AddHostedService(); builder.Services.AddSingleton(); @@ -53,5 +62,5 @@ var host = builder.Build(); host.Services.GetRequiredService().CreateLogger("Startup") - .LogServiceStarted(builder.Configuration.UseAws() ? "Aws" : "InMemory (no queue consumer)"); + .LogServiceStarted(builder.Configuration.UseAws() ? "Aws" : builder.Configuration.UsePostgresBackend() ? "Postgres" : "InMemory (no queue consumer)"); host.Run(); diff --git a/src/FanOut/Program.cs b/src/FanOut/Program.cs index 37c5359..e8d65cf 100644 --- a/src/FanOut/Program.cs +++ b/src/FanOut/Program.cs @@ -8,6 +8,7 @@ using WebhookGateway.FanOut.Source; using WebhookGateway.Hosting; using WebhookGateway.Postgres; +using WebhookGateway.Postgres.Backend; var builder = Host.CreateApplicationBuilder(args); builder.Logging.AddStructuredJsonConsole("FanOut"); @@ -37,7 +38,14 @@ builder.Services.AddHostedService(); } +if (builder.Configuration.UsePostgresBackend()) +{ + builder.Services.AddPostgresFanOut(builder.Configuration); + builder.Services.Configure(builder.Configuration.GetSection(RollupFlusherOptions.SectionName)); + builder.Services.AddHostedService(); +} + var host = builder.Build(); host.Services.GetRequiredService().CreateLogger("Startup") - .LogServiceStarted(builder.Configuration.UseAws() ? "Aws" : "InMemory"); + .LogServiceStarted(builder.Configuration.UseAws() ? "Aws" : builder.Configuration.UsePostgresBackend() ? "Postgres" : "InMemory"); host.Run(); diff --git a/src/Ingest/Program.cs b/src/Ingest/Program.cs index 6dee27a..753a961 100644 --- a/src/Ingest/Program.cs +++ b/src/Ingest/Program.cs @@ -5,6 +5,7 @@ using WebhookGateway.Ingest.Publishing; using WebhookGateway.Ingest.Tenancy; using WebhookGateway.Postgres; +using WebhookGateway.Postgres.Backend; using WebhookGateway.Valkey; var builder = WebApplication.CreateBuilder(args); @@ -33,6 +34,12 @@ builder.Services.AddSingleton(); } +// Backend:Mode=Postgres: events go to a PostgreSQL queue, idempotency keys and the monthly count live in PostgreSQL (D53). +if (builder.Configuration.UsePostgresBackend()) +{ + builder.Services.AddPostgresIngest(builder.Configuration); +} + // With Valkey:ConnectionString set, limits are shared across Ingest tasks; each falls back to local behavior if Valkey is down. if (builder.Configuration.UseValkey()) { @@ -54,5 +61,5 @@ app.MapGet("/healthz", () => Results.Ok(new { status = "ok", service = "Ingest" })); EventsEndpoint.Map(app); -app.Logger.LogServiceStarted(builder.Configuration.UseAws() ? "Aws" : "InMemory"); +app.Logger.LogServiceStarted(builder.Configuration.UseAws() ? "Aws" : builder.Configuration.UsePostgresBackend() ? "Postgres" : "InMemory"); app.Run(); diff --git a/src/Postgres/Backend/PostgresArchive.cs b/src/Postgres/Backend/PostgresArchive.cs new file mode 100644 index 0000000..8db7a8f --- /dev/null +++ b/src/Postgres/Backend/PostgresArchive.cs @@ -0,0 +1,119 @@ +using System.IO.Compression; +using System.Text; +using System.Text.Json; +using Microsoft.Extensions.Logging; +using Microsoft.Extensions.Options; +using Npgsql; +using WebhookGateway.Contracts; +using WebhookGateway.Contracts.Ids; +using WebhookGateway.Contracts.Messages; +using WebhookGateway.Dispatcher.Delivery; +using WebhookGateway.FanOut.Archive; + +namespace WebhookGateway.Postgres.Backend; + +/// +/// Event archive in PostgreSQL (D53): one gzip JSON Lines batch per row, the same format as the S3 archive (D33). The pointer is +/// pg://archive/<key>#<line>. Rows expire after the retention period and are removed by the maintenance job. +/// +public sealed partial class PostgresEventArchive( + NpgsqlDataSource db, + TimeProvider clock, + IOptions options, + ILogger logger) : IEventArchive +{ + public const string Scheme = "pg://archive/"; + private readonly string _workerId = Guid.NewGuid().ToString("N")[..12]; + private long _sequence; + + public async ValueTask> WriteBatchAsync(IReadOnlyList events, CancellationToken cancellationToken) + { + ArgumentNullException.ThrowIfNull(events); + var now = clock.GetUtcNow(); + var key = $"{now:yyyy-MM-dd}/{now:HH}/{_workerId}-{Interlocked.Increment(ref _sequence)}"; + var refs = new Dictionary(events.Count); + using var compressed = new MemoryStream(); + await using (var gzip = new GZipStream(compressed, CompressionLevel.Fastest, leaveOpen: true)) + await using (var writer = new StreamWriter(gzip, new UTF8Encoding(false))) + { + for (var line = 0; line < events.Count; line++) + { + await writer.WriteLineAsync(ContractJson.Serialize(events[line])).ConfigureAwait(false); + refs[events[line].EventId] = $"{Scheme}{key}#{line}"; + } + } + + await using var command = db.CreateCommand("INSERT INTO webhook.archive_batches (key, body, event_count, expires_at) VALUES (@key, @body, @count, @expires)"); + command.Parameters.AddWithValue("key", key); + command.Parameters.AddWithValue("body", compressed.ToArray()); + command.Parameters.AddWithValue("count", events.Count); + command.Parameters.AddWithValue("expires", now.Add(options.Value.ArchiveRetention)); + await command.ExecuteNonQueryAsync(cancellationToken).ConfigureAwait(false); + LogWritten(logger, events.Count, compressed.Length, key); + return refs; + } + + [LoggerMessage(Level = LogLevel.Information, Message = "Archived {EventCount} events ({Bytes} bytes compressed) as batch {Key}")] + private static partial void LogWritten(ILogger logger, int eventCount, long bytes, string key); +} + +/// Inline payload when present, otherwise the event is read back from the archive batch named by PayloadRef. +public sealed partial class PostgresPayloadResolver(NpgsqlDataSource db, ILogger logger) : IPayloadResolver +{ + public async ValueTask ResolveAsync(DeliveryMessage message, CancellationToken cancellationToken) + { + ArgumentNullException.ThrowIfNull(message); + if (message.Payload is not null) + { + return message.Payload; + } + + var reference = message.PayloadRef; + var hash = reference?.LastIndexOf('#') ?? -1; + if (reference is null || !reference.StartsWith(PostgresEventArchive.Scheme, StringComparison.Ordinal) || hash < 0 + || !int.TryParse(reference[(hash + 1)..], out var line)) + { + LogBadRef(logger, message.DeliveryId.Value); + return null; + } + + var key = reference[PostgresEventArchive.Scheme.Length..hash]; + await using var command = db.CreateCommand("SELECT body FROM webhook.archive_batches WHERE key = @key"); + command.Parameters.AddWithValue("key", key); + if (await command.ExecuteScalarAsync(cancellationToken).ConfigureAwait(false) is not byte[] body) + { + LogMissing(logger, message.DeliveryId.Value, key, line); + return null; + } + + using var stream = new MemoryStream(body); + await using var gzip = new GZipStream(stream, CompressionMode.Decompress); + using var reader = new StreamReader(gzip); + for (var i = 0; await reader.ReadLineAsync(cancellationToken).ConfigureAwait(false) is { } text; i++) + { + if (i == line) + { + var stored = ContractJson.Deserialize(text); + if (stored.EventId == message.EventId) + { + return stored.Payload; + } + + LogMismatch(logger, message.DeliveryId.Value, message.EventId.Value); + return null; + } + } + + LogMissing(logger, message.DeliveryId.Value, key, line); + return null; + } + + [LoggerMessage(Level = LogLevel.Error, Message = "Delivery {DeliveryId} has an unreadable payload reference")] + private static partial void LogBadRef(ILogger logger, string deliveryId); + + [LoggerMessage(Level = LogLevel.Error, Message = "Delivery {DeliveryId}: archive batch {Key} line {Line} not found")] + private static partial void LogMissing(ILogger logger, string deliveryId, string key, int line); + + [LoggerMessage(Level = LogLevel.Error, Message = "Delivery {DeliveryId}: archive line does not belong to event {EventId}")] + private static partial void LogMismatch(ILogger logger, string deliveryId, string eventId); +} diff --git a/src/Postgres/Backend/PostgresBackendExtensions.cs b/src/Postgres/Backend/PostgresBackendExtensions.cs new file mode 100644 index 0000000..8d877e6 --- /dev/null +++ b/src/Postgres/Backend/PostgresBackendExtensions.cs @@ -0,0 +1,68 @@ +using Microsoft.Extensions.Configuration; +using Microsoft.Extensions.DependencyInjection; +using Microsoft.Extensions.DependencyInjection.Extensions; +using WebhookGateway.Core.Rollups; +using WebhookGateway.Dispatcher.Delivery; +using WebhookGateway.Dispatcher.Source; +using WebhookGateway.FanOut.Archive; +using WebhookGateway.FanOut.Delivery; +using WebhookGateway.FanOut.Source; +using WebhookGateway.Ingest.Limits; +using WebhookGateway.Ingest.Publishing; + +namespace WebhookGateway.Postgres.Backend; + +/// +/// Backend:Mode=Postgres (D53): queues, delivery state, rollups, archive, idempotency and quota all live in the +/// config database, so the whole product runs with PostgreSQL alone. Each role registers only what it uses. +/// +public static class PostgresBackendExtensions +{ + public static bool UsePostgresBackend(this IConfiguration configuration) + { + ArgumentNullException.ThrowIfNull(configuration); + return string.Equals(configuration["Backend:Mode"], "Postgres", StringComparison.OrdinalIgnoreCase); + } + + public static IServiceCollection AddPostgresIngest(this IServiceCollection services, IConfiguration configuration) + { + services.AddPostgresShared(configuration); + services.Replace(ServiceDescriptor.Singleton()); + services.Replace(ServiceDescriptor.Singleton()); + services.Replace(ServiceDescriptor.Singleton()); + return services; + } + + public static IServiceCollection AddPostgresFanOut(this IServiceCollection services, IConfiguration configuration) + { + services.AddPostgresShared(configuration); + services.TryAddSingleton(); + services.Replace(ServiceDescriptor.Singleton()); + services.Replace(ServiceDescriptor.Singleton(sp => sp.GetRequiredService())); + services.Replace(ServiceDescriptor.Singleton()); + services.Replace(ServiceDescriptor.Singleton()); + return services; + } + + public static IServiceCollection AddPostgresDispatcher(this IServiceCollection services, IConfiguration configuration) + { + services.AddPostgresShared(configuration); + services.TryAddSingleton(); + services.TryAddSingleton(); + services.Replace(ServiceDescriptor.Singleton()); + services.Replace(ServiceDescriptor.Singleton(sp => sp.GetRequiredService())); + services.Replace(ServiceDescriptor.Singleton()); + services.Replace(ServiceDescriptor.Singleton(sp => sp.GetRequiredService())); + services.Replace(ServiceDescriptor.Singleton(sp => sp.GetRequiredService())); + services.Replace(ServiceDescriptor.Singleton()); + services.AddHostedService(); + return services; + } + + private static void AddPostgresShared(this IServiceCollection services, IConfiguration configuration) + { + services.AddPostgresDataSource(configuration); + services.Configure(configuration.GetSection(BackendOptions.SectionName)); + services.TryAddSingleton(); + } +} diff --git a/src/Postgres/Backend/PostgresDeliveryStateStore.cs b/src/Postgres/Backend/PostgresDeliveryStateStore.cs new file mode 100644 index 0000000..b9c7b56 --- /dev/null +++ b/src/Postgres/Backend/PostgresDeliveryStateStore.cs @@ -0,0 +1,128 @@ +using System.Globalization; +using System.Text.Json; +using Microsoft.Extensions.Logging; +using Microsoft.Extensions.Options; +using Npgsql; +using WebhookGateway.Contracts; +using WebhookGateway.Contracts.Ids; +using WebhookGateway.Dispatcher.Delivery; + +namespace WebhookGateway.Postgres.Backend; + +public sealed class BackendOptions +{ + public const string SectionName = "Backend:Postgres"; + + /// How long a delivery_state row lives (D34/D53: plan retention arrives with the plan lookup). + public TimeSpan DeliveryStateRetention { get; init; } = TimeSpan.FromDays(30); + + public TimeSpan RollupMinuteRetention { get; init; } = TimeSpan.FromDays(7); + + public TimeSpan ArchiveRetention { get; init; } = TimeSpan.FromDays(30); +} + +/// webhook.delivery_state: only non-success deliveries, plus the due index used by the retry scheduler. +public sealed partial class PostgresDeliveryStateStore( + NpgsqlDataSource db, + TimeProvider clock, + IOptions options, + ILogger logger) : IDeliveryStateStore, IDueDeliveryIndex +{ + private const string Columns = + "tenant_id, delivery_id, endpoint_id, event_id, event_type, occurred_at, status, attempt, tier, next_attempt_at, last_status_code, last_error, payload, payload_ref, created_at, updated_at"; + + private readonly BackendOptions _options = options.Value; + + public async ValueTask GetAsync(TenantId tenantId, DeliveryId deliveryId, CancellationToken cancellationToken) + { + await using var command = db.CreateCommand($"SELECT {Columns} FROM webhook.delivery_state WHERE tenant_id = @t AND delivery_id = @d"); + command.Parameters.AddWithValue("t", tenantId.Value); + command.Parameters.AddWithValue("d", deliveryId.Value); + await using var reader = await command.ExecuteReaderAsync(cancellationToken).ConfigureAwait(false); + return await reader.ReadAsync(cancellationToken).ConfigureAwait(false) ? Read(reader) : null; + } + + public async ValueTask PutAsync(DeliveryStateRecord record, CancellationToken cancellationToken) + { + ArgumentNullException.ThrowIfNull(record); + await using var command = db.CreateCommand( + $""" + INSERT INTO webhook.delivery_state ({Columns}, expires_at) + VALUES (@tenant, @delivery, @endpoint, @event, @type, @occurred, @status, @attempt, @tier, @next, @code, @error, @payload, @ref, @created, @updated, @expires) + ON CONFLICT (tenant_id, delivery_id) DO UPDATE SET + status = EXCLUDED.status, attempt = EXCLUDED.attempt, tier = EXCLUDED.tier, next_attempt_at = EXCLUDED.next_attempt_at, + last_status_code = EXCLUDED.last_status_code, last_error = EXCLUDED.last_error, payload = EXCLUDED.payload, + payload_ref = EXCLUDED.payload_ref, updated_at = EXCLUDED.updated_at, expires_at = EXCLUDED.expires_at + """); + var p = command.Parameters; + p.AddWithValue("tenant", record.TenantId.Value); + p.AddWithValue("delivery", record.DeliveryId.Value); + p.AddWithValue("endpoint", record.EndpointId.Value); + p.AddWithValue("event", record.EventId.Value); + p.AddWithValue("type", record.EventType); + p.AddWithValue("occurred", record.OccurredAt); + p.AddWithValue("status", record.Status.ToString()); + p.AddWithValue("attempt", record.Attempt); + p.AddWithValue("tier", record.Tier.ToString()); + p.AddWithValue("next", (object?)record.NextAttemptAt ?? DBNull.Value); + p.AddWithValue("code", (object?)record.LastStatusCode ?? DBNull.Value); + p.AddWithValue("error", (object?)record.LastError ?? DBNull.Value); + p.AddWithValue("payload", record.Payload is { } payload ? payload.GetRawText() : DBNull.Value); + p.AddWithValue("ref", (object?)record.PayloadRef ?? DBNull.Value); + p.AddWithValue("created", record.CreatedAt); + p.AddWithValue("updated", record.UpdatedAt); + p.AddWithValue("expires", clock.GetUtcNow().Add(_options.DeliveryStateRetention)); + await command.ExecuteNonQueryAsync(cancellationToken).ConfigureAwait(false); + LogStored(logger, record.TenantId.Value, record.DeliveryId.Value, record.Status, record.Attempt); + } + + public async ValueTask DeleteAsync(TenantId tenantId, DeliveryId deliveryId, CancellationToken cancellationToken) + { + await using var command = db.CreateCommand("DELETE FROM webhook.delivery_state WHERE tenant_id = @t AND delivery_id = @d"); + command.Parameters.AddWithValue("t", tenantId.Value); + command.Parameters.AddWithValue("d", deliveryId.Value); + await command.ExecuteNonQueryAsync(cancellationToken).ConfigureAwait(false); + LogDeleted(logger, tenantId.Value, deliveryId.Value); + } + + public async ValueTask> GetDueAsync(DateTimeOffset now, int limit, CancellationToken cancellationToken) + { + await using var command = db.CreateCommand( + $"SELECT {Columns} FROM webhook.delivery_state WHERE status = 'Scheduled' AND next_attempt_at <= @now ORDER BY next_attempt_at LIMIT @limit"); + command.Parameters.AddWithValue("now", now); + command.Parameters.AddWithValue("limit", limit); + var due = new List(); + await using var reader = await command.ExecuteReaderAsync(cancellationToken).ConfigureAwait(false); + while (await reader.ReadAsync(cancellationToken).ConfigureAwait(false)) + { + due.Add(Read(reader)); + } + + return due; + } + + private static DeliveryStateRecord Read(NpgsqlDataReader r) => + new( + TenantId.Parse(r.GetString(0)), + DeliveryId.Parse(r.GetString(1)), + EndpointId.Parse(r.GetString(2)), + EventId.Parse(r.GetString(3)), + r.GetString(4), + r.GetFieldValue(5), + Enum.Parse(r.GetString(6)), + r.GetInt32(7), + r.IsDBNull(9) ? null : r.GetFieldValue(9), + r.IsDBNull(10) ? null : r.GetInt32(10), + r.IsDBNull(11) ? null : r.GetString(11), + r.IsDBNull(12) ? null : JsonDocument.Parse(r.GetString(12)).RootElement.Clone(), + r.IsDBNull(13) ? null : r.GetString(13), + r.GetFieldValue(14), + r.GetFieldValue(15), + Enum.Parse(r.GetString(8))); + + [LoggerMessage(Level = LogLevel.Debug, Message = "delivery_state stored {DeliveryId} of tenant {TenantId}: {Status}, attempt {Attempt}")] + private static partial void LogStored(ILogger logger, string tenantId, string deliveryId, DeliveryStatus status, int attempt); + + [LoggerMessage(Level = LogLevel.Debug, Message = "delivery_state deleted {DeliveryId} of tenant {TenantId}")] + private static partial void LogDeleted(ILogger logger, string tenantId, string deliveryId); +} diff --git a/src/Postgres/Backend/PostgresLimits.cs b/src/Postgres/Backend/PostgresLimits.cs new file mode 100644 index 0000000..5782f70 --- /dev/null +++ b/src/Postgres/Backend/PostgresLimits.cs @@ -0,0 +1,71 @@ +using System.Globalization; +using Npgsql; +using WebhookGateway.Contracts.Ids; +using WebhookGateway.Ingest.Limits; + +namespace WebhookGateway.Postgres.Backend; + +/// +/// Idempotency keys in PostgreSQL: durable across restarts and shared by every Ingest instance without Valkey. +/// One statement claims a free or expired key; otherwise the existing event id is returned. +/// +public sealed class PostgresIdempotencyStore(NpgsqlDataSource db) : IIdempotencyStore +{ + public async ValueTask TryClaimAsync(TenantId tenantId, string key, EventId eventId, TimeSpan window, CancellationToken cancellationToken) + { + await using var claim = db.CreateCommand( + """ + INSERT INTO webhook.idempotency_keys (tenant_id, key, event_id, expires_at) + VALUES (@t, @k, @e, now() + make_interval(secs => @window)) + ON CONFLICT (tenant_id, key) DO UPDATE SET event_id = EXCLUDED.event_id, expires_at = EXCLUDED.expires_at + WHERE webhook.idempotency_keys.expires_at <= now() + RETURNING event_id + """); + claim.Parameters.AddWithValue("t", tenantId.Value); + claim.Parameters.AddWithValue("k", key); + claim.Parameters.AddWithValue("e", eventId.Value); + claim.Parameters.AddWithValue("window", window.TotalSeconds); + if (await claim.ExecuteScalarAsync(cancellationToken).ConfigureAwait(false) is string) + { + return new IdempotencyClaim(true, null); + } + + await using var read = db.CreateCommand("SELECT event_id FROM webhook.idempotency_keys WHERE tenant_id = @t AND key = @k"); + read.Parameters.AddWithValue("t", tenantId.Value); + read.Parameters.AddWithValue("k", key); + return await read.ExecuteScalarAsync(cancellationToken).ConfigureAwait(false) is string existing + ? new IdempotencyClaim(false, EventId.Parse(existing)) + : new IdempotencyClaim(true, null); + } + + public async ValueTask ReleaseAsync(TenantId tenantId, string key, CancellationToken cancellationToken) + { + await using var command = db.CreateCommand("DELETE FROM webhook.idempotency_keys WHERE tenant_id = @t AND key = @k"); + command.Parameters.AddWithValue("t", tenantId.Value); + command.Parameters.AddWithValue("k", key); + await command.ExecuteNonQueryAsync(cancellationToken).ConfigureAwait(false); + } +} + +/// Monthly accepted-event counter per tenant in PostgreSQL (durable, shared). +public sealed class PostgresQuotaCounter(NpgsqlDataSource db) : IQuotaCounter +{ + public async ValueTask IncrementAsync(TenantId tenantId, DateTimeOffset now, CancellationToken cancellationToken) + { + await using var command = db.CreateCommand( + "INSERT INTO webhook.usage_counters (tenant_id, month, accepted) VALUES (@t, @m, 1) ON CONFLICT (tenant_id, month) DO UPDATE SET accepted = webhook.usage_counters.accepted + 1 RETURNING accepted"); + command.Parameters.AddWithValue("t", tenantId.Value); + command.Parameters.AddWithValue("m", Month(now)); + return (long)(await command.ExecuteScalarAsync(cancellationToken).ConfigureAwait(false))!; + } + + public async ValueTask DecrementAsync(TenantId tenantId, DateTimeOffset now, CancellationToken cancellationToken) + { + await using var command = db.CreateCommand("UPDATE webhook.usage_counters SET accepted = GREATEST(0, accepted - 1) WHERE tenant_id = @t AND month = @m"); + command.Parameters.AddWithValue("t", tenantId.Value); + command.Parameters.AddWithValue("m", Month(now)); + await command.ExecuteNonQueryAsync(cancellationToken).ConfigureAwait(false); + } + + private static string Month(DateTimeOffset now) => now.UtcDateTime.ToString("yyyyMM", CultureInfo.InvariantCulture); +} diff --git a/src/Postgres/Backend/PostgresMaintenance.cs b/src/Postgres/Backend/PostgresMaintenance.cs new file mode 100644 index 0000000..c87d458 --- /dev/null +++ b/src/Postgres/Backend/PostgresMaintenance.cs @@ -0,0 +1,76 @@ +using Microsoft.Extensions.Hosting; +using Microsoft.Extensions.Logging; +using Npgsql; + +namespace WebhookGateway.Postgres.Backend; + +/// +/// Deletes expired rows (delivery state, archive batches, rollup minutes, idempotency keys) and old dead-lettered messages once an +/// hour. Safe to run in several instances: every statement is an idempotent delete. +/// +public sealed partial class PostgresMaintenance(NpgsqlDataSource db, ILogger logger) : BackgroundService +{ + private static readonly TimeSpan Interval = TimeSpan.FromHours(1); + + protected override async Task ExecuteAsync(CancellationToken stoppingToken) + { + using var timer = new PeriodicTimer(Interval); + do + { + try + { + await RunOnceAsync(stoppingToken).ConfigureAwait(false); + } + catch (Exception ex) when (ex is not OperationCanceledException) + { + LogFailed(logger, ex); + } + } + while (await WaitAsync(timer, stoppingToken).ConfigureAwait(false)); + } + + /// One cleanup pass. Exposed so tests can run it. + public async ValueTask RunOnceAsync(CancellationToken cancellationToken) + { + var total = 0; + foreach (var (table, column) in new[] + { + ("delivery_state", "expires_at"), + ("archive_batches", "expires_at"), + ("rollups", "expires_at"), + ("idempotency_keys", "expires_at"), + }) + { + await using var command = db.CreateCommand($"DELETE FROM webhook.{table} WHERE {column} <= now()"); + var deleted = await command.ExecuteNonQueryAsync(cancellationToken).ConfigureAwait(false); + if (deleted > 0) + { + LogDeleted(logger, deleted, table); + } + + total += deleted; + } + + await using var dlq = db.CreateCommand("DELETE FROM webhook.queue_messages WHERE queue LIKE '%-dlq' AND enqueued_at <= now() - interval '14 days'"); + total += await dlq.ExecuteNonQueryAsync(cancellationToken).ConfigureAwait(false); + return total; + } + + private static async Task WaitAsync(PeriodicTimer timer, CancellationToken cancellationToken) + { + try + { + return await timer.WaitForNextTickAsync(cancellationToken).ConfigureAwait(false); + } + catch (OperationCanceledException) + { + return false; + } + } + + [LoggerMessage(Level = LogLevel.Information, Message = "Maintenance removed {Count} expired rows from {Table}")] + private static partial void LogDeleted(ILogger logger, int count, string table); + + [LoggerMessage(Level = LogLevel.Error, Message = "Maintenance pass failed; will try again in an hour")] + private static partial void LogFailed(ILogger logger, Exception exception); +} diff --git a/src/Postgres/Backend/PostgresMessageStore.cs b/src/Postgres/Backend/PostgresMessageStore.cs new file mode 100644 index 0000000..55fd802 --- /dev/null +++ b/src/Postgres/Backend/PostgresMessageStore.cs @@ -0,0 +1,108 @@ +using Microsoft.Extensions.Logging; +using Npgsql; +using NpgsqlTypes; + +namespace WebhookGateway.Postgres.Backend; + +/// A message leased from a queue. proves the lease when deleting it. +public sealed record QueueMessage(long Id, string Receipt, string Body, int ReceiveCount); + +/// +/// A durable queue on one PostgreSQL table with the semantics the services rely on from SQS (D53): messages can be delayed, +/// a received message is invisible for the visibility timeout and reappears if it is not deleted, and after +/// receives it moves to the dead-letter queue <queue>-dlq. Receiving uses +/// FOR UPDATE SKIP LOCKED, so many workers can read the same queue without blocking each other. +/// +public sealed partial class PostgresMessageStore(NpgsqlDataSource db, ILogger logger) +{ + public const int MaxReceives = 5; + private static readonly TimeSpan PollInterval = TimeSpan.FromMilliseconds(200); + + public async ValueTask SendAsync(string queue, IReadOnlyList<(string Body, TimeSpan Delay)> messages, CancellationToken cancellationToken) + { + if (messages.Count == 0) + { + return; + } + + await using var command = db.CreateCommand( + "INSERT INTO webhook.queue_messages (queue, body, visible_at) SELECT @queue, b, now() + make_interval(secs => d) FROM unnest(@bodies, @delays) AS t(b, d)"); + command.Parameters.AddWithValue("queue", queue); + command.Parameters.Add(new NpgsqlParameter("bodies", NpgsqlDbType.Array | NpgsqlDbType.Text) { Value = messages.Select(m => m.Body).ToArray() }); + command.Parameters.Add(new NpgsqlParameter("delays", NpgsqlDbType.Array | NpgsqlDbType.Double) { Value = messages.Select(m => m.Delay.TotalSeconds).ToArray() }); + await command.ExecuteNonQueryAsync(cancellationToken).ConfigureAwait(false); + } + + /// Waits up to for messages (long poll) and leases up to of them. + public async ValueTask> ReceiveAsync( + string queue, + int maxCount, + TimeSpan visibilityTimeout, + TimeSpan wait, + CancellationToken cancellationToken) + { + var deadline = DateTimeOffset.UtcNow + wait; + while (true) + { + await DeadLetterExhaustedAsync(queue, cancellationToken).ConfigureAwait(false); + var messages = await LeaseAsync(queue, maxCount, visibilityTimeout, cancellationToken).ConfigureAwait(false); + if (messages.Count > 0 || DateTimeOffset.UtcNow >= deadline) + { + return messages; + } + + await Task.Delay(PollInterval, cancellationToken).ConfigureAwait(false); + } + } + + public async ValueTask DeleteAsync(QueueMessage message, CancellationToken cancellationToken) + { + ArgumentNullException.ThrowIfNull(message); + await using var command = db.CreateCommand("DELETE FROM webhook.queue_messages WHERE id = @id AND receipt = @receipt"); + command.Parameters.AddWithValue("id", message.Id); + command.Parameters.AddWithValue("receipt", Guid.Parse(message.Receipt)); + await command.ExecuteNonQueryAsync(cancellationToken).ConfigureAwait(false); + } + + private async ValueTask> LeaseAsync(string queue, int maxCount, TimeSpan visibilityTimeout, CancellationToken cancellationToken) + { + await using var command = db.CreateCommand( + """ + UPDATE webhook.queue_messages m + SET visible_at = now() + make_interval(secs => @visibility), receive_count = m.receive_count + 1, receipt = gen_random_uuid() + WHERE m.id IN ( + SELECT id FROM webhook.queue_messages + WHERE queue = @queue AND visible_at <= now() AND receive_count < @max + ORDER BY id LIMIT @limit FOR UPDATE SKIP LOCKED) + RETURNING m.id, m.receipt, m.body, m.receive_count + """); + command.Parameters.AddWithValue("queue", queue); + command.Parameters.AddWithValue("visibility", visibilityTimeout.TotalSeconds); + command.Parameters.AddWithValue("max", MaxReceives); + command.Parameters.AddWithValue("limit", maxCount); + var result = new List(); + await using var reader = await command.ExecuteReaderAsync(cancellationToken).ConfigureAwait(false); + while (await reader.ReadAsync(cancellationToken).ConfigureAwait(false)) + { + result.Add(new QueueMessage(reader.GetInt64(0), reader.GetGuid(1).ToString(), reader.GetString(2), reader.GetInt32(3))); + } + + return result; + } + + private async ValueTask DeadLetterExhaustedAsync(string queue, CancellationToken cancellationToken) + { + await using var command = db.CreateCommand( + "UPDATE webhook.queue_messages SET queue = queue || '-dlq', receive_count = 0, visible_at = now() WHERE queue = @queue AND visible_at <= now() AND receive_count >= @max"); + command.Parameters.AddWithValue("queue", queue); + command.Parameters.AddWithValue("max", MaxReceives); + var moved = await command.ExecuteNonQueryAsync(cancellationToken).ConfigureAwait(false); + if (moved > 0) + { + LogDeadLettered(logger, moved, queue); + } + } + + [LoggerMessage(Level = LogLevel.Error, Message = "{Count} messages from queue {Queue} failed 5 times and moved to the dead-letter queue")] + private static partial void LogDeadLettered(ILogger logger, int count, string queue); +} diff --git a/src/Postgres/Backend/PostgresQueueAdapters.cs b/src/Postgres/Backend/PostgresQueueAdapters.cs new file mode 100644 index 0000000..fccc8b9 --- /dev/null +++ b/src/Postgres/Backend/PostgresQueueAdapters.cs @@ -0,0 +1,134 @@ +using System.Text.Json; +using Microsoft.Extensions.Logging; +using WebhookGateway.Contracts; +using WebhookGateway.Contracts.Messages; +using WebhookGateway.Dispatcher.Delivery; +using WebhookGateway.Dispatcher.Source; +using WebhookGateway.FanOut.Delivery; +using WebhookGateway.FanOut.Source; +using WebhookGateway.Ingest.Publishing; + +namespace WebhookGateway.Postgres.Backend; + +/// Queue names are the same as with SQS (docs/data-plane.md). +public static class QueueNames +{ + public const string Events = "events"; + + public static string Deliveries(PlanTier tier) => $"deliveries-{tier.ToString().ToLowerInvariant()}"; +} + +public sealed class PostgresEventPublisher(PostgresMessageStore queue) : IEventPublisher +{ + public ValueTask PublishAsync(EventMessage message, CancellationToken cancellationToken) => + queue.SendAsync(QueueNames.Events, [(ContractJson.Serialize(message), TimeSpan.Zero)], cancellationToken); +} + +/// First attempts from Fan-out and retries from the Dispatcher, with the delay as queue visibility. +public sealed class PostgresDeliveryPublisher(PostgresMessageStore queue) : IDeliveryPublisher, IRetryPublisher +{ + public ValueTask PublishAsync(PlanTier tier, IReadOnlyList deliveries, CancellationToken cancellationToken) => + queue.SendAsync(QueueNames.Deliveries(tier), [.. deliveries.Select(d => (ContractJson.Serialize(d), TimeSpan.Zero))], cancellationToken); + + public ValueTask PublishAsync(PlanTier tier, DeliveryMessage nextAttempt, TimeSpan delay, CancellationToken cancellationToken) => + queue.SendAsync(QueueNames.Deliveries(tier), [(ContractJson.Serialize(nextAttempt), delay < TimeSpan.Zero ? TimeSpan.Zero : delay)], cancellationToken); +} + +public sealed partial class PostgresEventSource(PostgresMessageStore queue, ILogger logger, TimeSpan? wait = null) : IEventSource +{ + private static readonly TimeSpan Visibility = TimeSpan.FromSeconds(60); + private readonly Dictionary _leases = []; + + public async ValueTask> ReceiveAsync(int maxCount, CancellationToken cancellationToken) + { + var messages = await queue.ReceiveAsync(QueueNames.Events, Math.Clamp(maxCount, 1, 10), Visibility, wait ?? TimeSpan.FromSeconds(5), cancellationToken).ConfigureAwait(false); + var events = new List(); + foreach (var m in messages) + { + try + { + var receipt = $"{m.Id}:{m.Receipt}"; + events.Add(new ReceivedEvent(ContractJson.Deserialize(m.Body), receipt)); + lock (_leases) + { + _leases[receipt] = m; + } + } + catch (JsonException ex) + { + // Left leased: it reappears, and after five receives lands in the dead-letter queue. The body is never logged. + LogUnreadable(logger, ex, m.Id); + } + } + + return events; + } + + public async ValueTask AcknowledgeAsync(IReadOnlyList events, CancellationToken cancellationToken) + { + foreach (var e in events) + { + QueueMessage? lease; + lock (_leases) + { + _leases.Remove(e.Receipt, out lease); + } + + if (lease is not null) + { + await queue.DeleteAsync(lease, cancellationToken).ConfigureAwait(false); + } + } + } + + [LoggerMessage(Level = LogLevel.Error, Message = "Unreadable message {MessageId} left on the events queue for redelivery and the dead-letter queue")] + private static partial void LogUnreadable(ILogger logger, Exception exception, long messageId); +} + +public sealed partial class PostgresDeliverySource(PostgresMessageStore queue, ILogger logger, TimeSpan? wait = null) : IDeliverySource +{ + private static readonly TimeSpan Visibility = TimeSpan.FromSeconds(60); + private readonly Dictionary _leases = []; + + public async ValueTask> ReceiveAsync(PlanTier tier, int maxCount, CancellationToken cancellationToken) + { + var messages = await queue.ReceiveAsync(QueueNames.Deliveries(tier), Math.Clamp(maxCount, 1, 10), Visibility, wait ?? TimeSpan.FromSeconds(5), cancellationToken).ConfigureAwait(false); + var deliveries = new List(); + foreach (var m in messages) + { + try + { + var receipt = $"{m.Id}:{m.Receipt}"; + deliveries.Add(new ReceivedDelivery(ContractJson.Deserialize(m.Body), receipt)); + lock (_leases) + { + _leases[receipt] = m; + } + } + catch (JsonException ex) + { + LogUnreadable(logger, ex, m.Id, tier); + } + } + + return deliveries; + } + + public async ValueTask AcknowledgeAsync(PlanTier tier, ReceivedDelivery delivery, CancellationToken cancellationToken) + { + ArgumentNullException.ThrowIfNull(delivery); + QueueMessage? lease; + lock (_leases) + { + _leases.Remove(delivery.Receipt, out lease); + } + + if (lease is not null) + { + await queue.DeleteAsync(lease, cancellationToken).ConfigureAwait(false); + } + } + + [LoggerMessage(Level = LogLevel.Error, Message = "Unreadable message {MessageId} left on the {Tier} delivery queue for redelivery and the dead-letter queue")] + private static partial void LogUnreadable(ILogger logger, Exception exception, long messageId, PlanTier tier); +} diff --git a/src/Postgres/Backend/PostgresRollupWriter.cs b/src/Postgres/Backend/PostgresRollupWriter.cs new file mode 100644 index 0000000..d918c81 --- /dev/null +++ b/src/Postgres/Backend/PostgresRollupWriter.cs @@ -0,0 +1,101 @@ +using System.Globalization; +using System.Text; +using Microsoft.Extensions.Logging; +using Microsoft.Extensions.Options; +using Npgsql; +using WebhookGateway.Core.Rollups; + +namespace WebhookGateway.Postgres.Backend; + +/// +/// Adds drained counters to webhook.rollups with ON CONFLICT DO UPDATE col = col + excluded.col, so many +/// workers can add to the same minute row. Same row layout as the DynamoDB writer (D41); rows that fail are returned. +/// +public sealed partial class PostgresRollupWriter( + NpgsqlDataSource db, + TimeProvider clock, + IOptions options, + ILogger logger) : IRollupWriter +{ + private static readonly string[] CounterNames = + [ + "created", "attempts", "succeeded", "retried", "failed", "exhausted", "timeouts", "connection_ms", "rt_sum", "qw_sum", + .. Enumerable.Range(0, 13).Select(i => $"rt{i}"), + .. Enumerable.Range(0, 13).Select(i => $"qw{i}"), + ]; + + private static readonly string Sql = BuildSql(); + private readonly BackendOptions _options = options.Value; + + public async ValueTask> WriteAsync( + IReadOnlyDictionary rows, + CancellationToken cancellationToken) + { + ArgumentNullException.ThrowIfNull(rows); + var expires = clock.GetUtcNow().Add(_options.RollupMinuteRetention); + var failed = new Dictionary(); + Exception? lastError = null; + foreach (var (key, row) in rows) + { + try + { + await using var command = db.CreateCommand(Sql); + command.Parameters.AddWithValue("pk", $"{key.TenantId.Value}#{key.EndpointKey}"); + command.Parameters.AddWithValue("sk", $"1m#{key.Minute.UtcDateTime:yyyyMMddHHmm}"); + command.Parameters.AddWithValue("expires", expires); + var values = Values(row); + for (var i = 0; i < CounterNames.Length; i++) + { + command.Parameters.AddWithValue($"v{i}", values[i]); + } + + await command.ExecuteNonQueryAsync(cancellationToken).ConfigureAwait(false); + } + catch (Exception ex) when (ex is not OperationCanceledException) + { + failed[key] = row; + lastError = ex; + } + } + + if (lastError is not null) + { + LogFlushFailed(logger, lastError, failed.Count, rows.Count); + } + else + { + LogFlushed(logger, rows.Count); + } + + return failed; + } + + private static long[] Values(RollupRow row) + { + var values = new List + { + row.Created, row.Attempts, row.Succeeded, row.Retried, row.Failed, row.Exhausted, row.Timeouts, + row.ConnectionMs, row.ResponseTime.SumMs, row.QueueWait.SumMs, + }; + values.AddRange(row.ResponseTime.Counts); + values.AddRange(row.QueueWait.Counts); + return [.. values]; + } + + private static string BuildSql() + { + var columns = string.Join(", ", CounterNames); + var parameters = string.Join(", ", CounterNames.Select((_, i) => $"@v{i}")); + var updates = string.Join(", ", CounterNames.Select(c => $"{c} = webhook.rollups.{c} + EXCLUDED.{c}")); + return new StringBuilder() + .Append(CultureInfo.InvariantCulture, $"INSERT INTO webhook.rollups (pk, sk, expires_at, {columns}) VALUES (@pk, @sk, @expires, {parameters}) ") + .Append(CultureInfo.InvariantCulture, $"ON CONFLICT (pk, sk) DO UPDATE SET {updates}, expires_at = EXCLUDED.expires_at") + .ToString(); + } + + [LoggerMessage(Level = LogLevel.Debug, Message = "Rollups flushed: {RowCount} rows")] + private static partial void LogFlushed(ILogger logger, int rowCount); + + [LoggerMessage(Level = LogLevel.Error, Message = "Rollup flush failed for {Failed} of {Total} rows; they are kept for the next flush")] + private static partial void LogFlushFailed(ILogger logger, Exception exception, int failed, int total); +} diff --git a/src/Postgres/PostgresServiceCollectionExtensions.cs b/src/Postgres/PostgresServiceCollectionExtensions.cs index c2821cb..9b30383 100644 --- a/src/Postgres/PostgresServiceCollectionExtensions.cs +++ b/src/Postgres/PostgresServiceCollectionExtensions.cs @@ -1,5 +1,6 @@ using Microsoft.Extensions.Configuration; using Microsoft.Extensions.DependencyInjection; +using Microsoft.Extensions.DependencyInjection.Extensions; using Npgsql; namespace WebhookGateway.Postgres; @@ -17,7 +18,7 @@ public static IServiceCollection AddPostgresDataSource(this IServiceCollection s { services.Configure(configuration.GetSection(PostgresOptions.SectionName)); var connectionString = configuration[$"{PostgresOptions.SectionName}:ConnectionString"]!; - services.AddSingleton(_ => new NpgsqlDataSourceBuilder(connectionString).Build()); + services.TryAddSingleton(_ => new NpgsqlDataSourceBuilder(connectionString).Build()); return services; } } diff --git a/tests/IntegrationTests/Postgres/PostgresBackendTests.cs b/tests/IntegrationTests/Postgres/PostgresBackendTests.cs new file mode 100644 index 0000000..7c7983f --- /dev/null +++ b/tests/IntegrationTests/Postgres/PostgresBackendTests.cs @@ -0,0 +1,260 @@ +using System.Text.Json; +using Microsoft.Extensions.Logging.Abstractions; +using Microsoft.Extensions.Options; +using Microsoft.Extensions.Time.Testing; +using Npgsql; +using WebhookGateway.Contracts; +using WebhookGateway.Contracts.Ids; +using WebhookGateway.Contracts.Messages; +using WebhookGateway.Core.Rollups; +using WebhookGateway.Dispatcher.Delivery; +using WebhookGateway.Postgres; +using WebhookGateway.Postgres.Backend; + +namespace WebhookGateway.IntegrationTests.Postgres; + +/// Needs the local stack (./scripts/dev-up.sh, PostgreSQL on 5432); skipped when it is not reachable. +[Collection("postgres-queues")] +public sealed class PostgresBackendTests +{ + public const string Connection = "Host=localhost;Port=5432;Database=webhook;Username=webhook;Password=webhook;Timeout=2"; + private static readonly DateTimeOffset Now = new(2026, 10, 21, 12, 0, 0, TimeSpan.Zero); + + internal static async Task ConnectAsync() + { + var db = NpgsqlDataSource.Create(Connection); + try + { + await using var connection = await db.OpenConnectionAsync(); + var dir = AppContext.BaseDirectory; + while (dir is not null && !Directory.Exists(Path.Combine(dir, "db", "migrations"))) + { + dir = Path.GetDirectoryName(dir); + } + + await new MigrationRunner(db, NullLogger.Instance).RunAsync(Path.Combine(dir!, "db", "migrations"), CancellationToken.None); + await using var clear = db.CreateCommand("DELETE FROM webhook.queue_messages"); + await clear.ExecuteNonQueryAsync(); + return db; + } + catch (NpgsqlException) + { + await db.DisposeAsync(); + return null; + } + } + + private static PostgresMessageStore Messages(NpgsqlDataSource db) => new(db, NullLogger.Instance); + + [Fact] + public async Task A_delayed_message_is_invisible_until_due_and_a_deleted_message_is_gone() + { + await using var db = await ConnectAsync(); + if (db is null) + { + return; + } + + var store = Messages(db); + await store.SendAsync("t-delay", [("later", TimeSpan.FromMilliseconds(800)), ("now", TimeSpan.Zero)], CancellationToken.None); + + var first = await store.ReceiveAsync("t-delay", 10, TimeSpan.FromSeconds(30), TimeSpan.Zero, CancellationToken.None); + await Task.Delay(900); + var second = await store.ReceiveAsync("t-delay", 10, TimeSpan.FromSeconds(30), TimeSpan.Zero, CancellationToken.None); + + Assert.Equal("now", Assert.Single(first).Body); + Assert.Equal("later", Assert.Single(second).Body); + await store.DeleteAsync(first[0], CancellationToken.None); + await store.DeleteAsync(second[0], CancellationToken.None); + Assert.Empty(await store.ReceiveAsync("t-delay", 10, TimeSpan.FromMilliseconds(1), TimeSpan.Zero, CancellationToken.None)); + } + + [Fact] + public async Task An_unacknowledged_message_reappears_and_moves_to_the_dead_letter_queue_after_five_receives() + { + await using var db = await ConnectAsync(); + if (db is null) + { + return; + } + + var store = Messages(db); + await store.SendAsync("t-dlq", [("poison", TimeSpan.Zero)], CancellationToken.None); + + for (var i = 1; i <= PostgresMessageStore.MaxReceives; i++) + { + var got = await store.ReceiveAsync("t-dlq", 1, TimeSpan.FromMilliseconds(50), TimeSpan.Zero, CancellationToken.None); + Assert.Equal(i, Assert.Single(got).ReceiveCount); + await Task.Delay(80); + } + + Assert.Empty(await store.ReceiveAsync("t-dlq", 1, TimeSpan.FromSeconds(30), TimeSpan.Zero, CancellationToken.None)); + var dead = await store.ReceiveAsync("t-dlq-dlq", 1, TimeSpan.FromSeconds(30), TimeSpan.Zero, CancellationToken.None); + Assert.Equal("poison", Assert.Single(dead).Body); + } + + [Fact] + public async Task Two_workers_never_receive_the_same_message() + { + await using var db = await ConnectAsync(); + if (db is null) + { + return; + } + + var store = Messages(db); + await store.SendAsync("t-race", [.. Enumerable.Range(0, 40).Select(i => ($"m{i}", TimeSpan.Zero))], CancellationToken.None); + + var batches = await Task.WhenAll(Enumerable.Range(0, 8).Select(_ => + store.ReceiveAsync("t-race", 10, TimeSpan.FromSeconds(30), TimeSpan.Zero, CancellationToken.None).AsTask())); + + var all = batches.SelectMany(b => b).Select(m => m.Body).ToList(); + Assert.Equal(40, all.Count); + Assert.Equal(40, all.Distinct().Count()); + } + + [Fact] + public async Task Delivery_state_round_trips_and_only_due_scheduled_rows_are_in_the_index() + { + await using var db = await ConnectAsync(); + if (db is null) + { + return; + } + + var store = new PostgresDeliveryStateStore(db, new FakeTimeProvider(Now), Options.Create(new BackendOptions()), NullLogger.Instance); + DeliveryStateRecord Record(DeliveryStatus status, DateTimeOffset next) + { + var eventId = EventId.New(); + var endpointId = EndpointId.New(); + return new DeliveryStateRecord( + TenantId.New(), DeliveryId.For(eventId, endpointId), endpointId, eventId, "order.created", Now.AddMinutes(-5), status, 3, + next, 503, "http_503", JsonDocument.Parse("""{"a":1}""").RootElement.Clone(), null, Now.AddMinutes(-4), Now, PlanTier.Premium); + } + + var due = Record(DeliveryStatus.Scheduled, Now.AddMinutes(-1)); + var later = Record(DeliveryStatus.Scheduled, Now.AddHours(3)); + var other = Record(DeliveryStatus.RetryScheduled, Now.AddMinutes(-1)); + foreach (var r in new[] { due, later, other }) + { + await store.PutAsync(r, CancellationToken.None); + } + + var read = await store.GetAsync(due.TenantId, due.DeliveryId, CancellationToken.None); + var dueNow = await store.GetDueAsync(Now, 1000, CancellationToken.None); + await store.DeleteAsync(due.TenantId, due.DeliveryId, CancellationToken.None); + + Assert.Equal(PlanTier.Premium, read!.Tier); + Assert.Equal(1, read.Payload!.Value.GetProperty("a").GetInt32()); + Assert.Contains(dueNow, r => r.DeliveryId == due.DeliveryId); + Assert.DoesNotContain(dueNow, r => r.DeliveryId == later.DeliveryId || r.DeliveryId == other.DeliveryId); + Assert.Null(await store.GetAsync(due.TenantId, due.DeliveryId, CancellationToken.None)); + } + + [Fact] + public async Task Rollups_from_two_flushes_are_added_together() + { + await using var db = await ConnectAsync(); + if (db is null) + { + return; + } + + var writer = new PostgresRollupWriter(db, new FakeTimeProvider(Now), Options.Create(new BackendOptions()), NullLogger.Instance); + var tenant = TenantId.New(); + var endpoint = EndpointId.New(); + var accumulator = new RollupAccumulator(); + for (var i = 0; i < 2; i++) + { + accumulator.RecordAttempt(new AttemptSample(tenant, endpoint, DeliveryOutcome.Succeeded, 10, 40, false, Now)); + Assert.Empty(await writer.WriteAsync(accumulator.Drain(), CancellationToken.None)); + } + + await using var read = db.CreateCommand("SELECT attempts, succeeded, rt_sum FROM webhook.rollups WHERE pk = @pk AND sk = '1m#202610211200'"); + read.Parameters.AddWithValue("pk", $"{tenant.Value}#{endpoint.Value}"); + await using var reader = await read.ExecuteReaderAsync(); + Assert.True(await reader.ReadAsync()); + Assert.Equal((2L, 2L, 80L), (reader.GetInt64(0), reader.GetInt64(1), reader.GetInt64(2))); + } + + [Fact] + public async Task Archived_events_are_read_back_by_their_pointer() + { + await using var db = await ConnectAsync(); + if (db is null) + { + return; + } + + var archive = new PostgresEventArchive(db, new FakeTimeProvider(Now), Options.Create(new BackendOptions()), NullLogger.Instance); + EventMessage NewEvent(int n) => new( + EventMessage.CurrentSchemaVersion, EventId.New(), TenantId.New(), "order.created", Now, Now, null, JsonDocument.Parse($$"""{"n":{{n}}}""").RootElement.Clone()); + var events = new[] { NewEvent(0), NewEvent(1), NewEvent(2) }; + + var refs = await archive.WriteBatchAsync(events, CancellationToken.None); + var target = events[2]; + var delivery = new DeliveryMessage( + DeliveryMessage.CurrentSchemaVersion, target.TenantId, target.EventId, DeliveryId.For(target.EventId, EndpointId.New()), EndpointId.New(), + target.Type, Now, 1, Now, Now, null, refs[target.EventId]); + var payload = await new PostgresPayloadResolver(db, NullLogger.Instance).ResolveAsync(delivery, CancellationToken.None); + var missing = await new PostgresPayloadResolver(db, NullLogger.Instance) + .ResolveAsync(delivery with { PayloadRef = "pg://archive/nope#0" }, CancellationToken.None); + + Assert.Equal(2, payload!.Value.GetProperty("n").GetInt32()); + Assert.Null(missing); + } + + [Fact] + public async Task Idempotency_keys_and_the_monthly_count_are_durable_and_exact() + { + await using var db = await ConnectAsync(); + if (db is null) + { + return; + } + + var idempotency = new PostgresIdempotencyStore(db); + var quota = new PostgresQuotaCounter(db); + var tenant = TenantId.New(); + var first = EventId.New(); + + var a = await idempotency.TryClaimAsync(tenant, "k", first, TimeSpan.FromMinutes(5), CancellationToken.None); + var b = await idempotency.TryClaimAsync(tenant, "k", EventId.New(), TimeSpan.FromMinutes(5), CancellationToken.None); + await idempotency.ReleaseAsync(tenant, "k", CancellationToken.None); + var c = await idempotency.TryClaimAsync(tenant, "k", EventId.New(), TimeSpan.FromMinutes(5), CancellationToken.None); + await quota.IncrementAsync(tenant, Now, CancellationToken.None); + var two = await quota.IncrementAsync(tenant, Now, CancellationToken.None); + await quota.DecrementAsync(tenant, Now, CancellationToken.None); + var again = await quota.IncrementAsync(tenant, Now, CancellationToken.None); + + Assert.True(a.Claimed); + Assert.False(b.Claimed); + Assert.Equal(first, b.ExistingEventId); + Assert.True(c.Claimed); + Assert.Equal(2, two); + Assert.Equal(2, again); + } + + [Fact] + public async Task Maintenance_removes_expired_rows() + { + await using var db = await ConnectAsync(); + if (db is null) + { + return; + } + + var tenant = TenantId.New(); + await using (var insert = db.CreateCommand("INSERT INTO webhook.idempotency_keys (tenant_id, key, event_id, expires_at) VALUES (@t, 'old', 'evt_x', now() - interval '1 minute')")) + { + insert.Parameters.AddWithValue("t", tenant.Value); + await insert.ExecuteNonQueryAsync(); + } + + var removed = await new PostgresMaintenance(db, NullLogger.Instance).RunOnceAsync(CancellationToken.None); + + Assert.True(removed >= 1); + var claim = await new PostgresIdempotencyStore(db).TryClaimAsync(tenant, "old", EventId.New(), TimeSpan.FromMinutes(1), CancellationToken.None); + Assert.True(claim.Claimed); + } +} diff --git a/tests/IntegrationTests/Postgres/PostgresPipelineEndToEndTests.cs b/tests/IntegrationTests/Postgres/PostgresPipelineEndToEndTests.cs new file mode 100644 index 0000000..aef56d1 --- /dev/null +++ b/tests/IntegrationTests/Postgres/PostgresPipelineEndToEndTests.cs @@ -0,0 +1,185 @@ +using System.Net; +using System.Net.Http.Headers; +using System.Security.Cryptography; +using System.Text; +using System.Text.Json; +using Microsoft.AspNetCore.Hosting; +using Microsoft.Extensions.Configuration; +using Microsoft.Extensions.DependencyInjection; +using Microsoft.Extensions.Logging; +using Microsoft.Extensions.Options; +using Npgsql; +using WebhookGateway.Aws; +using WebhookGateway.Contracts; +using WebhookGateway.Contracts.Ids; +using WebhookGateway.Contracts.Messages; +using WebhookGateway.Core.Rollups; +using WebhookGateway.Dispatcher.Delivery; +using WebhookGateway.Dispatcher.Endpoints; +using WebhookGateway.Dispatcher.Outbound; +using WebhookGateway.Dispatcher.Retry; +using WebhookGateway.Dispatcher.Source; +using WebhookGateway.FanOut; +using WebhookGateway.FanOut.Routing; +using WebhookGateway.Ingest; +using WebhookGateway.Ingest.Tenancy; +using WebhookGateway.Postgres; +using WebhookGateway.Postgres.Backend; + +namespace WebhookGateway.IntegrationTests.Postgres; + +/// +/// The whole product on PostgreSQL alone (Backend:Mode=Postgres, no AWS emulator): a tenant, key and endpoint in PostgreSQL; an event posted to Ingest lands on SQS; +/// Fan-out archives it to S3 and routes it using the config database; the Dispatcher reads the delivery queue, loads the +/// endpoint and its decrypted secret, and "sends" through a recording sender (real HTTP is covered elsewhere and +/// loopback targets are blocked on purpose). Needs ./scripts/dev-up.sh (PostgreSQL); skipped otherwise. +/// +[Collection("postgres-queues")] +public sealed class PostgresPipelineEndToEndTests +{ + private const string Connection = "Host=localhost;Port=5432;Database=webhook;Username=webhook;Password=webhook;Timeout=2"; + + [Fact] + public async Task An_event_posted_to_ingest_is_delivered_to_the_tenants_endpoint() + { + await using var db = NpgsqlDataSource.Create(Connection); + try + { + await using var probe = await db.OpenConnectionAsync(); + } + catch (NpgsqlException) + { + return; // local stack not running + } + + var localKey = Convert.ToBase64String(RandomNumberGenerator.GetBytes(32)); + var protector = LocalSecretProtector.FromBase64(localKey); + var tenant = Guid.CreateVersion7(); + var endpoint = Guid.CreateVersion7(); + var apiKey = $"wgk_e2e_{Guid.NewGuid():N}"; + var signing = WebhookGateway.Dispatcher.Signing.SigningSecret.Generate(); + var stored = await protector.ProtectAsync(signing.ToText(), CancellationToken.None); + await Exec(db, "INSERT INTO webhook.tenants (id, name, plan_code) VALUES (@id, 'E2E', 'advanced')", ("id", tenant)); + await Exec(db, "INSERT INTO webhook.api_keys (id, tenant_id, key_hash, prefix) VALUES (@id, @t, @h, 'wgk_e2e_')", ("id", Guid.CreateVersion7()), ("t", tenant), ("h", ApiKeyHasher.Hash(apiKey))); + await using (var insert = db.CreateCommand("INSERT INTO webhook.endpoints (id, tenant_id, url, secret_ciphertext, secret_kms_key_id) VALUES (@id, @t, 'https://customer.example.com/hook', @c, @k)")) + { + insert.Parameters.AddWithValue("id", endpoint); + insert.Parameters.AddWithValue("t", tenant); + insert.Parameters.AddWithValue("c", stored.Ciphertext); + insert.Parameters.AddWithValue("k", stored.KeyId); + await insert.ExecuteNonQueryAsync(); + } + + await Exec(db, "INSERT INTO webhook.subscriptions (id, tenant_id, endpoint_id, event_type) VALUES (@id, @t, @e, 'order.created')", ("id", Guid.CreateVersion7()), ("t", tenant), ("e", endpoint)); + + try + { + var settings = new Dictionary + { + ["Backend:Mode"] = "Postgres", + ["Postgres:ConnectionString"] = Connection, + ["Secrets:LocalKey"] = localKey, + }; + await using (var clear = db.CreateCommand("DELETE FROM webhook.queue_messages")) + { + await clear.ExecuteNonQueryAsync(); + } + + + // 1. Ingest accepts the event. + using var ingest = new Microsoft.AspNetCore.Mvc.Testing.WebApplicationFactory().WithWebHostBuilder(b => + { + foreach (var (k, v) in settings) + { + b.UseSetting(k, v); + } + }); + using var client = ingest.CreateClient(); + client.DefaultRequestHeaders.Authorization = new AuthenticationHeaderValue("Bearer", apiKey); + using var content = new StringContent("""{"type":"order.created","data":{"orderId":42}}""", Encoding.UTF8, "application/json"); + using var accepted = await client.PostAsync(new Uri("/v1/events", UriKind.Relative), content); + Assert.Equal(HttpStatusCode.Accepted, accepted.StatusCode); + var eventId = JsonDocument.Parse(await accepted.Content.ReadAsStringAsync()).RootElement.GetProperty("eventId").GetString()!; + + // 2. Fan-out routes it using the config database. + var configuration = new ConfigurationBuilder().AddInMemoryCollection(settings).Build(); + await using var fanOut = BuildFanOut(configuration); + Assert.Equal(1, await fanOut.GetRequiredService().RunOnceAsync(CancellationToken.None)); + + // 3. The Dispatcher delivers it to the endpoint. + var sender = new RecordingSender(); + await using var dispatcher = BuildDispatcher(configuration, sender); + var consumer = dispatcher.GetRequiredService(); + Assert.Equal(1, await consumer.RunOnceAsync(PlanTier.Advanced, CancellationToken.None)); + + var call = Assert.Single(sender.Calls); + Assert.Equal($"ep_{endpoint}", call.Target.EndpointId.Value); + Assert.Equal(new Uri("https://customer.example.com/hook"), call.Target.Url); + Assert.Equal(signing.ToText(), Assert.Single(call.Target.Secrets).ToText()); + Assert.Equal(eventId, call.Message.EventId.Value); + Assert.Equal(42, call.Payload.GetProperty("orderId").GetInt32()); + } + finally + { + await Exec(db, "DELETE FROM webhook.tenants WHERE id = @id", ("id", tenant)); + } + } + + private static ServiceProvider BuildFanOut(IConfiguration configuration) + { + var services = new ServiceCollection(); + services.AddLogging(); + services.AddSingleton(configuration); + services.AddSingleton(TimeProvider.System); + services.AddSingleton(); + services.AddPostgresDataSource(configuration); + services.AddSingleton(); + services.AddPostgresFanOut(configuration); + services.AddSingleton(); + services.AddSingleton(); + return services.BuildServiceProvider(); + } + + private static ServiceProvider BuildDispatcher(IConfiguration configuration, RecordingSender sender) + { + var services = new ServiceCollection(); + services.AddLogging(); + services.AddSingleton(configuration); + services.AddSingleton(TimeProvider.System); + services.AddSingleton(); + services.AddSingleton(RetryPolicy.CreateDefault()); + services.Configure(_ => { }); + services.Configure(_ => { }); + services.AddPostgresDataSource(configuration); + services.AddSecretProtection(configuration); + services.AddSingleton(); + services.AddPostgresDispatcher(configuration); + services.AddSingleton(sender); + services.AddSingleton(); + services.AddSingleton(sp => sp.GetRequiredService()); + services.AddSingleton(); + return services.BuildServiceProvider(); + } + + private static async Task Exec(NpgsqlDataSource db, string sql, params (string, object)[] parameters) + { + await using var command = db.CreateCommand(sql); + foreach (var (name, value) in parameters) + { + command.Parameters.AddWithValue(name, value); + } + + await command.ExecuteNonQueryAsync(); + } + + private sealed class RecordingSender : IWebhookSender + { + public List<(EndpointTarget Target, DeliveryMessage Message, JsonElement Payload)> Calls { get; } = []; + + public ValueTask SendAsync(EndpointTarget target, DeliveryMessage message, JsonElement payload, CancellationToken cancellationToken) + { + Calls.Add((target, message, payload)); + return ValueTask.FromResult(new SendResult(SendOutcome.Succeeded, 200, 12, null)); + } + } +}