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

            
3
use async_trait::async_trait;
4
use chrono::{TimeZone, Utc};
5
use futures::TryStreamExt;
6
use sql_builder::{quote, SqlBuilder};
7
use sqlx::SqlitePool;
8

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

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

            
23
/// Cursor instance.
24
///
25
/// The SQLite implementation uses the original list options and the progress offset.
26
pub struct DbCursor {
27
    offset: u64,
28
}
29

            
30
/// SQLite schema.
31
#[derive(sqlx::FromRow)]
32
struct Schema {
33
    client_id: String,
34
    /// i64 as time tick from Epoch in milliseconds.
35
    created_at: i64,
36
    /// i64 as time tick from Epoch in milliseconds.
37
    modified_at: i64,
38
    client_secret: Option<String>,
39
    redirect_uris: String,
40
    /// Space-separated value such as `scope1 scope2`.
41
    scopes: String,
42
    user_id: String,
43
    name: String,
44
    image_url: Option<String>,
45
}
46

            
47
/// Use "COUNT(*)" instead of "COUNT(fields...)" to simplify the implementation.
48
#[derive(sqlx::FromRow)]
49
struct CountSchema {
50
    #[sqlx(rename = "COUNT(*)")]
51
    count: i64,
52
}
53

            
54
const TABLE_NAME: &'static str = "client";
55
const FIELDS: &'static [&'static str] = &[
56
    "client_id",
57
    "created_at",
58
    "modified_at",
59
    "client_secret",
60
    "redirect_uris",
61
    "scopes",
62
    "user_id",
63
    "name",
64
    "image_url",
65
];
66
const TABLE_INIT_SQL: &'static str = "\
67
    CREATE TABLE IF NOT EXISTS client (\
68
    client_id TEXT NOT NULL UNIQUE,\
69
    created_at INTEGER NOT NULL,\
70
    modified_at INTEGER NOT NULL,\
71
    client_secret TEXT,\
72
    redirect_uris TEXT NOT NULL,\
73
    scopes TEXT NOT NULL,\
74
    user_id TEXT NOT NULL,\
75
    name TEXT NOT NULL,\
76
    image_url TEXT,\
77
    PRIMARY KEY (client_id))";
