feat: add per-organization data retention with 1-year default (#144)

Add data_retention_days field to organizations (default 365, range 30-730).
DataRetentionWorker runs nightly at 1 AM to purge expired time-series data
across 8 tables, respecting each org's retention setting.

TimescaleDB global retention bumped from 90 to 730 days as safety net.
Resolved alerts are cleaned up; unresolved alerts kept regardless of age.

Reviewed-on: graham/towerops-web#144

config/runtime.exs

@@ -252,7 +252,9 @@ if config_env() == :prod do
          # Backhaul capacity utilization insights every 15 minutes
          {"*/15 * * * *", Towerops.Workers.CapacityInsightWorker},
          # Sync device usage to Stripe daily at 3 AM UTC
-         {"0 3 * * *", Towerops.Workers.BillingSyncWorker}
+         {"0 3 * * *", Towerops.Workers.BillingSyncWorker},
+         # Data retention cleanup nightly at 1 AM
+         {"0 1 * * *", Towerops.Workers.DataRetentionWorker}
        ]},
       # Automatically delete completed jobs after 60 seconds.
       # Default limit of 10k/30s (333/s) can't keep up with job throughput (~600+/s),

lib/towerops/organizations/organization.ex

@@ -66,6 +66,9 @@ defmodule Towerops.Organizations.Organization do
     field :last_billing_sync_at, :utc_datetime
     field :last_synced_device_count, :integer
 
+    # Data retention
+    field :data_retention_days, :integer, default: 365
+
     # Virtual fields (populated by queries)
     field :device_count, :integer, virtual: true
 
@@ -112,6 +115,7 @@ defmodule Towerops.Organizations.Organization do
           payment_method_status: String.t() | nil,
           last_billing_sync_at: DateTime.t() | nil,
           last_synced_device_count: integer() | nil,
+          data_retention_days: integer(),
           default_agent_token_id: Ecto.UUID.t() | nil,
           default_agent_token: NotLoaded.t() | AgentToken.t() | nil,
           default_escalation_policy_id: Ecto.UUID.t() | nil,
@@ -155,7 +159,8 @@ defmodule Towerops.Organizations.Organization do
       :last_billing_sync_at,
       :last_synced_device_count,
       :default_escalation_policy_id,
-      :alert_routing
+      :alert_routing,
+      :data_retention_days
     ])
     |> validate_required([:name])
     |> validate_length(:name, min: 2, max: 100)
@@ -165,6 +170,11 @@ defmodule Towerops.Organizations.Organization do
     )
     |> validate_inclusion(:snmp_version, ["1", "2c", "3"], message: "must be 1, 2c, or 3")
     |> validate_number(:snmp_port, greater_than: 0, less_than: 65_536)
+    |> validate_number(:data_retention_days,
+      greater_than_or_equal_to: 30,
+      less_than_or_equal_to: 730,
+      message: "must be between 30 and 730 days"
+    )
     |> validate_snmpv3_fields()
     |> validate_mikrotik_fields()
     |> generate_slug()

lib/towerops/workers/data_retention_worker.ex

