photon_runtime/admin/
snapshot.rs

1//! Compose [`AdminSnapshot`] from registries and storage checkpoint reads.
2
3use photon_backend::{
4    BackendCapabilities, HandlerDescriptor, HandlerRegistry, ShardConfig, TopicDescriptor,
5    shard_storage_key,
6};
7
8use crate::Photon;
9
10use super::types::{
11    AdminBackendSummary, AdminCheckpointSummary, AdminHandlerSummary, AdminSnapshot,
12    AdminTopicSummary,
13};
14
15/// Build a full admin snapshot for the given runtime (async checkpoint reads).
16///
17/// # Errors
18///
19/// Returns an error if a checkpoint load fails.
20pub async fn collect_admin_snapshot(photon: &Photon) -> photon_backend::Result<AdminSnapshot> {
21    let topics = collect_topics(photon);
22    let handlers = collect_handlers();
23    let backend = collect_backend(photon.backend_capabilities());
24    let checkpoints = collect_checkpoints(photon).await?;
25
26    Ok(AdminSnapshot {
27        backend,
28        topics,
29        handlers,
30        checkpoints,
31    })
32}
33
34fn collect_topics(photon: &Photon) -> Vec<AdminTopicSummary> {
35    photon
36        .registry()
37        .sorted_topic_names()
38        .into_iter()
39        .filter_map(|name| photon.registry().get(name).map(topic_from_descriptor))
40        .collect()
41}
42
43fn topic_from_descriptor(descriptor: &TopicDescriptor) -> AdminTopicSummary {
44    AdminTopicSummary {
45        topic_name: descriptor.topic_name.to_string(),
46        keyed_by: descriptor.keyed_by.map(str::to_string),
47        schema_json: descriptor
48            .schema_value()
49            .unwrap_or_else(|| serde_json::json!({})),
50    }
51}
52
53fn collect_handlers() -> Vec<AdminHandlerSummary> {
54    let registry = HandlerRegistry::auto_discover();
55    registry.iter().map(handler_from_descriptor).collect()
56}
57
58fn handler_from_descriptor(handler: &HandlerDescriptor) -> AdminHandlerSummary {
59    let is_group = handler.is_consumer_group();
60    AdminHandlerSummary {
61        topic_name: handler.topic_name.to_string(),
62        subscription_name: if is_group || handler.subscription_name.is_empty() {
63            None
64        } else {
65            Some(handler.subscription_name.to_string())
66        },
67        consumer_group: handler.consumer_group.map(str::to_string),
68        registry_key: handler.registry_key.to_string(),
69        mode: if is_group {
70            "consumer_group".to_string()
71        } else {
72            "durable".to_string()
73        },
74    }
75}
76
77fn collect_backend(caps: BackendCapabilities) -> AdminBackendSummary {
78    AdminBackendSummary {
79        telemetry_label: caps.telemetry_label.to_string(),
80        supports_get_event: caps.supports_get_event,
81        max_replay_window_secs: caps.max_replay_window.map(|d| d.as_secs()),
82    }
83}
84
85async fn collect_checkpoints(
86    photon: &Photon,
87) -> photon_backend::Result<Vec<AdminCheckpointSummary>> {
88    let registry = HandlerRegistry::auto_discover();
89    let mut out = Vec::new();
90
91    for handler in registry.iter() {
92        if handler.is_consumer_group() {
93            let group_id = handler.consumer_group.unwrap_or("");
94            let shard_count = group_shard_count(photon, handler);
95            for shard_id in 0..shard_count {
96                let shard_key = shard_storage_key(shard_id);
97                let last_seq = photon
98                    .get_checkpoint_seq(group_id, handler.topic_name, Some(&shard_key))
99                    .await?;
100                out.push(AdminCheckpointSummary {
101                    subscription_name: group_id.to_string(),
102                    topic_name: handler.topic_name.to_string(),
103                    topic_key: Some(shard_key),
104                    last_seq,
105                });
106            }
107        } else {
108            let last_seq = photon
109                .get_checkpoint_seq(handler.subscription_name, handler.topic_name, None)
110                .await?;
111            out.push(AdminCheckpointSummary {
112                subscription_name: handler.subscription_name.to_string(),
113                topic_name: handler.topic_name.to_string(),
114                topic_key: None,
115                last_seq,
116            });
117        }
118    }
119
120    Ok(out)
121}
122
123fn group_shard_count(photon: &Photon, handler: &'static HandlerDescriptor) -> u32 {
124    handler
125        .group_shard_count
126        .or_else(|| {
127            photon
128                .registry()
129                .get(handler.topic_name)
130                .and_then(|d| d.shard_config)
131                .map(|c| c.shard_count)
132        })
133        .unwrap_or(ShardConfig::DEFAULT_SHARD_COUNT)
134}
photon_runtime/admin/snapshot.rs

photon_runtime/admin/
snapshot.rs
