uqa_storage/notifications/publication/
subscription.rs

1//
2// Unified Query Algebra
3//
4// Copyright (c) 2023-2026 Cognica, Inc.
5//
6
7//! Committed subscription changes share the original publication identity and admission.
8
9use super::{
10    hex_digit, invalid, write_number, BudgetedVec, Cursor, NotificationListenerRow,
11    StorageReadControl, VersionResult, MAX_NOTIFICATION_CHANNEL_BYTES,
12};
13
14#[derive(Clone, Copy)]
15pub(super) struct Subscription {
16    owner_id: [u8; 16],
17    session_id: u64,
18    wake_port: u16,
19    next_sequence: u64,
20    position: u64,
21    channels_offset: usize,
22    channel_count: u64,
23}
24
25impl Subscription {
26    pub(super) fn view(self, bytes: &[u8]) -> NotificationSubscriptionView<'_> {
27        NotificationSubscriptionView {
28            owner_id: self.owner_id,
29            session_id: self.session_id,
30            wake_port: self.wake_port,
31            next_sequence: self.next_sequence,
32            position: self.position,
33            channel_bytes: &bytes[self.channels_offset..],
34            channel_count: self.channel_count,
35        }
36    }
37}
38
39/// An empty channel list removes the listener. Other changes end its SQL transaction and preserve the committed queue cursor captured before publication.
40#[derive(Clone, Copy)]
41pub struct NotificationSubscriptionView<'a> {
42    pub owner_id: [u8; 16],
43    pub session_id: u64,
44    pub wake_port: u16,
45    pub next_sequence: u64,
46    pub position: u64,
47    channel_bytes: &'a [u8],
48    channel_count: u64,
49}
50
51impl<'a> NotificationSubscriptionView<'a> {
52    pub const fn is_unlisten(&self) -> bool {
53        self.channel_count == 0
54    }
55
56    /// Reuse the common controlled JSON writer for registries that store a channel array. Iteration borrows the original record; only the encoded JSON buffer is allocated.
57    pub fn channels_json(&self, control: &StorageReadControl) -> VersionResult<BudgetedVec<u8>> {
58        struct Channels<'a>(NotificationSubscriptionView<'a>);
59        impl serde::Serialize for Channels<'_> {
60            fn serialize<S: serde::Serializer>(&self, serializer: S) -> Result<S::Ok, S::Error> {
61                use serde::ser::SerializeSeq;
62                let mut sequence = serializer.serialize_seq(None)?;
63                for channel in self.0.channels() {
64                    sequence.serialize_element(channel.map_err(serde::ser::Error::custom)?)?;
65                }
66                sequence.end()
67            }
68        }
69        crate::key_value::record_json::encode(&Channels(*self), control)
70    }
71
72    pub fn channels(&self) -> impl Iterator<Item = VersionResult<&'a str>> + 'a {
73        let mut cursor = Cursor {
74            rest: self.channel_bytes,
75        };
76        let mut remaining = self.channel_count;
77        std::iter::from_fn(move || {
78            if remaining == 0 {
79                return None;
80            }
81            remaining -= 1;
82            Some(cursor.text(MAX_NOTIFICATION_CHANNEL_BYTES))
83        })
84    }
85}
86
87pub(super) fn encode(
88    bytes: &mut BudgetedVec<u8>,
89    listener: Option<&NotificationListenerRow>,
90    process_id: i32,
91    control: &StorageReadControl,
92) -> VersionResult<()> {
93    let Some(listener) = listener else {
94        bytes.extend_from_slice(b"\n0\n")?;
95        return Ok(());
96    };
97    if listener.owner_id == [0; 16]
98        || listener.process_id != process_id
99        || listener.wake_port == 0
100        || listener.transaction_open
101        || i64::try_from(listener.next_sequence).is_err()
102        || i64::try_from(listener.position).is_err()
103    {
104        return Err(invalid());
105    }
106    bytes.extend_from_slice(b"\n1\n")?;
107    for byte in listener.owner_id {
108        bytes.push(b"0123456789abcdef"[usize::from(byte >> 4)])?;
109        bytes.push(b"0123456789abcdef"[usize::from(byte & 15)])?;
110    }
111    bytes.push(b'\n')?;
112    for value in [
113        listener.session_id,
114        u64::from(listener.wake_port),
115        listener.next_sequence,
116        listener.position,
117        u64::try_from(listener.channels.len()).map_err(|_| invalid())?,
118    ] {
119        write_number(bytes, value, b'\n')?;
120    }
121    for channel in &listener.channels {
122        control.cancellation().check()?;
123        if channel.is_empty() || channel.len() >= MAX_NOTIFICATION_CHANNEL_BYTES {
124            return Err(invalid());
125        }
126        write_number(
127            bytes,
128            u64::try_from(channel.len()).map_err(|_| invalid())?,
129            b':',
130        )?;
131        bytes.extend_from_slice(channel.as_bytes())?;
132    }
133    Ok(())
134}
135
136pub(super) fn decode(
137    cursor: &mut Cursor<'_>,
138    record_len: usize,
139    control: &StorageReadControl,
140) -> VersionResult<Option<Subscription>> {
141    if cursor.take(1)? != b"\n" {
142        return Err(invalid());
143    }
144    match cursor.number(b'\n')? {
145        0 => return Ok(None),
146        1 => {}
147        _ => return Err(invalid()),
148    }
149    let mut owner_id = [0; 16];
150    for byte in &mut owner_id {
151        let hex = cursor.take(2)?;
152        *byte = (hex_digit(hex[0])? << 4) | hex_digit(hex[1])?;
153    }
154    if cursor.take(1)? != b"\n" || owner_id == [0; 16] {
155        return Err(invalid());
156    }
157    let session_id = cursor.number(b'\n')?;
158    let wake_port = u16::try_from(cursor.number(b'\n')?).map_err(|_| invalid())?;
159    let next_sequence = cursor.number(b'\n')?;
160    let position = cursor.number(b'\n')?;
161    let channel_count = cursor.number(b'\n')?;
162    if wake_port == 0
163        || i64::try_from(next_sequence).is_err()
164        || i64::try_from(position).is_err()
165        || channel_count > u64::try_from(cursor.rest.len() / 3).map_err(|_| invalid())?
166    {
167        return Err(invalid());
168    }
169    let channels_offset = record_len - cursor.rest.len();
170    for _ in 0..channel_count {
171        control.cancellation().check()?;
172        if cursor.text(MAX_NOTIFICATION_CHANNEL_BYTES)?.is_empty() {
173            return Err(invalid());
174        }
175    }
176    Ok(Some(Subscription {
177        owner_id,
178        session_id,
179        wake_port,
180        next_sequence,
181        position,
182        channels_offset,
183        channel_count,
184    }))
185}
uqa_storage/notifications/publication/subscription.rs

uqa_storage/notifications/publication/
subscription.rs
