shared-framework 0.0.17

Reusable building blocks for HTTP services — Hyper routing, SeaORM data layer, validation, OpenAPI docs, jobs, queues, cache.
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
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
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
343
344
345
346
347
348
349
350
351
352
353
354
355
356
357
358
359
360
361
362
363
364
365
366
367
368
369
370
371
372
373
374
375
376
377
378
379
380
381
382
383
384
385
386
387
388
389
390
391
392
393
394
395
396
397
398
399
400
401
402
403
404
405
406
407
408
409
410
411
412
413
414
415
416
417
418
419
420
421
422
423
424
425
426
427
428
429
430
431
432
433
434
435
436
437
438
439
440
441
442
443
444
445
446
447
448
449
450
451
452
453
454
455
456
457
458
//! Query building, per-request options, and paginated results.
//!
//! Provides [`QueryData`] (a SeaORM `Select` wrapper with pagination and
//! relation-hydration steps), [`DeleteQueryData`] for bulk deletes,
//! [`RepositoryOptions`] for limit/cursor/user/correlation settings,
//! [`PageResult`] for cursor pages, and [`CursorData`] for opaque cursors.
//!
//! Build a [`QueryData`], optionally attach hydration with
//! [`QueryData::with_traverser`], then run it through
//! [`crate::data::PersistentRepository`] (`get_all`/`get_many`/`get_one`/
//! `get_paginated_view` and the `*_traversed` variants).
//!
//! Cursors are base64-encoded JSON holding a `limit` and an `id` cursor.
//! Selects filter on `id` in the direction recorded by the ordering
//! (ascending uses `id > cursor`, descending uses `id < cursor`);
//! deletes always use `id > cursor`.

use crate::data::BaseEntity;
use crate::logging::correlation::CorrelationContext;
use base64::{engine::general_purpose::STANDARD as BASE64, Engine};
use chrono::Utc;
use sea_orm::prelude::DateTimeWithTimeZone;
use sea_orm::{entity::EntityLoaderTrait, DatabaseTransaction, EntityTrait};
use serde::{Deserialize, Serialize};
use std::sync::Arc;

/// Default user type used when no caller identity is supplied.
///
/// Implements [`BaseEntity`] with `id` 0, a nil UUID, and current timestamps.
#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)]
pub struct NoUser;

impl BaseEntity for NoUser {
    type LoaderType = ();

    fn load() -> Self::LoaderType {}

    fn id(&self) -> i64 {
        0
    }
    fn uid(&self) -> uuid::Uuid {
        uuid::Uuid::nil()
    }
    fn created_at(&self) -> DateTimeWithTimeZone {
        Utc::now().into()
    }
    fn updated_at(&self) -> DateTimeWithTimeZone {
        Utc::now().into()
    }
}

/// Opaque pagination cursor carrying the page size and last-seen `id`.
///
/// Encoded as base64 JSON by [`CursorData::encode`]; decoded back by
/// [`CursorData::decode`], which returns `None` for malformed input.
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct CursorData {
    /// Page size the cursor was issued for.
    pub limit: usize,
    /// Last-seen row `id`; the next page continues after (or before) it.
    pub cursor: i64,
}

impl CursorData {
    /// Encodes this cursor as a base64 JSON string.
    pub fn encode(&self) -> String {
        let json = serde_json::to_string(self).unwrap_or_default();
        BASE64.encode(json.as_bytes())
    }
    /// Decodes a string produced by [`CursorData::encode`].
    ///
    /// Returns `None` when the input is not valid base64, not valid UTF-8,
    /// or not valid cursor JSON.
    pub fn decode(s: &str) -> Option<Self> {
        let bytes = BASE64.decode(s).ok()?;
        let json = String::from_utf8(bytes).ok()?;
        serde_json::from_str(&json).ok()
    }
}

