"""Background tracker — periodic coroutines that keep billing state consistent. The router keeps balances in memory (hot path). This module runs background coroutines that persist state to gists and (later) watch for payments. Coroutines: usage_flusher (30s) — write cached balances back to per-user gists payment_watcher(60s) — scan deposit addresses for incoming ETH → credit [Phase 5] price_refresher( 5m) — refresh ETH/USD price [Phase 6] Only the flusher runs in Phase 4; the others are stubbed for later phases. All coroutines are resilient: exceptions are logged, never crash the task. """ from __future__ import annotations import asyncio import logging from . import cache, registry logger = logging.getLogger("router.tracker") FLUSH_INTERVAL = 30.0 # write cached balances to gists every 30s PAYMENT_INTERVAL = 60.0 # poll deposit addresses for ETH every 60s PRICE_INTERVAL = 300.0 # refresh ETH/USD price every 5 minutes SWEEP_INTERVAL = 21600.0 # sweep stale users every 6 hours class Tracker: """Manages the background billing coroutines for the lifetime of the app.""" def __init__(self): self._tasks: list[asyncio.Task] = [] # running asyncio tasks self._running = False # guard against double-start def start(self) -> None: """Launch all background coroutines (called once at app startup).""" if self._running: # already started (e.g. test re-import) return # don't double-launch self._running = True # mark as started self._tasks = [ # create (but don't await) each coroutine task asyncio.create_task(self._run_flusher(), name="usage_flusher"), # balance persistence asyncio.create_task(self._run_payments(), name="payment_watcher"), # ETH deposit detection asyncio.create_task(self._run_price(), name="price_refresher"), # ETH/USD price cache asyncio.create_task(self._run_sweeper(), name="stale_sweeper"), # reaps inactive users ] logger.info("tracker started: %d coroutines", len(self._tasks)) # log what we launched async def stop(self) -> None: """Cancel all coroutines cleanly (called on app shutdown).""" for task in self._tasks: # signal each task to stop task.cancel() # cancel raises CancelledError inside the coroutine for task in self._tasks: # wait for them to actually stop try: # suppress the cancellation exception await task # let it finish/cancel except asyncio.CancelledError: # expected on cancel pass # normal shutdown self._tasks = [] # clear the list self._running = False # allow a restart logger.info("tracker stopped") # confirm async def _run_flusher(self) -> None: """Periodically flush cached balances to per-user gists (every 30s).""" while True: # run forever (until cancelled) try: # isolate errors so the loop never dies await asyncio.sleep(FLUSH_INTERVAL) # wait between flushes written = await cache.flush() # write all dirty balances if written: # only log if we actually wrote something logger.info("flusher: persisted %d user balances", written) # log the count except asyncio.CancelledError: # shutdown signal raise # re-raise so the task exits cleanly except Exception as exc: # any other error logger.error("flusher error: %s", exc) # log + keep looping async def _run_payments(self) -> None: """Poll deposit addresses on Base for incoming ETH → credits (every 60s).""" from . import payments # local import (avoids loading RPC deps at module import) while True: # run forever try: # isolate errors await asyncio.sleep(PAYMENT_INTERVAL) # wait between scans credited = await payments.scan_all() # scan all user addresses if credited: # at least one user was credited logger.info("payments: %d deposits processed", credited) # log except asyncio.CancelledError: # shutdown raise # exit cleanly except Exception as exc: # any error logger.error("payment watcher error: %s", exc) # log + keep looping async def _run_price(self) -> None: """Refresh the ETH/USD price from CoinGecko and cache it (every 5 min).""" from . import price # local import from ..forwarder import get_shared_client # shared httpx pool while True: # run forever try: # isolate errors await asyncio.sleep(PRICE_INTERVAL) # wait between refreshes client = get_shared_client() # borrow the shared pool usd = await price.fetch_eth_usd(client) # fetch from CoinGecko if usd and usd > 0: # got a valid price await registry.set_price(usd) # cache in the registry gist logger.info("price refreshed: ETH @ $%.2f", usd) # log except asyncio.CancelledError: # shutdown raise # exit cleanly except Exception as exc: # any error logger.error("price refresher error: %s", exc) # log + keep looping async def _run_sweeper(self) -> None: """Reap dormant accounts: users with 0 credits + no activity in 30 days. Marks them 'dormant' (not deleted — their balance + history are kept so a returning user can still deposit and resume). This keeps the registry lean and the payment-watcher loop from polling dead addresses forever. """ import time # stdlib epoch math DORMANT_SECONDS = 30 * 24 * 3600 # 30 days of inactivity while True: # run forever try: # isolate errors await asyncio.sleep(SWEEP_INTERVAL) # wait between sweeps (6h) reg = await registry.load() # fresh registry read now = time.time() # current epoch swept = 0 # count of accounts marked dormant this round for u in reg.get("users", []): # every registered user if u.get("status") != "active": # already dormant/closed continue # skip bal = await cache.get_balance(u["user_id"]) # their cached balance if bal > 0: # still has credits — keep active continue # skip last = u.get("last_active_epoch", 0) # last activity timestamp if not last: # never recorded (legacy signup) continue # skip (don't reap unknowns) if (now - last) < DORMANT_SECONDS: # active within the window continue # skip u["status"] = "dormant" # mark dormant (keeps data, stops polling) swept += 1 # tally if swept: # at least one account reaped await registry.save(reg) # persist the status changes logger.info("sweeper: %d dormant accounts marked", swept) # log except asyncio.CancelledError: # shutdown raise # exit cleanly except Exception as exc: # any error logger.error("sweeper error: %s", exc) # log + keep looping # module-level singleton (one tracker per process) TRACKER = Tracker()