# SPDX-License-Identifier: Apache-2.0
# Copyright 2026 Aaron K. Clark
"""Multi-tenant cron worker: poll every active tenant in one pass.

A bare ``applytrack poll`` (no ``--tenant``) drives this. Where
:func:`applytrack.poll.run_poll` serves one tenant end-to-end, the worker
fetches each enabled source **once** across all tenants and then routes the
shared buckets per tenant: the public job boards are the slow, rate-limited part,
and re-fetching Remotive once per user would be both wasteful and a fast way to
get the host blocked.

The one privileged, cross-tenant query lives here —
``SELECT id FROM users WHERE status = 'active'``. Everything downstream is built
per-tenant through a :class:`~applytrack.db.PollRepo`, so the ``WHERE tenant_id``
choke-point still holds for every read and write. A failure polling one tenant is
isolated (caught, the rest of the run continues) so one bad profile can't starve
the others.
"""

from __future__ import annotations

import logging
from collections.abc import Callable, Iterable
from datetime import datetime, timezone
from functools import partial
from typing import Protocol

import httpx
import psycopg

from applytrack import handshake, linkedin, mygreenhouse
from applytrack.criteria import AtsBoard, Criteria
from applytrack.db import PollRepo
from applytrack.linkcheck import BROWSER_HEADERS
from applytrack.poll import (
    SOURCE_FETCHERS,
    LeadRepo,
    LinkCache,
    Listing,
    ThreadClients,
    _host,
    _select_for_profile,
    make_ats_fetcher,
    make_rss_fetcher,
    run_bounded,
    score_and_stage,
)

logger = logging.getLogger(__name__)

_TENANT_POLL_LOCK_PREFIX = "applytrack:tenant-poll:"


class TenantRepo(LeadRepo, Protocol):
    """A :class:`~applytrack.poll.LeadRepo` that can also load its own profile.

    :class:`~applytrack.db.PollRepo` already satisfies this; the offline worker
    tests pass an in-memory fake shaped the same way.
    """

    def load_profile(self) -> Criteria:
        """Load this tenant's discovery :class:`~applytrack.criteria.Criteria`."""
        ...


def _active_tenant_ids(conn: psycopg.Connection) -> list[int]:
    """The privileged cross-tenant read: ids of every active user, ascending.

    This is the *only* query in either runtime not scoped to a single tenant; it
    exists so the cron worker knows whom to poll. Every read/write it fans out to
    is re-scoped through a per-tenant :class:`~applytrack.db.PollRepo`.
    """
    with conn.cursor() as cur:
        cur.execute("SELECT id FROM users WHERE status = 'active' ORDER BY id")
        return [int(row[0]) for row in cur.fetchall()]


def _try_tenant_poll_lock(conn: psycopg.Connection, tenant_id: int) -> bool:
    """Acquire this tenant's PostgreSQL session advisory lock without waiting."""
    with conn.cursor() as cur:
        cur.execute(
            "SELECT pg_try_advisory_lock(hashtextextended(%s, 0))",
            (f"{_TENANT_POLL_LOCK_PREFIX}{tenant_id}",),
        )
        row = cur.fetchone()
    return bool(row and row[0])


def _release_tenant_poll_lock(conn: psycopg.Connection, tenant_id: int) -> None:
    """Release a tenant poll lock acquired by :func:`_try_tenant_poll_lock`."""
    with conn.cursor() as cur:
        cur.execute(
            "SELECT pg_advisory_unlock(hashtextextended(%s, 0))",
            (f"{_TENANT_POLL_LOCK_PREFIX}{tenant_id}",),
        )