/// Per-request settings for repository queries.
///
/// `U` is the caller identity type and must implement [`BaseEntity`]; it
/// defaults to [`NoUser`]. An identity can be supplied either as a full
/// entity (`user`) or as a raw id (`user_id`).
///
/// An optional database transaction can be attached via
/// [`RepositoryOptions::with_transaction`]. When present, repository
/// operations run against that transaction; otherwise they use the
/// repository's own connection.
#[derive(Debug, Clone)]
pub struct RepositoryOptions<U = NoUser>
where
    U: BaseEntity + Clone + Send + Sync + 'static,
{
    /// Maximum rows for limited reads; default is 15.
    pub limit: usize,
    /// Opaque cursor from a previous page, if continuing pagination.
    pub cursor: Option<String>,
    /// Whether selects use `SELECT DISTINCT`.
    pub distinct: bool,
    /// Detach flag carried alongside the query; default is `true`.
    pub detach: bool,
    /// Caller identity as an entity, if supplied.
    pub user: Option<U>,
    /// Caller identity as a raw id, if supplied.
    pub user_id: Option<i64>,
    /// Request correlation propagated with the query, if supplied.
    pub correlation: Option<Arc<CorrelationContext>>,
    /// Optional transaction used as the query executor when set.
    ///
    /// Held behind an [`Arc`] so the options stay cheaply cloneable even
    /// though [`DatabaseTransaction`] itself is neither `Clone` nor `Send`-free
    /// to move. Share with `Arc::new(txn)` — note a begun transaction cannot
    /// be cloned out of SeaORM, so callers typically wrap it at begin time
    /// (see [`RepositoryOptions::with_transaction`]).
    pub txn: Option<Arc<DatabaseTransaction>>,
}

impl<U> Default for RepositoryOptions<U>
where
    U: BaseEntity + Clone + Send + Sync + 'static,
{
    fn default() -> Self {
        Self {
            limit: 15,
            cursor: None,
            distinct: false,
            detach: true,
            user: None,
            user_id: None,
            correlation: None,
            txn: None,
        }
    }
}

impl<U> RepositoryOptions<U>
where
    U: BaseEntity + Clone + Send + Sync + 'static,
{
    /// Creates options with defaults: limit 15, no cursor, `distinct` false, `detach` true.
    pub fn new() -> Self {
        Self::default()
    }

    /// Sets the maximum rows for limited reads.
    pub fn with_limit(mut self, limit: usize) -> Self {
        self.limit = limit;
        self
    }

    /// Sets the pagination cursor to continue from.
    pub fn with_cursor(mut self, cursor: impl Into<String>) -> Self {
        self.cursor = Some(cursor.into());
        self
    }

    /// Sets the caller identity as an entity; also records its `id` as `user_id`.
    pub fn with_user(mut self, user: U) -> Self {
        self.user_id = Some(user.id());
        self.user = Some(user);
        self
    }

    /// Sets the caller identity as a raw id without an entity.
    pub fn with_user_id(mut self, user_id: i64) -> Self {
        self.user_id = Some(user_id);
        self
    }

    /// Attaches request correlation context to the options.
    pub fn with_correlation(mut self, ctx: Arc<CorrelationContext>) -> Self {
        self.correlation = Some(ctx);
        self
    }

    /// Enables `SELECT DISTINCT` for the query.
    pub fn distinct(mut self) -> Self {
        self.distinct = true;
        self
    }

    /// Sets the detach flag carried alongside the query.
    pub fn with_detach(mut self, detach: bool) -> Self {
        self.detach = detach;
        self
    }

    /// Attaches a transaction used as the executor for database operations.
    ///
    /// When set, repository methods run against this transaction instead of
    /// the repository's own connection. When unset, they fall back to the
    /// repository connection.
    pub fn with_transaction(mut self, txn: Arc<DatabaseTransaction>) -> Self {
        self.txn = Some(txn);
        self
    }

    /// Alias for [`RepositoryOptions::with_transaction`].
    pub fn with_txn(mut self, txn: Arc<DatabaseTransaction>) -> Self {
        self.txn = Some(txn);
        self
    }

    /// Removes any attached transaction, restoring execution on the repository connection.
    pub fn without_transaction(mut self) -> Self {
        self.txn = None;
        self
    }

    /// Returns the attached transaction, if any.
    pub fn transaction(&self) -> Option<&DatabaseTransaction> {
        self.txn.as_deref()
    }

    /// Returns true when a transaction is attached.
    pub fn has_transaction(&self) -> bool {
        self.txn.is_some()
    }

    /// Returns the effective caller id: `user_id` when set, else the `user` entity id.
    pub fn user_id(&self) -> Option<i64> {
        self.user_id.or_else(|| self.user.as_ref().map(|u| u.id()))
    }

    /// Overlays `other` on top of `self`, joining two parameter sets as you please.
    ///
    /// Scalar fields (`limit`, `distinct`, `detach`) are taken from `other`;
    /// optional fields (`cursor`, `user`, `user_id`, `correlation`, `txn`) fall back
    /// to `self` when `other` leaves them unset. A replaced `user` entity
    /// clears a stale inherited `user_id` (the id is still derivable from the
    /// entity via [`RepositoryOptions::user_id`]).
    ///
    /// Typical use: request-derived base joined with caller overrides —
    /// `RepositoryOptions::from_ctx(ctx).join(RepositoryOptions::new().with_limit(50))`.
    pub fn join(mut self, other: Self) -> Self {
        self.limit = other.limit;
        self.distinct = other.distinct;
        self.detach = other.detach;
        self.cursor = other.cursor.or(self.cursor);
        if other.user.is_some() {
            self.user = other.user;
            // Keep the id consistent with the newly joined entity unless the
            // caller explicitly joined an id alongside it.
            if other.user_id.is_none() {
                self.user_id = None;
            } else {
                self.user_id = other.user_id;
            }
        } else {
            self.user_id = other.user_id.or(self.user_id);
        }
        self.correlation = other.correlation.or(self.correlation);
        self.txn = other.txn.or(self.txn);
        self
    }
}

