krabka_admin_ui/
server_fns.rs

1//! Server-function seam for the Dioxus admin UI shell.
2
3use std::{collections::BTreeMap, future::Future, pin::Pin};
4
5use krabka_client_admin::{
6    AclEntry, AclEntryFilter, AclOperation, AdminClient, AlterConfigOp, ConfigResource,
7    CreatePartitionsOp, CreateTopicSpec, IncrementalAlterConfigsOptions, PatternType,
8    PermissionType, QuotaOp, ResourceType, ScramDeletion, ScramUpsertion, TopicMutationOptions,
9    TopicPartitionReplica,
10};
11use serde::{Deserialize, Serialize};
12
13pub use crate::dto::{AclRow, QuotaRow, UserRow};
14use crate::{
15    admin::{AdminFacade, quota_mutation_outcome, resource_outcome_rows},
16    auth::{LoginBroker, LoginRequest, LoginSuccess, build_scram_sha512_security},
17    config::AdminUiConfig,
18    dto::{
19        AclRequestDto, AlterConfigRequestDto, CreatePartitionsRequestDto, CreateTopicRequestDto,
20        DeleteTopicRequestDto, GroupRow, LogDirMoveRequestDto, LogDirRow, QuotaDeleteDto,
21        QuotaUpsertDto, ResourceOutcome, ScramUserDeleteDto, ScramUserUpsertDto, TopicRow,
22    },
23    error::UiError,
24    server::AppState,
25    session::{SessionId, SessionRecord, SessionStore},
26};
27
28/// The configured topic-mutation timeout, as the quantity that `AdminClient` takes.
29///
30/// The broker gets this much time to finish a UI-issued topic mutation. After that
31/// time, the broker answers with a timeout error. `CreateTopics`, `DeleteTopics`,
32/// and `CreatePartitions` carry the value as Kafka's `timeout_ms` field.
33fn mutation_timeout(cfg: &AdminUiConfig) -> TopicMutationOptions {
34    TopicMutationOptions::with_timeout(cfg.topic_mutation_timeout)
35}
36
37#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
38pub struct CurrentSession {
39    pub username: String,
40    pub principal: String,
41}
42
43/// # Errors
44/// Returns an error when the request is invalid, authentication or session validation fails, or the broker admin operation reports a failure.
45pub fn logout<F>(context: &ServerFunctionContext<'_, F>) -> Result<(), UiError> {
46    logout_with_context(context)
47}
48
49/// # Errors
50/// Returns an error when the request is invalid, authentication or session validation fails, or the broker admin operation reports a failure.
51pub async fn create_topic<F: AdminSeamFactory>(
52    context: &ServerFunctionContext<'_, F>,
53    request: CreateTopicRequestDto,
54) -> Result<Vec<ResourceOutcome>, UiError> {
55    create_topic_with_context(context, request).await
56}
57
58/// # Errors
59/// Returns an error when the request is invalid, authentication or session validation fails, or the broker admin operation reports a failure.
60pub async fn delete_topic<F: AdminSeamFactory>(
61    context: &ServerFunctionContext<'_, F>,
62    request: DeleteTopicRequestDto,
63) -> Result<Vec<ResourceOutcome>, UiError> {
64    delete_topic_with_context(context, request).await
65}
66
67/// # Errors
68/// Returns an error when the request is invalid, authentication or session validation fails, or the broker admin operation reports a failure.
69pub async fn create_partitions<F: AdminSeamFactory>(
70    context: &ServerFunctionContext<'_, F>,
71    request: CreatePartitionsRequestDto,
72) -> Result<Vec<ResourceOutcome>, UiError> {
73    create_partitions_with_context(context, request).await
74}
75
76/// # Errors
77/// Returns an error when the request is invalid, authentication or session validation fails, or the broker admin operation reports a failure.
78pub async fn alter_configs<F: AdminSeamFactory>(
79    context: &ServerFunctionContext<'_, F>,
80    request: AlterConfigRequestDto,
81) -> Result<Vec<ResourceOutcome>, UiError> {
82    alter_configs_with_context(context, request).await
83}
84
85/// # Errors
86/// Returns an error when the request is invalid, authentication or session validation fails, or the broker admin operation reports a failure.
87pub async fn create_acl<F: AdminSeamFactory>(
88    context: &ServerFunctionContext<'_, F>,
89    request: AclRequestDto,
90) -> Result<Vec<ResourceOutcome>, UiError> {
91    create_acl_with_context(context, request).await
92}
93
94/// # Errors
95/// Returns an error when the request is invalid, authentication or session validation fails, or the broker admin operation reports a failure.
96pub async fn delete_acl<F: AdminSeamFactory>(
97    context: &ServerFunctionContext<'_, F>,
98    request: AclRequestDto,
99) -> Result<Vec<ResourceOutcome>, UiError> {
100    delete_acl_with_context(context, request).await
101}
102
103/// # Errors
104/// Returns an error when the request is invalid, authentication or session validation fails, or the broker admin operation reports a failure.
105pub async fn upsert_scram_sha512_user<F: AdminSeamFactory>(
106    context: &ServerFunctionContext<'_, F>,
107    request: ScramUserUpsertDto,
108) -> Result<Vec<ResourceOutcome>, UiError> {
109    upsert_scram_sha512_user_with_context(context, request).await
110}
111
112/// # Errors
113/// Returns an error when the request is invalid, authentication or session validation fails, or the broker admin operation reports a failure.
114pub async fn delete_scram_user<F: AdminSeamFactory>(
115    context: &ServerFunctionContext<'_, F>,
116    request: ScramUserDeleteDto,
117) -> Result<Vec<ResourceOutcome>, UiError> {
118    delete_scram_user_with_context(context, request).await
119}
120
121/// # Errors
122/// Returns an error when the request is invalid, authentication or session validation fails, or the broker admin operation reports a failure.
123pub async fn upsert_quota<F: AdminSeamFactory>(
124    context: &ServerFunctionContext<'_, F>,
125    request: QuotaUpsertDto,
126) -> Result<Vec<ResourceOutcome>, UiError> {
127    upsert_quota_with_context(context, request).await
128}
129
130/// # Errors
131/// Returns an error when the request is invalid, authentication or session validation fails, or the broker admin operation reports a failure.
132pub async fn delete_quota<F: AdminSeamFactory>(
133    context: &ServerFunctionContext<'_, F>,
134    request: QuotaDeleteDto,
135) -> Result<Vec<ResourceOutcome>, UiError> {
136    delete_quota_with_context(context, request).await
137}
138
139/// # Errors
140/// Returns an error when the request is invalid, authentication or session validation fails, or the broker admin operation reports a failure.
141pub async fn move_log_dir<F: AdminSeamFactory>(
142    context: &ServerFunctionContext<'_, F>,
143    request: LogDirMoveRequestDto,
144) -> Result<Vec<ResourceOutcome>, UiError> {
145    move_log_dir_with_context(context, request).await
146}
147
148pub trait AdminReadSeam {
149    fn topics<'a>(
150        &'a self,
151    ) -> Pin<Box<dyn Future<Output = Result<Vec<TopicRow>, UiError>> + Send + 'a>>;
152
153    fn groups<'a>(
154        &'a self,
155    ) -> Pin<Box<dyn Future<Output = Result<Vec<GroupRow>, UiError>> + Send + 'a>>;
156
157    fn acls<'a>(
158        &'a self,
159    ) -> Pin<Box<dyn Future<Output = Result<Vec<AclRow>, UiError>> + Send + 'a>>;
160
161    fn users<'a>(
162        &'a self,
163    ) -> Pin<Box<dyn Future<Output = Result<Vec<UserRow>, UiError>> + Send + 'a>>;
164
165    /// The quotas of `entity`, or of the signed-in operator when `entity` is
166    /// `None`.
167    fn quotas<'a>(
168        &'a self,
169        entity: Option<String>,
170    ) -> Pin<Box<dyn Future<Output = Result<Vec<QuotaRow>, UiError>> + Send + 'a>>;
171
172    fn log_dirs<'a>(
173        &'a self,
174    ) -> Pin<Box<dyn Future<Output = Result<Vec<LogDirRow>, UiError>> + Send + 'a>>;
175}
176
177pub trait AdminMutationSeam {
178    fn create_topic<'a>(
179        &'a self,
180        request: CreateTopicRequestDto,
181    ) -> Pin<Box<dyn Future<Output = Result<Vec<ResourceOutcome>, UiError>> + Send + 'a>>;
182
183    fn delete_topic<'a>(
184        &'a self,
185        request: DeleteTopicRequestDto,
186    ) -> Pin<Box<dyn Future<Output = Result<Vec<ResourceOutcome>, UiError>> + Send + 'a>>;
187
188    fn create_partitions<'a>(
189        &'a self,
190        request: CreatePartitionsRequestDto,
191    ) -> Pin<Box<dyn Future<Output = Result<Vec<ResourceOutcome>, UiError>> + Send + 'a>>;
192
193    fn alter_configs<'a>(
194        &'a self,
195        request: AlterConfigRequestDto,
196    ) -> Pin<Box<dyn Future<Output = Result<Vec<ResourceOutcome>, UiError>> + Send + 'a>>;
197
198    fn create_acl<'a>(
199        &'a self,
200        request: AclRequestDto,
201    ) -> Pin<Box<dyn Future<Output = Result<Vec<ResourceOutcome>, UiError>> + Send + 'a>>;
202
203    fn delete_acl<'a>(
204        &'a self,
205        request: AclRequestDto,
206    ) -> Pin<Box<dyn Future<Output = Result<Vec<ResourceOutcome>, UiError>> + Send + 'a>>;
207
208    fn upsert_scram_sha512_user<'a>(
209        &'a self,
210        request: ScramUserUpsertDto,
211    ) -> Pin<Box<dyn Future<Output = Result<Vec<ResourceOutcome>, UiError>> + Send + 'a>>;
212
213    fn delete_scram_user<'a>(
214        &'a self,
215        request: ScramUserDeleteDto,
216    ) -> Pin<Box<dyn Future<Output = Result<Vec<ResourceOutcome>, UiError>> + Send + 'a>>;
217
218    fn upsert_quota<'a>(
219        &'a self,
220        request: QuotaUpsertDto,
221    ) -> Pin<Box<dyn Future<Output = Result<Vec<ResourceOutcome>, UiError>> + Send + 'a>>;
222
223    fn delete_quota<'a>(
224        &'a self,
225        request: QuotaDeleteDto,
226    ) -> Pin<Box<dyn Future<Output = Result<Vec<ResourceOutcome>, UiError>> + Send + 'a>>;
227
228    fn move_log_dir<'a>(
229        &'a self,
230        request: LogDirMoveRequestDto,
231    ) -> Pin<Box<dyn Future<Output = Result<Vec<ResourceOutcome>, UiError>> + Send + 'a>>;
232}
233
234pub trait AdminSeamFactory {
235    type Reader<'a>: AdminReadSeam
236    where
237        Self: 'a;
238    type Mutations<'a>: AdminMutationSeam
239    where
240        Self: 'a;
241
242    /// # Errors
243    /// Returns an error when the request is invalid, authentication or session validation fails, or the broker admin operation reports a failure.
244    fn read_seam<'a>(
245        &'a self,
246        cfg: &AdminUiConfig,
247        record: &SessionRecord,
248    ) -> Result<Self::Reader<'a>, UiError>;
249
250    /// # Errors
251    /// Returns an error when the request is invalid, authentication or session validation fails, or the broker admin operation reports a failure.
252    fn mutation_seam<'a>(
253        &'a self,
254        cfg: &AdminUiConfig,
255        record: &SessionRecord,
256    ) -> Result<Self::Mutations<'a>, UiError>;
257}
258
259#[derive(Clone, Copy, Debug, Default)]
260pub struct BrokerAdminSeamFactory;
261
262impl AdminSeamFactory for BrokerAdminSeamFactory {
263    type Reader<'a> = BrokerAdminReadSeam;
264    type Mutations<'a> = BrokerAdminMutationSeam;
265
266    fn read_seam<'a>(
267        &'a self,
268        cfg: &AdminUiConfig,
269        record: &SessionRecord,
270    ) -> Result<Self::Reader<'a>, UiError> {
271        BrokerAdminReadSeam::from_session(cfg, record)
272    }
273
274    fn mutation_seam<'a>(
275        &'a self,
276        cfg: &AdminUiConfig,
277        record: &SessionRecord,
278    ) -> Result<Self::Mutations<'a>, UiError> {
279        BrokerAdminMutationSeam::from_session(cfg, record)
280    }
281}
282
283pub struct ServerFunctionContext<'a, F = BrokerAdminSeamFactory> {
284    pub cfg: &'a AdminUiConfig,
285    pub sessions: &'a SessionStore,
286    pub raw_session_id: Option<String>,
287    pub seam_factory: &'a F,
288}
289
290impl<'a, F> ServerFunctionContext<'a, F> {
291    #[must_use]
292    pub fn new(
293        cfg: &'a AdminUiConfig,
294        sessions: &'a SessionStore,
295        raw_session_id: Option<&str>,
296        seam_factory: &'a F,
297    ) -> Self {
298        Self {
299            cfg,
300            sessions,
301            raw_session_id: raw_session_id.map(str::to_owned),
302            seam_factory,
303        }
304    }
305}
306
307/// # Errors
308/// Returns an error when the request is invalid, authentication or session validation fails, or the broker admin operation reports a failure.
309pub async fn login_with_context<B: LoginBroker>(
310    cfg: &AdminUiConfig,
311    sessions: &SessionStore,
312    broker: &B,
313    request: LoginRequest,
314) -> Result<LoginSuccess, UiError> {
315    crate::auth::AuthService::new_with_broker(cfg, sessions, broker)
316        .login(request)
317        .await
318}
319
320/// # Errors
321/// Returns an error when the request is invalid, authentication or session validation fails, or the broker admin operation reports a failure.
322pub async fn login_with_app_state<B: LoginBroker>(
323    state: &AppState,
324    broker: &B,
325    request: LoginRequest,
326) -> Result<LoginSuccess, UiError> {
327    login_with_context(&state.cfg, &state.sessions, broker, request).await
328}
329
330/// # Errors
331/// Returns an error when the request is invalid, authentication or session validation fails, or the broker admin operation reports a failure.
332pub fn logout_with_context<F>(context: &ServerFunctionContext<'_, F>) -> Result<(), UiError> {
333    let session_id = require_session_id(context.sessions, context.raw_session_id.as_deref())?;
334
335    context.sessions.remove(&session_id);
336    Ok(())
337}
338
339/// # Errors
340/// Returns an error when the request is invalid, authentication or session validation fails, or the broker admin operation reports a failure.
341pub fn current_session_with_store(
342    sessions: &SessionStore,
343    raw_session_id: Option<&str>,
344) -> Result<CurrentSession, UiError> {
345    let record = require_session(sessions, raw_session_id)?;
346
347    Ok(CurrentSession {
348        username: record.user.username,
349        principal: record.user.principal,
350    })
351}
352
353/// The stored session behind a cookie value.
354///
355/// The HTTP layer needs the record itself: the record carries the operator's
356/// capabilities and the CSRF token that a mutation must repeat.
357///
358/// # Errors
359/// Returns an error when the request is invalid, authentication or session validation fails, or the broker admin operation reports a failure.
360pub fn session_record_with_store(
361    sessions: &SessionStore,
362    raw_session_id: Option<&str>,
363) -> Result<SessionRecord, UiError> {
364    require_session(sessions, raw_session_id)
365}
366
367/// # Errors
368/// Returns an error when the request is invalid, authentication or session validation fails, or the broker admin operation reports a failure.
369pub fn current_session_with_context<F>(
370    context: &ServerFunctionContext<'_, F>,
371) -> Result<CurrentSession, UiError> {
372    current_session_with_store(context.sessions, context.raw_session_id.as_deref())
373}
374
375/// # Errors
376/// Returns an error when the request is invalid, authentication or session validation fails, or the broker admin operation reports a failure.
377pub async fn list_topics_with_reader<R: AdminReadSeam>(
378    sessions: &SessionStore,
379    raw_session_id: Option<&str>,
380    reader: &R,
381) -> Result<Vec<TopicRow>, UiError> {
382    require_session(sessions, raw_session_id)?;
383
384    reader.topics().await
385}
386
387/// # Errors
388/// Returns an error when the request is invalid, authentication or session validation fails, or the broker admin operation reports a failure.
389pub async fn list_topics_with_context<F: AdminSeamFactory>(
390    context: &ServerFunctionContext<'_, F>,
391) -> Result<Vec<TopicRow>, UiError> {
392    let reader = read_seam_from_context(context)?;
393
394    reader.topics().await
395}
396
397/// # Errors
398/// Returns an error when the request is invalid, authentication or session validation fails, or the broker admin operation reports a failure.
399pub async fn list_groups_with_reader<R: AdminReadSeam>(
400    sessions: &SessionStore,
401    raw_session_id: Option<&str>,
402    reader: &R,
403) -> Result<Vec<GroupRow>, UiError> {
404    require_session(sessions, raw_session_id)?;
405
406    reader.groups().await
407}
408
409/// # Errors
410/// Returns an error when the request is invalid, authentication or session validation fails, or the broker admin operation reports a failure.
411pub async fn list_groups_with_context<F: AdminSeamFactory>(
412    context: &ServerFunctionContext<'_, F>,
413) -> Result<Vec<GroupRow>, UiError> {
414    let reader = read_seam_from_context(context)?;
415
416    reader.groups().await
417}
418
419/// # Errors
420/// Returns an error when the request is invalid, authentication or session validation fails, or the broker admin operation reports a failure.
421pub async fn list_acls_with_reader<R: AdminReadSeam>(
422    sessions: &SessionStore,
423    raw_session_id: Option<&str>,
424    reader: &R,
425) -> Result<Vec<AclRow>, UiError> {
426    require_session(sessions, raw_session_id)?;
427
428    reader.acls().await
429}
430
431/// # Errors
432/// Returns an error when the request is invalid, authentication or session validation fails, or the broker admin operation reports a failure.
433pub async fn list_acls<F: AdminSeamFactory>(
434    context: &ServerFunctionContext<'_, F>,
435) -> Result<Vec<AclRow>, UiError> {
436    let reader = read_seam_from_context(context)?;
437
438    reader.acls().await
439}
440
441/// # Errors
442/// Returns an error when the request is invalid, authentication or session validation fails, or the broker admin operation reports a failure.
443pub async fn list_users_with_reader<R: AdminReadSeam>(
444    sessions: &SessionStore,
445    raw_session_id: Option<&str>,
446    reader: &R,
447) -> Result<Vec<UserRow>, UiError> {
448    require_session(sessions, raw_session_id)?;
449
450    reader.users().await
451}
452
453/// # Errors
454/// Returns an error when the request is invalid, authentication or session validation fails, or the broker admin operation reports a failure.
455pub async fn list_users<F: AdminSeamFactory>(
456    context: &ServerFunctionContext<'_, F>,
457) -> Result<Vec<UserRow>, UiError> {
458    let reader = read_seam_from_context(context)?;
459
460    reader.users().await
461}
462
463/// # Errors
464/// Returns an error when the request is invalid, authentication or session validation fails, or the broker admin operation reports a failure.
465pub async fn list_quotas_with_reader<R: AdminReadSeam>(
466    sessions: &SessionStore,
467    raw_session_id: Option<&str>,
468    reader: &R,
469    entity: Option<String>,
470) -> Result<Vec<QuotaRow>, UiError> {
471    require_session(sessions, raw_session_id)?;
472
473    reader.quotas(entity).await
474}
475
476/// # Errors
477/// Returns an error when the request is invalid, authentication or session validation fails, or the broker admin operation reports a failure.
478pub async fn list_quotas<F: AdminSeamFactory>(
479    context: &ServerFunctionContext<'_, F>,
480    entity: Option<String>,
481) -> Result<Vec<QuotaRow>, UiError> {
482    let reader = read_seam_from_context(context)?;
483
484    reader.quotas(entity).await
485}
486
487/// # Errors
488/// Returns an error when the request is invalid, authentication or session validation fails, or the broker admin operation reports a failure.
489pub async fn list_log_dirs_with_reader<R: AdminReadSeam>(
490    sessions: &SessionStore,
491    raw_session_id: Option<&str>,
492    reader: &R,
493) -> Result<Vec<LogDirRow>, UiError> {
494    require_session(sessions, raw_session_id)?;
495
496    reader.log_dirs().await
497}
498
499/// # Errors
500/// Returns an error when the request is invalid, authentication or session validation fails, or the broker admin operation reports a failure.
501pub async fn list_log_dirs_with_context<F: AdminSeamFactory>(
502    context: &ServerFunctionContext<'_, F>,
503) -> Result<Vec<LogDirRow>, UiError> {
504    let reader = read_seam_from_context(context)?;
505
506    reader.log_dirs().await
507}
508
509/// # Errors
510/// Returns an error when the request is invalid, authentication or session validation fails, or the broker admin operation reports a failure.
511pub async fn create_topic_with_mutations<M: AdminMutationSeam>(
512    sessions: &SessionStore,
513    raw_session_id: Option<&str>,
514    mutations: &M,
515    request: CreateTopicRequestDto,
516) -> Result<Vec<ResourceOutcome>, UiError> {
517    require_session(sessions, raw_session_id)?;
518    ensure_valid_request(request.validate())?;
519
520    mutations.create_topic(request).await
521}
522
523/// # Errors
524/// Returns an error when the request is invalid, authentication or session validation fails, or the broker admin operation reports a failure.
525pub async fn create_topic_with_context<F: AdminSeamFactory>(
526    context: &ServerFunctionContext<'_, F>,
527    request: CreateTopicRequestDto,
528) -> Result<Vec<ResourceOutcome>, UiError> {
529    let mutations = mutation_seam_after_valid_request(context, request.validate())?;
530
531    mutations.create_topic(request).await
532}
533
534/// # Errors
535/// Returns an error when the request is invalid, authentication or session validation fails, or the broker admin operation reports a failure.
536pub async fn delete_topic_with_context<F: AdminSeamFactory>(
537    context: &ServerFunctionContext<'_, F>,
538    request: DeleteTopicRequestDto,
539) -> Result<Vec<ResourceOutcome>, UiError> {
540    let mutations = mutation_seam_after_valid_request(context, request.validate())?;
541
542    mutations.delete_topic(request).await
543}
544
545/// # Errors
546/// Returns an error when the request is invalid, authentication or session validation fails, or the broker admin operation reports a failure.
547pub async fn create_partitions_with_context<F: AdminSeamFactory>(
548    context: &ServerFunctionContext<'_, F>,
549    request: CreatePartitionsRequestDto,
550) -> Result<Vec<ResourceOutcome>, UiError> {
551    let mutations = mutation_seam_after_valid_request(context, request.validate())?;
552
553    mutations.create_partitions(request).await
554}
555
556/// # Errors
557/// Returns an error when the request is invalid, authentication or session validation fails, or the broker admin operation reports a failure.
558pub async fn alter_configs_with_context<F: AdminSeamFactory>(
559    context: &ServerFunctionContext<'_, F>,
560    request: AlterConfigRequestDto,
561) -> Result<Vec<ResourceOutcome>, UiError> {
562    let mutations = mutation_seam_after_valid_request(context, request.validate())?;
563
564    mutations.alter_configs(request).await
565}
566
567/// # Errors
568/// Returns an error when the request is invalid, authentication or session validation fails, or the broker admin operation reports a failure.
569pub async fn create_acl_with_context<F: AdminSeamFactory>(
570    context: &ServerFunctionContext<'_, F>,
571    request: AclRequestDto,
572) -> Result<Vec<ResourceOutcome>, UiError> {
573    let mutations = mutation_seam_after_valid_request(context, request.validate())?;
574
575    mutations.create_acl(request).await
576}
577
578/// # Errors
579/// Returns an error when the request is invalid, authentication or session validation fails, or the broker admin operation reports a failure.
580pub async fn delete_acl_with_context<F: AdminSeamFactory>(
581    context: &ServerFunctionContext<'_, F>,
582    request: AclRequestDto,
583) -> Result<Vec<ResourceOutcome>, UiError> {
584    let mutations = mutation_seam_after_valid_request(context, request.validate())?;
585
586    mutations.delete_acl(request).await
587}
588
589/// # Errors
590/// Returns an error when the request is invalid, authentication or session validation fails, or the broker admin operation reports a failure.
591pub async fn upsert_scram_sha512_user_with_context<F: AdminSeamFactory>(
592    context: &ServerFunctionContext<'_, F>,
593    request: ScramUserUpsertDto,
594) -> Result<Vec<ResourceOutcome>, UiError> {
595    let mutations = mutation_seam_after_valid_request(context, request.validate())?;
596
597    mutations.upsert_scram_sha512_user(request).await
598}
599
600/// # Errors
601/// Returns an error when the request is invalid, authentication or session validation fails, or the broker admin operation reports a failure.
602pub async fn delete_scram_user_with_context<F: AdminSeamFactory>(
603    context: &ServerFunctionContext<'_, F>,
604    request: ScramUserDeleteDto,
605) -> Result<Vec<ResourceOutcome>, UiError> {
606    let mutations = mutation_seam_after_valid_request(context, request.validate())?;
607
608    mutations.delete_scram_user(request).await
609}
610
611/// # Errors
612/// Returns an error when the request is invalid, authentication or session validation fails, or the broker admin operation reports a failure.
613pub async fn upsert_quota_with_context<F: AdminSeamFactory>(
614    context: &ServerFunctionContext<'_, F>,
615    request: QuotaUpsertDto,
616) -> Result<Vec<ResourceOutcome>, UiError> {
617    let mutations = mutation_seam_after_valid_request(context, request.validate())?;
618
619    mutations.upsert_quota(request).await
620}
621
622/// # Errors
623/// Returns an error when the request is invalid, authentication or session validation fails, or the broker admin operation reports a failure.
624pub async fn delete_quota_with_context<F: AdminSeamFactory>(
625    context: &ServerFunctionContext<'_, F>,
626    request: QuotaDeleteDto,
627) -> Result<Vec<ResourceOutcome>, UiError> {
628    let mutations = mutation_seam_after_valid_request(context, request.validate())?;
629
630    mutations.delete_quota(request).await
631}
632
633/// # Errors
634/// Returns an error when the request is invalid, authentication or session validation fails, or the broker admin operation reports a failure.
635pub async fn move_log_dir_with_context<F: AdminSeamFactory>(
636    context: &ServerFunctionContext<'_, F>,
637    request: LogDirMoveRequestDto,
638) -> Result<Vec<ResourceOutcome>, UiError> {
639    let mutations = mutation_seam_after_valid_request(context, request.validate())?;
640
641    mutations.move_log_dir(request).await
642}
643
644pub struct BrokerAdminReadSeam {
645    cfg: AdminUiConfig,
646    username: String,
647    password: String,
648}
649
650impl BrokerAdminReadSeam {
651    /// # Errors
652    /// Returns an error when the request is invalid, authentication or session validation fails, or the broker admin operation reports a failure.
653    pub fn from_session(cfg: &AdminUiConfig, record: &SessionRecord) -> Result<Self, UiError> {
654        let Some(credentials) = &record.credentials else {
655            return Err(UiError::NotAuthenticated);
656        };
657
658        Ok(Self {
659            cfg: cfg.clone(),
660            username: record.user.username.clone(),
661            password: credentials.password().to_string(),
662        })
663    }
664
665    async fn facade(&self) -> Result<AdminFacade, UiError> {
666        let security = build_scram_sha512_security(&self.cfg, &self.username, &self.password);
667        let client =
668            AdminClient::connect_secured(&self.cfg.bootstrap_addrs, Some(security)).await?;
669
670        Ok(AdminFacade::new(client))
671    }
672}
673
674impl AdminReadSeam for BrokerAdminReadSeam {
675    fn topics<'a>(
676        &'a self,
677    ) -> Pin<Box<dyn Future<Output = Result<Vec<TopicRow>, UiError>> + Send + 'a>> {
678        Box::pin(async move {
679            let mut facade = self.facade().await?;
680            Ok(facade.topics().await?)
681        })
682    }
683
684    fn groups<'a>(
685        &'a self,
686    ) -> Pin<Box<dyn Future<Output = Result<Vec<GroupRow>, UiError>> + Send + 'a>> {
687        Box::pin(async move {
688            let mut facade = self.facade().await?;
689            Ok(facade.groups().await?)
690        })
691    }
692
693    fn acls<'a>(
694        &'a self,
695    ) -> Pin<Box<dyn Future<Output = Result<Vec<AclRow>, UiError>> + Send + 'a>> {
696        Box::pin(async move {
697            let mut facade = self.facade().await?;
698            Ok(facade.acls().await?)
699        })
700    }
701
702    fn users<'a>(
703        &'a self,
704    ) -> Pin<Box<dyn Future<Output = Result<Vec<UserRow>, UiError>> + Send + 'a>> {
705        Box::pin(async move {
706            let mut facade = self.facade().await?;
707            Ok(facade.users().await?)
708        })
709    }
710
711    fn quotas<'a>(
712        &'a self,
713        entity: Option<String>,
714    ) -> Pin<Box<dyn Future<Output = Result<Vec<QuotaRow>, UiError>> + Send + 'a>> {
715        Box::pin(async move {
716            let mut facade = self.facade().await?;
717            let entity = entity.unwrap_or_else(|| self.username.clone());
718
719            Ok(facade.quotas_for_user(&entity).await?)
720        })
721    }
722
723    fn log_dirs<'a>(
724        &'a self,
725    ) -> Pin<Box<dyn Future<Output = Result<Vec<LogDirRow>, UiError>> + Send + 'a>> {
726        Box::pin(async move {
727            let mut facade = self.facade().await?;
728            Ok(facade.log_dirs().await?)
729        })
730    }
731}
732
733pub struct BrokerAdminMutationSeam(BrokerAdminReadSeam);
734
735impl BrokerAdminMutationSeam {
736    /// # Errors
737    /// Returns an error when the request is invalid, authentication or session validation fails, or the broker admin operation reports a failure.
738    pub fn from_session(cfg: &AdminUiConfig, record: &SessionRecord) -> Result<Self, UiError> {
739        Ok(Self(BrokerAdminReadSeam::from_session(cfg, record)?))
740    }
741}
742
743impl AdminMutationSeam for BrokerAdminMutationSeam {
744    fn create_topic<'a>(
745        &'a self,
746        request: CreateTopicRequestDto,
747    ) -> Pin<Box<dyn Future<Output = Result<Vec<ResourceOutcome>, UiError>> + Send + 'a>> {
748        Box::pin(async move {
749            let mut facade = self.0.facade().await?;
750            let configs = request
751                .configs
752                .into_iter()
753                .map(|config| (config.name, config.value))
754                .collect::<BTreeMap<_, _>>();
755            let outcomes = facade
756                .client_mut()
757                .create_topics(
758                    &[CreateTopicSpec {
759                        name: request.name,
760                        partitions: request.partitions,
761                        replicas: request.replicas,
762                        configs,
763                        replica_assignments: BTreeMap::new(),
764                    }],
765                    mutation_timeout(&self.0.cfg),
766                )
767                .await?;
768
769            Ok(resource_outcome_rows(outcomes))
770        })
771    }
772
773    fn delete_topic<'a>(
774        &'a self,
775        request: DeleteTopicRequestDto,
776    ) -> Pin<Box<dyn Future<Output = Result<Vec<ResourceOutcome>, UiError>> + Send + 'a>> {
777        Box::pin(async move {
778            let mut facade = self.0.facade().await?;
779            let outcomes = facade
780                .client_mut()
781                .delete_topics(&[request.name.as_str()], mutation_timeout(&self.0.cfg))
782                .await?;
783
784            Ok(resource_outcome_rows(outcomes))
785        })
786    }
787
788    fn create_partitions<'a>(
789        &'a self,
790        request: CreatePartitionsRequestDto,
791    ) -> Pin<Box<dyn Future<Output = Result<Vec<ResourceOutcome>, UiError>> + Send + 'a>> {
792        Box::pin(async move {
793            let mut facade = self.0.facade().await?;
794            let outcomes = facade
795                .client_mut()
796                .create_partitions(
797                    &[CreatePartitionsOp {
798                        name: request.topic,
799                        new_total_count: request.total_count,
800                        assignments: None,
801                    }],
802                    mutation_timeout(&self.0.cfg),
803                )
804                .await?;
805
806            Ok(resource_outcome_rows(outcomes))
807        })
808    }
809
810    fn alter_configs<'a>(
811        &'a self,
812        request: AlterConfigRequestDto,
813    ) -> Pin<Box<dyn Future<Output = Result<Vec<ResourceOutcome>, UiError>> + Send + 'a>> {
814        Box::pin(async move {
815            ensure_topic_config_resource(&request.resource_type, &request.resource_name)?;
816            let mut facade = self.0.facade().await?;
817            let ops = request
818                .configs
819                .into_iter()
820                .map(|config| AlterConfigOp::set(config.name, config.value))
821                .collect::<Vec<_>>();
822            let changes = BTreeMap::from([(ConfigResource::topic(request.resource_name), ops)]);
823            let outcomes = facade
824                .client_mut()
825                .incremental_alter_configs(&changes, IncrementalAlterConfigsOptions::default())
826                .await?;
827
828            Ok(resource_outcome_rows(outcomes.into_iter().collect()))
829        })
830    }
831
832    fn create_acl<'a>(
833        &'a self,
834        request: AclRequestDto,
835    ) -> Pin<Box<dyn Future<Output = Result<Vec<ResourceOutcome>, UiError>> + Send + 'a>> {
836        Box::pin(async move {
837            let mut facade = self.0.facade().await?;
838            let acl = acl_entry_from_request(&request)?;
839            let outcomes = facade.client_mut().create_acls(&[acl]).await?;
840
841            Ok(resource_outcome_rows(outcomes))
842        })
843    }
844
845    fn delete_acl<'a>(
846        &'a self,
847        request: AclRequestDto,
848    ) -> Pin<Box<dyn Future<Output = Result<Vec<ResourceOutcome>, UiError>> + Send + 'a>> {
849        Box::pin(async move {
850            let mut facade = self.0.facade().await?;
851            let filter = acl_filter_from_request(&request)?;
852            let outcomes = facade.client_mut().delete_acls(&[filter]).await?;
853
854            Ok(resource_outcome_rows(outcomes))
855        })
856    }
857
858    fn upsert_scram_sha512_user<'a>(
859        &'a self,
860        request: ScramUserUpsertDto,
861    ) -> Pin<Box<dyn Future<Output = Result<Vec<ResourceOutcome>, UiError>> + Send + 'a>> {
862        Box::pin(async move {
863            let mut facade = self.0.facade().await?;
864            let outcomes = facade
865                .client_mut()
866                .alter_user_scram_credentials_sha512(
867                    &[ScramUpsertion {
868                        username: request.username,
869                        password: request.password,
870                        iterations: request.iterations,
871                    }],
872                    &[],
873                )
874                .await?;
875
876            Ok(resource_outcome_rows(outcomes))
877        })
878    }
879
880    fn delete_scram_user<'a>(
881        &'a self,
882        request: ScramUserDeleteDto,
883    ) -> Pin<Box<dyn Future<Output = Result<Vec<ResourceOutcome>, UiError>> + Send + 'a>> {
884        Box::pin(async move {
885            let mut facade = self.0.facade().await?;
886            let outcomes = facade
887                .client_mut()
888                .alter_user_scram_credentials_sha512(
889                    &[],
890                    &[ScramDeletion {
891                        username: request.username,
892                    }],
893                )
894                .await?;
895
896            Ok(resource_outcome_rows(outcomes))
897        })
898    }
899
900    fn upsert_quota<'a>(
901        &'a self,
902        request: QuotaUpsertDto,
903    ) -> Pin<Box<dyn Future<Output = Result<Vec<ResourceOutcome>, UiError>> + Send + 'a>> {
904        Box::pin(async move {
905            let mut facade = self.0.facade().await?;
906            let error = facade
907                .client_mut()
908                .alter_user_quotas(
909                    &request.entity,
910                    &[QuotaOp::Set {
911                        key: request.quota_type.clone(),
912                        value: request.value,
913                    }],
914                    false,
915                )
916                .await?;
917
918            Ok(vec![quota_mutation_outcome(
919                &request.entity,
920                &request.quota_type,
921                error,
922            )])
923        })
924    }
925
926    fn delete_quota<'a>(
927        &'a self,
928        request: QuotaDeleteDto,
929    ) -> Pin<Box<dyn Future<Output = Result<Vec<ResourceOutcome>, UiError>> + Send + 'a>> {
930        Box::pin(async move {
931            let mut facade = self.0.facade().await?;
932            let error = facade
933                .client_mut()
934                .alter_user_quotas(
935                    &request.entity,
936                    &[QuotaOp::Remove {
937                        key: request.quota_type.clone(),
938                    }],
939                    false,
940                )
941                .await?;
942
943            Ok(vec![quota_mutation_outcome(
944                &request.entity,
945                &request.quota_type,
946                error,
947            )])
948        })
949    }
950
951    fn move_log_dir<'a>(
952        &'a self,
953        request: LogDirMoveRequestDto,
954    ) -> Pin<Box<dyn Future<Output = Result<Vec<ResourceOutcome>, UiError>> + Send + 'a>> {
955        Box::pin(async move {
956            let mut facade = self.0.facade().await?;
957            let assignments = BTreeMap::from([(
958                TopicPartitionReplica {
959                    topic: request.topic,
960                    partition: request.partition,
961                    broker_id: request.broker_id,
962                },
963                request.destination_log_dir,
964            )]);
965            let outcomes = facade
966                .client_mut()
967                .alter_replica_log_dirs(&assignments)
968                .await;
969
970            Ok(resource_outcome_rows(outcomes.into_iter().collect()))
971        })
972    }
973}
974
975fn ensure_valid_request(validation: Result<(), String>) -> Result<(), UiError> {
976    validation.map_err(UiError::InvalidRequest)
977}
978
979fn ensure_topic_config_resource(resource_type: &str, resource_name: &str) -> Result<(), UiError> {
980    if resource_type.eq_ignore_ascii_case("topic") {
981        return Ok(());
982    }
983
984    Err(UiError::InvalidRequest(format!(
985        "alter configs only supports topic resources through krabka-client-admin; {resource_type}:{resource_name} is unsupported"
986    )))
987}
988
989/// The ACL the request describes.
990///
991/// # Errors
992/// Returns [`UiError::InvalidRequest`] when a field names something Kafka does
993/// not define.
994pub fn acl_entry_from_request(request: &AclRequestDto) -> Result<AclEntry, UiError> {
995    Ok(AclEntry {
996        resource_type: parse_resource_type(&request.resource_type)?,
997        resource_name: request.resource_name.clone(),
998        pattern_type: parse_pattern_type(&request.pattern_type)?,
999        principal: request.principal.clone(),
1000        host: request.host.clone(),
1001        operation: parse_acl_operation(&request.operation)?,
1002        permission_type: parse_permission_type(&request.permission)?,
1003    })
1004}
1005
1006/// The delete filter the request describes.
1007///
1008/// Every axis is pinned, the pattern type included: `DeleteAcls` removes only
1009/// the entries a filter matches exactly, and a prefixed ACL does not match a
1010/// literal filter.
1011///
1012/// # Errors
1013/// Returns [`UiError::InvalidRequest`] when a field names something Kafka does
1014/// not define.
1015pub fn acl_filter_from_request(request: &AclRequestDto) -> Result<AclEntryFilter, UiError> {
1016    Ok(AclEntryFilter {
1017        resource_type: Some(parse_resource_type(&request.resource_type)?),
1018        resource_name: Some(request.resource_name.clone()),
1019        pattern_type: Some(parse_pattern_type(&request.pattern_type)?),
1020        principal: Some(request.principal.clone()),
1021        host: Some(request.host.clone()),
1022        operation: Some(parse_acl_operation(&request.operation)?),
1023        permission_type: Some(parse_permission_type(&request.permission)?),
1024    })
1025}
1026
1027fn parse_pattern_type(value: &str) -> Result<PatternType, UiError> {
1028    match value.to_ascii_lowercase().as_str() {
1029        "literal" => Ok(PatternType::Literal),
1030        "prefixed" => Ok(PatternType::Prefixed),
1031        _ => Err(UiError::InvalidRequest(format!(
1032            "unsupported ACL pattern type {value}"
1033        ))),
1034    }
1035}
1036
1037fn parse_resource_type(value: &str) -> Result<ResourceType, UiError> {
1038    match value.to_ascii_lowercase().as_str() {
1039        "topic" => Ok(ResourceType::Topic),
1040        "group" => Ok(ResourceType::Group),
1041        "cluster" => Ok(ResourceType::Cluster),
1042        "transactionalid" | "transactional_id" | "transactional-id" => {
1043            Ok(ResourceType::TransactionalId)
1044        }
1045        _ => Err(UiError::InvalidRequest(format!(
1046            "unsupported ACL resource type {value}"
1047        ))),
1048    }
1049}
1050
1051fn parse_acl_operation(value: &str) -> Result<AclOperation, UiError> {
1052    match value.to_ascii_lowercase().as_str() {
1053        "all" => Ok(AclOperation::All),
1054        "read" => Ok(AclOperation::Read),
1055        "write" => Ok(AclOperation::Write),
1056        "create" => Ok(AclOperation::Create),
1057        "delete" => Ok(AclOperation::Delete),
1058        "alter" => Ok(AclOperation::Alter),
1059        "describe" => Ok(AclOperation::Describe),
1060        "clusteraction" | "cluster_action" | "cluster-action" => Ok(AclOperation::ClusterAction),
1061        "describeconfigs" | "describe_configs" | "describe-configs" => {
1062            Ok(AclOperation::DescribeConfigs)
1063        }
1064        "alterconfigs" | "alter_configs" | "alter-configs" => Ok(AclOperation::AlterConfigs),
1065        "idempotentwrite" | "idempotent_write" | "idempotent-write" => {
1066            Ok(AclOperation::IdempotentWrite)
1067        }
1068        "twophasecommit" | "two_phase_commit" | "two-phase-commit" => {
1069            Ok(AclOperation::TwoPhaseCommit)
1070        }
1071        _ => Err(UiError::InvalidRequest(format!(
1072            "unsupported ACL operation {value}"
1073        ))),
1074    }
1075}
1076
1077fn parse_permission_type(value: &str) -> Result<PermissionType, UiError> {
1078    match value.to_ascii_lowercase().as_str() {
1079        "allow" => Ok(PermissionType::Allow),
1080        "deny" => Ok(PermissionType::Deny),
1081        _ => Err(UiError::InvalidRequest(format!(
1082            "unsupported ACL permission {value}"
1083        ))),
1084    }
1085}
1086
1087fn require_session_id(
1088    sessions: &SessionStore,
1089    raw_session_id: Option<&str>,
1090) -> Result<SessionId, UiError> {
1091    let Some(raw_session_id) = raw_session_id else {
1092        return Err(UiError::NotAuthenticated);
1093    };
1094    let Ok(session_id) = SessionId::try_from(raw_session_id) else {
1095        return Err(UiError::NotAuthenticated);
1096    };
1097
1098    if sessions.get(&session_id).is_none() {
1099        return Err(UiError::NotAuthenticated);
1100    }
1101
1102    Ok(session_id)
1103}
1104
1105fn require_session(
1106    sessions: &SessionStore,
1107    raw_session_id: Option<&str>,
1108) -> Result<SessionRecord, UiError> {
1109    let Some(raw_session_id) = raw_session_id else {
1110        return Err(UiError::NotAuthenticated);
1111    };
1112    let Ok(session_id) = SessionId::try_from(raw_session_id) else {
1113        return Err(UiError::NotAuthenticated);
1114    };
1115
1116    sessions.get(&session_id).ok_or(UiError::NotAuthenticated)
1117}
1118
1119fn read_seam_from_context<'a, F: AdminSeamFactory>(
1120    context: &'a ServerFunctionContext<'_, F>,
1121) -> Result<F::Reader<'a>, UiError> {
1122    let record = require_session(context.sessions, context.raw_session_id.as_deref())?;
1123
1124    context.seam_factory.read_seam(context.cfg, &record)
1125}
1126
1127fn mutation_seam_after_valid_request<'a, F: AdminSeamFactory>(
1128    context: &'a ServerFunctionContext<'_, F>,
1129    validation: Result<(), String>,
1130) -> Result<F::Mutations<'a>, UiError> {
1131    let record = require_session(context.sessions, context.raw_session_id.as_deref())?;
1132    ensure_valid_request(validation)?;
1133
1134    context.seam_factory.mutation_seam(context.cfg, &record)
1135}
krabka_admin_ui/server_fns.rs

krabka_admin_ui/
server_fns.rs