def _gather_by_source(
    profiles: Iterable[Criteria],
    limit: int,
    *,
    ats_only: bool = False,
    failures: dict[str, str] | None = None,
) -> dict[str, list[Listing]]:
    """Fetch every source enabled by *any* profile once, keyed by source id.

    Built-in sources key by name (``"remotive"``); ATS boards key by
    ``"{provider}:{slug}"`` and custom RSS feeds by ``"rss:{url}"`` — the same keys
    :func:`_select_for_profile` routes by, so two tenants following the same feed
    cost one request. A failing source yields an empty bucket and is logged at
    WARNING (rather than aborting the gather or vanishing silently), so a
    dead/renamed source is visible in the logs instead of looking like an empty
    market for every tenant.

    ``ats_only`` gathers *only* the tenants' ATS boards (Greenhouse/Lever/Ashby and
    the like), skipping the built-in aggregators and custom RSS feeds. It is the
    freshness fast lane: ATS boards close fastest, and hitting only the per-employer
    board APIs on a short cadence does not touch the shared aggregators' rate limits,
    so this can run far more often than the full hourly poll.

    ``failures``, when given, collects ``{source key: what went wrong}`` for each source
    that raised, so each tenant's ``poll_runs`` row can name its failing sources (#359).
    """
    builtin: set[str] = set()
    boards: dict[str, AtsBoard] = {}
    feeds: dict[str, str] = {}
    for profile in profiles:
        for name, on in profile.sources.items():
            if on and name in SOURCE_FETCHERS:
                builtin.add(name)
        for board in profile.ats_boards:
            boards[f"{board.provider}:{board.slug}"] = board
        for url in profile.rss_feeds:
            feeds[f"rss:{url}"] = url
    if ats_only:
        builtin.clear()
        feeds.clear()

    # (host key, source key, thunk). Every ATS board of one provider shares that
    # provider's API host, so they are gated as one host; each feed by its own host.
    tasks: list[tuple[str, str, Callable[[], list[Listing]]]] = []
    feed_clients = ThreadClients(timeout=20.0)
    gathered: dict[str, list[Listing]] = {}
    with httpx.Client(timeout=20.0, follow_redirects=True, headers=BROWSER_HEADERS) as client:
        for name in sorted(builtin):
            tasks.append((name, name, partial(SOURCE_FETCHERS[name], client, limit)))
        for key, board in boards.items():
            fetcher = make_ats_fetcher(board)
            if fetcher is None:
                continue
            tasks.append((board.provider, key, partial(fetcher, client, limit)))
        for key, url in feeds.items():
            # Not `client`: a user-supplied feed URL is fetched only through an
            # SSRF-guarded client (see poll.fetch_rss) — one per pool thread, reused
            # across that thread's feeds (#347).
            tasks.append((_host(url), key, partial(_fetch_feed, url, feed_clients, limit)))
        try:
            outcomes = run_bounded((host, thunk) for host, _, thunk in tasks)
        finally:
            feed_clients.close()
    for (_, key, _), outcome in zip(tasks, outcomes, strict=True):
        if isinstance(outcome, Exception):
            # One bad source must not abort the gather.
            gathered[key] = []
            logger.warning("poll source %s failed", key, exc_info=outcome)
            if failures is not None:
                failures[key] = _describe(outcome)
        else:
            gathered[key] = outcome
    return gathered


def _fetch_feed(url: str, clients: ThreadClients, limit: int) -> list[Listing]:
    """One custom feed, on this pool thread's own SSRF-safe client."""
    return make_rss_fetcher(url)(clients.get(), limit)


# How much of a failure's message a poll_runs row keeps: enough to tell a 403 from a
# timeout, not a stack trace.
_MAX_ERROR_CHARS = 300


def _describe(exc: BaseException) -> str:
    """One line for a failed source: the exception type and its message, capped."""
    text = f"{type(exc).__name__}: {exc}".strip().rstrip(":")
    return text[:_MAX_ERROR_CHARS]


def _source_keys(profile: Criteria, *, ats_only: bool) -> list[str]:
    """The shared-gather keys this tenant's profile draws on — the same keys
    :func:`_gather_by_source` fetches and :func:`_select_for_profile` routes by."""
    keys: list[str] = []
    if not ats_only:
        keys += [n for n, on in profile.sources.items() if on and n in SOURCE_FETCHERS]
    keys += [f"{b.provider}:{b.slug}" for b in profile.ats_boards]
    if not ats_only:
        keys += [f"rss:{url}" for url in profile.rss_feeds]
    return keys