@@ -0,0 +1,174 @@
+defmodule Towerops.Workers.DataRetentionWorker do
+  @moduledoc """
+  Oban worker for per-organization data retention.
+
+  Runs nightly to delete time-series data older than each organization's
+  `data_retention_days` setting (default: 365 days).
+
+  Tables cleaned:
+  - monitoring_checks (via device_id -> devices.organization_id)
+  - snmp_sensor_readings (via sensor -> snmp_device -> device -> org)
+  - snmp_interface_stats (via interface -> snmp_device -> device -> org)
+  - snmp_processor_readings (via processor -> snmp_device -> device -> org)
+  - snmp_storage_readings (via storage -> snmp_device -> device -> org)
+  - check_results (direct organization_id)
+  - wireless_client_readings (direct organization_id)
+  - alerts (direct organization_id, only resolved alerts)
+
+  TimescaleDB global retention (730 days) acts as a safety net.
+  """
+  use Oban.Worker, queue: :maintenance
+
+  alias Towerops.Repo
+
+  require Logger
+
+  @impl Oban.Worker
+  def perform(_job) do
+    retention_groups = fetch_retention_groups()
+
+    results =
+      Enum.flat_map(retention_groups, fn {retention_days, org_ids} ->
+        cutoff = DateTime.add(DateTime.utc_now(), -retention_days, :day)
+        purge_for_orgs(org_ids, cutoff)
+      end)
+
+    total = Enum.reduce(results, 0, fn {_table, count}, acc -> acc + count end)
+
+    if total > 0 do
+      summary = Enum.reject(results, fn {_table, count} -> count == 0 end)
+      Logger.info("Data retention: deleted #{total} total rows — #{inspect(summary)}")
+    end
+
+    {:ok, %{total_deleted: total, details: Map.new(results)}}
+  end
+
+  defp fetch_retention_groups do
+    result =
+      Repo.query!("""
+      SELECT data_retention_days, array_agg(id) AS org_ids
+      FROM organizations
+      GROUP BY data_retention_days
+      """)
+
+    Enum.map(result.rows, fn [days, ids] ->
+      # Keep raw binary UUIDs from PostgreSQL — Postgrex expects binary format
+      # for uuid[] parameters, not string-formatted UUIDs
+      {days, ids}
+    end)
+  end
+
+  defp purge_for_orgs(org_ids, cutoff) do
+    # Tables with direct organization_id
+    direct_results =
+      Enum.map(
+        [{"check_results", "checked_at"}, {"wireless_client_readings", "checked_at"}],
+        fn {table, time_col} ->
+          result =
+            Repo.query!(
+              "DELETE FROM #{table} WHERE organization_id = ANY($1::uuid[]) AND #{time_col} < $2",
+              [org_ids, cutoff]
+            )
+
+          {table, result.num_rows}
+        end
+      )
+
+    # Resolved alerts only (keep unresolved regardless of age)
+    alert_result =
+      Repo.query!(
+        """
+        DELETE FROM alerts
+        WHERE organization_id = ANY($1::uuid[])
+          AND inserted_at < $2
+          AND resolved_at IS NOT NULL
+        """,
+        [org_ids, cutoff]
+      )
+
+    # monitoring_checks: device_id -> devices.organization_id
+    mc_result =
+      Repo.query!(
+        """
+        DELETE FROM monitoring_checks
+        WHERE checked_at < $2
+          AND device_id IN (SELECT id FROM devices WHERE organization_id = ANY($1::uuid[]))
+        """,
+        [org_ids, cutoff]
+      )
+
+    # snmp_sensor_readings: sensor_id -> snmp_sensors -> snmp_devices -> devices -> org
+    sr_result =
+      Repo.query!(
+        """
+        DELETE FROM snmp_sensor_readings
+        WHERE checked_at < $2
+          AND sensor_id IN (
+            SELECT s.id FROM snmp_sensors s
+            JOIN snmp_devices sd ON sd.id = s.snmp_device_id
+            JOIN devices d ON d.id = sd.device_id
+            WHERE d.organization_id = ANY($1::uuid[])
+          )
+        """,
+        [org_ids, cutoff]
+      )
+
+    # snmp_interface_stats: interface_id -> snmp_interfaces -> snmp_devices -> devices -> org
+    is_result =
+      Repo.query!(
+        """
+        DELETE FROM snmp_interface_stats
+        WHERE checked_at < $2
+          AND interface_id IN (
+            SELECT i.id FROM snmp_interfaces i
+            JOIN snmp_devices sd ON sd.id = i.snmp_device_id
+            JOIN devices d ON d.id = sd.device_id
+            WHERE d.organization_id = ANY($1::uuid[])
+          )
+        """,
+        [org_ids, cutoff]
+      )
+
+    # snmp_processor_readings: processor_id -> snmp_processors -> snmp_devices -> devices -> org
+    pr_result =
+      Repo.query!(
+        """
+        DELETE FROM snmp_processor_readings
+        WHERE checked_at < $2
+          AND processor_id IN (
+            SELECT p.id FROM snmp_processors p
+            JOIN snmp_devices sd ON sd.id = p.snmp_device_id
+            JOIN devices d ON d.id = sd.device_id
+            WHERE d.organization_id = ANY($1::uuid[])
+          )
+        """,
+        [org_ids, cutoff]
+      )
+
+    # snmp_storage_readings: storage_id -> snmp_storage -> snmp_devices -> devices -> org
+    str_result =
+      Repo.query!(
+        """
+        DELETE FROM snmp_storage_readings
+        WHERE checked_at < $2
+          AND storage_id IN (
+            SELECT st.id FROM snmp_storage st
+            JOIN snmp_devices sd ON sd.id = st.snmp_device_id
+            JOIN devices d ON d.id = sd.device_id
+            WHERE d.organization_id = ANY($1::uuid[])
+          )
+        """,
+        [org_ids, cutoff]
+      )
+
+    direct_results ++
+      [
+        {"alerts_resolved", alert_result.num_rows},
+        {"monitoring_checks", mc_result.num_rows},
+        {"snmp_sensor_readings", sr_result.num_rows},
+        {"snmp_interface_stats", is_result.num_rows},
+        {"snmp_processor_readings", pr_result.num_rows},
+        {"snmp_storage_readings", str_result.num_rows}
+      ]
+  end
+end

