rusticity_core/
sns.rs

1//! Read-only SNS Topics inventory. ListTopics supplies all three columns;
2//! no per-topic GetTopicAttributes requests are needed.
3use crate::config::AwsConfig;
4use crate::pagination::{Pagination, StopReason};
5use anyhow::{bail, Context, Result};
6use std::time::Duration;
7
8const LIST_TIMEOUT: Duration = Duration::from_secs(30);
9// Fail explicitly at the bound rather than presenting an incomplete inventory.
10const MAX_PAGES: usize = 1000;
11
12#[derive(Clone, Debug, Default, PartialEq, Eq)]
13pub struct SnsTopic {
14    pub name: String,
15    pub arn: String,
16    pub topic_type: String,
17}
18impl SnsTopic {
19    pub fn from_arn(arn: &str) -> Result<Self> {
20        let parts: Vec<_> = arn.splitn(6, ':').collect();
21        if parts.first() != Some(&"arn") || parts.get(2) != Some(&"sns") {
22            bail!("not an SNS topic ARN: {arn:?}");
23        }
24        if parts.len() != 6
25            || parts[1].is_empty()
26            || parts[3].is_empty()
27            || parts[4].len() != 12
28            || !parts[4].bytes().all(|c| c.is_ascii_digit())
29        {
30            bail!("malformed SNS topic ARN: {arn:?}");
31        }
32        let name = parts[5];
33        let fifo = name.ends_with(".fifo");
34        let base_name = name.strip_suffix(".fifo").unwrap_or(name);
35        if base_name.is_empty()
36            || name.len() > 256
37            || !base_name
38                .bytes()
39                .all(|c| c.is_ascii_alphanumeric() || matches!(c, b'_' | b'-'))
40        {
41            bail!("malformed SNS topic ARN (invalid topic name): {arn:?}");
42        }
43        Ok(Self {
44            name: name.into(),
45            arn: arn.into(),
46            topic_type: if fifo { "FIFO" } else { "Standard" }.into(),
47        })
48    }
49}
50
51pub struct SnsClient {
52    config: AwsConfig,
53}
54impl SnsClient {
55    pub fn new(config: AwsConfig) -> Self {
56        Self { config }
57    }
58    pub async fn list_topics(&self) -> Result<Vec<SnsTopic>> {
59        tokio::time::timeout(LIST_TIMEOUT, self.list_topics_inner(MAX_PAGES))
60            .await
61            .map_err(|_| anyhow::anyhow!("SNS ListTopics timed out after {LIST_TIMEOUT:?}"))?
62    }
63    async fn list_topics_inner(&self, max_pages: usize) -> Result<Vec<SnsTopic>> {
64        let client = self.config.sns_client();
65        let mut topics = Vec::new();
66        let mut token = None;
67        let mut pagination = Pagination::new(max_pages);
68        for _ in 0..max_pages {
69            // ListTopics has a fixed page size; there is no maxResults parameter.
70            let page = client
71                .list_topics()
72                .set_next_token(token)
73                .send()
74                .await
75                .context("SNS ListTopics request failed")?;
76            for topic in page.topics() {
77                let arn = topic
78                    .topic_arn()
79                    .filter(|arn| !arn.is_empty())
80                    .context("SNS ListTopics returned a topic with no ARN")?;
81                topics.push(SnsTopic::from_arn(arn)?);
82            }
83            match pagination.advance(page.next_token) {
84                Ok(Some(next)) => token = Some(next),
85                Ok(None) => {
86                    topics.sort_by(|a, b| a.name.cmp(&b.name).then_with(|| a.arn.cmp(&b.arn)));
87                    return Ok(topics);
88                }
89                Err(StopReason::RepeatedToken) => {
90                    bail!("SNS ListTopics returned a repeated pagination token; aborting");
91                }
92                Err(StopReason::PageLimit | StopReason::ResultLimit) => break,
93            }
94        }
95        bail!("SNS ListTopics exceeded {max_pages} pages; inventory is incomplete")
96    }
97}
98#[cfg(test)]
99mod tests;
rusticity_core/sns.rs

rusticity_core/
sns.rs
