b64e2537 Andy McCall 1 month ago ↶
1
2
3
use super :: { now_ms, Pool} ;
4
use sqlx:: FromRow;
5
6
# [ derive ( Clone, Debug, FromRow ) ]
7
pub struct Hook {
8
pub id : i64 ,
9
pub repo_id : Option< i64 > ,
10
pub org_id : Option< i64 > ,
11
pub url : String,
12
pub secret : String,
13
pub events : String,
14
pub active : bool ,
15
pub created_ms : i64 ,
9756c67d Andy McCall 1 month ago ↶
16
17
pub format : String,
b64e2537 Andy McCall 1 month ago ↶
18
}
9756c67d Andy McCall 1 month ago ↶
19
20
21
pub const FORMATS : [ & str ; 3 ] = [ " oxygenhub" , " discord" , " slack" ] ;
b64e2537 Andy McCall 1 month ago ↶
22
9756c67d Andy McCall 1 month ago ↶
23
24
25
pub fn format_for_url ( url : & str ) -> & 'static str {
26
let host = url.split ( " ://" ) .nth ( 1 ) .unwrap_or ( " " ) .split ( [ ' /' , ' :' ] ) .next ( ) .unwrap_or ( " " ) .to_ascii_lowercase ( ) ;
27
if matches! ( host.as_str ( ) , " discord.com" | " discordapp.com" | " canary.discord.com" | " ptb.discord.com" )
28
& & url.contains ( " /api/webhooks/" )
29
{
30
" discord"
31
} else if host = = " hooks.slack.com" {
32
" slack"
33
} else {
34
" oxygenhub"
35
}
36
}
37
38
39
40
pub fn masked_url ( url : & str ) -> String {
41
let ( scheme, rest) = url.split_once ( " ://" ) .unwrap_or ( ( " " , url) ) ;
42
let ( path, query) = rest.split_once ( ' ?' ) .map ( | ( p , q ) | ( p, Some ( q) ) ) .unwrap_or ( ( rest, None ) ) ;
43
let parts: Vec< String> =
44
path.split ( ' /' )
45
.enumerate ( )
46
.map ( | ( i , seg ) | {
47
if i > 0 & & seg.chars ( ) .count ( ) > 14 {
48
format! ( " {} …" , seg.chars ( ) .take ( 4 ) .collect:: < String> ( ) )
49
} else {
50
seg.to_string ( )
51
}
52
} )
53
.collect ( ) ;
54
let mut out = if scheme.is_empty ( ) { parts.join ( " /" ) } else { format! ( " {scheme} ://{} " , parts.join ( " /" ) ) } ;
55
if query.is_some ( ) {
56
out.push_str ( " ?…" ) ;
57
}
58
out
59
}
60
b64e2537 Andy McCall 1 month ago ↶
61
impl Hook {
62
63
pub fn wants ( & self , event : & str ) -> bool {
64
self .active & & self .events.split ( ' ,' ) .map ( str :: trim) .any ( | e | e = = " *" | | event = = e | | event.starts_with ( & format! ( " {e} ." ) ) )
65
}
66
}
67
68
# [ derive ( Clone, Debug, FromRow ) ]
69
pub struct Delivery {
70
pub id : i64 ,
71
pub hook_id : i64 ,
72
pub delivery : String,
73
pub event : String,
74
pub payload : String,
75
pub created_ms : i64 ,
76
pub attempts : i64 ,
77
pub next_ms : i64 ,
78
pub status : Option< i64 > ,
79
pub response : String,
80
pub delivered_ms : Option< i64 > ,
81
}
82
83
pub async fn hooks_for_repo ( pool : & Pool, repo_id : i64 ) -> sqlx:: Result< Vec< Hook> > {
84
sqlx:: query_as( " SELECT * FROM hooks WHERE repo_id = ? ORDER BY id" ) .bind ( repo_id) .fetch_all ( pool) .await
85
}
86
87
pub async fn hooks_for_org ( pool : & Pool, org_id : i64 ) -> sqlx:: Result< Vec< Hook> > {
88
sqlx:: query_as( " SELECT * FROM hooks WHERE org_id = ? ORDER BY id" ) .bind ( org_id) .fetch_all ( pool) .await
89
}
90
91
92
93
pub async fn hooks_hearing ( pool : & Pool, repo_id : i64 , owner : & str ) -> sqlx:: Result< Vec< Hook> > {
94
let mut out = hooks_for_repo ( pool, repo_id) .await? ;
95
if let Some ( org) = super :: accounts:: org_by_name( pool, owner) .await? {
96
out.extend ( hooks_for_org ( pool, org.id) .await? ) ;
97
}
98
Ok ( out)
99
}
100
101
pub async fn hook ( pool : & Pool, id : i64 ) -> sqlx:: Result< Option< Hook> > {
102
sqlx:: query_as( " SELECT * FROM hooks WHERE id = ?" ) .bind ( id) .fetch_optional ( pool) .await
103
}
104
105
pub async fn create_hook (
106
pool : & Pool,
107
repo_id : Option< i64 > ,
108
org_id : Option< i64 > ,
109
url : & str ,
110
secret : & str ,
111
events : & str ,
112
) -> sqlx:: Result< Hook> {
9756c67d Andy McCall 1 month ago ↶
113
sqlx:: query_as( " INSERT INTO hooks (repo_id, org_id, url, secret, events, created_ms, format) VALUES (?, ?, ?, ?, ?, ?, ?) RETURNING *" )
b64e2537 Andy McCall 1 month ago ↶
114
.bind ( repo_id)
115
.bind ( org_id)
116
.bind ( url)
117
.bind ( secret)
118
.bind ( events)
119
.bind ( now_ms ( ) )
9756c67d Andy McCall 1 month ago ↶
120
.bind ( format_for_url ( url) )
b64e2537 Andy McCall 1 month ago ↶
121
.fetch_one ( pool)
9756c67d Andy McCall 1 month ago ↶
122
.await
123
}
124
125
126
pub async fn update_hook (
127
pool : & Pool,
128
id : i64 ,
129
url : Option< & str > ,
130
secret : Option< & str > ,
131
events : & str ,
132
format : & str ,
133
active : bool ,
134
) -> sqlx:: Result< ( ) > {
135
sqlx:: query( " UPDATE hooks SET url = COALESCE(?, url), secret = COALESCE(?, secret), events = ?, format = ?, active = ? WHERE id = ?" )
136
.bind ( url)
137
.bind ( secret)
138
.bind ( events)
139
.bind ( format)
140
.bind ( active)
141
.bind ( id)
142
.execute ( pool)
143
.await? ;
144
Ok ( ( ) )
145
}
146
147
pub async fn set_hook_format ( pool : & Pool, id : i64 , format : & str ) -> sqlx:: Result< ( ) > {
148
sqlx:: query( " UPDATE hooks SET format = ? WHERE id = ?" ) .bind ( format) .bind ( id) .execute ( pool) .await? ;
149
Ok ( ( ) )
150
}
151
152
153
pub async fn last_attempt ( pool : & Pool, hook_id : i64 ) -> sqlx:: Result< Option< Delivery> > {
154
sqlx:: query_as( " SELECT * FROM deliveries WHERE hook_id = ? AND attempts > 0 ORDER BY id DESC LIMIT 1" )
155
.bind ( hook_id)
156
.fetch_optional ( pool)
b64e2537 Andy McCall 1 month ago ↶
157
.await
158
}
159
160
pub async fn set_hook_active ( pool : & Pool, id : i64 , active : bool ) -> sqlx:: Result< ( ) > {
161
sqlx:: query( " UPDATE hooks SET active = ? WHERE id = ?" ) .bind ( active) .bind ( id) .execute ( pool) .await? ;
162
Ok ( ( ) )
163
}
164
165
pub async fn delete_hook ( pool : & Pool, id : i64 ) -> sqlx:: Result< ( ) > {
166
sqlx:: query( " DELETE FROM hooks WHERE id = ?" ) .bind ( id) .execute ( pool) .await? ;
167
Ok ( ( ) )
168
}
169
170
pub async fn queue_delivery ( pool : & Pool, hook_id : i64 , delivery : & str , event : & str , payload : & str ) -> sqlx:: Result< i64 > {
171
let now = now_ms ( ) ;
172
sqlx:: query_scalar(
173
" INSERT INTO deliveries (hook_id, delivery, event, payload, created_ms, next_ms) VALUES (?, ?, ?, ?, ?, ?) RETURNING id" ,
174
)
175
.bind ( hook_id)
176
.bind ( delivery)
177
.bind ( event)
178
.bind ( payload)
179
.bind ( now)
180
.bind ( now)
181
.fetch_one ( pool)
182
.await
183
}
184
185
pub async fn due_deliveries ( pool : & Pool, limit : i64 ) -> sqlx:: Result< Vec< Delivery> > {
186
sqlx:: query_as( " SELECT * FROM deliveries WHERE delivered_ms IS NULL AND next_ms <= ? ORDER BY next_ms LIMIT ?" )
187
.bind ( now_ms ( ) )
188
.bind ( limit)
189
.fetch_all ( pool)
190
.await
191
}
192
193
pub async fn deliveries_for_hook ( pool : & Pool, hook_id : i64 , limit : i64 ) -> sqlx:: Result< Vec< Delivery> > {
194
sqlx:: query_as( " SELECT * FROM deliveries WHERE hook_id = ? ORDER BY created_ms DESC, id DESC LIMIT ?" )
195
.bind ( hook_id)
196
.bind ( limit)
197
.fetch_all ( pool)
198
.await
199
}
200
201
pub async fn delivery ( pool : & Pool, id : i64 ) -> sqlx:: Result< Option< Delivery> > {
202
sqlx:: query_as( " SELECT * FROM deliveries WHERE id = ?" ) .bind ( id) .fetch_optional ( pool) .await
203
}
204
205
206
pub async fn record_attempt ( pool : & Pool, id : i64 , attempts : i64 , status : Option< i64 > , response : & str ) -> sqlx:: Result< ( ) > {
207
let ok = status.is_some_and ( | s | ( 200 .. 300 ) .contains ( & s) ) ;
208
let now = now_ms ( ) ;
209
let delivered: Option< i64 > = if ok { Some ( now) } else { None } ;
210
211
let backoff_ms = [ 60_000 i64 , 300_000 , 1_800_000 , 7_200_000 , 21_600_000 ] ;
212
let next = if ok | | attempts as usize > = backoff_ms.len ( ) { i64 :: MAX / 4 } else { now + backoff_ms[ attempts as usize ] } ;
213
sqlx:: query( " UPDATE deliveries SET attempts = ?, status = ?, response = ?, delivered_ms = ?, next_ms = ? WHERE id = ?" )
214
.bind ( attempts + 1 )
215
.bind ( status)
216
.bind ( & response.chars ( ) .take ( 2000 ) .collect:: < String> ( ) )
217
.bind ( delivered)
218
.bind ( next)
219
.bind ( id)
220
.execute ( pool)
221
.await? ;
222
Ok ( ( ) )
223
}
224
225
226
pub async fn redeliver ( pool : & Pool, id : i64 ) -> sqlx:: Result< ( ) > {
227
sqlx:: query( " UPDATE deliveries SET delivered_ms = NULL, next_ms = ? WHERE id = ?" ) .bind ( now_ms ( ) ) .bind ( id) .execute ( pool) .await? ;
228
Ok ( ( ) )
229
}
230
231
pub async fn prune_deliveries ( pool : & Pool, older_than_ms : i64 ) -> sqlx:: Result< ( ) > {
232
sqlx:: query( " DELETE FROM deliveries WHERE created_ms < ?" ) .bind ( older_than_ms) .execute ( pool) .await? ;
233
Ok ( ( ) )
234
}
235
236
237
238
# [ derive ( Clone, Debug, FromRow ) ]
239
pub struct Check {
240
pub id : i64 ,
241
pub repo_id : i64 ,
242
pub state : String,
243
pub name : String,
244
pub status : String,
245
pub summary : String,
246
pub url : String,
247
pub reporter_id : Option< i64 > ,
248
pub reporter : String,
249
pub seal : String,
250
pub started_ms : i64 ,
251
pub finished_ms : Option< i64 > ,
252
}
253
254
impl Check {
255
pub fn done ( & self ) -> bool {
256
matches! ( self .status.as_str ( ) , " passed" | " failed" | " cancelled" | " skipped" )
257
}
258
}
259
260
pub async fn checks_for_state ( pool : & Pool, repo_id : i64 , state : & str ) -> sqlx:: Result< Vec< Check> > {
261
sqlx:: query_as( " SELECT * FROM checks WHERE repo_id = ? AND state = ? ORDER BY name" ) .bind ( repo_id) .bind ( state) .fetch_all ( pool) .await
262
}
263
264
265
pub async fn upsert_check (
266
pool : & Pool,
267
repo_id : i64 ,
268
state : & str ,
269
name : & str ,
270
status : & str ,
271
summary : & str ,
272
url : & str ,
273
reporter : Option< ( i64 , & str ) > ,
274
seal : & str ,
275
) -> sqlx:: Result< Check> {
276
let now = now_ms ( ) ;
277
let finished: Option< i64 > = if matches! ( status, " passed" | " failed" | " cancelled" | " skipped" ) { Some ( now) } else { None } ;
278
sqlx:: query_as(
279
" INSERT INTO checks (repo_id, state, name, status, summary, url, reporter_id, reporter, seal, started_ms, finished_ms) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?) \
280
ON CONFLICT (repo_id, state, name) DO UPDATE SET status = excluded.status, summary = excluded.summary, url = excluded.url, reporter_id = excluded.reporter_id, reporter = excluded.reporter, seal = CASE WHEN excluded.seal = '' THEN checks.seal ELSE excluded.seal END, finished_ms = excluded.finished_ms RETURNING *" ,
281
)
282
.bind ( repo_id)
283
.bind ( state)
284
.bind ( name)
285
.bind ( status)
286
.bind ( summary)
287
.bind ( url)
288
.bind ( reporter.map ( | r | r.0 ) )
289
.bind ( reporter.map ( | r | r.1 ) .unwrap_or ( " " ) )
290
.bind ( seal)
291
.bind ( now)
292
.bind ( finished)
293
.fetch_one ( pool)
294
.await
295
}
296
297
pub async fn required_checks ( pool : & Pool, repo_id : i64 ) -> sqlx:: Result< Vec< String> > {
298
let s: String = sqlx:: query_scalar( " SELECT required_checks FROM repositories WHERE id = ?" ) .bind ( repo_id) .fetch_one ( pool) .await? ;
299
Ok ( s.split ( ' ,' ) .map ( str :: trim) .filter ( | x | ! x.is_empty ( ) ) .map ( String :: from) .collect ( ) )
300
}
301
302
pub async fn set_required_checks ( pool : & Pool, repo_id : i64 , names : & [String]) -> sqlx:: Result< ( ) > {
303
sqlx:: query( " UPDATE repositories SET required_checks = ? WHERE id = ?" ) .bind ( names.join ( " ," ) ) .bind ( repo_id) .execute ( pool) .await? ;
304
Ok ( ( ) )
305
}