def _record_run(
    repo: TenantRepo,
    tenant_id: int,
    started_at: datetime,
    *,
    ats_only: bool,
    leads_added: int,
    errors: list[dict[str, str]],
) -> None:
    """Write this tenant's ``poll_runs`` row (#359). Best effort, like :func:`_mark_polled`:
    a database the api has not migrated yet must not fail a poll, and the offline fakes
    may have no such method."""
    record = getattr(repo, "record_run", None)
    if record is None:
        return
    try:
        record(started_at, ats_only=ats_only, leads_added=leads_added, errors=errors)
    except Exception:  # noqa: BLE001 - the leads are in; the run row is bookkeeping
        logger.warning("could not record the poll run for tenant %s", tenant_id, exc_info=True)


def _mark_polled(repo: TenantRepo, tenant_id: int) -> None:
    """Record the finished pass for the operator's usage view (#358). Best effort: a
    database the api has not migrated yet (no ``last_polled_at``) must not fail a poll
    that already staged its leads, and the offline fakes have no such method."""
    mark = getattr(repo, "mark_polled", None)
    if mark is None:
        return
    try:
        mark()
    except Exception:  # noqa: BLE001 - the leads are in; the stamp is bookkeeping
        logger.warning("could not stamp last_polled_at for tenant %s", tenant_id, exc_info=True)


