Skip to main content

fakecloud_managedblockchain/
service.rs

1//! Amazon Managed Blockchain (`managedblockchain`) restJson1 dispatch + handlers.
2//!
3//! The full 27-operation Managed Blockchain control plane. Requests are routed
4//! to an operation by HTTP method + `@http` URI path under `/`; path labels are
5//! captured positionally (percent-decoded, so an ARN label whose slashes/colons
6//! arrive percent-encoded survives intact) and query parameters are read from
7//! the raw query string. State is account-partitioned and persisted; each
8//! resource is stored as its already-output-valid wire object so `Get*` echoes
9//! exactly what `Create*` persisted.
10
11use std::sync::Arc;
12
13use async_trait::async_trait;
14use http::{Method, StatusCode};
15use percent_encoding::percent_decode_str;
16use serde_json::{json, Map, Value};
17use tokio::sync::Mutex as AsyncMutex;
18
19use fakecloud_core::service::{AwsRequest, AwsResponse, AwsService, AwsServiceError};
20use fakecloud_persistence::SnapshotStore;
21
22use crate::persistence::save_snapshot;
23use crate::shared;
24use crate::state::{ManagedBlockchainData, SharedManagedBlockchainState};
25
26/// Every operation name in the Amazon Managed Blockchain Smithy model (27).
27pub const MANAGEDBLOCKCHAIN_ACTIONS: &[&str] = &[
28    "CreateAccessor",
29    "CreateMember",
30    "CreateNetwork",
31    "CreateNode",
32    "CreateProposal",
33    "DeleteAccessor",
34    "DeleteMember",
35    "DeleteNode",
36    "GetAccessor",
37    "GetMember",
38    "GetNetwork",
39    "GetNode",
40    "GetProposal",
41    "ListAccessors",
42    "ListInvitations",
43    "ListMembers",
44    "ListNetworks",
45    "ListNodes",
46    "ListProposalVotes",
47    "ListProposals",
48    "ListTagsForResource",
49    "RejectInvitation",
50    "TagResource",
51    "UntagResource",
52    "UpdateMember",
53    "UpdateNode",
54    "VoteOnProposal",
55];
56
57/// Operations that mutate persisted state on success (so a snapshot is taken).
58const MUTATING: &[&str] = &[
59    "CreateAccessor",
60    "CreateMember",
61    "CreateNetwork",
62    "CreateNode",
63    "CreateProposal",
64    "DeleteAccessor",
65    "DeleteMember",
66    "DeleteNode",
67    "RejectInvitation",
68    "TagResource",
69    "UntagResource",
70    "UpdateMember",
71    "UpdateNode",
72    "VoteOnProposal",
73];
74
75pub struct ManagedBlockchainService {
76    state: SharedManagedBlockchainState,
77    snapshot_store: Option<Arc<dyn SnapshotStore>>,
78    snapshot_lock: Arc<AsyncMutex<()>>,
79}
80
81impl ManagedBlockchainService {
82    pub fn new(state: SharedManagedBlockchainState) -> Self {
83        Self {
84            state,
85            snapshot_store: None,
86            snapshot_lock: Arc::new(AsyncMutex::new(())),
87        }
88    }
89
90    pub fn with_snapshot_store(mut self, store: Arc<dyn SnapshotStore>) -> Self {
91        self.snapshot_store = Some(store);
92        self
93    }
94
95    async fn save(&self) {
96        save_snapshot(
97            &self.state,
98            self.snapshot_store.clone(),
99            &self.snapshot_lock,
100        )
101        .await;
102    }
103
104    /// Persist hook for the CloudFormation provisioner; `None` in memory mode.
105    pub fn snapshot_hook(&self) -> Option<fakecloud_persistence::SnapshotHook> {
106        let store = self.snapshot_store.clone()?;
107        let state = self.state.clone();
108        let lock = self.snapshot_lock.clone();
109        Some(Arc::new(move || {
110            let state = state.clone();
111            let store = store.clone();
112            let lock = lock.clone();
113            Box::pin(async move {
114                save_snapshot(&state, Some(store), &lock).await;
115            })
116        }))
117    }
118
119    /// Settle any in-flight resource lifecycle transition for the account:
120    /// members and nodes created `CREATING` settle to `AVAILABLE`, and any
121    /// `IN_PROGRESS` proposal whose `ExpirationDate` has passed settles to
122    /// `EXPIRED`. Returns `true` if a transition fired (so the caller persists).
123    fn reconcile(&self, account: &str) -> bool {
124        let now = shared::iso_now();
125        let mut guard = self.state.write();
126        let data = guard.get_or_create(account);
127        let mut changed = false;
128        for m in data.members.values_mut() {
129            if promote_creating(m) {
130                changed = true;
131            }
132        }
133        for n in data.nodes.values_mut() {
134            if promote_creating(n) {
135                changed = true;
136            }
137        }
138        for p in data.proposals.values_mut() {
139            if let Some(obj) = p.as_object_mut() {
140                let status = obj.get("Status").and_then(Value::as_str).unwrap_or("");
141                let expired = obj
142                    .get("ExpirationDate")
143                    .and_then(Value::as_str)
144                    .map(|e| e < now.as_str())
145                    .unwrap_or(false);
146                if status == "IN_PROGRESS" && expired {
147                    obj.insert("Status".into(), json!("EXPIRED"));
148                    changed = true;
149                }
150            }
151        }
152        changed
153    }
154
155    /// Route a request to an operation name + captured path labels by HTTP
156    /// method + `@http` URI path. Returns `None` when no route matches.
157    fn resolve_action(req: &AwsRequest) -> Option<(&'static str, Vec<String>)> {
158        let raw = req.raw_path.split('?').next().unwrap_or(&req.raw_path);
159        let trimmed = raw.strip_prefix('/').unwrap_or(raw);
160        let segs: Vec<String> = if trimmed.is_empty() {
161            Vec::new()
162        } else {
163            trimmed
164                .split('/')
165                .map(|s| percent_decode_str(s).decode_utf8_lossy().into_owned())
166                .collect()
167        };
168        let s: Vec<&str> = segs.iter().map(String::as_str).collect();
169        let m = &req.method;
170        let get = m == Method::GET;
171        let post = m == Method::POST;
172        let del = m == Method::DELETE;
173        let patch = m == Method::PATCH;
174        let l = |v: &[&str]| v.iter().map(|x| x.to_string()).collect::<Vec<_>>();
175        let (action, labels): (&'static str, Vec<String>) = match s.as_slice() {
176            ["networks"] if post => ("CreateNetwork", vec![]),
177            ["networks"] if get => ("ListNetworks", vec![]),
178            ["networks", nid] if get => ("GetNetwork", l(&[nid])),
179            ["networks", nid, "members"] if post => ("CreateMember", l(&[nid])),
180            ["networks", nid, "members"] if get => ("ListMembers", l(&[nid])),
181            ["networks", nid, "members", mid] if get => ("GetMember", l(&[nid, mid])),
182            ["networks", nid, "members", mid] if patch => ("UpdateMember", l(&[nid, mid])),
183            ["networks", nid, "members", mid] if del => ("DeleteMember", l(&[nid, mid])),
184            ["networks", nid, "nodes"] if post => ("CreateNode", l(&[nid])),
185            ["networks", nid, "nodes"] if get => ("ListNodes", l(&[nid])),
186            ["networks", nid, "nodes", ndid] if get => ("GetNode", l(&[nid, ndid])),
187            ["networks", nid, "nodes", ndid] if patch => ("UpdateNode", l(&[nid, ndid])),
188            ["networks", nid, "nodes", ndid] if del => ("DeleteNode", l(&[nid, ndid])),
189            ["networks", nid, "proposals"] if post => ("CreateProposal", l(&[nid])),
190            ["networks", nid, "proposals"] if get => ("ListProposals", l(&[nid])),
191            ["networks", nid, "proposals", pid] if get => ("GetProposal", l(&[nid, pid])),
192            ["networks", nid, "proposals", pid, "votes"] if get => {
193                ("ListProposalVotes", l(&[nid, pid]))
194            }
195            ["networks", nid, "proposals", pid, "votes"] if post => {
196                ("VoteOnProposal", l(&[nid, pid]))
197            }
198            ["accessors"] if post => ("CreateAccessor", vec![]),
199            ["accessors"] if get => ("ListAccessors", vec![]),
200            ["accessors", aid] if get => ("GetAccessor", l(&[aid])),
201            ["accessors", aid] if del => ("DeleteAccessor", l(&[aid])),
202            ["invitations"] if get => ("ListInvitations", vec![]),
203            ["invitations", iid] if del => ("RejectInvitation", l(&[iid])),
204            ["tags", arn] if post => ("TagResource", l(&[arn])),
205            ["tags", arn] if get => ("ListTagsForResource", l(&[arn])),
206            ["tags", arn] if del => ("UntagResource", l(&[arn])),
207            _ => return None,
208        };
209        Some((action, labels))
210    }
211}
212
213#[async_trait]
214impl AwsService for ManagedBlockchainService {
215    fn service_name(&self) -> &str {
216        "managedblockchain"
217    }
218
219    async fn handle(&self, req: AwsRequest) -> Result<AwsResponse, AwsServiceError> {
220        let Some((action, labels)) = Self::resolve_action(&req) else {
221            return Err(AwsServiceError::aws_error(
222                StatusCode::NOT_FOUND,
223                "ResourceNotFoundException",
224                format!("Unknown operation: {} {}", req.method, req.raw_path),
225            ));
226        };
227        let (result, settled) = self.dispatch(action, &labels, &req);
228        let success = matches!(result.as_ref(), Ok(resp) if resp.status.is_success());
229        if settled || (MUTATING.contains(&action) && success) {
230            self.save().await;
231        }
232        result
233    }
234
235    fn supported_actions(&self) -> &[&str] {
236        MANAGEDBLOCKCHAIN_ACTIONS
237    }
238}
239
240/// Per-request account + region context.
241struct Ctx {
242    account: String,
243    region: String,
244}
245
246impl ManagedBlockchainService {
247    fn dispatch(
248        &self,
249        action: &str,
250        labels: &[String],
251        req: &AwsRequest,
252    ) -> (Result<AwsResponse, AwsServiceError>, bool) {
253        let body = match parse_body(req) {
254            Ok(b) => b,
255            Err(e) => return (Err(e), false),
256        };
257        if let Err(e) = crate::validate::validate_input(action, &body) {
258            return (Err(e), false);
259        }
260        let ctx = Ctx {
261            account: req.account_id.clone(),
262            region: req.region.clone(),
263        };
264        let q = parse_query(&req.raw_query);
265        let settled = self.reconcile(&ctx.account);
266        let a = |i: usize| labels.get(i).map(String::as_str).unwrap_or_default();
267        let result = match action {
268            "CreateNetwork" => self.create_network(&ctx, &body),
269            "GetNetwork" => self.get_network(&ctx, a(0)),
270            "ListNetworks" => self.list_networks(&ctx, &q),
271            "CreateMember" => self.create_member(&ctx, a(0), &body),
272            "GetMember" => self.get_member(&ctx, a(0), a(1)),
273            "UpdateMember" => self.update_member(&ctx, a(0), a(1), &body),
274            "DeleteMember" => self.delete_member(&ctx, a(0), a(1)),
275            "ListMembers" => self.list_members(&ctx, a(0), &q),
276            "CreateNode" => self.create_node(&ctx, a(0), &body),
277            "GetNode" => self.get_node(&ctx, a(0), a(1)),
278            "UpdateNode" => self.update_node(&ctx, a(0), a(1), &body),
279            "DeleteNode" => self.delete_node(&ctx, a(0), a(1)),
280            "ListNodes" => self.list_nodes(&ctx, a(0), &q),
281            "CreateProposal" => self.create_proposal(&ctx, a(0), &body),
282            "GetProposal" => self.get_proposal(&ctx, a(0), a(1)),
283            "ListProposals" => self.list_proposals(&ctx, a(0), &q),
284            "ListProposalVotes" => self.list_proposal_votes(&ctx, a(0), a(1), &q),
285            "VoteOnProposal" => self.vote_on_proposal(&ctx, a(0), a(1), &body),
286            "CreateAccessor" => self.create_accessor(&ctx, &body),
287            "GetAccessor" => self.get_accessor(&ctx, a(0)),
288            "DeleteAccessor" => self.delete_accessor(&ctx, a(0)),
289            "ListAccessors" => self.list_accessors(&ctx, &q),
290            "ListInvitations" => self.list_invitations(&ctx, &q),
291            "RejectInvitation" => self.reject_invitation(&ctx, a(0)),
292            "TagResource" => self.tag_resource(&ctx, a(0), &body),
293            "UntagResource" => self.untag_resource(&ctx, a(0), &q),
294            "ListTagsForResource" => self.list_tags(&ctx, a(0)),
295            _ => Err(AwsServiceError::action_not_implemented(
296                "managedblockchain",
297                action,
298            )),
299        };
300        (result, settled)
301    }
302
303    // ------------------------------ Networks ----------------------------
304
305    fn create_network(&self, ctx: &Ctx, body: &Value) -> Result<AwsResponse, AwsServiceError> {
306        let now = shared::iso_now();
307        let network_id = shared::new_network_id();
308        let framework = str_or(body, "Framework", "HYPERLEDGER_FABRIC");
309        let member_id = shared::new_member_id();
310
311        let mut net = Map::new();
312        net.insert("Id".into(), json!(network_id));
313        net.insert("Name".into(), json!(str_or(body, "Name", "")));
314        net.insert("Framework".into(), json!(framework));
315        net.insert(
316            "FrameworkVersion".into(),
317            json!(str_or(body, "FrameworkVersion", "")),
318        );
319        net.insert("Status".into(), json!("AVAILABLE"));
320        net.insert("CreationDate".into(), json!(now));
321        net.insert("Arn".into(), json!(shared::network_arn(&network_id)));
322        net.insert(
323            "VpcEndpointServiceName".into(),
324            json!(format!(
325                "com.amazonaws.{}.managedblockchain.{}",
326                ctx.region,
327                network_id.to_lowercase()
328            )),
329        );
330        echo(&mut net, body, &["Description", "VotingPolicy"]);
331        net.insert(
332            "FrameworkAttributes".into(),
333            network_framework_attributes(&framework, &network_id, body, &ctx.region),
334        );
335
336        let is_fabric = framework == "HYPERLEDGER_FABRIC";
337
338        let mut guard = self.state.write();
339        let data = guard.get_or_create(&ctx.account);
340        store_tags(data, &shared::network_arn(&network_id), body);
341        data.networks.insert(network_id.clone(), Value::Object(net));
342
343        // Hyperledger Fabric networks atomically create the requested first
344        // member; Ethereum networks do not.
345        let mut out = json!({ "NetworkId": network_id });
346        if is_fabric {
347            let member_config = body
348                .get("MemberConfiguration")
349                .cloned()
350                .unwrap_or(json!({}));
351            let member = build_member(ctx, &network_id, &member_id, &member_config, &now, true);
352            let arn = shared::member_arn(&ctx.region, &ctx.account, &member_id);
353            store_tags(data, &arn, &member_config);
354            data.members.insert(member_id.clone(), member);
355            out.as_object_mut()
356                .unwrap()
357                .insert("MemberId".into(), json!(member_id));
358        }
359        ok(out)
360    }
361
362    fn get_network(&self, ctx: &Ctx, network_id: &str) -> Result<AwsResponse, AwsServiceError> {
363        let guard = self.state.read();
364        let data = guard.get(&ctx.account);
365        match data.and_then(|d| d.networks.get(network_id)) {
366            Some(n) => {
367                let arn = shared::network_arn(network_id);
368                ok(json!({ "Network": with_tags(data.unwrap(), &arn, n) }))
369            }
370            None => Err(not_found(&format!("Network {network_id} was not found."))),
371        }
372    }
373
374    fn list_networks(
375        &self,
376        ctx: &Ctx,
377        q: &[(String, String)],
378    ) -> Result<AwsResponse, AwsServiceError> {
379        validate_list_query(
380            q,
381            10,
382            &[
383                ("framework", crate::validate::FRAMEWORK),
384                ("status", NETWORK_STATUS),
385            ],
386        )?;
387        let name = query_one(q, "name");
388        let framework = query_one(q, "framework");
389        let status = query_one(q, "status");
390        let guard = self.state.read();
391        let mut networks: Vec<Value> = guard
392            .get(&ctx.account)
393            .map(|d| {
394                d.networks
395                    .values()
396                    .filter(|n| {
397                        opt_eq(n, "Name", name)
398                            && opt_eq(n, "Framework", framework)
399                            && opt_eq(n, "Status", status)
400                    })
401                    .map(network_summary)
402                    .collect()
403            })
404            .unwrap_or_default();
405        networks.sort_by(|a, b| summary_id(a).cmp(summary_id(b)));
406        let (page, next) = paginate(networks, q)?;
407        let mut out = json!({ "Networks": page });
408        if let Some(n) = next {
409            out["NextToken"] = json!(n);
410        }
411        ok(out)
412    }
413
414    // ------------------------------ Members -----------------------------
415
416    fn create_member(
417        &self,
418        ctx: &Ctx,
419        network_id: &str,
420        body: &Value,
421    ) -> Result<AwsResponse, AwsServiceError> {
422        check_id(network_id, "NetworkId")?;
423        let invitation_id = str_or(body, "InvitationId", "");
424        let now = shared::iso_now();
425        let member_id = shared::new_member_id();
426        let member_config = body
427            .get("MemberConfiguration")
428            .cloned()
429            .unwrap_or(json!({}));
430
431        let mut guard = self.state.write();
432        let data = guard.get_or_create(&ctx.account);
433        // The invitation must exist and be pending; joining consumes it.
434        if let Some(inv) = data.invitations.get_mut(&invitation_id) {
435            if let Some(obj) = inv.as_object_mut() {
436                obj.insert("Status".into(), json!("ACCEPTED"));
437            }
438        } else {
439            return Err(invalid_request(&format!(
440                "Invitation {invitation_id} was not found or is not pending."
441            )));
442        }
443        let member = build_member(ctx, network_id, &member_id, &member_config, &now, true);
444        let arn = shared::member_arn(&ctx.region, &ctx.account, &member_id);
445        store_tags(data, &arn, &member_config);
446        data.members.insert(member_id.clone(), member);
447        ok(json!({ "MemberId": member_id }))
448    }
449
450    fn get_member(
451        &self,
452        ctx: &Ctx,
453        _network_id: &str,
454        member_id: &str,
455    ) -> Result<AwsResponse, AwsServiceError> {
456        let guard = self.state.read();
457        let data = guard.get(&ctx.account);
458        match data.and_then(|d| d.members.get(member_id)) {
459            Some(m) => {
460                let arn = shared::member_arn(&ctx.region, &ctx.account, member_id);
461                ok(json!({ "Member": with_tags(data.unwrap(), &arn, m) }))
462            }
463            None => Err(not_found(&format!("Member {member_id} was not found."))),
464        }
465    }
466
467    fn update_member(
468        &self,
469        ctx: &Ctx,
470        _network_id: &str,
471        member_id: &str,
472        body: &Value,
473    ) -> Result<AwsResponse, AwsServiceError> {
474        let mut guard = self.state.write();
475        let data = guard.get_or_create(&ctx.account);
476        let Some(m) = data
477            .members
478            .get_mut(member_id)
479            .and_then(Value::as_object_mut)
480        else {
481            return Err(not_found(&format!("Member {member_id} was not found.")));
482        };
483        if let Some(lpc) = body.get("LogPublishingConfiguration") {
484            m.insert("LogPublishingConfiguration".into(), lpc.clone());
485        }
486        ok_empty()
487    }
488
489    fn delete_member(
490        &self,
491        ctx: &Ctx,
492        _network_id: &str,
493        member_id: &str,
494    ) -> Result<AwsResponse, AwsServiceError> {
495        let mut guard = self.state.write();
496        let data = guard.get_or_create(&ctx.account);
497        let Some(m) = data
498            .members
499            .get_mut(member_id)
500            .and_then(Value::as_object_mut)
501        else {
502            return Err(not_found(&format!("Member {member_id} was not found.")));
503        };
504        m.insert("Status".into(), json!("DELETED"));
505        ok_empty()
506    }
507
508    fn list_members(
509        &self,
510        ctx: &Ctx,
511        network_id: &str,
512        q: &[(String, String)],
513    ) -> Result<AwsResponse, AwsServiceError> {
514        check_id(network_id, "NetworkId")?;
515        validate_list_query(q, 20, &[("status", MEMBER_STATUS)])?;
516        let name = query_one(q, "name");
517        let status = query_one(q, "status");
518        let guard = self.state.read();
519        let mut members: Vec<Value> = guard
520            .get(&ctx.account)
521            .map(|d| {
522                d.members
523                    .values()
524                    .filter(|m| {
525                        opt_eq(m, "NetworkId", Some(network_id))
526                            && opt_eq(m, "Name", name)
527                            && opt_eq(m, "Status", status)
528                    })
529                    .map(|m| member_summary(m, &ctx.account, &ctx.region))
530                    .collect()
531            })
532            .unwrap_or_default();
533        members.sort_by(|a, b| summary_id(a).cmp(summary_id(b)));
534        let (page, next) = paginate(members, q)?;
535        let mut out = json!({ "Members": page });
536        if let Some(n) = next {
537            out["NextToken"] = json!(n);
538        }
539        ok(out)
540    }
541
542    // ------------------------------- Nodes ------------------------------
543
544    fn create_node(
545        &self,
546        ctx: &Ctx,
547        network_id: &str,
548        body: &Value,
549    ) -> Result<AwsResponse, AwsServiceError> {
550        check_id(network_id, "NetworkId")?;
551        let now = shared::iso_now();
552        let node_id = shared::new_node_id();
553        let member_id = body
554            .get("MemberId")
555            .and_then(Value::as_str)
556            .map(str::to_string);
557        let node_config = body.get("NodeConfiguration").cloned().unwrap_or(json!({}));
558
559        let mut guard = self.state.write();
560        let data = guard.get_or_create(&ctx.account);
561        let framework = data
562            .networks
563            .get(network_id)
564            .and_then(|n| n.get("Framework"))
565            .and_then(Value::as_str)
566            .unwrap_or("HYPERLEDGER_FABRIC")
567            .to_string();
568        let node = build_node(
569            ctx,
570            network_id,
571            member_id.as_deref(),
572            &node_id,
573            &node_config,
574            &framework,
575            &now,
576        );
577        let arn = shared::node_arn(&ctx.region, &ctx.account, &node_id);
578        store_tags(data, &arn, body);
579        data.nodes.insert(node_id.clone(), node);
580        ok(json!({ "NodeId": node_id }))
581    }
582
583    fn get_node(
584        &self,
585        ctx: &Ctx,
586        _network_id: &str,
587        node_id: &str,
588    ) -> Result<AwsResponse, AwsServiceError> {
589        let guard = self.state.read();
590        let data = guard.get(&ctx.account);
591        match data.and_then(|d| d.nodes.get(node_id)) {
592            Some(n) => {
593                let arn = shared::node_arn(&ctx.region, &ctx.account, node_id);
594                ok(json!({ "Node": with_tags(data.unwrap(), &arn, n) }))
595            }
596            None => Err(not_found(&format!("Node {node_id} was not found."))),
597        }
598    }
599
600    fn update_node(
601        &self,
602        ctx: &Ctx,
603        _network_id: &str,
604        node_id: &str,
605        body: &Value,
606    ) -> Result<AwsResponse, AwsServiceError> {
607        let mut guard = self.state.write();
608        let data = guard.get_or_create(&ctx.account);
609        let Some(n) = data.nodes.get_mut(node_id).and_then(Value::as_object_mut) else {
610            return Err(not_found(&format!("Node {node_id} was not found.")));
611        };
612        if let Some(lpc) = body.get("LogPublishingConfiguration") {
613            n.insert("LogPublishingConfiguration".into(), lpc.clone());
614        }
615        ok_empty()
616    }
617
618    fn delete_node(
619        &self,
620        ctx: &Ctx,
621        _network_id: &str,
622        node_id: &str,
623    ) -> Result<AwsResponse, AwsServiceError> {
624        let mut guard = self.state.write();
625        let data = guard.get_or_create(&ctx.account);
626        let Some(n) = data.nodes.get_mut(node_id).and_then(Value::as_object_mut) else {
627            return Err(not_found(&format!("Node {node_id} was not found.")));
628        };
629        n.insert("Status".into(), json!("DELETED"));
630        ok_empty()
631    }
632
633    fn list_nodes(
634        &self,
635        ctx: &Ctx,
636        network_id: &str,
637        q: &[(String, String)],
638    ) -> Result<AwsResponse, AwsServiceError> {
639        check_id(network_id, "NetworkId")?;
640        validate_list_query(q, 20, &[("status", NODE_STATUS)])?;
641        // `memberId` is a `ResourceIdString` (1..=32); validate it even when
642        // present-but-empty (a too-short negative variant), which `query_one`
643        // would otherwise drop.
644        if let Some((_, mid)) = q.iter().find(|(k, _)| k == "memberId") {
645            check_id(mid, "MemberId")?;
646        }
647        let member_id = query_one(q, "memberId");
648        let status = query_one(q, "status");
649        let guard = self.state.read();
650        let mut nodes: Vec<Value> = guard
651            .get(&ctx.account)
652            .map(|d| {
653                d.nodes
654                    .values()
655                    .filter(|n| {
656                        opt_eq(n, "NetworkId", Some(network_id))
657                            && opt_eq(n, "MemberId", member_id)
658                            && opt_eq(n, "Status", status)
659                    })
660                    .map(node_summary)
661                    .collect()
662            })
663            .unwrap_or_default();
664        nodes.sort_by(|a, b| summary_id(a).cmp(summary_id(b)));
665        let (page, next) = paginate(nodes, q)?;
666        let mut out = json!({ "Nodes": page });
667        if let Some(n) = next {
668            out["NextToken"] = json!(n);
669        }
670        ok(out)
671    }
672
673    // ----------------------------- Proposals ----------------------------
674
675    fn create_proposal(
676        &self,
677        ctx: &Ctx,
678        network_id: &str,
679        body: &Value,
680    ) -> Result<AwsResponse, AwsServiceError> {
681        let now = shared::iso_now();
682        let proposal_id = shared::new_proposal_id();
683        let member_id = str_or(body, "MemberId", "");
684        let actions = body.get("Actions").cloned().unwrap_or(json!({}));
685
686        let mut guard = self.state.write();
687        let data = guard.get_or_create(&ctx.account);
688        if !data.networks.contains_key(network_id) {
689            return Err(not_found(&format!("Network {network_id} was not found.")));
690        }
691        let proposed_by_name = data
692            .members
693            .get(&member_id)
694            .and_then(|m| m.get("Name"))
695            .and_then(Value::as_str)
696            .unwrap_or("")
697            .to_string();
698        let duration = proposal_duration_hours(data.networks.get(network_id));
699        let outstanding = eligible_member_count(data, network_id);
700        let expiration = (chrono::Utc::now() + chrono::Duration::hours(duration))
701            .format("%Y-%m-%dT%H:%M:%S%.3fZ")
702            .to_string();
703
704        let mut p = Map::new();
705        p.insert("ProposalId".into(), json!(proposal_id));
706        p.insert("NetworkId".into(), json!(network_id));
707        p.insert("Actions".into(), actions);
708        p.insert("ProposedByMemberId".into(), json!(member_id));
709        p.insert("ProposedByMemberName".into(), json!(proposed_by_name));
710        p.insert("Status".into(), json!("IN_PROGRESS"));
711        p.insert("CreationDate".into(), json!(now));
712        p.insert("ExpirationDate".into(), json!(expiration));
713        p.insert("YesVoteCount".into(), json!(0));
714        p.insert("NoVoteCount".into(), json!(0));
715        p.insert("OutstandingVoteCount".into(), json!(outstanding));
716        p.insert(
717            "Arn".into(),
718            json!(shared::proposal_arn(
719                &ctx.region,
720                &ctx.account,
721                &proposal_id
722            )),
723        );
724        if let Some(d) = body.get("Description") {
725            p.insert("Description".into(), d.clone());
726        }
727        store_tags(
728            data,
729            &shared::proposal_arn(&ctx.region, &ctx.account, &proposal_id),
730            body,
731        );
732        data.proposals.insert(proposal_id.clone(), Value::Object(p));
733        data.votes.insert(proposal_id.clone(), Vec::new());
734        ok(json!({ "ProposalId": proposal_id }))
735    }
736
737    fn get_proposal(
738        &self,
739        ctx: &Ctx,
740        _network_id: &str,
741        proposal_id: &str,
742    ) -> Result<AwsResponse, AwsServiceError> {
743        let guard = self.state.read();
744        let data = guard.get(&ctx.account);
745        match data.and_then(|d| d.proposals.get(proposal_id)) {
746            Some(p) => {
747                let arn = shared::proposal_arn(&ctx.region, &ctx.account, proposal_id);
748                ok(json!({ "Proposal": with_tags(data.unwrap(), &arn, p) }))
749            }
750            None => Err(not_found(&format!("Proposal {proposal_id} was not found."))),
751        }
752    }
753
754    fn list_proposals(
755        &self,
756        ctx: &Ctx,
757        network_id: &str,
758        q: &[(String, String)],
759    ) -> Result<AwsResponse, AwsServiceError> {
760        check_id(network_id, "NetworkId")?;
761        validate_list_query(q, 100, &[])?;
762        let guard = self.state.read();
763        let data = guard.get(&ctx.account);
764        if let Some(d) = data {
765            if !d.networks.contains_key(network_id) {
766                return Err(not_found(&format!("Network {network_id} was not found.")));
767            }
768        }
769        let mut proposals: Vec<Value> = data
770            .map(|d| {
771                d.proposals
772                    .values()
773                    .filter(|p| opt_eq(p, "NetworkId", Some(network_id)))
774                    .map(proposal_summary)
775                    .collect()
776            })
777            .unwrap_or_default();
778        proposals.sort_by(|a, b| {
779            a.get("ProposalId")
780                .and_then(Value::as_str)
781                .unwrap_or("")
782                .cmp(b.get("ProposalId").and_then(Value::as_str).unwrap_or(""))
783        });
784        let (page, next) = paginate(proposals, q)?;
785        let mut out = json!({ "Proposals": page });
786        if let Some(n) = next {
787            out["NextToken"] = json!(n);
788        }
789        ok(out)
790    }
791
792    fn list_proposal_votes(
793        &self,
794        ctx: &Ctx,
795        network_id: &str,
796        proposal_id: &str,
797        q: &[(String, String)],
798    ) -> Result<AwsResponse, AwsServiceError> {
799        check_id(network_id, "NetworkId")?;
800        check_id(proposal_id, "ProposalId")?;
801        validate_list_query(q, 100, &[])?;
802        let guard = self.state.read();
803        let votes: Vec<Value> = guard
804            .get(&ctx.account)
805            .and_then(|d| d.votes.get(proposal_id))
806            .cloned()
807            .unwrap_or_default();
808        let (page, next) = paginate(votes, q)?;
809        let mut out = json!({ "ProposalVotes": page });
810        if let Some(n) = next {
811            out["NextToken"] = json!(n);
812        }
813        ok(out)
814    }
815
816    fn vote_on_proposal(
817        &self,
818        ctx: &Ctx,
819        network_id: &str,
820        proposal_id: &str,
821        body: &Value,
822    ) -> Result<AwsResponse, AwsServiceError> {
823        let voter = str_or(body, "VoterMemberId", "");
824        let vote = str_or(body, "Vote", "");
825
826        let mut guard = self.state.write();
827        // Collect the invitations to materialise across accounts once the
828        // proposal is decided; execute them after the caller's borrow ends.
829        let mut invitations_to_send: Vec<(String, Value)> = Vec::new();
830        {
831            let region = ctx.region.clone();
832            let data = guard.get_or_create(&ctx.account);
833            let network = data.networks.get(network_id).cloned();
834            if network.is_none() {
835                return Err(not_found(&format!("Network {network_id} was not found.")));
836            }
837            if !data.members.contains_key(&voter) {
838                return Err(not_found(&format!("Member {voter} was not found.")));
839            }
840            let member_name = data
841                .members
842                .get(&voter)
843                .and_then(|m| m.get("Name"))
844                .and_then(Value::as_str)
845                .unwrap_or("")
846                .to_string();
847
848            {
849                let proposal = data
850                    .proposals
851                    .get(proposal_id)
852                    .ok_or_else(|| not_found(&format!("Proposal {proposal_id} was not found.")))?;
853                let status = proposal.get("Status").and_then(Value::as_str).unwrap_or("");
854                if status != "IN_PROGRESS" {
855                    return Err(illegal_action(&format!(
856                        "Proposal {proposal_id} is not open for voting (status {status})."
857                    )));
858                }
859            }
860
861            let votes = data.votes.entry(proposal_id.to_string()).or_default();
862            if votes
863                .iter()
864                .any(|v| v.get("MemberId").and_then(Value::as_str) == Some(voter.as_str()))
865            {
866                return Err(illegal_action(&format!(
867                    "Member {voter} has already voted on proposal {proposal_id}."
868                )));
869            }
870            votes.push(json!({
871                "Vote": vote,
872                "MemberId": voter,
873                "MemberName": member_name,
874            }));
875            let yes = votes
876                .iter()
877                .filter(|v| v.get("Vote").and_then(Value::as_str) == Some("YES"))
878                .count() as i64;
879            let no = votes
880                .iter()
881                .filter(|v| v.get("Vote").and_then(Value::as_str) == Some("NO"))
882                .count() as i64;
883
884            let total = eligible_member_count(data, network_id);
885            let (threshold, comparator) = threshold_policy(network.as_ref());
886            let outstanding = (total - yes - no).max(0);
887            let decision = decide_proposal(yes, outstanding, total, threshold, &comparator);
888
889            // Apply the decision to the proposal + gather any materialisations.
890            let actions = data
891                .proposals
892                .get(proposal_id)
893                .and_then(|p| p.get("Actions"))
894                .cloned()
895                .unwrap_or(json!({}));
896            if let Some(p) = data
897                .proposals
898                .get_mut(proposal_id)
899                .and_then(Value::as_object_mut)
900            {
901                p.insert("YesVoteCount".into(), json!(yes));
902                p.insert("NoVoteCount".into(), json!(no));
903                p.insert("OutstandingVoteCount".into(), json!(outstanding));
904                p.insert("Status".into(), json!(decision.clone()));
905            }
906
907            if decision == "APPROVED" {
908                // Removals target members in this account/network.
909                if let Some(removals) = actions.get("Removals").and_then(Value::as_array) {
910                    for r in removals {
911                        if let Some(mid) = r.get("MemberId").and_then(Value::as_str) {
912                            if let Some(m) =
913                                data.members.get_mut(mid).and_then(Value::as_object_mut)
914                            {
915                                m.insert("Status".into(), json!("DELETED"));
916                            }
917                        }
918                    }
919                }
920                // Invitations become new `Invitation` records in the invited
921                // principal's account.
922                if let Some(invites) = actions.get("Invitations").and_then(Value::as_array) {
923                    let net = network.as_ref().unwrap();
924                    for inv in invites {
925                        if let Some(principal) = inv.get("Principal").and_then(Value::as_str) {
926                            let invitation = build_invitation(&region, principal, net);
927                            invitations_to_send.push((principal.to_string(), invitation));
928                        }
929                    }
930                }
931            }
932        }
933
934        for (principal, invitation) in invitations_to_send {
935            let inv_id = invitation
936                .get("InvitationId")
937                .and_then(Value::as_str)
938                .unwrap_or_default()
939                .to_string();
940            let acct = guard.get_or_create(&principal);
941            acct.invitations.insert(inv_id, invitation);
942        }
943
944        ok_empty()
945    }
946
947    // ------------------------------ Accessors ---------------------------
948
949    fn create_accessor(&self, ctx: &Ctx, body: &Value) -> Result<AwsResponse, AwsServiceError> {
950        let now = shared::iso_now();
951        let accessor_id = shared::new_accessor_id();
952        let billing_token = shared::new_billing_token();
953        let network_type = body
954            .get("NetworkType")
955            .and_then(Value::as_str)
956            .map(str::to_string);
957
958        let mut acc = Map::new();
959        acc.insert("Id".into(), json!(accessor_id));
960        acc.insert(
961            "Type".into(),
962            json!(str_or(body, "AccessorType", "BILLING_TOKEN")),
963        );
964        acc.insert("BillingToken".into(), json!(billing_token));
965        acc.insert("Status".into(), json!("AVAILABLE"));
966        acc.insert("CreationDate".into(), json!(now));
967        acc.insert(
968            "Arn".into(),
969            json!(shared::accessor_arn(
970                &ctx.region,
971                &ctx.account,
972                &accessor_id
973            )),
974        );
975        if let Some(nt) = &network_type {
976            acc.insert("NetworkType".into(), json!(nt));
977        }
978
979        let mut guard = self.state.write();
980        let data = guard.get_or_create(&ctx.account);
981        store_tags(
982            data,
983            &shared::accessor_arn(&ctx.region, &ctx.account, &accessor_id),
984            body,
985        );
986        data.accessors
987            .insert(accessor_id.clone(), Value::Object(acc));
988
989        let mut out = json!({ "AccessorId": accessor_id, "BillingToken": billing_token });
990        if let Some(nt) = network_type {
991            out["NetworkType"] = json!(nt);
992        }
993        ok(out)
994    }
995
996    fn get_accessor(&self, ctx: &Ctx, accessor_id: &str) -> Result<AwsResponse, AwsServiceError> {
997        let guard = self.state.read();
998        let data = guard.get(&ctx.account);
999        match data.and_then(|d| d.accessors.get(accessor_id)) {
1000            Some(a) => {
1001                let arn = shared::accessor_arn(&ctx.region, &ctx.account, accessor_id);
1002                ok(json!({ "Accessor": with_tags(data.unwrap(), &arn, a) }))
1003            }
1004            None => Err(not_found(&format!("Accessor {accessor_id} was not found."))),
1005        }
1006    }
1007
1008    fn delete_accessor(
1009        &self,
1010        ctx: &Ctx,
1011        accessor_id: &str,
1012    ) -> Result<AwsResponse, AwsServiceError> {
1013        let mut guard = self.state.write();
1014        let data = guard.get_or_create(&ctx.account);
1015        let Some(a) = data
1016            .accessors
1017            .get_mut(accessor_id)
1018            .and_then(Value::as_object_mut)
1019        else {
1020            return Err(not_found(&format!("Accessor {accessor_id} was not found.")));
1021        };
1022        a.insert("Status".into(), json!("PENDING_DELETION"));
1023        ok_empty()
1024    }
1025
1026    fn list_accessors(
1027        &self,
1028        ctx: &Ctx,
1029        q: &[(String, String)],
1030    ) -> Result<AwsResponse, AwsServiceError> {
1031        validate_list_query(
1032            q,
1033            50,
1034            &[("networkType", crate::validate::ACCESSOR_NETWORK_TYPE)],
1035        )?;
1036        let network_type = query_one(q, "networkType");
1037        let guard = self.state.read();
1038        let mut accessors: Vec<Value> = guard
1039            .get(&ctx.account)
1040            .map(|d| {
1041                d.accessors
1042                    .values()
1043                    .filter(|a| opt_eq(a, "NetworkType", network_type))
1044                    .map(accessor_summary)
1045                    .collect()
1046            })
1047            .unwrap_or_default();
1048        accessors.sort_by(|a, b| summary_id(a).cmp(summary_id(b)));
1049        let (page, next) = paginate(accessors, q)?;
1050        let mut out = json!({ "Accessors": page });
1051        if let Some(n) = next {
1052            out["NextToken"] = json!(n);
1053        }
1054        ok(out)
1055    }
1056
1057    // ----------------------------- Invitations --------------------------
1058
1059    fn list_invitations(
1060        &self,
1061        ctx: &Ctx,
1062        q: &[(String, String)],
1063    ) -> Result<AwsResponse, AwsServiceError> {
1064        validate_list_query(q, 100, &[])?;
1065        let guard = self.state.read();
1066        let mut invitations: Vec<Value> = guard
1067            .get(&ctx.account)
1068            .map(|d| d.invitations.values().cloned().collect())
1069            .unwrap_or_default();
1070        invitations.sort_by(|a, b| {
1071            a.get("InvitationId")
1072                .and_then(Value::as_str)
1073                .unwrap_or("")
1074                .cmp(b.get("InvitationId").and_then(Value::as_str).unwrap_or(""))
1075        });
1076        let (page, next) = paginate(invitations, q)?;
1077        let mut out = json!({ "Invitations": page });
1078        if let Some(n) = next {
1079            out["NextToken"] = json!(n);
1080        }
1081        ok(out)
1082    }
1083
1084    fn reject_invitation(
1085        &self,
1086        ctx: &Ctx,
1087        invitation_id: &str,
1088    ) -> Result<AwsResponse, AwsServiceError> {
1089        let mut guard = self.state.write();
1090        let data = guard.get_or_create(&ctx.account);
1091        let Some(inv) = data
1092            .invitations
1093            .get_mut(invitation_id)
1094            .and_then(Value::as_object_mut)
1095        else {
1096            return Err(not_found(&format!(
1097                "Invitation {invitation_id} was not found."
1098            )));
1099        };
1100        let status = inv.get("Status").and_then(Value::as_str).unwrap_or("");
1101        if status != "PENDING" {
1102            return Err(illegal_action(&format!(
1103                "Invitation {invitation_id} is not pending (status {status})."
1104            )));
1105        }
1106        inv.insert("Status".into(), json!("REJECTED"));
1107        ok_empty()
1108    }
1109
1110    // -------------------------------- Tags ------------------------------
1111
1112    fn tag_resource(
1113        &self,
1114        ctx: &Ctx,
1115        arn: &str,
1116        body: &Value,
1117    ) -> Result<AwsResponse, AwsServiceError> {
1118        check_arn(arn)?;
1119        let mut guard = self.state.write();
1120        let data = guard.get_or_create(&ctx.account);
1121        let entry = data.tags.entry(arn.to_string()).or_default();
1122        if let Some(tags) = body.get("Tags").and_then(Value::as_object) {
1123            for (k, v) in tags {
1124                if let Some(s) = v.as_str() {
1125                    entry.insert(k.clone(), s.to_string());
1126                }
1127            }
1128        }
1129        ok_empty()
1130    }
1131
1132    fn untag_resource(
1133        &self,
1134        ctx: &Ctx,
1135        arn: &str,
1136        q: &[(String, String)],
1137    ) -> Result<AwsResponse, AwsServiceError> {
1138        check_arn(arn)?;
1139        let mut guard = self.state.write();
1140        let data = guard.get_or_create(&ctx.account);
1141        if let Some(entry) = data.tags.get_mut(arn) {
1142            for (k, v) in q {
1143                if k == "tagKeys" {
1144                    entry.remove(v);
1145                }
1146            }
1147            if entry.is_empty() {
1148                data.tags.remove(arn);
1149            }
1150        }
1151        ok_empty()
1152    }
1153
1154    fn list_tags(&self, ctx: &Ctx, arn: &str) -> Result<AwsResponse, AwsServiceError> {
1155        check_arn(arn)?;
1156        let guard = self.state.read();
1157        let tags = guard
1158            .get(&ctx.account)
1159            .and_then(|d| d.tags.get(arn))
1160            .cloned()
1161            .unwrap_or_default();
1162        ok(json!({ "Tags": tags }))
1163    }
1164}
1165
1166// ============================= builders ==============================
1167
1168/// Build a `Member` wire object. Fabric members carry a CA endpoint + admin
1169/// username in their framework attributes.
1170fn build_member(
1171    ctx: &Ctx,
1172    network_id: &str,
1173    member_id: &str,
1174    config: &Value,
1175    now: &str,
1176    fabric: bool,
1177) -> Value {
1178    let admin_username = config
1179        .get("FrameworkConfiguration")
1180        .and_then(|f| f.get("Fabric"))
1181        .and_then(|f| f.get("AdminUsername"))
1182        .and_then(Value::as_str)
1183        .unwrap_or("admin")
1184        .to_string();
1185    let mut m = Map::new();
1186    m.insert("NetworkId".into(), json!(network_id));
1187    m.insert("Id".into(), json!(member_id));
1188    m.insert(
1189        "Name".into(),
1190        json!(config.get("Name").and_then(Value::as_str).unwrap_or("")),
1191    );
1192    m.insert("Status".into(), json!("CREATING"));
1193    m.insert("CreationDate".into(), json!(now));
1194    m.insert(
1195        "Arn".into(),
1196        json!(shared::member_arn(&ctx.region, &ctx.account, member_id)),
1197    );
1198    if let Some(d) = config.get("Description") {
1199        m.insert("Description".into(), d.clone());
1200    }
1201    if let Some(lpc) = config.get("LogPublishingConfiguration") {
1202        m.insert("LogPublishingConfiguration".into(), lpc.clone());
1203    }
1204    if let Some(kms) = config.get("KmsKeyArn") {
1205        m.insert("KmsKeyArn".into(), kms.clone());
1206    }
1207    if fabric {
1208        m.insert(
1209            "FrameworkAttributes".into(),
1210            json!({
1211                "Fabric": {
1212                    "AdminUsername": admin_username,
1213                    "CaEndpoint": shared::fabric_ca_endpoint(member_id, &ctx.region),
1214                }
1215            }),
1216        );
1217    }
1218    Value::Object(m)
1219}
1220
1221/// Build a `Node` wire object, deriving the framework-specific endpoints.
1222fn build_node(
1223    ctx: &Ctx,
1224    network_id: &str,
1225    member_id: Option<&str>,
1226    node_id: &str,
1227    config: &Value,
1228    framework: &str,
1229    now: &str,
1230) -> Value {
1231    let mut n = Map::new();
1232    n.insert("NetworkId".into(), json!(network_id));
1233    if let Some(mid) = member_id {
1234        n.insert("MemberId".into(), json!(mid));
1235    }
1236    n.insert("Id".into(), json!(node_id));
1237    n.insert(
1238        "InstanceType".into(),
1239        json!(config
1240            .get("InstanceType")
1241            .and_then(Value::as_str)
1242            .unwrap_or("")),
1243    );
1244    if let Some(az) = config.get("AvailabilityZone") {
1245        n.insert("AvailabilityZone".into(), az.clone());
1246    } else {
1247        n.insert("AvailabilityZone".into(), json!(format!("{}a", ctx.region)));
1248    }
1249    if let Some(sdb) = config.get("StateDB") {
1250        n.insert("StateDB".into(), sdb.clone());
1251    }
1252    n.insert("Status".into(), json!("CREATING"));
1253    n.insert("CreationDate".into(), json!(now));
1254    n.insert(
1255        "Arn".into(),
1256        json!(shared::node_arn(&ctx.region, &ctx.account, node_id)),
1257    );
1258    if let Some(lpc) = config.get("LogPublishingConfiguration") {
1259        n.insert("LogPublishingConfiguration".into(), lpc.clone());
1260    }
1261    let attrs = if framework == "ETHEREUM" {
1262        json!({
1263            "Ethereum": {
1264                "HttpEndpoint": shared::ethereum_http_endpoint(node_id, &ctx.region),
1265                "WebSocketEndpoint": shared::ethereum_ws_endpoint(node_id, &ctx.region),
1266            }
1267        })
1268    } else {
1269        json!({
1270            "Fabric": {
1271                "PeerEndpoint": shared::fabric_peer_endpoint(node_id, &ctx.region),
1272                "PeerEventEndpoint": shared::fabric_peer_event_endpoint(node_id, &ctx.region),
1273            }
1274        })
1275    };
1276    n.insert("FrameworkAttributes".into(), attrs);
1277    Value::Object(n)
1278}
1279
1280/// Build an `Invitation` wire object for a materialised proposal invitation.
1281fn build_invitation(region: &str, principal: &str, network: &Value) -> Value {
1282    let id = shared::new_invitation_id();
1283    let now = shared::iso_now();
1284    let expiration = (chrono::Utc::now() + chrono::Duration::days(7))
1285        .format("%Y-%m-%dT%H:%M:%S%.3fZ")
1286        .to_string();
1287    json!({
1288        "InvitationId": id,
1289        "CreationDate": now,
1290        "ExpirationDate": expiration,
1291        "Status": "PENDING",
1292        "NetworkSummary": network_summary(network),
1293        "Arn": shared::invitation_arn(region, principal, &id),
1294    })
1295}
1296
1297/// The network-level `FrameworkAttributes` (Fabric ordering endpoint + edition,
1298/// or Ethereum chain id).
1299fn network_framework_attributes(
1300    framework: &str,
1301    network_id: &str,
1302    body: &Value,
1303    region: &str,
1304) -> Value {
1305    if framework == "ETHEREUM" {
1306        // Ethereum network chain id is derived from the framework version.
1307        let chain_id = match body.get("FrameworkVersion").and_then(Value::as_str) {
1308            Some("1.0") | Some("mainnet") => "1",
1309            _ => "1",
1310        };
1311        json!({ "Ethereum": { "ChainId": chain_id } })
1312    } else {
1313        let edition = body
1314            .get("FrameworkConfiguration")
1315            .and_then(|f| f.get("Fabric"))
1316            .and_then(|f| f.get("Edition"))
1317            .and_then(Value::as_str)
1318            .unwrap_or("STARTER");
1319        json!({
1320            "Fabric": {
1321                "OrderingServiceEndpoint": shared::fabric_ordering_endpoint(network_id, region),
1322                "Edition": edition,
1323            }
1324        })
1325    }
1326}
1327
1328// ============================= summaries =============================
1329
1330fn network_summary(n: &Value) -> Value {
1331    let mut s = Map::new();
1332    for k in [
1333        "Id",
1334        "Name",
1335        "Description",
1336        "Framework",
1337        "FrameworkVersion",
1338        "Status",
1339        "CreationDate",
1340        "Arn",
1341    ] {
1342        if let Some(v) = n.get(k) {
1343            s.insert(k.to_string(), v.clone());
1344        }
1345    }
1346    Value::Object(s)
1347}
1348
1349fn member_summary(m: &Value, account: &str, region: &str) -> Value {
1350    let id = m.get("Id").and_then(Value::as_str).unwrap_or("");
1351    let mut s = Map::new();
1352    for k in ["Id", "Name", "Description", "Status", "CreationDate"] {
1353        if let Some(v) = m.get(k) {
1354            s.insert(k.to_string(), v.clone());
1355        }
1356    }
1357    s.insert("IsOwned".into(), json!(true));
1358    s.insert("Arn".into(), json!(shared::member_arn(region, account, id)));
1359    Value::Object(s)
1360}
1361
1362fn node_summary(n: &Value) -> Value {
1363    let mut s = Map::new();
1364    for k in [
1365        "Id",
1366        "Status",
1367        "CreationDate",
1368        "AvailabilityZone",
1369        "InstanceType",
1370        "Arn",
1371    ] {
1372        if let Some(v) = n.get(k) {
1373            s.insert(k.to_string(), v.clone());
1374        }
1375    }
1376    Value::Object(s)
1377}
1378
1379fn proposal_summary(p: &Value) -> Value {
1380    let mut s = Map::new();
1381    for k in [
1382        "ProposalId",
1383        "Description",
1384        "ProposedByMemberId",
1385        "ProposedByMemberName",
1386        "Status",
1387        "CreationDate",
1388        "ExpirationDate",
1389        "Arn",
1390    ] {
1391        if let Some(v) = p.get(k) {
1392            s.insert(k.to_string(), v.clone());
1393        }
1394    }
1395    Value::Object(s)
1396}
1397
1398fn accessor_summary(a: &Value) -> Value {
1399    let mut s = Map::new();
1400    for k in ["Id", "Type", "Status", "CreationDate", "Arn", "NetworkType"] {
1401        if let Some(v) = a.get(k) {
1402            s.insert(k.to_string(), v.clone());
1403        }
1404    }
1405    Value::Object(s)
1406}
1407
1408// ============================= voting logic ==========================
1409
1410/// Extract `(ThresholdPercentage, ThresholdComparator)` from a network's voting
1411/// policy, defaulting to a simple majority (50 %, GREATER_THAN).
1412fn threshold_policy(network: Option<&Value>) -> (i64, String) {
1413    let policy = network
1414        .and_then(|n| n.get("VotingPolicy"))
1415        .and_then(|v| v.get("ApprovalThresholdPolicy"));
1416    let threshold = policy
1417        .and_then(|p| p.get("ThresholdPercentage"))
1418        .and_then(Value::as_i64)
1419        .unwrap_or(50);
1420    let comparator = policy
1421        .and_then(|p| p.get("ThresholdComparator"))
1422        .and_then(Value::as_str)
1423        .unwrap_or("GREATER_THAN")
1424        .to_string();
1425    (threshold, comparator)
1426}
1427
1428/// The `ProposalDurationInHours` from a network's voting policy (default 24).
1429fn proposal_duration_hours(network: Option<&Value>) -> i64 {
1430    network
1431        .and_then(|n| n.get("VotingPolicy"))
1432        .and_then(|v| v.get("ApprovalThresholdPolicy"))
1433        .and_then(|p| p.get("ProposalDurationInHours"))
1434        .and_then(Value::as_i64)
1435        .unwrap_or(24)
1436}
1437
1438/// Number of members eligible to vote in a network (not deleted).
1439fn eligible_member_count(data: &ManagedBlockchainData, network_id: &str) -> i64 {
1440    data.members
1441        .values()
1442        .filter(|m| {
1443            m.get("NetworkId").and_then(Value::as_str) == Some(network_id)
1444                && m.get("Status").and_then(Value::as_str) != Some("DELETED")
1445        })
1446        .count() as i64
1447}
1448
1449/// Decide a proposal's status from the current tally. Approves as soon as the
1450/// yes votes meet the threshold of the total electorate; rejects once approval
1451/// is arithmetically impossible even if every outstanding vote were `YES`.
1452fn decide_proposal(
1453    yes: i64,
1454    outstanding: i64,
1455    total: i64,
1456    threshold: i64,
1457    comparator: &str,
1458) -> String {
1459    if total <= 0 {
1460        return "IN_PROGRESS".to_string();
1461    }
1462    let meets = |y: i64| {
1463        let pct = (y as f64) * 100.0 / (total as f64);
1464        match comparator {
1465            "GREATER_THAN_OR_EQUAL_TO" => pct >= threshold as f64,
1466            _ => pct > threshold as f64,
1467        }
1468    };
1469    if meets(yes) {
1470        "APPROVED".to_string()
1471    } else if !meets(yes + outstanding) {
1472        "REJECTED".to_string()
1473    } else {
1474        "IN_PROGRESS".to_string()
1475    }
1476}
1477
1478/// Promote a `CREATING` member/node wire object to `AVAILABLE`. Returns `true`
1479/// on change.
1480fn promote_creating(v: &mut Value) -> bool {
1481    if let Some(obj) = v.as_object_mut() {
1482        if obj.get("Status").and_then(Value::as_str) == Some("CREATING") {
1483            obj.insert("Status".into(), json!("AVAILABLE"));
1484            return true;
1485        }
1486    }
1487    false
1488}
1489
1490// ============================= helpers ==============================
1491
1492fn ok(v: Value) -> Result<AwsResponse, AwsServiceError> {
1493    Ok(AwsResponse::json_value(StatusCode::OK, v))
1494}
1495
1496fn ok_empty() -> Result<AwsResponse, AwsServiceError> {
1497    Ok(AwsResponse::json(StatusCode::OK, "{}"))
1498}
1499
1500fn parse_body(req: &AwsRequest) -> Result<Value, AwsServiceError> {
1501    if req.body.is_empty() {
1502        return Ok(json!({}));
1503    }
1504    serde_json::from_slice(&req.body)
1505        .map_err(|e| invalid_request(&format!("The request body is malformed: {e}")))
1506}
1507
1508fn invalid_request(msg: &str) -> AwsServiceError {
1509    AwsServiceError::aws_error(StatusCode::BAD_REQUEST, "InvalidRequestException", msg)
1510}
1511
1512fn not_found(msg: &str) -> AwsServiceError {
1513    AwsServiceError::aws_error(StatusCode::NOT_FOUND, "ResourceNotFoundException", msg)
1514}
1515
1516fn illegal_action(msg: &str) -> AwsServiceError {
1517    AwsServiceError::aws_error(StatusCode::CONFLICT, "IllegalActionException", msg)
1518}
1519
1520/// Reject a path-label ARN that is absent or not an ARN with
1521/// `InvalidRequestException`.
1522fn check_arn(arn: &str) -> Result<(), AwsServiceError> {
1523    if arn.is_empty() || !arn.starts_with("arn:") {
1524        return Err(invalid_request(
1525            "The request failed because it is missing a valid resource ARN.",
1526        ));
1527    }
1528    Ok(())
1529}
1530
1531fn str_or(body: &Value, key: &str, default: &str) -> String {
1532    body.get(key)
1533        .and_then(Value::as_str)
1534        .unwrap_or(default)
1535        .to_string()
1536}
1537
1538/// Copy each named optional member from `body` into `out` verbatim when present
1539/// and non-null.
1540fn echo(out: &mut Map<String, Value>, body: &Value, keys: &[&str]) {
1541    for key in keys {
1542        if let Some(v) = body.get(*key) {
1543            if !v.is_null() {
1544                out.insert((*key).to_string(), v.clone());
1545            }
1546        }
1547    }
1548}
1549
1550/// Whether an object's string member equals a filter (or the filter is `None`).
1551fn opt_eq(v: &Value, key: &str, filter: Option<&str>) -> bool {
1552    match filter {
1553        None => true,
1554        Some(f) => v.get(key).and_then(Value::as_str) == Some(f),
1555    }
1556}
1557
1558fn summary_id(v: &Value) -> &str {
1559    v.get("Id").and_then(Value::as_str).unwrap_or("")
1560}
1561
1562/// Store the create-time `Tags` map (if any) under a resource ARN.
1563fn store_tags(data: &mut ManagedBlockchainData, arn: &str, body: &Value) {
1564    if let Some(tags) = body.get("Tags").and_then(Value::as_object) {
1565        if tags.is_empty() {
1566            return;
1567        }
1568        let entry = data.tags.entry(arn.to_string()).or_default();
1569        for (k, v) in tags {
1570            if let Some(s) = v.as_str() {
1571                entry.insert(k.clone(), s.to_string());
1572            }
1573        }
1574    }
1575}
1576
1577/// Attach the resource's `Tags` (from the ARN-keyed tag store) to a clone of its
1578/// wire object for a read response.
1579fn with_tags(data: &ManagedBlockchainData, arn: &str, obj: &Value) -> Value {
1580    let mut out = obj.clone();
1581    let tags = data.tags.get(arn).cloned().unwrap_or_default();
1582    if let Some(o) = out.as_object_mut() {
1583        o.insert("Tags".into(), json!(tags));
1584    }
1585    out
1586}
1587
1588fn parse_query(raw: &str) -> Vec<(String, String)> {
1589    raw.split('&')
1590        .filter(|p| !p.is_empty())
1591        .map(|pair| {
1592            let (k, v) = pair.split_once('=').unwrap_or((pair, ""));
1593            (
1594                percent_decode_str(k).decode_utf8_lossy().into_owned(),
1595                percent_decode_str(v).decode_utf8_lossy().into_owned(),
1596            )
1597        })
1598        .collect()
1599}
1600
1601fn query_one<'a>(q: &'a [(String, String)], key: &str) -> Option<&'a str> {
1602    q.iter()
1603        .find(|(k, _)| k == key)
1604        .map(|(_, v)| v.as_str())
1605        .filter(|v| !v.is_empty())
1606}
1607
1608/// Managed Blockchain resource ids (`ResourceIdString`) are 1..=32 chars. The
1609/// pagination token (`PaginationToken`) is at most 128 chars.
1610const ID_MAX_LEN: usize = 32;
1611const TOKEN_MAX_LEN: usize = 128;
1612
1613pub const NETWORK_STATUS: &[&str] = &[
1614    "CREATING",
1615    "AVAILABLE",
1616    "CREATE_FAILED",
1617    "DELETING",
1618    "DELETED",
1619];
1620pub const MEMBER_STATUS: &[&str] = &[
1621    "CREATING",
1622    "AVAILABLE",
1623    "CREATE_FAILED",
1624    "UPDATING",
1625    "DELETING",
1626    "DELETED",
1627    "INACCESSIBLE_ENCRYPTION_KEY",
1628];
1629pub const NODE_STATUS: &[&str] = &[
1630    "CREATING",
1631    "AVAILABLE",
1632    "UNHEALTHY",
1633    "CREATE_FAILED",
1634    "UPDATING",
1635    "DELETING",
1636    "DELETED",
1637    "FAILED",
1638    "INACCESSIBLE_ENCRYPTION_KEY",
1639];
1640
1641/// Reject a path-label resource id whose length is outside `1..=32`, or which
1642/// still carries an unfilled `{Placeholder}` (an omitted required label), with
1643/// `InvalidRequestException`.
1644fn check_id(id: &str, field: &str) -> Result<(), AwsServiceError> {
1645    let n = id.chars().count();
1646    if !(1..=ID_MAX_LEN).contains(&n) {
1647        return Err(invalid_request(&format!(
1648            "'{field}' length must be between 1 and {ID_MAX_LEN}, got {n}."
1649        )));
1650    }
1651    if id.contains('{') || id.contains('}') {
1652        return Err(invalid_request(&format!(
1653            "'{field}' is not a valid resource id: {id}"
1654        )));
1655    }
1656    Ok(())
1657}
1658
1659/// Validate the common list query params: `maxResults` (within `1..=cap`),
1660/// `nextToken` (<= 128 chars), plus any `(param, allowed-enum)` constraints.
1661fn validate_list_query(
1662    q: &[(String, String)],
1663    cap: i64,
1664    enums: &[(&str, &[&str])],
1665) -> Result<(), AwsServiceError> {
1666    if let Some(v) = query_one(q, "maxResults") {
1667        let n: i64 = v
1668            .parse()
1669            .map_err(|_| invalid_request("maxResults must be an integer."))?;
1670        if n < 1 || n > cap {
1671            return Err(invalid_request(&format!(
1672                "maxResults must be between 1 and {cap}, got {n}."
1673            )));
1674        }
1675    }
1676    if let Some(t) = query_one(q, "nextToken") {
1677        if t.chars().count() > TOKEN_MAX_LEN {
1678            return Err(invalid_request(&format!(
1679                "nextToken must be at most {TOKEN_MAX_LEN} characters."
1680            )));
1681        }
1682    }
1683    for (key, allowed) in enums {
1684        if let Some(v) = query_one(q, key) {
1685            if !allowed.contains(&v) {
1686                return Err(invalid_request(&format!(
1687                    "'{key}' must be one of [{}], got '{v}'.",
1688                    allowed.join(", ")
1689                )));
1690            }
1691        }
1692    }
1693    Ok(())
1694}
1695
1696/// Paginate wire objects using the request's `maxResults` / `nextToken` query
1697/// params (`nextToken` is a plain decimal offset). Range/length validation is
1698/// performed up front by `validate_list_query`.
1699fn paginate(
1700    items: Vec<Value>,
1701    q: &[(String, String)],
1702) -> Result<(Vec<Value>, Option<String>), AwsServiceError> {
1703    let max = query_one(q, "maxResults")
1704        .and_then(|v| v.parse::<usize>().ok())
1705        .unwrap_or(usize::MAX);
1706    let start = query_one(q, "nextToken")
1707        .and_then(|v| v.parse::<usize>().ok())
1708        .unwrap_or(0);
1709    let start = start.min(items.len());
1710    let end = start.saturating_add(max).min(items.len());
1711    let page: Vec<Value> = items.get(start..end).unwrap_or(&[]).to_vec();
1712    let next = if end < items.len() {
1713        Some(end.to_string())
1714    } else {
1715        None
1716    };
1717    Ok((page, next))
1718}
1719
1720#[cfg(test)]
1721mod tests {
1722    use super::*;
1723    use fakecloud_core::multi_account::MultiAccountState;
1724    use parking_lot::RwLock;
1725
1726    fn svc() -> ManagedBlockchainService {
1727        ManagedBlockchainService::new(Arc::new(RwLock::new(MultiAccountState::new(
1728            "000000000000",
1729            "us-east-1",
1730            "",
1731        ))))
1732    }
1733
1734    fn ctx() -> Ctx {
1735        Ctx {
1736            account: "000000000000".to_string(),
1737            region: "us-east-1".to_string(),
1738        }
1739    }
1740
1741    fn body_of(resp: &AwsResponse) -> Value {
1742        serde_json::from_slice(resp.body.expect_bytes()).unwrap()
1743    }
1744
1745    fn expect_err(r: Result<AwsResponse, AwsServiceError>) -> AwsServiceError {
1746        match r {
1747            Ok(_) => panic!("expected an error response"),
1748            Err(e) => e,
1749        }
1750    }
1751
1752    fn fabric_network(s: &ManagedBlockchainService) -> (String, String) {
1753        let resp = s
1754            .create_network(
1755                &ctx(),
1756                &json!({
1757                    "ClientRequestToken": "t",
1758                    "Name": "net",
1759                    "Framework": "HYPERLEDGER_FABRIC",
1760                    "FrameworkVersion": "2.2",
1761                    "FrameworkConfiguration": { "Fabric": { "Edition": "STARTER" } },
1762                    "VotingPolicy": { "ApprovalThresholdPolicy": {
1763                        "ThresholdPercentage": 50,
1764                        "ProposalDurationInHours": 24,
1765                        "ThresholdComparator": "GREATER_THAN"
1766                    } },
1767                    "MemberConfiguration": {
1768                        "Name": "member1",
1769                        "FrameworkConfiguration": { "Fabric": {
1770                            "AdminUsername": "admin", "AdminPassword": "Password123!"
1771                        } }
1772                    },
1773                    "Description": "desc"
1774                }),
1775            )
1776            .unwrap();
1777        let out = body_of(&resp);
1778        (
1779            out["NetworkId"].as_str().unwrap().to_string(),
1780            out["MemberId"].as_str().unwrap().to_string(),
1781        )
1782    }
1783
1784    #[test]
1785    fn create_network_atomically_creates_first_member() {
1786        let s = svc();
1787        let (net_id, member_id) = fabric_network(&s);
1788        assert!(net_id.starts_with("n-"));
1789        assert!(member_id.starts_with("m-"));
1790
1791        // GetNetwork echoes name + description + status AVAILABLE.
1792        let g = body_of(&s.get_network(&ctx(), &net_id).unwrap());
1793        assert_eq!(g["Network"]["Name"], "net");
1794        assert_eq!(g["Network"]["Description"], "desc");
1795        assert_eq!(g["Network"]["Status"], "AVAILABLE");
1796        assert_eq!(
1797            g["Network"]["FrameworkAttributes"]["Fabric"]["Edition"],
1798            "STARTER"
1799        );
1800
1801        // GetMember settles to AVAILABLE on read and carries a CA endpoint.
1802        s.reconcile(&ctx().account);
1803        let m = body_of(&s.get_member(&ctx(), &net_id, &member_id).unwrap());
1804        assert_eq!(m["Member"]["Status"], "AVAILABLE");
1805        assert!(m["Member"]["FrameworkAttributes"]["Fabric"]["CaEndpoint"].is_string());
1806    }
1807
1808    #[test]
1809    fn node_lifecycle() {
1810        let s = svc();
1811        let (net_id, _member) = fabric_network(&s);
1812        let created = body_of(
1813            &s.create_node(
1814                &ctx(),
1815                &net_id,
1816                &json!({
1817                    "ClientRequestToken": "t2",
1818                    "NodeConfiguration": { "InstanceType": "bc.t3.small", "StateDB": "CouchDB" }
1819                }),
1820            )
1821            .unwrap(),
1822        );
1823        let node_id = created["NodeId"].as_str().unwrap().to_string();
1824        assert!(node_id.starts_with("nd-"));
1825        s.reconcile(&ctx().account);
1826        let g = body_of(&s.get_node(&ctx(), &net_id, &node_id).unwrap());
1827        assert_eq!(g["Node"]["Status"], "AVAILABLE");
1828        assert!(g["Node"]["FrameworkAttributes"]["Fabric"]["PeerEndpoint"].is_string());
1829        // Delete flips to DELETED.
1830        s.delete_node(&ctx(), &net_id, &node_id).unwrap();
1831        let g2 = body_of(&s.get_node(&ctx(), &net_id, &node_id).unwrap());
1832        assert_eq!(g2["Node"]["Status"], "DELETED");
1833    }
1834
1835    #[test]
1836    fn proposal_vote_approves_and_materialises_invitation() {
1837        let s = svc();
1838        let (net_id, member_id) = fabric_network(&s);
1839        // Propose inviting our own account so ListInvitations can see it.
1840        let prop = body_of(
1841            &s.create_proposal(
1842                &ctx(),
1843                &net_id,
1844                &json!({
1845                    "ClientRequestToken": "t3",
1846                    "MemberId": member_id,
1847                    "Actions": { "Invitations": [ { "Principal": "000000000000" } ] }
1848                }),
1849            )
1850            .unwrap(),
1851        );
1852        let proposal_id = prop["ProposalId"].as_str().unwrap().to_string();
1853        // One member, threshold 50% GREATER_THAN -> a single YES = 100% approves.
1854        s.vote_on_proposal(
1855            &ctx(),
1856            &net_id,
1857            &proposal_id,
1858            &json!({ "VoterMemberId": member_id, "Vote": "YES" }),
1859        )
1860        .unwrap();
1861        let g = body_of(&s.get_proposal(&ctx(), &net_id, &proposal_id).unwrap());
1862        assert_eq!(g["Proposal"]["Status"], "APPROVED");
1863        assert_eq!(g["Proposal"]["YesVoteCount"], 1);
1864        // Votes are recorded.
1865        let votes = body_of(
1866            &s.list_proposal_votes(&ctx(), &net_id, &proposal_id, &[])
1867                .unwrap(),
1868        );
1869        assert_eq!(votes["ProposalVotes"].as_array().unwrap().len(), 1);
1870        // The invitation was materialised.
1871        let invs = body_of(&s.list_invitations(&ctx(), &[]).unwrap());
1872        assert_eq!(invs["Invitations"].as_array().unwrap().len(), 1);
1873        assert_eq!(invs["Invitations"][0]["Status"], "PENDING");
1874    }
1875
1876    #[test]
1877    fn duplicate_vote_is_illegal_action() {
1878        let s = svc();
1879        let (net_id, member_id) = fabric_network(&s);
1880        let prop = body_of(
1881            &s.create_proposal(
1882                &ctx(),
1883                &net_id,
1884                &json!({
1885                    "ClientRequestToken": "t",
1886                    "MemberId": member_id,
1887                    "Actions": { "Removals": [ { "MemberId": "m-OTHER" } ] }
1888                }),
1889            )
1890            .unwrap(),
1891        );
1892        let pid = prop["ProposalId"].as_str().unwrap().to_string();
1893        s.vote_on_proposal(
1894            &ctx(),
1895            &net_id,
1896            &pid,
1897            &json!({ "VoterMemberId": member_id, "Vote": "YES" }),
1898        )
1899        .unwrap();
1900        // Proposal already APPROVED -> further votes are illegal.
1901        let err = expect_err(s.vote_on_proposal(
1902            &ctx(),
1903            &net_id,
1904            &pid,
1905            &json!({ "VoterMemberId": member_id, "Vote": "NO" }),
1906        ));
1907        assert!(format!("{err:?}").contains("IllegalActionException"));
1908    }
1909
1910    #[test]
1911    fn accessor_lifecycle() {
1912        let s = svc();
1913        let created = body_of(
1914            &s.create_accessor(
1915                &ctx(),
1916                &json!({ "ClientRequestToken": "t", "AccessorType": "BILLING_TOKEN", "NetworkType": "ETHEREUM_MAINNET" }),
1917            )
1918            .unwrap(),
1919        );
1920        let id = created["AccessorId"].as_str().unwrap().to_string();
1921        assert!(created["BillingToken"].is_string());
1922        let g = body_of(&s.get_accessor(&ctx(), &id).unwrap());
1923        assert_eq!(g["Accessor"]["Status"], "AVAILABLE");
1924        assert_eq!(g["Accessor"]["NetworkType"], "ETHEREUM_MAINNET");
1925        s.delete_accessor(&ctx(), &id).unwrap();
1926        let g2 = body_of(&s.get_accessor(&ctx(), &id).unwrap());
1927        assert_eq!(g2["Accessor"]["Status"], "PENDING_DELETION");
1928    }
1929
1930    #[test]
1931    fn list_networks_filters_and_paginates() {
1932        let s = svc();
1933        fabric_network(&s);
1934        fabric_network(&s);
1935        let all = body_of(&s.list_networks(&ctx(), &[]).unwrap());
1936        assert_eq!(all["Networks"].as_array().unwrap().len(), 2);
1937        // Paginate.
1938        let page1 = body_of(
1939            &s.list_networks(&ctx(), &[("maxResults".into(), "1".into())])
1940                .unwrap(),
1941        );
1942        assert_eq!(page1["Networks"].as_array().unwrap().len(), 1);
1943        assert!(page1["NextToken"].is_string());
1944        // Filter by framework.
1945        let eth = body_of(
1946            &s.list_networks(&ctx(), &[("framework".into(), "ETHEREUM".into())])
1947                .unwrap(),
1948        );
1949        assert_eq!(eth["Networks"].as_array().unwrap().len(), 0);
1950    }
1951
1952    #[test]
1953    fn unknown_network_is_resource_not_found() {
1954        let s = svc();
1955        let err = expect_err(s.get_network(&ctx(), "n-NOPE"));
1956        assert!(format!("{err:?}").contains("ResourceNotFoundException"));
1957    }
1958
1959    #[test]
1960    fn tag_untag_round_trips() {
1961        let s = svc();
1962        let (net_id, _m) = fabric_network(&s);
1963        let arn = shared::network_arn(&net_id);
1964        s.tag_resource(&ctx(), &arn, &json!({ "Tags": { "k": "v" } }))
1965            .unwrap();
1966        let listed = body_of(&s.list_tags(&ctx(), &arn).unwrap());
1967        assert_eq!(listed["Tags"]["k"], "v");
1968        s.untag_resource(&ctx(), &arn, &[("tagKeys".into(), "k".into())])
1969            .unwrap();
1970        let after = body_of(&s.list_tags(&ctx(), &arn).unwrap());
1971        assert!(after["Tags"].as_object().unwrap().is_empty());
1972    }
1973}