//! Postgres write-throughput benchmark for `horizon-sdk`.
//!
//! Drives every entity type with single-row inserts and (for the high-volume
//! time-series tables) varying batch sizes against a live local Postgres
//! instance. Mirrors the workload shape that the parallel Python timing
//! script executes so per-variant throughput numbers between the two
//! SDKs reflect runtime differences and not workload differences.
//!
//! # Running
//!
//! ```bash
//! # against the local docker compose Postgres
//! DATABASE_URL=postgresql://horizon_owner:insecure_password@localhost:5432/horizon?sslmode=disable \
//!   cargo bench --bench postgres_write_throughput
//!
//! # output JSON for downstream sticky-comment / sign-off tooling
//! HORIZON_BENCH_JSON=/tmp/horizon-sdk-bench.json \
//!   cargo bench --bench postgres_write_throughput
//! ```
//!
//! Criterion writes HTML reports to `target/criterion/`. When
//! `HORIZON_BENCH_JSON` is set, a summary line per variant is appended
//! to that path as newline-delimited JSON for downstream consumption.
//!
//! # Environment
//!
//! - `DATABASE_URL`:Postgres connection string (required). Default in
//!   local dev: `postgresql://horizon_owner:insecure_password@localhost:5432/horizon?sslmode=disable`.
//! - `HORIZON_BENCH_ROLE`:Postgres role to `SET ROLE` per connection.
//!   Defaults to empty (no role switch, connection stays as the role
//!   from the URL, typically `horizon_owner`) so the bench measures
//!   the production write path used by Bytewax / Dagster producers.
//!   Set to `horizon_visitor` to exercise the RLS-aware app-user write
//!   path. Note that the visitor role only has column-level INSERT
//!   grants on most tables, so a visitor-mode run requires careful
//!   workload alignment with what visitor is permitted to write.
//! - `HORIZON_BENCH_DATA_ROW_BATCH_SIZES`:comma-separated list of
//!   batch sizes for the `data_row` variant. Defaults to
//!   `1,100,1000,5000`. Larger values (10000, 50000) are recommended on
//!   CI hardware but skipped by default to keep the local run fast.
//! - `HORIZON_BENCH_METADATA_ROW_BATCH_SIZES`:same shape, defaults to
//!   `1,100,1000`.
//! - `HORIZON_BENCH_ANNOTATION_BATCH_SIZES`:same shape, batch sizes for
//!   the `annotation` variant. Defaults to `1,100,1000`.
//! - `HORIZON_BENCH_JSON`:optional output file for newline-delimited
//!   JSON summaries (see `bench_summary` below).
//! - `HORIZON_BENCH_VECTOR_LENGTH`:`DataRow.vector` length. Default
//!   `512`; production rows typically sit between 256 and 4096.
//! - `HORIZON_BENCH_SEED`:`BenchSeed.seed` value. Default `0`. Change
//!   to isolate concurrent runs on the same database.
//!
//! Every bench variant cleans up its parent platform after the run so
//! repeated invocations do not accumulate fixture data. Time-series rows
//! are cleaned up by `ON DELETE CASCADE` on the parent `data_stream`.

#![allow(
    clippy::absolute_paths,
    clippy::arbitrary_source_item_ordering,
    clippy::as_conversions,
    clippy::expect_used,
    clippy::let_underscore_must_use,
    clippy::let_underscore_untyped,
    clippy::missing_docs_in_private_items,
    clippy::print_stderr,
    clippy::ref_option,
    clippy::shadow_reuse,
    clippy::significant_drop_tightening,
    clippy::single_call_fn,
    clippy::struct_field_names,
    clippy::too_many_lines,
    reason = "criterion harness uses .expect() on repo calls and `let _ =` cleanup throughout, each per-variant helper is called once from a single dispatcher, and criterion's API requires `&Option<T>` closure shapes that trip restriction lints"
)]

use std::env;
use std::fs::OpenOptions;
use std::io::Write as _;
use std::time::Duration as StdDuration;

use criterion::measurement::Measurement;
use criterion::{
    BenchmarkGroup, BenchmarkId, Criterion, Throughput, criterion_group, criterion_main,
};
use horizon_sdk::bench_support::{
    self, BenchSeed, annotation, audio_specification, beamgram_specification,
    bearing_time_record_specification, data_row, data_stream, metadata_row, mission,
    mission_platform, ontology, ontology_class, platform, platform_audio_specification,
    platform_beamgram_specification, platform_bearing_time_record_specification,
    platform_information, platform_kind, platform_spectrogram_specification,
    spectrogram_specification, uuid_for,
};
use horizon_sdk::postgres::repository::PostgresRepository;
use sqlx::PgPool;
use sqlx::postgres::{PgConnectOptions, PgPoolOptions};
use tokio::runtime::Runtime;
use uuid::Uuid;

/// Default `DATABASE_URL` matching the local docker-compose Postgres.
const DEFAULT_DATABASE_URL: &str =
    "postgresql://horizon_owner:insecure_password@localhost:5432/horizon?sslmode=disable";
/// Default Postgres role override for pooled connections. Empty by
/// default so the connection stays as the URL role (typically
/// `horizon_owner`), matching the production write path used by
/// Bytewax / Dagster producers. Override via `HORIZON_BENCH_ROLE` to
/// measure under a different role (e.g. `horizon_visitor` for the
/// RLS-aware app-user path).
const DEFAULT_ROLE: &str = "";
/// Default batch sizes for the `data_row` variant.
const DEFAULT_DATA_ROW_BATCH_SIZES: &[usize] = &[1, 100, 1_000, 5_000];
/// Default batch sizes for the `metadata_row` variant.
const DEFAULT_METADATA_ROW_BATCH_SIZES: &[usize] = &[1, 100, 1_000];
/// Default batch sizes for the `annotation` variant.
const DEFAULT_ANNOTATION_BATCH_SIZES: &[usize] = &[1, 100, 1_000];
/// Maximum `counter` value pre-cleaned at variant init for single-row
/// entity benches. A previously-panicked run leaves rows behind whose
/// deterministic IDs collide with the next run; clearing IDs for
/// indices `0..=PRECLEAN_BOUND` guarantees the first iteration measures
/// a clean INSERT path rather than an upsert-conflict or unique-key
/// failure. Sized well above the typical criterion iteration count for
/// the fastest variants (~3 ms/iter, ~3 000 iters per run).
const PRECLEAN_BOUND: u64 = 0x4000;

