Créer un worker asynchrone de bout en bout (capstone)¶

crates/gitrust-core/src/migrations/m20260501_000023_create_notify_jobs.rs

use sea_orm_migration::prelude::*;

#[derive(DeriveMigrationName)]
pub struct Migration;

#[async_trait::async_trait]
impl MigrationTrait for Migration {
    async fn up(&self, manager: &SchemaManager) -> Result<(), DbErr> {
        manager
            .create_table(
                Table::create()
                    .table(NotifyJobs::Table)
                    .if_not_exists()
                    .col(ColumnDef::new(NotifyJobs::Id).uuid().not_null().primary_key())
                    .col(ColumnDef::new(NotifyJobs::OwnerId).uuid().not_null())
                    .col(ColumnDef::new(NotifyJobs::TargetUrl).text().not_null())
                    .col(ColumnDef::new(NotifyJobs::Payload).text().not_null())
                    .col(
                        ColumnDef::new(NotifyJobs::Status)
                            .string()
                            .not_null()
                            .default("pending"),
                    )
                    .col(ColumnDef::new(NotifyJobs::ErrorMessage).text().null())
                    .col(ColumnDef::new(NotifyJobs::StartedAt).timestamp_with_time_zone().null())
                    .col(ColumnDef::new(NotifyJobs::FinishedAt).timestamp_with_time_zone().null())
                    .col(
                        ColumnDef::new(NotifyJobs::CreatedAt)
                            .timestamp_with_time_zone()
                            .not_null(),
                    )
                    .col(
                        ColumnDef::new(NotifyJobs::UpdatedAt)
                            .timestamp_with_time_zone()
                            .not_null(),
                    )
                    .to_owned(),
            )
            .await?;

        manager
            .create_index(
                Index::create()
                    .table(NotifyJobs::Table)
                    .name("notify_jobs_owner_status_idx")
                    .col(NotifyJobs::OwnerId)
                    .col(NotifyJobs::Status)
                    .to_owned(),
            )
            .await
    }

    async fn down(&self, manager: &SchemaManager) -> Result<(), DbErr> {
        manager
            .drop_table(Table::drop().table(NotifyJobs::Table).to_owned())
            .await
    }
}

#[derive(DeriveIden)]
enum NotifyJobs {
    Table,
    Id,
    OwnerId,
    TargetUrl,
    Payload,
    Status,
    ErrorMessage,
    StartedAt,
    FinishedAt,
    CreatedAt,
    UpdatedAt,
}

#[tokio::test]
async fn migration_creates_notify_jobs_table() {
    let db = setup_test_db().await;
    // Si la migration s'est appliquée, la requête ne renvoie pas d'erreur.
    let count: i64 = db
        .query_one(Statement::from_string(
            DatabaseBackend::Postgres,
            "SELECT COUNT(*) FROM notify_jobs".to_owned(),
        ))
        .await
        .expect("La table notify_jobs doit exister")
        .unwrap()
        .try_get("", "count")
        .unwrap();
    assert_eq!(count, 0);
}

use serde::{Deserialize, Serialize};

/// Statut d'un job de notification — state machine à 5 états.
#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
#[serde(rename_all = "snake_case")]
pub enum NotifyStatus {
    Pending,
    Running,
    Success,
    Failed,
    Cancelled,
}

impl NotifyStatus {
    pub fn as_str(&self) -> &'static str {
        match self {
            Self::Pending => "pending",
            Self::Running => "running",
            Self::Success => "success",
            Self::Failed => "failed",
            Self::Cancelled => "cancelled",
        }
    }

    pub fn from_str(s: &str) -> Option<Self> {
        match s {
            "pending" => Some(Self::Pending),
            "running" => Some(Self::Running),
            "success" => Some(Self::Success),
            "failed" => Some(Self::Failed),
            "cancelled" => Some(Self::Cancelled),
            _ => None,
        }
    }
}

/// Payload envoyé via SSE au client.
#[derive(Debug, Serialize)]
pub struct NotifyJobSse {
    pub status: String,
    pub error_message: Option<String>,
}

/// Input pour créer un job.
#[derive(Debug, Deserialize)]
pub struct CreateNotifyJobInput {
    pub target_url: String,
    pub payload: String,
}

