1use crate::domain::{BranchName, EventType, RepoUrl, TargetRepo};
4use crate::model::{
5 Branch, CreateSubscription, Subscription, SubscriptionWithBranch, TriggerQueueItem,
6 UpdateSubscription,
7};
8use crate::repository::{
9 RepositoryError,
10 branch::BranchRepository,
11 subscription::SubscriptionRepository,
12 trigger::{TriggerRepository, UpdateRetryStatus},
13};
14use async_trait::async_trait;
15use futures::future::BoxFuture;
16use sqlx::{SqliteConnection, SqlitePool};
17
18#[derive(Debug)]
19pub struct SqliteRepository {
21 pool: SqlitePool,
23}
24
25impl SqliteRepository {
26 pub fn new(pool: SqlitePool) -> Self {
28 Self { pool }
29 }
30
31 pub fn get_pool(&self) -> &SqlitePool {
33 &self.pool
34 }
35
36 pub async fn run_in_transaction<'a, F, T, E>(&self, f: F) -> Result<T, E>
38 where
39 F: for<'b> FnOnce(&'b mut SqliteConnection) -> BoxFuture<'b, Result<T, E>> + Send + 'a,
40 E: From<sqlx::Error> + Send + 'a,
41 T: Send + 'a,
42 {
43 let mut tx = self.pool.begin().await?;
44 let result = f(&mut tx).await?;
45 tx.commit().await?;
46 Ok(result)
47 }
48}
49
50#[async_trait]
51impl BranchRepository for SqliteRepository {
52 async fn get_all(&self) -> Result<Vec<Branch>, RepositoryError> {
53 sqlx::query_as::<_, Branch>("SELECT * FROM branches")
54 .fetch_all(&self.pool)
55 .await
56 .map_err(RepositoryError::Database)
57 }
58
59 async fn find_by_id(&self, id: i64) -> Result<Option<Branch>, RepositoryError> {
60 sqlx::query_as::<_, Branch>("SELECT * FROM branches WHERE id = ?")
61 .bind(id)
62 .fetch_optional(&self.pool)
63 .await
64 .map_err(RepositoryError::Database)
65 }
66
67 async fn delete_by_id(&self, id: i64) -> Result<(), RepositoryError> {
68 let result = sqlx::query!("DELETE FROM branches WHERE id = ?", id)
69 .execute(&self.pool)
70 .await
71 .map_err(RepositoryError::Database)?;
72
73 if result.rows_affected() == 0 {
74 return Err(RepositoryError::NotFound);
75 }
76 Ok(())
77 }
78
79 async fn update_last_commit_hash(
80 &self,
81 id: i64,
82 hash: &crate::domain::CommitHash,
83 ) -> Result<(), RepositoryError> {
84 sqlx::query!(
85 "UPDATE branches SET last_commit_hash = ?, updated_at = CURRENT_TIMESTAMP WHERE id = ?",
86 hash,
87 id
88 )
89 .execute(&self.pool)
90 .await
91 .map_err(RepositoryError::Database)?;
92 Ok(())
93 }
94
95 async fn update_last_commit_hash_in_tx(
96 &self,
97 id: i64,
98 hash: &crate::domain::CommitHash,
99 tx: &mut sqlx::SqliteConnection,
100 ) -> Result<(), RepositoryError> {
101 sqlx::query!(
102 "UPDATE branches SET last_commit_hash = ?, updated_at = CURRENT_TIMESTAMP WHERE id = ?",
103 hash,
104 id
105 )
106 .execute(tx)
107 .await
108 .map_err(RepositoryError::Database)?;
109 Ok(())
110 }
111}
112
113#[async_trait]
114impl SubscriptionRepository for SqliteRepository {
115 async fn create(
116 &self,
117 subscription_payload: &CreateSubscription,
118 ) -> Result<SubscriptionWithBranch, RepositoryError> {
119 let mut transaction = self.pool.begin().await.map_err(RepositoryError::Database)?;
120
121 let branch_id = sqlx::query_scalar::<_, i64>(
122 "INSERT INTO branches (repo_url, name) VALUES (?, ?) \
123 ON CONFLICT(repo_url, name) DO UPDATE SET repo_url=excluded.repo_url \
124 RETURNING id",
125 )
126 .bind(&subscription_payload.source_repo_url)
127 .bind(&subscription_payload.source_branch_name)
128 .fetch_one(&mut *transaction)
129 .await
130 .map_err(RepositoryError::Database)?;
131
132 let subscription = sqlx::query_as::<_, Subscription>(
133 "INSERT INTO subscriptions (branch_id, target_repo, event_type, gh_app_installation_id) VALUES (?, ?, ?, ?) RETURNING *",
134 )
135 .bind(branch_id)
136 .bind(&subscription_payload.target_repo)
137 .bind(&subscription_payload.event_type)
138 .bind(subscription_payload.gh_app_installation_id)
139 .fetch_one(&mut *transaction)
140 .await
141 .map_err(RepositoryError::Database)?;
142
143 transaction
144 .commit()
145 .await
146 .map_err(RepositoryError::Database)?;
147
148 Ok(SubscriptionWithBranch {
149 subscription,
150 source_branch: crate::model::SourceBranchInfo {
151 repo_url: subscription_payload.source_repo_url.clone(),
152 name: subscription_payload.source_branch_name.clone(),
153 },
154 })
155 }
156
157 async fn get_by_id(&self, id: i64) -> Result<Option<Subscription>, RepositoryError> {
158 sqlx::query_as::<_, Subscription>("SELECT * FROM subscriptions WHERE id = ?")
159 .bind(id)
160 .fetch_optional(&self.pool)
161 .await
162 .map_err(RepositoryError::Database)
163 }
164
165 async fn get_by_id_with_branch(
166 &self,
167 id: i64,
168 ) -> Result<Option<SubscriptionWithBranch>, RepositoryError> {
169 let row = sqlx::query!(
170 "SELECT s.*, b.repo_url as branch_repo_url, b.name as branch_name \
171 FROM subscriptions s \
172 JOIN branches b ON s.branch_id = b.id \
173 WHERE s.id = ?",
174 id
175 )
176 .fetch_optional(&self.pool)
177 .await
178 .map_err(RepositoryError::Database)?;
179
180 match row {
181 Some(row) => Ok(Some(SubscriptionWithBranch {
182 subscription: Subscription {
183 id: row.id,
184 branch_id: row.branch_id,
185 target_repo: crate::domain::TargetRepo::new(row.target_repo)
186 .map_err(|e| RepositoryError::Mapping(e.to_string()))?,
187 event_type: crate::domain::EventType::new(row.event_type)
188 .map_err(|e| RepositoryError::Mapping(e.to_string()))?,
189 gh_app_installation_id: row.gh_app_installation_id,
190 created_at: row.created_at.and_utc(),
191 updated_at: row.updated_at.and_utc(),
192 },
193 source_branch: crate::model::SourceBranchInfo {
194 repo_url: crate::domain::RepoUrl::new(row.branch_repo_url)
195 .map_err(|e| RepositoryError::Mapping(e.to_string()))?,
196 name: crate::domain::BranchName::new(row.branch_name)
197 .map_err(|e| RepositoryError::Mapping(e.to_string()))?,
198 },
199 })),
200 None => Ok(None),
201 }
202 }
203
204 async fn get_by_keys_with_branch(
205 &self,
206 branch_id: i64,
207 target_repo: &TargetRepo,
208 event_type: &EventType,
209 ) -> Result<Option<SubscriptionWithBranch>, RepositoryError> {
210 let row = sqlx::query!(
211 "SELECT s.*, b.repo_url as branch_repo_url, b.name as branch_name \
212 FROM subscriptions s \
213 JOIN branches b ON s.branch_id = b.id \
214 WHERE s.branch_id = ? AND s.target_repo = ? AND s.event_type = ?",
215 branch_id,
216 target_repo,
217 event_type
218 )
219 .fetch_optional(&self.pool)
220 .await
221 .map_err(RepositoryError::Database)?;
222
223 match row {
224 Some(row) => Ok(Some(SubscriptionWithBranch {
225 subscription: Subscription {
226 id: row.id,
227 branch_id: row.branch_id,
228 target_repo: crate::domain::TargetRepo::new(row.target_repo)
229 .map_err(|e| RepositoryError::Mapping(e.to_string()))?,
230 event_type: crate::domain::EventType::new(row.event_type)
231 .map_err(|e| RepositoryError::Mapping(e.to_string()))?,
232 gh_app_installation_id: row.gh_app_installation_id,
233 created_at: row.created_at.and_utc(),
234 updated_at: row.updated_at.and_utc(),
235 },
236 source_branch: crate::model::SourceBranchInfo {
237 repo_url: crate::domain::RepoUrl::new(row.branch_repo_url)
238 .map_err(|e| RepositoryError::Mapping(e.to_string()))?,
239 name: crate::domain::BranchName::new(row.branch_name)
240 .map_err(|e| RepositoryError::Mapping(e.to_string()))?,
241 },
242 })),
243 None => Ok(None),
244 }
245 }
246
247 async fn list_paginated(
248 &self,
249 last_id: i64,
250 limit: i64,
251 ) -> Result<Vec<Subscription>, RepositoryError> {
252 sqlx::query_as::<_, Subscription>(
253 "SELECT * FROM subscriptions WHERE id > ? ORDER BY id ASC LIMIT ?",
254 )
255 .bind(last_id)
256 .bind(limit)
257 .fetch_all(&self.pool)
258 .await
259 .map_err(RepositoryError::Database)
260 }
261
262 async fn list_paginated_with_branches(
263 &self,
264 last_id: i64,
265 limit: i64,
266 ) -> Result<Vec<SubscriptionWithBranch>, RepositoryError> {
267 let rows = sqlx::query!(
268 "SELECT s.*, b.repo_url as branch_repo_url, b.name as branch_name \
269 FROM subscriptions s \
270 JOIN branches b ON s.branch_id = b.id \
271 WHERE s.id > ? ORDER BY s.id ASC LIMIT ?",
272 last_id,
273 limit
274 )
275 .fetch_all(&self.pool)
276 .await
277 .map_err(RepositoryError::Database)?;
278
279 let subscriptions: Result<Vec<SubscriptionWithBranch>, RepositoryError> = rows
280 .into_iter()
281 .map(|row| {
282 Ok(SubscriptionWithBranch {
283 subscription: Subscription {
284 id: row.id,
285 branch_id: row.branch_id,
286 target_repo: TargetRepo::new(row.target_repo)
287 .map_err(|e| RepositoryError::Mapping(e.to_string()))?,
288 event_type: EventType::new(row.event_type)
289 .map_err(|e| RepositoryError::Mapping(e.to_string()))?,
290 gh_app_installation_id: row.gh_app_installation_id,
291 created_at: row.created_at.and_utc(),
292 updated_at: row.updated_at.and_utc(),
293 },
294 source_branch: crate::model::SourceBranchInfo {
295 repo_url: RepoUrl::new(row.branch_repo_url)
296 .map_err(|e| RepositoryError::Mapping(e.to_string()))?,
297 name: BranchName::new(row.branch_name)
298 .map_err(|e| RepositoryError::Mapping(e.to_string()))?,
299 },
300 })
301 })
302 .collect();
303 subscriptions
304 }
305
306 async fn count_remaining(&self, last_id: i64) -> Result<i64, RepositoryError> {
307 sqlx::query_scalar!("SELECT COUNT(*) FROM subscriptions WHERE id > ?", last_id)
308 .fetch_one(&self.pool)
309 .await
310 .map_err(RepositoryError::Database)
311 }
312
313 async fn update(
314 &self,
315 id: i64,
316 subscription: &UpdateSubscription,
317 ) -> Result<Subscription, RepositoryError> {
318 let mut query_builder = sqlx::QueryBuilder::new("UPDATE subscriptions SET ");
319 let mut separated = query_builder.separated(", ");
320
321 if let Some(target_repo) = &subscription.target_repo {
322 separated
323 .push("target_repo = ")
324 .push_bind_unseparated(target_repo);
325 }
326 if let Some(event_type) = &subscription.event_type {
327 separated
328 .push("event_type = ")
329 .push_bind_unseparated(event_type);
330 }
331 if let Some(gh_app_installation_id) = subscription.gh_app_installation_id {
332 separated
333 .push("gh_app_installation_id = ")
334 .push_bind_unseparated(gh_app_installation_id);
335 }
336
337 separated.push("updated_at = CURRENT_TIMESTAMP");
338
339 query_builder.push(" WHERE id = ");
340 query_builder.push_bind(id);
341 query_builder.push(" RETURNING *");
342
343 query_builder
344 .build_query_as::<Subscription>()
345 .fetch_optional(&self.pool)
346 .await
347 .map_err(RepositoryError::Database)?
348 .ok_or(RepositoryError::NotFound)
349 }
350
351 async fn delete(&self, id: i64) -> Result<(), RepositoryError> {
352 let result = sqlx::query!("DELETE FROM subscriptions WHERE id = ?", id)
353 .execute(&self.pool)
354 .await
355 .map_err(RepositoryError::Database)?;
356
357 if result.rows_affected() == 0 {
358 return Err(RepositoryError::NotFound);
359 }
360 Ok(())
361 }
362
363 async fn get_branch_id_by_subscription_id(
364 &self,
365 id: i64,
366 ) -> Result<Option<i64>, RepositoryError> {
367 sqlx::query_scalar!("SELECT branch_id FROM subscriptions WHERE id = ?", id)
368 .fetch_optional(&self.pool)
369 .await
370 .map_err(RepositoryError::Database)
371 }
372
373 async fn count_subscriptions_by_branch_id(
374 &self,
375 branch_id: i64,
376 ) -> Result<i64, RepositoryError> {
377 sqlx::query_scalar!(
378 "SELECT COUNT(*) FROM subscriptions WHERE branch_id = ?",
379 branch_id
380 )
381 .fetch_one(&self.pool)
382 .await
383 .map_err(RepositoryError::Database)
384 }
385
386 async fn delete_subscription_and_cascade(&self, id: i64) -> Result<(), RepositoryError> {
387 self.run_in_transaction(|tx| {
388 Box::pin(async move {
389 let branch_id = sqlx::query_scalar!(
390 "DELETE FROM subscriptions WHERE id = ? RETURNING branch_id",
391 id
392 )
393 .fetch_optional(&mut *tx)
394 .await
395 .map_err(RepositoryError::Database)?
396 .ok_or(RepositoryError::NotFound)?;
397
398 let remaining_subscriptions = sqlx::query_scalar!(
399 "SELECT COUNT(*) FROM subscriptions WHERE branch_id = ?",
400 branch_id
401 )
402 .fetch_one(&mut *tx)
403 .await
404 .map_err(RepositoryError::Database)?;
405
406 if remaining_subscriptions == 0 {
407 sqlx::query!("DELETE FROM branches WHERE id = ?", branch_id)
408 .execute(&mut *tx)
409 .await
410 .map_err(RepositoryError::Database)?;
411 }
412
413 Ok(())
414 })
415 })
416 .await
417 }
418}
419
420#[async_trait]
421impl TriggerRepository for SqliteRepository {
422 async fn get_all(&self) -> Result<Vec<TriggerQueueItem>, RepositoryError> {
423 sqlx::query_as::<_, TriggerQueueItem>("SELECT * FROM trigger_queue")
424 .fetch_all(&self.pool)
425 .await
426 .map_err(RepositoryError::Database)
427 }
428
429 async fn delete_by_id(&self, id: i64) -> Result<(), RepositoryError> {
430 sqlx::query!("DELETE FROM trigger_queue WHERE id = ?", id)
431 .execute(&self.pool)
432 .await
433 .map_err(RepositoryError::Database)?;
434 Ok(())
435 }
436
437 async fn find_oldest_pending_and_mark_processing(
438 &self,
439 ) -> Result<Option<TriggerQueueItem>, RepositoryError> {
440 let trigger = sqlx::query_as::<_, TriggerQueueItem>(
441 "UPDATE trigger_queue
442 SET status = 'PROCESSING', status_updated_at = CURRENT_TIMESTAMP
443 WHERE id = (
444 SELECT id FROM trigger_queue
445 WHERE status IN ('PENDING') AND next_retry_at <= CURRENT_TIMESTAMP
446 ORDER BY next_retry_at ASC LIMIT 1
447 )
448 RETURNING id, branch_id, new_hash, retry_count, target_repo, event_type, gh_app_installation_id",
449 )
450 .fetch_optional(&self.pool)
451 .await
452 .map_err(RepositoryError::Database)?;
453
454 Ok(trigger)
455 }
456
457 async fn update_retry_status(&self, params: UpdateRetryStatus) -> Result<(), RepositoryError> {
458 let next_retry_count = params.retry_count + 1;
459
460 if next_retry_count as u32 >= params.max_attempts {
461 sqlx::query!(
462 "UPDATE trigger_queue SET status = 'FAILED', retry_count = ? WHERE id = ?",
463 next_retry_count,
464 params.id
465 )
466 .execute(&self.pool)
467 .await
468 .map_err(RepositoryError::Database)?;
469 } else {
470 let backoff_secs = (params.backoff_base_secs * (1 << (next_retry_count - 1))) as i64;
471 sqlx::query!(
472 "UPDATE trigger_queue SET status = 'PENDING', retry_count = ?, next_retry_at = datetime('now', ? || ' seconds') WHERE id = ?",
473 next_retry_count,
474 backoff_secs,
475 params.id
476 )
477 .execute(&self.pool)
478 .await
479 .map_err(RepositoryError::Database)?;
480 }
481 Ok(())
482 }
483
484 async fn recover_stuck_tasks(&self, threshold_seconds: u64) -> Result<(), RepositoryError> {
485 let threshold_str = format!("-{} seconds", threshold_seconds);
486
487 sqlx::query!(
488 "UPDATE trigger_queue
489 SET status = 'PENDING', status_updated_at = CURRENT_TIMESTAMP
490 WHERE status = 'PROCESSING'
491 AND status_updated_at < DATETIME('now', ?)",
492 threshold_str
493 )
494 .execute(&self.pool)
495 .await
496 .map_err(RepositoryError::Database)?;
497 Ok(())
498 }
499
500 async fn queue_triggers_for_branch(
501 &self,
502 branch_id: i64,
503 new_hash: &crate::domain::CommitHash,
504 executor: &mut sqlx::SqliteConnection,
505 ) -> Result<(), RepositoryError> {
506 sqlx::query!(
507 "INSERT INTO trigger_queue (branch_id, new_hash, target_repo, event_type, gh_app_installation_id)
508 SELECT ?, ?, s.target_repo, s.event_type, s.gh_app_installation_id
509 FROM subscriptions s
510 WHERE s.branch_id = ?
511 ON CONFLICT(branch_id, target_repo, event_type) WHERE status = 'PENDING'
512 DO UPDATE SET new_hash = excluded.new_hash, status_updated_at = CURRENT_TIMESTAMP",
513 branch_id,
514 new_hash,
515 branch_id
516 )
517 .execute(executor)
518 .await
519 .map_err(RepositoryError::Database)?;
520 Ok(())
521 }
522}