src/services/billing/tracker.py

"""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()