1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
|
"""Scheduled cleanup tasks — IP log purge, and unhosted group collection."""
import asyncio
import logging
from datetime import datetime, timedelta, timezone
from sqlalchemy import delete, select
from sqlalchemy.ext.asyncio import AsyncSession
from meshbay_hub.db.models import EmailVerification, Group, IPLog, User
log = logging.getLogger(__name__)
RETENTION_DAYS = 365
CLEANUP_INTERVAL_HOURS = 24
async def purge_old_ip_logs(db: AsyncSession, retention_days: int = RETENTION_DAYS) -> int:
cutoff = datetime.now(timezone.utc) - timedelta(days=retention_days)
result = await db.execute(delete(IPLog).where(IPLog.timestamp < cutoff))
await db.commit()
return result.rowcount
PENDING_USER_EXPIRY_DAYS = 7
async def purge_expired_verifications(db: AsyncSession) -> int:
now = datetime.now(timezone.utc)
result = await db.execute(
delete(EmailVerification).where(EmailVerification.expires_at < now))
await db.commit()
return result.rowcount
async def purge_stale_pending_users(db: AsyncSession,
expiry_days: int = PENDING_USER_EXPIRY_DAYS) -> int:
cutoff = datetime.now(timezone.utc) - timedelta(days=expiry_days)
result = await db.execute(
delete(User).where(User.status == "pending", User.created_at < cutoff))
await db.commit()
return result.rowcount
async def cleanup_loop(get_session):
"""Run cleanup once at startup, then every 24 hours."""
try:
while True:
try:
async with get_session() as db:
deleted = await purge_old_ip_logs(db)
if deleted:
log.info("Purged %d IP log entries older than %d days", deleted, RETENTION_DAYS)
expired = await purge_expired_verifications(db)
if expired:
log.info("Purged %d expired email verifications", expired)
stale = await purge_stale_pending_users(db)
if stale:
log.info("Purged %d stale pending users", stale)
# One row per recipient the hub has written to, and the
# window is a day: without this the table grows for the
# life of the instance and nothing ever reads the old rows.
from meshbay_hub import mail
quota = await mail.purge_expired_quota(db)
if quota:
log.info("Purged %d expired mail counters", quota)
except asyncio.CancelledError:
raise
except Exception as e:
log.error("IP log cleanup failed: %s", e)
await asyncio.sleep(CLEANUP_INTERVAL_HOURS * 3600)
except asyncio.CancelledError:
return
# ── Groups that never got a node ──────────────────────────────────────────────
UNHOSTED_GRACE_DAYS = 7
async def find_unhosted_groups(db: AsyncSession, grace_days: int = UNHOSTED_GRACE_DAYS):
"""Groups created more than `grace_days` ago that no node has ever announced.
`hosted_at` is set the first time a node registers claiming the group and is
never cleared, so this finds groups that were created and then abandoned —
not ones whose node happens to be offline today. That distinction is the
whole reason the column exists rather than a check against the live socket
registry, which would delete every group during a hub restart.
"""
cutoff = datetime.now(timezone.utc) - timedelta(days=grace_days)
result = await db.execute(
select(Group).where(Group.hosted_at.is_(None), Group.created_at < cutoff))
return list(result.scalars().all())
async def prune_unhosted_groups(db: AsyncSession, grace_days: int = UNHOSTED_GRACE_DAYS,
dry_run: bool = False) -> list[tuple[str, str]]:
"""Delete abandoned groups. Returns [(id, name)] of what was (or would be) removed.
Everything on the hub that points at the group goes with it (`purge_groups`):
there is no cascade configured, an orphan membership would keep the group in
everyone's /mine query through the join, and on PostgreSQL any remaining
reference refuses the deletion outright. Nothing on a node is touched: the
hub does not command those machines, and by definition no node ever claimed
this group anyway.
"""
doomed = await find_unhosted_groups(db, grace_days)
if not doomed or dry_run:
return [(g.id, g.name) for g in doomed]
from meshbay_hub.db.purge import purge_groups
await purge_groups(db, [g.id for g in doomed])
await db.commit()
log.info("Pruned %d group(s) that no node ever hosted", len(doomed))
return [(g.id, g.name) for g in doomed]
|