def run_all_tenants(
    conn: psycopg.Connection | None = None,
    *,
    limit_per_source: int = 40,
    tenant_ids: Iterable[int] | None = None,
    repo_for: Callable[[int], TenantRepo] | None = None,
    gathered: dict[str, list[Listing]] | None = None,
    verify_links: bool = True,
    locked_tenant_ids: set[int] | None = None,
    ats_only: bool = False,
) -> dict[int, list[str]]:
    """Poll every active tenant, returning ``{tenant_id: [staged slug names]}``.

    Production passes only ``conn`` (autocommit): the active-tenant list and each
    tenant's :class:`~applytrack.db.PollRepo` are derived from it, sources are
    fetched once, and the shared buckets are routed per profile. The ``tenant_ids``
    / ``repo_for`` / ``gathered`` seams let the offline tests drive the same
    fan-out with in-memory fakes and a fixed listing set (no DB, no network).

    Production also holds a PostgreSQL session advisory lock for each tenant from
    profile setup through source gathering and staging. A concurrent drain,
    scheduled run, or second poller container skips an already-locked tenant
    without waiting. ``locked_tenant_ids`` lets the queue drainer re-enqueue those
    skipped requests for a later attempt.

    Per-tenant failures are isolated: a tenant whose poll raises is recorded with
    an empty result and the run continues.

    ``ats_only`` restricts the shared gather to the tenants' ATS boards — the
    freshness fast lane (see :func:`_gather_by_source`). It is ignored when a
    ready-made ``gathered`` is supplied.
    """
    started_at = datetime.now(timezone.utc)
    if repo_for is None:
        if conn is None:
            raise ValueError("run_all_tenants needs either conn or repo_for")
        bound = conn

        def repo_for(tid: int) -> TenantRepo:
            return PollRepo(bound, tid)

    if tenant_ids is None:
        if conn is None:
            raise ValueError("run_all_tenants needs either conn or tenant_ids")
        tenant_ids = _active_tenant_ids(conn)
    tenant_ids = list(tenant_ids)

    acquired_locks: list[int] = []
    pollable_tenant_ids: list[int] = []
    for tid in tenant_ids:
        if conn is None:
            pollable_tenant_ids.append(tid)
            continue
        try:
            if _try_tenant_poll_lock(conn, tid):
                acquired_locks.append(tid)
                pollable_tenant_ids.append(tid)
                continue
            if locked_tenant_ids is not None:
                locked_tenant_ids.add(tid)
            logger.info(
                "poll skipped for tenant %s: another poll is already active",
                tid,
            )
        except Exception:  # noqa: BLE001 - one lock failure must not abort the run
            logger.warning("poll lock acquisition failed for tenant %s", tid, exc_info=True)

    # Build the per-tenant repo + profile up front: this is where each tenant's
    # WHERE tenant_id scoping is fixed for the rest of the run. A tenant whose
    # setup raises is isolated here so the shared gather still runs for the rest.
    repos: dict[int, TenantRepo] = {}
    profiles: dict[int, Criteria] = {}
    # What failed, for each tenant's poll_runs row (#359): the shared sources once, the
    # signed-in ones per tenant.
    shared_failures: dict[str, str] = {}
    results: dict[int, list[str]] = {
        tid: [] for tid in tenant_ids if tid not in pollable_tenant_ids
    }
    try:
        for tid in pollable_tenant_ids:
            repo = None
            try:
                repo = repo_for(tid)
                profiles[tid] = repo.load_profile()
                repos[tid] = repo
            except Exception as exc:  # noqa: BLE001 - isolate one tenant's failure
                results[tid] = []
                logger.warning("poll setup failed for tenant %s", tid, exc_info=True)
                if repo is not None:
                    _record_run(
                        repo, tid, started_at, ats_only=ats_only, leads_added=0,
                        errors=[{"source": "poll", "error": _describe(exc)}],
                    )

        if gathered is None:
            gathered = _gather_by_source(
                profiles.values(), limit_per_source, ats_only=ats_only, failures=shared_failures
            )
        # Every tenant stages from the same shared listings: what one tenant's link
        # checks learned about a URL holds for the rest of the run (#347).
        link_cache = LinkCache()

        for tid in pollable_tenant_ids:
            if tid not in repos:
                continue  # setup failed above; its empty result is already recorded
            own_failures: dict[str, str] = {}
            try:
                listings = _select_for_profile(gathered, profiles[tid])
                # A signed-in source is the tenant's own, never shared: gathered here,
                # under a per-tenant key, only for a tenant that turned it on (#218).
                if profiles[tid].sources.get(mygreenhouse.SOURCE) and not ats_only:
                    key = f"{mygreenhouse.SOURCE}:{tid}"
                    if key not in gathered:
                        gathered[key] = _gather_portal(
                            repos[tid], profiles[tid], limit_per_source, own_failures
                        )
                    listings = [*listings, *gathered[key]]
                if profiles[tid].sources.get(linkedin.SOURCE) and not ats_only:
                    key = f"{linkedin.SOURCE}:{tid}"
                    if key not in gathered:
                        gathered[key] = _gather_linkedin(
                            repos[tid], profiles[tid], limit_per_source, own_failures
                        )
                    listings = [*listings, *gathered[key]]
                if profiles[tid].sources.get(handshake.SOURCE) and not ats_only:
                    key = f"{handshake.SOURCE}:{tid}"
                    if key not in gathered:
                        gathered[key] = _gather_handshake(
                            repos[tid], profiles[tid], limit_per_source, own_failures
                        )
                    listings = [*listings, *gathered[key]]
                results[tid] = score_and_stage(
                    repos[tid],
                    profiles[tid],
                    listings,
                    verify_links=verify_links,
                    link_cache=link_cache,
                )
            except Exception as exc:  # noqa: BLE001 - isolate one tenant's failure
                results[tid] = []
                logger.warning("poll failed for tenant %s", tid, exc_info=True)
                own_failures["poll"] = _describe(exc)
                _record_run(
                    repos[tid], tid, started_at, ats_only=ats_only, leads_added=0,
                    errors=_run_errors(profiles[tid], ats_only, shared_failures, own_failures),
                )
                continue
            _mark_polled(repos[tid], tid)
            _record_run(
                repos[tid], tid, started_at, ats_only=ats_only, leads_added=len(results[tid]),
                errors=_run_errors(profiles[tid], ats_only, shared_failures, own_failures),
            )
        return results
    finally:
        if conn is not None:
            for tid in acquired_locks:
                try:
                    _release_tenant_poll_lock(conn, tid)
                except Exception:  # noqa: BLE001 - release the remaining locks
                    logger.warning("poll lock release failed for tenant %s", tid, exc_info=True)


def _run_errors(
    profile: Criteria,
    ats_only: bool,
    shared_failures: dict[str, str],
    own_failures: dict[str, str],
) -> list[dict[str, str]]:
    """This tenant's failing sources for its ``poll_runs`` row: the shared ones it draws
    on, then its own signed-in ones and the pass itself. Never another tenant's source."""
    errors = [
        {"source": key, "error": shared_failures[key]}
        for key in _source_keys(profile, ats_only=ats_only)
        if key in shared_failures
    ]
    errors += [{"source": key, "error": err} for key, err in own_failures.items()]
    return errors


