Parse Fivetran CSV files (#26137)

GitOrigin-RevId: 797fc70b09ec53c31adc7a030e7079165795ca02

Author: Nicolapps

Cargo.lock

@@ -4,12 +4,15 @@ resolver = "2"
 exclude = [ "crates/fivetran_source", "crates/py_client", "crates/python_client_tests" ]
 
 [workspace.dependencies]
+aes = { version = "0.8.4" }
 anyhow = "1"
 async-broadcast = "0.7.0"
 async-channel = "1.9.0"
+async-compression = { version = "0.4.8", features = [ "tokio", "zstd", "gzip" ] }
 async-recursion = "1.1.1"
 async-trait = "0.1"
 async_zip = { version = "0.0.9", default-features = false, features = [ "zstd", "deflate" ] }
+cbc = { version = "0.1.2" }
 csv-async = "1.2"
 atomic_refcell = "0.1.10"
 axum = { version = "0.6", features = [ "headers", "ws", "original-uri", "macros", "multipart" ] }
@@ -98,6 +101,7 @@ ring = "0.17.0"
 rsa = "0.9.0"
 rusqlite = { version = "0.30", features = [ "bundled" ] }
 saffron = { git = "https://github.com/get-convex/saffron", rev = "1d842379919fb5c1988ac127cebd6167b1eb9bec", features = [ "std" ] }
+schemars = { version = "0.8" }
 semver = { version = "1", features = [ "serde" ] }
 sentry = { version = "0.31", features = [ "anyhow", "tower", "tower-http" ] }
 serde = { version = "1", features = [ "derive" ] }

Cargo.toml

@@ -0,0 +1,59 @@
+[package]
+name = "convex_fivetran_destination"
+description = "Fivetran destination for Convex (convex.dev)"
+version = "0.0.1"
+authors = ["Convex, Inc. <[email protected]>"]
+edition = "2021"
+resolver = "2"
+license = "Apache-2.0"
+repository = "https://github.com/get-convex/convex-fivetran-source"
+homepage = "https://www.convex.dev/"
+
+[lib]
+name = "convex_fivetran_destination"
+path = "src/lib.rs"
+
+[dependencies]
+aes = { workspace = true }
+anyhow = { workspace = true }
+async-compression = { workspace = true }
+async-trait = { workspace = true }
+base64 = { workspace = true }
+cbc = { workspace = true }
+chrono = { workspace = true }
+common = { path = "../common" }
+csv-async = { workspace = true }
+derive_more = { workspace = true }
+futures = { workspace = true }
+futures-async-stream = { workspace = true }
+maplit = { workspace = true }
+prost = { workspace = true }
+prost-types = { workspace = true }
+reqwest = { workspace = true, features = ["json", "native-tls-vendored"] }
+serde = { workspace = true, features = ["derive"] }
+serde_json = { workspace = true }
+thiserror = { workspace = true }
+tokio = { workspace = true }
+tokio-stream = { workspace = true }
+tonic = { workspace = true, features = ["gzip"] }
+url = { workspace = true }
+
+[build-dependencies]
+cfg-if = { workspace = true }
+tonic-build = { workspace = true }
+
+[dev-dependencies]
+common = { path = "../common", features = ["testing"] }
+proptest = { workspace = true }
+proptest-derive = { workspace = true }
+rand = { workspace = true }
+uuid = { workspace = true }
+
+[package.metadata.cargo-machete]
+ignored = [
+    # Build dependencies not understood
+    "cfg_if",
+    "tonic_build",
+    # Prost required via tonic macro
+    "prost",
+]

crates/fivetran_destination/Cargo.toml

@@ -0,0 +1,39 @@
+use std::{
+    io::Result,
+    path::Path,
+};
+
+cfg_if::cfg_if! {
+    if #[cfg(target_os = "macos")] {
+        const PROTOC_BINARY_NAME: &str = "protoc-macos-universal";
+    } else if #[cfg(all(target_os = "linux", target_arch = "aarch64"))] {
+        const PROTOC_BINARY_NAME: &str = "protoc-linux-aarch64";
+    } else if #[cfg(all(target_os = "linux", target_arch = "x86_64"))] {
+        const PROTOC_BINARY_NAME: &str = "protoc-linux-x86_64";
+    } else {
+        panic!("no protoc binary available for this architecture");
+    }
+}
+
+fn set_protoc_path() {
+    let root = Path::new("../pb_build/protoc");
+    if root.exists() {
+        let include_path = std::fs::canonicalize(root.join("include"))
+            .expect("Failed to canonicalize protoc include path");
+        std::env::set_var("PROTOC_INCLUDE", include_path);
+        let binary_path = std::fs::canonicalize(root.join(PROTOC_BINARY_NAME))
+            .expect("Failed to canonicalize protoc path");
+        std::env::set_var("PROTOC", binary_path);
+    }
+}
+
+fn main() -> Result<()> {
+    set_protoc_path();
+
+    tonic_build::configure().btree_map(["."]).compile(
+        &["protos/common.proto", "protos/destination_sdk.proto"],
+        &["protos/"],
+    )?;
+
+    Ok(())
+}