"""Secondary-side polling task.

Owns ONE long-lived `psycopg.Connection` to a read-only role on the

cluster. Every `poll_interval_ms` it executes:

SELECT id, topic, payload, recorded_at

FROM events

WHERE id > %s

ORDER BY id ASC

LIMIT 1000;

…where `%s` is the highest id we've seen, then publishes each new row

to the broadcast Hub. New rows replicated from the primary appear here

within (replication lag + one poll interval) — typically <100ms on a

healthy cluster.

The connection is opened ONCE at startup and held for the process

lifetime. The mutual-RA-TLS handshake happens once. Per-poll cost is

just a single SELECT round-trip over an existing TLS session.

On `psycopg.OperationalError` (leader rotation, sidecar restart, network

glitch) we close the dead connection, call `connection.open_secondary`

again to re-handshake, and resume. That's the only place we eat a

fresh handshake in steady state.

"""

from __future__ import annotations

import asyncio

import logging

from typing import Any

import psycopg

from .broadcast import Hub

from .config import Settings

from .connection import open_secondary

log = logging.getLogger(__name__)

class Poller:

"""Async task that drains new `events` rows from the secondary.

Construct with the Settings + Hub. Call `start()` to spawn the

background task; `stop()` to cancel and close the connection.

"""

def __init__(self, settings: Settings, hub: Hub) -> None:

self._settings = settings

self._hub = hub

self._conn: psycopg.Connection | None = None

self._task: asyncio.Task[None] | None = None

self._last_seen: int = 0

self._stop_event = asyncio.Event()

async def start(self) -> None:

"""Open the long-lived secondary connection and spawn the task."""

# connect_via_manifest is sync; it blocks the loop briefly during

# the RA-TLS handshake. Acceptable at startup; not in the hot path.

loop = asyncio.get_running_loop()

self._conn = await loop.run_in_executor(None, open_secondary, self._settings)

# Bootstrap the cursor — start from the current max(id) so the

# first poll doesn't replay every historical event to the

# WebSocket. New WS connections that want history can do a

# SELECT at subscribe time.

await loop.run_in_executor(None, self._init_cursor)

self._task = asyncio.create_task(self._run(), name="teesql-example-poller")

log.info(

"poller started: poll_interval=%dms last_seen=%d",

self._settings.poll_interval_ms,

self._last_seen,

)

async def stop(self) -> None:

"""Cancel the task and close the connection cleanly."""

self._stop_event.set()

if self._task is not None:

self._task.cancel()

try:

await self._task

except asyncio.CancelledError:

pass

self._task = None

if self._conn is not None:

try:

self._conn.close()

except Exception as e: # noqa: BLE001

log.warning("error closing secondary connection: %s", e)

self._conn = None

log.info("poller stopped")

# ── internals ────────────────────────────────────────────────────────

def _init_cursor(self) -> None:

"""Set self._last_seen = current max(id), or 0 if table is empty."""

assert self._conn is not None

with self._conn.cursor() as cur:

cur.execute("SELECT COALESCE(max(id), 0) AS m FROM events")

row = cur.fetchone()

# Possible to be in a tx after the SELECT; commit to release any locks.

self._conn.commit()

self._last_seen = int(row["m"]) if row else 0

def _poll_once_sync(self) -> list[dict[str, Any]]:

"""Run one SELECT in the executor's thread. Returns new rows."""

assert self._conn is not None

with self._conn.cursor() as cur:

cur.execute(

"SELECT id, topic, payload, recorded_at "

"FROM events WHERE id > %s "

"ORDER BY id ASC LIMIT 1000",

(self._last_seen,),

)

rows = list(cur.fetchall())

# Read-only role → still good practice to commit so we don't

# accidentally hold a transaction snapshot open.

self._conn.commit()

return rows

async def _reconnect(self) -> None:

"""Tear down and re-open the secondary connection.

Called on OperationalError. Eats one mutual-RA-TLS handshake,

which is the price of leader rotation. last_seen is preserved

across reconnect so we don't replay or lose events.

"""

log.warning("poller: reconnecting secondary (last_seen=%d)", self._last_seen)

if self._conn is not None:

try:

self._conn.close()

except Exception: # noqa: BLE001

pass

self._conn = None

loop = asyncio.get_running_loop()

# Brief backoff so we don't hammer a freshly-rotated cluster.

await asyncio.sleep(0.5)

self._conn = await loop.run_in_executor(None, open_secondary, self._settings)

async def _run(self) -> None:

interval = self._settings.poll_interval_ms / 1000.0

loop = asyncio.get_running_loop()

while not self._stop_event.is_set():

try:

rows = await loop.run_in_executor(None, self._poll_once_sync)

except psycopg.OperationalError as e:

log.warning("poller: OperationalError %s; reconnecting", e)

try:

await self._reconnect()

except Exception as re: # noqa: BLE001

log.error("poller: reconnect failed (%s); will retry next tick", re)

await asyncio.sleep(interval)

continue

except Exception as e: # noqa: BLE001

# Any other error: log loud, sleep, keep looping. The next

# tick will retry; if the error is persistent the operator

# sees it in logs.

log.exception("poller: unexpected error: %s", e)

await asyncio.sleep(interval)

continue

if rows:

# Last id wins because rows are id-ordered.

self._last_seen = int(rows[-1]["id"])

for row in rows:

# Make the row JSON-friendly for the WebSocket.

# `payload` is already a dict (psycopg decodes JSONB).

# `recorded_at` is a datetime — isoformat for the wire.

await self._hub.publish(

{

"id": int(row["id"]),

"topic": row["topic"],

"payload": row["payload"],

"recorded_at": row["recorded_at"].isoformat(),

}

)

log.debug("poller: published %d rows (last_seen=%d)", len(rows), self._last_seen)

await asyncio.sleep(interval)