[PATCH datacenter-manager v3 10/12] subscription: add Adopt All bulk action

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

Add a server endpoint plus CLI / UI wiring for importing every
foreign live subscription in one transaction. The typical use case
is connecting an existing fleet of PVE/PBS nodes to PDM for the
first time: rather than clicking Adopt Key per-node, the operator
runs Adopt All once and the pool catches up with the deployed
subscriptions in a single call.

The candidate set is recomputed under the config lock, so a
parallel Assign / Adopt landing between the network read and the
lock cannot race-import a key that has just been bound. Candidates
are silently skipped on missing per-remote PRIV_RESOURCE_MODIFY,
on a conflicting pool state ((remote, node) target already bound,
or the live key already bound elsewhere), or on a remote-supplied
key or node name failing schema validation; the UI dialog
enumerates the same set.

Signed-off-by: Thomas Lamprecht <t.lamprecht@proxmox.com>
---

New in v3.

 cli/client/src/subscriptions.rs               |  29 ++++
 docs/subscription-registry.rst                |   7 +
 lib/pdm-api-types/src/subscription.rs         |  17 ++
 lib/pdm-client/src/lib.rs                     |  22 +++
 server/src/api/subscriptions/mod.rs           | 161 +++++++++++++++++-
 ui/src/configuration/subscription_registry.rs | 160 +++++++++++++++++
 6 files changed, 394 insertions(+), 2 deletions(-)

diff --git a/cli/client/src/subscriptions.rs b/cli/client/src/subscriptions.rs
index c9ba5e4c..469f0841 100644
--- a/cli/client/src/subscriptions.rs
+++ b/cli/client/src/subscriptions.rs
@@ -52,6 +52,7 @@ pub fn cli() -> CommandLineInterface {
             "adopt-key",
             CliCommand::new(&API_METHOD_ADOPT_KEY).arg_param(&["remote", "node"]),
         )
+        .insert("adopt-all", CliCommand::new(&API_METHOD_ADOPT_ALL))
         .into()
 }
 
@@ -342,6 +343,34 @@ async fn adopt_key(
     Ok(())
 }
 
+#[api(
+    input: {
+        properties: {
+            digest: {
+                schema: PROXMOX_CONFIG_DIGEST_SCHEMA,
+                optional: true,
+            },
+        },
+    },
+)]
+/// Adopt every foreign live subscription into the pool in one transaction.
+///
+/// Walks all remotes the caller can audit, imports any (remote, node) with a live current key
+/// and no pool binding. Candidates the caller has no modify privilege on, or whose key is
+/// already bound elsewhere in the pool, are silently skipped.
+async fn adopt_all(digest: Option<String>) -> Result<(), Error> {
+    let digest = digest.map(ConfigDigest::from);
+    let adopted = client()?.subscription_adopt_all(digest).await?;
+    if adopted.is_empty() {
+        println!("No foreign live subscriptions to adopt.");
+        return Ok(());
+    }
+    println!("Adopted {} live subscription(s):", adopted.len());
+    for e in &adopted {
+        println!("  {}/{} -> {}", e.remote, e.node, e.key);
+    }
+    Ok(())
+}
 
 #[api(
     input: {
diff --git a/docs/subscription-registry.rst b/docs/subscription-registry.rst
index 4c31c9a6..6d599fe2 100644
--- a/docs/subscription-registry.rst
+++ b/docs/subscription-registry.rst
@@ -52,6 +52,13 @@ adoption are highlighted with a download hint icon in the Node Subscription Stat
 the pool grid carries a hidden-by-default Source column distinguishing manually-added from
 adopted entries, which can be enabled via the column picker if the distinction matters.
 
+The Adopt All action runs the same import across every remote the operator can audit in one
+transaction. Use it after first connecting an existing fleet of nodes to PDM so the pool
+catches up with the live subscriptions already deployed, without having to click through
+Adopt Key for each node. Candidates the operator has no modify privilege on, whose key is
+already bound elsewhere in the pool, whose (remote, node) target is already bound by another
+pool entry, or whose key or node name fails schema validation are skipped silently.
+
 The proposed plan can be inspected before it is applied. Apply Pending walks the queue in
 order; if any push or clear fails the remaining queue is kept intact for retry. Discard Pending
 drops the plan without touching any remote.
diff --git a/lib/pdm-api-types/src/subscription.rs b/lib/pdm-api-types/src/subscription.rs
index 8a0a7977..df1fec1c 100644
--- a/lib/pdm-api-types/src/subscription.rs
+++ b/lib/pdm-api-types/src/subscription.rs
@@ -569,6 +569,23 @@ pub struct ClearPendingResult {
     pub cleared: u32,
 }
 
+#[api(
+    properties: {
+        "key": { schema: SUBSCRIPTION_KEY_SCHEMA },
+    },
+)]
+#[derive(Clone, Debug, Serialize, Deserialize, PartialEq, Eq)]
+#[serde(rename_all = "kebab-case")]
+/// One entry imported by the bulk Adopt-All endpoint.
+pub struct AdoptedEntry {
+    /// Remote the live subscription was running on.
+    pub remote: String,
+    /// Node within the remote.
+    pub node: String,
+    /// The adopted subscription key.
+    pub key: String,
+}
+
 #[api]
 #[derive(Clone, Debug, Default, Serialize, Deserialize, PartialEq, Eq)]
 #[serde(rename_all = "kebab-case")]