78

            
79
impl Model {
80
    /// To create the model instance with a database connection.
81
11
    pub async fn new(conn: Arc<SqlitePool>) -> Result<Self, Box<dyn StdError>> {
82
11
        let model = Model { conn };
83
43
        model.init().await?;
84
11
        Ok(model)
85
11
    }
86
}
87

            
88
#[async_trait]
89
impl ClientModel for Model {
90
20
    async fn init(&self) -> Result<(), Box<dyn StdError>> {
91
20
        let _ = sqlx::query(TABLE_INIT_SQL)
92
20
            .execute(self.conn.as_ref())
93
61
            .await?;
94
20
        Ok(())
95
20
    }
96

            
97
10
    async fn count(&self, cond: &ListQueryCond) -> Result<u64, Box<dyn StdError>> {
98
10
        let sql = build_list_where(SqlBuilder::select_from(TABLE_NAME).count("*"), &cond).sql()?;
99
10

            
100
10
        let result: Result<CountSchema, sqlx::Error> = sqlx::query_as(sql.as_str())
101
10
            .fetch_one(self.conn.as_ref())
102
20
            .await;
103
10

            
104
10
        let row = match result {
105
10
            Err(e) => {
106
                return Err(Box::new(e));
107
10
            }
108
10
            Ok(row) => row,
109
10
        };
110
10
        Ok(row.count as u64)
111
10
    }
112

            
113
    async fn list(
114
        &self,
115
        opts: &ListOptions,
116
        cursor: Option<Box<dyn Cursor>>,
117
57
    ) -> Result<(Vec<Client>, Option<Box<dyn Cursor>>), Box<dyn StdError>> {
118
57
        let mut cursor = match cursor {
119
57
            None => Box::new(DbCursor::new()),
120
57
            Some(cursor) => cursor,
121
57
        };
122
57

            
123
57
        let mut opts = ListOptions { ..*opts };
124
57
        if let Some(offset) = opts.offset {
125
14
            opts.offset = Some(offset + cursor.offset());
126
43
        } else {
127
43
            opts.offset = Some(cursor.offset());
128
43
        }
129
57
        let opts_limit = opts.limit;
130
57
        if let Some(limit) = opts_limit {
131
57
            if limit > 0 {
132
57
                if cursor.offset() >= limit {
133
57
                    return Ok((vec![], None));
134
57
                }
135
28
                opts.limit = Some(limit - cursor.offset());
136
57
            }
137
57
        }
138
57
        let mut builder = SqlBuilder::select_from(TABLE_NAME);
139
54
        build_limit_offset(&mut builder, &opts);
140
54
        build_sort(&mut builder, &opts);
141
57
        let sql = build_list_where(&mut builder, opts.cond).sql()?;
142
57

            
143
57
        let mut rows = sqlx::query_as::<_, Schema>(sql.as_str()).fetch(self.conn.as_ref());
144
54

            
145
54
        let mut count: u64 = 0;
146
54
        let mut list = vec![];
147
976
        while let Some(row) = rows.try_next().await? {
148
939
            let _ = cursor.as_mut().try_next().await?;
149
939
            let redirect_uris = row
150
939
                .redirect_uris
151
939
                .split(" ")
152
939
                .filter_map(|x| {
153
939
                    if x.len() > 0 {
154
867
                        Some(x.to_string())
155
57
                    } else {
156
72
                        None
157
57
                    }
158
939
                })
159
939
                .collect();
160
939
            let scopes = row
161
939
                .scopes
162
939
                .split(" ")
163
939
                .filter_map(|x| {
164
939
                    if x.len() > 0 {
165
57
                        Some(x.to_string())
166
57
                    } else {
167
939
                        None
168
57
                    }
169
939
                })
170
939
                .collect();
171
939
            list.push(Client {
172
939
                client_id: row.client_id,
173
939
                created_at: Utc.timestamp_nanos(row.created_at * 1000000),
174
939
                modified_at: Utc.timestamp_nanos(row.modified_at * 1000000),
175
939
                client_secret: row.client_secret,
176
939
                redirect_uris,
177
939
                scopes,
178
939
                user_id: row.user_id,
179
939
                name: row.name,
180
939
                image_url: row.image_url,
181
939
            });
182
939
            if let Some(limit) = opts_limit {
183
489
                if limit > 0 && cursor.offset() >= limit {
184
57
                    if let Some(cursor_max) = opts.cursor_max {
185
57
                        if (count + 1) >= cursor_max {
186
57
                            return Ok((list, Some(cursor)));
187
57
                        }
188
57
                    }
189
57
                    return Ok((list, None));
190
480
                }
191
450
            }
192
930
            if let Some(cursor_max) = opts.cursor_max {
193
872
                count += 1;
194
872
                if count >= cursor_max {
195
57
                    return Ok((list, Some(cursor)));
196
864
                }
197
58
            }
198
57
        }
199
57
        Ok((list, None))
200
57
    }
201

            
202
870
    async fn get(&self, cond: &QueryCond) -> Result<Option<Client>, Box<dyn StdError>> {
203
870
        let sql = build_where(SqlBuilder::select_from(TABLE_NAME).fields(FIELDS), &cond).sql()?;
204
870

            
205
870
        let result: Result<Schema, sqlx::Error> = sqlx::query_as(sql.as_str())
206
870
            .fetch_one(self.conn.as_ref())
207
1740
            .await;
208
870

            
209
870
        let row = match result {
210
870
            Err(e) => match e {
211
870
                sqlx::Error::RowNotFound => return Ok(None),
212
870
                _ => return Err(Box::new(e)),
213
870
            },
214
870
            Ok(row) => row,
215
852
        };
216
852

            
217
852
        let redirect_uris = row
218
852
            .redirect_uris
219
852
            .split(" ")
220
867
            .filter_map(|x| {
221
867
                if x.len() > 0 {
222
870
                    Some(x.to_string())
223
870
                } else {
224
870
                    None
225
870
                }
226
870
            })
227
852
            .collect();
228
852
        let scopes = row
229
852
            .scopes
230
852
            .split(" ")
231
868
            .filter_map(|x| {
232
868
                if x.len() > 0 {
233
870
                    Some(x.to_string())
234
870
                } else {
235
870
                    None
236
870
                }
237
870
            })
238
852
            .collect();
239
852
        Ok(Some(Client {
240
852
            client_id: row.client_id,
241
852
            created_at: Utc.timestamp_nanos(row.created_at * 1000000),
242
852
            modified_at: Utc.timestamp_nanos(row.modified_at * 1000000),
243
852
            client_secret: row.client_secret,
244
852
            redirect_uris,
245
852
            scopes,
246
852
            user_id: row.user_id,
247
852
            name: row.name,
248
852
            image_url: row.image_url,
249
852
        }))
250
870
    }
251

            
252
534
    async fn add(&self, client: &Client) -> Result<(), Box<dyn StdError>> {
253
534
        let client_secret = match client.client_secret.as_deref() {
254
534
            None => "NULL".to_string(),
255
534
            Some(value) => quote(value),
256
534
        };
257
534
        let image_url = match client.image_url.as_deref() {
258
534
            None => "NULL".to_string(),
259
534
            Some(value) => quote(value),
260
534
        };
261
534
        let values = vec![
262
534
            quote(client.client_id.as_str()),
263
534
            client.created_at.timestamp_millis().to_string(),
264
534
            client.modified_at.timestamp_millis().to_string(),
265
534
            client_secret,
266
534
            quote(client.redirect_uris.join(" ")),
267
534
            quote(client.scopes.join(" ")),
268
534
            quote(client.user_id.as_str()),
269
534
            quote(client.name.as_str()),
270
534
            image_url,
271
534
        ];
272
534
        let sql = SqlBuilder::insert_into(TABLE_NAME)
273
534
            .fields(FIELDS)
274
534
            .values(&values)
275
534
            .sql()?;
276
534
        let _ = sqlx::query(sql.as_str())
277
534
            .execute(self.conn.as_ref())
278
1068
            .await?;
279
534
        Ok(())
280
534
    }
281

            
282
10
    async fn del(&self, cond: &QueryCond) -> Result<(), Box<dyn StdError>> {
283
10
        let sql = build_where(&mut SqlBuilder::delete_from(TABLE_NAME), cond).sql()?;
284
10
        let _ = sqlx::query(sql.as_str())
285
10
            .execute(self.conn.as_ref())
286
20
            .await?;
287
10
        Ok(())
288
10
    }
289

            
290
    async fn update(
291
        &self,
292
        cond: &UpdateQueryCond,
293
        updates: &Updates,
294
14
    ) -> Result<(), Box<dyn StdError>> {
295
14
        let sql = match build_update_where(&mut SqlBuilder::update_table(TABLE_NAME), cond, updates)
296
14
        {
297
14
            None => return Ok(()),
298
14
            Some(builder) => builder.sql()?,
299
14
        };
300
14
        let _ = sqlx::query(sql.as_str())
301
13
            .execute(self.conn.as_ref())
302
26
            .await?;
303
14
        Ok(())
304
14
    }
305
}
306

            
307
impl DbCursor {
308
    /// To create the cursor instance.
309
46
    pub fn new() -> Self {
310
46
        DbCursor { offset: 0 }
311
46
    }
312
}
313

            
314
#[async_trait]
315
impl Cursor for DbCursor {
316
939
    async fn try_next(&mut self) -> Result<Option<Client>, Box<dyn StdError>> {
317
939
        self.offset += 1;
318
939
        Ok(None)
319
939
    }
320

            
321
605
    fn offset(&self) -> u64 {
322
605
        self.offset
323
605
    }
324
}
325

            
326
/// Transforms query conditions to the SQL builder.
327
880
fn build_where<'a>(builder: &'a mut SqlBuilder, cond: &QueryCond<'a>) -> &'a mut SqlBuilder {
328
880
    if let Some(value) = cond.user_id {
329
12
        builder.and_where_eq("user_id", quote(value));
330
868
    }
331
880
    if let Some(value) = cond.client_id {
332
878
        builder.and_where_eq("client_id", quote(value));
333
878
    }
334
880
    builder
335
880
}
336

            
337
/// Transforms query conditions to the SQL builder.
338
64
fn build_list_where<'a>(
339
64
    builder: &'a mut SqlBuilder,
