use dbsp::{Runtime, utils::Tup1};
use feldera_sqllib::Variant;
use feldera_types::{
    deserialize_table_record,
    program_schema::{Relation, SqlIdentifier},
    serde_with_context::{SerializeWithContext, SqlSerdeConfig},
    serialize_table_record,
    transport::postgres::{PostgresTlsConfig, PostgresWriteMode},
};
use pg::PostgresTestStruct;
use serde_json::json;
use serial_test::serial;
use std::{collections::BTreeMap, io::Write, path::Path};
use tempfile::NamedTempFile;

use crate::{
    Catalog, Controller,
    integrated::postgres::test::pg::PostgresTestStructCdc,
    test::{TestStruct, wait},
};

fn postgres_url() -> String {
    std::env::var("POSTGRES_URL")
        .unwrap_or("postgres://postgres:password@localhost:5432".to_string())
}

/// Returns the SSL connection URL and TLS config if the SSL environment
/// variables are set (`POSTGRES_SSL_URL`, `POSTGRES_SSL_CA_LOCATION`,
/// `POSTGRES_SSL_CLIENT_LOCATION`, `POSTGRES_SSL_CLIENT_KEY_LOCATION`).
fn postgres_ssl_config() -> (String, PostgresTlsConfig) {
    let url = std::env::var("POSTGRES_SSL_URL")
        .unwrap_or("postgres://postgres:secret@localhost:5432/postgres".to_string());
    let ssl_ca_pem = std::env::var("POSTGRES_SSL_CA_PEM").ok();
    let ssl_ca_location = std::env::var("POSTGRES_SSL_CA_LOCATION").ok();
    let ssl_client_pem = std::env::var("POSTGRES_SSL_CLIENT_CERT_PEM").ok();
    let ssl_client_location = std::env::var("POSTGRES_SSL_CLIENT_LOCATION").ok();
    let ssl_client_key = std::env::var("POSTGRES_SSL_CLIENT_KEY_PEM").ok();
    let ssl_client_key_location = std::env::var("POSTGRES_SSL_CLIENT_KEY_LOCATION").ok();
    let ssl_certificate_chain_location =
        std::env::var("POSTGRES_SSL_CERTIFICATE_CHAIN_LOCATION").ok();
    let verify_hostname = std::env::var("POSTGRES_SSL_VERIFY_HOSTNAME")
        .ok()
        .and_then(|s| s.parse::<bool>().ok());

    (
        url,
        PostgresTlsConfig {
            ssl_ca_location,
            ssl_ca_pem,
            ssl_client_location,
            ssl_client_key_location,
            ssl_client_pem,
            ssl_client_key,
            ssl_certificate_chain_location,
            verify_hostname,
        },
    )
}

mod pg {
    use std::collections::BTreeMap;

    use chrono::SubsecRound;
    use dbsp::{Runtime, utils::Tup1};
    use feldera_macros::IsNone;
    use feldera_sqllib::{F32, F64, SqlDecimal, SqlString, Variant};
    use feldera_types::{
        config::PipelineConfig,
        deserialize_table_record,
        program_schema::{ColumnType, Field, Relation, SqlIdentifier},
        serialize_table_record,
        transport::postgres::PostgresTlsConfig,
    };
    use postgres::{NoTls, Row};
    use rand::{Rng, distributions::Standard, prelude::Distribution};

    use crate::{
        Catalog, Controller, integrated::postgres::tls::make_tls_connector, test::TestStruct,
    };

    #[derive(Debug, postgres_types::ToSql, postgres_types::FromSql)]
    #[postgres(name = "test_struct")]
    struct TempTestStruct {
        id: Option<i32>,
        i: Option<i64>,
        b: Option<bool>,
        s: Option<String>,
    }

    impl From<TempTestStruct> for TestStruct {
        fn from(val: TempTestStruct) -> Self {
            TestStruct {
                id: val.id.unwrap() as u32,
                i: val.i,
                b: val.b.unwrap(),
                s: val.s.unwrap(),
            }
        }
    }