/// Parse a comma-separated list of usize values from an env var, falling
/// back to `default_sizes` when the variable is unset or empty.
fn parse_batch_sizes_env(var: &str, default_sizes: &[usize]) -> Vec<usize> {
    env::var(var)
        .ok()
        .filter(|raw| !raw.trim().is_empty())
        .map(|raw| {
            raw.split(',')
                .filter_map(|part| part.trim().parse::<usize>().ok())
                .filter(|size| *size > 0)
                .collect::<Vec<_>>()
        })
        .filter(|parsed| !parsed.is_empty())
        .unwrap_or_else(|| default_sizes.to_vec())
}

/// A Postgres role name validated for safe interpolation into `SET ROLE`.
///
/// The only constructor (`from_env`) rejects anything outside ASCII
/// alphanumeric and underscore, so holding a `BenchRole` is proof the
/// inner string cannot break out of the `SET ROLE` statement. Keeping
/// the guarantee in the type means the validation cannot drift away
/// from the interpolation site even as the two move apart in the file.
struct BenchRole(String);

impl BenchRole {
    /// Build a role from a raw environment value, returning `None` when
    /// the value is empty (no role switch) and panicking when it carries
    /// characters that could escape the `SET ROLE` identifier.
    fn from_env(raw: &str) -> Option<Self> {
        let trimmed = raw.trim();
        if trimmed.is_empty() {
            return None;
        }
        assert!(
            trimmed
                .chars()
                .all(|c| c.is_ascii_alphanumeric() || c == '_'),
            "HORIZON_BENCH_ROLE must contain only ASCII alphanumeric and underscore; got: {trimmed:?}"
        );
        Some(Self(trimmed.to_owned()))
    }

    fn as_str(&self) -> &str {
        &self.0
    }
}

/// Resolved benchmark configuration read once from the environment.
struct BenchConfig {
    annotation_batch_sizes: Vec<usize>,
    database_url: String,
    data_row_batch_sizes: Vec<usize>,
    json_output_path: Option<String>,
    metadata_row_batch_sizes: Vec<usize>,
    role: Option<BenchRole>,
    seed: BenchSeed,
}

impl BenchConfig {
    fn from_env() -> Self {
        let database_url =
            env::var("DATABASE_URL").unwrap_or_else(|_| DEFAULT_DATABASE_URL.to_owned());
        let role_raw = env::var("HORIZON_BENCH_ROLE").unwrap_or_else(|_| DEFAULT_ROLE.to_owned());
        let role = BenchRole::from_env(&role_raw);
        let vector_length = env::var("HORIZON_BENCH_VECTOR_LENGTH")
            .ok()
            .and_then(|raw| raw.parse::<usize>().ok())
            .unwrap_or(512);
        let seed_value = env::var("HORIZON_BENCH_SEED")
            .ok()
            .and_then(|raw| raw.parse::<u64>().ok())
            .unwrap_or(0);
        let seed = BenchSeed {
            seed: seed_value,
            vector_length,
            ..BenchSeed::default()
        };
        Self {
            annotation_batch_sizes: parse_batch_sizes_env(
                "HORIZON_BENCH_ANNOTATION_BATCH_SIZES",
                DEFAULT_ANNOTATION_BATCH_SIZES,
            ),
            database_url,
            data_row_batch_sizes: parse_batch_sizes_env(
                "HORIZON_BENCH_DATA_ROW_BATCH_SIZES",
                DEFAULT_DATA_ROW_BATCH_SIZES,
            ),
            json_output_path: env::var("HORIZON_BENCH_JSON").ok(),
            metadata_row_batch_sizes: parse_batch_sizes_env(
                "HORIZON_BENCH_METADATA_ROW_BATCH_SIZES",
                DEFAULT_METADATA_ROW_BATCH_SIZES,
            ),
            role,
            seed,
        }
    }
}

/// Build a connection pool against `database_url`. When `role` is
/// `Some`, every connection runs `SET ROLE <role>` after connecting so
/// measured writes traverse the requested role's permission path. When
/// `role` is `None`, the pool stays as the URL's credentials (typically
/// `horizon_owner`) so it can drive fixture setup and teardown without
/// hitting RLS predicates.
async fn build_pool(database_url: &str, role: Option<&BenchRole>) -> PgPool {
    let connect_options: PgConnectOptions = database_url
        .parse()
        .expect("DATABASE_URL must be a valid Postgres URL");
    let role = role.map(|role| role.as_str().to_owned());
    PgPoolOptions::new()
        .max_connections(8)
        .after_connect(move |conn, _meta| {
            let role = role.clone();
            Box::pin(async move {
                if let Some(role) = role.as_ref() {
                    let query = format!("SET ROLE {role}");
                    sqlx::query(&query).execute(&mut *conn).await?;
                }
                Ok(())
            })
        })
        .connect_with(connect_options)
        .await
        .expect("benchmark requires a live Postgres at DATABASE_URL")
}

/// Owner-credentials pool used for fixture lifecycle (warmup, parent
/// seeding, pre-clean of leftover deterministic rows, post-bench
/// cleanup). RLS predicates on `horizon_visitor` would block these
/// operations, so the lifecycle path never inherits the role override.
async fn build_owner_pool(config: &BenchConfig) -> PgPool {
    build_pool(&config.database_url, None).await
}

/// Role-switched pool used exclusively inside `bencher.iter()` so the
/// measured write path reflects whatever role the run targets
/// (`horizon_owner` by default, `horizon_visitor` when
/// `HORIZON_BENCH_ROLE` is set).
async fn build_role_pool(config: &BenchConfig) -> PgPool {
    build_pool(&config.database_url, config.role.as_ref()).await
}

/// Bench-side fixture lifecycle + measurement plumbing. The owner pool
/// is reserved for setup, warmup, pre-clean, and teardown; the role
/// pool is used only inside `bencher.iter()` so the measured path
/// reflects the requested role's permission set.
struct BenchRepos {
    owner: PostgresRepository,
    owner_pool: PgPool,
    role: PostgresRepository,
}

impl BenchRepos {
    async fn build(config: &BenchConfig) -> Self {
        let owner_pool = build_owner_pool(config).await;
        let role_pool = build_role_pool(config).await;
        Self {
            owner: PostgresRepository::new(owner_pool.clone(), None),
            owner_pool,
            role: PostgresRepository::new(role_pool, None),
        }
    }
}

