[PATCH 6/8] cli: add subscription key pool management subcommands

[Thomas Lamprecht <t.lamprecht@proxmox.com>]

Expose the new key-pool API to the CLI so that the subscriptions
command group gains the following sub commands:

  list-keys
  add-keys (variadic)
  assign-key
  clear-key
  remove-key
  auto-assign
  apply-pending
  clear-pending

The pre-existing `status` subcommand becomes a sibling under this
`subscriptions` group.

Signed-off-by: Thomas Lamprecht <t.lamprecht@proxmox.com>
---
 cli/client/src/subscriptions.rs | 195 +++++++++++++++++++++++++++++++-
 lib/pdm-client/src/lib.rs       | 126 ++++++++++++++++++++-
 2 files changed, 316 insertions(+), 5 deletions(-)

diff --git a/cli/client/src/subscriptions.rs b/cli/client/src/subscriptions.rs
index d8bf1e0..5d5532b 100644
--- a/cli/client/src/subscriptions.rs
+++ b/cli/client/src/subscriptions.rs
@@ -1,18 +1,44 @@
 use anyhow::Error;
 
 use proxmox_router::cli::{
-    format_and_print_result, CliCommand, CommandLineInterface, OutputFormat,
+    format_and_print_result, CliCommand, CliCommandMap, CommandLineInterface, OutputFormat,
 };
 use proxmox_schema::api;
 
-use pdm_api_types::subscription::RemoteSubscriptionState;
+use pdm_api_types::remotes::REMOTE_ID_SCHEMA;
+use pdm_api_types::subscription::{RemoteSubscriptionState, SUBSCRIPTION_KEY_SCHEMA};
 use pdm_api_types::VIEW_ID_SCHEMA;
 
 use crate::env::emoji;
 use crate::{client, env};
 
 pub fn cli() -> CommandLineInterface {
-    CliCommand::new(&API_METHOD_GET_SUBSCRIPTION_STATUS).into()
+    CliCommandMap::new()
+        .insert(
+            "status",
+            CliCommand::new(&API_METHOD_GET_SUBSCRIPTION_STATUS),
+        )
+        .insert("list-keys", CliCommand::new(&API_METHOD_LIST_KEYS))
+        .insert(
+            "add-keys",
+            CliCommand::new(&API_METHOD_ADD_KEYS).arg_param(&["keys"]),
+        )
+        .insert(
+            "assign-key",
+            CliCommand::new(&API_METHOD_ASSIGN_KEY).arg_param(&["key"]),
+        )
+        .insert(
+            "clear-key",
+            CliCommand::new(&API_METHOD_CLEAR_KEY).arg_param(&["key"]),
+        )
+        .insert(
+            "remove-key",
+            CliCommand::new(&API_METHOD_REMOVE_KEY).arg_param(&["key"]),
+        )
+        .insert("auto-assign", CliCommand::new(&API_METHOD_AUTO_ASSIGN))
+        .insert("apply-pending", CliCommand::new(&API_METHOD_APPLY_PENDING))
+        .insert("clear-pending", CliCommand::new(&API_METHOD_CLEAR_PENDING))
+        .into()
 }
 
 #[api(
@@ -37,7 +63,7 @@ pub fn cli() -> CommandLineInterface {
         },
     }
 )]
-/// List all the remotes this instance is managing.
+/// Show the subscription status of all remotes.
 async fn get_subscription_status(
     max_age: Option<u64>,
     verbose: Option<bool>,
@@ -106,3 +132,164 @@ async fn get_subscription_status(
     }
     Ok(())
 }
