feat(zebra): implement data deletion worker

Author: loadez

zebra/lib/zebra/models/job.ex

@@ -356,6 +356,49 @@ defmodule Zebra.Models.Job do
     )
   end
 
+  def delete_old_job_stop_requests(limit) do
+    import Ecto.Query,
+      only: [from: 2, where: 3, subquery: 1, limit: 2, order_by: 2]
+
+    jobs_subquery =
+      from(j in Zebra.Models.Job,
+        where: not is_nil(j.expires_at) and j.expires_at <= fragment("CURRENT_TIMESTAMP"),
+        order_by: [asc: j.created_at],
+        limit: ^limit,
+        select: j.id
+      )
+
+    query =
+      from(jsr in Zebra.Models.JobStopRequest,
+        where: jsr.job_id in subquery(jobs_subquery)
+      )
+
+    {deleted_count, _} = Zebra.LegacyRepo.delete_all(query)
+
+    {:ok, deleted_count}
+  end
+
+  def delete_old_jobs(limit) do
+    import Ecto.Query, only: [from: 2]
+
+    jobs_subquery =
+      from(j in Zebra.Models.Job,
+        where: not is_nil(j.expires_at) and j.expires_at <= fragment("CURRENT_TIMESTAMP"),
+        order_by: [asc: j.created_at],
+        limit: ^limit,
+        select: j.id
+      )
+
+    query =
+      from(j in Zebra.Models.Job,
+        where: j.id in subquery(jobs_subquery)
+      )
+
+    {deleted_count, _} = Zebra.LegacyRepo.delete_all(query)
+
+    {:ok, deleted_count}
+  end
+
   def wait_for_agent(job) do
     if valid_transition?(job.aasm_state, state_waiting_for_agent()) do
       update(job, %{aasm_state: state_waiting_for_agent()})

zebra/lib/zebra/workers.ex

@@ -1,5 +1,6 @@
 defmodule Zebra.Workers do
   @all [
+    %{name: Zebra.Workers.JobDeletionPolicyWorker, flag: "START_JOB_DELETION_POLICY_WORKER"},
     %{name: Zebra.Workers.JobStartedCallbackWorker, flag: "START_JOB_STARTED_CALLBACK_WORKER"},
     %{name: Zebra.Workers.JobFinishedCallbackWorker, flag: "START_JOB_FINISHED_CALLBACK_WORKER"},
     %{name: Zebra.Workers.JobTeardownCallbackWorker, flag: "START_JOB_TEARDOWN_CALLBACK_WORKER"},

zebra/lib/zebra/workers/job_deletion_policy_worker.ex

@@ -0,0 +1,42 @@
+defmodule Zebra.Workers.JobDeletionPolicyWorker do
+  require Logger
+
+  @self_hosted_prefix "s1-%"
+
+  defstruct [
+    # period of sleep between worker ticks
+    :naptime,
+    # limit for deletions per batch
+    :limit,
+  ]
+
+  def start_link(worker) do
+    pid =
+      spawn_link(fn ->
+        loop(worker)
+      end)
+
+    {:ok, pid}
+  end
+
+  def loop(worker) do
+    # Perform a tick (cleaning operation)
+    Task.async(fn -> tick(worker) end) |> Task.await(:infinity)
+
+    :timer.sleep(worker.naptime)
+
+    # Recursively call loop to continue periodic execution
+    loop(worker)
+  end
+
+  def tick(worker) do
+    Logger.info("Starting cleanup tick...")
+
+    limit = worker.limit
+
+    Zebra.Models.Job.delete_old_job_stop_requests(limit)
+    Zebra.Models.Job.delete_old_jobs(limit)
+
+    Logger.info("Cleanup tick completed.")
+  end
+end

zebra/priv/legacy_repo/migrations/20251118091829_add_expires_at_to_jobs.exs

@@ -0,0 +1,14 @@
+defmodule Zebra.LegacyRepo.Migrations.AddExpiresAtToJobs do
+  use Ecto.Migration
+
+  def change do
+    alter table(:jobs) do
+      add :expires_at, :utc_datetime
+    end
+    create index(:jobs, [:expires_at, :created_at],
+      name: "index_jobs_on_expires_created_not_null",
+      concurrently: true,
+      where: "expires_at IS NOT NULL"
+    )
+  end
+end