photon-backend-fluvio 0.1.2

Fluvio storage adapter for Photon
Documentation 
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110

//! Fluvio topic naming and checkpoint key layout for Photon events.

use std::hash::{Hash, Hasher};

/// Topic prefix for all Photon topics (single-shard mode).
pub const TOPIC_PREFIX: &str = "photon";

/// Sharded-mode topic prefix (must not overlap legacy single-shard layout).
pub const SHARDED_TOPIC_PREFIX: &str = "photon-s";

/// Header carrying the Photon event UUID.
#[allow(dead_code)] // reserved for record metadata wiring
pub const HEADER_EVENT_ID: &str = "Photon-Event-Id";

/// Header carrying the per-partition monotonic sequence.
#[allow(dead_code)] // reserved for record metadata wiring
pub const HEADER_SEQ: &str = "Photon-Seq";

/// Header carrying the optional topic partition key.
#[allow(dead_code)] // reserved for record metadata wiring
pub const HEADER_TOPIC_KEY: &str = "Photon-Topic-Key";

/// Separator for checkpoint keys (Fluvio record keys).
const KV_SEP: &str = "/";

/// Max Fluvio topic name length (cluster limit).
const MAX_FLUVIO_TOPIC_LEN: usize = 63;

/// Sanitize a logical topic name for Fluvio topic naming rules.
///
/// Characters that previously collapsed onto `-` (`.`, `/`, `:`) are escaped so distinct
/// logical names cannot collide (`a.b` vs `a/b`). Escapes use `xNN` (not `~NN`) because Fluvio
/// topic names only allow alphanumeric characters and `-`.
#[must_use]
pub fn sanitize_fluvio_topic_name(topic_name: &str) -> String {
    let mut out = String::with_capacity(topic_name.len());
    for c in topic_name.chars() {
        match c {
            '.' => out.push_str("x2e"),
            '/' => out.push_str("x2f"),
            ':' => out.push_str("x3a"),
            'x' => out.push_str("x78"),
            c if c.is_ascii_lowercase() || c.is_ascii_digit() || c == '-' => out.push(c),
            c if c.is_ascii_uppercase() => out.push(c.to_ascii_lowercase()),
            _ => out.push('-'),
        }
    }
    out
}

pub fn truncate_or_hash(prefix: &str, topic_name: &str) -> String {
    let sanitized = sanitize_fluvio_topic_name(topic_name);
    let candidate = format!("{prefix}-{sanitized}");
    if candidate.len() <= MAX_FLUVIO_TOPIC_LEN {
        return candidate;
    }
    let mut hasher = std::collections::hash_map::DefaultHasher::new();
    topic_name.hash(&mut hasher);
    format!("{prefix}-{:016x}", hasher.finish())
}

/// Build the Fluvio topic for a logical topic (`photon-{topic_name}`).
#[must_use]
pub fn photon_topic(topic_name: &str) -> String {
    truncate_or_hash(TOPIC_PREFIX, topic_name)
}

/// Checkpoint key (logical `{sub}:{topic}:{key}`; wire uses `/` separators).
#[must_use]
pub fn checkpoint_key(sub: &str, topic: &str, topic_key: Option<&str>) -> String {
    format!(
        "{sub}{}{topic}{}{}",
        KV_SEP,
        KV_SEP,
        topic_key.unwrap_or("__null__")
    )
}

/// Per-shard checkpoint suffix for keyed subscriptions.
#[must_use]
pub fn checkpoint_key_sharded(
    sub: &str,
    topic: &str,
    topic_key: Option<&str>,
    topic_shard: u32,
) -> String {
    format!("{}/s{topic_shard}", checkpoint_key(sub, topic, topic_key))
}

/// Checkpoint key for unkeyed durable checkpoints across topic shards (JSON map).
#[must_use]
pub fn checkpoint_key_unkeyed_shards(sub: &str, topic: &str) -> String {
    format!("{sub}{KV_SEP}__shards__{topic}")
}

#[cfg(test)]
mod tests {
    use super::*;

    #[test]
    fn sanitizer_does_not_collide_dot_slash_colon() {
        let a = sanitize_fluvio_topic_name("a.b");
        let b = sanitize_fluvio_topic_name("a/b");
        let c = sanitize_fluvio_topic_name("a:b");
        assert_ne!(a, b);
        assert_ne!(a, c);
        assert_ne!(b, c);
        assert_eq!(photon_topic("a.b"), format!("photon-{a}"));
    }
}