krabka_admin_ui/
dto.rs

1use std::fmt;
2
3use serde::{Deserialize, Serialize, de::DeserializeOwned};
4
5/// A request shape that an HTML form can carry.
6///
7/// A browser form posts flat `application/x-www-form-urlencoded` pairs, so a
8/// request that holds a list needs a flat twin. Each mutation endpoint names
9/// the form shape it accepts, and the server turns that shape into the request
10/// the mutation takes. A request that is already flat is its own form shape.
11pub trait MutationForm: DeserializeOwned {
12    /// The request the mutation takes.
13    type Request: DeserializeOwned;
14
15    /// # Errors
16    /// Returns the reason the submitted form does not describe a request.
17    fn into_request(self) -> Result<Self::Request, String>;
18}
19
20/// Reads `name=value,name=value` out of one form field.
21///
22/// # Errors
23/// Returns the reason an entry does not read `name=value`.
24fn config_entries_from_form(value: &str) -> Result<Vec<ConfigEntryDto>, String> {
25    value
26        .split(',')
27        .map(str::trim)
28        .filter(|entry| !entry.is_empty())
29        .map(|entry| {
30            let Some((name, value)) = entry.split_once('=') else {
31                return Err(format!("config entry \"{entry}\" must read name=value"));
32            };
33
34            Ok(ConfigEntryDto {
35                name: name.trim().to_string(),
36                value: value.trim().to_string(),
37            })
38        })
39        .collect()
40}
41
42#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
43pub struct ConfigEntryDto {
44    pub name: String,
45    pub value: String,
46}
47
48impl ConfigEntryDto {
49    /// # Errors
50    /// Returns an error when the request is invalid, authentication or session validation fails, or the broker admin operation reports a failure.
51    pub fn validate(&self) -> Result<(), String> {
52        ensure_not_blank("config name", &self.name)
53    }
54}
55
56#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
57pub struct CreateTopicRequestDto {
58    pub name: String,
59    pub partitions: i32,
60    pub replicas: i32,
61    pub configs: Vec<ConfigEntryDto>,
62}
63
64impl CreateTopicRequestDto {
65    /// # Errors
66    /// Returns an error when the request is invalid, authentication or session validation fails, or the broker admin operation reports a failure.
67    pub fn validate(&self) -> Result<(), String> {
68        ensure_not_blank("topic name", &self.name)?;
69        ensure_positive("partition count", self.partitions)?;
70        ensure_positive("replica count", self.replicas)?;
71
72        for config in &self.configs {
73            config.validate()?;
74        }
75
76        Ok(())
77    }
78}
79
80#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
81pub struct DeleteTopicRequestDto {
82    pub name: String,
83}
84
85impl DeleteTopicRequestDto {
86    /// # Errors
87    /// Returns an error when the request is invalid, authentication or session validation fails, or the broker admin operation reports a failure.
88    pub fn validate(&self) -> Result<(), String> {
89        ensure_not_blank("topic name", &self.name)
90    }
91}
92
93#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
94pub struct CreatePartitionsRequestDto {
95    pub topic: String,
96    pub total_count: i32,
97}
98
99impl CreatePartitionsRequestDto {
100    /// # Errors
101    /// Returns an error when the request is invalid, authentication or session validation fails, or the broker admin operation reports a failure.
102    pub fn validate(&self) -> Result<(), String> {
103        ensure_not_blank("topic name", &self.topic)?;
104        ensure_positive("partition count", self.total_count)
105    }
106}
107
108#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
109pub struct AlterConfigRequestDto {
110    pub resource_type: String,
111    pub resource_name: String,
112    pub configs: Vec<ConfigEntryDto>,
113}
114
115impl AlterConfigRequestDto {
116    /// # Errors
117    /// Returns an error when the request is invalid, authentication or session validation fails, or the broker admin operation reports a failure.
118    pub fn validate(&self) -> Result<(), String> {
119        ensure_not_blank("config resource type", &self.resource_type)?;
120        ensure_not_blank("config resource name", &self.resource_name)?;
121
122        for config in &self.configs {
123            config.validate()?;
124        }
125
126        Ok(())
127    }
128}
129
130#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
131pub struct AclRequestDto {
132    pub resource_type: String,
133    pub resource_name: String,
134    /// `literal` or `prefixed`. A prefixed ACL only matches a filter that
135    /// names the same pattern type, so a delete needs it as much as a create.
136    pub pattern_type: String,
137    pub principal: String,
138    pub operation: String,
139    pub permission: String,
140    pub host: String,
141}
142
143impl AclRequestDto {
144    /// # Errors
145    /// Returns an error when the request is invalid, authentication or session validation fails, or the broker admin operation reports a failure.
146    pub fn validate(&self) -> Result<(), String> {
147        ensure_not_blank("ACL resource type", &self.resource_type)?;
148        ensure_not_blank("ACL resource name", &self.resource_name)?;
149        ensure_not_blank("ACL pattern type", &self.pattern_type)?;
150        ensure_not_blank("ACL principal", &self.principal)?;
151        ensure_not_blank("ACL operation", &self.operation)?;
152        ensure_not_blank("ACL permission", &self.permission)?;
153        ensure_not_blank("ACL host", &self.host)
154    }
155}
156
157#[derive(Clone, PartialEq, Eq, Serialize, Deserialize)]
158pub struct ScramUserUpsertDto {
159    pub username: String,
160    pub password: String,
161    pub iterations: i32,
162}
163
164impl ScramUserUpsertDto {
165    /// # Errors
166    /// Returns an error when the request is invalid, authentication or session validation fails, or the broker admin operation reports a failure.
167    pub fn validate(&self) -> Result<(), String> {
168        ensure_not_blank("SCRAM username", &self.username)?;
169        ensure_not_blank("SCRAM password", &self.password)?;
170        ensure_positive("SCRAM iterations", self.iterations)
171    }
172}
173
174impl fmt::Debug for ScramUserUpsertDto {
175    fn fmt(&self, formatter: &mut fmt::Formatter<'_>) -> fmt::Result {
176        formatter
177            .debug_struct("ScramUserUpsertDto")
178            .field("username", &self.username)
179            .field("password", &"<redacted>")
180            .field("iterations", &self.iterations)
181            .finish()
182    }
183}
184
185#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
186pub struct ScramUserDeleteDto {
187    pub username: String,
188}
189
190impl ScramUserDeleteDto {
191    /// # Errors
192    /// Returns an error when the request is invalid, authentication or session validation fails, or the broker admin operation reports a failure.
193    pub fn validate(&self) -> Result<(), String> {
194        ensure_not_blank("SCRAM username", &self.username)
195    }
196}
197
198#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)]
199pub struct QuotaUpsertDto {
200    pub entity: String,
201    pub quota_type: String,
202    pub value: f64,
203}
204
205impl QuotaUpsertDto {
206    /// # Errors
207    /// Returns an error when the request is invalid, authentication or session validation fails, or the broker admin operation reports a failure.
208    pub fn validate(&self) -> Result<(), String> {
209        ensure_not_blank("quota entity", &self.entity)?;
210        ensure_not_blank("quota type", &self.quota_type)?;
211        ensure_finite("quota value", self.value)
212    }
213}
214
215#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
216pub struct QuotaDeleteDto {
217    pub entity: String,
218    pub quota_type: String,
219}
220
221impl QuotaDeleteDto {
222    /// # Errors
223    /// Returns an error when the request is invalid, authentication or session validation fails, or the broker admin operation reports a failure.
224    pub fn validate(&self) -> Result<(), String> {
225        ensure_not_blank("quota entity", &self.entity)?;
226        ensure_not_blank("quota type", &self.quota_type)
227    }
228}
229
230#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
231pub struct LogDirMoveRequestDto {
232    pub topic: String,
233    pub partition: i32,
234    /// The broker whose replica moves.
235    pub broker_id: i32,
236    pub destination_log_dir: String,
237}
238
239impl LogDirMoveRequestDto {
240    /// # Errors
241    /// Returns an error when the request is invalid, authentication or session validation fails, or the broker admin operation reports a failure.
242    pub fn validate(&self) -> Result<(), String> {
243        ensure_not_blank("log-dir move topic", &self.topic)?;
244        ensure_nonnegative("log-dir move partition", self.partition)?;
245        ensure_nonnegative("log-dir move broker id", self.broker_id)?;
246        ensure_not_blank("destination log dir", &self.destination_log_dir)
247    }
248}
249
250/// The flat form twin of [`CreateTopicRequestDto`].
251#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
252pub struct CreateTopicFormDto {
253    pub name: String,
254    pub partitions: i32,
255    pub replicas: i32,
256    /// `name=value,name=value`, or empty for the broker defaults.
257    pub configs: String,
258}
259
260impl MutationForm for CreateTopicFormDto {
261    type Request = CreateTopicRequestDto;
262
263    fn into_request(self) -> Result<Self::Request, String> {
264        Ok(CreateTopicRequestDto {
265            name: self.name,
266            partitions: self.partitions,
267            replicas: self.replicas,
268            configs: config_entries_from_form(&self.configs)?,
269        })
270    }
271}
272
273/// The flat form twin of [`AlterConfigRequestDto`].
274#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
275pub struct AlterConfigFormDto {
276    pub resource_type: String,
277    pub resource_name: String,
278    /// `name=value,name=value`.
279    pub configs: String,
280}
281
282impl MutationForm for AlterConfigFormDto {
283    type Request = AlterConfigRequestDto;
284
285    fn into_request(self) -> Result<Self::Request, String> {
286        Ok(AlterConfigRequestDto {
287            resource_type: self.resource_type,
288            resource_name: self.resource_name,
289            configs: config_entries_from_form(&self.configs)?,
290        })
291    }
292}
293
294macro_rules! flat_mutation_form {
295    ($($request:ty),+ $(,)?) => {
296        $(
297            impl MutationForm for $request {
298                type Request = Self;
299
300                fn into_request(self) -> Result<Self::Request, String> {
301                    Ok(self)
302                }
303            }
304        )+
305    };
306}
307
308flat_mutation_form!(
309    DeleteTopicRequestDto,
310    CreatePartitionsRequestDto,
311    AclRequestDto,
312    ScramUserUpsertDto,
313    ScramUserDeleteDto,
314    QuotaUpsertDto,
315    QuotaDeleteDto,
316    LogDirMoveRequestDto,
317);
318
319fn ensure_not_blank(field: &str, value: &str) -> Result<(), String> {
320    if value.trim().is_empty() {
321        return Err(format!("{field} must not be blank"));
322    }
323
324    Ok(())
325}
326
327fn ensure_positive(field: &str, value: i32) -> Result<(), String> {
328    if value <= 0 {
329        return Err(format!("{field} must be positive"));
330    }
331
332    Ok(())
333}
334
335fn ensure_nonnegative(field: &str, value: i32) -> Result<(), String> {
336    if value < 0 {
337        return Err(format!("{field} must be nonnegative"));
338    }
339
340    Ok(())
341}
342
343fn ensure_finite(field: &str, value: f64) -> Result<(), String> {
344    if !value.is_finite() {
345        return Err(format!("{field} must be finite"));
346    }
347
348    Ok(())
349}
350
351#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
352pub struct KafkaErrorDto {
353    pub code: i16,
354    pub name: String,
355    pub message: Option<String>,
356}
357
358#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
359pub struct ResourceOutcome {
360    pub resource: String,
361    pub error: Option<KafkaErrorDto>,
362}
363
364impl ResourceOutcome {
365    #[must_use]
366    pub fn ok(resource: impl Into<String>) -> Self {
367        Self {
368            resource: resource.into(),
369            error: None,
370        }
371    }
372
373    #[must_use]
374    pub fn failed(resource: impl Into<String>, error: KafkaErrorDto) -> Self {
375        Self {
376            resource: resource.into(),
377            error: Some(error),
378        }
379    }
380
381    #[must_use]
382    pub const fn has_error(&self) -> bool {
383        self.error.is_some()
384    }
385}
386
387#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
388pub struct TopicRow {
389    pub name: String,
390    pub topic_id: Option<String>,
391    pub partition_count: i32,
392    pub replication_factor: i32,
393    pub error: Option<KafkaErrorDto>,
394}
395
396#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
397pub struct GroupRow {
398    pub group_id: String,
399}
400
401#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
402pub struct AclRow {
403    pub resource: String,
404    pub pattern_type: String,
405    pub principal: String,
406    /// The host the rule applies to. `*` is every host. Two rules that differ
407    /// only here are different rules.
408    pub host: String,
409    pub operation: String,
410    pub permission: String,
411}
412
413#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
414pub struct UserRow {
415    pub username: String,
416    pub principal: String,
417}
418
419#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
420pub struct QuotaRow {
421    pub entity: String,
422    pub quota_type: String,
423    pub value: String,
424}
425
426#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
427pub struct LogDirRow {
428    pub broker_id: i32,
429    pub log_dir: String,
430    pub topic: String,
431    pub partition: i32,
432    pub partition_size: i64,
433    pub offset_lag: i64,
434    pub is_future_key: bool,
435    pub error: Option<KafkaErrorDto>,
436}
krabka_admin_ui/dto.rs

krabka_admin_ui/
dto.rs
