magma_plugin/
import.rs

1//! magma-plugin::import — the typed `ImportResourceState` gRPC client
2//! call.
3//!
4//! This is the missing keystone of magma's adopt-a-pre-existing-resource
5//! capability. Given a dialed provider channel, a resource `type_name`,
6//! and a provider-specific `id`, it drives the provider's
7//! `ImportResourceState` RPC and decodes each returned `DynamicValue`
8//! state into typed `magma_types::ImportedInstance` values.
9//!
10//! It follows the exact RPC-call pattern the lifecycle integration
11//! tests use for `PlanResourceChange` / `ApplyResourceChange`:
12//! `ProviderClient::new(channel)` then `client.method(req).await`. The
13//! `DynamicValue` decode uses the same `json`-field path
14//! `PlanResourceChange` uses on the wire (magma has no msgpack decoder;
15//! a state that arrives only as msgpack surfaces a typed error rather
16//! than a silent wrong answer).
17
18use magma_protocol::tfplugin6;
19use magma_protocol::tfplugin6::provider_client::ProviderClient;
20use magma_types::ImportedInstance;
21
22use crate::{H2Channel, PluginError};
23
24/// Decode a tfplugin6 `DynamicValue` into a typed `serde_json::Value`.
25///
26/// magma's wire path uses the `json` field (the same field
27/// `PlanResourceChange`/`ApplyResourceChange` round-trip in the
28/// lifecycle integration suite). A `DynamicValue` carrying ONLY
29/// msgpack surfaces a typed error — magma has no msgpack decoder yet,
30/// and a silent wrong answer is forbidden (theory/MAGMA.md §IX +
31/// the TYPED-SPEC rule).
32fn decode_dynamic_value(dv: &tfplugin6::DynamicValue) -> Result<serde_json::Value, PluginError> {
33    if !dv.json.is_empty() {
34        return serde_json::from_slice(&dv.json).map_err(|e| {
35            PluginError::ImportDecode(format!("DynamicValue.json is not valid JSON: {e}"))
36        });
37    }
38    if !dv.msgpack.is_empty() {
39        return Err(PluginError::ImportDecode(
40            "provider returned imported state as msgpack only; magma's wire decoder \
41             reads the DynamicValue.json field (no msgpack decoder yet)"
42                .into(),
43        ));
44    }
45    // An empty DynamicValue is a legitimate "null" state — decode to
46    // JSON null so the caller absorbs an empty-attributes instance
47    // rather than erroring.
48    Ok(serde_json::Value::Null)
49}
50
51/// Render the provider's `Diagnostic` list into one error string. Only
52/// `ERROR`-severity diagnostics are fatal; warnings/info are dropped
53/// (they don't fail an import).
54fn fatal_diagnostics(diags: &[tfplugin6::Diagnostic]) -> Option<String> {
55    let errors: Vec<String> = diags
56        .iter()
57        .filter(|d| d.severity == tfplugin6::diagnostic::Severity::Error as i32)
58        .map(|d| {
59            if d.detail.is_empty() {
60                d.summary.clone()
61            } else {
62                format!("{}: {}", d.summary, d.detail)
63            }
64        })
65        .collect();
66    if errors.is_empty() {
67        None
68    } else {
69        Some(errors.join("; "))
70    }
71}
72
73/// Drive the provider's `ImportResourceState` RPC for `(type_name, id)`
74/// over a dialed gRPC `channel`, returning the typed imported
75/// resources.
76///
77/// This is the in-process equivalent of `tofu import <addr> <id>`: it
78/// asks the provider "given this id, what is the live resource's
79/// state?" and hands back the decoded attributes the apply prepass
80/// absorbs into the working state.
81///
82/// # Errors
83///
84/// * `PluginError::ImportRpc` — the gRPC call itself failed (transport
85///   error, provider crashed, etc.).
86/// * `PluginError::ImportRejected` — the provider returned ERROR-level
87///   diagnostics (bad id, resource not found, type doesn't support
88///   import).
89/// * `PluginError::ImportDecode` — a returned `DynamicValue` couldn't
90///   be decoded via the json path.
91pub async fn import_resource_state(
92    channel: H2Channel,
93    type_name: &str,
94    id: &str,
95) -> Result<Vec<ImportedInstance>, PluginError> {
96    let mut client = ProviderClient::new(channel);
97
98    let req = tfplugin6::import_resource_state::Request {
99        type_name: type_name.to_string(),
100        id: id.to_string(),
101        client_capabilities: crate::provider::client_caps_v6(),
102        identity: None,
103    };
104
105    let resp = client
106        .import_resource_state(req)
107        .await
108        .map_err(|status| {
109            PluginError::ImportRpc(format!(
110                "ImportResourceState RPC for {type_name} id={id:?} failed: {status}"
111            ))
112        })?
113        .into_inner();
114
115    if let Some(reason) = fatal_diagnostics(&resp.diagnostics) {
116        return Err(PluginError::ImportRejected {
117            type_name: type_name.to_string(),
118            id: id.to_string(),
119            reason,
120        });
121    }
122
123    let mut imported = Vec::with_capacity(resp.imported_resources.len());
124    for res in resp.imported_resources {
125        let attributes = match res.state.as_ref() {
126            Some(dv) => decode_dynamic_value(dv)?,
127            None => serde_json::Value::Null,
128        };
129        imported.push(ImportedInstance {
130            type_name: res.type_name,
131            attributes,
132            private: res.private,
133        });
134    }
135
136    Ok(imported)
137}
138
139#[cfg(test)]
140mod tests {
141    use super::*;
142
143    #[test]
144    fn decode_json_dynamic_value() {
145        let dv = tfplugin6::DynamicValue {
146            msgpack: vec![],
147            json: br#"{"id":"role-1","name":"cluster-role"}"#.to_vec(),
148        };
149        let v = decode_dynamic_value(&dv).unwrap();
150        assert_eq!(v["id"], "role-1");
151        assert_eq!(v["name"], "cluster-role");
152    }
153
154    #[test]
155    fn decode_empty_dynamic_value_is_null() {
156        let dv = tfplugin6::DynamicValue {
157            msgpack: vec![],
158            json: vec![],
159        };
160        assert_eq!(decode_dynamic_value(&dv).unwrap(), serde_json::Value::Null);
161    }
162
163    #[test]
164    fn decode_msgpack_only_is_typed_error() {
165        let dv = tfplugin6::DynamicValue {
166            msgpack: vec![0x81, 0xa2, 0x69, 0x64],
167            json: vec![],
168        };
169        assert!(matches!(
170            decode_dynamic_value(&dv),
171            Err(PluginError::ImportDecode(_))
172        ));
173    }
174
175    #[test]
176    fn fatal_diagnostics_extracts_errors_only() {
177        let diags = vec![
178            tfplugin6::Diagnostic {
179                severity: tfplugin6::diagnostic::Severity::Warning as i32,
180                summary: "a warning".into(),
181                detail: String::new(),
182                attribute: None,
183            },
184            tfplugin6::Diagnostic {
185                severity: tfplugin6::diagnostic::Severity::Error as i32,
186                summary: "bad id".into(),
187                detail: "no such resource".into(),
188                attribute: None,
189            },
190        ];
191        let reason = fatal_diagnostics(&diags).unwrap();
192        assert!(reason.contains("bad id"));
193        assert!(reason.contains("no such resource"));
194        assert!(!reason.contains("a warning"));
195    }
196
197    #[test]
198    fn no_fatal_diagnostics_when_all_warnings() {
199        let diags = vec![tfplugin6::Diagnostic {
200            severity: tfplugin6::diagnostic::Severity::Warning as i32,
201            summary: "just a warning".into(),
202            detail: String::new(),
203            attribute: None,
204        }];
205        assert!(fatal_diagnostics(&diags).is_none());
206    }
207}
magma_plugin/import.rs

magma_plugin/
import.rs