+
+#[api]
+/// List all subscription keys in the pool.
+async fn list_keys() -> Result<(), Error> {
+    let (keys, _digest) = client()?.list_subscription_keys().await?;
+
+    let output_format = env().format_args.output_format;
+    if output_format == OutputFormat::Text {
+        if keys.is_empty() {
+            println!("No keys in pool.");
+            return Ok(());
+        }
+        let key_width = keys.iter().map(|k| k.key.len()).max().unwrap_or(20);
+        for key in &keys {
+            let assignment = match (&key.remote, &key.node) {
+                (Some(r), Some(n)) => format!("{r}/{n}"),
+                _ => "(unassigned)".to_string(),
+            };
+            println!(
+                "  {key:<kw$}  {product:<5}  {level:<10}  {status:<10}  {assignment}",
+                key = key.key,
+                kw = key_width,
+                product = key.product_type.to_string(),
+                level = key.level.to_string(),
+                status = key.status.to_string(),
+            );
+        }
+    } else {
+        format_and_print_result(&keys, &output_format.to_string());
+    }
+    Ok(())
+}
+
+#[api(
+    input: {
+        properties: {
+            keys: {
+                type: Array,
+                description: "Subscription keys to add to the pool.",
+                items: { schema: SUBSCRIPTION_KEY_SCHEMA },
+            },
+        },
+    },
+)]
+/// Add one or more subscription keys to the pool.
+async fn add_keys(keys: Vec<String>) -> Result<(), Error> {
+    client()?.add_subscription_keys(&keys, None).await?;
+    let n = keys.len();
+    if n == 1 {
+        println!("Added {} to pool.", keys[0]);
+    } else {
+        println!("Added {n} keys to pool.");
+    }
+    Ok(())
+}
+
+#[api(
+    input: {
+        properties: {
+            key: { schema: SUBSCRIPTION_KEY_SCHEMA },
+            remote: { schema: REMOTE_ID_SCHEMA },
+            node: {
+                type: String,
+                description: "Node name within the remote.",
+            },
+        },
+    },
+)]
+/// Assign a key from the pool to a remote node.
+async fn assign_key(key: String, remote: String, node: String) -> Result<(), Error> {
+    client()?
+        .assign_subscription_key(&key, Some(&remote), Some(&node), None)
+        .await?;
+    println!("Assigned {key} to {remote}/{node}.");
+    Ok(())
+}
+
+#[api(
+    input: {
+        properties: {
+            key: { schema: SUBSCRIPTION_KEY_SCHEMA },
+        },
+    },
+)]
+/// Clear the assignment of a key (unassign from its remote node).
+async fn clear_key(key: String) -> Result<(), Error> {
+    client()?
+        .assign_subscription_key(&key, None, None, None)
+        .await?;
+    println!("Cleared assignment for {key}.");
+    Ok(())
+}
+
+#[api(
+    input: {
+        properties: {
+            key: { schema: SUBSCRIPTION_KEY_SCHEMA },
+        },
+    },
+)]
+/// Remove a key from the pool entirely.
+async fn remove_key(key: String) -> Result<(), Error> {
+    client()?.delete_subscription_key(&key).await?;
+    println!("Removed {key} from pool.");
+    Ok(())
+}
+
+#[api(
+    input: {
+        properties: {
+            apply: {
+                type: bool,
+                optional: true,
+                default: false,
+                description: "Apply the proposed assignments immediately.",
+            },
+        },
+    },
+)]
+/// Propose or apply automatic key-to-node assignments.
+async fn auto_assign(apply: Option<bool>) -> Result<(), Error> {
+    let apply = apply.unwrap_or(false);
+    let proposals = client()?.subscription_auto_assign(apply).await?;
+
+    if proposals.is_empty() {
+        println!("No suitable keys available for unsubscribed nodes.");
+        return Ok(());
+    }
+
+    for p in &proposals {
+        let verb = if apply { "assigned" } else { "proposed" };
+        println!("  {verb}: {} -> {}/{}", p.key, p.remote, p.node);
+    }
+
+    if !apply {
+        println!("\nRe-run with --apply to apply these assignments.");
+    }
+    Ok(())
+}
+
+#[api]
+/// Push all pending key assignments to remotes as a worker task.
+async fn apply_pending() -> Result<(), Error> {
+    match client()?.subscription_apply_pending().await? {
+        None => println!("No pending assignments to apply."),
+        Some(upid) => println!("Task started: {upid}"),
+    }
+    Ok(())
+}
+
+#[api]
+/// Clear every pending assignment in one bulk transaction.
+async fn clear_pending() -> Result<(), Error> {
+    let cleared = client()?.subscription_clear_pending().await?;
+    if cleared == 0 {
+        println!("No pending assignments to clear.");
+    } else {
+        println!("Cleared {cleared} pending assignment(s).");
+    }
+    Ok(())
+}
diff --git a/lib/pdm-client/src/lib.rs b/lib/pdm-client/src/lib.rs
index 76b33ef..b0527b1 100644
--- a/lib/pdm-client/src/lib.rs
+++ b/lib/pdm-client/src/lib.rs
@@ -76,7 +76,10 @@ pub mod types {
 
     pub use pve_api_types::StorageStatus as PveStorageStatus;
 
-    pub use pdm_api_types::subscription::{RemoteSubscriptionState, RemoteSubscriptions};
+    pub use pdm_api_types::subscription::{
+        ClearPendingResult, ProductType, ProposedAssignment, RemoteNodeStatus,
+        RemoteSubscriptionState, RemoteSubscriptions, SubscriptionKeyEntry, SubscriptionKeySource,
+    };
 
     pub use pve_api_types::{SdnVnetMacVrf, SdnZoneIpVrf};
 }
@@ -1089,6 +1092,127 @@ impl<T: HttpApiClient> PdmClient<T> {
         Ok(self.0.get(&path).await?.expect_json()?.data)
     }
 
