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

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

            
12
use super::super::user::{
13
    Cursor, ListOptions, ListQueryCond, QueryCond, SortKey, Updates, User, UserModel,
14
};
15

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

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

            
30
/// MongoDB schema.
31
22036
#[derive(Deserialize, Serialize)]
32
struct Schema {
33
    #[serde(rename = "userId")]
34
    user_id: String,
35
    account: String,
36
    #[serde(rename = "createdAt")]
37
    created_at: DateTime,
38
    #[serde(rename = "modifiedAt")]
39
    modified_at: DateTime,
40
    #[serde(rename = "verifiedAt")]
41
    verified_at: Option<DateTime>,
42
    #[serde(rename = "expiredAt")]
43
    expired_at: Option<DateTime>,
44
    #[serde(rename = "disabledAt")]
45
    disabled_at: Option<DateTime>,
46
    roles: HashMap<String, bool>,
47
    password: String,
48
    salt: String,
49
    name: String,
50
    info: Document,
51
}
52

            
53
const COL_NAME: &'static str = "user";
54

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

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

            
85
21
    async fn count(&self, cond: &ListQueryCond) -> Result<u64, Box<dyn StdError>> {
86
21
        let filter = get_list_query_filter(cond);
87
21
        let count = self
88
21
            .conn
89
21
            .collection::<Schema>(COL_NAME)
90
21
            .count_documents(filter)
91
42
            .await?;
92
21
        Ok(count)
93
42
    }
94

            
95
    async fn list(
96
        &self,
97
        opts: &ListOptions,
98
        cursor: Option<Box<dyn Cursor>>,
99
83
    ) -> Result<(Vec<User>, Option<Box<dyn Cursor>>), Box<dyn StdError>> {
100
83
        let mut cursor = match cursor {
101
            None => {
102
72
                let filter = get_list_query_filter(opts.cond);
103
72
                Box::new(DbCursor::new(
104
72
                    build_find_options(opts, self.conn.collection::<Schema>(COL_NAME).find(filter))
105
144
                        .await?,
106
                ))
107
            }
108
11
            Some(cursor) => cursor,
109
        };
110

            
111
83
        let mut count: u64 = 0;
112
83
        let mut list = Vec::new();
113
1208
        while let Some(item) = cursor.try_next().await? {
114
1136
            list.push(item);
115
1136
            if let Some(cursor_max) = opts.cursor_max {
116
1032
                count += 1;
117
1032
                if count >= cursor_max {
118
11
                    return Ok((list, Some(cursor)));
119
1021
                }
120
104
            }
121
        }
122
72
        Ok((list, None))
123
166
    }
124

            
125
449
    async fn get(&self, cond: &QueryCond) -> Result<Option<User>, Box<dyn StdError>> {
126
449
        let filter = get_query_filter(cond);
127
449
        let mut cursor = self
128
449
            .conn
129
449
            .collection::<Schema>(COL_NAME)
130
449
            .find(filter)
131
897
            .await?;
132
449
        if let Some(user) = cursor.try_next().await? {
133
            return Ok(Some(User {
134
438
                user_id: user.user_id,
135
438
                account: user.account,
136
438
                created_at: user.created_at.into(),
137
438
                modified_at: user.modified_at.into(),
138
438
                verified_at: match user.verified_at {
139
12
                    None => None,
140
426
                    Some(value) => Some(value.into()),
141
                },
142
438
                expired_at: match user.expired_at {
143
428
                    None => None,
144
10
                    Some(value) => Some(value.into()),
145
                },
146
438
                disabled_at: match user.disabled_at {
147
427
                    None => None,
148
11
                    Some(value) => Some(value.into()),
149
                },
150
438
                roles: user.roles,
151
438
                password: user.password,
152
438
                salt: user.salt,
153
438
                name: user.name,
154
438
                info: bson::from_document(user.info)?,
155
            }));
156
11
        }
157
11
        Ok(None)
158
898
    }
159

            
160
549
    async fn add(&self, user: &User) -> Result<(), Box<dyn StdError>> {
161
549
        let item = Schema {
162
549
            user_id: user.user_id.clone(),
163
549
            account: user.account.to_lowercase(),
164
549
            created_at: user.created_at.into(),
165
549
            modified_at: user.modified_at.into(),
166
549
            verified_at: match user.verified_at {
167
31
                None => None,
168
518
                Some(value) => Some(value.into()),
169
            },
170
549
            expired_at: match user.expired_at {
171
538
                None => None,
172
11
                Some(value) => Some(value.into()),
173
            },
174
549
            disabled_at: match user.disabled_at {
175
541
                None => None,
176
8
                Some(value) => Some(value.into()),
177
            },
178
549
            roles: user.roles.clone(),
179
549
            password: user.password.clone(),
180
549
            salt: user.salt.clone(),
181
549
            name: user.name.clone(),
182
549
            info: bson::to_document(&user.info)?,
183
        };
184
549
        self.conn
185
549
            .collection::<Schema>(COL_NAME)
186
549
            .insert_one(item)
187
1099
            .await?;
188
547
        Ok(())
189
1098
    }
