Implement prune job

migrations/2025-02-14-115440_add_prune_job_type/down.sql

@@ -0,0 +1,5 @@
+-- Remove the constraint for job kinds
+ALTER TABLE jobs DROP CONSTRAINT chk_job_kind;
+
+-- Add back the original constraint without the prune job type
+ALTER TABLE jobs ADD CONSTRAINT chk_job_kind CHECK (kind IN (0, 1, 2, 3, 4)); -- Excluding 5 (Prune)

migrations/2025-02-14-115440_add_prune_job_type/up.sql

@@ -0,0 +1,4 @@
+-- Migration to add the "prune" job type to the jobs table
+
+ALTER TABLE jobs
+ADD CONSTRAINT chk_job_kind CHECK (kind IN (0, 1, 2, 3, 4)); -- Assuming 0-3 are existing job types, 4 is for "prune"

src/api/mod.rs

@@ -1,5 +1,6 @@
 pub mod build;
 pub mod delta;
+pub mod prune;
 pub mod repo;
 pub mod status;
 pub mod tokens;

src/api/prune.rs

@@ -0,0 +1,34 @@
+use actix_web::{web, HttpRequest, HttpResponse};
+use futures3::TryFutureExt;
+use serde::Deserialize;
+
+use crate::db::Db;
+use crate::tokens::{ClaimsScope, ClaimsValidator};
+use crate::errors::ApiError;
+
+#[derive(Deserialize)]
+pub struct PruneArgs {
+    repo: String,
+}
+
+pub fn handle_prune(
+    args: web::Json<PruneArgs>,
+    db: web::Data<Db>,
+    req: HttpRequest,
+) -> impl futures::Future<Item = HttpResponse, Error = ApiError> {
+    Box::pin(handle_prune_async(args, db, req)).compat()
+}
+
+async fn handle_prune_async(
+    args: web::Json<PruneArgs>,
+    db: web::Data<Db>,
+    req: HttpRequest,
+) -> Result<HttpResponse, ApiError> {
+    // Verify token has TokenManagement scope
+    req.has_token_claims("", ClaimsScope::TokenManagement)?;
+
+    // Create a new prune job
+    let job = db.get_ref().start_prune_job(args.repo.clone()).await?;
+
+    Ok(HttpResponse::Ok().json(job))
+}

src/app.rs

@@ -181,6 +181,10 @@ pub fn create_app(
                     .service(
                         web::resource("/delta/upload/{repo}")
                             .route(web::post().to_async(api::delta::delta_upload)),
+                    )
+                    .service(
+                        web::resource("/prune")
+                            .route(web::post().to_async(api::prune::handle_prune)),
                     ),
             )
             .service(

src/db.rs

@@ -272,6 +272,21 @@ impl Db {
         .await
     }
 
+    pub async fn start_prune_job(&self, repo: String) -> Result<Job, ApiError> {
+        self.run_in_transaction(move |conn| {
+            diesel::insert_into(schema::jobs::table)
+                .values(NewJob {
+                    kind: JobKind::Prune.to_db(),
+                    contents: json!({}).to_string(),
+                    start_after: None,
+                    repo: Some(repo),
+                })
+                .get_result(conn)
+                .map_err(ApiError::from)
+        })
+        .await
+    }
+
     /* Checks */
 
     pub async fn get_check_by_job_id(&self, job: i32) -> Result<Check, ApiError> {

src/jobs/job_instance.rs

@@ -5,6 +5,7 @@ use diesel::pg::PgConnection;
 use super::check_job::CheckJobInstance;
 use super::commit_job::CommitJobInstance;
 use super::job_executor::JobExecutor;
+use super::prune_job::PruneJobInstance;
 use super::publish_job::PublishJobInstance;
 use super::republish_job::RepublishJobInstance;
 use super::update_repo_job::UpdateRepoJobInstance;
@@ -18,6 +19,7 @@ pub fn new_job_instance(executor: &JobExecutor, job: Job) -> Box<dyn JobInstance
         }
         Some(JobKind::Republish) => RepublishJobInstance::new(job),
         Some(JobKind::Check) => CheckJobInstance::new(job),
+        Some(JobKind::Prune) => PruneJobInstance::new(job),
         _ => InvalidJobInstance::new(job, JobError::new("Unknown job type")),
     }
 }