// Convenience for NoUser — with_user_id without entity
impl RepositoryOptions<NoUser> {
    /// Builds options from the request context: pagination limit and cursor,
    /// the numeric user id when the context carries one, and the correlation itself.
    /// Join caller overrides on top via [`RepositoryOptions::join`].
    pub fn from_ctx(ctx: Arc<CorrelationContext>) -> Self {
        Self {
            limit: ctx.pagination_limit(),
            cursor: ctx.pagination_cursor(),
            user_id: ctx.user_id().and_then(|s| s.parse().ok()),
            correlation: Some(ctx),
            ..Self::default()
        }
    }

    /// Replaces the [`NoUser`] identity with the given entity, keeping other settings.
    pub fn with_user_entity<U>(self, user: U) -> RepositoryOptions<U>
    where
        U: BaseEntity + Clone + Send + Sync + 'static,
    {
        RepositoryOptions {
            limit: self.limit,
            cursor: self.cursor,
            distinct: self.distinct,
            detach: self.detach,
            user: Some(user),
            user_id: None,
            correlation: self.correlation,
            txn: self.txn,
        }
    }
}

/// One cursor page of rows.
///
/// `T` is the row model type. `data` serializes as `items`, `has_next` as
/// `hasNext`, `next_cursor` as `next`, `total` as `count`, and `limit` as `limit`.
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct PageResult<T> {
    /// Rows in this page, serialized as `items`.
    #[serde(rename = "items")]
    pub data: Vec<T>,

    /// Whether another page follows, serialized as `hasNext`.
    #[serde(rename = "hasNext")]
    pub has_next: bool,

    /// Cursor that produced this page, if any.
    #[serde(rename = "cursor", skip_serializing_if = "Option::is_none")]
    pub cursor: Option<String>,

    /// Cursor for the next page (`None` when `has_next` is false), serialized as `next`.
    #[serde(rename = "next", skip_serializing_if = "Option::is_none")]
    pub next_cursor: Option<String>,

    /// Total matching rows ignoring cursor and limit, serialized as `count`.
    #[serde(rename = "count")]
    pub total: i64,

    /// Page size used for this page.
    #[serde(rename = "limit")]
    pub limit: usize,
}

impl<T> PageResult<T> {
    /// Returns the rows in this page.
    pub fn items(&self) -> &[T] {
        &self.data
    }
    /// Returns the page size recorded in the next-page cursor, if present and valid.
    pub fn limit(&self) -> Option<usize> {
        self.next_cursor
            .as_ref()
            .and_then(|c| CursorData::decode(c))
            .map(|d| d.limit)
    }
}