190

            
191
5
    async fn del(&self, user_id: &str) -> Result<(), Box<dyn StdError>> {
192
5
        let filter = doc! {"userId": user_id};
193
5
        self.conn
194
5
            .collection::<Schema>(COL_NAME)
195
5
            .delete_one(filter)
196
10
            .await?;
197
5
        Ok(())
198
10
    }
199

            
200
24
    async fn update(&self, user_id: &str, updates: &Updates) -> Result<(), Box<dyn StdError>> {
201
24
        let filter = doc! {"userId": user_id};
202
24
        if let Some(updates) = get_update_doc(updates) {
203
23
            self.conn
204
23
                .collection::<Schema>(COL_NAME)
205
23
                .update_one(filter, updates)
206
46
                .await?;
207
1
        }
208
24
        return Ok(());
209
48
    }
210
}
211

            
212
impl DbCursor {
213
    /// To create the cursor instance with a collection cursor.
214
72
    pub fn new(cursor: MongoDbCursor<Schema>) -> Self {
215
72
        DbCursor { cursor, offset: 0 }
216
72
    }
217
}
218

            
219
#[async_trait]
220
impl Cursor for DbCursor {
221
1208
    async fn try_next(&mut self) -> Result<Option<User>, Box<dyn StdError>> {
222
1208
        if let Some(item) = self.cursor.try_next().await? {
223
1136
            self.offset += 1;
224
1136
            return Ok(Some(User {
225
1136
                user_id: item.user_id,
226
1136
                account: item.account,
227
1136
                created_at: item.created_at.into(),
228
1136
                modified_at: item.modified_at.into(),
229
1136
                verified_at: match item.verified_at {
230
117
                    None => None,
231
1019
                    Some(value) => Some(value.into()),
232
                },
233
1136
                expired_at: match item.expired_at {
234
1077
                    None => None,
235
59
                    Some(value) => Some(value.into()),
236
                },
237
1136
                disabled_at: match item.disabled_at {
238
1075
                    None => None,
239
61
                    Some(value) => Some(value.into()),
240
                },
241
1136
                roles: item.roles,
242
1136
                password: item.password,
243
1136
                salt: item.salt,
244
1136
                name: item.name,
245
1136
                info: bson::from_document(item.info)?,
246
            }));
247
72
        }
248
72
        Ok(None)
249
2416
    }
250

            
251
4
    fn offset(&self) -> u64 {
252
4
        self.offset
253
4
    }
254
}
255

            
256
/// Transforms query conditions to the MongoDB document.
257
449
fn get_query_filter(cond: &QueryCond) -> Document {
258
449
    let mut filter = Document::new();
259
449
    if let Some(value) = cond.user_id {
260
299
        filter.insert("userId", value);
261
299
    }
262
449
    if let Some(value) = cond.account {
263
150
        filter.insert("account", value.to_lowercase().as_str());
264
299
    }
265
449
    filter
266
449
}
267

            
268
/// Transforms query conditions to the MongoDB document.
269
93
fn get_list_query_filter(cond: &ListQueryCond) -> Document {
270
93
    let mut filter = Document::new();
271
93
    if let Some(value) = cond.user_id {
272
2
        filter.insert("userId", value);
273
91
    }
274
93
    if let Some(value) = cond.account {
275
11
        filter.insert("account", value.to_lowercase().as_str());
276
82
    }
277
93
    if let Some(value) = cond.account_contains {
278
30
        filter.insert(
279
30
            "account",
280
30
            Regex {
281
30
                pattern: value.to_lowercase(),
282
30
                options: "i".to_string(),
283
30
            },
284
30
        );
285
63
    }
286
93
    if let Some(value) = cond.verified_at {
287
6
        if value {
288
4
            filter.insert("verifiedAt", doc! {"$ne": Bson::Null});
289
4
        } else {
290
2
            filter.insert("verifiedAt", Bson::Null);
291
2
        }
292
87
    }
293
93
    if let Some(value) = cond.disabled_at {
294
4
        if value {
295
2
            filter.insert("disabledAt", doc! {"$ne": Bson::Null});
296
2
        } else {
297
2
            filter.insert("disabledAt", Bson::Null);
298
2
        }
299
89
    }
300
93
    if let Some(value) = cond.name_contains {
301
6
        filter.insert(
302
6
            "name",
303
6
            Regex {
304
6
                pattern: value.to_string(),
305
6
                options: "i".to_string(),
306
6
            },
307
6
        );
308
87
    }
309
93
    filter
310
93
}
311

            
312
/// Transforms model options to the options.
313
72
fn build_find_options<'a, T>(opts: &ListOptions, mut find: Find<'a, T>) -> Find<'a, T>
314
72
where
315
72
    T: Send + Sync,