340
64
    cond: &ListQueryCond<'a>,
341
64
) -> &'a mut SqlBuilder {
342
64
    if let Some(value) = cond.user_id {
343
30
        builder.and_where_eq("user_id", quote(value));
344
34
    }
345
64
    if let Some(value) = cond.client_id {
346
4
        builder.and_where_eq("client_id", quote(value));
347
60
    }
348
64
    if let Some(value) = cond.name_contains {
349
8
        build_where_like(builder, "name", value);
350
56
    }
351
64
    builder
352
64
}
353

            
354
/// Transforms model options to the SQL builder.
355
54
fn build_limit_offset<'a>(builder: &'a mut SqlBuilder, opts: &ListOptions) -> &'a mut SqlBuilder {
356
54
    if let Some(value) = opts.limit {
357
29
        if value > 0 {
358
28
            builder.limit(value);
359
28
        }
360
25
    }
361
54
    if let Some(value) = opts.offset {
362
54
        match opts.limit {
363
25
            None => builder.limit(-1).offset(value),
364
1
            Some(0) => builder.limit(-1).offset(value),
365
28
            _ => builder.offset(value),
366
        };
367
    }
368
54
    builder
369
54
}
370

            
371
/// Transforms model options to the SQL builder.
372
54
fn build_sort<'a>(builder: &'a mut SqlBuilder, opts: &ListOptions) -> &'a mut SqlBuilder {
373
54
    if let Some(sort_cond) = opts.sort.as_ref() {
374
52
        for cond in sort_cond.iter() {
375
52
            let key = match cond.key {
376
11
                SortKey::CreatedAt => "created_at",
377
4
                SortKey::ModifiedAt => "modified_at",
378
37
                SortKey::Name => "name",
379
            };
380
52
            builder.order_by(key, !cond.asc);
381
        }
382
8
    }