def _gather_portal(
    repo: TenantRepo, profile: Criteria, limit: int, failures: dict[str, str] | None = None
) -> list[Listing]:
    """One tenant's MyGreenhouse listings (#218), searched with the session the agent's
    browser keeps on the board-account row (#221). The poller never signs in itself: the
    portal answers a plain client's request for a security code with a redirect and no
    mail, so a missing or bounced session is cleared here and left for the agent to
    renew. Every failure is logged and yields an empty bucket — the rest of the poll goes on."""
    account_of = getattr(repo, "portal_account", None)
    if account_of is None:
        return []
    try:
        account = account_of()
        if account is None:
            logger.warning(
                "mygreenhouse is on but no board account for greenhouse.io is saved — "
                "add one under Settings · Agent · Board accounts"
            )
            return []
        keep = getattr(repo, "save_portal_session", None)
        if not account.session_fresh:
            logger.info(
                "mygreenhouse: no live session for %s — waiting for the agent's browser to "
                "sign in and keep one (#221)",
                account.email,
            )
            return []
        with httpx.Client(timeout=30.0, follow_redirects=False, headers=BROWSER_HEADERS) as client:
            portal = mygreenhouse.Portal(client, account)
            try:
                return mygreenhouse.fetch_listings(
                    portal, profile.keywords, remote_only=profile.remote_only, limit=limit
                )
            except mygreenhouse.NeedsSignIn:
                logger.info(
                    "mygreenhouse: the kept session for %s was bounced; cleared for the "
                    "agent's browser to renew (#221)",
                    account.email,
                )
                if keep is not None:
                    keep("", None)
                return []
    except Exception as exc:  # noqa: BLE001 - one tenant's portal must not abort the poll
        logger.warning("poll source mygreenhouse failed", exc_info=True)
        if failures is not None:
            failures[mygreenhouse.SOURCE] = _describe(exc)
        return []


def _gather_linkedin(
    repo: TenantRepo, profile: Criteria, limit: int, failures: dict[str, str] | None = None
) -> list[Listing]:
    """One tenant's LinkedIn listings (#233): the offsite-apply postings behind the
    keyword searches, with the employer's link. Signed in with the session the agent's
    browser keeps on the board-account row when there is a live one; LinkedIn's guest
    search otherwise. A bounced session is cleared for the agent to renew. Every failure
    is logged and yields an empty bucket — the rest of the poll goes on."""
    account_of = getattr(repo, "linkedin_account", None)
    try:
        account = account_of() if account_of is not None else None
        keep = getattr(repo, "save_linkedin_session", None)
        if account is None:
            logger.info(
                "linkedin is on with no board account for linkedin.com — searching as a guest "
                "(rate-limited); save one under Settings · Agent · Board accounts"
            )
        elif not account.session_fresh:
            logger.info(
                "linkedin: no live session for %s — searching as a guest until the agent's "
                "browser signs in and keeps one",
                account.email,
            )
            account = None
        seen_url = getattr(repo, "seen_url", None) or (lambda _url: False)
        easy_of = getattr(repo, "linkedin_easy_enabled", None)
        easy_apply = bool(easy_of()) if easy_of is not None else False

        def remember(url: str) -> None:
            from applytrack.poll import _norm_url

            repo.mark_seen_many([(_norm_url(url), "")])

        with httpx.Client(timeout=30.0, follow_redirects=False, headers=BROWSER_HEADERS) as client:
            li = linkedin.Client(client, account)
            try:
                return linkedin.fetch_listings(
                    li, profile.keywords, remote_only=profile.remote_only, limit=limit,
                    already_seen=seen_url, remember=remember, easy_apply=easy_apply,
                )
            except linkedin.NeedsSignIn:
                logger.info(
                    "linkedin: the kept session for %s was bounced; cleared for the agent's "
                    "browser to renew",
                    account.email if account else "?",
                )
                if keep is not None and account is not None:
                    keep("", None)
                return []
    except Exception as exc:  # noqa: BLE001 - one tenant's LinkedIn must not abort the poll
        logger.warning("poll source linkedin failed", exc_info=True)
        if failures is not None:
            failures[linkedin.SOURCE] = _describe(exc)
        return []