/// Builder wrapping a SeaORM entity loader with pagination and hydration settings.
///
/// `E` is the SeaORM entity being queried; its model must implement [`BaseEntity`]
/// so cursor pagination can use `id` comparisons. `order_asc` records the
/// cursor direction (`Some(true)` ascending, `Some(false)` descending).
pub struct QueryData<E>
where
    E: EntityTrait,
    E::ModelEx: BaseEntity,
    <E::ModelEx as BaseEntity>::LoaderType: EntityLoaderTrait<E, ModelEx = E::ModelEx>,
{
    /// The entity loader used to load relations for this entity type.
    pub loader: <E::ModelEx as BaseEntity>::LoaderType,

    /// Limit/cursor/distinct/detach settings applied at execution time.
    pub opts: RepositoryOptions<NoUser>,

    /// Recorded ordering direction for cursor filtering, if any ordering was set.
    pub order_asc: Option<bool>,
}

impl<E> QueryData<E>
where
    E: EntityTrait,
    E::ModelEx: BaseEntity,
    <E::ModelEx as BaseEntity>::LoaderType: EntityLoaderTrait<E, ModelEx = E::ModelEx>,
{
    /// Wraps an entity loader with default options and no ordering.
    pub fn new(loader: <E::ModelEx as BaseEntity>::LoaderType) -> Self {
        Self {
            loader,
            opts: RepositoryOptions::default(),
            order_asc: None,
        }
    }

    /// Copies limit, cursor, distinct, detach, correlation, and transaction settings from the given options.
    pub fn with_options<U>(mut self, opts: RepositoryOptions<U>) -> Self
    where
        U: BaseEntity + Clone + Send + Sync + 'static,
    {
        self.opts.limit = opts.limit;
        self.opts.cursor = opts.cursor;
        self.opts.distinct = opts.distinct;
        self.opts.detach = opts.detach;
        self.opts.correlation = opts.correlation;
        self.opts.txn = opts.txn;
        self
    }

    /// Orders by `id` and records the direction for cursor filtering.
    pub fn order_by(mut self, asc: bool) -> Self {
        if asc {
            self.loader = self.loader.order_by_id_asc();
        } else {
            self.loader = self.loader.order_by_id_desc();
        }
        self.order_asc = Some(asc);
        self
    }

    /// Orders by ascending `id` and records ascending cursor direction.
    pub fn order_by_asc(mut self) -> Self {
        self.loader = self.loader.order_by_id_asc();
        self.order_asc = Some(true);
        self
    }

    /// Orders by descending `id` and records descending cursor direction.
    pub fn order_by_desc(mut self) -> Self {
        self.loader = self.loader.order_by_id_desc();
        self.order_asc = Some(false);
        self
    }

    /// Marks the select distinct and records the distinct flag in options.
    pub fn distinct(mut self) -> Self {
        self.opts.distinct = true;
        self
    }

    /// Applies a filter to the loader via the given closure.
    pub fn filter<F>(mut self, f: F) -> Self
    where
        F: FnOnce(<E::ModelEx as BaseEntity>::LoaderType) -> <E::ModelEx as BaseEntity>::LoaderType,
    {
        self.loader = f(self.loader);
        self
    }

    /// Sets the select limit and records it in options.
    pub fn with_limit(mut self, limit: u64) -> Self {
        self.opts.limit = limit as usize;
        self
    }
}

/// Bulk-delete wrapper around a SeaORM `DeleteMany`.
///
/// `E` is the SeaORM entity being deleted. Cursor filtering for deletes
/// always uses `id > cursor`, regardless of ordering.
pub struct DeleteQueryData<E: EntityTrait> {
    /// The underlying delete being built.
    pub delete: sea_orm::DeleteMany<E>,
}

impl<E: EntityTrait> DeleteQueryData<E> {
    /// Wraps a SeaORM delete-many for repository execution.
    pub fn new(delete: sea_orm::DeleteMany<E>) -> Self {
        Self { delete }
    }

    /// Applies a filter to the delete via the given closure.
    pub fn filter<F>(mut self, f: F) -> Self
    where
        F: FnOnce(sea_orm::DeleteMany<E>) -> sea_orm::DeleteMany<E>,
    {
        self.delete = f(self.delete);
        self
    }
}