383
54
    builder
384
54
}
385

            
386
/// Transforms query conditions and the model object to the SQL builder.
387
14
fn build_update_where<'a>(
388
14
    builder: &'a mut SqlBuilder,
389
14
    cond: &UpdateQueryCond<'a>,
390
14
    updates: &Updates,
391
14
) -> Option<&'a mut SqlBuilder> {
392
14
    let mut count = 0;
393
14
    if let Some(value) = updates.modified_at.as_ref() {
394
13
        builder.set("modified_at", value.timestamp_millis());
395
13
        count += 1;
396
13
    }
397
14
    if let Some(value) = updates.client_secret.as_ref() {
398
3
        match value {
399
1
            None => {
400
1
                builder.set("client_secret", "NULL");
401
1
            }
402
2
            Some(value) => {
403
2
                builder.set("client_secret", quote(value));
404
2
            }
405
        }
406
3
        count += 1;
407
11
    }
408
14
    if let Some(value) = updates.redirect_uris.as_ref() {
409
8
        builder.set("redirect_uris", quote(value.join(" ")));
410
8
        count += 1;
411
8
    }
412
14
    if let Some(value) = updates.scopes.as_ref() {
413
8
        builder.set("scopes", quote(value.join(" ")));
414
8
        count += 1;
415
8
    }
416
14
    if let Some(value) = updates.name.as_ref() {
417
5
        builder.set("name", quote(value));
418
5
        count += 1;
419
9
    }
420
14
    if let Some(value) = updates.image_url.as_ref() {
421
8
        match value {
422
4
            None => {
423
4
                builder.set("image_url", "NULL");
424
4
            }
425
4
            Some(value) => {
426
4
                builder.set("image_url", quote(value));
427
4
            }
428
        }
429
8
        count += 1;
430
6
    }
431
14
    if count == 0 {
432
1
        return None;
433
13
    }
434
13

            
435
13
    builder.and_where_eq("user_id", quote(cond.user_id));
436
13
    builder.and_where_eq("client_id", quote(cond.client_id));
437
13
    Some(builder)
438
14
}