"""Ingestion worker: pair discovery (hourly), OHLC accumulation (every N s), WS streaming, retention."""

from __future__ import annotations

import asyncio
import time

import redis.asyncio as aioredis

from ktrade.config import Settings
from ktrade.db.pool import create_pool
from ktrade.ingestion.backfill import backfill_all
from ktrade.ingestion.pairs import refresh_pairs
from ktrade.ingestion.stream import StreamIngestor
from ktrade.kraken.models import PairMeta
from ktrade.kraken.rest import KrakenREST
from ktrade.logging import get_logger

log = get_logger("worker.ingest")


async def run_worker(s: Settings) -> None:
    pool = await create_pool(s.database_url, max_size=8)
    r = aioredis.from_url(s.redis_url, decode_responses=True)
    rest = KrakenREST(s.kraken_rest_url, tier=s.kraken_tier)  # public endpoints only — no credentials in the ingest worker
    pairs: dict[str, PairMeta] = {}
    while not pairs:
        try:
            pairs = await refresh_pairs(rest, pool, s.universe)
        except Exception as e:  # noqa: BLE001
            log.warning("worker.pairs.retry", error=str(e))
            await asyncio.sleep(10)
    # Kraken WS v2 accepts a single ohlc interval per symbol per connection → stream 1-min bars only; REST accumulates the rest.
    stream = StreamIngestor(pool, r, s.kraken_ws_url, pairs, s.book_depth, [1])

    async def backfill_loop() -> None:
        while True:
            t0 = time.time()
            try:
                res = await backfill_all(rest, pool, pairs, s.candle_intervals)
                await r.set("ktrade:backfill:last", str(time.time()))
                log.info("worker.backfill.done", pairs=len(pairs), rows=sum(res.values()), seconds=round(time.time() - t0, 1))
            except Exception as e:  # noqa: BLE001
                log.warning("worker.backfill.error", error=str(e))
            await asyncio.sleep(max(5, s.backfill_poll_seconds - (time.time() - t0)))

    async def pairs_loop() -> None:
        while True:
            await asyncio.sleep(3600)
            try:
                new = await refresh_pairs(rest, pool, s.universe)
                pairs.update(new)
            except Exception as e:  # noqa: BLE001
                log.warning("worker.pairs.error", error=str(e))

    async def retention_loop() -> None:
        while True:
            try:
                await pool.execute("DELETE FROM book_samples WHERE ts < $1", int(time.time()) - 14 * 86400)
                await pool.execute("DELETE FROM trades WHERE ts < $1", time.time() - 30 * 86400)
            except Exception as e:  # noqa: BLE001
                log.warning("worker.retention.error", error=str(e))
            await asyncio.sleep(3600)

    log.info("worker.started", universe=list(pairs), intervals=s.candle_intervals)
    await asyncio.gather(backfill_loop(), pairs_loop(), retention_loop(), stream.run())