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
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
use anyhow::anyhow;
use async_graphql::{
    connection::{
        query,
        Connection,
        CursorType,
        Edge,
        EmptyFields,
    },
    MergedObject,
    MergedSubscription,
    OutputType,
    Schema,
    SchemaBuilder,
};
use fuel_core_storage::{
    iter::IterDirection,
    Result as StorageResult,
};
use itertools::Itertools;

pub mod balance;
pub mod block;
pub mod chain;
pub mod coins;
pub mod contract;
pub mod dap;
pub mod health;
pub mod message;
pub mod node_info;

pub mod gas_price;
pub mod scalars;
pub mod tx;

pub mod relayed_tx;

#[derive(MergedObject, Default)]
pub struct Query(
    dap::DapQuery,
    balance::BalanceQuery,
    block::BlockQuery,
    chain::ChainQuery,
    tx::TxQuery,
    health::HealthQuery,
    coins::CoinQuery,
    contract::ContractQuery,
    contract::ContractBalanceQuery,
    node_info::NodeQuery,
    gas_price::LatestGasPriceQuery,
    gas_price::EstimateGasPriceQuery,
    message::MessageQuery,
    relayed_tx::RelayedTransactionQuery,
);

#[derive(MergedObject, Default)]
pub struct Mutation(dap::DapMutation, tx::TxMutation, block::BlockMutation);

#[derive(MergedSubscription, Default)]
pub struct Subscription(tx::TxStatusSubscription);

pub type CoreSchema = Schema<Query, Mutation, Subscription>;
pub type CoreSchemaBuilder = SchemaBuilder<Query, Mutation, Subscription>;

pub fn build_schema() -> CoreSchemaBuilder {
    Schema::build_with_ignore_name_conflicts(
        Query::default(),
        Mutation::default(),
        Subscription::default(),
        ["TransactionConnection", "MessageConnection"],
    )
}

async fn query_pagination<F, Entries, SchemaKey, SchemaValue>(
    after: Option<String>,
    before: Option<String>,
    first: Option<i32>,
    last: Option<i32>,
    entries: F,
) -> async_graphql::Result<Connection<SchemaKey, SchemaValue, EmptyFields, EmptyFields>>
where
    SchemaKey: CursorType + Send + Sync,
    <SchemaKey as CursorType>::Error: core::fmt::Display + Send + Sync + 'static,
    SchemaValue: OutputType,
    // TODO: Optimization: Support `count` here including skipping of entities.
    //  It means also returning `has_previous_page` and `has_next_page` values.
    // entries(start_key: Option<DBKey>)
    F: FnOnce(&Option<SchemaKey>, IterDirection) -> StorageResult<Entries>,
    Entries: Iterator<Item = StorageResult<(SchemaKey, SchemaValue)>>,
    SchemaKey: Eq,
{
    match (after.as_ref(), before.as_ref(), first, last) {
        (_, _, Some(first), Some(last)) => {
            return Err(anyhow!(
                "Either first `{first}` or latest `{last}` elements, not both"
            )
            .into())
        }
        (Some(after), _, _, Some(last)) => {
            return Err(anyhow!(
                "After `{after:?}` with last `{last}` elements is not supported"
            )
            .into())
        }
        (_, Some(before), Some(first), _) => {
            return Err(anyhow!(
                "Before `{before:?}` with first `{first}` elements is not supported"
            )
            .into())
        }
        (None, None, None, None) => {
            return Err(anyhow!("The queries for the whole range is not supported").into())
        }
        (_, _, _, _) => { /* Other combinations are allowed */ }
    };

    query(
        after,
        before,
        first,
        last,
        |after: Option<SchemaKey>, before: Option<SchemaKey>, first, last| async move {
            let (count, direction) = if let Some(first) = first {
                (first, IterDirection::Forward)
            } else if let Some(last) = last {
                (last, IterDirection::Reverse)
            } else {
                // Unreachable because of the check `(None, None, None, None)` above
                unreachable!()
            };

            let start;
            let end;

            if direction == IterDirection::Forward {
                start = after;
                end = before;
            } else {
                start = before;
                end = after;
            }

            let entries = entries(&start, direction)?;
            let mut has_previous_page = false;
            let mut has_next_page = false;

            // TODO: Add support of `skip` field for pages with huge list of entities with
            //  the same `SchemaKey`.
            let entries = entries.skip_while(|result| {
                if let Ok((key, _)) = result {
                    // TODO: `entries` should return information about `has_previous_page` for wild
                    //  queries
                    if let Some(start) = start.as_ref() {
                        // Skip until start + 1
                        if key == start {
                            has_previous_page = true;
                            return true
                        }
                    }
                }
                false
            });

            let mut count = count.saturating_add(1) /* for `has_next_page` */;
            let entries = entries.take(count).take_while(|result| {
                if let Ok((key, _)) = result {
                    if let Some(end) = end.as_ref() {
                        // take until we've reached the end
                        if key == end {
                            has_next_page = true;
                            return false
                        }
                    }
                    count = count.saturating_sub(1);
                    has_next_page |= count == 0;
                    count != 0
                } else {
                    // We want to stop immediately in the case of error
                    false
                }
            });

            let entries: Vec<_> = entries.try_collect()?;
            let entries = entries.into_iter();

            let mut connection = Connection::new(has_previous_page, has_next_page);

            connection.edges.extend(
                entries
                    .into_iter()
                    .map(|(key, value)| Edge::new(key, value)),
            );

            Ok::<Connection<SchemaKey, SchemaValue>, anyhow::Error>(connection)
        },
    )
    .await
}