Fivetran: use POST requests for list_snapshot and document_deltas (#37305)

This switches to using POST requests for `list_snapshot` and `document_deltas`.  This improves the type safety by using the shared `Args` objects moved in #37281, and allows for longer request bodies.

Since we no longer do requests with query parameters I also simplified the implementation of `.get` to remove the `parameters` parameter.

I verified manually that `convex_api` still works with the Convex backend.

GitOrigin-RevId: 8d8fd85df57569da6a33951bc99984070b4fcc05

Author: Nicolapps

crates/fivetran_source/Cargo.toml

@@ -19,7 +19,6 @@ derive_more = { workspace = true }
 futures = { workspace = true }
 futures-async-stream = { workspace = true }
 headers = { workspace = true }
-maplit = { workspace = true }
 prost = { workspace = true }
 prost-types = { workspace = true }
 reqwest = { workspace = true }
@@ -40,6 +39,7 @@ tonic-build = { workspace = true }
 cmd_util = { path = "../cmd_util" }
 convex = { path = "../convex", features = ["testing"] }
 convex_fivetran_common = { path = "../fivetran_common", features = ["testing"] }
+maplit = { workspace = true }
 proptest = { workspace = true }
 proptest-derive = { workspace = true }
 rand = { workspace = true }

crates/fivetran_source/src/convex_api.rs

@@ -7,6 +7,10 @@ use std::{
 use anyhow::Context;
 use async_trait::async_trait;
 use convex_fivetran_common::config::Config;
+use convex_fivetran_source::api_types::{
+    DocumentDeltasArgs,
+    ListSnapshotArgs,
+};
 use derive_more::{
     Display,
     From,
@@ -16,7 +20,6 @@ use headers::{
     HeaderName,
     HeaderValue,
 };
-use maplit::btreemap;
 use serde::{
     de::DeserializeOwned,
     Deserialize,
@@ -68,35 +71,70 @@ pub struct ConvexApi {
 }
 
 impl ConvexApi {
-    /// Performs a GET HTTP request to a given endpoint of the Convex API using
-    /// the given query parameters.
-    async fn get<T: DeserializeOwned>(
+    /// Performs a GET HTTP request to a given endpoint of the Convex API.
+    async fn get<T: DeserializeOwned>(&self, endpoint: &str) -> anyhow::Result<T> {
+        let url = self
+            .config
+            .deploy_url
+            .join("api/")
+            .unwrap()
+            .join(endpoint)
+            .unwrap();
+
+        match reqwest::Client::new()
+            .get(url)
+            .header(CONVEX_CLIENT_HEADER, &*CONVEX_CLIENT_HEADER_VALUE)
+            .header(
+                reqwest::header::AUTHORIZATION,
+                format!("Convex {}", self.config.deploy_key),
+            )
+            .send()
+            .await
+        {
+            Ok(resp) if resp.status().is_success() => Ok(resp
+                .json::<T>()
+                .await
+                .context("Failed to deserialize query result")?),
+            Ok(resp) => {
+                if let Ok(text) = resp.text().await {
+                    anyhow::bail!(
+                        "Call to {endpoint} on {} returned an unsuccessful response: {text}",
+                        self.config.deploy_url
+                    )
+                } else {
+                    anyhow::bail!(
+                        "Call to {endpoint} on {} returned no response",
+                        self.config.deploy_url
+                    )
+                }
+            },
+            Err(e) => anyhow::bail!(e.to_string()),
+        }
+    }
+
+    /// Performs a POST HTTP request to a given endpoint of the Convex API using
+    /// the given parameters as a JSON body.
+    async fn post<P: Serialize, T: DeserializeOwned>(
         &self,
         endpoint: &str,
-        parameters: BTreeMap<&str, Option<String>>,
+        parameters: P,
     ) -> anyhow::Result<T> {
-        let non_null_parameters: BTreeMap<&str, String> = parameters
-            .into_iter()
-            .filter_map(|(key, value)| value.map(|value| (key, value)))
-            .collect();
-
-        let mut url = self
+        let url = self
             .config
             .deploy_url
             .join("api/")
             .unwrap()
             .join(endpoint)
             .unwrap();
 
-        url.query_pairs_mut().extend_pairs(non_null_parameters);
-
         match reqwest::Client::new()
-            .get(url)
+            .post(url)
             .header(CONVEX_CLIENT_HEADER, &*CONVEX_CLIENT_HEADER_VALUE)
             .header(
                 reqwest::header::AUTHORIZATION,
                 format!("Convex {}", self.config.deploy_key),
             )
+            .json(&parameters)
             .send()
             .await
         {
@@ -125,8 +163,7 @@ impl ConvexApi {
 #[async_trait]
 impl Source for ConvexApi {
     async fn test_streaming_export_connection(&self) -> anyhow::Result<()> {
-        self.get("test_streaming_export_connection", btreemap! {})
-            .await
+        self.get("test_streaming_export_connection").await
     }
 
     async fn list_snapshot(
@@ -135,13 +172,14 @@ impl Source for ConvexApi {
         cursor: Option<ListSnapshotCursor>,
         table_name: Option<String>,
     ) -> anyhow::Result<ListSnapshotResponse> {
-        self.get(
+        self.post(
             "list_snapshot",
-            btreemap! {
-                "snapshot" => snapshot.map(|n| n.to_string()),
-                "cursor" => cursor.map(|n| n.to_string()),
-                "tableName" => table_name,
-                "format" => Some("convex_encoded_json".to_string()),
+            ListSnapshotArgs {
+                snapshot,
+                cursor: cursor.map(|c| c.into()),
+                table_name,
+                component: None,
+                format: Some("convex_encoded_json".to_string()),
             },
         )
         .await
@@ -152,20 +190,21 @@ impl Source for ConvexApi {
         cursor: DocumentDeltasCursor,
         table_name: Option<String>,
     ) -> anyhow::Result<DocumentDeltasResponse> {
-        self.get(
+        self.post(
             "document_deltas",
-            btreemap! {
-                "cursor" => Some(cursor.to_string()),
-                "tableName" => table_name,
-                "format" => Some("convex_encoded_json".to_string()),
+            DocumentDeltasArgs {
+                cursor: Some(cursor.into()),
+                table_name,
+                component: None,
+                format: Some("convex_encoded_json".to_string()),
             },
         )
         .await
     }
 
     async fn get_tables_and_columns(&self) -> anyhow::Result<BTreeMap<TableName, Vec<FieldName>>> {
         let tables_to_columns: BTreeMap<TableName, Vec<String>> =
-            self.get("get_tables_and_columns", btreemap! {}).await?;
+            self.get("get_tables_and_columns").await?;
 
         tables_to_columns
             .into_iter()