1#![cfg_attr(
12 not(test),
13 deny(
14 clippy::unwrap_used,
15 clippy::unreachable,
16 clippy::expect_used,
17 clippy::panic,
18 clippy::unimplemented,
19 clippy::todo
20 )
21)]
22
23use std::future::Future;
24
25#[cfg(feature = "db-writer")]
26mod db_writer;
27
28use aster_forge_runtime::{
29 RuntimeComponentBundleRegistration, RuntimeComponentKind, RuntimeComponentRegistry,
30 StartupPhaseFailurePolicy, runtime_component,
31};
32
33#[cfg(feature = "db-writer")]
34pub use db_writer::{
35 AuditLogBufferConfig, AuditLogManager, DEFAULT_AUDIT_LOG_BATCH_SIZE,
36 DEFAULT_AUDIT_LOG_DELAYED_FLUSH_AFTER, DEFAULT_AUDIT_LOG_QUEUE_CAPACITY,
37 flush_global_audit_log_manager, global_audit_log_manager, init_global_audit_log_manager,
38 record_audit_log, shutdown_global_audit_log_manager, write_audit_log_direct,
39};
40
41pub const AUDIT_LOGS_COMPONENT: &str = "audit_logs";
43pub const AUDIT_MANAGER_COMPONENT: &str = "audit_manager";
45pub const SERVER_START_AUDIT_PHASE: &str = "server_start_audit";
47pub const SERVER_SHUTDOWN_AUDIT_PHASE: &str = "server_shutdown_audit";
49pub const AUDIT_MANAGER_FLUSH_SHUTDOWN_PHASE: &str = "audit_manager_flush";
51
52#[cfg(feature = "mail-outbox-dependency")]
53const DEFAULT_SERVER_SHUTDOWN_AUDIT_DEPENDENCIES: &[&str] =
54 &[aster_forge_mail::MAIL_OUTBOX_COMPONENT];
55#[cfg(not(feature = "mail-outbox-dependency"))]
56const DEFAULT_SERVER_SHUTDOWN_AUDIT_DEPENDENCIES: &[&str] = &[];
57
58pub fn audit_component<T, StartFn, StartFut, ShutdownFn, ShutdownFut, FlushFn, FlushFut>(
67 resources: T,
68 record_server_start: StartFn,
69 record_server_shutdown: ShutdownFn,
70 flush_audit_manager: FlushFn,
71) -> RuntimeComponentBundleRegistration<impl aster_forge_runtime::RuntimeComponentBundle>
72where
73 T: Clone + Send + 'static,
74 StartFn: FnOnce(T) -> StartFut + Send + Sync + 'static,
75 StartFut: Future<Output = Result<(), String>> + Send + 'static,
76 ShutdownFn: FnOnce(T) -> ShutdownFut + Send + Sync + 'static,
77 ShutdownFut: Future<Output = Result<(), String>> + Send + 'static,
78 FlushFn: FnOnce(()) -> FlushFut + Send + Sync + 'static,
79 FlushFut: Future<Output = Result<(), String>> + Send + 'static,
80{
81 audit_component_after(
82 resources,
83 DEFAULT_SERVER_SHUTDOWN_AUDIT_DEPENDENCIES,
84 record_server_start,
85 record_server_shutdown,
86 flush_audit_manager,
87 )
88}
89
90pub fn audit_component_infallible<
97 T,
98 StartFn,
99 StartFut,
100 ShutdownFn,
101 ShutdownFut,
102 FlushFn,
103 FlushFut,
104>(
105 resources: T,
106 record_server_start: StartFn,
107 record_server_shutdown: ShutdownFn,
108 flush_audit_manager: FlushFn,
109) -> RuntimeComponentBundleRegistration<impl aster_forge_runtime::RuntimeComponentBundle>
110where
111 T: Clone + Send + 'static,
112 StartFn: FnOnce(T) -> StartFut + Send + Sync + 'static,
113 StartFut: Future<Output = ()> + Send + 'static,
114 ShutdownFn: FnOnce(T) -> ShutdownFut + Send + Sync + 'static,
115 ShutdownFut: Future<Output = ()> + Send + 'static,
116 FlushFn: FnOnce(()) -> FlushFut + Send + Sync + 'static,
117 FlushFut: Future<Output = ()> + Send + 'static,
118{
119 audit_component_after_infallible(
120 resources,
121 DEFAULT_SERVER_SHUTDOWN_AUDIT_DEPENDENCIES,
122 record_server_start,
123 record_server_shutdown,
124 flush_audit_manager,
125 )
126}
127
128pub fn audit_component_after<T, StartFn, StartFut, ShutdownFn, ShutdownFut, FlushFn, FlushFut>(
135 resources: T,
136 shutdown_dependencies: &'static [&'static str],
137 record_server_start: StartFn,
138 record_server_shutdown: ShutdownFn,
139 flush_audit_manager: FlushFn,
140) -> RuntimeComponentBundleRegistration<impl aster_forge_runtime::RuntimeComponentBundle>
141where
142 T: Clone + Send + 'static,
143 StartFn: FnOnce(T) -> StartFut + Send + Sync + 'static,
144 StartFut: Future<Output = Result<(), String>> + Send + 'static,
145 ShutdownFn: FnOnce(T) -> ShutdownFut + Send + Sync + 'static,
146 ShutdownFut: Future<Output = Result<(), String>> + Send + 'static,
147 FlushFn: FnOnce(()) -> FlushFut + Send + Sync + 'static,
148 FlushFut: Future<Output = Result<(), String>> + Send + 'static,
149{
150 let startup_resources = resources.clone();
151 let server_start = server_start_audit_component(startup_resources, record_server_start);
152 let server_shutdown = server_shutdown_audit_component_after(
153 resources,
154 shutdown_dependencies,
155 record_server_shutdown,
156 );
157 let audit_manager = audit_manager_component(flush_audit_manager);
158
159 runtime_component(move |registry: &mut RuntimeComponentRegistry| {
160 registry
161 .register_bundle(server_start)
162 .register_bundle(server_shutdown)
163 .register_bundle(audit_manager);
164 })
165}
166
167pub fn audit_component_after_infallible<
173 T,
174 StartFn,
175 StartFut,
176 ShutdownFn,
177 ShutdownFut,
178 FlushFn,
179 FlushFut,
180>(
181 resources: T,
182 shutdown_dependencies: &'static [&'static str],
183 record_server_start: StartFn,
184 record_server_shutdown: ShutdownFn,
185 flush_audit_manager: FlushFn,
186) -> RuntimeComponentBundleRegistration<impl aster_forge_runtime::RuntimeComponentBundle>
187where
188 T: Clone + Send + 'static,
189 StartFn: FnOnce(T) -> StartFut + Send + Sync + 'static,
190 StartFut: Future<Output = ()> + Send + 'static,
191 ShutdownFn: FnOnce(T) -> ShutdownFut + Send + Sync + 'static,
192 ShutdownFut: Future<Output = ()> + Send + 'static,
193 FlushFn: FnOnce(()) -> FlushFut + Send + Sync + 'static,
194 FlushFut: Future<Output = ()> + Send + 'static,
195{
196 audit_component_after(
197 resources,
198 shutdown_dependencies,
199 move |resources| async move {
200 record_server_start(resources).await;
201 Ok(())
202 },
203 move |resources| async move {
204 record_server_shutdown(resources).await;
205 Ok(())
206 },
207 move |()| async move {
208 flush_audit_manager(()).await;
209 Ok(())
210 },
211 )
212}
213
214pub fn shutdown_audit_component<T, RecordFn, RecordFut, FlushFn, FlushFut>(
222 resources: T,
223 record_server_shutdown: RecordFn,
224 flush_audit_manager: FlushFn,
225) -> RuntimeComponentBundleRegistration<impl aster_forge_runtime::RuntimeComponentBundle>
226where
227 T: Send + 'static,
228 RecordFn: FnOnce(T) -> RecordFut + Send + Sync + 'static,
229 RecordFut: Future<Output = Result<(), String>> + Send + 'static,
230 FlushFn: FnOnce(()) -> FlushFut + Send + Sync + 'static,
231 FlushFut: Future<Output = Result<(), String>> + Send + 'static,
232{
233 shutdown_audit_component_after(
234 resources,
235 DEFAULT_SERVER_SHUTDOWN_AUDIT_DEPENDENCIES,
236 record_server_shutdown,
237 flush_audit_manager,
238 )
239}
240
241pub fn shutdown_audit_component_after<T, RecordFn, RecordFut, FlushFn, FlushFut>(
244 resources: T,
245 shutdown_dependencies: &'static [&'static str],
246 record_server_shutdown: RecordFn,
247 flush_audit_manager: FlushFn,
248) -> RuntimeComponentBundleRegistration<impl aster_forge_runtime::RuntimeComponentBundle>
249where
250 T: Send + 'static,
251 RecordFn: FnOnce(T) -> RecordFut + Send + Sync + 'static,
252 RecordFut: Future<Output = Result<(), String>> + Send + 'static,
253 FlushFn: FnOnce(()) -> FlushFut + Send + Sync + 'static,
254 FlushFut: Future<Output = Result<(), String>> + Send + 'static,
255{
256 let server_shutdown = server_shutdown_audit_component_after(
257 resources,
258 shutdown_dependencies,
259 record_server_shutdown,
260 );
261 let audit_manager = audit_manager_component(flush_audit_manager);
262
263 runtime_component(move |registry: &mut RuntimeComponentRegistry| {
264 registry
265 .register_bundle(server_shutdown)
266 .register_bundle(audit_manager);
267 })
268}
269
270pub fn server_start_audit_component<T, F, Fut>(
272 resources: T,
273 record_server_start: F,
274) -> RuntimeComponentBundleRegistration<impl aster_forge_runtime::RuntimeComponentBundle>
275where
276 T: Send + 'static,
277 F: FnOnce(T) -> Fut + Send + Sync + 'static,
278 Fut: Future<Output = Result<(), String>> + Send + 'static,
279{
280 runtime_component(move |registry: &mut RuntimeComponentRegistry| {
281 let mut resources = Some(resources);
282 let mut record_server_start = Some(record_server_start);
283 registry.component_startup(
284 AUDIT_LOGS_COMPONENT,
285 RuntimeComponentKind::Product,
286 SERVER_START_AUDIT_PHASE,
287 StartupPhaseFailurePolicy::Required,
288 move || {
289 let resources = resources.take();
290 let record_server_start = record_server_start.take();
291 async move {
292 let Some(resources) = resources else {
293 return Err(
294 "server start audit startup phase resources already consumed"
295 .to_string(),
296 );
297 };
298 let Some(record_server_start) = record_server_start else {
299 return Err("server start audit startup phase callback already consumed"
300 .to_string());
301 };
302 record_server_start(resources).await
303 }
304 },
305 );
306 })
307}
308
309pub fn server_shutdown_audit_component<T, F, Fut>(
315 resources: T,
316 record_server_shutdown: F,
317) -> RuntimeComponentBundleRegistration<aster_forge_runtime::ShutdownResourceComponent<T>>
318where
319 T: Send + 'static,
320 F: FnOnce(T) -> Fut + Send + Sync + 'static,
321 Fut: Future<Output = Result<(), String>> + Send + 'static,
322{
323 server_shutdown_audit_component_after(
324 resources,
325 DEFAULT_SERVER_SHUTDOWN_AUDIT_DEPENDENCIES,
326 record_server_shutdown,
327 )
328}
329
330pub fn server_shutdown_audit_component_after<T, F, Fut>(
332 resources: T,
333 shutdown_dependencies: &'static [&'static str],
334 record_server_shutdown: F,
335) -> RuntimeComponentBundleRegistration<aster_forge_runtime::ShutdownResourceComponent<T>>
336where
337 T: Send + 'static,
338 F: FnOnce(T) -> Fut + Send + Sync + 'static,
339 Fut: Future<Output = Result<(), String>> + Send + 'static,
340{
341 aster_forge_runtime::shutdown_resource_component_after(
342 AUDIT_LOGS_COMPONENT,
343 RuntimeComponentKind::Product,
344 SERVER_SHUTDOWN_AUDIT_PHASE,
345 shutdown_dependencies,
346 resources,
347 record_server_shutdown,
348 )
349}
350
351pub fn audit_manager_component<F, Fut>(
353 flush_audit_manager: F,
354) -> RuntimeComponentBundleRegistration<aster_forge_runtime::ShutdownResourceComponent<()>>
355where
356 F: FnOnce(()) -> Fut + Send + Sync + 'static,
357 Fut: Future<Output = Result<(), String>> + Send + 'static,
358{
359 aster_forge_runtime::shutdown_resource_component_after(
360 AUDIT_MANAGER_COMPONENT,
361 RuntimeComponentKind::Product,
362 AUDIT_MANAGER_FLUSH_SHUTDOWN_PHASE,
363 &[AUDIT_LOGS_COMPONENT],
364 (),
365 flush_audit_manager,
366 )
367}
368
369#[cfg(test)]
370mod tests {
371 use std::sync::{
372 Arc,
373 atomic::{AtomicUsize, Ordering},
374 };
375
376 use aster_forge_runtime::RuntimeComponentBundle;
377
378 use super::{
379 AUDIT_LOGS_COMPONENT, AUDIT_MANAGER_COMPONENT, AUDIT_MANAGER_FLUSH_SHUTDOWN_PHASE,
380 SERVER_SHUTDOWN_AUDIT_PHASE, SERVER_START_AUDIT_PHASE, audit_component,
381 audit_component_after, audit_component_after_infallible, audit_component_infallible,
382 audit_manager_component, server_shutdown_audit_component,
383 server_shutdown_audit_component_after, server_start_audit_component,
384 shutdown_audit_component,
385 };
386
387 #[cfg(not(feature = "mail-outbox-dependency"))]
388 #[test]
389 fn server_shutdown_audit_component_registers_without_dependencies() {
390 let registry = aster_forge_runtime::RuntimeComponentRegistry::configured(|registry| {
391 server_shutdown_audit_component((), |()| async { Ok(()) }).register(registry);
392 });
393
394 let descriptor = registry
395 .descriptor(AUDIT_LOGS_COMPONENT)
396 .expect("audit logs component should be registered");
397 assert_eq!(
398 descriptor.kind,
399 aster_forge_runtime::RuntimeComponentKind::Product
400 );
401 assert!(descriptor.dependencies.is_empty());
402 assert_eq!(
403 descriptor
404 .shutdown
405 .first()
406 .expect("audit logs shutdown should be registered")
407 .phase_name,
408 SERVER_SHUTDOWN_AUDIT_PHASE
409 );
410 }
411
412 #[cfg(feature = "mail-outbox-dependency")]
413 #[test]
414 fn server_shutdown_audit_component_registers_mail_outbox_dependency() {
415 let registry = aster_forge_runtime::RuntimeComponentRegistry::configured(|registry| {
416 server_shutdown_audit_component((), |()| async { Ok(()) }).register(registry);
417 });
418
419 let descriptor = registry
420 .descriptor(AUDIT_LOGS_COMPONENT)
421 .expect("audit logs component should be registered");
422 assert_eq!(
423 descriptor.kind,
424 aster_forge_runtime::RuntimeComponentKind::Product
425 );
426 assert_eq!(
427 descriptor.dependencies,
428 vec![aster_forge_mail::MAIL_OUTBOX_COMPONENT]
429 );
430 assert_eq!(
431 descriptor
432 .shutdown
433 .first()
434 .expect("audit logs shutdown should be registered")
435 .phase_name,
436 SERVER_SHUTDOWN_AUDIT_PHASE
437 );
438 }
439
440 #[test]
441 fn server_shutdown_audit_component_after_registers_caller_dependencies() {
442 let registry = aster_forge_runtime::RuntimeComponentRegistry::configured(|registry| {
443 server_shutdown_audit_component_after(
444 (),
445 &["background_tasks", "mail_outbox"],
446 |()| async { Ok(()) },
447 )
448 .register(registry);
449 });
450
451 let descriptor = registry
452 .descriptor(AUDIT_LOGS_COMPONENT)
453 .expect("audit logs component should be registered");
454 assert_eq!(
455 descriptor.dependencies,
456 vec!["background_tasks", "mail_outbox"]
457 );
458 }
459
460 #[tokio::test]
461 async fn server_start_audit_component_registers_and_runs_startup_phase() {
462 let calls = Arc::new(AtomicUsize::new(0));
463 let calls_for_start = calls.clone();
464 let mut registry = aster_forge_runtime::RuntimeComponentRegistry::configured(|registry| {
465 server_start_audit_component((), move |()| {
466 let calls = calls_for_start.clone();
467 async move {
468 calls.fetch_add(1, Ordering::SeqCst);
469 Ok(())
470 }
471 })
472 .register(registry);
473 });
474
475 let descriptor = registry
476 .descriptor(AUDIT_LOGS_COMPONENT)
477 .expect("audit logs component should be registered");
478 assert_eq!(descriptor.startup.len(), 1);
479 assert_eq!(descriptor.startup[0].phase_name, SERVER_START_AUDIT_PHASE);
480
481 let report = registry.startup().await;
482
483 assert!(!report.aborted());
484 assert_eq!(calls.load(Ordering::SeqCst), 1);
485 }
486
487 #[test]
488 fn audit_manager_component_registers_after_audit_logs() {
489 let registry = aster_forge_runtime::RuntimeComponentRegistry::configured(|registry| {
490 audit_manager_component(|()| async { Ok(()) }).register(registry);
491 });
492
493 let descriptor = registry
494 .descriptor(AUDIT_MANAGER_COMPONENT)
495 .expect("audit manager component should be registered");
496 assert_eq!(descriptor.dependencies, vec![AUDIT_LOGS_COMPONENT]);
497 assert_eq!(
498 descriptor
499 .shutdown
500 .first()
501 .expect("audit manager shutdown should be registered")
502 .phase_name,
503 AUDIT_MANAGER_FLUSH_SHUTDOWN_PHASE
504 );
505 }
506
507 #[tokio::test]
508 async fn audit_component_registers_startup_shutdown_and_manager_flush() {
509 let order = Arc::new(std::sync::Mutex::new(Vec::new()));
510 let order_for_start = order.clone();
511 let order_for_record = order.clone();
512 let order_for_flush = order.clone();
513
514 let mut registry = aster_forge_runtime::RuntimeComponentRegistry::configured(|registry| {
515 audit_component(
516 (),
517 move |()| {
518 let order = order_for_start.clone();
519 async move {
520 order
521 .lock()
522 .expect("audit component test order should lock")
523 .push("start");
524 Ok(())
525 }
526 },
527 move |()| {
528 let order = order_for_record.clone();
529 async move {
530 order
531 .lock()
532 .expect("audit component test order should lock")
533 .push("record");
534 Ok(())
535 }
536 },
537 move |()| {
538 let order = order_for_flush.clone();
539 async move {
540 order
541 .lock()
542 .expect("audit component test order should lock")
543 .push("flush");
544 Ok(())
545 }
546 },
547 )
548 .register(registry);
549 });
550
551 let startup_report = registry.startup().await;
552 assert!(!startup_report.aborted());
553 let shutdown_report = registry.shutdown().await;
554
555 assert!(!shutdown_report.has_failures());
556 assert_eq!(
557 order
558 .lock()
559 .expect("audit component test order should lock")
560 .as_slice(),
561 ["start", "record", "flush"]
562 );
563 }
564
565 #[tokio::test]
566 async fn infallible_audit_component_runs_unit_returning_hooks_in_order() {
567 let order = Arc::new(std::sync::Mutex::new(Vec::new()));
568 let order_for_start = order.clone();
569 let order_for_record = order.clone();
570 let order_for_flush = order.clone();
571
572 let mut registry = aster_forge_runtime::RuntimeComponentRegistry::configured(|registry| {
573 audit_component_infallible(
574 (),
575 move |()| {
576 let order = order_for_start.clone();
577 async move {
578 order
579 .lock()
580 .expect("audit component test order should lock")
581 .push("start");
582 }
583 },
584 move |()| {
585 let order = order_for_record.clone();
586 async move {
587 order
588 .lock()
589 .expect("audit component test order should lock")
590 .push("record");
591 }
592 },
593 move |()| {
594 let order = order_for_flush.clone();
595 async move {
596 order
597 .lock()
598 .expect("audit component test order should lock")
599 .push("flush");
600 }
601 },
602 )
603 .register(registry);
604 });
605
606 let startup_report = registry.startup().await;
607 assert!(!startup_report.aborted());
608 let shutdown_report = registry.shutdown().await;
609
610 assert!(!shutdown_report.has_failures());
611 assert_eq!(
612 order
613 .lock()
614 .expect("audit component test order should lock")
615 .as_slice(),
616 ["start", "record", "flush"]
617 );
618 }
619
620 #[test]
621 fn audit_component_after_registers_shutdown_dependencies() {
622 let registry = aster_forge_runtime::RuntimeComponentRegistry::configured(|registry| {
623 audit_component_after(
624 (),
625 &["mail_outbox"],
626 |()| async { Ok(()) },
627 |()| async { Ok(()) },
628 |()| async { Ok(()) },
629 )
630 .register(registry);
631 });
632
633 let descriptor = registry
634 .descriptor(AUDIT_LOGS_COMPONENT)
635 .expect("audit logs component should be registered");
636 assert_eq!(descriptor.dependencies, vec!["mail_outbox"]);
637 }
638
639 #[test]
640 fn infallible_audit_component_after_registers_shutdown_dependencies() {
641 let registry = aster_forge_runtime::RuntimeComponentRegistry::configured(|registry| {
642 audit_component_after_infallible(
643 (),
644 &["background_tasks"],
645 |()| async {},
646 |()| async {},
647 |()| async {},
648 )
649 .register(registry);
650 });
651
652 let descriptor = registry
653 .descriptor(AUDIT_LOGS_COMPONENT)
654 .expect("audit logs component should be registered");
655 assert_eq!(descriptor.dependencies, vec!["background_tasks"]);
656 }
657
658 #[cfg(feature = "mail-outbox-dependency")]
659 #[test]
660 fn audit_component_registers_mail_outbox_dependency_when_feature_enabled() {
661 let registry = aster_forge_runtime::RuntimeComponentRegistry::configured(|registry| {
662 audit_component(
663 (),
664 |()| async { Ok(()) },
665 |()| async { Ok(()) },
666 |()| async { Ok(()) },
667 )
668 .register(registry);
669 });
670
671 let descriptor = registry
672 .descriptor(AUDIT_LOGS_COMPONENT)
673 .expect("audit logs component should be registered");
674 assert_eq!(
675 descriptor.dependencies,
676 vec![aster_forge_mail::MAIL_OUTBOX_COMPONENT]
677 );
678 }
679
680 #[tokio::test]
681 async fn shutdown_audit_component_runs_shutdown_record_before_manager_flush() {
682 let order = Arc::new(std::sync::Mutex::new(Vec::new()));
683 let order_for_record = order.clone();
684 let order_for_flush = order.clone();
685
686 let mut registry = aster_forge_runtime::RuntimeComponentRegistry::new();
687 registry.register_bundle(shutdown_audit_component(
688 (),
689 move |()| {
690 let order = order_for_record.clone();
691 async move {
692 order
693 .lock()
694 .expect("audit component test order should lock")
695 .push("record");
696 Ok(())
697 }
698 },
699 move |()| {
700 let order = order_for_flush.clone();
701 async move {
702 order
703 .lock()
704 .expect("audit component test order should lock")
705 .push("flush");
706 Ok(())
707 }
708 },
709 ));
710 let report = registry.shutdown().await;
711
712 assert!(!report.has_failures());
713 assert_eq!(
714 order
715 .lock()
716 .expect("audit component test order should lock")
717 .as_slice(),
718 ["record", "flush"]
719 );
720 }
721}