diff --git a/lib/pdm-client/src/lib.rs b/lib/pdm-client/src/lib.rs
index 6c764c00..f03f6c40 100644
--- a/lib/pdm-client/src/lib.rs
+++ b/lib/pdm-client/src/lib.rs
@@ -1303,6 +1303,28 @@ impl<T: HttpApiClient> PdmClient<T> {
             .nodata()
     }
 
+    /// Adopt every foreign live subscription that the caller can modify, in one transaction.
+    /// Returns the list of `(remote, node, key)` tuples that were imported into the pool;
+    /// candidates the caller has no `PRIV_RESOURCE_MODIFY` on (or that fail validation, or that
+    /// are already bound elsewhere in the pool) are silently skipped. See the server endpoint
+    /// docs for the full skip rules.
+    pub async fn subscription_adopt_all(
+        &self,
+        digest: Option<ConfigDigest>,
+    ) -> Result<Vec<pdm_api_types::subscription::AdoptedEntry>, Error> {
+        #[derive(Serialize)]
+        struct AdoptAllArgs {
+            #[serde(skip_serializing_if = "Option::is_none")]
+            digest: Option<ConfigDigest>,
+        }
+        Ok(self
+            .0
+            .post("/api2/extjs/subscriptions/adopt-all", &AdoptAllArgs { digest })
+            .await?
+            .expect_json()?
+            .data)
+    }
+
     /// Queue a clear for the subscription on `remote`/`node`. Apply Pending later removes the
     /// subscription from the node so the key can be reassigned elsewhere; Discard Pending
     /// undoes the queueing without touching the remote. Returns `BAD_REQUEST` if no pool entry
