feat(maintenance): scheduled + manual DB VACUUM ANALYZE + bloat readout

The TABLESAMPLE showcase reads physical blocks (bloat-sensitive), and the
periodic prune/backfill/recovery tasks churn dead tuples faster than
autovacuum always keeps up — so explicit maintenance earns its keep here.

- tasks.maintenance.vacuum_analyze: VACUUM (ANALYZE) over high-churn tables
  (VACUUM_TABLES) on an AUTOCOMMIT connection (VACUUM can't run in a txn).
  Scheduled weekly via Beat; also operator-triggerable.
- _sync_engine.get_sync_engine(): expose the process engine for the
  autocommit connection.
- GET  /api/admin/maintenance/db-stats: per-table n_live/n_dead/dead_pct +
  last (auto)vacuum/analyze from pg_stat_user_tables — visibility, not a
  black box.
- POST /api/admin/maintenance/vacuum: enqueue the task on demand.

Tests: vacuum task runs + reports tables; db-stats shape; trigger queues.

Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>

tests/test_api_admin.py

@@ -409,3 +409,29 @@ async def test_purge_legacy_commit_deletes_only_legacy(client, db):
         )
     )).scalar_one()
     assert gone == 0
+
+
+# --- DB maintenance: bloat readout + manual VACUUM trigger ----------
+
+
+@pytest.mark.asyncio
+async def test_db_stats_returns_table_bloat_shape(client):
+    resp = await client.get("/api/admin/maintenance/db-stats")
+    assert resp.status_code == 200
+    body = await resp.get_json()
+    assert "tables" in body
+    names = {t["table"] for t in body["tables"]}
+    assert "image_record" in names
+    for t in body["tables"]:
+        assert {"table", "live", "dead", "dead_pct", "last_vacuum"} <= set(t)
+
+
+@pytest.mark.asyncio
+async def test_trigger_vacuum_queues_the_task(client, monkeypatch):
+    from backend.app.tasks import maintenance
+
+    calls = []
+    monkeypatch.setattr(maintenance.vacuum_analyze, "delay", lambda: calls.append(1))
+    resp = await client.post("/api/admin/maintenance/vacuum")
+    assert resp.status_code == 202
+    assert calls == [1]

tests/test_maintenance.py

@@ -638,3 +638,12 @@ def test_recover_stalled_download_dedupes_per_source(db_sync):
         select(Source.consecutive_failures).where(Source.id == sid)
     ).scalar_one()
     assert failures == 1
+
+
+def test_vacuum_analyze_runs_over_high_churn_tables():
+    """VACUUM (ANALYZE) runs (on its own AUTOCOMMIT connection) and reports the
+    tables it touched."""
+    from backend.app.tasks.maintenance import VACUUM_TABLES, vacuum_analyze
+
+    result = vacuum_analyze.apply().get()
+    assert result["vacuumed"] == list(VACUUM_TABLES)