def _gather_handshake(
    repo: TenantRepo, profile: Criteria, limit: int, failures: dict[str, str] | None = None
) -> list[Listing]:
    """One tenant's Handshake listings (#267): the externally-applied postings behind the
    keyword searches, with the employer's own link. Unlike LinkedIn there is no guest
    path — Handshake's public pages expose a fraction of the board and hide the employer's
    URL, so with no live session this yields nothing and says so rather than staging leads
    that cannot be applied to. A bounced session is cleared for the agent to renew. Every
    failure is logged and yields an empty bucket — the rest of the poll goes on."""
    account_of = getattr(repo, "handshake_account", None)
    try:
        account = account_of() if account_of is not None else None
        keep = getattr(repo, "save_handshake_session", None)
        if account is None:
            logger.info(
                "handshake is on with no board account for joinhandshake.com — save one under "
                "Settings · Agent · Board accounts; Handshake has no guest search"
            )
            return []
        if not account.session_fresh:
            logger.info(
                "handshake: no live session for %s — waiting for the agent's browser to sign "
                "in and keep one",
                account.username,
            )
            return []
        seen_url = getattr(repo, "seen_url", None) or (lambda _url: False)

        with httpx.Client(timeout=30.0, follow_redirects=False, headers=BROWSER_HEADERS) as client:
            hs = handshake.Client(client, account)
            try:
                return handshake.fetch_listings(
                    hs, profile.keywords, limit=limit, already_seen=seen_url,
                )
            except handshake.SchemaError as exc:
                # Loud on purpose: a schema break otherwise looks exactly like a board
                # with no jobs, and the source would return nothing indefinitely.
                logger.error("handshake: %s", exc)
                if failures is not None:
                    failures[handshake.SOURCE] = _describe(exc)
                return []
            except handshake.NeedsSignIn:
                logger.info(
                    "handshake: the kept session for %s was bounced; cleared for the agent's "
                    "browser to renew",
                    account.username,
                )
                if keep is not None:
                    keep("", None)
                return []
    except Exception as exc:  # noqa: BLE001 - one tenant's Handshake must not abort the poll
        logger.warning("poll source handshake failed", exc_info=True)
        if failures is not None:
            failures[handshake.SOURCE] = _describe(exc)
        return []


def drain_requests(
    conn: psycopg.Connection,
    *,
    limit_per_source: int = 40,
    repo_for: Callable[[int], TenantRepo] | None = None,
) -> dict[int, list[str]]:
    """Service the on-demand poll queue: claim every request and poll those tenants.

    ``POST /api/poll`` (the SPA's button) enqueues a ``poll_requests`` row; a fast
    cron runs this — typically every minute — so the button doesn't wait for the
    hourly pass. Rows are claimed with ``DELETE ... RETURNING``: atomic, so a crash
    or a second drainer can't double-process, and each distinct tenant is polled
    once. The connection must be in **autocommit** mode so the claim commits before
    the (slower) poll runs. This global claim is privileged like
    :func:`_active_tenant_ids`; everything it fans out to is re-scoped per tenant
    through :func:`run_all_tenants`.
    """
    with conn.cursor() as cur:
        cur.execute("DELETE FROM poll_requests RETURNING tenant_id")
        tenant_ids = sorted({int(row[0]) for row in cur.fetchall()})
    if not tenant_ids:
        return {}
    locked_tenant_ids: set[int] = set()
    results = run_all_tenants(
        conn,
        limit_per_source=limit_per_source,
        tenant_ids=tenant_ids,
        repo_for=repo_for,
        locked_tenant_ids=locked_tenant_ids,
    )
    if locked_tenant_ids:
        with conn.cursor() as cur:
            for tenant_id in sorted(locked_tenant_ids):
                cur.execute(
                    "INSERT INTO poll_requests (tenant_id) VALUES (%s)",
                    (tenant_id,),
                )
        logger.info(
            "requeued %s tenant poll request(s) blocked by active polls",
            len(locked_tenant_ids),
        )
    return results