feat(sink): write events directly to PostgreSQL

README.md

@@ -8,7 +8,7 @@ Rust + axum, deployed as an ARM64 container on ECS Fargate.
 
 | Method | Path | Description |
 |---|---|---|
-| GET | `/health` | Liveness probe → `{"status":"healthy"}`. Touches no downstream; returns `503` once the Kafka producer has hit a fatal error, so ECS replaces the task. |
+| GET | `/health` | Liveness probe → `{"status":"healthy"}`. Touches no downstream; returns `503` once the sink has hit a fatal error (the Kafka producer's, or a ClickHouse config error), so ECS replaces the task. |
 | POST | `/v1/events` | Ingest an event batch. Requires the `X-Environment-Key` header. |
 
 `POST /v1/events` validates each event independently — invalid events are rejected individually and never block valid ones in the same batch.
@@ -52,7 +52,7 @@ On the wire to Kafka, `traits` and `metadata` are produced as JSON-encoded strin
 
 | Env var | Required | Default |
 |---|---|---|
-| `SINK` | no | `kafka` (or `clickhouse`) |
+| `SINK` | no | `kafka` (or `clickhouse`, `postgres`) |
 | `KAFKA_BOOTSTRAP_SERVERS` | with `kafka` | — |
 | `KAFKA_TOPIC` | with `kafka` | — |
 | `KAFKA_EXTERNAL_TOPIC` | with `kafka` | — |
@@ -69,7 +69,7 @@ On the wire to Kafka, `traits` and `metadata` are produced as JSON-encoded strin
 | `DATABASE_URL` | yes | — |
 | `LISTEN_ADDR` | no | `0.0.0.0:8080` |
 
-With `SINK=kafka`, events from environments that use an external warehouse go to `KAFKA_EXTERNAL_TOPIC`; all others go to `KAFKA_TOPIC`. With `SINK=clickhouse`, every event goes to the one ClickHouse table. The server looks presented keys up in the `experimentation_environment_keys` view in Flagsmith's Postgres, at `DATABASE_URL`.
+With `SINK=kafka`, events from environments that use an external warehouse go to `KAFKA_EXTERNAL_TOPIC`; all others go to `KAFKA_TOPIC`. With `SINK=clickhouse`, every event goes to the one ClickHouse table, and with `SINK=postgres`, to `flagsmith_exp.events` at `DATABASE_URL`. The server looks presented keys up in the `experimentation_environment_keys` view in Flagsmith's Postgres, at `DATABASE_URL`.
 
 ### Database role
 
@@ -88,7 +88,16 @@ With `KAFKA_AUTH=scram`, the producer authenticates with SASL/SCRAM-SHA-512 over
 
 ### ClickHouse sink
 
-With `SINK=clickhouse`, `202` means the events are buffered in memory, not yet stored. A background task takes the whole buffer every `SINK_FLUSH_INTERVAL_MS`, or sooner once `SINK_FLUSH_MAX_EVENTS` events are waiting, and writes it as one `INSERT`. A failed batch is retried 3 times and then dropped with an error log. The buffer is unbounded: requests never wait or get `503`, so while a write is slow or failing (up to about 2 minutes with retries) it grows by the ingest rate times the write time, and the next flush takes it all. A normal shutdown (SIGTERM) flushes the buffer, but events still buffered are lost if the process crashes.
+With `SINK=clickhouse`, `202` means the events are buffered in memory, not yet stored. A background task takes the whole buffer every `SINK_FLUSH_INTERVAL_MS`, or sooner once `SINK_FLUSH_MAX_EVENTS` events are waiting, and writes it as one `INSERT`. A batch that fails with a transient error (network, timeout, overload, or anything unrecognised) is retried 5 times over about 17 seconds, enough to ride out a ClickHouse restart, and then dropped with an error log; these never affect `/health`. Rows the table rejects (parse, type or constraint errors) drop the batch at once, with an error log, and leave `/health` alone. A config error (bad credentials or grants, or a missing database, table or column) drops the batch at once and fails `/health` until restart. At startup the server checks that ClickHouse is reachable, accepts the credentials, and has a table whose columns match, and exits with an error if not. The buffer is unbounded: requests never wait or get `503`, so while a write is slow or failing (up to about 3 minutes with retries) it grows by the ingest rate times the write time, and the next flush takes it all. A normal shutdown (SIGTERM) flushes the buffer, but events still buffered are lost if the process crashes.
+
+### PostgreSQL sink
+
+With `SINK=postgres`, events are buffered, flushed and retried as with the ClickHouse sink, and each batch is written to `flagsmith_exp.events` in Flagsmith's Postgres, at `DATABASE_URL`, as one `INSERT` in a transaction. Flagsmith's migrations create the table. The sink shares the server's connection pool. A missing table or column, or a permission or authentication error, fails `/health` until restart. If the server's role does not own the table, grant it:
+
+```sql
+GRANT USAGE ON SCHEMA flagsmith_exp TO ingestion_server;
+GRANT INSERT ON flagsmith_exp.events TO ingestion_server;
+```
 
 ## Run locally
 

src/config.rs

@@ -18,6 +18,9 @@ pub enum SinkConfig {
         clickhouse: ClickHouseConfig,
         batch: BatchConfig,
     },
+    Postgres {
+        batch: BatchConfig,
+    },
 }
 
 #[derive(Debug, Clone)]
