Updated the scheduled backup rention policy to all from last 24 hours, daily from last 7 days, and weekly from last 4 weeks

Author: blackcoffeexbt

flake.nix

@@ -21,7 +21,7 @@
 
   outputs = { self, nixpkgs, raspberry-pi-nix, lnbits, spark-sidecar, ... }:
   let
-    version = "0.9.1";  # Bump before each release tag to match the next tag name
+    version = "0.9.3";  # Bump before each release tag to match the next tag name
     system = "aarch64-linux";
   in
   {

nixos/admin-app/app.py

@@ -50,6 +50,7 @@
     package_encrypted_backup,
     package_plain_backup,
     read_json_file,
+    select_scheduled_backups_to_keep,
     utc_now_iso,
     validate_manifest_files,
     write_json_file,
@@ -804,13 +805,7 @@ def _local_backup_manifest(path: Path) -> dict[str, Any] | None:
 
 def _prune_scheduled_backups():
     now = datetime.now().astimezone()
-    checkpoints = [
-        ("daily", now.timestamp() - 86400),
-        ("weekly", now.timestamp() - (7 * 86400)),
-        ("monthly", now.timestamp() - (30 * 86400)),
-    ]
-    keep: set[Path] = set()
-    candidates: list[tuple[Path, float]] = []
+    candidates: list[tuple[Path, datetime]] = []
 
     for path in RECOVERY_BACKUP_DIR.glob("lnbitsbox-recovery-*.zip"):
         manifest = _local_backup_manifest(path)
@@ -819,18 +814,12 @@ def _prune_scheduled_backups():
         created_at = parse_iso_datetime(manifest.get("created_at"))
         if created_at is None:
             continue
-        candidates.append((path, created_at.timestamp()))
+        candidates.append((path, created_at))
 
     if not candidates:
         return
 
-    candidates.sort(key=lambda item: item[1], reverse=True)
-    keep.add(candidates[0][0])
-
-    for _, threshold in checkpoints:
-        eligible = [item for item in candidates if item[1] <= threshold]
-        if eligible:
-            keep.add(max(eligible, key=lambda item: item[1])[0])
+    keep = select_scheduled_backups_to_keep(candidates, now=now)
 
     for path, _ in candidates:
         if path in keep:

nixos/admin-app/recovery_utils.py

@@ -4,7 +4,7 @@
 import json
 import secrets
 import zipfile
-from datetime import datetime, timezone
+from datetime import datetime, timedelta, timezone
 from pathlib import Path
 from typing import Any
 
@@ -217,3 +217,54 @@ def parse_iso_datetime(value: str | None) -> datetime | None:
     if parsed.tzinfo is None:
         return parsed.replace(tzinfo=timezone.utc)
     return parsed
+
+
+def select_scheduled_backups_to_keep(
+    backups: list[tuple[Path, datetime]],
+    *,
+    now: datetime | None = None,
+) -> set[Path]:
+    if not backups:
+        return set()
+
+    if now is None:
+        now = datetime.now(timezone.utc)
+    elif now.tzinfo is None:
+        now = now.replace(tzinfo=timezone.utc)
+
+    normalized = []
+    for path, created_at in backups:
+        if created_at.tzinfo is None:
+            created_at = created_at.replace(tzinfo=timezone.utc)
+        normalized.append((path, created_at.astimezone(now.tzinfo)))
+
+    normalized.sort(key=lambda item: item[1], reverse=True)
+    keep: set[Path] = {normalized[0][0]}
+
+    hourly_cutoff = now - timedelta(hours=24)
+    daily_cutoff = now - timedelta(days=7)
+    weekly_cutoff = now - timedelta(weeks=4)
+
+    daily_buckets: dict[datetime.date, tuple[Path, datetime]] = {}
+    weekly_buckets: dict[tuple[int, int], tuple[Path, datetime]] = {}
+
+    for path, created_at in normalized:
+        if created_at >= hourly_cutoff:
+            keep.add(path)
+            continue
+        if created_at >= daily_cutoff:
+            bucket = created_at.date()
+            existing = daily_buckets.get(bucket)
+            if existing is None or created_at > existing[1]:
+                daily_buckets[bucket] = (path, created_at)
+            continue
+        if created_at >= weekly_cutoff:
+            iso_year, iso_week, _ = created_at.isocalendar()
+            bucket = (iso_year, iso_week)
+            existing = weekly_buckets.get(bucket)
+            if existing is None or created_at > existing[1]:
+                weekly_buckets[bucket] = (path, created_at)
+
+    keep.update(path for path, _ in daily_buckets.values())
+    keep.update(path for path, _ in weekly_buckets.values())
+    return keep

nixos/admin-app/tests/test_recovery_utils.py

@@ -1,6 +1,8 @@
 import io
 import unittest
 import zipfile
+from datetime import datetime, timedelta, timezone
+from pathlib import Path
 
 try:
     from recovery_utils import (
@@ -11,6 +13,7 @@
         load_backup_container,
         package_encrypted_backup,
         package_plain_backup,
+        select_scheduled_backups_to_keep,
         validate_manifest_files,
     )
     RECOVERY_IMPORT_ERROR = None
@@ -104,6 +107,43 @@ def test_compatibility_report(self):
         self.assertEqual(compatibility_report("0.2.0", "0.1.49")["level"], "warn")
         self.assertEqual(compatibility_report("1.0.0", "0.1.49")["level"], "error")
 
+    def test_scheduled_backup_retention_keeps_recent_daily_and_weekly_windows(self):
+        now = datetime(2026, 3, 17, 12, 0, tzinfo=timezone.utc)
+        backups = [
+            (Path("/tmp/hourly-1.zip"), now - timedelta(hours=1)),
+            (Path("/tmp/hourly-20.zip"), now - timedelta(hours=20)),
+            (Path("/tmp/day-2-new.zip"), now - timedelta(days=2, hours=1)),
+            (Path("/tmp/day-2-old.zip"), now - timedelta(days=2, hours=6)),
+            (Path("/tmp/day-6.zip"), now - timedelta(days=6, hours=2)),
+            (Path("/tmp/week-2-new.zip"), now - timedelta(days=10, hours=1)),
+            (Path("/tmp/week-2-old.zip"), now - timedelta(days=12)),
+            (Path("/tmp/week-4.zip"), now - timedelta(days=24)),
+            (Path("/tmp/older-than-window.zip"), now - timedelta(days=40)),
+        ]
+
+        keep = select_scheduled_backups_to_keep(backups, now=now)
+
+        self.assertIn(Path("/tmp/hourly-1.zip"), keep)
+        self.assertIn(Path("/tmp/hourly-20.zip"), keep)
+        self.assertIn(Path("/tmp/day-2-new.zip"), keep)
+        self.assertNotIn(Path("/tmp/day-2-old.zip"), keep)
+        self.assertIn(Path("/tmp/day-6.zip"), keep)
+        self.assertIn(Path("/tmp/week-2-new.zip"), keep)
+        self.assertNotIn(Path("/tmp/week-2-old.zip"), keep)
+        self.assertIn(Path("/tmp/week-4.zip"), keep)
+        self.assertNotIn(Path("/tmp/older-than-window.zip"), keep)
+
+    def test_scheduled_backup_retention_keeps_latest_even_when_outside_windows(self):
+        now = datetime(2026, 3, 17, 12, 0, tzinfo=timezone.utc)
+        backups = [
+            (Path("/tmp/latest-old.zip"), now - timedelta(days=60)),
+            (Path("/tmp/older.zip"), now - timedelta(days=90)),
+        ]
+
+        keep = select_scheduled_backups_to_keep(backups, now=now)
+
+        self.assertEqual(keep, {Path("/tmp/latest-old.zip")})
+
 
 if __name__ == "__main__":
     unittest.main()