priv/repo/migrations/20260324193531_add_data_retention_days_to_organizations.exs

@@ -0,0 +1,9 @@
+defmodule Towerops.Repo.Migrations.AddDataRetentionDaysToOrganizations do
+  use Ecto.Migration
+
+  def change do
+    alter table(:organizations) do
+      add :data_retention_days, :integer, default: 365, null: false
+    end
+  end
+end

priv/repo/migrations/20260324193820_update_timescaledb_retention_to_730_days.exs

@@ -0,0 +1,59 @@
+defmodule Towerops.Repo.Migrations.UpdateTimescaledbRetentionTo730Days do
+  @moduledoc """
+  Updates TimescaleDB retention policies from 90 days to 730 days.
+
+  Per-organization retention is now handled by DataRetentionWorker (default 365 days).
+  The 730-day TimescaleDB policy acts as a safety net for orphaned data from
+  deleted organizations or data that the worker misses.
+  """
+  use Ecto.Migration
+
+  @disable_ddl_transaction true
+  @disable_migration_lock true
+
+  @hypertables ~w(monitoring_checks snmp_sensor_readings snmp_interface_stats wireless_client_readings)
+
+  def up do
+    if timescaledb_available?() do
+      # Update existing retention policies from 90 days to 730 days
+      for table <- @hypertables do
+        execute("SELECT remove_retention_policy('#{table}', if_exists => true)")
+
+        execute(
+          "SELECT add_retention_policy('#{table}', INTERVAL '730 days', if_not_exists => true)"
+        )
+      end
+
+      # Add missing retention policy for check_results
+      execute(
+        "SELECT add_retention_policy('check_results', INTERVAL '730 days', if_not_exists => true)"
+      )
+    end
+  end
+
+  def down do
+    if timescaledb_available?() do
+      # Revert to original 90-day retention
+      for table <- @hypertables do
+        execute("SELECT remove_retention_policy('#{table}', if_exists => true)")
+
+        execute(
+          "SELECT add_retention_policy('#{table}', INTERVAL '90 days', if_not_exists => true)"
+        )
+      end
+
+      # Remove check_results retention (it had none before)
+      execute("SELECT remove_retention_policy('check_results', if_exists => true)")
+    end
+  end
+
+  defp timescaledb_available? do
+    # Check for actual hypertables, not just the extension being installed.
+    # The extension may be available but tables might not be hypertables
+    # (e.g., CI/staging environments that skip hypertable creation).
+    case repo().query("SELECT 1 FROM timescaledb_information.hypertables LIMIT 1") do
+      {:ok, %{num_rows: n}} when n > 0 -> true
+      _ -> false
+    end
+  end
+end

test/towerops/workers/data_retention_worker_test.exs