#[cfg(test)]
mod tests {
    use super::*;

    #[test]
    fn status_round_trip() {
        for (s, expected) in [
            ("pending", NotifyStatus::Pending),
            ("running", NotifyStatus::Running),
            ("success", NotifyStatus::Success),
            ("failed", NotifyStatus::Failed),
            ("cancelled", NotifyStatus::Cancelled),
        ] {
            let status = NotifyStatus::from_str(s).expect("doit parser");
            assert_eq!(status, expected);
            assert_eq!(status.as_str(), s);
        }
    }

    #[test]
    fn unknown_status_returns_none() {
        assert!(NotifyStatus::from_str("unknown").is_none());
    }
}

use chrono::Utc;
use sea_orm::{ActiveModelTrait, ActiveValue::Set, DatabaseConnection, EntityTrait};
use uuid::Uuid;

use crate::{
    dto::notify_dto::{CreateNotifyJobInput, NotifyStatus},
    error::GitrustError,
    models::notify_job,
};

pub struct NotifyService;

impl NotifyService {
    /// Crée un job en statut `pending` et retourne son ID.
    pub async fn create_job(
        db: &DatabaseConnection,
        owner_id: Uuid,
        input: CreateNotifyJobInput,
    ) -> Result<Uuid, GitrustError> {
        let id = Uuid::new_v4();
        let now = Utc::now();
        let model = notify_job::ActiveModel {
            id: Set(id),
            owner_id: Set(owner_id),
            target_url: Set(input.target_url),
            payload: Set(input.payload),
            status: Set(NotifyStatus::Pending.as_str().to_owned()),
            error_message: Set(None),
            started_at: Set(None),
            finished_at: Set(None),
            created_at: Set(now),
            updated_at: Set(now),
        };
        model.insert(db).await.map_err(GitrustError::Database)?;
        Ok(id)
    }

    /// Passe le job en `running`.
    pub async fn mark_running(db: &DatabaseConnection, id: Uuid) -> Result<(), GitrustError> {
        Self::update_status(db, id, NotifyStatus::Running, None).await
    }

    /// Passe le job en `success`.
    pub async fn mark_success(db: &DatabaseConnection, id: Uuid) -> Result<(), GitrustError> {
        Self::update_status(db, id, NotifyStatus::Success, None).await
    }

    /// Passe le job en `failed` avec un message d'erreur.
    pub async fn mark_failed(
        db: &DatabaseConnection,
        id: Uuid,
        error: &str,
    ) -> Result<(), GitrustError> {
        Self::update_status(db, id, NotifyStatus::Failed, Some(error)).await
    }

    /// Passe le job en `cancelled`.
    pub async fn mark_cancelled(db: &DatabaseConnection, id: Uuid) -> Result<(), GitrustError> {
        Self::update_status(db, id, NotifyStatus::Cancelled, None).await
    }

    /// Vérifie si le job a été annulé (pour le poll du worker).
    pub async fn is_cancelled(
        db: &DatabaseConnection,
        id: Uuid,
    ) -> Result<bool, GitrustError> {
        let job = notify_job::Entity::find_by_id(id)
            .one(db)
            .await
            .map_err(GitrustError::Database)?
            .ok_or_else(|| GitrustError::NotFound("job introuvable".into()))?;
        Ok(job.status == NotifyStatus::Cancelled.as_str())
    }

    async fn update_status(
        db: &DatabaseConnection,
        id: Uuid,
        status: NotifyStatus,
        error: Option<&str>,
    ) -> Result<(), GitrustError> {
        use sea_orm::ActiveModelTrait;
        let now = Utc::now();
        let mut model: notify_job::ActiveModel = notify_job::Entity::find_by_id(id)
            .one(db)
            .await
            .map_err(GitrustError::Database)?
            .ok_or_else(|| GitrustError::NotFound("job introuvable".into()))?
            .into();

        model.status = Set(status.as_str().to_owned());
        model.updated_at = Set(now);
        model.error_message = Set(error.map(str::to_owned));

        if status == NotifyStatus::Running {
            model.started_at = Set(Some(now));
        }
        if matches!(status, NotifyStatus::Success | NotifyStatus::Failed | NotifyStatus::Cancelled) {
            model.finished_at = Set(Some(now));
        }

        model.update(db).await.map_err(GitrustError::Database)?;
        Ok(())
    }
}

