feat(zebra): implement data deletion worker

Author: loadez

zebra/config/config.exs

@@ -24,6 +24,11 @@ config :zebra, Zebra.Workers.TaskFinisher, timeout: 10_000
 config :zebra, Zebra.Workers.Dispatcher, timeout: 1_000
 config :zebra, Zebra.Workers.Monitor, timeout: 60_000
 
+config :zebra, Zebra.Workers.JobDeletionPolicyWorker,
+  naptime: 1_000, # 1 second
+  longnaptime: 3_600_000, # 1 hour
+  limit: 100
+
 config :zebra, Zebra.Workers.Scheduler,
   cooldown_period: 1_000,
   batch_size: 3

zebra/lib/zebra/models/job.ex

@@ -79,6 +79,7 @@ defmodule Zebra.Models.Job do
     field(:scheduled_at, :utc_datetime)
     field(:started_at, :utc_datetime)
     field(:finished_at, :utc_datetime)
+    field(:expires_at, :utc_datetime)
   end
 
   def create(params) do
@@ -356,6 +357,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,62 @@
+defmodule Zebra.Workers.JobDeletionPolicyWorker do
+  require Logger
+
+  @self_hosted_prefix "s1-%"
+
+  defstruct [
+    # period of sleep between worker ticks
+    :naptime,
+    # longer period of sleep when there is nothing to delete
+    :longnaptime,
+    # limit for deletions per batch
+    :limit
+  ]
+
+  def start_link do
+    worker_config = Application.fetch_env!(:zebra, __MODULE__)
+    worker = struct(__MODULE__, worker_config)
+
+    pid =
+      spawn_link(fn ->
+        loop(worker)
+      end)
+
+    {:ok, pid}
+  end
+
+  def loop(worker) do
+    # Perform a tick (cleaning operation)
+    deleted_any? = Task.async(fn -> tick(worker) end) |> Task.await(:infinity)
+
+    sleep_for =
+      if deleted_any? do
+        worker.naptime
+      else
+        worker.longnaptime || worker.naptime
+      end
+
+    :timer.sleep(sleep_for)
+
+    # Recursively call loop to continue periodic execution
+    loop(worker)
+  end
+
+  def tick(worker) do
+    Logger.info("Starting cleanup tick...")
+
+    limit = worker.limit
+
+    {:ok, deleted_stop_requests} = Zebra.Models.Job.delete_old_job_stop_requests(limit)
+    {:ok, deleted_jobs} = Zebra.Models.Job.delete_old_jobs(limit)
+
+    total_deleted = deleted_stop_requests + deleted_jobs
+
+    if total_deleted == 0 do
+      Logger.info("No jobs found for deletion.")
+      false
+    else
+      Logger.info("Deleted #{deleted_stop_requests} job stop requests and #{deleted_jobs} jobs.")
+      true
+    end
+  end
+end

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

@@ -0,0 +1,9 @@
+defmodule Zebra.LegacyRepo.Migrations.AddExpiresAtToJobs do
+  use Ecto.Migration
+
+  def change do
+    alter table(:jobs) do
+      add :expires_at, :utc_datetime
+    end
+  end
+end

zebra/priv/legacy_repo/migrations/20251119103829_add_expires_created_index_at_jobs_table.exs

@@ -0,0 +1,13 @@
+defmodule Zebra.LegacyRepo.Migrations.AddExpiresCreatedIndexAtJobsTable do
+  use Ecto.Migration
+  @disable_migration_lock true
+  @disable_ddl_transaction true
+
+  def change do
+    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

zebra/priv/legacy_repo/migrations/20251119104017_add_organization_created_index_at_jobs_table.exs

@@ -0,0 +1,13 @@
+defmodule Zebra.LegacyRepo.Migrations.AddOrganizationCreatedIndexAtJobsTable do
+  use Ecto.Migration
+  @disable_migration_lock true
+  @disable_ddl_transaction true
+
+  def change do
+    create index(:jobs, [:organization_id, :created_at],
+      name: "index_jobs_on_organization_created_expires_is_null",
+      concurrently: true,
+      where: "expires_at IS NULL"
+    )
+  end
+end