/// Bulk-delete leftover deterministic rows from a prior run for the
/// given table. Computes the same IDs the upcoming bench iter would
/// generate for indices `0..=PRECLEAN_BOUND` and issues a single
/// `DELETE … WHERE id = ANY($1::uuid[])` against the owner pool. The
/// SDK exposes delete-by-id, not bulk delete, so the raw SQL is
/// scoped to bench fixtures only. Every call site passes a hardcoded
/// literal, and the charset assertion keeps that guarantee next to the
/// interpolation so a future caller cannot reach this sink with an
/// identifier sourced from config or the environment.
async fn preclean_table_by_label(pool: &PgPool, table: &str, seed: u64, label_prefix: &str) {
    assert!(
        table.chars().all(|c| c.is_ascii_alphanumeric() || c == '_'),
        "preclean table must contain only ASCII alphanumeric and underscore; got: {table:?}"
    );
    let ids: Vec<Uuid> = (0_u64..=PRECLEAN_BOUND)
        .map(|index| uuid_for(seed, &format!("{label_prefix}:{index}")))
        .collect();
    let query = format!("DELETE FROM horizon_public.{table} WHERE id = ANY($1::uuid[])");
    let _ = sqlx::query(&query).bind(&ids).execute(pool).await;
}

/// Warm the pool, sqlx prepared-statement cache, and Postgres buffer
/// cache before measurement begins. The first acquisition on a freshly
/// built pool pays connect-time overhead (TLS handshake, role-set per
/// `after_connect`, and Postgres backend startup) that would otherwise
/// inflate the first measured variant's timing. A handful of throwaway
/// upserts on a dedicated warmup row absorbs that cost upfront.
async fn warm_pool(repo: &PostgresRepository, seed: BenchSeed) {
    let warmup_row = platform(seed, u64::MAX - 100);
    let warmup_id = warmup_row.id.expect("warmup platform row carries id");
    for _ in 0_usize..5_usize {
        let _ = repo.upsert_platform(&warmup_row).await;
    }
    let _ = repo.delete_platform(warmup_id).await;
}

/// Warm both the owner and role pools so the first measured iteration
/// does not pay TLS handshake, `SET ROLE`, or sqlx prepared-statement
/// compile cost on whichever pool drives the timed write path. Each
/// upsert is best-effort; visitor-role configurations that lack write
/// grants on `platform` simply skip the role-side warmup silently.
async fn warm_pools(repos: &BenchRepos, seed: BenchSeed) {
    warm_pool(&repos.owner, seed).await;
    warm_pool(&repos.role, seed).await;
}

/// Open a long-lived fixture platform + data stream + spec graph and
/// return the IDs needed by the time-series variants. The owner role
/// (default `horizon_owner` via the pool URL) is used to seed fixtures
/// so RLS policies on `horizon_visitor` do not block setup.
struct TimeSeriesFixture {
    data_stream_id: Uuid,
    platform_id: Uuid,
    specification_id: Uuid,
}

async fn seed_time_series_fixture(repo: &PostgresRepository, seed: BenchSeed) -> TimeSeriesFixture {
    let platform_row = bench_support::platform(seed, 0);
    let platform_id = repo
        .upsert_platform(&platform_row)
        .await
        .expect("seed platform")
        .id
        .expect("upserted platform returns id");
    let data_stream_row = bench_support::data_stream(seed, platform_id, 0);
    let data_stream_id = repo
        .upsert_data_stream(&data_stream_row)
        .await
        .expect("seed data stream")
        .id
        .expect("upserted data stream returns id");
    let spec_row = bench_support::spectrogram_specification(seed, 0);
    let specification_id = repo
        .upsert_spectrogram_specification(&spec_row)
        .await
        .expect("seed spectrogram specification")
        .id
        .expect("upserted spec returns id");
    TimeSeriesFixture {
        data_stream_id,
        platform_id,
        specification_id,
    }
}

/// Best-effort cleanup of the fixture graph. `ON DELETE CASCADE` removes
/// the data stream and all time-series rows when the platform is deleted.
async fn teardown_time_series_fixture(repo: &PostgresRepository, fixture: &TimeSeriesFixture) {
    let _ = repo.delete_platform(fixture.platform_id).await;
    let _ = repo
        .delete_spectrogram_specification(fixture.specification_id)
        .await;
}

/// Append a single benchmark summary line to the JSON output file when
/// `HORIZON_BENCH_JSON` is set. Each line is a self-contained JSON
/// object so downstream consumers can stream-parse without buffering.
fn write_json_summary(
    path: &Option<String>,
    variant: &str,
    batch_size: Option<usize>,
    iterations: u64,
    total_rows: u64,
) {
    let Some(path) = path else {
        return;
    };
    let mut file = match OpenOptions::new().create(true).append(true).open(path) {
        Ok(file) => file,
        Err(error) => {
            eprintln!("HORIZON_BENCH_JSON open failed for {path}: {error}");
            return;
        }
    };
    let line = serde_json::json!({
        "variant": variant,
        "batch_size": batch_size,
        "iterations": iterations,
        "total_rows": total_rows,
    });
    if let Err(error) = writeln!(file, "{line}") {
        eprintln!("HORIZON_BENCH_JSON write failed: {error}");
    }
}