// crates/gitrust-core/src/services/notify_worker.rs

use std::sync::Arc;

use sea_orm::DatabaseConnection;
use tokio::sync::{mpsc, Semaphore};
use uuid::Uuid;

use super::notify_service::NotifyService;

/// Tâche envoyée via le canal mpsc au worker.
pub struct NotifyTask {
    pub job_id: Uuid,
    pub target_url: String,
    pub payload: String,
}

pub struct NotifyWorkerConfig {
    /// Nombre maximal de jobs concurrents.
    pub max_concurrent: usize,
    /// Timeout en secondes par job.
    pub timeout_secs: u64,
}

/// Boucle principale du worker. À appeler avec `tokio::spawn`.
pub async fn run(
    db: DatabaseConnection,
    mut rx: mpsc::Receiver<NotifyTask>,
    config: NotifyWorkerConfig,
) {
    let semaphore = Arc::new(Semaphore::new(config.max_concurrent));

    while let Some(task) = rx.recv().await {
        let permit = semaphore.clone().acquire_owned().await.expect("semaphore fermé");
        let db = db.clone();
        let timeout = config.timeout_secs;

        tokio::spawn(async move {
            let _permit = permit; // libéré à la fin du bloc
            execute_task(&db, task, timeout).await;
        });
    }
}

async fn execute_task(db: &DatabaseConnection, task: NotifyTask, timeout_secs: u64) {
    // Vérifier si annulé avant de démarrer
    match NotifyService::is_cancelled(db, task.job_id).await {
        Ok(true) => return,
        Err(e) => {
            tracing::error!(job_id = %task.job_id, error = %e, "Erreur lecture job");
            return;
        }
        Ok(false) => {}
    }

    if let Err(e) = NotifyService::mark_running(db, task.job_id).await {
        tracing::error!(job_id = %task.job_id, error = %e, "Impossible de marquer running");
        return;
    }

    let result = tokio::time::timeout(
        std::time::Duration::from_secs(timeout_secs),
        send_notification(&task.target_url, &task.payload),
    )
    .await;

    match result {
        Ok(Ok(())) => {
            if let Err(e) = NotifyService::mark_success(db, task.job_id).await {
                tracing::error!(job_id = %task.job_id, error = %e, "Impossible de marquer success");
            }
        }
        Ok(Err(e)) => {
            let msg = format!("Erreur HTTP: {e}");
            if let Err(db_err) = NotifyService::mark_failed(db, task.job_id, &msg).await {
                tracing::error!(job_id = %task.job_id, error = %db_err, "Impossible de marquer failed");
            }
        }
        Err(_elapsed) => {
            let msg = format!("Timeout après {timeout_secs}s");
            if let Err(e) = NotifyService::mark_failed(db, task.job_id, &msg).await {
                tracing::error!(job_id = %task.job_id, error = %e, "Impossible de marquer failed (timeout)");
            }
        }
    }
}

async fn send_notification(url: &str, payload: &str) -> Result<(), reqwest::Error> {
    let client = reqwest::Client::new();
    client
        .post(url)
        .header("Content-Type", "application/json")
        .body(payload.to_owned())
        .send()
        .await?
        .error_for_status()?;
    Ok(())
}

use axum::{
    extract::{Path, State},
    response::{sse::{Event, KeepAlive, Sse}, Redirect},
    Extension, Form,
};
use sea_orm::DatabaseConnection;
use tokio::time::{interval, Duration};
use tokio_stream::wrappers::IntervalStream;
use tokio_stream::StreamExt;
use uuid::Uuid;

use gitrust_core::{
    dto::notify_dto::CreateNotifyJobInput,
    services::{notify_service::NotifyService, notify_worker::NotifyTask},
};

use crate::auth::AuthUser;

pub type NotifySender = tokio::sync::mpsc::Sender<NotifyTask>;

