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
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
|
"""Scheduled cleanup tasks — IP log purge, and unhosted group collection."""
import asyncio
import logging
from datetime import UTC, datetime, timedelta
from sqlalchemy import delete, select
from sqlalchemy.ext.asyncio import AsyncSession
from meshbay_hub.db.models import EmailVerification, Group, GroupInviteLink, 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(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(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(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 purge_invite_links(db: AsyncSession) -> int:
"""Unused invitation links once expired; used ones after a month."""
from meshbay_hub.api.invite_links import KEEP_REDEEMED
now = datetime.now(UTC)
result = await db.execute(delete(GroupInviteLink).where(
(GroupInviteLink.redeemed_by.is_(None) & (GroupInviteLink.expires_at <= now))
| (GroupInviteLink.redeemed_at < now - KEEP_REDEEMED)))
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)
# Every name anybody types at the sign-in form is a row,
# real or not; once its window has passed, nothing reads it.
from meshbay_hub import login_throttle
throttled = await login_throttle.purge_expired(db)
if throttled:
log.info("Purged %d expired sign-in counters", throttled)
links = await purge_invite_links(db)
if links:
log.info("Purged %d spent or expired invitation links", links)
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(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]
|