/// Bench `data_row` inserts at each configured batch size. `insert_data_row_batch`
/// is the primary write path that downstream perf PRs will replace with
/// `COPY FROM` and per-spec parallel connections.
fn bench_data_row_inserts(criterion: &mut Criterion, runtime: &Runtime, config: &BenchConfig) {
    let repos = runtime.block_on(BenchRepos::build(config));
    runtime.block_on(warm_pools(&repos, config.seed));
    let fixture = runtime.block_on(seed_time_series_fixture(&repos.owner, config.seed));

    let mut group = criterion.benchmark_group("postgres_data_row_insert");
    group.measurement_time(StdDuration::from_secs(20));
    group.sample_size(10);

    // Advance the row index across iterations so every insert lands on a
    // fresh timestamp. Re-inserting an identical batch would pile duplicate
    // rows into a single Timescale chunk, bloating the chunk and its indexes
    // mid-run and drifting the measured latency away from the write path.
    let mut iteration_offset = 0_u64;
    for &batch_size in &config.data_row_batch_sizes {
        group.throughput(Throughput::Elements(batch_size as u64));
        let batch_size_u64 = batch_size as u64;
        group.bench_function(BenchmarkId::from_parameter(batch_size), |bencher| {
            let mut iterations = 0_u64;
            bencher.iter(|| {
                let offset = iteration_offset;
                let batch: Vec<_> = (0_u64..batch_size_u64)
                    .map(|index| {
                        data_row(
                            config.seed,
                            fixture.data_stream_id,
                            fixture.specification_id,
                            offset.saturating_add(index),
                        )
                    })
                    .collect();
                iteration_offset = iteration_offset.saturating_add(batch_size_u64);
                runtime.block_on(async {
                    repos
                        .role
                        .insert_data_row_batch(&batch)
                        .await
                        .expect("insert_data_row_batch");
                });
                iterations = iterations.saturating_add(1);
            });
            write_json_summary(
                &config.json_output_path,
                "data_row_batch",
                Some(batch_size),
                iterations,
                iterations.saturating_mul(batch_size_u64),
            );
        });
    }
    group.finish();

    runtime.block_on(teardown_time_series_fixture(&repos.owner, &fixture));
}

/// Bench `metadata_row` inserts at each configured batch size. The
/// metadata-row path is structurally identical to `data_row` except the
/// vector column is absent, so it isolates per-row protocol overhead
/// from vector-encoding cost.
fn bench_metadata_row_inserts(criterion: &mut Criterion, runtime: &Runtime, config: &BenchConfig) {
    let repos = runtime.block_on(BenchRepos::build(config));
    runtime.block_on(warm_pools(&repos, config.seed));
    let fixture = runtime.block_on(seed_time_series_fixture(&repos.owner, config.seed));

    let mut group = criterion.benchmark_group("postgres_metadata_row_insert");
    group.measurement_time(StdDuration::from_secs(15));
    group.sample_size(10);

    // Advance the row index across iterations so every insert lands on a
    // fresh timestamp, for the same Timescale chunk-bloat reason as the
    // `data_row` bench above.
    let mut iteration_offset = 0_u64;
    for &batch_size in &config.metadata_row_batch_sizes {
        group.throughput(Throughput::Elements(batch_size as u64));
        let batch_size_u64 = batch_size as u64;
        group.bench_function(BenchmarkId::from_parameter(batch_size), |bencher| {
            let mut iterations = 0_u64;
            bencher.iter(|| {
                let offset = iteration_offset;
                let batch: Vec<_> = (0_u64..batch_size_u64)
                    .map(|index| {
                        metadata_row(
                            config.seed,
                            fixture.data_stream_id,
                            offset.saturating_add(index),
                        )
                    })
                    .collect();
                iteration_offset = iteration_offset.saturating_add(batch_size_u64);
                runtime.block_on(async {
                    repos
                        .role
                        .insert_metadata_row_batch(&batch)
                        .await
                        .expect("insert_metadata_row_batch");
                });
                iterations = iterations.saturating_add(1);
            });
            write_json_summary(
                &config.json_output_path,
                "metadata_row_batch",
                Some(batch_size),
                iterations,
                iterations.saturating_mul(batch_size_u64),
            );
        });
    }
    group.finish();

    runtime.block_on(teardown_time_series_fixture(&repos.owner, &fixture));
}

/// Bench `annotation` batch inserts at each configured batch size.
fn bench_annotation_inserts(criterion: &mut Criterion, runtime: &Runtime, config: &BenchConfig) {
    let repos = runtime.block_on(BenchRepos::build(config));
    runtime.block_on(warm_pools(&repos, config.seed));
    let parent_platform_id = runtime.block_on(async {
        // Cascade-clean any leftover annotations from a previously
        // panicked run before re-seeding the parent; otherwise the
        // deterministic annotation primary keys collide on the first
        // iteration.
        let leftover_id = platform(config.seed, u64::MAX - 6)
            .id
            .expect("parent platform row carries id");
        let _ = repos.owner.delete_platform(leftover_id).await;
        repos
            .owner
            .upsert_platform(&platform(config.seed, u64::MAX - 6))
            .await
            .expect("seed parent platform for annotation_inserts")
            .id
            .expect("parent platform id")
    });

    let mut group = criterion.benchmark_group("postgres_annotation_insert");
    group.measurement_time(StdDuration::from_secs(15));
    group.sample_size(10);

    let mut iteration_offset = 0_u64;
    for &batch_size in &config.annotation_batch_sizes {
        group.throughput(Throughput::Elements(batch_size as u64));
        let batch_size_u64 = batch_size as u64;
        group.bench_function(BenchmarkId::from_parameter(batch_size), |bencher| {
            let mut iterations = 0_u64;
            bencher.iter(|| {
                let offset = iteration_offset;
                let batch: Vec<_> = (0_u64..batch_size_u64)
                    .map(|index| {
                        annotation(
                            config.seed,
                            parent_platform_id,
                            offset.saturating_add(index),
                        )
                    })
                    .collect();
                iteration_offset = iteration_offset.saturating_add(batch_size_u64);
                runtime.block_on(async {
                    repos
                        .role
                        .insert_annotation_batch(&batch)
                        .await
                        .expect("insert_annotation_batch");
                });
                iterations = iterations.saturating_add(1);
            });
            write_json_summary(
                &config.json_output_path,
                "annotation_batch",
                Some(batch_size),
                iterations,
                iterations.saturating_mul(batch_size_u64),
            );
        });
    }
    group.finish();

    runtime.block_on(async {
        let _ = repos.owner.delete_platform(parent_platform_id).await;
    });
}

