1
use std::{error::Error as StdError, sync::Arc};
2

            
3
use async_trait::async_trait;
4
use futures::TryStreamExt;
5
use mongodb::{
6
    Cursor as MongoDbCursor, Database,
7
    action::Find,
8
    bson::{Bson, DateTime, Document, Regex, doc, raw::CString},
9
};
10
use serde::{Deserialize, Serialize};
11

            
12
use super::super::client::{
13
    Client, ClientModel, Cursor, ListOptions, ListQueryCond, QueryCond, SortKey, UpdateQueryCond,
14
    Updates,
15
};
16

            
17
/// Model instance.
18
pub struct Model {
19
    /// The associated database connection.
20
    conn: Arc<Database>,
21
}
22

            
23
/// Cursor instance.
24
struct DbCursor {
25
    /// The associated collection cursor.
26
    cursor: MongoDbCursor<Schema>,
27
    /// (Useless) only for Cursor trait implementation.
28
    offset: u64,
29
}
30

            
31
/// MongoDB schema.
32
#[derive(Deserialize, Serialize)]
33
struct Schema {
34
    #[serde(rename = "clientId")]
35
    client_id: String,
36
    #[serde(rename = "createdAt")]
37
    created_at: DateTime,
38
    #[serde(rename = "modifiedAt")]
39
    modified_at: DateTime,
40
    #[serde(rename = "clientSecret")]
41
    client_secret: Option<String>,
42
    #[serde(rename = "redirectUris")]
43
    redirect_uris: Vec<String>,
44
    scopes: Vec<String>,
45
    #[serde(rename = "userId")]
46
    user_id: String,
47
    name: String,
48
    #[serde(rename = "imageUrl")]
49
    image_url: Option<String>,
50
}
51

            
52
const COL_NAME: &'static str = "client";
53

            
54
impl Model {
55
    /// To create the model instance with a database connection.
56
14
    pub async fn new(conn: Arc<Database>) -> Result<Self, Box<dyn StdError>> {
57
14
        let model = Model { conn };
58
14
        model.init().await?;
59
14
        Ok(model)
60
14
    }
61
}
62

            
63
#[async_trait]
64
impl ClientModel for Model {
65
24
    async fn init(&self) -> Result<(), Box<dyn StdError>> {
66
        let indexes = vec![
67
            doc! {"name": "clientId_1", "key": {"clientId": 1}, "unique": true},
68
            doc! {"name": "createdAt_1", "key": {"createdAt": 1}},
69
            doc! {"name": "modifiedAt_1", "key": {"modifiedAt": 1}},
70
            doc! {"name": "userId_1", "key": {"userId": 1}},
71
            doc! {"name": "name_1", "key": {"name": 1}},
72
        ];
73
        let command = doc! {
74
            "createIndexes": COL_NAME,
75
            "indexes": indexes,
76
        };
77
        self.conn.run_command(command).await?;
78
        Ok(())
79
24
    }
80

            
81
20
    async fn count(&self, cond: &ListQueryCond) -> Result<u64, Box<dyn StdError>> {
82
        let filter = get_list_query_filter(cond);
83
        let count = self
84
            .conn
85
            .collection::<Schema>(COL_NAME)
86
            .count_documents(filter)
87
            .await?;
88
        Ok(count)
89
20
    }
90

            
91
    async fn list(
92
        &self,
93
        opts: &ListOptions,
94
        cursor: Option<Box<dyn Cursor>>,
95
114
    ) -> Result<(Vec<Client>, Option<Box<dyn Cursor>>), Box<dyn StdError>> {
96
        let mut cursor = match cursor {
97
            None => {
98
                let filter = get_list_query_filter(opts.cond);
99
                Box::new(DbCursor::new(
100
                    build_find_options(opts, self.conn.collection::<Schema>(COL_NAME).find(filter))
101
                        .await?,
102
                ))
103
            }
104
            Some(cursor) => cursor,
105
        };
106

            
107
        let mut count: u64 = 0;
108
        let mut list = Vec::new();
109
        while let Some(item) = cursor.try_next().await? {
110
            list.push(item);
111
114
            if let Some(cursor_max) = opts.cursor_max {
112
                count += 1;
113
                if count >= cursor_max {
114
                    return Ok((list, Some(cursor)));
115
                }
116
            }
117
        }
118
        Ok((list, None))
119
114
    }
120

            
121
1726
    async fn get(&self, cond: &QueryCond) -> Result<Option<Client>, Box<dyn StdError>> {
122
        let filter = get_query_filter(cond);
123
        let mut cursor = self
124
            .conn
125
            .collection::<Schema>(COL_NAME)
126
            .find(filter)
127
            .await?;
128
        if let Some(item) = cursor.try_next().await? {
129
            return Ok(Some(Client {
130
                client_id: item.client_id,
131
                created_at: item.created_at.into(),
132
                modified_at: item.modified_at.into(),
133
                client_secret: item.client_secret,
134
                redirect_uris: item.redirect_uris,
135
                scopes: item.scopes,
136
                user_id: item.user_id,
137
                name: item.name,
138
                image_url: item.image_url,
139
            }));
140
        }
141
        Ok(None)
142
1726
    }
143

            
144
1068
    async fn add(&self, client: &Client) -> Result<(), Box<dyn StdError>> {
145
        let item = Schema {
146
            client_id: client.client_id.clone(),
147
            created_at: client.created_at.into(),
148
            modified_at: client.modified_at.into(),
149
            client_secret: client.client_secret.clone(),
150
            redirect_uris: client.redirect_uris.clone(),
151
            scopes: client.scopes.clone(),
152
            user_id: client.user_id.clone(),
153
            name: client.name.clone(),
154
            image_url: client.image_url.clone(),
155
        };
156
        self.conn
157
            .collection::<Schema>(COL_NAME)
158
            .insert_one(item)
159
            .await?;
160
        Ok(())
161
1068
    }
162

            
163
20
    async fn del(&self, cond: &QueryCond) -> Result<(), Box<dyn StdError>> {
164
        let filter = get_query_filter(cond);
165
        self.conn
166
            .collection::<Schema>(COL_NAME)
167
            .delete_many(filter)
168
            .await?;
169
        Ok(())
170
20
    }
171

            
172
    async fn update(
173
        &self,
174
        cond: &UpdateQueryCond,
175
        updates: &Updates,
176
28
    ) -> Result<(), Box<dyn StdError>> {
177
        let filter = get_update_query_filter(cond);
178
        if let Some(updates) = get_update_doc(updates) {
179
            self.conn
180
                .collection::<Schema>(COL_NAME)
181
                .update_one(filter, updates)
182
                .await?;
183
        }
184
        return Ok(());
185
28
    }
186
}
187

            
188
impl DbCursor {
189
    /// To create the cursor instance with a collection cursor.
190
92
    pub fn new(cursor: MongoDbCursor<Schema>) -> Self {
191
92
        DbCursor { cursor, offset: 0 }
192
92
    }
193
}
194

            
195
#[async_trait]
196
impl Cursor for DbCursor {
197
1970
    async fn try_next(&mut self) -> Result<Option<Client>, Box<dyn StdError>> {
198
        if let Some(item) = self.cursor.try_next().await? {
199
            self.offset += 1;
200
            return Ok(Some(Client {
201
                client_id: item.client_id,
202
                created_at: item.created_at.into(),
203
                modified_at: item.modified_at.into(),
204
                client_secret: item.client_secret,
205
                redirect_uris: item.redirect_uris,
206
                scopes: item.scopes,
207
                user_id: item.user_id,
208
                name: item.name,
209
                image_url: item.image_url,
210
            }));
211
        }
212
        Ok(None)
213
1970
    }
214

            
215
8
    fn offset(&self) -> u64 {
216
8
        self.offset
217
8
    }
218
}
219

            
220
/// Transforms query conditions to the MongoDB document.
221
1746
fn get_query_filter(cond: &QueryCond) -> Document {
222
1746
    let mut filter = Document::new();
223
1746
    if let Some(value) = cond.user_id {
224
24
        filter.insert("userId", value);
225
1722
    }
226
1746
    if let Some(value) = cond.client_id {
227
1742
        filter.insert("clientId", value);
228
1742
    }
229
1746
    filter
230
1746
}
231

            
232
/// Transforms query conditions to the MongoDB document.
233
112
fn get_list_query_filter(cond: &ListQueryCond) -> Document {
234
112
    let mut filter = Document::new();
235
112
    if let Some(value) = cond.user_id {
236
50
        filter.insert("userId", value);
237
62
    }
238
112
    if let Some(value) = cond.client_id {
239
8
        filter.insert("clientId", value);
240
104
    }
241
112
    if let Some(value) = cond.name_contains {
242
16
        if let Ok(pattern) = CString::try_from(value) {
243
16
            if let Ok(options) = CString::try_from("i") {
244
16
                filter.insert("name", Regex { pattern, options });
245
16
            }
246
        }
247
96
    }
248
112
    filter
249
112
}
250

            
251
/// Transforms model options to the options.
252
92
fn build_find_options<'a, T>(opts: &ListOptions, mut find: Find<'a, T>) -> Find<'a, T>
253
92
where
254
92
    T: Send + Sync,
