cat_gateway/rbac/
get_chain.rs

1//! Utilities for obtaining a RBAC registration chain (`RegistrationChain`).
2
3use anyhow::{bail, Context, Result};
4use cardano_chain_follower::{ChainFollower, Network, Point, Slot, StakeAddress, TxnIndex};
5use catalyst_types::catalyst_id::CatalystId;
6use futures::{future::try_join, TryFutureExt, TryStreamExt};
7use rbac_registration::{cardano::cip509::Cip509, registration::cardano::RegistrationChain};
8
9use crate::{
10    db::index::{
11        queries::rbac::{
12            get_catalyst_id_from_stake_address::Query as CatalystIdQuery,
13            get_rbac_registrations::{Query as RbacQuery, QueryParams as RbacQueryParams},
14        },
15        session::CassandraSession,
16    },
17    rbac::{
18        chains_cache::{cache_persistent_rbac_chain, cached_persistent_rbac_chain},
19        ChainInfo,
20    },
21    settings::Settings,
22};
23
24/// Returns the latest (including the volatile part) registration chain by the given
25/// Catalyst ID.
26pub async fn latest_rbac_chain(id: &CatalystId) -> Result<Option<ChainInfo>> {
27    let id = id.as_short_id();
28
29    let volatile_session =
30        CassandraSession::get(false).context("Failed to get volatile Cassandra session")?;
31    // Get the persistent part of the chain and volatile registrations. Both of these parts
32    // can be non-existing.
33    let (chain, volatile_regs) = try_join(
34        persistent_rbac_chain(&id),
35        indexed_regs(&volatile_session, &id),
36    )
37    .await?;
38
39    let mut last_persistent_txn = None;
40    let mut last_persistent_slot = 0.into();
41
42    // Either update the persistent chain or build a new one.
43    let chain = match chain {
44        Some(c) => {
45            last_persistent_txn = Some(c.current_tx_id_hash());
46            last_persistent_slot = c.current_point().slot_or_default();
47            Some(apply_regs(c, volatile_regs).await?)
48        },
49        None => build_rbac_chain(volatile_regs).await?,
50    };
51
52    Ok(chain.map(|chain| {
53        let last_txn = Some(chain.current_tx_id_hash());
54        // If the last persistent transaction ID is the same as the last one, then there are no
55        // volatile registrations in this chain.
56        let last_volatile_txn = if last_persistent_txn == last_txn {
57            None
58        } else {
59            last_txn
60        };
61
62        ChainInfo {
63            chain,
64            last_persistent_txn,
65            last_volatile_txn,
66            last_persistent_slot,
67        }
68    }))
69}
70
71/// Returns the latest (including the volatile part) registration chain by the given stake
72/// address.
73pub async fn latest_rbac_chain_by_address(address: &StakeAddress) -> Result<Option<ChainInfo>> {
74    let persistent_session =
75        CassandraSession::get(true).context("Failed to get persistent Cassandra session")?;
76    let volatile_session =
77        CassandraSession::get(false).context("Failed to get volatile Cassandra session")?;
78
79    // We always check the latest (volatile) data first.
80    let id = match CatalystIdQuery::latest(&volatile_session, address).await? {
81        Some(id) => id,
82        None => {
83            match CatalystIdQuery::latest(&persistent_session, address).await? {
84                Some(id) => id,
85                None => return Ok(None),
86            }
87        },
88    };
89
90    latest_rbac_chain(&id).await
91}
92
93/// Returns only the persistent part of a registration chain by the given Catalyst ID.
94pub async fn persistent_rbac_chain(id: &CatalystId) -> Result<Option<RegistrationChain>> {
95    let session = CassandraSession::get(true).context("Failed to get Cassandra session")?;
96
97    let id = id.as_short_id();
98
99    if let Some(chain) = cached_persistent_rbac_chain(&session, &id) {
100        return Ok(Some(chain));
101    }
102
103    let regs = indexed_regs(&session, &id).await?;
104    let chain = build_rbac_chain(regs).await?.inspect(|c| {
105        cache_persistent_rbac_chain(id.clone(), c.clone());
106    });
107    Ok(chain)
108}
109
110/// Queries indexed RBAC registrations from the database.
111async fn indexed_regs(session: &CassandraSession, id: &CatalystId) -> Result<Vec<RbacQuery>> {
112    RbacQuery::execute(session, RbacQueryParams {
113        catalyst_id: id.clone().into(),
114    })
115    .and_then(|r| r.try_collect().map_err(Into::into))
116    .await
117}
118
119/// Builds a chain from the given registrations.
120pub async fn build_rbac_chain(
121    regs: impl IntoIterator<Item = RbacQuery>,
122) -> Result<Option<RegistrationChain>> {
123    let mut regs = regs.into_iter();
124    let Some(root) = regs.next() else {
125        return Ok(None);
126    };
127    if !root.removed_stake_addresses.is_empty() {
128        // This set contains addresses that were removed from the chain. It is impossible to
129        // remove an address before the chain was even started.
130        bail!("The root registration shouldn't contain removed stake addresses");
131    }
132    let root = cip509(
133        Settings::cardano_network(),
134        root.slot_no.into(),
135        root.txn_index.into(),
136    )
137    .await?;
138
139    let chain = RegistrationChain::new(root).context("Failed to start registration chain")?;
140    let chain = apply_regs(chain, regs).await?;
141    Ok(Some(chain))
142}
143
144/// Applies the given registration to the given chain.
145pub async fn apply_regs(
146    mut chain: RegistrationChain, regs: impl IntoIterator<Item = RbacQuery>,
147) -> Result<RegistrationChain> {
148    let network = Settings::cardano_network();
149
150    for reg in regs {
151        if !reg.removed_stake_addresses.is_empty() {
152            // TODO: This should be handled as a part of the
153            // https://github.com/input-output-hk/catalyst-voices/issues/2599 task.
154            continue;
155        }
156        let reg = cip509(network, reg.slot_no.into(), reg.txn_index.into()).await?;
157        chain = chain
158            .update(reg)
159            .context("Failed to update registration chain")?;
160    }
161
162    Ok(chain)
163}
164
165/// Loads and parses a `Cip509` registration from a block using chain follower.
166async fn cip509(network: Network, slot: Slot, txn_index: TxnIndex) -> Result<Cip509> {
167    let point = Point::fuzzy(slot);
168    let block = ChainFollower::get_block(network, point)
169        .await
170        .context("Unable to get block")?
171        .data;
172    if block.point().slot_or_default() != slot {
173        // The `ChainFollower::get_block` function can return the next consecutive block if it
174        // cannot find the exact one. This shouldn't happen, but we need to check anyway.
175        bail!(
176            "Unable to find exact {slot:?} block. Found block slot {:?}",
177            block.point().slot_or_default()
178        );
179    }
180    // We perform validation during indexing, so this normally should never fail.
181    Cip509::new(&block, txn_index, &[])
182        .with_context(|| {
183            format!("Invalid RBAC registration, slot = {slot:?}, transaction index = {txn_index:?}")
184        })?
185        .with_context(|| {
186            format!("No RBAC registration, slot = {slot:?}, transaction index = {txn_index:?}")
187        })
188}
cat_gateway/rbac/get_chain.rs

cat_gateway/rbac/
get_chain.rs