diff --git a/server/src/api/subscriptions/mod.rs b/server/src/api/subscriptions/mod.rs
index cc46806c..a8f5cfc5 100644
--- a/server/src/api/subscriptions/mod.rs
+++ b/server/src/api/subscriptions/mod.rs
@@ -21,8 +21,8 @@ use proxmox_sortable_macro::sortable;
 
 use pdm_api_types::remotes::{Remote, REMOTE_ID_SCHEMA};
 use pdm_api_types::subscription::{
-    pick_best_pve_socket_key, socket_count_from_key, AutoAssignProposal, ClearPendingResult,
-    ProductType, ProposedAssignment, RemoteNodeStatus, SubscriptionKeyEntry,
+    pick_best_pve_socket_key, socket_count_from_key, AdoptedEntry, AutoAssignProposal,
+    ClearPendingResult, ProductType, ProposedAssignment, RemoteNodeStatus, SubscriptionKeyEntry,
     SubscriptionKeySource, SubscriptionLevel, SUBSCRIPTION_KEY_SCHEMA,
 };
 use pdm_api_types::{
@@ -39,6 +39,7 @@ pub const ROUTER: Router = Router::new()
 
 #[sortable]
 const SUBDIRS: SubdirMap = &sorted!([
+    ("adopt-all", &Router::new().post(&API_METHOD_ADOPT_ALL)),
     ("adopt-key", &Router::new().post(&API_METHOD_ADOPT_KEY)),
     (
         "apply-pending",
@@ -1112,6 +1113,162 @@ async fn adopt_key(
     Ok(())
 }
 
+#[api(
+    input: {
+        properties: {
+            digest: {
+                type: ConfigDigest,
+                optional: true,
+            },
+        },
+    },
+    returns: {
+        type: Array,
+        description: "List of (remote, node, key) tuples that were adopted into the pool.",
+        items: { type: AdoptedEntry },
+    },
+    access: {
+        permission: &Permission::Privilege(&["system"], PRIV_SYS_MODIFY, false),
+    },
+)]
+/// Adopt every foreign live subscription in one transaction.
+///
+/// Walks the node-status view (so only remotes the caller can audit are considered), collects
+/// every (remote, node) that has a live current key but no pool entry bound to it, and imports
+/// each one into the pool with source = `Adopted`. Candidates are skipped (not adopted, not an
+/// error) when:
+///
+/// - The caller has no `PRIV_RESOURCE_MODIFY` on the candidate's remote: an audit-only operator
+///   should not be able to materialise pool state for a remote they cannot manage.
+/// - The live key is already in the pool but bound elsewhere: leaving the rebind as a manual
+///   step keeps the bulk action from silently competing with a deliberate prior assignment.
+/// - The live key fails schema validation or its prefix is unknown: a buggy or malicious
+///   remote should not be able to inject garbage into the pool through a bulk shortcut.
+///
+/// Successfully-adopted entries are returned so the caller (CLI / UI) can summarise the outcome
+/// without needing a separate refresh round-trip.
+async fn adopt_all(
+    digest: Option<ConfigDigest>,
+    rpcenv: &mut dyn RpcEnvironment,
+) -> Result<Vec<AdoptedEntry>, Error> {
+    let auth_id: Authid = rpcenv
+        .get_auth_id()
+        .context("no authid available")?
+        .parse()?;
+
+    // Use a fresh node-status snapshot: a cached entry from minutes ago could miss a live
+    // subscription that was just installed on a remote, or vice-versa, claim a subscription
+    // that has since been removed. Adopting bogus or already-cleared keys would be a footgun.
+    let node_statuses = collect_node_status(FRESH_NODE_STATUS_MAX_AGE, rpcenv).await?;
+
+    // Lock + sync IO under spawn_blocking. The closure re-resolves the candidate set under the
+    // lock: a parallel admin's Assign / Adopt between the network read above and the lock
+    // acquisition here would otherwise let us race-import a key that has just been bound by
+    // them.
+    let (adopted, new_digest_opt) = tokio::task::spawn_blocking(
+        move || -> Result<(Vec<AdoptedEntry>, Option<ConfigDigest>), Error> {
+            let user_info = CachedUserInfo::new()?;
+            let _lock = pdm_config::subscriptions::lock_config()?;
+            let (mut config, config_digest) = pdm_config::subscriptions::config()?;
+            config_digest.detect_modification(digest.as_ref())?;
+
+            let mut adopted: Vec<AdoptedEntry> = Vec::new();
+            for n in &node_statuses {
+                let Some(current_key) = n.current_key.as_deref() else {
+                    continue;
+                };
+                if n.assigned_key.is_some() {
+                    continue;
+                }
+                if user_info.lookup_privs(&auth_id, &["resource", &n.remote])
+                    & PRIV_RESOURCE_MODIFY
+                    == 0
+                {
+                    continue;
+                }
+                // Re-validate foreign node name: later interpolated into remote URL.
+                if NODE_SCHEMA.parse_simple_value(&n.node).is_err() {
+                    warn!(
+                        "skipping adopt-all candidate on {}/{}: node name fails schema",
+                        n.remote, n.node,
+                    );
+                    continue;
+                }
+                // Re-check binding state under the lock - between the network read and here a
+                // parallel Adopt / Assign on the same target could have created a pool entry
+                // bound to (remote, node) that the cached node-status snapshot did not see.
+                let target_bound = config.iter().any(|(_, e)| {
+                    e.remote.as_deref() == Some(n.remote.as_str())
+                        && e.node.as_deref() == Some(n.node.as_str())
+                });
+                if target_bound {
+                    continue;
+                }
+
+                if let Some(existing) = config.get_mut(current_key) {
+                    if existing.remote.is_some() || existing.node.is_some() {
+                        // Bound elsewhere: leave the rebind as an explicit operator decision.
+                        continue;
+                    }
+                    existing.remote = Some(n.remote.clone());
+                    existing.node = Some(n.node.clone());
+                } else {
+                    if SUBSCRIPTION_KEY_SCHEMA
+                        .parse_simple_value(current_key)
+                        .is_err()
+                    {
+                        warn!(
+                            "skipping adopt-all candidate on {}/{}: key '{}' fails schema",
+                            n.remote,
+                            n.node,
+                            redact_key(current_key),
+                        );
+                        continue;
+                    }
+                    let Some(product_type) = ProductType::from_key(current_key) else {
+                        warn!(
+                            "skipping adopt-all candidate on {}/{}: unrecognised key prefix \
+                             '{}'",
+                            n.remote,
+                            n.node,
+                            redact_key(current_key),
+                        );
+                        continue;
+                    };
+                    let entry = SubscriptionKeyEntry {
+                        key: current_key.to_string(),
+                        product_type,
+                        level: SubscriptionLevel::from_key(Some(current_key)),
+                        source: SubscriptionKeySource::Adopted,
+                        remote: Some(n.remote.clone()),
+                        node: Some(n.node.clone()),
+                        ..Default::default()
+                    };
+                    config.insert(current_key.to_string(), entry);
+                }
+                adopted.push(AdoptedEntry {
+                    remote: n.remote.clone(),
+                    node: n.node.clone(),
+                    key: current_key.to_string(),
+                });
+            }
+
+            let new_digest = if adopted.is_empty() {
+                None
+            } else {
+                Some(pdm_config::subscriptions::save_config(&config)?)
+            };
+            Ok((adopted, new_digest))
+        },
+    )
+    .await??;
+
+    if let Some(new_digest) = new_digest_opt {
+        rpcenv["digest"] = new_digest.to_hex().into();
+    }
+    Ok(adopted)
+}
+
 #[api(
     input: {
         properties: {
diff --git a/ui/src/configuration/subscription_registry.rs b/ui/src/configuration/subscription_registry.rs
index 7d79370b..b84ddb36 100644
--- a/ui/src/configuration/subscription_registry.rs
+++ b/ui/src/configuration/subscription_registry.rs
@@ -99,6 +99,14 @@ fn pending_badge(push_count: u32, clear_count: u32) -> Row {
     row
 }
 
+/// Row shape for the Adopt All preview table.
+#[derive(Clone, PartialEq)]
+struct AdoptCandidate {
+    remote: String,
+    node: String,
+    key: String,
+}
+
 #[derive(Clone, Debug, PartialEq)]
 enum NodeTreeEntry {
     Root,
@@ -238,6 +246,8 @@ pub enum Msg {
     /// Open the confirmation dialog for adopting the live subscription on the selected node
     /// into the pool.
     AdoptKeyForSelectedNode,
+    /// Open the confirmation dialog for adopting every foreign live subscription into the pool.
+    AdoptAllPreview,
 }
 
 #[derive(PartialEq)]
@@ -259,6 +269,12 @@ pub enum ViewState {
         node: String,
         current_key: String,
     },
+    /// Pending confirmation to bulk-adopt every foreign live subscription. The candidate list
+    /// is captured at view-open time so the dialog body can show the operator exactly what
+    /// will be imported; the server re-computes the set under the lock at commit time.
+    ConfirmAdoptAll {
+        candidates: Vec<(String, String, String)>,
+    },
     /// Assign a pool key to the given node. Opens from the right panel's Assign Key button.
     AssignKeyToNode {
         remote: String,
@@ -274,6 +290,7 @@ pub struct SubscriptionRegistryComp {
     tree_store: TreeStore<NodeTreeEntry>,
     tree_columns: Rc<Vec<DataTableHeader<NodeTreeEntry>>>,
     proposal_columns: Rc<Vec<DataTableHeader<ProposedAssignment>>>,
+    adopt_columns: Rc<Vec<DataTableHeader<AdoptCandidate>>>,
     node_selection: Selection,
     last_node_data: Vec<RemoteNodeStatus>,
     /// Canonical pool snapshot. Passed down to the key grid (display) and shared with the
@@ -458,6 +475,24 @@ impl SubscriptionRegistryComp {
                 .into(),
         ])
     }
+
+    fn adopt_columns() -> Rc<Vec<DataTableHeader<AdoptCandidate>>> {
+        Rc::new(vec![
+            DataTableColumn::new(tr!("Remote / Node"))
+                .flex(2)
+                .render(|c: &AdoptCandidate| format!("{} / {}", c.remote, c.node).into())
+                .into(),
+            DataTableColumn::new(tr!("Key"))
+                .flex(2)
+                .render(|c: &AdoptCandidate| {
+                    Container::from_tag("span")
+                        .class(pwt::css::FontStyle::LabelMedium)
+                        .with_child(c.key.clone())
+                        .into()
+                })
+                .into(),
+        ])
+    }
 }
 
 fn key_cell(n: &RemoteNodeStatus) -> Html {
@@ -542,6 +577,7 @@ impl LoadableComponent for SubscriptionRegistryComp {
             tree_store: store.clone(),
             tree_columns: Self::tree_columns(store),
             proposal_columns: Self::proposal_columns(),
+            adopt_columns: Self::adopt_columns(),
             node_selection,
             last_node_data: Vec::new(),
             pool_keys: Rc::new(Vec::new()),
@@ -692,6 +728,14 @@ impl LoadableComponent for SubscriptionRegistryComp {
                     current_key,
                 }));
             }
+            Msg::AdoptAllPreview => {
+                let candidates = self.adopt_all_candidates();
+                if candidates.is_empty() {
+                    return false;
+                }
+                ctx.link()
+                    .change_view(Some(ViewState::ConfirmAdoptAll { candidates }));
+            }
         }
         true
     }
@@ -699,6 +743,7 @@ impl LoadableComponent for SubscriptionRegistryComp {
     fn toolbar(&self, ctx: &LoadableComponentContext<Self>) -> Option<Html> {
         let link = ctx.link();
         let (push_count, clear_count) = self.pending_counts();
+        let adopt_all_count = self.adopt_all_candidates().len();
         let mut toolbar = Toolbar::new()
             .border_bottom(true)
             .with_child(
@@ -712,6 +757,18 @@ impl LoadableComponent for SubscriptionRegistryComp {
                      subscription, then queue it pending Apply."
                 )),
             )
+            .with_child(
+                Tooltip::new(
+                    Button::new(tr!("Adopt All"))
+                        .icon_class("fa fa-download")
+                        .disabled(adopt_all_count == 0)
+                        .on_activate(link.callback(|_| Msg::AdoptAllPreview)),
+                )
+                .tip(tr!(
+                    "Import every foreign live subscription that is not yet tracked by the \
+                     pool. The remote is not contacted; only the pool config is updated."
+                )),
+            )
             .with_spacer()
             .with_child(
                 Tooltip::new(
@@ -860,6 +917,9 @@ impl LoadableComponent for SubscriptionRegistryComp {
             ViewState::ConfirmAutoAssign(proposal) => {
                 Some(self.render_auto_assign_dialog(ctx, proposal))
             }
+            ViewState::ConfirmAdoptAll { candidates } => {
+                Some(self.render_adopt_all_dialog(ctx, candidates))
+            }
             ViewState::ConfirmAdoptKey {
                 remote,
                 node,
@@ -1175,6 +1235,25 @@ impl SubscriptionRegistryComp {
         Some((n.remote.clone(), n.node.clone(), current_key))
     }
 
+    /// Iterate the loaded node-status snapshot and return every `(remote, node, current_key)`
+    /// eligible for bulk Adopt-All (live key set, no pool binding). Used both for the toolbar
+    /// disabled gate and for the preview list in the confirm dialog; the authoritative set is
+    /// recomputed by the server under the lock at commit time, so this view is a hint, not a
+    /// contract.
+    fn adopt_all_candidates(&self) -> Vec<(String, String, String)> {
+        self.last_node_data
+            .iter()
+            .filter_map(|n| {
+                if n.assigned_key.is_some() {
+                    return None;
+                }
+                n.current_key
+                    .clone()
+                    .map(|k| (n.remote.clone(), n.node.clone(), k))
+            })
+            .collect()
+    }
+
     /// Returns `(remote, node, type, node_sockets)` for the right-panel Assign button:
     /// selected row is a node, no assigned key in the pool yet, and no live active subscription.
     /// Refusing earlier than the server keeps the button-disable affordance honest.
@@ -1250,4 +1329,85 @@ impl SubscriptionRegistryComp {
             .with_child(body)
             .into()
     }
+
+    fn render_adopt_all_dialog(
+        &self,
+        ctx: &LoadableComponentContext<Self>,
+        candidates: &[(String, String, String)],
+    ) -> Html {
+        use pwt::widget::Dialog;
+
+        let rows: Vec<AdoptCandidate> = candidates
+            .iter()
+            .map(|(r, n, k)| AdoptCandidate {
+                remote: r.clone(),
+                node: n.clone(),
+                key: k.clone(),
+            })
+            .collect();
+        let n = rows.len();
+        let store: Store<AdoptCandidate> = Store::with_extract_key(|c: &AdoptCandidate| {
+            format!("{}/{}", c.remote, c.node).into()
+        });
+        store.set_data(rows);
+
+        let link_close = ctx.link().clone();
+        let link_apply = ctx.link().clone();
+        let digest = self.pool_digest.clone();
+        let body = Column::new()
+            .class(Flex::Fill)
+            .class(Overflow::Hidden)
+            .min_height(0)
+            .padding(2)
+            .gap(2)
+            .min_width(600)
+            .with_child(Container::from_tag("p").with_child(tr!(
+                "The following {n} live subscription(s) will be imported into the pool; \
+                 the remote is not contacted.",
+                n = n,
+            )))
+            .with_child(
+                DataTable::new(self.adopt_columns.clone(), store)
+                    .striped(true)
+                    .class(FlexFit)
+                    .min_height(140),
+            )
+            .with_child(
+                Row::new()
+                    .class(JustifyContent::FlexEnd)
+                    .gap(2)
+                    .padding_top(2)
+                    .with_child(
+                        Button::new(tr!("Cancel"))
+                            .on_activate(move |_| link_close.change_view(None)),
+                    )
+                    .with_child(Button::new(tr!("Adopt")).on_activate(move |_| {
+                        let link = link_apply.clone();
+                        let digest = digest.clone();
+                        link.clone().spawn(async move {
+                            let digest = digest.map(pdm_client::ConfigDigest::from);
+                            if let Err(err) =
+                                crate::pdm_client().subscription_adopt_all(digest).await
+                            {
+                                link.show_error(tr!("Adopt All"), err.to_string(), true);
+                            }
+                            link.change_view(None);
+                            link.send_reload();
+                        });
+                    })),
+            );
+
+        Dialog::new(tr!("Adopt All"))
+            .resizable(true)
+            .width(700)
+            .min_width(500)
+            .min_height(300)
+            .max_height("80vh")
+            .on_close({
+                let link = ctx.link().clone();
+                move |_| link.change_view(None)
+            })
+            .with_child(body)
+            .into()
+    }
 }
-- 
2.47.3