+    /// List all keys in the subscription pool. Returns the entries plus the matching
+    /// `ConfigDigest` so the caller can chain a digest-aware add / assign / delete back.
+    pub async fn list_subscription_keys(
+        &self,
+    ) -> Result<(Vec<SubscriptionKeyEntry>, Option<ConfigDigest>), Error> {
+        let mut res = self
+            .0
+            .get("/api2/extjs/subscriptions/keys")
+            .await?
+            .expect_json()?;
+        Ok((res.data, res.attribs.remove("digest").map(ConfigDigest)))
+    }
+
+    /// Add one or more keys to the pool. See the daemon-side endpoint for the all-or-nothing
+    /// validation semantics.
+    pub async fn add_subscription_keys(
+        &self,
+        keys: &[String],
+        digest: Option<ConfigDigest>,
+    ) -> Result<(), Error> {
+        #[derive(Serialize)]
+        struct AddArgs<'a> {
+            keys: &'a [String],
+            #[serde(skip_serializing_if = "Option::is_none")]
+            digest: Option<ConfigDigest>,
+        }
+        self.0
+            .post("/api2/extjs/subscriptions/keys", &AddArgs { keys, digest })
+            .await?
+            .nodata()
+    }
+
+    /// Assign a key to a remote node. Pass `None` for both `remote` and `node` to clear the
+    /// assignment instead.
+    pub async fn assign_subscription_key(
+        &self,
+        key: &str,
+        remote: Option<&str>,
+        node: Option<&str>,
+        digest: Option<ConfigDigest>,
+    ) -> Result<(), Error> {
+        #[derive(Serialize)]
+        struct AssignArgs<'a> {
+            #[serde(skip_serializing_if = "Option::is_none")]
+            remote: Option<&'a str>,
+            #[serde(skip_serializing_if = "Option::is_none")]
+            node: Option<&'a str>,
+            #[serde(skip_serializing_if = "Option::is_none")]
+            digest: Option<ConfigDigest>,
+        }
+        let path = format!("/api2/extjs/subscriptions/keys/{key}");
+        self.0
+            .put(
+                &path,
+                &AssignArgs {
+                    remote,
+                    node,
+                    digest,
+                },
+            )
+            .await?
+            .nodata()
+    }
+
+    /// Remove a key from the pool entirely.
+    ///
+    /// No digest parameter: deletion is a point-of-no-return operation and the typed-client
+    /// surface elsewhere (delete_remote, delete_user, ...) does not round-trip a digest on
+    /// DELETE either. External REST callers can still pass `digest` via the URL query if they
+    /// want optimistic concurrency on deletion; the server-side endpoint accepts it.
+    pub async fn delete_subscription_key(&self, key: &str) -> Result<(), Error> {
+        let path = format!("/api2/extjs/subscriptions/keys/{key}");
+        self.0.delete(&path).await?.nodata()
+    }
+
+    /// Combined remote/node subscription status, filtered to remotes the caller has audit
+    /// privilege on.
+    pub async fn subscription_node_status(
+        &self,
+        max_age: Option<u64>,
+    ) -> Result<Vec<RemoteNodeStatus>, Error> {
+        let path = ApiPathBuilder::new("/api2/extjs/subscriptions/node-status")
+            .maybe_arg("max-age", &max_age)
+            .build();
+        Ok(self.0.get(&path).await?.expect_json()?.data)
+    }
+
+    /// Propose (or apply) automatic key-to-node assignments.
+    pub async fn subscription_auto_assign(
+        &self,
+        apply: bool,
+    ) -> Result<Vec<ProposedAssignment>, Error> {
+        let path = ApiPathBuilder::new("/api2/extjs/subscriptions/auto-assign")
+            .arg("apply", &apply)
+            .build();
+        Ok(self.0.post(&path, &json!({})).await?.expect_json()?.data)
+    }
+
+    /// Push every pending assignment. Returns the worker UPID, or `None` when there is nothing
+    /// to do.
+    pub async fn subscription_apply_pending(&self) -> Result<Option<String>, Error> {
+        Ok(self
+            .0
+            .post("/api2/extjs/subscriptions/apply-pending", &json!({}))
+            .await?
+            .expect_json()?
+            .data)
+    }
+
+    /// Clear every pending assignment in one bulk transaction; returns the count of cleared
+    /// entries.
+    pub async fn subscription_clear_pending(&self) -> Result<u32, Error> {
+        let result: types::ClearPendingResult = self
+            .0
+            .post("/api2/extjs/subscriptions/clear-pending", &json!({}))
+            .await?
+            .expect_json()?
+            .data;
+        Ok(result.cleared)
+    }
+
     pub async fn pve_list_networks(
         &self,
         remote: &str,
-- 
2.47.3