@@ -36,6 +39,9 @@ impl Config {
                 clickhouse: clickhouse_config_from_env()?,
                 batch: batch_config_from_env()?,
             },
+            SinkKind::Postgres => SinkConfig::Postgres {
+                batch: batch_config_from_env()?,
+            },
         };
 
         Ok(Config {
@@ -71,13 +77,15 @@ fn required(name: &'static str) -> Result<String, ConfigError> {
 enum SinkKind {
     Kafka,
     ClickHouse,
+    Postgres,
 }
 
 /// Which sink `SINK` selects; unset means Kafka.
 fn parse_sink(value: Option<&str>) -> Result<SinkKind, ConfigError> {
     match value.unwrap_or("kafka") {
         "kafka" => Ok(SinkKind::Kafka),
         "clickhouse" => Ok(SinkKind::ClickHouse),
+        "postgres" => Ok(SinkKind::Postgres),
         other => Err(ConfigError::InvalidSink(other.to_string())),
     }
 }
@@ -134,7 +142,7 @@ pub enum ConfigError {
     MissingRequired(&'static str),
     #[error("Invalid value for {0}: {1:?} (expected scram or none)")]
     Invalid(&'static str, String),
-    #[error("Invalid value for SINK: {0:?} (expected kafka or clickhouse)")]
+    #[error("Invalid value for SINK: {0:?} (expected kafka, clickhouse or postgres)")]
     InvalidSink(String),
     #[error("Invalid value for {0}: {1:?} (expected a positive integer)")]
     InvalidNumber(&'static str, String),
@@ -189,13 +197,14 @@ mod tests {
             parse_sink(Some("clickhouse")).unwrap(),
             SinkKind::ClickHouse
         );
+        assert_eq!(parse_sink(Some("postgres")).unwrap(), SinkKind::Postgres);
     }
 
     #[test]
     fn parse_sink_rejects_unknown_value() {
         assert!(matches!(
-            parse_sink(Some("postgres")),
-            Err(ConfigError::InvalidSink(v)) if v == "postgres"
+            parse_sink(Some("bigquery")),
+            Err(ConfigError::InvalidSink(v)) if v == "bigquery"
         ));
     }
 

src/main.rs

@@ -11,14 +11,15 @@ mod test_helpers;
 use std::sync::Arc;
 use std::time::Duration;
 
+use anyhow::Context;
 use sqlx::postgres::PgPoolOptions;
 use tokio::signal;
 use tracing::info;
 
 use auth::ContextState;
 use config::{Config, SinkConfig};
 use environment_keys::EnvironmentKeys;
-use sink::{BatchingSink, ClickHouseWriter, KafkaSink, KafkaSinkConfig, Sink};
+use sink::{BatchingSink, ClickHouseWriter, KafkaSink, KafkaSinkConfig, PostgresWriter, Sink};
 
 const ENVIRONMENT_KEY_REFRESH_AFTER: Duration = Duration::from_secs(300);
 
@@ -33,6 +34,10 @@ async fn main() -> anyhow::Result<()> {
         .init();
 
     let config = Config::from_env()?;
+    let pool = PgPoolOptions::new()
+        .max_connections(8)
+        .acquire_timeout(Duration::from_secs(1))
+        .connect_lazy(&config.database_url)?;
     let (sink, external_topic) = match &config.sink {
         SinkConfig::Kafka(kafka) => {
             info!(
@@ -57,17 +62,30 @@ async fn main() -> anyhow::Result<()> {
                 "Starting flagsmith-analytics server"
             );
             let writer = ClickHouseWriter::new(clickhouse.clone());
+            writer
+                .check()
+                .await
+                .context("ClickHouse startup check failed")?;
             (
                 Sink::ClickHouse(BatchingSink::new(writer, batch.clone())),
                 None,
             )
         }
+        SinkConfig::Postgres { batch } => {
+            info!(
+                listen_addr = %config.listen_addr,
+                sink = "postgres",
+                batch = ?batch,
+                "Starting flagsmith-analytics server"
+            );
+            let writer = PostgresWriter::new(pool.clone());
+            (
+                Sink::Postgres(BatchingSink::new(writer, batch.clone())),
+                None,
+            )
+        }
     };
 
-    let pool = PgPoolOptions::new()
-        .max_connections(8)
-        .acquire_timeout(Duration::from_secs(1))
-        .connect_lazy(&config.database_url)?;
     let context_state = ContextState {
         environment_keys: EnvironmentKeys::new(pool, ENVIRONMENT_KEY_REFRESH_AFTER),
         external_topic: external_topic.map(Arc::from),

src/routes/health.rs

@@ -10,7 +10,7 @@ pub struct HealthResponse {
     error: Option<String>,
 }
 
-/// Liveness plus one local check: a fatal producer error is permanent, so
+/// Liveness plus one local check: a fatal sink error is permanent, so
 /// report it and let ECS replace the task. No downstream call, so this cannot hang.
 pub async fn health_check(State(sink): State<Sink>) -> (StatusCode, Json<HealthResponse>) {
     match sink.fatal_error() {