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
use crate::models::active_backend_modules::SqlAppModule;
use futures::future::join_all;
use myc_core::domain::dtos::webhook::{
WebHookExecutionStatus, WebHookRetryPolicy,
};
use myc_core::domain::entities::WebHookUpdating;
use myc_core::models::CoreConfig;
use myc_core::{
domain::entities::{EncryptionKeyFetching, WebHookFetching},
use_cases::dispatch_webhooks,
};
use mycelium_base::entities::FetchManyResponseKind;
use rand::Rng;
use shaku::HasComponent;
use std::collections::HashMap;
use std::sync::Arc;
use std::time::Duration;
/// Dispatch webhooks
///
/// Spawns a new thread to consume messages from the webhook queue.
///
#[tracing::instrument(name = "webhook_dispatcher", skip_all)]
pub(crate) async fn webhook_dispatcher(
config: CoreConfig,
app_modules: Arc<SqlAppModule>,
) {
tokio::spawn(async move {
tracing::info!("Starting webhook dispatcher");
let webhook_config = config.webhook.clone();
let read_repo: &dyn WebHookFetching = app_modules.resolve_ref();
let write_repo: &dyn WebHookUpdating = app_modules.resolve_ref();
let enc_key_repo: &dyn EncryptionKeyFetching =
app_modules.resolve_ref();
let child_read_repo = Box::new(read_repo);
let child_write_repo = Box::new(write_repo);
let child_enc_key_repo = Box::new(enc_key_repo);
let mut interval =
actix_rt::time::interval(Duration::from_secs(match webhook_config
.consume_interval_in_secs
.async_get_or_error()
.await
{
Ok(interval) => interval,
Err(err) => {
panic!("Error on get consume interval: {err}");
}
}));
//
// Skip the first tick to avoid fetching events that were created in the
// same second as the dispatcher start.
//
interval.tick().await;
//
// Wait for a random time between 1 and the consume interval. This only
// staggers this dispatcher against the email one so they do not both
// wake on the same second; it is NOT what keeps two replicas off the
// same event. That is the repository's claim (`FOR UPDATE SKIP
// LOCKED`), which jitter alone never provided.
//
let random_time =
rand::thread_rng().gen_range(1..=interval.period().as_secs());
tokio::time::sleep(Duration::from_secs(random_time)).await;
loop {
interval.tick().await;
//
// Fetch webhook dispatch events
//
let retry_policy = WebHookRetryPolicy::new(
webhook_config
.retry_base_in_secs
.async_get_or_error()
.await
.unwrap_or(30),
webhook_config
.retry_cap_in_secs
.async_get_or_error()
.await
.unwrap_or(3600),
webhook_config
.visibility_timeout_in_secs
.async_get_or_error()
.await
.unwrap_or(900),
);
//
// `Processing` is deliberately absent from this filter. A row a
// live pod is working on must not be handed out again, and one left
// behind by a pod that died is picked up by the repository's own
// stale-claim branch, which keys off the lease clock rather than
// off this list.
//
let events_response = match read_repo
.fetch_execution_event(
webhook_config
.consume_batch_size
.async_get_or_error()
.await
.unwrap_or(10) as u32,
webhook_config
.max_attempts
.async_get_or_error()
.await
.unwrap_or(3) as u32,
Some(vec![
WebHookExecutionStatus::Pending,
WebHookExecutionStatus::Failed,
]),
retry_policy,
)
.await
{
Ok(events) => events,
Err(err) => {
tracing::error!("Error on fetch execution event: {err}");
continue;
}
};
let events = match events_response {
FetchManyResponseKind::NotFound => {
continue;
}
FetchManyResponseKind::Found(events) => events,
FetchManyResponseKind::FoundPaginated { records, .. } => {
records
}
};
//
// Fold events by trigger
//
let events_by_trigger =
events.into_iter().fold(HashMap::new(), |mut acc, event| {
let id = event.id.unwrap_or_else(|| {
panic!("Webhook artifact id is required");
});
acc.entry((event.trigger.clone(), id))
.or_insert_with(Vec::new)
.push(event);
acc
});
if events_by_trigger.is_empty() {
continue;
}
//
// Dispatch webhooks
//
for ((trigger, id), artifacts) in events_by_trigger {
tracing::info!(
"Dispatch webhooks for trigger {trigger} and id {id}: {artifacts}",
trigger = trigger,
id = id,
artifacts = artifacts.len()
);
let dispatching_events =
join_all(artifacts.into_iter().map(|artifact| {
dispatch_webhooks(
trigger.to_owned(),
artifact,
config.clone(),
child_read_repo.clone(),
child_write_repo.clone(),
child_enc_key_repo.clone(),
)
}))
.await;
for event in dispatching_events {
if let Err(err) = event {
tracing::error!("Error on dispatch webhook: {err}");
}
}
}
}
});
}