signer-daemon 0.3.2

Signer daemon package.
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

use crate::{SignerDaemonCore, entity::crdt_event};
use sea_orm::{QueryOrder as _, prelude::*};

pub struct CrdtAdapterController {
    core: SignerDaemonCore,
}

impl CrdtAdapterController {
    pub fn new(core: SignerDaemonCore) -> Self {
        Self { core }
    }

    pub async fn apply_all(&mut self) -> crate::DaemonResult<()> {
        let anchor = crdt_event::Entity::find()
            .filter(crdt_event::Column::Revert.is_null())
            .order_by_asc(crdt_event::Column::Clock)
            .order_by_asc(crdt_event::Column::Peer)
            .one(&self.core.db)
            .await
            .map_err(|e| {
                crate::DaemonError::Signer(crate::SignerError::Msg(format!(
                    "find anchor failed: {}",
                    e
                )))
            })?;

        if let Some(anchor) = &anchor {
            let revert_vec = crdt_event::Entity::find()
                .filter(
                    crdt_event::Column::Revert.is_not_null().and(
                        crdt_event::Column::Clock
                            .gt(anchor.clock)
                            .or(crdt_event::Column::Clock
                                .eq(anchor.clock)
                                .and(crdt_event::Column::Peer.gt(&anchor.peer))),
                    ),
                )
                .order_by_desc(crdt_event::Column::Clock)
                .order_by_desc(crdt_event::Column::Peer)
                .all(&self.core.db)
                .await
                .map_err(|e| {
                    crate::DaemonError::Signer(crate::SignerError::Msg(format!(
                        "find revert events failed: {}",
                        e
                    )))
                })?;
            for revert in revert_vec {
                revert.revert(&self.core).await.map_err(|e| {
                    crate::DaemonError::Signer(crate::SignerError::Msg(format!(
                        "apply event {:?} failed: {}",
                        revert, e
                    )))
                })?;
            }
        }

        let apply_vec = match anchor {
            Some(anchor) => crdt_event::Entity::find()
                .filter(
                    crdt_event::Column::Clock
                        .gt(anchor.clock)
                        .or(crdt_event::Column::Clock
                            .eq(anchor.clock)
                            .and(crdt_event::Column::Peer.gte(anchor.peer))),
                )
                .order_by_asc(crdt_event::Column::Clock)
                .order_by_asc(crdt_event::Column::Peer)
                .all(&self.core.db)
                .await
                .map_err(|e| {
                    crate::DaemonError::Signer(crate::SignerError::Msg(format!(
                        "find apply events failed: {}",
                        e
                    )))
                })?,
            None => crdt_event::Entity::find()
                .order_by_asc(crdt_event::Column::Clock)
                .order_by_asc(crdt_event::Column::Peer)
                .all(&self.core.db)
                .await
                .map_err(|e| {
                    crate::DaemonError::Signer(crate::SignerError::Msg(format!(
                        "find apply events failed: {}",
                        e
                    )))
                })?,
        };

        for apply in apply_vec {
            apply.apply(&self.core).await.map_err(|e| {
                crate::DaemonError::Signer(crate::SignerError::Msg(format!(
                    "apply event {:?} failed: {}",
                    apply, e
                )))
            })?;
        }

        Ok(())
    }
}