reliar-store-postgres 0.9.0

PostgreSQL provider for the Reliar transactional outbox and inbox: migrations, migrate(), enqueue and the SKIP LOCKED claim.
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

//! `InboxDeadLetters`'s three statements: `list_dead`/`retry_dead`/`purge_dead`.

use super::rows::InboxRow;

/// [`list_dead_rows`]'s bind parameters.
pub(in crate::inbox) struct ListDeadRowsParams<'a> {
    pub(in crate::inbox) scope: Option<&'a str>,

    pub(in crate::inbox) message_type: Option<&'a str>,

    pub(in crate::inbox) dead_before: Option<time::OffsetDateTime>,

    pub(in crate::inbox) after_dead_at: Option<time::OffsetDateTime>,

    pub(in crate::inbox) after_id: Option<uuid::Uuid>,

    pub(in crate::inbox) limit: i64,
}

/// `list_dead`'s query. `ORDER BY dead_at, id` is normative: database-authored death time is the
/// keyset's leading column, and the unique row id breaks ties.
pub(in crate::inbox) async fn list_dead_rows<'e>(
    executor: impl sqlx::PgExecutor<'e>,
    params: ListDeadRowsParams<'_>,
) -> Result<Vec<InboxRow>, sqlx::Error> {
    sqlx::query_as!(
        InboxRow,
        r#"SELECT id, scope, message_id, message_type, message_version, conversation_id,
                  correlation_id, causation_id, received_at, updated_at, completed_at, dead_at,
                  attempts, last_error
             FROM inbox
            WHERE dead_at IS NOT NULL
              AND ($1::text IS NULL OR scope = $1)
              AND ($2::text IS NULL OR message_type = $2)
              AND ($3::timestamptz IS NULL OR dead_at < $3)
              AND ($4::timestamptz IS NULL OR (dead_at, id) > ($4, $5::uuid))
            ORDER BY dead_at, id
            LIMIT $6"#,
        params.scope,
        params.message_type,
        params.dead_before,
        params.after_dead_at,
        params.after_id,
        params.limit,
    )
    .fetch_all(executor)
    .await
}

/// [`retry_dead_rows`]'s bind parameters.
pub(in crate::inbox) struct RetryDeadRowsParams<'a> {
    pub(in crate::inbox) ids: &'a [uuid::Uuid],
}

pub(in crate::inbox) async fn retry_dead_rows<'e>(
    executor: impl sqlx::PgExecutor<'e>,
    params: RetryDeadRowsParams<'_>,
) -> Result<u64, sqlx::Error> {
    let result = sqlx::query!(
        r#"UPDATE inbox SET dead_at = NULL, attempts = 0, updated_at = now()
            WHERE id = ANY($1) AND dead_at IS NOT NULL"#,
        params.ids,
    )
    .execute(executor)
    .await?;

    Ok(result.rows_affected())
}

/// [`purge_dead_rows`]'s bind parameters.
pub(in crate::inbox) struct PurgeDeadRowsParams<'a> {
    pub(in crate::inbox) ids: &'a [uuid::Uuid],
}

pub(in crate::inbox) async fn purge_dead_rows<'e>(
    executor: impl sqlx::PgExecutor<'e>,
    params: PurgeDeadRowsParams<'_>,
) -> Result<u64, sqlx::Error> {
    let result = sqlx::query!(
        r#"DELETE FROM inbox WHERE id = ANY($1) AND dead_at IS NOT NULL"#,
        params.ids,
    )
    .execute(executor)
    .await?;

    Ok(result.rows_affected())
}