/// Bench single-row inserts and upserts on every non-time-series entity.
/// Each variant uses a fresh deterministic UUID per iteration so the
/// insert path (not the upsert-conflict path) is exercised; the matching
/// `upsert_*` call runs immediately after as the second sub-bench so the
/// conflict path is measured against the same row.
fn bench_entity_writes(criterion: &mut Criterion, runtime: &Runtime, config: &BenchConfig) {
    let repos = runtime.block_on(BenchRepos::build(config));
    runtime.block_on(warm_pools(&repos, config.seed));

    // Pre-clean leftover deterministic rows from a previously-panicked
    // run so the first iteration of each single-row entity variant
    // measures a clean INSERT rather than an upsert-conflict or
    // unique-key collision. Entities with a parent row inherit cleanup
    // via `ON DELETE CASCADE` and are not listed here.
    runtime.block_on(async {
        preclean_table_by_label(&repos.owner_pool, "platform", config.seed.seed, "platform").await;
        preclean_table_by_label(&repos.owner_pool, "mission", config.seed.seed, "mission").await;
        preclean_table_by_label(&repos.owner_pool, "ontology", config.seed.seed, "ontology").await;
        preclean_table_by_label(
            &repos.owner_pool,
            "audio_specification",
            config.seed.seed,
            "audio_specification",
        )
        .await;
        preclean_table_by_label(
            &repos.owner_pool,
            "beamgram_specification",
            config.seed.seed,
            "beamgram_specification",
        )
        .await;
        preclean_table_by_label(
            &repos.owner_pool,
            "bearing_time_record_specification",
            config.seed.seed,
            "btr_specification",
        )
        .await;
        preclean_table_by_label(
            &repos.owner_pool,
            "spectrogram_specification",
            config.seed.seed,
            "spectrogram_specification",
        )
        .await;
        preclean_table_by_label(
            &repos.owner_pool,
            "platform_kind",
            config.seed.seed,
            "platform_kind",
        )
        .await;
    });

    let mut group = criterion.benchmark_group("postgres_entity_write");
    group.measurement_time(StdDuration::from_secs(10));
    group.sample_size(20);

    bench_platform(&mut group, runtime, &repos, config);
    bench_data_stream(&mut group, runtime, &repos, config);
    bench_mission(&mut group, runtime, &repos, config);
    bench_ontology(&mut group, runtime, &repos, config);
    bench_ontology_class(&mut group, runtime, &repos, config);
    bench_audio_specification(&mut group, runtime, &repos, config);
    bench_beamgram_specification(&mut group, runtime, &repos, config);
    bench_btr_specification(&mut group, runtime, &repos, config);
    bench_spectrogram_specification(&mut group, runtime, &repos, config);
    bench_annotation(&mut group, runtime, &repos, config);
    bench_platform_information(&mut group, runtime, &repos, config);
    bench_platform_kind(&mut group, runtime, &repos, config);
    bench_mission_platform(&mut group, runtime, &repos, config);
    bench_platform_spec_joins(&mut group, runtime, &repos, config);

    group.finish();
}

fn bench_platform<M: Measurement>(
    group: &mut BenchmarkGroup<'_, M>,
    runtime: &Runtime,
    repos: &BenchRepos,
    config: &BenchConfig,
) {
    let mut counter = 0_u64;
    group.bench_function("platform_upsert", |bencher| {
        bencher.iter(|| {
            counter = counter.wrapping_add(1);
            let row = platform(config.seed, counter);
            runtime.block_on(async {
                repos
                    .role
                    .upsert_platform(&row)
                    .await
                    .expect("upsert_platform");
            });
        });
    });
    // Best-effort cleanup of every row generated above.
    runtime.block_on(async {
        for index in 0..=counter {
            let id = uuid_for(config.seed.seed, &format!("platform:{index}"));
            let _ = repos.owner.delete_platform(id).await;
        }
    });
    write_json_summary(
        &config.json_output_path,
        "platform_upsert",
        None,
        counter,
        counter,
    );
}

fn bench_data_stream<M: Measurement>(
    group: &mut BenchmarkGroup<'_, M>,
    runtime: &Runtime,
    repos: &BenchRepos,
    config: &BenchConfig,
) {
    let parent = runtime.block_on(async {
        repos
            .owner
            .upsert_platform(&platform(config.seed, u64::MAX - 1))
            .await
            .expect("seed parent platform")
            .id
            .expect("parent platform id")
    });
    let mut counter = 0_u64;
    group.bench_function("data_stream_upsert", |bencher| {
        bencher.iter(|| {
            counter = counter.wrapping_add(1);
            let row = data_stream(config.seed, parent, counter);
            runtime.block_on(async {
                repos
                    .role
                    .upsert_data_stream(&row)
                    .await
                    .expect("upsert_data_stream");
            });
        });
    });
    runtime.block_on(async {
        let _ = repos.owner.delete_platform(parent).await;
    });
    write_json_summary(
        &config.json_output_path,
        "data_stream_upsert",
        None,
        counter,
        counter,
    );
}

fn bench_mission<M: Measurement>(
    group: &mut BenchmarkGroup<'_, M>,
    runtime: &Runtime,
    repos: &BenchRepos,
    config: &BenchConfig,
) {
    let mut counter = 0_u64;
    group.bench_function("mission_upsert", |bencher| {
        bencher.iter(|| {
            counter = counter.wrapping_add(1);
            let row = mission(config.seed, counter);
            runtime.block_on(async {
                repos
                    .role
                    .upsert_mission(&row)
                    .await
                    .expect("upsert_mission");
            });
        });
    });
    runtime.block_on(async {
        for index in 0..=counter {
            let id = uuid_for(config.seed.seed, &format!("mission:{index}"));
            let _ = repos.owner.delete_mission(id).await;
        }
    });
    write_json_summary(
        &config.json_output_path,
        "mission_upsert",
        None,
        counter,
        counter,
    );
}

fn bench_ontology<M: Measurement>(
    group: &mut BenchmarkGroup<'_, M>,
    runtime: &Runtime,
    repos: &BenchRepos,
    config: &BenchConfig,
) {
    let mut counter = 0_u64;
    group.bench_function("ontology_upsert", |bencher| {
        bencher.iter(|| {
            counter = counter.wrapping_add(1);
            let row = ontology(config.seed, counter);
            runtime.block_on(async {
                repos
                    .role
                    .upsert_ontology(&row)
                    .await
                    .expect("upsert_ontology");
            });
        });
    });
    runtime.block_on(async {
        for index in 0..=counter {
            let id = uuid_for(config.seed.seed, &format!("ontology:{index}"));
            let _ = repos.owner.delete_ontology(id).await;
        }
    });
    write_json_summary(
        &config.json_output_path,
        "ontology_upsert",
        None,
        counter,
        counter,
    );
}

