From 02b19f5b6464b47689960b2808736dfeba8853b1 Mon Sep 17 00:00:00 2001
From: Alan Chen <alan@infinyon.com>
Date: Tue, 4 Jun 2024 12:07:11 -0700
Subject: [PATCH] add(hub-util): sdf endpoint consts and refactor

---
 crates/fluvio-hub-protocol/src/constants.rs | 11 ++++++++
 crates/fluvio-hub-util/src/utils.rs         | 28 +++++++++++++++++++++
 2 files changed, 39 insertions(+)

diff --git a/crates/fluvio-hub-protocol/src/constants.rs b/crates/fluvio-hub-protocol/src/constants.rs
index 6bf7cc4dce5..e426e5d61b7 100644
--- a/crates/fluvio-hub-protocol/src/constants.rs
+++ b/crates/fluvio-hub-protocol/src/constants.rs
@@ -16,6 +16,12 @@ pub const HUB_API_LIST_META: &str = concatcp!(HUB_API_V, "/list_with_meta");
 pub const HUB_API_CONN_PKG: &str = concatcp!(HUB_API_V, "/connector/pkg");
 pub const HUB_API_CONN_LIST: &str = concatcp!(HUB_API_V, "/connector/list");
 
+// sdf specific api
+pub const HUB_API_SDF_PKG: &str = concatcp!(HUB_API_V, "/sdf/pkg");
+pub const HUB_API_SDF_LIST: &str = concatcp!(HUB_API_V, "/sdf/list");
+pub const HUB_API_SDF_PKG_PUB: &str = concatcp!(HUB_API_V, "/sdf/pkg/pub/pkg");
+pub const HUB_API_SDF_DATAFLOW_PUB: &str = concatcp!(HUB_API_V, "/sdf/pkg/pub/dataflow");
+
 pub const HUB_MANIFEST_BLOB: &str = "manifest.tar.gz";
 pub const HUB_PACKAGE_EXT: &str = "ipkg";
 pub const HUB_PACKAGE_META: &str = "package-meta.yaml";
@@ -27,3 +33,8 @@ pub const HUB_SIGNFILE_BASE: &str = "signature";
 pub const DEF_CARGO_TOML_PATH: &str = "Cargo.toml";
 pub const DEF_HUB_INIT_DIR: &str = ".hub";
 pub const DEF_HUB_PKG_META: &str = concatcp!(DEF_HUB_INIT_DIR, "/", HUB_PACKAGE_META); // .hub/package-meta.yaml
+
+// This is required in sdf hub package_meta manifests
+pub const SDF_PKG_KIND: &str = "sdf-kind";
+pub const SDF_PKG_KIND_DATAFLOW: &str = "dataflow";
+pub const SDF_PKG_KIND_PACKAGE: &str = "pkg";
diff --git a/crates/fluvio-hub-util/src/utils.rs b/crates/fluvio-hub-util/src/utils.rs
index 510067cfa16..2cc16cf7098 100644
--- a/crates/fluvio-hub-util/src/utils.rs
+++ b/crates/fluvio-hub-util/src/utils.rs
@@ -168,6 +168,34 @@ pub async fn push_package_conn(pkgpath: &str, access: &HubAccess, target: &str)
     push_package_api(&url, pkgpath, access).await
 }
 
+/// push package to connector api
+pub async fn push_package_sdf(pkgpath: &str, access: &HubAccess, target: &str) -> Result<()> {
+    use crate::{
+        SDF_PKG_KIND, SDF_PKG_KIND_DATAFLOW, SDF_PKG_KIND_PACKAGE, HUB_API_SDF_DATAFLOW_PUB,
+        HUB_API_SDF_PKG_PUB,
+    };
+
+    info!("sdf package target: {target}");
+    let pm = package_get_meta(pkgpath)?;
+    let Some(sdf_kind) = pm.tag_get(SDF_PKG_KIND) else {
+        let msg = format!("Invalid sdf hub package_meta: missing tag for {SDF_PKG_KIND}");
+        return Err(HubError::PackagePublish(msg));
+    };
+    let sdf_kind = sdf_kind.first().cloned().unwrap_or_default().value;
+    let endpoint = match sdf_kind.as_str() {
+        SDF_PKG_KIND_DATAFLOW => HUB_API_SDF_DATAFLOW_PUB,
+        SDF_PKG_KIND_PACKAGE => HUB_API_SDF_PKG_PUB,
+        _ => {
+            let msg = format!("Invalid sdf hub package_meta {SDF_PKG_KIND}: {sdf_kind}");
+            return Err(HubError::PackagePublish(msg));
+        }
+    };
+    let host = &access.remote;
+    let url = format!("{host}/{endpoint}/{}/{}/{}", pm.group, pm.name, pm.version);
+    debug!(url, "package url");
+    push_package_api(&url, pkgpath, access).await
+}
+
 pub async fn push_package_api(put_url: &str, pkgpath: &str, access: &HubAccess) -> Result<()> {
     let pm = package_get_meta(pkgpath)?;
     packagename_validate(&pm.name)?;