feat(trash): restore/list_trash/purge/purge_expired + alive() helper

src/fabledassistant/services/trash.py

@@ -94,3 +94,96 @@ async def delete(user_id: int, entity_type: str, entity_id: int) -> str | None:
         await _cascade(session, user_id, entity_type, entity_id, batch, now)
         await session.commit()
     return batch
+
+
+# All soft-deletable models, and their trash-listing type label.
+_ALL = [Note, Event, Project, Milestone, Rulebook, RulebookTopic, Rule]
+_TYPE = {
+    Note: "note", Event: "event", Project: "project", Milestone: "milestone",
+    Rulebook: "rulebook", RulebookTopic: "topic", Rule: "rule",
+}
+
+
+def alive(stmt, model):
+    """Append a 'not trashed' filter to a select. Use in every live read."""
+    return stmt.where(model.deleted_at.is_(None))
+
+
+async def restore(user_id: int, batch_id: str) -> int:
+    """Clear deleted_at for every row in the batch. Returns rows restored."""
+    n = 0
+    async with async_session() as session:
+        for model in _ALL:
+            res = await session.execute(
+                update(model)
+                .where(model.deleted_batch_id == batch_id)
+                .values(deleted_at=None, deleted_batch_id=None)
+            )
+            n += res.rowcount or 0
+        await session.commit()
+    return n
+
+
+async def purge(user_id: int, batch_id: str) -> int:
+    """Hard-delete every row in the batch. Irreversible."""
+    from sqlalchemy import delete as sql_delete
+    n = 0
+    async with async_session() as session:
+        for model in _ALL:
+            res = await session.execute(
+                sql_delete(model).where(model.deleted_batch_id == batch_id)
+            )
+            n += res.rowcount or 0
+        await session.commit()
+    return n
+
+
+async def list_trash(user_id: int) -> list[dict]:
+    """Trashed rows across all soft-deletable tables, grouped by batch_id."""
+    batches: dict[str, dict] = {}
+    async with async_session() as session:
+        for model in _ALL:
+            rows = (await session.execute(
+                select(model).where(model.deleted_at.isnot(None))
+            )).scalars().all()
+            for r in rows:
+                grp = batches.setdefault(
+                    r.deleted_batch_id,
+                    {"batch_id": r.deleted_batch_id,
+                     "deleted_at": r.deleted_at.isoformat() if r.deleted_at else None,
+                     "items": []},
+                )
+                grp["items"].append({
+                    "type": _TYPE[model], "id": r.id,
+                    "title": getattr(r, "title", "") or "",
+                })
+    out = list(batches.values())
+    for g in out:
+        lead = g["items"][0]
+        g["summary"] = lead["title"] or f"{lead['type']} {lead['id']}"
+        g["count"] = len(g["items"])
+    out.sort(key=lambda g: g["deleted_at"] or "", reverse=True)
+    return out
+
+
+async def purge_expired(retention_days: int) -> int:
+    """Cron entry: hard-delete rows trashed more than retention_days ago.
+
+    retention_days <= 0 disables auto-purge (returns 0 without touching anything).
+    """
+    from datetime import timedelta
+    from sqlalchemy import delete as sql_delete
+    if retention_days <= 0:
+        return 0
+    cutoff = datetime.now(timezone.utc) - timedelta(days=retention_days)
+    n = 0
+    async with async_session() as session:
+        for model in _ALL:
+            res = await session.execute(
+                sql_delete(model).where(
+                    model.deleted_at.isnot(None), model.deleted_at < cutoff
+                )
+            )
+            n += res.rowcount or 0
+        await session.commit()
+    return n

tests/test_services_trash.py

@@ -77,3 +77,74 @@ async def test_delete_rulebook_cascades_topics_and_rules():
         batch = await delete(user_id=7, entity_type="rulebook", entity_id=2)
     assert isinstance(batch, str)
     assert session.execute.await_count == 5
+
+
+def _rowcount_result(n):
+    r = MagicMock()
+    r.rowcount = n
+    return r
+
+
+@pytest.mark.asyncio
+async def test_restore_clears_batch_across_all_models():
+    session = _make_mock_session()
+    # 7 models, each returns a rowcount
+    session.execute = AsyncMock(side_effect=[_rowcount_result(i) for i in [2, 0, 1, 1, 0, 0, 0]])
+    with patch("fabledassistant.services.trash.async_session") as cls:
+        cls.return_value = session
+        from fabledassistant.services.trash import restore
+        n = await restore(user_id=1, batch_id="abc")
+    assert n == 4
+    assert session.execute.await_count == 7
+    assert session.commit.called
+
+
+@pytest.mark.asyncio
+async def test_purge_expired_skips_when_retention_zero():
+    session = _make_mock_session()
+    session.execute = AsyncMock()
+    with patch("fabledassistant.services.trash.async_session") as cls:
+        cls.return_value = session
+        from fabledassistant.services.trash import purge_expired
+        n = await purge_expired(0)
+    assert n == 0
+    assert not session.execute.called  # never opens a delete
+
+
+@pytest.mark.asyncio
+async def test_purge_expired_deletes_across_models_when_positive():
+    session = _make_mock_session()
+    session.execute = AsyncMock(side_effect=[_rowcount_result(1) for _ in range(7)])
+    with patch("fabledassistant.services.trash.async_session") as cls:
+        cls.return_value = session
+        from fabledassistant.services.trash import purge_expired
+        n = await purge_expired(90)
+    assert n == 7
+    assert session.execute.await_count == 7
+
+
+@pytest.mark.asyncio
+async def test_list_trash_groups_by_batch():
+    session = _make_mock_session()
+
+    def _note(id, batch, title):
+        from datetime import datetime, timezone
+        n = MagicMock()
+        n.id = id; n.deleted_batch_id = batch; n.title = title
+        n.deleted_at = datetime(2026, 5, 28, tzinfo=timezone.utc)
+        return n
+
+    # Note model returns 2 rows in one batch; the other 6 models return none.
+    note_rows = MagicMock()
+    note_rows.scalars.return_value.all.return_value = [_note(1, "b1", "Task A"), _note(2, "b1", "Sub")]
+    empty = MagicMock()
+    empty.scalars.return_value.all.return_value = []
+    session.execute = AsyncMock(side_effect=[note_rows] + [empty] * 6)
+    with patch("fabledassistant.services.trash.async_session") as cls:
+        cls.return_value = session
+        from fabledassistant.services.trash import list_trash
+        out = await list_trash(user_id=1)
+    assert len(out) == 1
+    assert out[0]["batch_id"] == "b1"
+    assert out[0]["count"] == 2
+    assert out[0]["summary"] == "Task A"