fn bench_ontology_class<M: Measurement>(
    group: &mut BenchmarkGroup<'_, M>,
    runtime: &Runtime,
    repos: &BenchRepos,
    config: &BenchConfig,
) {
    let parent = runtime.block_on(async {
        repos
            .owner
            .upsert_ontology(&ontology(config.seed, u64::MAX - 1))
            .await
            .expect("seed parent ontology")
            .id
            .expect("parent ontology id")
    });
    let mut counter = 0_u64;
    group.bench_function("ontology_class_upsert", |bencher| {
        bencher.iter(|| {
            counter = counter.wrapping_add(1);
            let row = ontology_class(config.seed, parent, counter);
            runtime.block_on(async {
                repos
                    .role
                    .upsert_ontology_class(&row)
                    .await
                    .expect("upsert_ontology_class");
            });
        });
    });
    runtime.block_on(async {
        let _ = repos.owner.delete_ontology(parent).await;
    });
    write_json_summary(
        &config.json_output_path,
        "ontology_class_upsert",
        None,
        counter,
        counter,
    );
}

fn bench_audio_specification<M: Measurement>(
    group: &mut BenchmarkGroup<'_, M>,
    runtime: &Runtime,
    repos: &BenchRepos,
    config: &BenchConfig,
) {
    let mut counter = 0_u64;
    group.bench_function("audio_specification_upsert", |bencher| {
        bencher.iter(|| {
            counter = counter.wrapping_add(1);
            let row = audio_specification(config.seed, counter);
            runtime.block_on(async {
                repos
                    .role
                    .upsert_audio_specification(&row)
                    .await
                    .expect("upsert_audio_specification");
            });
        });
    });
    runtime.block_on(async {
        for index in 0..=counter {
            let id = uuid_for(config.seed.seed, &format!("audio_specification:{index}"));
            let _ = repos.owner.delete_audio_specification(id).await;
        }
    });
    write_json_summary(
        &config.json_output_path,
        "audio_specification_upsert",
        None,
        counter,
        counter,
    );
}

fn bench_beamgram_specification<M: Measurement>(
    group: &mut BenchmarkGroup<'_, M>,
    runtime: &Runtime,
    repos: &BenchRepos,
    config: &BenchConfig,
) {
    let mut counter = 0_u64;
    group.bench_function("beamgram_specification_upsert", |bencher| {
        bencher.iter(|| {
            counter = counter.wrapping_add(1);
            let row = beamgram_specification(config.seed, counter);
            runtime.block_on(async {
                repos
                    .role
                    .upsert_beamgram_specification(&row)
                    .await
                    .expect("upsert_beamgram_specification");
            });
        });
    });
    runtime.block_on(async {
        for index in 0..=counter {
            let id = uuid_for(config.seed.seed, &format!("beamgram_specification:{index}"));
            let _ = repos.owner.delete_beamgram_specification(id).await;
        }
    });
    write_json_summary(
        &config.json_output_path,
        "beamgram_specification_upsert",
        None,
        counter,
        counter,
    );
}

fn bench_btr_specification<M: Measurement>(
    group: &mut BenchmarkGroup<'_, M>,
    runtime: &Runtime,
    repos: &BenchRepos,
    config: &BenchConfig,
) {
    let mut counter = 0_u64;
    group.bench_function("bearing_time_record_specification_upsert", |bencher| {
        bencher.iter(|| {
            counter = counter.wrapping_add(1);
            let row = bearing_time_record_specification(config.seed, counter);
            runtime.block_on(async {
                repos
                    .role
                    .upsert_bearing_time_record_specification(&row)
                    .await
                    .expect("upsert_bearing_time_record_specification");
            });
        });
    });
    runtime.block_on(async {
        for index in 0..=counter {
            let id = uuid_for(config.seed.seed, &format!("btr_specification:{index}"));
            let _ = repos
                .owner
                .delete_bearing_time_record_specification(id)
                .await;
        }
    });
    write_json_summary(
        &config.json_output_path,
        "bearing_time_record_specification_upsert",
        None,
        counter,
        counter,
    );
}

fn bench_spectrogram_specification<M: Measurement>(
    group: &mut BenchmarkGroup<'_, M>,
    runtime: &Runtime,
    repos: &BenchRepos,
    config: &BenchConfig,
) {
    let mut counter = 0_u64;
    group.bench_function("spectrogram_specification_upsert", |bencher| {
        bencher.iter(|| {
            counter = counter.wrapping_add(1);
            let row = spectrogram_specification(config.seed, counter);
            runtime.block_on(async {
                repos
                    .role
                    .upsert_spectrogram_specification(&row)
                    .await
                    .expect("upsert_spectrogram_specification");
            });
        });
    });
    runtime.block_on(async {
        for index in 0..=counter {
            let id = uuid_for(
                config.seed.seed,
                &format!("spectrogram_specification:{index}"),
            );
            let _ = repos.owner.delete_spectrogram_specification(id).await;
        }
    });
    write_json_summary(
        &config.json_output_path,
        "spectrogram_specification_upsert",
        None,
        counter,
        counter,
    );
}

fn bench_annotation<M: Measurement>(
    group: &mut BenchmarkGroup<'_, M>,
    runtime: &Runtime,
    repos: &BenchRepos,
    config: &BenchConfig,
) {
    let parent_platform_id = runtime.block_on(async {
        // Cascade-clean leftover annotations from a previously panicked run.
        let leftover_id = platform(config.seed, u64::MAX - 3)
            .id
            .expect("parent platform row carries id");
        let _ = repos.owner.delete_platform(leftover_id).await;
        repos
            .owner
            .upsert_platform(&platform(config.seed, u64::MAX - 3))
            .await
            .expect("seed parent platform for annotation")
            .id
            .expect("parent platform id")
    });
    let mut counter = 0_u64;
    group.bench_function("annotation_insert", |bencher| {
        bencher.iter(|| {
            counter = counter.wrapping_add(1);
            let row = annotation(config.seed, parent_platform_id, counter);
            runtime.block_on(async {
                repos
                    .role
                    .insert_annotation(&row)
                    .await
                    .expect("insert_annotation");
            });
        });
    });
    runtime.block_on(async {
        let _ = repos.owner.delete_platform(parent_platform_id).await;
    });
    write_json_summary(
        &config.json_output_path,
        "annotation_insert",
        None,
        counter,
        counter,
    );
}