/// POST /notify — crée un job et redirige vers la page de suivi.
pub async fn create_notify_job(
    State(db): State<DatabaseConnection>,
    Extension(tx): Extension<NotifySender>,
    user: AuthUser,
    Form(input): Form<CreateNotifyJobInput>,
) -> Result<Redirect, crate::error::AppError> {
    let job_id = NotifyService::create_job(&db, user.user_id, input.clone()).await?;

    // Essai d'envoi dans le canal — si plein, le job reste `pending` jusqu'au
    // prochain redémarrage (les jobs `pending` peuvent être re-envoyés au démarrage).
    let _ = tx.try_send(NotifyTask {
        job_id,
        target_url: input.target_url,
        payload: input.payload,
    });

    Ok(Redirect::to(&format!("/notify/{job_id}")))
}

/// GET /notify/:id/stream — SSE : pousse les mises à jour de statut au client.
pub async fn notify_job_stream(
    State(db): State<DatabaseConnection>,
    Path(id): Path<Uuid>,
) -> Sse<impl tokio_stream::Stream<Item = Result<Event, std::convert::Infallible>>> {
    let stream = IntervalStream::new(interval(Duration::from_secs(1)))
        .map(move |_| {
            let db = db.clone();
            async move {
                // Poll la DB pour l'état courant
                use gitrust_core::models::notify_job;
                use sea_orm::EntityTrait;
                let job = notify_job::Entity::find_by_id(id)
                    .one(&db)
                    .await
                    .ok()
                    .flatten();

                let data = match job {
                    Some(j) => format!(
                        r#"{{"status":"{}","error":{}}}"#,
                        j.status,
                        j.error_message
                            .as_deref()
                            .map(|e| format!(r#""{}""#, e.replace('"', "\\\"")))
                            .unwrap_or("null".to_owned())
                    ),
                    None => r#"{"status":"not_found"}"#.to_owned(),
                };

                Ok(Event::default().data(data))
            }
        })
        .then(|fut| fut);

    Sse::new(stream).keep_alive(KeepAlive::default())
}

.route("/notify", post(handlers::notify::create_notify_job))
.route("/notify/:id", get(handlers::notify::notify_job_page))
.route("/notify/:id/stream", get(handlers::notify::notify_job_stream))
.route("/notify/:id/cancel", post(handlers::notify::cancel_notify_job))

// Dans la fonction main(), après la construction de l'app :
let (notify_tx, notify_rx) = tokio::sync::mpsc::channel::<NotifyTask>(100);

let notify_db = db.clone();
tokio::spawn(gitrust_core::services::notify_worker::run(
    notify_db,
    notify_rx,
    NotifyWorkerConfig {
        max_concurrent: std::env::var("NOTIFY_MAX_CONCURRENT")
            .unwrap_or("4".into())
            .parse()
            .unwrap_or(4),
        timeout_secs: std::env::var("NOTIFY_TIMEOUT_SECS")
            .unwrap_or("30".into())
            .parse()
            .unwrap_or(30),
    },
));

// Injecter l'émetteur dans le router via Extension
let app = routes::routes(&static_path)
    .layer(Extension(notify_tx))
    .with_state(db);

#[tokio::test]
async fn create_job_inserts_with_pending_status() {
    let db = setup_test_db().await;
    let id = NotifyService::create_job(
        &db,
        Uuid::new_v4(),
        CreateNotifyJobInput {
            target_url: "https://example.com/hook".into(),
            payload: r#"{"event":"test"}"#.into(),
        },
    )
    .await
    .expect("create_job doit réussir");

    let job = notify_job::Entity::find_by_id(id)
        .one(&db)
        .await
        .unwrap()
        .unwrap();
    assert_eq!(job.status, "pending");
    assert!(job.started_at.is_none());
}

#[tokio::test]
async fn mark_running_sets_started_at() {
    let db = setup_test_db().await;
    let id = create_test_job(&db).await;

    NotifyService::mark_running(&db, id).await.unwrap();

    let job = notify_job::Entity::find_by_id(id).one(&db).await.unwrap().unwrap();
    assert_eq!(job.status, "running");
    assert!(job.started_at.is_some());
}

#[tokio::test]
async fn cancel_before_running_is_terminal() {
    let db = setup_test_db().await;
    let id = create_test_job(&db).await;

    NotifyService::mark_cancelled(&db, id).await.unwrap();

    let is_cancelled = NotifyService::is_cancelled(&db, id).await.unwrap();
    assert!(is_cancelled);
}