keri_core/processor/escrow/
delegation_escrow.rs

1use std::{sync::Arc, time::Duration};
2
3use said::SelfAddressingIdentifier;
4
5use crate::{
6    actor::prelude::EventStorage,
7    database::{EscrowCreator, EscrowDatabase, EventDatabase},
8    error::Error,
9    event::{
10        event_data::EventData,
11        sections::seal::{EventSeal, Seal, SourceSeal},
12    },
13    event_message::signed_event_message::SignedEventMessage,
14    prefix::IdentifierPrefix,
15    processor::{
16        notification::{Notification, NotificationBus, Notifier},
17        validator::EventValidator,
18    },
19};
20
21/// Stores delegated events until delegating event is provided
22pub struct DelegationEscrow<D: EventDatabase + EscrowCreator> {
23    db: Arc<D>,
24    // Key of this escrow is (delegator's identifier, delegator's event sn if available).
25    pub delegation_escrow: D::EscrowDatabaseType,
26}
27
28impl<D: EventDatabase + EscrowCreator + 'static> DelegationEscrow<D> {
29    pub fn new(db: Arc<D>, _duration: Duration) -> Self {
30        let escrow_db = db.create_escrow_db("delegation_escrow");
31        Self {
32            db,
33            delegation_escrow: escrow_db,
34        }
35    }
36
37    pub fn get_event_by_sn_and_digest(
38        &self,
39        sn: u64,
40        delegator_id: &IdentifierPrefix,
41        event_digest: &SelfAddressingIdentifier,
42    ) -> Option<SignedEventMessage> {
43        self.delegation_escrow
44            .get(delegator_id, sn)
45            .ok()
46            .and_then(|mut events| {
47                events.find(|event| {
48                    event.event_message.data.sn == sn
49                        && event.event_message.digest().ok().as_ref() == Some(event_digest)
50                })
51            })
52    }
53
54    pub fn process_delegation_events(
55        &self,
56        bus: &NotificationBus,
57        delegator_id: &IdentifierPrefix,
58        anchored_seals: Vec<EventSeal>,
59        potential_delegator_seal: SourceSeal,
60    ) -> Result<(), Error> {
61        if let Ok(esc) = self.delegation_escrow.get_from_sn(delegator_id, 0) {
62            for event in esc {
63                let event_digest = event.event_message.digest()?;
64                let seal = anchored_seals.iter().find(|seal| {
65                    seal.event_digest() == event_digest
66                        && seal.sn == event.event_message.data.get_sn()
67                        && seal.prefix == event.event_message.data.get_prefix()
68                });
69                let delegated_event = match seal {
70                    Some(_s) => SignedEventMessage {
71                        delegator_seal: Some(potential_delegator_seal.clone()),
72                        ..event.clone()
73                    },
74                    None => event.clone(),
75                };
76                let validator = EventValidator::new(self.db.clone());
77                match validator.validate_event(&delegated_event) {
78                    Ok(_) => {
79                        // add to kel
80                        let child_id = event.event_message.data.get_prefix();
81                        self.db
82                            .add_kel_finalized_event(delegated_event.clone(), &child_id)
83                            .map_err(|_| Error::DbError)?;
84                        // remove from escrow
85                        self.delegation_escrow.remove(&event.event_message);
86                        bus.notify(&Notification::KeyEventAdded(event))?;
87                        // stop processing the escrow if kel was updated. It needs to start again.
88                        break;
89                    }
90                    Err(Error::SignatureVerificationError) => {
91                        // remove from escrow
92                        self.delegation_escrow.remove(&event.event_message);
93                    }
94                    Err(Error::NotEnoughReceiptsError) => {
95                        // remove from escrow
96                        self.delegation_escrow.remove(&event.event_message);
97                        bus.notify(&Notification::PartiallyWitnessed(delegated_event))?;
98                    }
99                    Err(_e) => (), // keep in escrow,
100                }
101            }
102        };
103
104        Ok(())
105    }
106}
107
108impl<D: EventDatabase + EscrowCreator + 'static> Notifier for DelegationEscrow<D> {
109    fn notify(&self, notification: &Notification, bus: &NotificationBus) -> Result<(), Error> {
110        match notification {
111            Notification::KeyEventAdded(ev_message) => {
112                // delegator's prefix
113                let id = ev_message.event_message.data.get_prefix();
114                // get anchored data
115                let anchored_data: Vec<Seal> = match &ev_message.event_message.data.event_data {
116                    EventData::Icp(icp) => icp.data.clone(),
117                    EventData::Rot(rot) => rot.data.clone(),
118                    EventData::Ixn(ixn) => ixn.data.clone(),
119                    EventData::Dip(dip) => dip.inception_data.data.clone(),
120                    EventData::Drt(drt) => drt.data.clone(),
121                };
122
123                let seals: Vec<EventSeal> = anchored_data
124                    .into_iter()
125                    .filter_map(|seal| match seal {
126                        Seal::Event(es) => Some(es),
127                        _ => None,
128                    })
129                    .collect();
130                if !seals.is_empty() {
131                    let potential_delegator_seal = SourceSeal::new(
132                        ev_message.event_message.data.get_sn(),
133                        ev_message.event_message.digest()?,
134                    );
135                    self.process_delegation_events(bus, &id, seals, potential_delegator_seal)?;
136                }
137            }
138            Notification::MissingDelegatingEvent(signed_event) => {
139                // ignore events with no signatures
140                if !signed_event.signatures.is_empty() {
141                    let delegator_id = match &signed_event.event_message.data.event_data {
142                        EventData::Dip(dip) => Ok(dip.delegator.clone()),
143                        EventData::Drt(_drt) => {
144                            let storage = EventStorage::new(self.db.clone());
145                            storage
146                                .get_state(&signed_event.event_message.data.get_prefix())
147                                .ok_or(Error::MissingDelegatingEventError)?
148                                .delegator
149                                .ok_or(Error::MissingDelegatingEventError)
150                        }
151                        _ => {
152                            // not delegated event
153                            Err(Error::SemanticError("Not delegated event".to_string()))
154                        }
155                    }?;
156                    let delegator_seal = signed_event.delegator_seal.clone();
157                    let sn = if let Some(delegator_seal) = delegator_seal {
158                        delegator_seal.sn
159                    } else {
160                        0
161                    };
162                    self.delegation_escrow
163                        .insert_key_value(&delegator_id, sn, signed_event)
164                        .map_err(|_| Error::DbError)?;
165                }
166            }
167            _ => return Err(Error::SemanticError("Wrong notification".into())),
168        }
169
170        Ok(())
171    }
172}
keri_core/processor/escrow/delegation_escrow.rs

keri_core/processor/escrow/
delegation_escrow.rs