fn bench_platform_information<M: Measurement>(
    group: &mut BenchmarkGroup<'_, M>,
    runtime: &Runtime,
    repos: &BenchRepos,
    config: &BenchConfig,
) {
    let parent_platform_id = runtime.block_on(async {
        // Cascade-clean leftover platform_information rows from a
        // previously panicked run.
        let leftover_id = platform(config.seed, u64::MAX - 5)
            .id
            .expect("parent platform row carries id");
        let _ = repos.owner.delete_platform(leftover_id).await;
        repos
            .owner
            .upsert_platform(&platform(config.seed, u64::MAX - 5))
            .await
            .expect("seed parent platform for platform_information")
            .id
            .expect("parent platform id")
    });
    let mut counter = 0_u64;
    group.bench_function("platform_information_insert", |bencher| {
        bencher.iter(|| {
            counter = counter.wrapping_add(1);
            let row = platform_information(config.seed, parent_platform_id, counter);
            runtime.block_on(async {
                repos
                    .role
                    .insert_platform_information(&row)
                    .await
                    .expect("insert_platform_information");
            });
        });
    });
    runtime.block_on(async {
        let _ = repos.owner.delete_platform(parent_platform_id).await;
    });
    write_json_summary(
        &config.json_output_path,
        "platform_information_insert",
        None,
        counter,
        counter,
    );
}

fn bench_platform_kind<M: Measurement>(
    group: &mut BenchmarkGroup<'_, M>,
    runtime: &Runtime,
    repos: &BenchRepos,
    config: &BenchConfig,
) {
    let mut counter = 0_u64;
    group.bench_function("platform_kind_insert", |bencher| {
        bencher.iter(|| {
            counter = counter.wrapping_add(1);
            let row = platform_kind(config.seed, counter);
            runtime.block_on(async {
                repos
                    .role
                    .insert_platform_kind(&row)
                    .await
                    .expect("insert_platform_kind");
            });
        });
    });
    runtime.block_on(async {
        for index in 0..=counter {
            let id = uuid_for(config.seed.seed, &format!("platform_kind:{index}"));
            let _ = repos.owner.delete_platform_kind(id).await;
        }
    });
    write_json_summary(
        &config.json_output_path,
        "platform_kind_insert",
        None,
        counter,
        counter,
    );
}

fn bench_mission_platform<M: Measurement>(
    group: &mut BenchmarkGroup<'_, M>,
    runtime: &Runtime,
    repos: &BenchRepos,
    config: &BenchConfig,
) {
    let (parent_mission_id, parent_platform_id) = runtime.block_on(async {
        // Cascade-clean any leftover mission_platform rows from a
        // previously panicked run by deleting either parent. Both
        // parents key the deterministic mission_platform id, so wiping
        // either side is enough; we wipe both for symmetry.
        let leftover_mission = mission(config.seed, u64::MAX - 4)
            .id
            .expect("parent mission row carries id");
        let leftover_platform = platform(config.seed, u64::MAX - 4)
            .id
            .expect("parent platform row carries id");
        let _ = repos.owner.delete_mission(leftover_mission).await;
        let _ = repos.owner.delete_platform(leftover_platform).await;
        let mission_id = repos
            .owner
            .upsert_mission(&mission(config.seed, u64::MAX - 4))
            .await
            .expect("seed parent mission for mission_platform")
            .id
            .expect("parent mission id");
        let platform_id = repos
            .owner
            .upsert_platform(&platform(config.seed, u64::MAX - 4))
            .await
            .expect("seed parent platform for mission_platform")
            .id
            .expect("parent platform id");
        (mission_id, platform_id)
    });
    let mut counter = 0_u64;
    group.bench_function("mission_platform_insert", |bencher| {
        bencher.iter(|| {
            counter = counter.wrapping_add(1);
            let row = mission_platform(config.seed, parent_mission_id, parent_platform_id, counter);
            runtime.block_on(async {
                repos
                    .role
                    .insert_mission_platform(&row)
                    .await
                    .expect("insert_mission_platform");
            });
        });
    });
    runtime.block_on(async {
        let _ = repos.owner.delete_platform(parent_platform_id).await;
        let _ = repos.owner.delete_mission(parent_mission_id).await;
    });
    write_json_summary(
        &config.json_output_path,
        "mission_platform_insert",
        None,
        counter,
        counter,
    );
}

