rostrum 14.0.1

An efficient implementation of Electrum Server with token support
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
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122

use std::sync::Arc;

use crate::{
    chaindef::{OutPointHash, ScriptHash},
    query::{queryutil::outpoints_are_spent, BUFFER_SIZE, CHUNK_SIZE},
    store::{DBContents, DBStore},
};

use super::{queryfilter::QueryFilter, status::scan_for_outputs};

use anyhow::Result;
use futures::{stream, StreamExt};

#[derive(Serialize)]
pub struct ScriptHashVolume {
    confirmed_sats_in: u64,
    confirmed_sats_out: u64,
    mempool_sats_in: u64,
    mempool_sats_out: u64,
}

pub async fn historical_volume(
    confirmed: &Arc<DBStore>,
    mempool: &Arc<DBStore>,
    scripthash: ScriptHash,
    filter: &QueryFilter,
) -> Result<ScriptHashVolume> {
    assert!(confirmed.contents == DBContents::ConfirmedIndex);
    assert!(mempool.contents == DBContents::MempoolIndex);

    // Find in and out sats in confirmed index
    let (confirmed_in, confirmed_out, confirmed_unspent) = {
        let (outputs_query, outputs_stream) = scan_for_outputs(confirmed, scripthash, filter).await;
        let (c_in, c_out, unspent) = outputs_stream
            .map(|(o, _)| (o.value(), o.take_hash()))
            .ready_chunks(CHUNK_SIZE)
            .then(|chunk| {
                let (values, outpoints): (Vec<u64>, Vec<OutPointHash>) = chunk.into_iter().unzip();
                let db = Arc::clone(confirmed);
                async move {
                    let spent_flags = outpoints_are_spent(&db, &outpoints).await;
                    stream::iter(values.into_iter().zip(outpoints).zip(spent_flags).map(
                        |((val, op), spent)| {
                            if spent {
                                // This outpoint is spent in confirmed
                                // (returning in and out; as we want to know total going in)
                                (val, val, None)
                            } else {
                                // This outpoint is still unspent in confirmed
                                (val, 0, Some((op, val)))
                            }
                        },
                    ))
                }
            })
            .flatten_unordered(BUFFER_SIZE)
            .fold(
                (0u64, 0u64, Vec::new()),
                |(sum_in, sum_out, mut unspent), (vin, vout, maybe_op_val)| async move {
                    if let Some(op_val) = maybe_op_val {
                        unspent.push(op_val);
                    }
                    (sum_in + vin, sum_out + vout, unspent)
                },
            )
            .await;
        outputs_query.await??;
        (c_in, c_out, unspent)
    };

    // Find spends in mempool of confirmed utxos
    let mempool_out_from_confirmed = if confirmed_unspent.is_empty() {
        0u64
    } else {
        let (ops, amounts): (Vec<OutPointHash>, Vec<u64>) =
            confirmed_unspent.iter().cloned().unzip();
        let spent_flags = outpoints_are_spent(mempool, &ops).await;
        let mut total_spent_in_mempool = 0u64;
        for (val, spent) in amounts.into_iter().zip(spent_flags) {
            if spent {
                total_spent_in_mempool += val;
            }
        }
        total_spent_in_mempool
    };

    // Find in/out in mempool
    let (mempool_in, mempool_out) = {
        let (mp_query, mp_stream) = scan_for_outputs(mempool, scripthash, filter).await;
        let (m_in, m_out) = mp_stream
            .map(|(o, _)| (o.value(), o.take_hash()))
            .ready_chunks(CHUNK_SIZE)
            .then(|chunk| {
                let (values, outpoints): (Vec<u64>, Vec<OutPointHash>) = chunk.into_iter().unzip();
                let db = Arc::clone(mempool);
                async move {
                    let spent_flags = outpoints_are_spent(&db, &outpoints).await;
                    stream::iter(values.into_iter().zip(spent_flags).map(|(val, spent)| {
                        if spent {
                            (val, val)
                        } else {
                            (val, 0u64)
                        }
                    }))
                }
            })
            .flatten_unordered(BUFFER_SIZE)
            .fold((0u64, 0u64), |(sum_in, sum_out), (vin, vout)| async move {
                (sum_in + vin, sum_out + vout)
            })
            .await;
        mp_query.await??;
        (m_in, m_out)
    };

    Ok(ScriptHashVolume {
        confirmed_sats_in: confirmed_in,
        confirmed_sats_out: confirmed_out,
        mempool_sats_in: mempool_in,
        mempool_sats_out: mempool_out + mempool_out_from_confirmed,
    })
}