Spaces:

apiarium
/

labs

Sleeping

labs / src /services /billing /tracker.py

deploy: unified router + dreamy website (2026-06-16T09:46:52Z)

c1a683f 5 months ago

7.7 kB

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

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