graphile_worker_utils 0.1.1

Worker utility helpers for graphile_worker
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

use graphile_worker_database::{DbExecutor, DbValue};
use indoc::formatdoc;

use super::client::WorkerUtils;
use super::types::RescheduleJobOptions;
use graphile_worker_job::DbJob;
use graphile_worker_queries::errors::GraphileWorkerError;
use graphile_worker_queries::schema_names::WorkerFunction;

pub(super) async fn remove_job(
    utils: &WorkerUtils,
    job_key: &str,
) -> Result<(), GraphileWorkerError> {
    let remove_job = WorkerFunction::RemoveJob.qualified(&utils.schema);
    let sql = formatdoc!(
        r#"
            select * from {remove_job}($1::text);
        "#
    );

    utils
        .database
        .execute(&sql, vec![DbValue::Text(job_key.to_string())].into())
        .await?;

    Ok(())
}

pub(super) async fn complete_jobs(
    utils: &WorkerUtils,
    ids: &[i64],
) -> Result<Vec<DbJob>, GraphileWorkerError> {
    let complete_jobs = WorkerFunction::CompleteJobs.qualified(&utils.schema);
    let sql = formatdoc!(
        r#"
            select * from {complete_jobs}($1::bigint[]);
        "#
    );

    fetch_db_jobs(utils, &sql, vec![DbValue::I64Array(ids.to_vec())]).await
}

pub(super) async fn permanently_fail_jobs(
    utils: &WorkerUtils,
    ids: &[i64],
    reason: &str,
) -> Result<Vec<DbJob>, GraphileWorkerError> {
    let permanently_fail_jobs = WorkerFunction::PermanentlyFailJobs.qualified(&utils.schema);
    let sql = formatdoc!(
        r#"
            select * from {permanently_fail_jobs}($1::bigint[], $2::text);
        "#
    );

    fetch_db_jobs(
        utils,
        &sql,
        vec![
            DbValue::I64Array(ids.to_vec()),
            DbValue::Text(reason.to_string()),
        ],
    )
    .await
}

pub(super) async fn reschedule_jobs(
    utils: &WorkerUtils,
    ids: &[i64],
    options: RescheduleJobOptions,
) -> Result<Vec<DbJob>, GraphileWorkerError> {
    let reschedule_jobs = WorkerFunction::RescheduleJobs.qualified(&utils.schema);
    let sql = formatdoc!(
        r#"
            select * from {reschedule_jobs}(
                $1::bigint[],
                run_at := $2::timestamptz,
                priority := $3::int,
                attempts := $4::int,
                max_attempts := $5::int
            );
        "#
    );

    fetch_db_jobs(
        utils,
        &sql,
        vec![
            DbValue::I64Array(ids.to_vec()),
            DbValue::TimestampTzOpt(options.run_at),
            DbValue::I32Opt(options.priority.map(i32::from)),
            DbValue::I32Opt(options.attempts.map(i32::from)),
            DbValue::I32Opt(options.max_attempts.map(i32::from)),
        ],
    )
    .await
}

pub(super) async fn force_unlock_workers(
    utils: &WorkerUtils,
    worker_ids: &[&str],
) -> Result<(), GraphileWorkerError> {
    let force_unlock_workers = WorkerFunction::ForceUnlockWorkers.qualified(&utils.schema);
    let sql = formatdoc!(
        r#"
            select * from {force_unlock_workers}($1::text[]);
        "#
    );

    utils
        .database
        .execute(
            &sql,
            vec![DbValue::TextArray(
                worker_ids
                    .iter()
                    .map(|worker_id| worker_id.to_string())
                    .collect(),
            )]
            .into(),
        )
        .await?;

    Ok(())
}

async fn fetch_db_jobs(
    utils: &WorkerUtils,
    sql: &str,
    params: Vec<DbValue>,
) -> Result<Vec<DbJob>, GraphileWorkerError> {
    let jobs = utils
        .database
        .fetch_all(sql, params.into())
        .await?
        .iter()
        .map(graphile_worker_queries::rows::db_job_from_row)
        .collect::<std::result::Result<Vec<_>, _>>()?;

    Ok(jobs)
}