fn bench_platform_spec_joins<M: Measurement>(
    group: &mut BenchmarkGroup<'_, M>,
    runtime: &Runtime,
    repos: &BenchRepos,
    config: &BenchConfig,
) {
    let (platform_id, audio_spec_id, beamgram_spec_id, btr_spec_id, spectrogram_spec_id) = runtime
        .block_on(async {
            let platform_id = repos
                .owner
                .upsert_platform(&platform(config.seed, u64::MAX - 2))
                .await
                .expect("seed parent platform")
                .id
                .expect("platform id");
            let audio_id = repos
                .owner
                .upsert_audio_specification(&audio_specification(config.seed, u64::MAX - 2))
                .await
                .expect("seed audio spec")
                .id
                .expect("audio spec id");
            let beamgram_id = repos
                .owner
                .upsert_beamgram_specification(&beamgram_specification(config.seed, u64::MAX - 2))
                .await
                .expect("seed beamgram spec")
                .id
                .expect("beamgram spec id");
            let btr_id = repos
                .owner
                .upsert_bearing_time_record_specification(&bearing_time_record_specification(
                    config.seed,
                    u64::MAX - 2,
                ))
                .await
                .expect("seed btr spec")
                .id
                .expect("btr spec id");
            let spectrogram_id = repos
                .owner
                .upsert_spectrogram_specification(&spectrogram_specification(
                    config.seed,
                    u64::MAX - 2,
                ))
                .await
                .expect("seed spectrogram spec")
                .id
                .expect("spectrogram spec id");
            (platform_id, audio_id, beamgram_id, btr_id, spectrogram_id)
        });

    // Clean up any leftover join rows from a previously-panicked run
    // that shared the same deterministic FK pairs. Without this, the
    // first iteration of each variant (where `previous_*_join_id` is
    // None) would collide with the leftover row on the
    // (platform_id, spec_id) unique constraint. Bench-only raw SQL
    // because the SDK exposes delete-by-id, not delete-by-FK-pair.
    runtime.block_on(async {
        let _ = sqlx::query(
            "DELETE FROM horizon_public.platform_audio_specification \
             WHERE platform_id = $1 AND audio_specification_id = $2",
        )
        .bind(platform_id)
        .bind(audio_spec_id)
        .execute(&repos.owner_pool)
        .await;
        let _ = sqlx::query(
            "DELETE FROM horizon_public.platform_beamgram_specification \
             WHERE platform_id = $1 AND beamgram_specification_id = $2",
        )
        .bind(platform_id)
        .bind(beamgram_spec_id)
        .execute(&repos.owner_pool)
        .await;
        let _ = sqlx::query(
            "DELETE FROM horizon_public.platform_bearing_time_record_specification \
             WHERE platform_id = $1 AND bearing_time_record_specification_id = $2",
        )
        .bind(platform_id)
        .bind(btr_spec_id)
        .execute(&repos.owner_pool)
        .await;
        let _ = sqlx::query(
            "DELETE FROM horizon_public.platform_spectrogram_specification \
             WHERE platform_id = $1 AND spectrogram_specification_id = $2",
        )
        .bind(platform_id)
        .bind(spectrogram_spec_id)
        .execute(&repos.owner_pool)
        .await;
    });

    let mut counter = 0_u64;
    let mut previous_audio_join_id: Option<Uuid> = None;
    group.bench_function("platform_audio_specification_insert", |bencher| {
        bencher.iter(|| {
            counter = counter.wrapping_add(1);
            let row =
                platform_audio_specification(config.seed, platform_id, audio_spec_id, counter);
            runtime.block_on(async {
                if let Some(prev_id) = previous_audio_join_id {
                    let _ = repos
                        .role
                        .delete_platform_audio_specification(prev_id)
                        .await;
                }
                let inserted = repos
                    .role
                    .insert_platform_audio_specification(&row)
                    .await
                    .expect("insert_platform_audio_specification");
                previous_audio_join_id = inserted.id;
            });
        });
    });

    let mut beamgram_counter = 0_u64;
    let mut previous_beamgram_join_id: Option<Uuid> = None;
    group.bench_function("platform_beamgram_specification_insert", |bencher| {
        bencher.iter(|| {
            beamgram_counter = beamgram_counter.wrapping_add(1);
            let row = platform_beamgram_specification(
                config.seed,
                platform_id,
                beamgram_spec_id,
                beamgram_counter,
            );
            runtime.block_on(async {
                if let Some(prev_id) = previous_beamgram_join_id {
                    let _ = repos
                        .role
                        .delete_platform_beamgram_specification(prev_id)
                        .await;
                }
                let inserted = repos
                    .role
                    .insert_platform_beamgram_specification(&row)
                    .await
                    .expect("insert_platform_beamgram_specification");
                previous_beamgram_join_id = inserted.id;
            });
        });
    });

    let mut btr_counter = 0_u64;
    let mut previous_btr_join_id: Option<Uuid> = None;
    group.bench_function(
        "platform_bearing_time_record_specification_insert",
        |bencher| {
            bencher.iter(|| {
                btr_counter = btr_counter.wrapping_add(1);
                let row = platform_bearing_time_record_specification(
                    config.seed,
                    platform_id,
                    btr_spec_id,
                    btr_counter,
                );
                runtime.block_on(async {
                    if let Some(prev_id) = previous_btr_join_id {
                        let _ = repos
                            .role
                            .delete_platform_bearing_time_record_specification(prev_id)
                            .await;
                    }
                    let inserted = repos
                        .role
                        .insert_platform_bearing_time_record_specification(&row)
                        .await
                        .expect("insert_platform_bearing_time_record_specification");
                    previous_btr_join_id = inserted.id;
                });
            });
        },
    );

    let mut spectrogram_counter = 0_u64;
    let mut previous_spectrogram_join_id: Option<Uuid> = None;
    group.bench_function("platform_spectrogram_specification_insert", |bencher| {
        bencher.iter(|| {
            spectrogram_counter = spectrogram_counter.wrapping_add(1);
            let row = platform_spectrogram_specification(
                config.seed,
                platform_id,
                spectrogram_spec_id,
                spectrogram_counter,
            );
            runtime.block_on(async {
                if let Some(prev_id) = previous_spectrogram_join_id {
                    let _ = repos
                        .role
                        .delete_platform_spectrogram_specification(prev_id)
                        .await;
                }
                let inserted = repos
                    .role
                    .insert_platform_spectrogram_specification(&row)
                    .await
                    .expect("insert_platform_spectrogram_specification");
                previous_spectrogram_join_id = inserted.id;
            });
        });
    });

    runtime.block_on(async {
        let _ = repos.owner.delete_platform(platform_id).await;
        let _ = repos.owner.delete_audio_specification(audio_spec_id).await;
        let _ = repos
            .owner
            .delete_beamgram_specification(beamgram_spec_id)
            .await;
        let _ = repos
            .owner
            .delete_bearing_time_record_specification(btr_spec_id)
            .await;
        let _ = repos
            .owner
            .delete_spectrogram_specification(spectrogram_spec_id)
            .await;
    });

    write_json_summary(
        &config.json_output_path,
        "platform_audio_specification_insert",
        None,
        counter,
        counter,
    );
    write_json_summary(
        &config.json_output_path,
        "platform_beamgram_specification_insert",
        None,
        beamgram_counter,
        beamgram_counter,
    );
    write_json_summary(
        &config.json_output_path,
        "platform_bearing_time_record_specification_insert",
        None,
        btr_counter,
        btr_counter,
    );
    write_json_summary(
        &config.json_output_path,
        "platform_spectrogram_specification_insert",
        None,
        spectrogram_counter,
        spectrogram_counter,
    );
}

/// Top-level criterion entry point. Reads env, builds a single Tokio
/// runtime shared across every variant, and dispatches to each
/// per-entity bench function.
fn run_benchmarks(criterion: &mut Criterion) {
    let config = BenchConfig::from_env();
    let runtime = tokio::runtime::Builder::new_multi_thread()
        .enable_all()
        .build()
        .expect("build tokio runtime");

    bench_data_row_inserts(criterion, &runtime, &config);
    bench_metadata_row_inserts(criterion, &runtime, &config);
    bench_annotation_inserts(criterion, &runtime, &config);
    bench_entity_writes(criterion, &runtime, &config);
}

criterion_group!(name = postgres_write_throughput; config = Criterion::default(); targets = run_benchmarks);
criterion_main!(postgres_write_throughput);
