Implement prune job (#147)

Author: barthalion

flat-manager-client

@@ -965,6 +965,20 @@ async def purge_command(session, args):
     return job
 
 
+async def prune_command(session, args):
+    resp = await session.post(
+        urljoin(args.manager_url, "api/v1/prune"),
+        headers={"Authorization": "Bearer " + args.token},
+        json={"repo": args.repo},
+    )
+    async with resp:
+        if resp.status >= 500:
+            raise ServerApiError(resp, await resp.text())
+        elif resp.status != 200:
+            raise ApiError(resp, await resp.text())
+        return await resp.json()
+
+
 async def create_token_command(session, args):
     data = await create_token(
         session,
@@ -1096,6 +1110,11 @@ if __name__ == "__main__":
     purge_parser.add_argument("build_url", help="remote build url")
     purge_parser.set_defaults(func=purge_command)
 
+    prune_parser = subparsers.add_parser("prune", help="Prune repository")
+    prune_parser.add_argument("manager_url", help="remote repo manager url")
+    prune_parser.add_argument("repo", help="repo name")
+    prune_parser.set_defaults(func=prune_command)
+
     create_token_parser = subparsers.add_parser(
         "create-token", help="Create subset token"
     )

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, 5)); -- 0=commit, 1=publish, 2=updaterepo, 3=republish, 4=check, 5=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,40 @@
+use actix_web::{web, HttpRequest, HttpResponse};
+use futures3::TryFutureExt;
+use serde::Deserialize;
+
+use crate::db::Db;
+use crate::errors::ApiError;
+use crate::tokens::{ClaimsScope, ClaimsValidator};
+
+#[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> {
+    req.validate_claims(|claims| {
+        if !claims.scope.contains(&ClaimsScope::TokenManagement) {
+            return Err(ApiError::NotEnoughPermissions(
+                "Missing TokenManagement scope".to_string(),
+            ));
+        }
+        Ok(())
+    })?;
+
+    // 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,88 @@
+use crate::errors::{JobError, JobResult};
+use crate::jobs::job_executor::JobExecutor;
+use crate::jobs::job_instance::JobInstance;
+use crate::models::Job;
+use diesel::pg::PgConnection;
+use log::info;
+use serde::{Deserialize, Serialize};
+use std::process::Command;
+
+#[derive(Debug, Serialize, Deserialize)]
+pub struct PruneJob {}
+
+pub struct PruneJobInstance {
+    job: Job,
+}
+
+impl PruneJobInstance {
+    #[allow(clippy::new_ret_no_self)]
+    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
+    }
+}