src/jobs/mod.rs

@@ -6,6 +6,7 @@ mod commit_job;
 mod job_executor;
 mod job_instance;
 mod job_queue;
+mod prune_job;
 mod publish_job;
 mod republish_job;
 mod update_repo_job;

src/jobs/prune_job.rs

@@ -0,0 +1,83 @@
+use crate::errors::{JobResult, JobError};
+use crate::jobs::job_executor::JobExecutor;
+use crate::jobs::job_instance::JobInstance;
+use crate::models::Job;
+use diesel::pg::PgConnection;
+use serde::{Deserialize, Serialize};
+use std::process::Command;
+use log::info;
+
+#[derive(Debug, Serialize, Deserialize)]
+pub struct PruneJob {}
+
+pub struct PruneJobInstance {
+    job: Job,
+}
+
+impl PruneJobInstance {
+    pub fn new(job: Job) -> Box<dyn JobInstance> {
+        match serde_json::from_str::<PruneJob>(&job.contents) {
+            Ok(_) => Box::new(PruneJobInstance { job }),
+            Err(e) => super::job_instance::InvalidJobInstance::new(
+                job,
+                JobError::new(&format!("Invalid prune job contents: {}", e))
+            ),
+        }
+    }
+}
+
+impl JobInstance for PruneJobInstance {
+    fn get_job_id(&self) -> i32 {
+        self.job.id
+    }
+
+    fn handle_job(
+        &mut self,
+        executor: &JobExecutor,
+        conn: &mut PgConnection,
+    ) -> JobResult<serde_json::Value> {
+        info!("#{}: Handling Job Prune", &self.job.id);
+
+        // Get repo config
+        let config = &executor.config;
+        let repo = self.job.repo.as_ref().ok_or_else(|| JobError::new("No repo specified"))?;
+        let repoconfig = config
+            .get_repoconfig(repo)
+            .map_err(|_e| JobError::new(&format!("Can't find repo {}", repo)))?;
+
+        let repo_path = repoconfig.get_abs_repo_path();
+
+        // First do a dry run
+        job_log_and_info!(self.job.id, conn, "Running prune dry-run");
+        let mut cmd = Command::new("flatpak");
+        cmd.arg("build-update-repo")
+            .arg("-v")
+            .arg("--no-update-summary")
+            .arg("--no-update-appstream")
+            .arg("--prune-dry-run")
+            .arg("--prune-depth=3")
+            .arg(&repo_path);
+
+        super::utils::do_command(cmd)?;
+
+        // Then do the actual prune
+        job_log_and_info!(self.job.id, conn, "Running actual prune");
+        let mut cmd = Command::new("flatpak");
+        cmd.arg("build-update-repo")
+            .arg("-v")
+            .arg("--no-update-summary")
+            .arg("--no-update-appstream")
+            .arg("--prune")
+            .arg("--prune-depth=3")
+            .arg(&repo_path);
+
+        super::utils::do_command(cmd)?;
+
+        Ok(serde_json::json!({}))
+    }
+
+    // Higher order than Publish/UpdateRepo to prevent them from running while prune is in queue
+    fn order(&self) -> i32 {
+        100
+    }
+}

src/models.rs

@@ -194,6 +194,7 @@ pub enum JobKind {
     UpdateRepo,
     Republish,
     Check,
+    Prune,
 }
 
 impl JobKind {
@@ -204,6 +205,7 @@ impl JobKind {
             JobKind::UpdateRepo => 2,
             JobKind::Republish => 3,
             JobKind::Check => 4,
+            JobKind::Prune => 5,
         }
     }
 
@@ -214,6 +216,7 @@ impl JobKind {
             2 => Some(JobKind::UpdateRepo),
             3 => Some(JobKind::Republish),
             4 => Some(JobKind::Check),
+            5 => Some(JobKind::Prune),
             _ => None,
         }
     }