@@ -0,0 +1,303 @@
+defmodule Towerops.Workers.DataRetentionWorkerTest do
+  use Towerops.DataCase, async: false
+
+  import Towerops.AccountsFixtures
+  import Towerops.DevicesFixtures
+  import Towerops.OrganizationsFixtures
+
+  alias Towerops.Organizations
+  alias Towerops.Repo
+  alias Towerops.Workers.DataRetentionWorker
+
+  describe "perform/1" do
+    test "deletes monitoring_checks older than org retention period" do
+      user = user_fixture()
+      org = organization_fixture(user.id)
+      {:ok, org} = Organizations.update_organization(org, %{data_retention_days: 30})
+      device = device_fixture(%{organization_id: org.id})
+
+      now = DateTime.truncate(DateTime.utc_now(), :second)
+      old_time = DateTime.add(now, -31, :day)
+      recent_time = DateTime.add(now, -10, :day)
+
+      old_id = Ecto.UUID.generate()
+      recent_id = Ecto.UUID.generate()
+
+      Repo.insert_all("monitoring_checks", [
+        %{
+          id: Ecto.UUID.dump!(old_id),
+          device_id: Ecto.UUID.dump!(device.id),
+          status: "success",
+          checked_at: old_time,
+          inserted_at: old_time
+        },
+        %{
+          id: Ecto.UUID.dump!(recent_id),
+          device_id: Ecto.UUID.dump!(device.id),
+          status: "success",
+          checked_at: recent_time,
+          inserted_at: recent_time
+        }
+      ])
+
+      assert {:ok, result} = DataRetentionWorker.perform(%Oban.Job{args: %{}})
+      assert result.total_deleted >= 1
+
+      # Old record should be deleted
+      remaining =
+        Repo.all(
+          from(m in "monitoring_checks",
+            where: m.device_id == type(^device.id, :binary_id),
+            select: m.id
+          )
+        )
+
+      remaining_uuids = Enum.map(remaining, &Ecto.UUID.cast!/1)
+      refute old_id in remaining_uuids
+      assert recent_id in remaining_uuids
+    end
+
+    test "deletes check_results older than org retention (direct org_id)" do
+      user = user_fixture()
+      org = organization_fixture(user.id)
+      {:ok, org} = Organizations.update_organization(org, %{data_retention_days: 30})
+
+      now = DateTime.truncate(DateTime.utc_now(), :second)
+      old_time = DateTime.add(now, -31, :day)
+      recent_time = DateTime.add(now, -10, :day)
+
+      # Create a check for the org
+      {:ok, check} =
+        Towerops.Monitoring.create_check(%{
+          organization_id: org.id,
+          check_type: "http",
+          name: "Test Check",
+          interval_seconds: 60,
+          config: %{"url" => "http://example.com"}
+        })
+
+      old_id = Ecto.UUID.generate()
+      recent_id = Ecto.UUID.generate()
+
+      Repo.insert_all("check_results", [
+        %{
+          id: Ecto.UUID.dump!(old_id),
+          organization_id: Ecto.UUID.dump!(org.id),
+          check_id: Ecto.UUID.dump!(check.id),
+          status: 0,
+          checked_at: old_time
+        },
+        %{
+          id: Ecto.UUID.dump!(recent_id),
+          organization_id: Ecto.UUID.dump!(org.id),
+          check_id: Ecto.UUID.dump!(check.id),
+          status: 0,
+          checked_at: recent_time
+        }
+      ])
+
+      assert {:ok, result} = DataRetentionWorker.perform(%Oban.Job{args: %{}})
+      assert result.total_deleted >= 1
+
+      remaining =
+        Repo.all(
+          from(cr in "check_results",
+            where: cr.organization_id == type(^org.id, :binary_id),
+            select: cr.id
+          )
+        )
+
+      remaining_uuids = Enum.map(remaining, &Ecto.UUID.cast!/1)
+      refute old_id in remaining_uuids
+      assert recent_id in remaining_uuids
+    end
+
+    test "respects different retention periods per org" do
+      user1 = user_fixture()
+      org1 = organization_fixture(user1.id)
+      {:ok, org1} = Organizations.update_organization(org1, %{data_retention_days: 60})
+      device1 = device_fixture(%{organization_id: org1.id})
+
+      user2 = user_fixture()
+      org2 = organization_fixture(user2.id)
+      {:ok, org2} = Organizations.update_organization(org2, %{data_retention_days: 90})
+      device2 = device_fixture(%{organization_id: org2.id})
+
+      now = DateTime.truncate(DateTime.utc_now(), :second)
+      # 75 days old: past org1's 60-day retention, within org2's 90-day retention
+      mid_time = DateTime.add(now, -75, :day)
+
+      org1_check_id = Ecto.UUID.generate()
+      org2_check_id = Ecto.UUID.generate()
+
+      Repo.insert_all("monitoring_checks", [
+        %{
+          id: Ecto.UUID.dump!(org1_check_id),
+          device_id: Ecto.UUID.dump!(device1.id),
+          status: "success",
+          checked_at: mid_time,
+          inserted_at: mid_time
+        },
+        %{
+          id: Ecto.UUID.dump!(org2_check_id),
+          device_id: Ecto.UUID.dump!(device2.id),
+          status: "success",
+          checked_at: mid_time,
+          inserted_at: mid_time
+        }
+      ])
+
+      assert {:ok, _result} = DataRetentionWorker.perform(%Oban.Job{args: %{}})
+
+      # org1 record should be deleted (75 days > 60 day retention)
+      org1_remaining =
+        Repo.all(
+          from(m in "monitoring_checks",
+            where: m.device_id == type(^device1.id, :binary_id),
+            select: m.id
+          )
+        )
+
+      assert org1_remaining == []
+
+      # org2 record should remain (75 days < 90 day retention)
+      org2_remaining =
+        Repo.all(
+          from(m in "monitoring_checks",
+            where: m.device_id == type(^device2.id, :binary_id),
+            select: m.id
+          )
+        )
+
+      org2_remaining_uuids = Enum.map(org2_remaining, &Ecto.UUID.cast!/1)
+      assert org2_check_id in org2_remaining_uuids
+    end
+
+    test "does NOT delete data within retention window" do
+      user = user_fixture()
+      org = organization_fixture(user.id)
+      # Default 365-day retention
+      device = device_fixture(%{organization_id: org.id})
+
+      now = DateTime.truncate(DateTime.utc_now(), :second)
+      recent_time = DateTime.add(now, -100, :day)
+
+      recent_id = Ecto.UUID.generate()
+
+      Repo.insert_all("monitoring_checks", [
+        %{
+          id: Ecto.UUID.dump!(recent_id),
+          device_id: Ecto.UUID.dump!(device.id),
+          status: "success",
+          checked_at: recent_time,
+          inserted_at: recent_time
+        }
+      ])
+
+      assert {:ok, result} = DataRetentionWorker.perform(%Oban.Job{args: %{}})
+
+      remaining =
+        Repo.all(
+          from(m in "monitoring_checks",
+            where: m.device_id == type(^device.id, :binary_id),
+            select: m.id
+          )
+        )
+
+      remaining_uuids = Enum.map(remaining, &Ecto.UUID.cast!/1)
+      assert recent_id in remaining_uuids
+      assert result.total_deleted == 0
+    end
+
+    test "returns summary with deletion counts" do
+      user = user_fixture()
+      org = organization_fixture(user.id)
+      {:ok, org} = Organizations.update_organization(org, %{data_retention_days: 30})
+      device = device_fixture(%{organization_id: org.id})
+
+      now = DateTime.truncate(DateTime.utc_now(), :second)
+      old_time = DateTime.add(now, -31, :day)
+
+      Repo.insert_all("monitoring_checks", [
+        %{
+          id: Ecto.UUID.dump!(Ecto.UUID.generate()),
+          device_id: Ecto.UUID.dump!(device.id),
+          status: "success",
+          checked_at: old_time,
+          inserted_at: old_time
+        },
+        %{
+          id: Ecto.UUID.dump!(Ecto.UUID.generate()),
+          device_id: Ecto.UUID.dump!(device.id),
+          status: "success",
+          checked_at: old_time,
+          inserted_at: old_time
+        }
+      ])
+
+      assert {:ok, result} = DataRetentionWorker.perform(%Oban.Job{args: %{}})
+      assert result.total_deleted >= 2
+      assert is_map(result.details)
+      assert Map.get(result.details, "monitoring_checks") >= 2
+    end
+
+    test "only deletes resolved alerts, keeps unresolved regardless of age" do
+      user = user_fixture()
+      org = organization_fixture(user.id)
+      {:ok, org} = Organizations.update_organization(org, %{data_retention_days: 30})
+      device = device_fixture(%{organization_id: org.id})
+
+      now = DateTime.truncate(DateTime.utc_now(), :second)
+      old_time = DateTime.add(now, -31, :day)
+
+      resolved_id = Ecto.UUID.generate()
+      unresolved_id = Ecto.UUID.generate()
+
+      Repo.insert_all("alerts", [
+        %{
+          id: Ecto.UUID.dump!(resolved_id),
+          organization_id: Ecto.UUID.dump!(org.id),
+          device_id: Ecto.UUID.dump!(device.id),
+          alert_type: "device_down",
+          triggered_at: old_time,
+          resolved_at: old_time,
+          storm_suppressed: false,
+          notification_sent: false,
+          inserted_at: old_time,
+          updated_at: old_time
+        },
+        %{
+          id: Ecto.UUID.dump!(unresolved_id),
+          organization_id: Ecto.UUID.dump!(org.id),
+          device_id: Ecto.UUID.dump!(device.id),
+          alert_type: "device_down",
+          triggered_at: old_time,
+          resolved_at: nil,
+          storm_suppressed: false,
+          notification_sent: false,
+          inserted_at: old_time,
+          updated_at: old_time
+        }
+      ])
+
+      assert {:ok, _result} = DataRetentionWorker.perform(%Oban.Job{args: %{}})
+
+      remaining =
+        Repo.all(
+          from(a in "alerts",
+            where: a.organization_id == type(^org.id, :binary_id),
+            select: a.id
+          )
+        )
+
+      remaining_uuids = Enum.map(remaining, &Ecto.UUID.cast!/1)
+      refute resolved_id in remaining_uuids
+      assert unresolved_id in remaining_uuids
+    end
+
+    test "works when no organizations exist" do
+      assert {:ok, result} = DataRetentionWorker.perform(%Oban.Job{args: %{}})
+      assert result.total_deleted == 0
+    end
+  end
+end