ave_core/node/
register.rs

1use async_trait::async_trait;
2use ave_actors::{
3    Actor, ActorContext, ActorError, ActorPath, Event, Handler, Message,
4    NotPersistentActor, Response,
5};
6use ave_common::SchemaType;
7use serde::{Deserialize, Serialize};
8use tracing::{Span, debug, error, info_span};
9
10use crate::model::common::emit_fail;
11
12#[derive(Clone, Debug, Serialize, Deserialize)]
13pub struct RegisterDataSubj {
14    pub schema_id: SchemaType,
15    pub active: bool,
16    pub namespace: String,
17    pub name: Option<String>,
18    pub description: Option<String>,
19}
20
21#[derive(Clone, Debug, Serialize, Deserialize)]
22pub struct RegisterDataGov {
23    pub active: bool,
24    pub name: Option<String>,
25    pub description: Option<String>,
26}
27
28#[derive(Clone, Debug, Default)]
29pub struct Register;
30
31#[derive(Debug, Clone)]
32pub enum RegisterMessage {
33    RegisterGov {
34        gov_id: String,
35        name: Option<String>,
36        description: Option<String>,
37    },
38    EOLGov {
39        gov_id: String,
40    },
41    RegisterSubj {
42        gov_id: String,
43        subject_id: String,
44        schema_id: SchemaType,
45        namespace: String,
46        name: Option<String>,
47        description: Option<String>,
48    },
49    EOLSubj {
50        gov_id: String,
51        subj_id: String,
52    },
53}
54
55impl Message for RegisterMessage {
56    fn is_critical(&self) -> bool {
57        true
58    }
59}
60
61#[derive(Debug, Clone)]
62pub enum RegisterResponse {
63    None,
64}
65
66impl Response for RegisterResponse {}
67
68#[derive(Debug, Clone, Serialize, Deserialize)]
69pub enum RegisterEvent {
70    RegisterGov {
71        gov_id: String,
72        data: RegisterDataGov,
73    },
74    EOLGov {
75        gov_id: String,
76    },
77    RegisterSubj {
78        gov_id: String,
79        subject_id: String,
80        data: RegisterDataSubj,
81    },
82    EOLSubj {
83        gov_id: String,
84        subj_id: String,
85    },
86}
87
88impl Event for RegisterEvent {}
89
90impl NotPersistentActor for Register {}
91
92#[async_trait]
93impl Actor for Register {
94    type Event = RegisterEvent;
95    type Message = RegisterMessage;
96    type Response = RegisterResponse;
97
98    fn get_span(_id: &str, parent_span: Option<Span>) -> tracing::Span {
99        parent_span.map_or_else(
100            || info_span!("Register"),
101            |parent_span| info_span!(parent: parent_span, "Register"),
102        )
103    }
104}
105
106#[async_trait]
107impl Handler<Self> for Register {
108    async fn handle_message(
109        &mut self,
110        _sender: ActorPath,
111        msg: RegisterMessage,
112        ctx: &mut ave_actors::ActorContext<Self>,
113    ) -> Result<RegisterResponse, ActorError> {
114        let event = match msg {
115            RegisterMessage::RegisterGov {
116                gov_id,
117                description,
118                name,
119            } => {
120                debug!(
121                    msg_type = "RegisterGov",
122                    gov_id = %gov_id,
123                    "Governance registration enqueued"
124                );
125                RegisterEvent::RegisterGov {
126                    gov_id,
127                    data: RegisterDataGov {
128                        active: true,
129                        name,
130                        description,
131                    },
132                }
133            }
134            RegisterMessage::EOLGov { gov_id } => {
135                debug!(
136                    msg_type = "EOLGov",
137                    gov_id = %gov_id,
138                    "Governance EOL enqueued"
139                );
140                RegisterEvent::EOLGov { gov_id }
141            }
142            RegisterMessage::RegisterSubj {
143                gov_id,
144                subject_id,
145                schema_id,
146                namespace,
147                name,
148                description,
149            } => {
150                debug!(
151                    msg_type = "RegisterSubj",
152                    gov_id = %gov_id,
153                    subject_id = %subject_id,
154                    schema_id = %schema_id,
155                    "Subject registration enqueued"
156                );
157                RegisterEvent::RegisterSubj {
158                    gov_id,
159                    subject_id,
160                    data: RegisterDataSubj {
161                        schema_id,
162                        active: true,
163                        namespace,
164                        name,
165                        description,
166                    },
167                }
168            }
169            RegisterMessage::EOLSubj { gov_id, subj_id } => {
170                debug!(
171                    msg_type = "EOLSubj",
172                    gov_id = %gov_id,
173                    subj_id = %subj_id,
174                    "Subject EOL enqueued"
175                );
176                RegisterEvent::EOLSubj { gov_id, subj_id }
177            }
178        };
179
180        self.on_event(event, ctx).await;
181        Ok(RegisterResponse::None)
182    }
183
184    async fn on_event(
185        &mut self,
186        event: RegisterEvent,
187        ctx: &mut ActorContext<Self>,
188    ) {
189        if let Err(e) = ctx.publish_event(event.clone()).await {
190            error!(error = %e, event = ?event, "Failed to publish register event");
191            emit_fail(ctx, e).await;
192        } else {
193            debug!(event = ?event, "Register event published successfully");
194        }
195    }
196}
ave_core/node/register.rs

ave_core/node/
register.rs