255
{
256
92
    if let Some(offset) = opts.offset {
257
20
        find = find.skip(offset);
258
72
    }
259
92
    if let Some(limit) = opts.limit {
260
50
        if limit > 0 {
261
48
            find = find.limit(limit as i64);
262
48
        }
263
42
    }
264
92
    if let Some(sort_list) = opts.sort.as_ref() {
265
76
        if sort_list.len() > 0 {
266
74
            let mut sort_opts = Document::new();
267
88
            for cond in sort_list.iter() {
268
88
                let key = match cond.key {
269
22
                    SortKey::CreatedAt => "createdAt",
270
8
                    SortKey::ModifiedAt => "modifiedAt",
271
58
                    SortKey::Name => "name",
272
                };
273
88
                if cond.asc {
274
74
                    sort_opts.insert(key.to_string(), 1);
275
74
                } else {
276
14
                    sort_opts.insert(key.to_string(), -1);
277
14
                }
278
            }
279
74
            find = find.sort(sort_opts);
280
2
        }
281
16
    }
282
92
    find
283
92
}
284

            
285
/// Transforms query conditions to the MongoDB document.
286
28
fn get_update_query_filter(cond: &UpdateQueryCond) -> Document {
287
28
    doc! {
288
28
        "userId": cond.user_id,
289
28
        "clientId": cond.client_id,
290
    }
291
28
}
292

            
293
/// Transforms the model object to the MongoDB document.
294
28
fn get_update_doc(updates: &Updates) -> Option<Document> {
295
28
    let mut count = 0;
296
28
    let mut document = Document::new();
297
28
    if let Some(value) = updates.modified_at.as_ref() {
298
26
        document.insert(
299
26
            "modifiedAt",
300
26
            DateTime::from_millis(value.timestamp_millis()),
301
26
        );
302
26
        count += 1;
303
26
    }
304
28
    if let Some(value) = updates.client_secret.as_ref() {
305
6
        match value {
306
2
            None => {
307
2
                document.insert("clientSecret", Bson::Null);
308
2
            }
309
4
            Some(value) => {
310
4
                document.insert("clientSecret", value);
311
4
            }
312
        }
313
6
        count += 1;
314
22
    }
315
28
    if let Some(value) = updates.redirect_uris {
316
16
        document.insert("redirectUris", value);
317
16
        count += 1;
318
16
    }
319
28
    if let Some(value) = updates.scopes {
320
16
        document.insert("scopes", value);
321
16
        count += 1;
322
16
    }
323
28
    if let Some(value) = updates.name {
324
10
        document.insert("name", value);
325
10
        count += 1;
326
18
    }
327
28
    if let Some(value) = updates.image_url.as_ref() {
328
16
        match value {
329
8
            None => {
330
8
                document.insert("imageUrl", Bson::Null);
331
8
            }
332
8
            Some(value) => {
333
8
                document.insert("imageUrl", value);
334
8
            }
335
        }
336
16
        count += 1;
337
12
    }
338
28
    if count == 0 {
339
2
        return None;
340
26
    }
341
26
    Some(doc! {"$set": document})
342
28
}