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
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151

use std::sync::Arc;

use anyhow::{anyhow, Result};
use tokio::sync::mpsc;

use crate::{
    core::{
        common::logger::get_logger,
        consensus::{
            beacon::EpochTime, state::beacon::ImmutableState as BeaconState, verifier::Verifier,
        },
    },
    modules::rofl::types::Register,
};

use super::{client::SubmitTxOpts, processor, App, Environment};

/// Registration task.
pub(super) struct Task<A: App> {
    imp: Option<Impl<A>>,
    tx: mpsc::Sender<()>,
}

impl<A> Task<A>
where
    A: App,
{
    /// Create a registration task.
    pub(super) fn new(state: Arc<processor::State<A>>, env: Environment<A>) -> Self {
        let (tx, rx) = mpsc::channel(1);

        let imp = Impl {
            state,
            env,
            logger: get_logger("modules/rofl/app/registration"),
            notify: rx,
            last_registration_epoch: None,
        };

        Self { imp: Some(imp), tx }
    }

    /// Start the registration task.
    pub(super) fn start(&mut self) {
        if let Some(imp) = self.imp.take() {
            imp.start();
        }
    }

    /// Ask the registration task to refresh the registration.
    pub(super) fn refresh(&self) {
        let _ = self.tx.try_send(());
    }
}

struct Impl<A: App> {
    state: Arc<processor::State<A>>,
    env: Environment<A>,
    logger: slog::Logger,

    notify: mpsc::Receiver<()>,
    last_registration_epoch: Option<EpochTime>,
}

impl<A> Impl<A>
where
    A: App,
{
    /// Start the registration task.
    pub(super) fn start(self) {
        tokio::task::spawn(self.run());
    }

    /// Run the registration task.
    async fn run(mut self) {
        slog::info!(self.logger, "starting registration task");

        // TODO: Handle retries etc.
        while self.notify.recv().await.is_some() {
            if let Err(err) = self.refresh_registration().await {
                slog::error!(self.logger, "failed to refresh registration";
                    "err" => ?err,
                );
            }
        }

        slog::info!(self.logger, "registration task stopped");
    }

    /// Perform application registration refresh.
    async fn refresh_registration(&mut self) -> Result<()> {
        // Determine current epoch.
        let state = self.state.consensus_verifier.latest_state().await?;
        let epoch = tokio::task::spawn_blocking(move || {
            let beacon = BeaconState::new(&state);
            beacon.epoch()
        })
        .await??;

        // Skip refresh in case epoch has not changed.
        if self.last_registration_epoch == Some(epoch) {
            return Ok(());
        }

        slog::info!(self.logger, "refreshing registration";
            "last_registration_epoch" => self.last_registration_epoch,
            "epoch" => epoch,
        );

        // Refresh registration.
        let ect = self
            .state
            .identity
            .endorsed_capability_tee()
            .ok_or(anyhow!("endorsed TEE capability not available"))?;
        let register = Register {
            app: A::id(),
            ect,
            expiration: epoch + 2,
            extra_keys: vec![self.env.signer().public_key()],
        };

        let tx = self.state.app.new_transaction("rofl.Register", register);
        let result = self
            .env
            .client()
            .multi_sign_and_submit_tx_opts(
                &[self.state.identity.clone(), self.env.signer()],
                tx,
                SubmitTxOpts {
                    encrypt: false, // Needed for initial fee payments.
                    ..Default::default()
                },
            )
            .await?
            .ok()?;

        slog::info!(self.logger, "refreshed registration"; "result" => ?result);

        if self.last_registration_epoch.is_none() {
            // If this is the first registration, notify processor that initial registration has
            // been completed so it can do other stuff.
            self.env
                .send_command(processor::Command::InitialRegistrationCompleted)
                .await?;
        }
        self.last_registration_epoch = Some(epoch);

        Ok(())
    }
}