    #[derive(
        Debug,
        Default,
        PartialEq,
        Eq,
        PartialOrd,
        Ord,
        Clone,
        Hash,
        size_of::SizeOf,
        rkyv::Archive,
        rkyv::Serialize,
        rkyv::Deserialize,
        IsNone,
    )]
    #[archive_attr(derive(Ord, Eq, PartialEq, PartialOrd))]
    pub(super) struct PostgresTestStruct {
        pub boolean_: bool,
        pub tinyint_: i8,
        pub smallint_: i16,
        pub int_: i32,
        pub bigint_: i64,
        pub decimal_: SqlDecimal<38, 10>,
        pub float_: F32,
        pub double_: F64,
        pub varchar_: SqlString,
        pub time_: feldera_sqllib::Time,
        pub date_: feldera_sqllib::Date,
        pub timestamp_: feldera_sqllib::Timestamp,
        pub variant_: feldera_sqllib::Variant,
        pub uuid_: feldera_sqllib::Uuid,
        pub varbinary_: feldera_sqllib::ByteArray,
        pub struct_: TestStruct,
        pub string_array_: Vec<feldera_sqllib::SqlString>,
        pub struct_array_: Vec<TestStruct>,
        pub map_: BTreeMap<feldera_sqllib::SqlString, TestStruct>,
    }

    #[derive(Debug, Default, PartialEq, Eq, PartialOrd, Ord, Clone)]
    pub(super) struct PostgresTestStructCdc {
        pub boolean_: bool,
        pub tinyint_: i8,
        pub smallint_: i16,
        pub int_: i32,
        pub bigint_: i64,
        pub decimal_: SqlDecimal<38, 10>,
        pub float_: F32,
        pub double_: F64,
        pub varchar_: SqlString,
        pub time_: feldera_sqllib::Time,
        pub date_: feldera_sqllib::Date,
        pub timestamp_: feldera_sqllib::Timestamp,
        pub variant_: feldera_sqllib::Variant,
        pub uuid_: feldera_sqllib::Uuid,
        pub varbinary_: feldera_sqllib::ByteArray,
        pub struct_: TestStruct,
        pub string_array_: Vec<feldera_sqllib::SqlString>,
        pub struct_array_: Vec<TestStruct>,
        pub map_: BTreeMap<feldera_sqllib::SqlString, TestStruct>,
        pub __feldera_op: String,
        pub __feldera_ts: feldera_sqllib::Timestamp,
    }

    impl PartialEq<PostgresTestStruct> for PostgresTestStructCdc {
        fn eq(&self, other: &PostgresTestStruct) -> bool {
            self.boolean_ == other.boolean_
                && self.tinyint_ == other.tinyint_
                && self.smallint_ == other.smallint_
                && self.int_ == other.int_
                && self.bigint_ == other.bigint_
                && self.decimal_ == other.decimal_
                && self.float_ == other.float_
                && self.double_ == other.double_
                && self.varchar_ == other.varchar_
                && self.time_ == other.time_
                && self.date_ == other.date_
                && self.timestamp_ == other.timestamp_
                && self.variant_ == other.variant_
                && self.uuid_ == other.uuid_
                && self.varbinary_ == other.varbinary_
                && self.struct_ == other.struct_
                && self.string_array_ == other.string_array_
                && self.struct_array_ == other.struct_array_
                && self.map_ == other.map_
        }
    }

    serialize_table_record!(PostgresTestStruct[19]{
        boolean_["boolean_"]: bool,
        tinyint_["tinyint_"]: i8,
        smallint_["smallint_"]: i16,
        int_["int_"]: i32,
        bigint_["bigint_"]: i64,
        decimal_["decimal_"]: SqlDecimal,
        float_["float_"]: F32,
        double_["double_"]: F64,
        varchar_["varchar_"]: SqlString,
        time_["time_"]: feldera_sqllib::Time,
        date_["date_"]: feldera_sqllib::Date,
        timestamp_["timestamp_"]: feldera_sqllib::Timestamp,
        variant_["variant_"]: feldera_sqllib::Variant,
        uuid_["uuid_"]: feldera_sqllib::Uuid,
        varbinary_["varbinary_"]: feldera_sqllib::ByteArray,
        struct_["struct_"]: TestStruct,
        string_array_["string_array_"]: Vec<feldera_sqllib::SqlString>,
        struct_array_["struct_array_"]: Vec<TestStruct>,
        map_["map_"]: BTreeMap<feldera_sqllib::SqlString, TestStruct>
    });

    deserialize_table_record!(PostgresTestStruct["PostgresTestStruct", Variant, 19] {
        (boolean_, "boolean_", false, bool, |_| None),
        (tinyint_, "tinyint_", false, i8, |_| None),
        (smallint_, "smallint_", false, i16, |_| None),
        (int_, "int_", false, i32, |_| None),
        (bigint_, "bigint_", false, i64, |_| None),
        (decimal_, "decimal_", false, SqlDecimal<38, 10>, |_| None),
        (float_, "float_", false, F32, |_| None),
        (double_, "double_", false, F64, |_| None),
        (varchar_, "varchar_", false, SqlString, |_| None),
        (time_, "time_", false, feldera_sqllib::Time, |_| None),
        (date_, "date_", false, feldera_sqllib::Date, |_| None),
        (timestamp_, "timestamp_", false, feldera_sqllib::Timestamp, |_| None),
        (variant_, "variant_", false, feldera_sqllib::Variant, |_| None),
        (uuid_, "uuid_", false, feldera_sqllib::Uuid, |_| None),
        (varbinary_, "varbinary_", false, feldera_sqllib::ByteArray, |_| None),
        (struct_, "struct_", false, TestStruct, |_| None),
        (string_array_, "string_array_", false, Vec<feldera_sqllib::SqlString>, |_| None),
        (struct_array_, "struct_array_", false, Vec<TestStruct>, |_| None),
        (map_, "map_", false, BTreeMap<feldera_sqllib::SqlString, TestStruct>, |_| None)
    });

    pub(super) struct TempPgTable {
        client: postgres::Client,
        name: String,
    }

    /// Connects a sync postgres client, with TLS if config is provided.
    pub(super) fn pg_connect(uri: &str, tls: &Option<PostgresTlsConfig>) -> postgres::Client {
        if let Some(tls) = tls {
            let connector = make_tls_connector(tls, "test")
                .expect("failed to build TLS connector")
                .expect("TLS config should produce a connector");
            postgres::Client::connect(uri, connector).expect("failed to connect with TLS")
        } else {
            postgres::Client::connect(uri, NoTls).expect("failed to connect")
        }
    }

    impl TempPgTable {
        fn new(name: &str, uri: String, pk: bool, tls: &Option<PostgresTlsConfig>) -> Self {
            let mut client = pg_connect(&uri, tls);

            let pk = if pk { "PRIMARY KEY" } else { "" };

            client
                .execute("DROP TYPE IF EXISTS test_struct", &[])
                .unwrap();

            client
                .execute(
                    r#"CREATE TYPE test_struct AS (
    id INTEGER,
    b BOOL,
    i BIGINT,
    s VARCHAR
)"#,
                    &[],
                )
                .unwrap();

            client
                .execute(
                    &format!(
                        r#"
CREATE TABLE {name} (
    boolean_      BOOLEAN,
    tinyint_      SMALLINT,
    smallint_     SMALLINT,
    int_          INTEGER,
    bigint_       BIGINT {pk},
    decimal_      DECIMAL,
    float_        REAL,
    double_       DOUBLE PRECISION,
    varchar_      TEXT,
    time_         TIME,
    date_         DATE,
    timestamp_    TIMESTAMP,
    variant_      JSONB,
    uuid_         UUID,
    varbinary_    BYTEA,
    string_array_ VARCHAR ARRAY,
    struct_       test_struct,
    struct_array_ test_struct ARRAY,
    map_          JSONB,
    __feldera_op  CHAR,
    __feldera_ts  BIGINT
)"#
                    ),
                    &[],
                )
                .unwrap_or_else(|_| panic!("failed to create table {name}"));

            TempPgTable {
                client,
                name: name.to_owned(),
            }
        }

        pub fn query(&mut self) -> Vec<Row> {
            self.client
                .query(
                    &format!(
                        "SELECT *, CAST(decimal_ AS VARCHAR) AS decimal_str FROM {}",
                        self.name
                    ),
                    &[],
                )
                .expect("failed to query table")
        }
    }

    impl Drop for TempPgTable {
        fn drop(&mut self) {
            self.client
                .execute(&format!("DROP TABLE {}", self.name), &[])
                .unwrap_or_else(|_| panic!("failed to drop table {}", self.name));

            self.client
                .execute("DROP TYPE test_struct", &[])
                .unwrap_or_else(|_| panic!("failed to drop type test_struct"));
        }
    }

    impl From<Row> for PostgresTestStruct {
        fn from(r: Row) -> Self {
            PostgresTestStruct {
                boolean_: r.get("boolean_"),
                tinyint_: r.get::<_, i16>("tinyint_") as i8,
                smallint_: r.get("smallint_"),
                int_: r.get("int_"),
                bigint_: r.get("bigint_"),
                decimal_: r
                    .get::<_, String>("decimal_str")
                    .parse::<SqlDecimal<38, 10>>()
                    .unwrap(),
                float_: F32::new(r.get("float_")),
                double_: F64::new(r.get("double_")),
                varchar_: SqlString::from_ref(r.get("varchar_")),
                time_: feldera_sqllib::Time::from_time(r.get("time_")),
                date_: feldera_sqllib::Date::from_date(r.get("date_")),
                timestamp_: feldera_sqllib::Timestamp::from_naiveDateTime(r.get("timestamp_")),
                variant_: feldera_sqllib::from_json_string(
                    &r.get::<_, serde_json::Value>("variant_").to_string(),
                )
                .unwrap(),
                uuid_: r.get::<_, uuid::Uuid>("uuid_").into(),
                varbinary_: feldera_sqllib::ByteArray::from_vec(r.get::<_, Vec<u8>>("varbinary_")),
                struct_: { r.get::<_, TempTestStruct>("struct_").into() },
                string_array_: r
                    .get::<_, Vec<String>>("string_array_")
                    .into_iter()
                    .map(|s| s.into())
                    .collect(),
                struct_array_: r
                    .get::<_, Vec<TempTestStruct>>("struct_array_")
                    .into_iter()
                    .map(TestStruct::from)
                    .collect(),
                map_: serde_json::from_value(r.get::<_, serde_json::Value>("map_")).unwrap(),
            }
        }
    }

    impl From<Row> for PostgresTestStructCdc {
        fn from(r: Row) -> Self {
            PostgresTestStructCdc {
                boolean_: r.get("boolean_"),
                tinyint_: r.get::<_, i16>("tinyint_") as i8,
                smallint_: r.get("smallint_"),
                int_: r.get("int_"),
                bigint_: r.get("bigint_"),
                decimal_: r
                    .get::<_, String>("decimal_str")
                    .parse::<SqlDecimal<38, 10>>()
                    .unwrap(),
                float_: F32::new(r.get("float_")),
                double_: F64::new(r.get("double_")),
                varchar_: SqlString::from_ref(r.get("varchar_")),
                time_: feldera_sqllib::Time::from_time(r.get("time_")),
                date_: feldera_sqllib::Date::from_date(r.get("date_")),
                timestamp_: feldera_sqllib::Timestamp::from_naiveDateTime(r.get("timestamp_")),
                variant_: feldera_sqllib::from_json_string(
                    &r.get::<_, serde_json::Value>("variant_").to_string(),
                )
                .unwrap(),
                uuid_: r.get::<_, uuid::Uuid>("uuid_").into(),
                varbinary_: feldera_sqllib::ByteArray::from_vec(r.get::<_, Vec<u8>>("varbinary_")),
                struct_: { r.get::<_, TempTestStruct>("struct_").into() },
                string_array_: r
                    .get::<_, Vec<String>>("string_array_")
                    .into_iter()
                    .map(|s| s.into())
                    .collect(),
                struct_array_: r
                    .get::<_, Vec<TempTestStruct>>("struct_array_")
                    .into_iter()
                    .map(TestStruct::from)
                    .collect(),
                map_: serde_json::from_value(r.get::<_, serde_json::Value>("map_")).unwrap(),
                __feldera_op: r.get("__feldera_op"),
                __feldera_ts: feldera_sqllib::Timestamp::from_dateTime(
                    chrono::DateTime::from_timestamp_micros(r.get::<_, i64>("__feldera_ts"))
                        .unwrap(),
                ),
            }
        }
    }

    impl PostgresTestStruct {
        pub fn schema() -> Vec<Field> {
            vec![
                Field::new("boolean_".into(), ColumnType::boolean(false)),
                Field::new("tinyint_".into(), ColumnType::tinyint(false)),
                Field::new("smallint_".into(), ColumnType::smallint(false)),
                Field::new("int_".into(), ColumnType::int(false)),
                Field::new("bigint_".into(), ColumnType::bigint(false)),
                Field::new("decimal_".into(), ColumnType::decimal(10, 3, false)),
                Field::new("float_".into(), ColumnType::real(false)),
                Field::new("double_".into(), ColumnType::double(false)),
                Field::new("varchar_".into(), ColumnType::varchar(false)),
                Field::new("time_".into(), ColumnType::time(false)),
                Field::new("date_".into(), ColumnType::date(false)),
                Field::new("timestamp_".into(), ColumnType::timestamp(false)),
                Field::new("variant_".into(), ColumnType::variant(false)),
                Field::new("uuid_".into(), ColumnType::uuid(false)),
                Field::new("varbinary_".into(), ColumnType::varbinary(false)),
                Field::new(
                    "struct_".into(),
                    ColumnType::structure(false, &TestStruct::schema()),
                ),
                Field::new(
                    "string_array_".into(),
                    ColumnType::array(false, ColumnType::varchar(false)),
                ),
                Field::new(
                    "struct_array_".into(),
                    ColumnType::array(false, ColumnType::structure(false, &TestStruct::schema())),
                ),
                Field::new(
                    "map_".into(),
                    ColumnType::map(
                        false,
                        ColumnType::varchar(false),
                        ColumnType::structure(false, &TestStruct::schema()),
                    ),
                ),
            ]
        }

        pub fn create_table(
            name: &str,
            uri: String,
            pk: bool,
            tls: &Option<PostgresTlsConfig>,
        ) -> TempPgTable {
            TempPgTable::new(name, uri, pk, tls)
        }

        pub fn test_circuit(
            config: PipelineConfig,
        ) -> (Controller, crossbeam::channel::Receiver<String>) {
            let schema = PostgresTestStruct::schema();
            let (err_sender, err_receiver) = crossbeam::channel::unbounded();
            let controller = Controller::with_test_config(
                move |workers| {
                    Ok({
                        let (circuit, catalog) = Runtime::init_circuit(workers, move |circuit| {
                            let mut catalog = Catalog::new();
                            let (input, hinput) = circuit.add_input_zset::<PostgresTestStruct>();

                            let input_schema = serde_json::to_string(&Relation::new(
                                "test_input1".into(),
                                schema.clone(),
                                false,
                                BTreeMap::new(),
                            ))
                            .unwrap();

                            let output_schema = serde_json::to_string(&Relation::new(
                                "test_output1".into(),
                                schema,
                                false,
                                BTreeMap::new(),
                            ))
                            .unwrap();

                            catalog.register_materialized_input_zset::<_, PostgresTestStruct>
                            (
                                input.clone(),
                                hinput,
                                &input_schema,
                            );

                            #[derive(Clone, Debug, Eq, PartialEq, Default)]
                            pub struct KeyStruct {
                                field0: i64,
                            }

                            impl From<KeyStruct> for Tup1<i64> {
                                fn from(t: KeyStruct) -> Self {
                                    Tup1::new(t.field0)
                                }
                            }
                            impl From<Tup1<i64>> for KeyStruct {
                                fn from(t: Tup1<i64>) -> Self {
                                    Self { field0: t.0 }
                                }
                            }

                            deserialize_table_record!(KeyStruct["idx", Variant, 1] {
                                (field0, "bigint_", false, i64, |_| None)
                            });
                            serialize_table_record!(KeyStruct[1]{
                                field0["bigint_"]: i64
                            });

                            let indexed_input = input
                                .map_index(|r| (Tup1(r.bigint_), r.to_owned()));
                            catalog.register_materialized_output_zset::<_, PostgresTestStruct>(input, &output_schema);

                            catalog
                                .register_index::<Tup1<i64>, KeyStruct, PostgresTestStruct, PostgresTestStruct>(
                                    indexed_input.clone(),
                                    &SqlIdentifier::from("idx"),
                                    &SqlIdentifier::from("test_output1"),
                                    &[&SqlIdentifier::from("bigint_")],
                                )
                                .expect("failed to register index");

                            Ok(catalog)
                        })
                        .unwrap();
                        (circuit, Box::new(catalog))
                    })
                },
                &config,
                Box::new(move |e, _| {
                    let msg = format!("postgres_output_test: error: {}", e);
                    println!("{msg}");
                    err_sender.send(msg).unwrap()
                }),
            )
            .unwrap();

            (controller, err_receiver)
        }
    }

    impl Distribution<PostgresTestStruct> for Standard {
        fn sample<R: Rng + ?Sized>(&self, rng: &mut R) -> PostgresTestStruct {
            PostgresTestStruct {
                boolean_: rng.r#gen(),
                tinyint_: rng.r#gen(),
                smallint_: rng.r#gen(),
                int_: rng.r#gen(),
                bigint_: rng.r#gen(),
                decimal_: SqlDecimal::<38, 10>::new(rng.gen_range::<i128, _>(-100..100), 3)
                    .unwrap(),
                float_: F32::new((rng.r#gen::<f32>() * 1000.0).trunc() / 1000.0),
                double_: F64::new((rng.r#gen::<f64>() * 1000.0).trunc() / 1000.0),
                varchar_: rng.r#gen::<u32>().to_string().into(),
                time_: feldera_sqllib::Time::from_time(
                    chrono::Utc::now().naive_utc().time().round_subsecs(5),
                ),
                date_: feldera_sqllib::Date::from_date(chrono::Utc::now().date_naive()),
                timestamp_: feldera_sqllib::Timestamp::from_naiveDateTime(
                    chrono::Utc::now().naive_utc(),
                ),
                variant_: feldera_sqllib::Variant::String(rng.r#gen::<u32>().to_string().into()),
                uuid_: uuid::Uuid::new_v4().into(),
                varbinary_: feldera_sqllib::ByteArray::from_vec(vec![rng.gen_range(0..127)]),
                struct_: rng.r#gen(),
                string_array_: vec![],
                struct_array_: vec![rng.r#gen()],
                map_: {
                    let mut map = BTreeMap::new();
                    map.insert(rng.r#gen::<u32>().to_string().into(), rng.r#gen());
                    map
                },
            }
        }
    }
}

fn test_pg_on_conflict(on_conflict_do_nothing: bool, mode: PostgresWriteMode) {
    let table_name = "test_pg_on_conflict";
    let url = postgres_url();
    let max_buffer_size_bytes = 1024;

    let mut data: Vec<PostgresTestStruct> = (0..10000).map(|_| rand::random()).collect();
    let mut insert_file = NamedTempFile::new().unwrap();

    for datum in data.iter() {
        let buffer: Vec<u8> = Vec::new();
        let mut serializer = serde_json::Serializer::new(buffer);
        datum
            .serialize_with_context(&mut serializer, &SqlSerdeConfig::default())
            .unwrap();
        insert_file
            .as_file_mut()
            .write_all(&serializer.into_inner())
            .unwrap();
        insert_file.write_all(b"\n").unwrap();
    }

    let mut upsert_data = data
        .clone()
        .into_iter()
        .map(|mut d| {
            d.varchar_ = "updated".into();
            d
        })
        .collect::<Vec<_>>();

    let mut upsert_file = NamedTempFile::new().unwrap();

    for new in upsert_data.iter() {
        let buffer: Vec<u8> = Vec::new();
        let mut serializer = serde_json::Serializer::new(buffer);
        new.serialize_with_context(&mut serializer, &SqlSerdeConfig::default())
            .unwrap();
        upsert_file
            .as_file_mut()
            .write_all(&serializer.into_inner())
            .unwrap();
    }

    let url_clone = url.clone();

    let get_config = |in_file: &Path| {
        serde_json::from_value(json!({
            "name": "test",
            "workers": 4,
            "inputs": {
                "ins": {
                    "stream": "test_input1",
                    "transport": {
                        "name": "file_input",
                        "config": {
                            "path": in_file,
                        },
                    },
                    "format": {
                        "name": "json",
                        "config": {
                            "update_format": "raw",
                            "array": false
                        }
                    }
                }
            },
            "outputs": {
                "test_output1": {
                    "stream": "test_output1",
                    "transport": {
                        "name": "postgres_output",
                        "config": {
                            "uri": url_clone,
                            "table": table_name,
                            "max_buffer_size_bytes": max_buffer_size_bytes,
                            "on_conflict_do_nothing": on_conflict_do_nothing,
                            "mode": match mode {
                                PostgresWriteMode::Materialized => "materialized",
                                PostgresWriteMode::Cdc => "cdc",
                            }
                        }
                    },
                    "index": "idx"
                }
            }
        }))
        .unwrap()
    };

    let mut table = PostgresTestStruct::create_table(
        table_name,
        url,
        matches!(mode, PostgresWriteMode::Materialized),
        &None,
    );

    let (controller, err_receiver) =
        PostgresTestStruct::test_circuit(get_config(insert_file.path()));

    controller.start();
    data.sort();

    wait(
        || match mode {
            PostgresWriteMode::Materialized => {
                let rows = table.query();

                let mut got = rows
                    .into_iter()
                    .map(PostgresTestStruct::from)
                    .collect::<Vec<_>>();

                got.sort();

                data == got || !err_receiver.is_empty()
            }
            PostgresWriteMode::Cdc => {
                let rows = table.query();

                let mut got = rows
                    .into_iter()
                    .map(PostgresTestStructCdc::from)
                    .collect::<Vec<_>>();

                got.sort();

                got == data && got.iter().all(|r| r.__feldera_op.as_str() == "i")
                    || !err_receiver.is_empty()
            }
        },
        40_000,
    )
    .expect("timeout: failed to insert data into postgres");

    controller.stop().unwrap();

    let (controller, err_receiver) =
        PostgresTestStruct::test_circuit(get_config(upsert_file.path()));

    controller.start();
    upsert_data.sort();

    let expected = if on_conflict_do_nothing && matches!(mode, PostgresWriteMode::Materialized) {
        data
    } else {
        upsert_data
    };

    match mode {
        PostgresWriteMode::Materialized => {
            wait(
                || {
                    let rows = table.query();

                    let mut got = rows
                        .into_iter()
                        .map(PostgresTestStruct::from)
                        .collect::<Vec<_>>();

                    got.sort();

                    got == expected || !err_receiver.is_empty()
                },
                40_000,
            )
            .expect("timeout: failed to update data into postgres");
        }
        PostgresWriteMode::Cdc => {
            wait(
                || {
                    let rows = table.query();

                    let got = rows
                        .into_iter()
                        .map(PostgresTestStructCdc::from)
                        .collect::<Vec<_>>();

                    let mut got: Vec<_> = got
                        .iter()
                        .filter(|r| r.varchar_ == "updated".into())
                        .cloned()
                        .collect();

                    got.sort();

                    got == expected && got.iter().all(|r| r.__feldera_op.as_str() == "i")
                        || !err_receiver.is_empty()
                },
                40_000,
            )
            .expect("timeout: failed to update data into postgres");
        }
    }

    controller.stop().unwrap();
}

#[test]
#[serial]
fn test_pg_on_conflict_do_update() {
    test_pg_on_conflict(false, PostgresWriteMode::Materialized);
}

#[test]
#[serial]
fn test_pg_on_conflict_do_nothing() {
    test_pg_on_conflict(true, PostgresWriteMode::Materialized);
}

#[test]
#[serial]
fn test_pg_insert0() {
    pg_insert0(PostgresWriteMode::Materialized);
}

#[test]
#[serial]
fn test_pg_insert0_cdc() {
    pg_insert0(PostgresWriteMode::Cdc);
}

fn pg_insert0(mode: PostgresWriteMode) {
    let table_name = "test_pg_insert";
    let url = postgres_url();
    let max_buffer_size_bytes = 1024;

    // On average 1 record is about 590 bytes, so 2000 records produce a buffer
    // of over 1 MiB
    let mut data: Vec<PostgresTestStruct> = (0..10000).map(|_| rand::random()).collect();

    let mut temp_file = NamedTempFile::new().unwrap();

    for datum in data.iter() {
        let buffer: Vec<u8> = Vec::new();
        let mut serializer = serde_json::Serializer::new(buffer);
        datum
            .serialize_with_context(&mut serializer, &SqlSerdeConfig::default())
            .unwrap();
        let inner = &serializer.into_inner();
        temp_file.as_file_mut().write_all(inner).unwrap();
        temp_file.write_all(b"\n").unwrap();
    }

    let config = serde_json::from_value(json!({
      "name": "test",
      "workers": 4,
      "inputs": {
        "ins": {
          "stream": "test_input1",
          "transport": {
            "name": "file_input",
            "config": {
              "path": temp_file.path(),
              "byte_size_buffer": 2097152
            }
          },
          "format": {
            "name": "json",
            "config": {
              "update_format": "raw",
              "array": false
            }
          }
        }
      },
      "outputs": {
        "test_output1": {
          "stream": "test_output1",
          "transport": {
            "name": "postgres_output",
            "config": {
              "uri": url,
              "table": table_name,
              "max_buffer_size_bytes": max_buffer_size_bytes,
              "mode": match mode {
                  PostgresWriteMode::Materialized => "materialized",
                  PostgresWriteMode::Cdc => "cdc",
              }
            }
          },
          "index": "idx"
        }
      }
    }))
    .unwrap();

    let mut table = PostgresTestStruct::create_table(
        table_name,
        url,
        matches!(mode, PostgresWriteMode::Materialized),
        &None,
    );

    let (controller, err_receiver) = PostgresTestStruct::test_circuit(config);

    controller.start();
    data.sort();

    wait(
        || match mode {
            PostgresWriteMode::Materialized => {
                let rows = table.query();

                let mut got = rows
                    .into_iter()
                    .map(PostgresTestStruct::from)
                    .collect::<Vec<_>>();

                got.sort();

                data == got || !err_receiver.is_empty()
            }
            PostgresWriteMode::Cdc => {
                let rows = table.query();

                let mut got = rows
                    .into_iter()
                    .map(PostgresTestStructCdc::from)
                    .collect::<Vec<_>>();

                got.sort();

                got == data && got.iter().all(|r| r.__feldera_op.as_str() == "i")
                    || !err_receiver.is_empty()
            }
        },
        40_000,
    )
    .expect("timeout: failed to insert data into postgres");
}

#[test]
#[serial]
fn test_pg_insert() {
    let table_name = "test_pg_insert";
    let url = postgres_url();
    let max_records_in_buffer = 1000;

    // On average 1 record is about 590 bytes, so 2000 records produce a buffer
    // of over 1 MiB
    let mut data: Vec<PostgresTestStruct> = (0..10000).map(|_| rand::random()).collect();

    let mut temp_file = NamedTempFile::new().unwrap();

    for datum in data.iter() {
        let buffer: Vec<u8> = Vec::new();
        let mut serializer = serde_json::Serializer::new(buffer);
        datum
            .serialize_with_context(&mut serializer, &SqlSerdeConfig::default())
            .unwrap();
        let inner = &serializer.into_inner();
        temp_file.as_file_mut().write_all(inner).unwrap();
        temp_file.write_all(b"\n").unwrap();
    }

    let config = serde_json::from_value(json!({
        "name": "test",
        "workers": 4,
        "inputs": {
            "ins": {
                "stream": "test_input1",
                "transport": {
                    "name": "file_input",
                    "config": {
                        "path": temp_file.path(),
                        "byte_size_buffer": 2097152
                    }
                },
                "format": {
                    "name": "json",
                    "config": {
                        "update_format": "raw",
                        "array": false
                    }
                }
            }
        },
        "outputs": {
            "test_output1": {
                "stream": "test_output1",
                "transport": {
                    "name": "postgres_output",
                    "config": {
                        "uri": url,
                        "table": table_name,
                        "max_records_in_buffer": max_records_in_buffer
                    }
                },
                "index": "idx"
            }
        }
    }))
    .unwrap();

    let mut table = PostgresTestStruct::create_table(table_name, url, true, &None);

    let (controller, err_receiver) = PostgresTestStruct::test_circuit(config);

    controller.start();

    wait(
        || {
            let rows = table.query();

            let mut got = rows
                .into_iter()
                .map(PostgresTestStruct::from)
                .collect::<Vec<_>>();

            got.sort();
            data.sort();

            data == got || !err_receiver.is_empty()
        },
        40_000,
    )
    .expect("timeout: failed to insert data into postgres");
}

#[test]
#[serial]
fn test_pg_upsert0() {
    pg_upsert0(PostgresWriteMode::Materialized);
}

#[test]
#[serial]
fn test_pg_upsert0_cdc() {
    pg_upsert0(PostgresWriteMode::Cdc);
}

fn pg_upsert0(mode: PostgresWriteMode) {
    let table_name = "test_pg_upsert";
    let url = postgres_url();
    let max_buffer_size_bytes = 1024;

    let mut data: Vec<PostgresTestStruct> = (0..10000).map(|_| rand::random()).collect();
    let mut insert_file = NamedTempFile::new().unwrap();

    for datum in data.iter() {
        let buffer: Vec<u8> = Vec::new();
        let mut serializer = serde_json::Serializer::new(buffer);
        datum
            .serialize_with_context(&mut serializer, &SqlSerdeConfig::default())
            .unwrap();
        insert_file
            .as_file_mut()
            .write_all(&serializer.into_inner())
            .unwrap();
        insert_file.write_all(b"\n").unwrap();
    }

    let mut upsert_data = data
        .clone()
        .into_iter()
        .map(|mut d| {
            d.varchar_ = "updated".into();
            d
        })
        .collect::<Vec<_>>();

    let mut upsert_file = NamedTempFile::new().unwrap();

    for old in data.iter() {
        let buffer: Vec<u8> = Vec::new();
        let mut serializer = serde_json::Serializer::new(buffer);
        old.serialize_with_context(&mut serializer, &SqlSerdeConfig::default())
            .unwrap();
        upsert_file
            .as_file_mut()
            .write_all(br#"{"delete": "#)
            .unwrap();
        upsert_file
            .as_file_mut()
            .write_all(&serializer.into_inner())
            .unwrap();
        upsert_file.write_all(b"}\n").unwrap();
    }

    for new in upsert_data.iter() {
        let buffer: Vec<u8> = Vec::new();
        let mut serializer = serde_json::Serializer::new(buffer);
        new.serialize_with_context(&mut serializer, &SqlSerdeConfig::default())
            .unwrap();
        upsert_file
            .as_file_mut()
            .write_all(br#"{"insert": "#)
            .unwrap();
        upsert_file
            .as_file_mut()
            .write_all(&serializer.into_inner())
            .unwrap();
        upsert_file.write_all(b"}\n").unwrap();
    }

    let config = serde_json::from_value(json!({
        "name": "test",
        "workers": 4,
        "inputs": {
            "ins": {
                "stream": "test_input1",
                "transport": {
                    "name": "file_input",
                    "config": {
                        "path": insert_file.path(),
                    }
                },
                "format": {
                    "name": "json",
                    "config": {
                        "update_format": "raw",
                        "array": false
                    }
                }
            },
            "ups": {
                "paused": true,
                "stream": "test_input1",
                "transport": {
                    "name": "file_input",
                    "config": {
                        "path": upsert_file.path(),
                    }
                },
                "format": {
                    "name": "json",
                    "config": {
                        "update_format": "insert_delete",
                        "array": false
                    }
                }
            }
        },
        "outputs": {
            "test_output1": {
                "stream": "test_output1",
                "transport": {
                    "name": "postgres_output",
                    "config": {
                        "uri": url,
                        "table": table_name,
                        "max_buffer_size_bytes": max_buffer_size_bytes,
                        "mode": match mode {
                            PostgresWriteMode::Materialized => "materialized",
                            PostgresWriteMode::Cdc => "cdc",
                        }

                    }
                },
                "index": "idx"
            }
        }
    }))
    .unwrap();

    let mut table = PostgresTestStruct::create_table(
        table_name,
        url,
        matches!(mode, PostgresWriteMode::Materialized),
        &None,
    );

    let (controller, err_receiver) = PostgresTestStruct::test_circuit(config);

    controller.start();
    data.sort();

    wait(
        || match mode {
            PostgresWriteMode::Materialized => {
                let rows = table.query();

                let mut got = rows
                    .into_iter()
                    .map(PostgresTestStruct::from)
                    .collect::<Vec<_>>();

                got.sort();

                data == got || !err_receiver.is_empty()
            }
            PostgresWriteMode::Cdc => {
                let rows = table.query();

                let mut got = rows
                    .into_iter()
                    .map(PostgresTestStructCdc::from)
                    .collect::<Vec<_>>();

                got.sort();

                got == data && got.iter().all(|r| r.__feldera_op.as_str() == "i")
                    || !err_receiver.is_empty()
            }
        },
        40_000,
    )
    .expect("timeout: failed to insert data into postgres");

    controller
        .start_input_endpoint("ups")
        .expect("failed to start upsert input file connector");

    upsert_data.sort();

    wait(
        || match mode {
            PostgresWriteMode::Materialized => {
                let rows = table.query();

                let mut got = rows
                    .into_iter()
                    .map(PostgresTestStruct::from)
                    .collect::<Vec<_>>();

                got.sort();

                got == upsert_data || !err_receiver.is_empty()
            }
            PostgresWriteMode::Cdc => {
                let rows = table.query();

                let got = rows
                    .into_iter()
                    .map(PostgresTestStructCdc::from)
                    .collect::<Vec<_>>();

                let mut deletes = got
                    .iter()
                    .filter(|r| r.__feldera_op.as_str() == "d")
                    .cloned()
                    .collect::<Vec<_>>();
                deletes.sort();

                let mut updates = got
                    .iter()
                    .filter(|r| r.varchar_ == "updated".into() && r.__feldera_op.as_str() == "i")
                    .cloned()
                    .collect::<Vec<_>>();
                updates.sort();

                deletes == data && updates == upsert_data || !err_receiver.is_empty()
            }
        },
        40_000,
    )
    .expect("timeout: failed to update data into postgres");
}

#[test]
#[serial]
fn test_pg_upsert() {
    let table_name = "test_pg_upsert";
    let url = postgres_url();
    let max_records_in_buffer = 1000;

    let mut data: Vec<PostgresTestStruct> = (0..10000).map(|_| rand::random()).collect();
    let mut insert_file = NamedTempFile::new().unwrap();

    for datum in data.iter() {
        let buffer: Vec<u8> = Vec::new();
        let mut serializer = serde_json::Serializer::new(buffer);
        datum
            .serialize_with_context(&mut serializer, &SqlSerdeConfig::default())
            .unwrap();
        insert_file
            .as_file_mut()
            .write_all(&serializer.into_inner())
            .unwrap();
        insert_file.write_all(b"\n").unwrap();
    }

    let mut upsert_data = data
        .clone()
        .into_iter()
        .map(|mut d| {
            d.varchar_ = "updated".into();
            d
        })
        .collect::<Vec<_>>();

    let mut upsert_file = NamedTempFile::new().unwrap();

    for old in data.iter() {
        let buffer: Vec<u8> = Vec::new();
        let mut serializer = serde_json::Serializer::new(buffer);
        old.serialize_with_context(&mut serializer, &SqlSerdeConfig::default())
            .unwrap();
        upsert_file
            .as_file_mut()
            .write_all(br#"{"delete": "#)
            .unwrap();
        upsert_file
            .as_file_mut()
            .write_all(&serializer.into_inner())
            .unwrap();
        upsert_file.write_all(b"}\n").unwrap();
    }

    for new in upsert_data.iter() {
        let buffer: Vec<u8> = Vec::new();
        let mut serializer = serde_json::Serializer::new(buffer);
        new.serialize_with_context(&mut serializer, &SqlSerdeConfig::default())
            .unwrap();
        upsert_file
            .as_file_mut()
            .write_all(br#"{"insert": "#)
            .unwrap();
        upsert_file
            .as_file_mut()
            .write_all(&serializer.into_inner())
            .unwrap();
        upsert_file.write_all(b"}\n").unwrap();
    }

    let config = serde_json::from_value(json!({
        "name": "test",
        "workers": 4,
        "inputs": {
            "ins": {
                "stream": "test_input1",
                "transport": {
                    "name": "file_input",
                    "config": {
                        "path": insert_file.path(),
                    }
                },
                "format": {
                    "name": "json",
                    "config": {
                        "update_format": "raw",
                        "array": false
                    }
                }
            },
            "ups": {
                "paused": true,
                "stream": "test_input1",
                "transport": {
                    "name": "file_input",
                    "config": {
                        "path": upsert_file.path(),
                    }
                },
                "format": {
                    "name": "json",
                    "config": {
                        "update_format": "insert_delete",
                        "array": false
                    }
                }
            }
        },
        "outputs": {
            "test_output1": {
                "stream": "test_output1",
                "transport": {
                    "name": "postgres_output",
                    "config": {
                        "uri": url,
                        "table": table_name,
                        "max_records_in_buffer": max_records_in_buffer
                    }
                },
                "index": "idx"
            }
        }
    }))
    .unwrap();

    let mut table = PostgresTestStruct::create_table(table_name, url, true, &None);

    let (controller, err_receiver) = PostgresTestStruct::test_circuit(config);

    controller.start();

    wait(
        || {
            let rows = table.query();

            let mut got = rows
                .into_iter()
                .map(PostgresTestStruct::from)
                .collect::<Vec<_>>();

            got.sort();
            data.sort();

            data == got || !err_receiver.is_empty()
        },
        40_000,
    )
    .expect("timeout: failed to insert data into postgres");

    controller
        .start_input_endpoint("ups")
        .expect("failed to start upsert input file connector");

    wait(
        || {
            let rows = table.query();

            let mut got = rows
                .into_iter()
                .map(PostgresTestStruct::from)
                .collect::<Vec<_>>();

            got.sort();
            upsert_data.sort();

            got == upsert_data || !err_receiver.is_empty()
        },
        40_000,
    )
    .expect("timeout: failed to update data into postgres");
}

#[test]
#[serial]
fn test_pg_delete() {
    pg_delete(PostgresWriteMode::Materialized)
}

#[test]
#[serial]
fn test_pg_delete_cdc() {
    pg_delete(PostgresWriteMode::Cdc)
}

fn pg_delete(mode: PostgresWriteMode) {
    let table_name = "test_pg_delete";
    let url = postgres_url();

    let mut data: Vec<PostgresTestStruct> = (0..10000).map(|_| rand::random()).collect();
    let mut insert_file = NamedTempFile::new().unwrap();

    for datum in data.iter() {
        let buffer: Vec<u8> = Vec::new();
        let mut serializer = serde_json::Serializer::new(buffer);
        datum
            .serialize_with_context(&mut serializer, &SqlSerdeConfig::default())
            .unwrap();
        insert_file
            .as_file_mut()
            .write_all(&serializer.into_inner())
            .unwrap();
        insert_file.write_all(b"\n").unwrap();
    }

    let mut delete_file = NamedTempFile::new().unwrap();

    for old in data.iter() {
        let buffer: Vec<u8> = Vec::new();
        let mut serializer = serde_json::Serializer::new(buffer);
        old.serialize_with_context(&mut serializer, &SqlSerdeConfig::default())
            .unwrap();
        delete_file
            .as_file_mut()
            .write_all(br#"{"delete": "#)
            .unwrap();
        delete_file
            .as_file_mut()
            .write_all(&serializer.into_inner())
            .unwrap();
        delete_file.write_all(b"}\n").unwrap();
    }

    let config = serde_json::from_value(json!({
      "name": "test",
      "workers": 4,
      "inputs": {
        "ins": {
          "stream": "test_input1",
          "transport": {
            "name": "file_input",
            "config": {
              "path": insert_file.path()
            }
          },
          "format": {
            "name": "json",
            "config": {
              "update_format": "raw",
              "array": false
            }
          }
        },
        "dels": {
          "paused": true,
          "stream": "test_input1",
          "transport": {
            "name": "file_input",
            "config": {
              "path": delete_file.path()
            }
          },
          "format": {
            "name": "json",
            "config": {
              "update_format": "insert_delete",
              "array": false
            }
          }
        }
      },
      "outputs": {
        "test_output1": {
          "stream": "test_output1",
          "transport": {
            "name": "postgres_output",
            "config": {
              "uri": url,
              "table": table_name,
              "mode": match mode {
                  PostgresWriteMode::Materialized => "materialized",
                  PostgresWriteMode::Cdc => "cdc",
              }
            }
          },
          "index": "idx"
        }
      }
    }))
    .unwrap();

    let mut table = PostgresTestStruct::create_table(
        table_name,
        url,
        matches!(mode, PostgresWriteMode::Materialized),
        &None,
    );

    let (controller, err_receiver) = PostgresTestStruct::test_circuit(config);

    controller.start();
    data.sort();

    wait(
        || match mode {
            PostgresWriteMode::Materialized => {
                let rows = table.query();

                let mut got = rows
                    .into_iter()
                    .map(PostgresTestStruct::from)
                    .collect::<Vec<_>>();

                got.sort();

                data == got || !err_receiver.is_empty()
            }
            PostgresWriteMode::Cdc => {
                let rows = table.query();

                let mut got = rows
                    .into_iter()
                    .map(PostgresTestStructCdc::from)
                    .collect::<Vec<_>>();

                got.sort();

                got == data && got.iter().all(|r| r.__feldera_op.as_str() == "i")
                    || !err_receiver.is_empty()
            }
        },
        40_000,
    )
    .expect("timeout: failed to insert data into postgres");

    controller
        .start_input_endpoint("dels")
        .expect("failed to start delete input file connector");

    wait(
        || match mode {
            PostgresWriteMode::Materialized => {
                let rows = table.query();

                let got = rows
                    .into_iter()
                    .map(PostgresTestStruct::from)
                    .collect::<Vec<_>>();

                got.is_empty() || !err_receiver.is_empty()
            }
            PostgresWriteMode::Cdc => {
                let rows = table.query();

                let got = rows
                    .into_iter()
                    .map(PostgresTestStructCdc::from)
                    .collect::<Vec<_>>();

                let mut got = got
                    .iter()
                    .filter(|r| r.__feldera_op.as_str() != "i")
                    .cloned()
                    .collect::<Vec<_>>();

                got.sort();

                got.len() == data.len()
                    && got
                        .iter()
                        .skip(data.len())
                        .all(|r| r.__feldera_op.as_str() == "d")
                    || !err_receiver.is_empty()
            }
        },
        40_000,
    )
    .expect("timeout: failed to delete data from postgres");
}

fn pg_simple(url: String, tls: Option<PostgresTlsConfig>) {
    let table_name = "simple_test";

    let mut client = pg::pg_connect(&url, &tls);
    client
        .execute(
            &format!(
                r#"CREATE TABLE IF NOT EXISTS {table_name} (
    id int primary key,
    b bool not null,
    i int8,
    s varchar not null
)"#
            ),
            &[],
        )
        .expect("failed to create test table in postgres");

    client
        .execute(&format!("TRUNCATE {table_name}"), &[])
        .unwrap();

    let input_file = NamedTempFile::new().unwrap();
    let data = vec![
        TestStruct {
            id: 1,
            b: true,
            i: None,
            s: "test".to_owned(),
        },
        TestStruct {
            id: 2,
            b: false,
            i: Some(1),
            s: "test".to_owned(),
        },
    ];

    input_file
        .as_file()
        .write_all(&serde_json::to_vec(&data).unwrap())
        .unwrap();

    let upsert_file = NamedTempFile::new().unwrap();
    let upsert_data: Vec<serde_json::Value> = data
        .clone()
        .into_iter()
        .flat_map(|d| {
            let updated = {
                let mut n = d.clone();
                n.s = "updated".to_owned();
                n
            };
            [
                json!({
                    "delete": d,
                }),
                json!({
                    "insert": updated,
                }),
            ]
        })
        .collect();

    upsert_file
        .as_file()
        .write_all(&serde_json::to_vec(&upsert_data).unwrap())
        .unwrap();

    let delete_file = NamedTempFile::new().unwrap();
    delete_file
        .as_file()
        .write_all(
            &serde_json::to_vec(
                &data
                    .clone()
                    .into_iter()
                    .map(|d| json!({"delete": d}))
                    .collect::<Vec<_>>(),
            )
            .unwrap(),
        )
        .unwrap();

    let idx = "v1_idx";

    let mut pg_output_config = json!({
        "uri": url,
        "table": table_name,
    });
    if let Some(ref tls) = tls {
        let tls_json = serde_json::to_value(tls).unwrap();
        pg_output_config
            .as_object_mut()
            .unwrap()
            .extend(tls_json.as_object().unwrap().clone());
    }

    let schema = TestStruct::schema();
    let config = serde_json::from_value(json!({
      "name": "test",
      "workers": 4,
      "inputs": {
        "ins": {
          "stream": "test_input1",
          "transport": {
            "name": "file_input",
            "config": {
              "path": input_file.path(),
            }
          },
          "format": {
            "name": "json",
            "config": {
              "update_format": "raw",
              "array": true
            }
          }
        },
        "ups": {
          "paused": true,
          "stream": "test_input1",
          "transport": {
            "name": "file_input",
            "config": {
              "path": upsert_file.path()
            }
          },
          "format": {
            "name": "json",
            "config": {
              "update_format": "insert_delete",
              "array": true
            }
          }
        },
        "del": {
          "paused": true,
          "stream": "test_input1",
          "transport": {
            "name": "file_input",
            "config": {
              "path": delete_file.path()
            }
          },
          "format": {
            "name": "json",
            "config": {
              "update_format": "insert_delete",
              "array": true
            }
          }
        }
      },
      "outputs": {
        "test_output1": {
          "stream": "test_output1",
          "transport": {
            "name": "postgres_output",
            "config": pg_output_config,
          },
          "index": idx
        }
      }
    }))
    .unwrap();

    let (err_sender, err_receiver) = crossbeam::channel::unbounded();
    let controller = Controller::with_test_config(
        move |workers| {
            Ok({
                let (circuit, catalog) = Runtime::init_circuit(workers, move |circuit| {
                    let mut catalog = Catalog::new();
                    let (input, hinput) = circuit.add_input_zset::<TestStruct>();

                    let input_schema = serde_json::to_string(&Relation::new(
                        "test_input1".into(),
                        schema.clone(),
                        false,
                        BTreeMap::new(),
                    ))
                    .unwrap();

                    let output_schema = serde_json::to_string(&Relation::new(
                        "test_output1".into(),
                        schema,
                        false,
                        BTreeMap::new(),
                    ))
                    .unwrap();

                    catalog.register_materialized_input_zset::<_, TestStruct>(
                        input.clone(),
                        hinput,
                        &input_schema,
                    );

                    #[derive(Clone, Debug, Eq, PartialEq, Default)]
                    pub struct KeyStruct {
                        field0: u32,
                    }
                    impl From<KeyStruct> for Tup1<u32> {
                        fn from(t: KeyStruct) -> Self {
                            Tup1::new(t.field0)
                        }
                    }
                    impl From<Tup1<u32>> for KeyStruct {
                        fn from(t: Tup1<u32>) -> Self {
                            Self { field0: t.0 }
                        }
                    }
                    deserialize_table_record!(KeyStruct["v1_idx", Variant, 1] {
                        (field0, "id", false, u32, |_| None)
                    });
                    serialize_table_record!(KeyStruct[1]{
                        field0["id"]: u32
                    });

                    let indexed_input = input.map_index(|r| (Tup1(r.id), r.to_owned()));

                    catalog
                        .register_materialized_output_zset::<_, TestStruct>(input, &output_schema);

                    catalog
                        .register_index::<Tup1<u32>, KeyStruct, TestStruct, TestStruct>(
                            indexed_input.clone(),
                            &SqlIdentifier::from(idx),
                            &SqlIdentifier::from("test_output1"),
                            &[&SqlIdentifier::from("id")],
                        )
                        .expect("failed to register index");

                    Ok(catalog)
                })
                .unwrap();
                (circuit, Box::new(catalog))
            })
        },
        &config,
        Box::new(move |e, _| {
            let msg = format!("postgres_output_test: error: {e}");
            println!("{msg}");
            err_sender.send(msg).unwrap()
        }),
    )
    .unwrap();

    controller.start();

    wait(
        ||
            {
                let rows = client
                    .query(
                        &format!("SELECT * FROM {table_name} ORDER BY id"),
                        &[],
                    )
                    .unwrap();
                let got: Vec<TestStruct> = rows
                    .into_iter()
                    .map(|row| {
                        let id: i32 = row.get(0);
                        let b: bool = row.get(1);
                        let i: Option<i64> = row.get(2);
                        let s: String = row.get(3);

                        TestStruct {
                            id: id as u32,
                            b,
                            i,
                            s,
                        }
                    })
                    .collect();

                got == data
            }
            || !err_receiver.is_empty(),
        1_000,
    )
    .expect("timeout: inserting records into postgres failed");

    controller.start_input_endpoint("ups").unwrap();

    wait(
        ||
            {
                client
                    .query(&format!("SELECT * FROM {table_name}"), &[])
                    .unwrap()
                    .iter()
                    .all(|r| {
                        let s: String = r.get("s");
                        s.as_str() == "updated"
                    })
            }
            || !err_receiver.is_empty(),
        1_000,
    )
    .expect("timeout");

    controller.start_input_endpoint("del").unwrap();

    wait(
        || {
            client
                .query(&format!("SELECT * FROM {table_name}"), &[])
                .unwrap()
                .is_empty()
                || !err_receiver.is_empty()
        },
        1_000,
    )
    .expect("timedout: failed to delete records");

    client
        .execute(&format!("DROP TABLE {table_name}"), &[])
        .unwrap();
}

#[test]
fn test_pg_simple() {
    pg_simple(postgres_url(), None);
}

/// Same as `test_pg_simple` but over a TLS connection.
#[test]
fn test_pg_simple_tls() {
    let (url, tls) = postgres_ssl_config();
    pg_simple(url, Some(tls));
}

/// Test the postgres *input* connector over a TLS connection.
///
/// Inserts rows into TLS-enabled postgres directly, then reads them back
/// through the `postgres_input` transport and writes them to a file.
#[test]
#[serial]
fn test_pg_input_tls() {
    let (url, tls) = postgres_ssl_config();
    let table_name = "test_pg_input_tls";

    // -- seed the table via the sync client --
    let mut client = pg::pg_connect(&url, &Some(tls.clone()));
    client
        .execute(
            &format!(
                r#"CREATE TABLE IF NOT EXISTS {table_name} (
    id int primary key,
    b bool not null,
    i int8,
    s varchar not null
)"#
            ),
            &[],
        )
        .expect("failed to create test table");
    client
        .execute(&format!("TRUNCATE {table_name}"), &[])
        .unwrap();
    client
        .execute(
            &format!("INSERT INTO {table_name} VALUES (1, true, NULL, 'hello')"),
            &[],
        )
        .unwrap();
    client
        .execute(
            &format!("INSERT INTO {table_name} VALUES (2, false, 42, 'world')"),
            &[],
        )
        .unwrap();

    // -- set up the pipeline: postgres_input -> file_output --
    let output_file = NamedTempFile::new().unwrap();
    let output_path = output_file.path().to_owned();

    let mut pg_input_config = json!({
        "uri": url,
        "query": format!("SELECT * FROM {table_name} ORDER BY id"),
    });
    let tls_json = serde_json::to_value(&tls).unwrap();
    pg_input_config
        .as_object_mut()
        .unwrap()
        .extend(tls_json.as_object().unwrap().clone());

    let schema = TestStruct::schema();
    let config = serde_json::from_value(json!({
        "name": "test",
        "workers": 1,
        "inputs": {
            "pg_in": {
                "stream": "test_input1",
                "transport": {
                    "name": "postgres_input",
                    "config": pg_input_config,
                },
            },
        },
        "outputs": {
            "test_output1": {
                "stream": "test_output1",
                "transport": {
                    "name": "file_output",
                    "config": {
                        "path": output_path,
                    }
                },
                "format": {
                    "name": "json",
                    "config": {
                        "update_format": "insert_delete",
                        "array": false,
                    }
                }
            }
        }
    }))
    .unwrap();

    let (err_sender, err_receiver) = crossbeam::channel::unbounded();
    let controller = Controller::with_test_config(
        move |workers| {
            Ok({
                let (circuit, catalog) = Runtime::init_circuit(workers, move |circuit| {
                    let mut catalog = Catalog::new();
                    let (input, hinput) = circuit.add_input_zset::<TestStruct>();

                    let input_schema = serde_json::to_string(&Relation::new(
                        "test_input1".into(),
                        schema.clone(),
                        false,
                        BTreeMap::new(),
                    ))
                    .unwrap();

                    let output_schema = serde_json::to_string(&Relation::new(
                        "test_output1".into(),
                        schema,
                        false,
                        BTreeMap::new(),
                    ))
                    .unwrap();

                    catalog.register_materialized_input_zset::<_, TestStruct>(
                        input.clone(),
                        hinput,
                        &input_schema,
                    );

                    catalog
                        .register_materialized_output_zset::<_, TestStruct>(input, &output_schema);

                    Ok(catalog)
                })
                .unwrap();
                (circuit, Box::new(catalog))
            })
        },
        &config,
        Box::new(move |e, _| {
            let msg = format!("test_pg_input_tls: error: {e}");
            println!("{msg}");
            err_sender.send(msg).unwrap()
        }),
    )
    .unwrap();

    controller.start();

    // Wait for the output file to contain the expected data.
    wait(
        || {
            let content = std::fs::read_to_string(&output_path).unwrap_or_default();
            let lines: Vec<&str> = content.lines().filter(|l| !l.is_empty()).collect();
            lines.len() >= 2 || !err_receiver.is_empty()
        },
        10_000,
    )
    .expect("timeout: postgres input TLS test did not produce output");

    assert!(err_receiver.is_empty(), "unexpected errors in pipeline");

    let content = std::fs::read_to_string(&output_path).unwrap();
    let rows: Vec<serde_json::Value> = content
        .lines()
        .filter(|l| !l.is_empty())
        .map(|l| serde_json::from_str(l).unwrap())
        .collect();

    assert_eq!(rows.len(), 2);
    assert_eq!(rows[0]["insert"]["id"], 1);
    assert_eq!(rows[0]["insert"]["b"], true);
    assert!(rows[0]["insert"]["i"].is_null());
    assert_eq!(rows[0]["insert"]["s"], "hello");
    assert_eq!(rows[1]["insert"]["id"], 2);
    assert_eq!(rows[1]["insert"]["b"], false);
    assert_eq!(rows[1]["insert"]["i"], 42);
    assert_eq!(rows[1]["insert"]["s"], "world");

    controller.stop().unwrap();

    client
        .execute(&format!("DROP TABLE {table_name}"), &[])
        .unwrap();
}