316
72
{
317
72
    if let Some(offset) = opts.offset {
318
10
        find = find.skip(offset);
319
62
    }
320
72
    if let Some(limit) = opts.limit {
321
38
        if limit > 0 {
322
37
            find = find.limit(limit as i64);
323
37
        }
324
34
    }
325
72
    if let Some(sort_list) = opts.sort.as_ref() {
326
60
        if sort_list.len() > 0 {
327
59
            let mut sort_opts = Document::new();
328
65
            for cond in sort_list.iter() {
329
65
                let key = match cond.key {
330
41
                    SortKey::Account => "account",
331
6
                    SortKey::CreatedAt => "createdAt",
332
4
                    SortKey::ModifiedAt => "modifiedAt",
333
4
                    SortKey::VerifiedAt => "verifiedAt",
334
2
                    SortKey::ExpiredAt => "expiredAt",
335
2
                    SortKey::DisabledAt => "disabledAt",
336
6
                    SortKey::Name => "name",
337
                };
338
65
                if cond.asc {
339
52
                    sort_opts.insert(key.to_string(), 1);
340
52
                } else {
341
13
                    sort_opts.insert(key.to_string(), -1);
342
13
                }
343
            }
344
59
            find = find.sort(sort_opts);
345
1
        }
346
12
    }
347
72
    find
348
72
}
349

            
350
/// Transforms the model object to the MongoDB document.
351
24
fn get_update_doc(updates: &Updates) -> Option<Document> {
352
24
    let mut count = 0;
353
24
    let mut document = Document::new();
354
24
    if let Some(value) = updates.modified_at.as_ref() {
355
23
        document.insert(
356
23
            "modifiedAt",
357
23
            DateTime::from_millis(value.timestamp_millis()),
358
23
        );
359
23
        count += 1;
360
23
    }
361
24
    if let Some(value) = updates.verified_at.as_ref() {
362
4
        document.insert(
363
4
            "verifiedAt",
364
4
            DateTime::from_millis(value.timestamp_millis()),
365
4
        );
366
4
        count += 1;
367
20
    }
368
24
    if let Some(value) = updates.expired_at.as_ref() {
369
4
        match value {
370
3
            None => {
371
3
                document.insert("expiredAt", Bson::Null);
372
3
            }
373
1
            Some(value) => {
374
1
                document.insert("expiredAt", DateTime::from_millis(value.timestamp_millis()));
375
1
            }
376
        }
377
4
        count += 1;
378
20
    }
379
24
    if let Some(value) = updates.disabled_at.as_ref() {
380
8
        match value {
381
4
            None => {
382
4
                document.insert("disabledAt", Bson::Null);
383
4
            }
384
4
            Some(value) => {
385
4
                document.insert(
386
4
                    "disabledAt",
387
4
                    DateTime::from_millis(value.timestamp_millis()),
388
4
                );
389
4
            }
390
        }
391
8
        count += 1;
392
16
    }
393
24
    if let Some(value) = updates.roles {
394
6
        let mut doc = Document::new();
395
17
        for (k, v) in value {
396
11
            doc.insert(k, v);
397
11
        }
398
6
        document.insert("roles", doc);
399
6
        count += 1;
400
18
    }
401
24
    if let Some(value) = updates.password.as_ref() {
402
7
        document.insert("password", value);
403
7
        count += 1;
404
17
    }
405
24
    if let Some(value) = updates.salt.as_ref() {
406
7
        document.insert("salt", value);
407
7
        count += 1;
408
17
    }
409
24
    if let Some(value) = updates.name {
410
16
        document.insert("name", value);
411
16
        count += 1;
412
16
    }
413
24
    if let Some(value) = updates.info {
414
16
        document.insert(
415
16
            "info",
416
16
            match bson::to_document(value) {
417
                Err(_) => return None,
418
16
                Ok(doc) => doc,
419
16
            },
420
16
        );
421
16
        count += 1;
422
8
    }
423
24
    if count == 0 {
424
1
        return None;
425
23
    }
426
23
    Some(doc! {"$set": document})
427
24
}