mycelium-notifier 7.7.2

Provide message adapters to the mycelium project.
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 chrono::Local;
use myc_core::domain::dtos::message::{MessageSendingEvent, MessageStatus};
use myc_core::domain::entities::{
    LocalMessageReading, LocalMessageWrite, RemoteMessageWrite,
};
use mycelium_base::entities::FetchManyResponseKind;
use mycelium_base::{
    entities::CreateResponseKind,
    utils::errors::{creation_err, MappedErrors},
};
use std::sync::Arc;
use uuid::Uuid;

/// Consumes messages from the message queue
///
/// This function consumes messages from the message queue sending by smtp.
#[tracing::instrument(
    name = "consume_messages",
    skip(
        local_message_read_repo,
        local_message_write_repo,
        remote_message_write_repo
    )
)]
pub async fn consume_messages(
    queue_name: String,
    local_message_read_repo: Arc<dyn LocalMessageReading>,
    local_message_write_repo: Arc<dyn LocalMessageWrite>,
    remote_message_write_repo: Arc<dyn RemoteMessageWrite>,
) -> Result<(i32, i32), MappedErrors> {
    let max_retries = 3;
    let mut retries = 0;
    let mut processed_messages_success = 0;
    let mut processed_messages_failed = vec![];

    //
    // Consume the queue up to the end
    //
    loop {
        //
        // Update retries counter and check if the maximum
        // number of retries was reached
        //
        retries += 1;

        if retries >= max_retries {
            break;
        }

        let events = match local_message_read_repo
            .list_oldest_messages(25, MessageStatus::Queued)
            .await?
        {
            FetchManyResponseKind::NotFound => break,
            FetchManyResponseKind::Found(messages) => messages,
            FetchManyResponseKind::FoundPaginated {
                records, count, ..
            } => {
                if count == 0 {
                    break;
                }

                records
            }
        };

        for event in events {
            let mut _message = event.to_owned();

            if let Err(err) =
                process_record(event, remote_message_write_repo.clone()).await
            {
                _message.attempted = Some(Local::now());
                _message.error = Some(err.to_string());
                _message.attempts += 1;

                if _message.attempts >= 5 {
                    _message.status = MessageStatus::Failed;
                }

                processed_messages_failed.push(_message.id);

                if let Err(err) = local_message_write_repo
                    .update_message_event(_message)
                    .await
                {
                    panic!("Failed to update message: {err}");
                }
            } else {
                if !processed_messages_failed.contains(&_message.id) {
                    processed_messages_failed = processed_messages_failed
                        .into_iter()
                        .filter(|id| id != &_message.id)
                        .collect();
                }

                processed_messages_success += 1;

                if let Err(err) = local_message_write_repo
                    .delete_message_event(_message.id)
                    .await
                {
                    panic!("Failed to delete message: {err}");
                }
            }
        }

        //
        // If there are failed messages, do not break the loop, try again
        //
        if !processed_messages_failed.is_empty() {
            continue;
        }

        //
        // If there are no failed messages, break the loop
        //
        break;
    }

    Ok((
        processed_messages_success,
        processed_messages_failed.len() as i32,
    ))
}

async fn process_record(
    record: MessageSendingEvent,
    message_sending_repo: Arc<dyn RemoteMessageWrite>,
) -> Result<Uuid, MappedErrors> {
    if let CreateResponseKind::NotCreated(_, _) =
        message_sending_repo.send(record.message).await?
    {
        return creation_err("Failed to send message")
            .with_exp_true()
            .as_error();
    }

    Ok(record.id)
}