# SPDX-License-Identifier: Apache-2.0
# Copyright 2026 Aaron K. Clark
"""Hourly job-discovery poller.

Pulls remote roles from public job-board APIs/feeds, scores each against the
user's keyword list, and adds *new* matches as ``lead`` ``.md`` files. It never
drafts a cover letter and never submits anything: drafting is on demand.

What counts as a match is no longer hardcoded — it comes from a tenant's
:class:`applytrack.criteria.Criteria`, which the .NET API persists as a
``search_profiles`` row and the poller loads through
:class:`applytrack.db.PollRepo`: the flat keyword list, the minimum fit score, a
remote-only toggle + location blocklist, which sources are enabled, and which
company ATS boards to scan.

This runtime no longer touches the disk store; it shares one Postgres with the
API and is driven against a tenant-scoped repo (see :class:`LeadRepo`).

Deduplication is load-bearing — the same posting must never be staged twice. For
each run an in-memory ledger is seeded from the tenant's existing ``applications``
rows. The posting URL is the authoritative key: a listing with a *distinct* URL is
a distinct lead, even when its ``company + role`` slug matches one already seen
(e.g. one role posted per city). The slug is only the fallback key for the
minority of sources that give no URL. Keys are recorded for any listing we
evaluate — staged, filtered, or link-dead — so we don't re-process it; the lone
exception is a genuine *write* failure, which leaves the listing unrecorded so a
later run can retry it.
"""

from __future__ import annotations

import json
import logging
import os
import re
import threading
from collections.abc import Callable, Collection, Iterable
from concurrent.futures import ThreadPoolExecutor
from dataclasses import dataclass, field, replace
from functools import lru_cache, partial
from html import unescape
from itertools import zip_longest
from typing import Protocol, TypeVar
from urllib.parse import urljoin, urlsplit

# Stdlib import is the Element *type* only — never a parser entry point; all parsing
# below goes through defusedxml, which is XXE / billion-laughs safe.
from xml.etree.ElementTree import Element  # nosec B405

import defusedxml.ElementTree as ET  # hardened XML parser (forbids entities/external)
import httpx
import psycopg

from applytrack.criteria import AtsBoard, Criteria, is_board_slug
from applytrack.linkcheck import (
    BROWSER_HEADERS,
    PublicFetchError,
    fetch_public,
    is_reachable,
    is_ssrf_safe,
    probe,
    ssrf_safe_client,
)
from applytrack.store import AppFields

logger = logging.getLogger(__name__)

_TAG_RE = re.compile(r"<[^>]+>")
_WS_RE = re.compile(r"\s+")
_SLUG_RE = re.compile(r"[^a-z0-9]+")
_URL_RE = re.compile(r"https?://[^\s<>\"')]+")

# Signals that a role is remote-friendly, for the optional remote-only filter.
_REMOTE_HINTS = ("remote", "anywhere", "worldwide", "distributed", "work from home")


@dataclass
class Listing:
    """A raw role pulled from a job board, before scoring/dedup."""

    company: str
    role: str
    link: str = ""
    location: str = ""
    salary: str = ""
    source: str = ""
    description: str = ""
    #: The aggregator's own "apply" link when its API hands one over (remoteOK's
    #: ``apply_url``): a redirect to the employer, cheaper and surer than scraping
    #: the listing page for it.
    apply_link: str = ""


# A fetcher takes an http client + per-source scan cap and returns listings.

# -- aggregator listings -----------------------------------------------------
#
# A lead from remoteOK, Remotive and their kind carries the aggregator's listing page
# as its link. There is no application form on that page: the agent judged such a
# lead, built a packet, opened it in the browser and found nothing to fill — a wasted
# model call and browser run per lead (#191). At discovery time the listing's apply
# link is followed to the employer's posting (through the same SSRF guard as the link
# check) and THAT is stored; when it cannot be, the lead keeps the aggregator link and
# the .NET side reads it as apply-by-hand. Both runtimes carry this host list — the
# schema is the contract, and this is a reading of its ``link`` column; keep it in
# step with ``AtsProvider.AggregatorHosts``.
AGGREGATOR_HOSTS = frozenset({
    "remoteok.com", "remoteok.io", "remotive.com", "remotive.io", "jobicy.com",
    "arbeitnow.com", "weworkremotely.com", "remotefirstjobs.com", "workanywhere.pro",
    "news.ycombinator.com",
    # A LinkedIn posting is a listing of the employer's (#233): its offsite Apply link
    # is the lead's ``apply_link`` and what gets stored; Easy Apply ones are never staged.
    "linkedin.com",
    # Jobright reposts employers' openings on LinkedIn and elsewhere with its own
    # "send my profile" flow in place of a form (#238).
    "jobright.ai",
    # Sundayy lists employers' openings and its Apply is an ad-monetised redirect chain
    # (sundayy.com -> thebigjobsite.com -> appcast.io -> lensa.com) through more
    # aggregators, never a form; all three job-board hosts are apply-by-hand (#243).
    "sundayy.com", "thebigjobsite.com", "lensa.com",
    # FetchJobs.co is the same chain under another name: a script-rendered listing whose
    # Apply leads off to us.thebigjobsite.com, never a form (#251).
    "fetchjobs.co",
    # A Handshake posting is a listing of the employer's (#267): its external Apply URL
    # is the lead's ``apply_link`` and what gets stored; Handshake-hosted ones are never
    # staged.
    "joinhandshake.com",
    # BestJobTool reposts job listings across boards without direct employer forms.
    "bestjobtool.com",
    # Adzuna lists other boards' postings; its "Apply" is a tracked redirect off-site.
    "adzuna.com",
})

# Aggregators whose Apply never reaches anything a person or a browser can fill: one
# ad-monetised redirect network under four names, its hop through us.thebigjobsite.com
# behind a bot check, its listings routinely another employer's posting under an invented
# title (#281). Not apply-by-hand — not staged at all. lensa.com, where the chain ends, is
# a board a person can sign up to and stays a plain aggregator. Keep in step with
# ``AtsProvider.DeadEndHosts``.
DEAD_END_HOSTS = frozenset({"sundayy.com", "thebigjobsite.com", "fetchjobs.co", "bestjobtool.com"})


def is_dead_end_link(url: str) -> bool:
    """True when ``url`` is a listing on a dead-end aggregator (#281)."""
    host = (urlsplit(url or "").hostname or "").lower()
    return any(host == h or host.endswith("." + h) for h in DEAD_END_HOSTS)


_APPLY_ANCHOR_RE = re.compile(
    r"<a\b[^>]*?href=[\"']([^\"']+)[\"'][^>]*>(.*?)</a>", re.IGNORECASE | re.DOTALL
)
_APPLY_WORDS_RE = re.compile(r"\bapply\b|i['’]?m interested", re.IGNORECASE)
_MAX_APPLY_CANDIDATES = 3


def is_aggregator_link(url: str) -> bool:
    """True when ``url`` is on a job aggregator's host rather than an employer's."""
    host = (urlsplit(url or "").hostname or "").lower()
    return any(host == h or host.endswith("." + h) for h in AGGREGATOR_HOSTS)


def _apply_candidates(html: str, base: str) -> list[str]:
    """The anchors on a listing page that read as its Apply link, absolute, in order."""
    out: list[str] = []
    for href, inner in _APPLY_ANCHOR_RE.findall(html):
        label = _strip_html(inner)
        if not (_APPLY_WORDS_RE.search(label) or _APPLY_WORDS_RE.search(href)):
            continue
        url = urljoin(base, unescape(href).strip())
        if url.startswith(("http://", "https://")) and url not in out:
            out.append(url)
        if len(out) >= _MAX_APPLY_CANDIDATES:
            break
    return out


# An employer's careers site that answers a bot with one of these is refusing the
# probe, not retiring the posting: Meta, Workday tenants and most Cloudflare-fronted
# boards do it to every unattended client, and a person's browser opens the same URL.
_REFUSED_STATUSES = frozenset({401, 403, 405, 406, 429, 451, 503})


def _resolve_employer(item: Listing, client: httpx.Client) -> tuple[str | None, bool]:
    """The employer's posting behind an aggregator listing, and whether it was *seen* live.

    ``(url, True)`` for the first candidate whose final URL is live and off every
    aggregator's host. Failing that, ``(url, False)`` for the first one the employer's
    server merely refused to show a bot (#276): the link LinkedIn handed over is still
    the form to fill, and falling back to the listing page traded a posting the browser
    can drive for one it never will. A 404, a redirect home or a generic index is dead
    either way, and ``(None, False)`` leaves the listing's own link in place.
    """
    candidates = [item.apply_link] if item.apply_link else []
    if not candidates:
        try:
            html = fetch_public(item.link, client=client).decode("utf-8", "replace")
        except PublicFetchError:
            return None, False
        candidates = _apply_candidates(html, item.link)
    refused: str | None = None
    for url in candidates[:_MAX_APPLY_CANDIDATES]:
        status = probe(url, client=client)
        final = status.final_url or url
        if is_aggregator_link(final):
            continue
        if status.ok:
            return final, True
        if (
            refused is None
            and status.status_code in _REFUSED_STATUSES
            and not status.redirected_to_home
            and not status.generic_listing
        ):
            refused = final
    return refused, False


def resolve_employer_link(item: Listing, client: httpx.Client) -> str | None:
    """Follow an aggregator listing to the employer's own posting, or ``None``.

    The aggregator's apply link when its API gave one, else the Apply anchors on the
    listing page; each is walked through the SSRF-guarded probe (every redirect hop
    validated) and the first live final URL that is *not* on an aggregator's host is
    the employer's posting — or, when none is live, the first the employer only refused
    to show a bot. Bounded to a few candidates; any failure is ``None``.
    """
    return _resolve_employer(item, client)[0]

Fetcher = Callable[[httpx.Client, int], list[Listing]]


# -- text helpers -----------------------------------------------------------


def _strip_html(text: str) -> str:
    return _WS_RE.sub(" ", _TAG_RE.sub(" ", text or "")).strip()


def _norm_url(url: str) -> str:
    """Scheme/host/trailing-slash-insensitive URL key (query & fragment dropped)."""
    url = (url or "").strip()
    if not url:
        return ""
    parts = urlsplit(url if "//" in url else "//" + url)
    host = parts.netloc.lower().removeprefix("www.")
    path = parts.path.rstrip("/").lower()
    return f"{host}{path}" or url.lower()


def _norm_slug(company: str, role: str) -> str:
    """Normalized ``company + role`` key, robust to spacing/punctuation.

    The *full* role is kept (parentheticals included), so per-city variants like
    ``Engineer (Campinas)`` vs ``Engineer (São Paulo)`` are distinct keys — they
    are distinct openings. This is the *fallback* dedup key, used only for
    listings that carry no URL; when a URL is present it is authoritative (see
    :meth:`Seen.has`), so two distinct URLs are never collapsed by this slug.
    """
    return _SLUG_RE.sub("-", f"{company} {role}".lower()).strip("-")


def _norm_company(company: str) -> str:
    """Normalized company key for the blacklist (spacing/punctuation-insensitive)."""
    return _SLUG_RE.sub("-", (company or "").lower()).strip("-")


def _ats_label(slug: str) -> str:
    """Turn an ATS board slug into a display company name (``foo-bar`` -> ``Foo Bar``)."""
    return re.sub(r"[-_]+", " ", slug).strip().title() or slug


# -- scoring ----------------------------------------------------------------


@lru_cache(maxsize=1024)
def _keyword_re(keyword: str) -> re.Pattern[str]:
    """Compile one keyword into a word-aware matcher (cached; keyword sets repeat).

    Plain ``in`` matching made short keywords fire on the middle of unrelated
    words. ``rag`` was the worst offender by far -- it hit storage,
    leveraging, coverage, encourage -- and was single-handedly
    responsible for most borderline scores on the live feeds, pulling in account
    executives and customer-experience roles on the strength of a benefits blurb.

    So a match may not *begin* mid-word. It may still run past the end, because
    that is where the useful inflections live: ``backend engineer`` should match
    "backend engineers" and ``prompt engineer`` should match "prompt engineering".
    An arbitrary tail is not allowed, or ``ai`` would match "aid" -- only a short
    set of English suffixes.

    The suffix allowance is only granted to keywords of four or more characters.
    Short acronyms have no inflections of their own, and granting them one just
    re-opens the mid-word problem from the other side: ``rag`` would match "rages",
    ``ml`` "mled", ``go`` "going".

    Deliberately not a regex word-boundary anchor on the front: keywords like
    ``.net`` and ``c#`` start with punctuation, which such an anchor mishandles. The
    rule is the narrower "not preceded by a letter or digit". For a
    punctuation-leading keyword that rule is relaxed to "not preceded by a run of
    four or more": ``.net`` must still not match example.net, but an unconditional
    lookbehind also blocked the ``P`` in ASP.NET, the ``B`` in VB.NET and the ``O``
    in ADO.NET. Those framework prefixes are all short; hostnames' second-level
    labels almost never are.
    """
    prefix = r"(?<![a-z0-9])" if keyword[:1].isalnum() else r"(?<![a-z0-9]{4})"
    suffix = r"(?:e?s|ing|ed)?" if len(keyword) >= 4 else ""
    return re.compile(rf"{prefix}{re.escape(keyword)}{suffix}(?![a-z0-9])")


def classify(title: str, description: str, keywords: Iterable[str]) -> tuple[int, list[str]]:
    """Score a role 0-100 against the flat keyword list.

    Returns ``(score, matched_keywords)``; ``(0, [])`` when nothing matched.
    Keywords match on word starts, not raw substrings -- see :func:`_keyword_re`.

    A hit in the title is the strong signal and scores as it always has: base 50,
    +9 per title keyword, +3 per keyword found only in the body. A listing with
    *no* title hit is weaker evidence -- a generic role whose description happens
    to name a technology -- so it starts from 40 and needs corroboration to clear
    a typical floor: three body keywords reach 55, one reaches 45. Under the old
    flat base a single passing mention scored 53, which sat just under the default
    floor by luck rather than by judgement.

    The result is clamped to 100. It used to be able to exceed it (six title hits
    scored 104), which contradicted both this docstring and the 0-100 slider the
    SPA offers for ``min_fit_score``.
    """
    title_l = (title or "").lower()
    text_l = f"{title} {description}".lower()
    kws = [k.lower() for k in keywords if k]
    hits = [k for k in kws if _keyword_re(k).search(text_l)]
    if not hits:
        return 0, []
    title_hits = sum(1 for k in kws if _keyword_re(k).search(title_l))
    body_hits = len(hits) - title_hits
    score = 50 + 9 * title_hits + 3 * body_hits if title_hits else 40 + 5 * body_hits
    return min(100, score), hits


# -- semantic fit (opt-in, #300) ----------------------------------------------
#
# The keyword regex can only ask "does this posting share words with the list?" Jev
# (TypeSafe's System One model) is asked the question the regex approximates -- is this
# posting's own work the kind of work the keywords describe? -- and answers with a
# calibrated probability. It is one plain REST call, off unless a tenant turns it on
# AND the operator sets TYPESAFE_API_KEY, and any failure falls back to classify() for
# the rest of the run: a self-hoster with no key loses nothing. Measured against the
# keyword score on 257 hand-labelled live listings in tools/jev_eval/.

JEV_INSTRUCTIONS = (
    "`profile.keywords` lists the kinds of work a job seeker does. Is `posting` a job whose "
    "actual day-to-day work is one of those kinds of work? Judge the role itself -- what the "
    "hired person will spend their time doing. A posting that only mentions a keyword in the "
    "company blurb, the product description, the benefits, or a list of tools the team happens "
    "to use is NOT a match: a salesperson, recruiter, product manager or support agent at an AI "
    "or data company is not doing AI or data work."
)
_JEV_CRITERIA = {
    "true": "The role's own work is one of the kinds of work in `profile.keywords`",
    "false": "A different kind of job, even if it shares words with `profile.keywords`",
}
# Well under the model's 32k-token state limit, and what the evaluation measured.
_JEV_MAX_DESCRIPTION = 6000


class JevJudge:
    """Asks Jev for a 0-100 fit per listing; ``None`` means "use the keyword score"."""

    def __init__(self, api_key: str, *, base_url: str = "", model: str = "",
                 client: httpx.Client | None = None) -> None:
        self._url = (base_url or "https://api.typesafe.ai").rstrip("/") + "/v1/systemone"
        self._model = model or "jev-latest"
        self._headers = {"Authorization": f"Bearer {api_key}"}
        self._client = client or httpx.Client(timeout=15.0)
        self.broken = False

    def fit(self, item: Listing, keywords: Iterable[str]) -> int | None:
        if self.broken:
            return None
        body = {
            "model": self._model,
            "state": {
                "profile": {"keywords": [k for k in keywords if k]},
                "posting": {
                    "title": item.role, "company": item.company, "location": item.location,
                    "description": item.description[:_JEV_MAX_DESCRIPTION],
                },
            },
            "questions": {"fit": {"type": "noul", "instructions": JEV_INSTRUCTIONS,
                                  "criteria": _JEV_CRITERIA}},
        }
        try:
            res = self._client.post(self._url, json=body, headers=self._headers)
            res.raise_for_status()
            p = float(res.json()["answers"]["fit"]["noul"])
        except Exception as exc:  # noqa: BLE001 - any failure means "keywords decide"
            # One failure ends Jev for this run: an outage or a revoked key must not cost a
            # timeout per listing. The keyword score decides the rest of the pass.
            self.broken = True
            logger.warning("jev: %s — keyword scoring for the rest of this run", exc)
            return None
        if not 0.0 <= p <= 1.0:
            self.broken = True
            return None
        return round(100 * p)

    def close(self) -> None:
        self._client.close()


def jev_judge_for(repo: object) -> JevJudge | None:
    """A judge when this tenant opted in (``agent_settings.jev_classify``) and the operator
    configured a key; otherwise None, and :func:`classify` alone decides."""
    enabled = getattr(repo, "jev_classify_enabled", None)
    key = os.environ.get("TYPESAFE_API_KEY", "").strip()
    if not key or enabled is None:
        return None
    try:
        if not enabled():
            return None
    except Exception:  # noqa: BLE001 - an unreadable setting is an off setting
        logger.warning("jev: could not read the tenant's setting; using keywords", exc_info=True)
        return None
    return JevJudge(
        key,
        base_url=os.environ.get("TYPESAFE_BASE_URL", ""),
        model=os.environ.get("JEV_MODEL", ""),
    )


def _looks_remote(item: Listing) -> bool:
    """Best-effort: does this listing read as remote-friendly?

    A blank location is treated as remote (most sources here are remote-only
    boards that omit it); otherwise we look for an explicit remote signal in the
    location or role text.
    """
    loc = (item.location or "").strip().lower()
    if not loc:
        return True
    blob = f"{loc} {item.role}".lower()
    return any(h in blob for h in _REMOTE_HINTS)


def _passes_location(item: Listing, criteria: Criteria) -> bool:
    """Apply the remote-only toggle + exclude-locations blocklist."""
    if criteria.remote_only and not _looks_remote(item):
        return False
    loc = (item.location or "").lower()
    return not any(term.lower() in loc for term in criteria.exclude_locations if term)


# -- ledger -----------------------------------------------------------------


@dataclass
class Seen:
    """The per-run dedup ledger, backed by the ``seen`` table.

    ``urls`` / ``slugs`` hold the normalized keys already pinged — loaded from the
    ``seen`` table and seeded from the tenant's existing applications. ``sink``, if
    set, persists each *newly* seen key back to the table so a company is never
    re-pinged on a later run, even after its lead is deleted.
    """

    urls: set[str]
    slugs: set[str]
    sink: Callable[[str, str], None] | None = None

    def has(self, url: str, slug: str) -> bool:
        """Have we already processed this listing?

        A listing is seen if its canonical URL has already been recorded in
        the ledger, or if its normalized ``company + role`` slug already matches
        an existing application or seen opportunity. This prevents re-posted
        listings (e.g. from LinkedIn with changing job IDs) from staging
        duplicate leads when the opportunity is already tracked.
        """
        nu = _norm_url(url)
        if nu and nu in self.urls:
            return True
        return bool(slug) and slug in self.slugs

    def note(self, url: str, slug: str) -> tuple[str, str]:
        """Record keys in memory only; return the normalized keys newly added."""
        nu = _norm_url(url)
        new_url = nu if nu and nu not in self.urls else ""
        new_slug = slug if slug and slug not in self.slugs else ""
        if nu:
            self.urls.add(nu)
        if slug:
            self.slugs.add(slug)
        return new_url, new_slug

    def add(self, url: str, slug: str) -> None:
        """Record keys and persist the newly seen ones through ``sink``."""
        new_url, new_slug = self.note(url, slug)
        if self.sink is not None and (new_url or new_slug):
            self.sink(new_url, new_slug)


@dataclass
class Blacklist:
    """Companies to skip entirely, read from the ``blacklist`` table.

    Unlike :class:`Seen` (which blocks one URL/role), a blacklisted company is
    dropped wholesale, so no role from it is ever staged. The .NET API owns the
    writes (``/api/blacklist``); the poller only reads, comparing on the same
    normalized key.
    """

    companies: set[str]

    def has(self, company: str) -> bool:
        key = _norm_company(company)
        return bool(key) and key in self.companies


class LeadRepo(Protocol):
    """The tenant-scoped data access ``run_poll`` needs (see :class:`applytrack.db.PollRepo`).

    Kept narrow so the offline tests can drive ``run_poll`` with an in-memory
    fake — no Postgres, no network.
    """

    def iter_existing(self) -> Iterable[tuple[str, str, str]]:
        """Yield ``(link, company, role)`` for every application already stored."""
        ...

    def blacklist_companies(self) -> Iterable[str]:
        """Return the tenant's blacklisted company keys."""
        ...

    def load_seen(
        self, url_keys: Collection[str], slug_keys: Collection[str]
    ) -> tuple[set[str], set[str]]:
        """Return which of these ``(url_keys, slug_keys)`` the dedup ledger already holds."""
        ...

    def mark_seen_many(self, keys: Iterable[tuple[str, str]]) -> None:
        """Persist newly seen normalized ``(url_key, slug_key)`` pairs (either may be empty)."""
        ...

    def add_lead(self, fields: AppFields) -> str:
        """Stage one new lead and return its slug ``name``."""
        ...


def _seed_from_existing(repo: LeadRepo, seen: Seen) -> None:
    """Fold every application already stored for the tenant into the ledger.

    Guarantees we never re-add a role you already have a row for — even one you
    created by hand in the SPA — using the same URL + ``company + role`` keys the
    live listings are checked against. In-memory only (``note``): existing rows are
    not re-persisted to the ``seen`` table."""
    for link, company, role in repo.iter_existing():
        seen.note(link, _norm_slug(company, role))


# -- fetchers: remote job boards --------------------------------------------


def _fmt_salary(lo: object, hi: object) -> str:
    def _int(v: object) -> int:
        try:
            return int(v) if isinstance(v, (int, float, str)) and v not in ("", None) else 0
        except (TypeError, ValueError):
            return 0

    lo_i, hi_i = _int(lo), _int(hi)
    if lo_i and hi_i:
        return f"${lo_i:,}–${hi_i:,}"
    return f"${lo_i:,}" if lo_i else (f"${hi_i:,}" if hi_i else "")


# Remotive's public JSON API is rationed to ~28 jobs; the website itself renders
# 35 fresh jobs per category into a `window.__INITIAL_SEARCH_RESULTS__` blob. We
# scrape page 1 of each relevant category to pull hundreds instead of dozens.
REMOTIVE_CATEGORIES = (
    "software-development",
    "data",
    "devops-sysadmin",
    "product",
    "all-others",
)
_REMOTIVE_BLOB_RE = re.compile(r"__INITIAL_SEARCH_RESULTS__\s*=\s*(\{)")


def _extract_js_object(html: str, start_brace: int) -> str | None:
    """Return the balanced ``{...}`` JSON object beginning at ``start_brace``."""
    depth = 0
    in_str = False
    esc = False
    for i in range(start_brace, len(html)):
        ch = html[i]
        if in_str:
            if esc:
                esc = False
            elif ch == "\\":
                esc = True
            elif ch == '"':
                in_str = False
        elif ch == '"':
            in_str = True
        elif ch == "{":
            depth += 1
        elif ch == "}":
            depth -= 1
            if depth == 0:
                return html[start_brace : i + 1]
    return None


def _remotive_hits(html: str) -> list[dict[str, object]]:
    m = _REMOTIVE_BLOB_RE.search(html)
    if not m:
        return []
    blob = _extract_js_object(html, m.start(1))
    if not blob:
        return []
    try:
        data = json.loads(blob)
    except ValueError:
        return []
    results = data.get("results") if isinstance(data, dict) else None
    envelope = results[0] if isinstance(results, list) and results else results
    hits = envelope.get("hits") if isinstance(envelope, dict) else None
    return [h for h in hits if isinstance(h, dict)] if isinstance(hits, list) else []


def _remotive_location(hit: dict[str, object]) -> str:
    locs = hit.get("locations")
    if isinstance(locs, list):
        names = [str(x.get("name", x)) if isinstance(x, dict) else str(x) for x in locs]
        return ", ".join(n for n in names if n)
    return str(locs) if locs else ""


def _listing_from_remotive_hit(hit: dict[str, object]) -> Listing | None:
    company = str(hit.get("company_name", "")).strip()
    role = str(hit.get("title", "")).strip()
    if not company or not role:
        return None
    skills = hit.get("skills")
    skill_txt = " ".join(str(s) for s in skills) if isinstance(skills, list) else ""
    desc = " ".join(
        str(hit.get(k, "")) for k in ("category", "occupation", "seniority", "job_type")
    )
    return Listing(
        company=company,
        role=role,
        link=str(hit.get("url", "")).strip(),
        location=_remotive_location(hit),
        salary=str(hit.get("salary", "") or "").strip(),
        source="remotive",
        description=f"{skill_txt} {desc}".strip(),
    )


def fetch_remotive(client: httpx.Client, limit: int) -> list[Listing]:
    out: list[Listing] = []
    seen_ids: set[object] = set()
    for cat in REMOTIVE_CATEGORIES:
        if len(out) >= limit:
            break
        try:
            r = client.get(f"https://remotive.com/remote-jobs/{cat}")
            r.raise_for_status()
        except httpx.HTTPError:
            continue
        for hit in _remotive_hits(r.text):
            jid = hit.get("id", hit.get("url"))
            if jid in seen_ids:
                continue
            seen_ids.add(jid)
            listing = _listing_from_remotive_hit(hit)
            if listing:
                out.append(listing)
                if len(out) >= limit:
                    break
    return out


def fetch_remoteok(client: httpx.Client, limit: int) -> list[Listing]:
    # RemoteOK blocks non-browser agents and prefixes the array with a legal
    # notice; the client already sends a browser User-Agent (see _gather).
    r = client.get("https://remoteok.com/api")
    r.raise_for_status()
    rows = [row for row in r.json() if isinstance(row, dict) and row.get("position")]
    out: list[Listing] = []
    for j in rows[:limit]:
        out.append(
            Listing(
                company=str(j.get("company", "")).strip(),
                role=str(j.get("position", "")).strip(),
                link=str(j.get("url", "")).strip(),
                location=str(j.get("location", "") or "Remote").strip(),
                salary=_fmt_salary(j.get("salary_min"), j.get("salary_max")),
                source="remoteok",
                description=_strip_html(str(j.get("description", ""))),
                apply_link=str(j.get("apply_url", "") or "").strip(),
            )
        )
    return out


def fetch_arbeitnow(client: httpx.Client, limit: int) -> list[Listing]:
    r = client.get("https://www.arbeitnow.com/api/job-board-api")
    r.raise_for_status()
    out: list[Listing] = []
    for j in r.json().get("data", [])[:limit]:
        tags = j.get("tags") or []
        types = j.get("job_types") or []
        extra = " ".join(str(t) for t in [*tags, *types])
        loc = str(j.get("location", "") or ("Remote" if j.get("remote") else "")).strip()
        out.append(
            Listing(
                company=str(j.get("company_name", "")).strip(),
                role=str(j.get("title", "")).strip(),
                link=str(j.get("url", "")).strip(),
                location=loc,
                source="arbeitnow",
                description=f"{extra} {_strip_html(str(j.get('description', '')))}".strip(),
            )
        )
    return out


def fetch_jobicy(client: httpx.Client, limit: int) -> list[Listing]:
    r = client.get("https://jobicy.com/api/v2/remote-jobs", params={"count": min(limit, 50)})
    r.raise_for_status()
    out: list[Listing] = []
    for j in r.json().get("jobs", [])[:limit]:
        body = str(j.get("jobExcerpt", "") or j.get("jobDescription", ""))
        out.append(
            Listing(
                company=str(j.get("companyName", "")).strip(),
                role=str(j.get("jobTitle", "")).strip(),
                link=str(j.get("url", "")).strip(),
                location=str(j.get("jobGeo", "") or "Remote").strip(),
                salary=str(j.get("annualSalaryMin", "") or "").strip(),
                source="jobicy",
                description=_strip_html(body),
            )
        )
    return out


# -- fetchers: RSS job boards -----------------------------------------------
#
# Three boards publish RSS rather than JSON, each splitting its listings across
# per-stack category feeds. They need no bespoke parsing: :func:`parse_job_feed`
# (defined with the custom-feed plumbing below) already reads both title
# conventions these boards use -- "Company: Role" and "Role at Company".

# We Work Remotely category feeds. Item titles are "Company: Role".
WWR_FEEDS = (
    "https://weworkremotely.com/categories/remote-programming-jobs.rss",
    "https://weworkremotely.com/categories/remote-back-end-programming-jobs.rss",
    "https://weworkremotely.com/categories/remote-full-stack-programming-jobs.rss",
    "https://weworkremotely.com/categories/remote-front-end-programming-jobs.rss",
    "https://weworkremotely.com/categories/remote-devops-sysadmin-jobs.rss",
)

# RemoteFirstJobs per-stack feeds. Item titles are "Role at Company".
REMOTEFIRSTJOBS_FEEDS = (
    "https://remotefirstjobs.com/rss/jobs/ai.rss",
    "https://remotefirstjobs.com/rss/jobs/python.rss",
    "https://remotefirstjobs.com/rss/jobs/react.rss",
    "https://remotefirstjobs.com/rss/jobs/golang.rss",
    "https://remotefirstjobs.com/rss/jobs/data-science.rss",
)

# WorkAnywhere.pro role-family feeds; it aggregates and de-dupes other boards.
# Item titles are "Role — Company" (em dash) -- see :func:`_retitle_em_dash`.
#
# This host is fronted by Vercel and rate-limits harder than the others: hammered
# a few times in a row it starts answering with an HTTP 429 "Security Checkpoint"
# JS interstitial instead of the feed, and it 429s a request with no User-Agent
# outright. The pooled client's BROWSER_HEADERS and the hourly poll cadence both
# stay well inside that, and a 429 costs a logged WARNING, not a failed run.
WORKANYWHERE_FEEDS = (
    "https://workanywhere.pro/rss/developer.xml",
    "https://workanywhere.pro/rss/engineer.xml",
    "https://workanywhere.pro/rss/data-ai.xml",
)


def _retitle_em_dash(listing: Listing) -> None:
    """Recover ``(role, company)`` from a "Role — Company" title, in place.

    :func:`_split_company_role` knows "Company: Role" and "Role at Company"; a
    board that uses an em dash instead falls through to its last resort, which
    attributes every lead to the *feed's* own title and leaves the employer
    buried in the role. Splitting on the last " — " fixes that, and only that:
    a role legitimately containing a hyphen ("Developer Advocate - AI &
    Developer Experiences — Snowflake") keeps it.

    Deliberately *not* folded into the shared splitter. Custom tenant feeds run
    through the same parser, and "Senior Engineer — Remote" is a common enough
    title there that a global em-dash rule would invent "Remote" as an employer.
    """
    role, sep, company = listing.role.rpartition(" — ")
    if sep and role.strip() and company.strip():
        listing.role, listing.company = role.strip(), company.strip()


def _fetch_feed_set(
    client: httpx.Client,
    feeds: tuple[str, ...],
    limit: int,
    source: str,
    retitle: Callable[[Listing], None] | None = None,
) -> list[Listing]:
    """Walk one board's category feeds through the shared feed parser.

    The per-source cap is split *evenly across feeds* rather than consumed
    first-come. These boards are partitioned by stack and wildly uneven -- WWR's
    full-stack feed carried ~118 items against front-end's ~4 when this was
    written -- so a first-come budget lets one busy category swallow the whole cap
    and starve every later feed, which is exactly the per-stack coverage the feeds
    exist to provide.

    Two things differ from a tenant's custom feeds. These URLs are hard-coded
    rather than user-supplied, so they use the pooled client instead of
    :func:`~applytrack.linkcheck.fetch_public`'s SSRF-guarded transport. And the
    ``source`` id is fixed, so leads are attributed to the board rather than to
    :func:`_feed_source`'s per-host ``rss:<host>``.

    ``retitle`` optionally repairs a board whose title convention the shared
    splitter doesn't recognize.

    A posting cross-listed in several of a board's categories is taken once. The
    staging ledger would drop the repeat anyway, but only after it had consumed a
    slot: counting duplicates against a feed's share silently shrinks how much of
    the board a poll actually covers. Measured on the live feeds when this landed,
    7 of RemoteFirstJobs' 40 slots were going to postings an earlier category had
    already supplied.
    """
    per_feed = max(1, limit // len(feeds))
    out: list[Listing] = []
    taken: set[str] = set()
    for feed in feeds:
        if len(out) >= limit:
            break
        try:
            r = client.get(feed)
            r.raise_for_status()
        except httpx.HTTPError:
            # One dead category must not cost the board its other feeds; _gather
            # only sees (and names) a failure of the whole source.
            logger.warning("source %s: feed %s unreachable", source, feed)
            continue
        # Parsed past this feed's share by exactly the number of postings an earlier
        # feed has already taken, so duplicates are skipped over rather than counted
        # against the share -- without materializing the whole-run cap per feed.
        fresh = 0
        for listing in parse_job_feed(r.content, feed, per_feed + len(taken)):
            if fresh >= per_feed or len(out) >= limit:
                break
            listing.source = source
            if retitle is not None:
                retitle(listing)
            # The URL is authoritative; a linkless entry falls back to company +
            # role -- the same rule Seen.has applies once the listing reaches the
            # ledger. Computed after retitle so the slug sees the split title.
            key = _norm_url(listing.link) or _norm_slug(listing.company, listing.role)
            if key in taken:
                continue
            taken.add(key)
            out.append(listing)
            fresh += 1
    return out


def fetch_weworkremotely(client: httpx.Client, limit: int) -> list[Listing]:
    return _fetch_feed_set(client, WWR_FEEDS, limit, "weworkremotely")


def fetch_remotefirstjobs(client: httpx.Client, limit: int) -> list[Listing]:
    return _fetch_feed_set(client, REMOTEFIRSTJOBS_FEEDS, limit, "remotefirstjobs")


def fetch_workanywhere(client: httpx.Client, limit: int) -> list[Listing]:
    return _fetch_feed_set(
        client, WORKANYWHERE_FEEDS, limit, "workanywhere", retitle=_retitle_em_dash
    )


# -- fetcher: Hacker News "Who is hiring?" ----------------------------------


def _latest_whoishiring_id(client: httpx.Client) -> str | None:
    """Find the objectID of the most recent monthly "Who is hiring?" thread."""
    r = client.get(
        "https://hn.algolia.com/api/v1/search",
        params={"tags": "story,author_whoishiring", "query": "who is hiring", "hitsPerPage": 10},
    )
    r.raise_for_status()
    best: tuple[int, str] | None = None
    for hit in r.json().get("hits", []):
        title = str(hit.get("title", "")).lower()
        if "who is hiring" not in title or "wants to be hired" in title:
            continue
        created = int(hit.get("created_at_i", 0))
        oid = str(hit.get("objectID", ""))
        if oid and (best is None or created > best[0]):
            best = (created, oid)
    return best[1] if best else None


def _parse_hn_comment(text_html: str) -> Listing | None:
    """Best-effort parse of one "Who is hiring?" comment into a listing.

    These are free-form, but the convention is a pipe-delimited first line like
    ``Company | Role | Location | REMOTE | ...``. We require at least a company
    and a role from that header; prose-only comments are skipped (the keyword
    filter downstream would mostly drop them anyway).
    """
    if not text_html:
        return None
    first_para = re.split(r"<p>|\n", text_html, maxsplit=1)[0]
    header = _strip_html(first_para)
    parts = [p.strip() for p in header.split("|") if p.strip()]
    if len(parts) < 2:
        return None
    company, role = parts[0], parts[1]
    if not company or not role or len(company) > 80:
        return None
    location = next((p for p in parts[2:] if any(h in p.lower() for h in _REMOTE_HINTS)), "")
    if not location:
        location = parts[2] if len(parts) > 2 else ""
    url_match = _URL_RE.search(text_html)
    return Listing(
        company=company,
        role=role,
        link=url_match.group(0) if url_match else "",
        location=location,
        source="hn_whoishiring",
        description=_strip_html(text_html)[:600],
    )


def fetch_hn_whoishiring(client: httpx.Client, limit: int) -> list[Listing]:
    thread_id = _latest_whoishiring_id(client)
    if not thread_id:
        return []
    r = client.get(f"https://hn.algolia.com/api/v1/items/{thread_id}")
    r.raise_for_status()
    children = r.json().get("children", [])
    out: list[Listing] = []
    for child in children:
        if not isinstance(child, dict) or not child.get("text"):
            continue
        listing = _parse_hn_comment(str(child["text"]))
        if listing:
            out.append(listing)
            if len(out) >= limit:
                break
    return out


# -- fetchers: company ATS boards (Greenhouse / Lever) ----------------------


def _require_board_slug(slug: str) -> None:
    # The slug becomes a path segment on the provider's host; a ``/``, ``..`` or ``?``
    # in it would rewrite the request (#346). Criteria parsing already drops such
    # boards — this is the last check before the URL is built.
    if not is_board_slug(slug):
        raise ValueError(f"invalid ATS board slug: {slug!r}")


def fetch_greenhouse(client: httpx.Client, limit: int, slug: str) -> list[Listing]:
    _require_board_slug(slug)
    r = client.get(
        f"https://boards-api.greenhouse.io/v1/boards/{slug}/jobs", params={"content": "true"}
    )
    r.raise_for_status()
    company = _ats_label(slug)
    out: list[Listing] = []
    for j in r.json().get("jobs", [])[:limit]:
        loc = j.get("location") or {}
        out.append(
            Listing(
                company=company,
                role=str(j.get("title", "")).strip(),
                link=str(j.get("absolute_url", "")).strip(),
                location=str(loc.get("name", "")).strip() if isinstance(loc, dict) else "",
                source=f"greenhouse:{slug}",
                description=_strip_html(str(j.get("content", ""))),
            )
        )
    return out


def fetch_lever(client: httpx.Client, limit: int, slug: str) -> list[Listing]:
    _require_board_slug(slug)
    r = client.get(f"https://api.lever.co/v0/postings/{slug}", params={"mode": "json"})
    r.raise_for_status()
    rows = r.json()
    company = _ats_label(slug)
    out: list[Listing] = []
    for j in rows[:limit] if isinstance(rows, list) else []:
        cats = j.get("categories") or {}
        location = str(cats.get("location", "")).strip() if isinstance(cats, dict) else ""
        team = str(cats.get("team", "")).strip() if isinstance(cats, dict) else ""
        out.append(
            Listing(
                company=company,
                role=str(j.get("text", "")).strip(),
                link=str(j.get("hostedUrl", "")).strip(),
                location=location,
                source=f"lever:{slug}",
                description=f"{team} {_strip_html(str(j.get('descriptionPlain', '')))}".strip(),
            )
        )
    return out


# Paylocity renders its public board as a React app, so there is no JSON API to
# call -- but the server embeds the whole job list in a ``window.pageData = {...}``
# assignment in the HTML it ships. That blob is the board: parsing it is a plain
# JSON decode, not HTML scraping, and it carries the same fields the page shows.
_PAYLOCITY_PAGE_DATA = "window.pageData = "

# A Paylocity board is addressed by the company's recruiting GUID (the long id in
# the middle of a board URL). Anchored here because the slug is user-supplied and
# gets interpolated into a URL path -- a GUID cannot traverse or add a query.
_PAYLOCITY_SLUG_RE = re.compile(r"^[0-9a-fA-F]{8}(?:-[0-9a-fA-F]{4}){3}-[0-9a-fA-F]{12}$")


def paylocity_slug(value: str) -> str:
    """The company GUID for a Paylocity board, from a raw GUID or a board URL.

    Accepts what a user actually has in hand -- the address bar of a company's
    Paylocity careers page (``.../recruiting/jobs/All/<guid>/Acme``) -- and
    returns just the GUID, or ``""`` when there is no GUID to be found.
    """
    value = (value or "").strip()
    if _PAYLOCITY_SLUG_RE.match(value):
        return value.lower()
    for part in re.split(r"[/?#]", value):
        if _PAYLOCITY_SLUG_RE.match(part):
            return part.lower()
    return ""


def _paylocity_page_data(html: str) -> dict[str, object]:
    """Pull the ``window.pageData`` object out of a Paylocity board page."""
    start = html.find(_PAYLOCITY_PAGE_DATA)
    if start < 0:
        return {}
    try:
        data, _ = json.JSONDecoder().raw_decode(html[start + len(_PAYLOCITY_PAGE_DATA) :])
    except ValueError:
        return {}
    return data if isinstance(data, dict) else {}


def _paylocity_location(job: dict[str, object]) -> str:
    """Prefer the display location; fall back to the structured city/state."""
    name = str(job.get("LocationName") or "").strip()
    if not name:
        loc = job.get("JobLocation")
        if isinstance(loc, dict):
            parts = [str(loc.get(k) or "").strip() for k in ("City", "State", "Country")]
            name = ", ".join(p for p in parts if p)
    # The board flags remote separately from the (often headquarters) location, so
    # say so in the location text -- that is what _looks_remote reads.
    if job.get("IsRemote") and "remote" not in name.lower():
        name = f"{name} (Remote)".strip() if name else "Remote"
    return name


def fetch_paylocity(client: httpx.Client, limit: int, slug: str) -> list[Listing]:
    """Scan one company's public Paylocity recruiting board.

    ``slug`` is the company GUID. The host is fixed and the GUID shape is
    validated, so this can share the pooled client like the other ATS boards.
    """
    guid = paylocity_slug(slug)
    if not guid:
        return []
    r = client.get(
        f"https://recruiting.paylocity.com/recruiting/jobs/All/{guid}",
        headers=BROWSER_HEADERS,
    )
    r.raise_for_status()
    data = _paylocity_page_data(r.text)
    jobs = data.get("Jobs")
    if not isinstance(jobs, list):
        return []
    company = str(data.get("ModuleTitle") or "").strip() or _ats_label(guid)
    out: list[Listing] = []
    for j in jobs[:limit]:
        if not isinstance(j, dict) or j.get("IsInternal"):
            continue
        job_id = str(j.get("JobId") or "").strip()
        out.append(
            Listing(
                company=company,
                role=str(j.get("JobTitle", "")).strip(),
                link=(
                    f"https://recruiting.paylocity.com/Recruiting/Jobs/Details/{job_id}"
                    if job_id
                    else ""
                ),
                location=_paylocity_location(j),
                source=f"paylocity:{guid}",
                description=_strip_html(str(j.get("Description", ""))),
            )
        )
    return out


# -- fetcher: custom RSS / Atom feeds ---------------------------------------
#
# Unlike every fetcher above, the URL here comes from the user (Settings ·
# Criteria), so it is attacker-influenced: the fetch goes through
# linkcheck.fetch_public, whose transport pins each hop to a validated public
# address, and the document is parsed by defusedxml. One feed key is
# "rss:{url}" — the same key the worker's shared gather routes by.

# A "Company: Role" head longer than this is almost certainly a sentence, not a
# company name; fall back to the feed's own title instead.
_MAX_COMPANY_LEN = 80

# What a careers feed tends to wrap its own title in ("Jobs at Hooli", "Hooli
# Careers"), stripped so the fallback company reads as just the company.
_FEED_TITLE_HEAD_RE = re.compile(
    r"^\s*(?:jobs?|careers?|openings?|vacancies|positions?)\s+at\s+", re.I
)
_FEED_TITLE_TAIL_RE = re.compile(
    r"[\s\-–—|:]*\b(?:jobs?|careers?|job\s+board|openings?|vacancies|positions?)\b\s*$", re.I
)
_ROLE_AT_COMPANY_RE = re.compile(r"\s+at\s+", re.I)


def _local(tag: object) -> str:
    """Local name of an XML tag, any ``{namespace}`` prefix discarded, lowercased."""
    return str(tag).rsplit("}", 1)[-1].lower()


def _child(node: Element, name: str) -> Element | None:
    """First *direct* child with the given local name (namespace-agnostic)."""
    return next((el for el in node if _local(el.tag) == name), None)


def _child_text(node: Element, *names: str) -> str:
    """Text of the first direct child matching any of ``names``, else ``""``.

    Nested markup is flattened (``itertext``) so an Atom ``<content type="xhtml">``
    yields its prose rather than nothing.
    """
    for name in names:
        el = _child(node, name)
        if el is not None:
            text = _WS_RE.sub(" ", "".join(el.itertext())).strip()
            if text:
                return text
    return ""


def _entry_link(node: Element) -> str:
    """The item's URL: an RSS ``<link>`` body, an Atom ``<link href>``, or a permalink guid."""
    fallback = ""
    for el in node:
        if _local(el.tag) != "link":
            continue
        href = (el.get("href") or "").strip()
        if href:
            # Atom: prefer rel="alternate" (the human page); keep others as backup.
            if (el.get("rel") or "alternate").strip().lower() == "alternate":
                return href
            fallback = fallback or href
            continue
        text = (el.text or "").strip()
        if text:
            return text
    if not fallback:
        guid = _child(node, "guid")
        text = (guid.text or "").strip() if guid is not None else ""
        if text.startswith(("http://", "https://")):
            return text
    return fallback


def _feed_company(feed_title: str) -> str:
    """Strip the boilerplate off a feed's own title to leave a usable company name."""
    title = _FEED_TITLE_HEAD_RE.sub("", feed_title or "").strip()
    return _FEED_TITLE_TAIL_RE.sub("", title).strip(" -–—|:").strip()


def _split_company_role(title: str, feed_company: str) -> tuple[str, str]:
    """Best-effort ``(company, role)`` from one feed item's title.

    Three shapes cover nearly every job feed: ``Company: Role`` (We Work Remotely
    and friends), ``Role at Company``, and a bare role on a company's own careers
    feed — where the company is the feed's title. A pair we can't split at all
    yields an empty company, and the caller drops the listing.
    """
    title = _WS_RE.sub(" ", title or "").strip()
    if not title:
        return "", ""

    company, sep, role = title.partition(":")
    company, role = company.strip(), role.strip()
    if sep and company and role and len(company) <= _MAX_COMPANY_LEN:
        return company, role

    # First " at " wins: a role prefix almost never contains one, while a company
    # name sometimes does — so "Staff Engineer at Data at Scale" keeps the company whole.
    split = _ROLE_AT_COMPANY_RE.search(title)
    if split is not None:
        role, company = title[: split.start()].strip(), title[split.end() :].strip()
        if company and role and len(company) <= _MAX_COMPANY_LEN:
            return company, role

    return feed_company, title


def _feed_body_text(raw: str) -> str:
    """Flatten a feed item's body to prose.

    Unlike the JSON sources, an RSS/Atom body is *escaped* HTML: the XML parser
    hands back the markup as text, so the tags are dropped first and the entities
    decoded after — otherwise every ``&`` and curly quote reaches the note (and the
    cover-letter prompt) as ``&amp;`` / ``&#8217;``.
    """
    return _WS_RE.sub(" ", unescape(_TAG_RE.sub(" ", raw or ""))).strip()


def _feed_source(feed_url: str) -> str:
    """Source id for a custom feed's leads: ``rss:<host>`` (``auto:rss:<host>`` on the entry)."""
    host = urlsplit(feed_url).netloc.lower().removeprefix("www.")
    return f"rss:{host}" if host else "rss"


def parse_job_feed(raw: bytes, feed_url: str, limit: int) -> list[Listing]:
    """Map an RSS 2.0 or Atom document onto listings — pure, no network.

    Parsed with defusedxml, so entity expansion and external references are refused;
    the document comes from a URL the user supplied, not from us. A document we
    can't parse yields no listings rather than raising, so one malformed feed never
    aborts a poll.
    """
    try:
        root = ET.fromstring(raw)
    except ET.ParseError:
        logger.warning("could not parse feed %s as RSS or Atom", feed_url)
        return []

    channel = _child(root, "channel")
    container = root if channel is None else channel
    feed_company = _feed_company(_child_text(container, "title"))
    source = _feed_source(feed_url)

    out: list[Listing] = []
    for node in root.iter():
        if _local(node.tag) not in ("item", "entry"):
            continue
        company, role = _split_company_role(_child_text(node, "title"), feed_company)
        if not company or not role:
            continue
        out.append(
            Listing(
                company=company,
                role=role,
                link=_entry_link(node),
                location=_child_text(node, "location", "region"),
                source=source,
                description=_feed_body_text(
                    _child_text(node, "description", "summary", "content", "encoded")
                ),
            )
        )
        if len(out) >= limit:
            break
    return out


def fetch_rss(client: httpx.Client, limit: int, url: str) -> list[Listing]:
    """Fetch and parse one custom RSS/Atom job feed.

    ``client`` — the shared plain client :func:`_gather` hands every fetcher — is
    deliberately unused: this URL is user-supplied, so the request must go through
    :func:`~applytrack.linkcheck.fetch_public`, whose transport connects only to
    validated public addresses. The built-in sources are hard-coded and can safely
    share the pooled client. An SSRF-safe client (the worker hands each of its
    threads one) *is* used, so a run's feeds share its connections and DNS pins
    instead of a fresh TLS handshake apiece (#347).
    """
    if is_ssrf_safe(client):
        return parse_job_feed(fetch_public(url, client=client), url, limit)
    return parse_job_feed(fetch_public(url), url, limit)


def make_rss_fetcher(url: str) -> Fetcher:
    """Bind one feed URL into a plain :data:`Fetcher`, named ``rss:<url>`` for logs."""

    def _fetch(client: httpx.Client, limit: int) -> list[Listing]:
        return fetch_rss(client, limit, url)

    _fetch.__name__ = f"rss:{url}"
    return _fetch


SOURCE_FETCHERS: dict[str, Fetcher] = {
    "remotive": fetch_remotive,
    "remoteok": fetch_remoteok,
    "arbeitnow": fetch_arbeitnow,
    "jobicy": fetch_jobicy,
    "weworkremotely": fetch_weworkremotely,
    "remotefirstjobs": fetch_remotefirstjobs,
    "workanywhere": fetch_workanywhere,
    "hn_whoishiring": fetch_hn_whoishiring,
}


def make_ats_fetcher(board: AtsBoard) -> Fetcher | None:
    """Bind a company slug to its provider's fetcher, yielding a plain Fetcher.

    The returned callable carries a ``provider:slug`` ``__name__`` so a failure of
    one board is named in the poll diagnostics rather than logged as ``<lambda>``.
    """
    slug = board.slug
    if board.provider == "greenhouse":

        def _fetch(client: httpx.Client, limit: int) -> list[Listing]:
            return fetch_greenhouse(client, limit, slug)

    elif board.provider == "lever":

        def _fetch(client: httpx.Client, limit: int) -> list[Listing]:
            return fetch_lever(client, limit, slug)

    elif board.provider == "paylocity":

        def _fetch(client: httpx.Client, limit: int) -> list[Listing]:
            return fetch_paylocity(client, limit, slug)

    else:
        return None
    _fetch.__name__ = f"{board.provider}:{slug}"
    return _fetch


def build_fetchers(criteria: Criteria) -> list[Fetcher]:
    """The fetchers to run for this config: enabled sources, ATS boards, custom feeds."""
    out: list[Fetcher] = [
        SOURCE_FETCHERS[name]
        for name, on in criteria.sources.items()
        if on and name in SOURCE_FETCHERS
    ]
    for board in criteria.ats_boards:
        fetcher = make_ats_fetcher(board)
        if fetcher is not None:
            out.append(fetcher)
    out.extend(make_rss_fetcher(url) for url in criteria.rss_feeds)
    return out


def _gather(fetchers: Iterable[Fetcher], limit: int) -> list[Listing]:
    """Run each fetcher, tolerating per-source failures (offline, 4xx, parse).

    A failing source is skipped but logged at WARNING with its name, so a
    dead/renamed source is visible in the worker logs instead of silently looking
    like an empty job market. Any exception is caught — one broken source (e.g. a
    changed upstream schema) must never abort the whole run.
    """
    listings: list[Listing] = []
    with httpx.Client(timeout=20.0, follow_redirects=True, headers=BROWSER_HEADERS) as client:
        for fetch in fetchers:
            label = getattr(fetch, "__name__", repr(fetch)).removeprefix("fetch_")
            try:
                listings.extend(fetch(client, limit))
            except Exception:  # noqa: BLE001 - one bad source must not abort the run
                logger.warning("poll source %s failed", label, exc_info=True)
    return listings


# -- bounded concurrency (#347) ---------------------------------------------

# A run's probes and source fetches go out this many at a time, and at most
# PER_HOST_WORKERS to any one host: parallel enough that 80 slow employer sites don't
# queue behind each other at 12 s apiece, polite enough not to earn a 429.
NETWORK_WORKERS = 8
PER_HOST_WORKERS = 2

_T = TypeVar("_T")
_K = TypeVar("_K")


def run_bounded(
    tasks: Iterable[tuple[str, Callable[[], _T]]],
    *,
    workers: int = NETWORK_WORKERS,
    per_host: int = PER_HOST_WORKERS,
) -> list[_T | Exception]:
    """Run ``(host, thunk)`` tasks on a bounded pool, results in task order.

    No more than ``per_host`` tasks for one host key run at once. A task that raises
    yields its exception in its slot instead, so one failure never costs the others.
    """
    tasks = list(tasks)
    gates = {host: threading.BoundedSemaphore(per_host) for host, _ in tasks}

    def run(host: str, thunk: Callable[[], _T]) -> _T | Exception:
        try:
            with gates[host]:
                return thunk()
        except Exception as exc:  # noqa: BLE001 - reported in the task's slot
            return exc

    if len(tasks) <= 1 or workers <= 1:
        return [run(host, thunk) for host, thunk in tasks]
    # Submit round-robin across hosts, so a run of one host's tasks doesn't fill the
    # pool with threads parked on its gate while other hosts' work waits in the queue.
    by_host: dict[str, list[int]] = {}
    for i, (host, _) in enumerate(tasks):
        by_host.setdefault(host, []).append(i)
    order = [i for row in zip_longest(*by_host.values()) for i in row if i is not None]
    with ThreadPoolExecutor(max_workers=min(workers, len(tasks))) as pool:
        futures = {i: pool.submit(run, *tasks[i]) for i in order}
        return [futures[i].result() for i in range(len(tasks))]


class ThreadClients:
    """One :func:`~applytrack.linkcheck.ssrf_safe_client` per worker thread, closed together.

    A pinned client arms one fetch deadline at a time, so two threads can't share one;
    one per thread still reuses its connections and DNS pins across every fetch that
    thread makes, rather than a fresh client and TLS handshake per URL (#347).
    """

    def __init__(self, timeout: float = 12.0) -> None:
        self._timeout = timeout
        self._local = threading.local()
        self._lock = threading.Lock()
        self._all: list[httpx.Client] = []

    def get(self) -> httpx.Client:
        client: httpx.Client | None = getattr(self._local, "client", None)
        if client is None:
            client = ssrf_safe_client(timeout=self._timeout)
            self._local.client = client
            with self._lock:
                self._all.append(client)
        return client

    def close(self) -> None:
        with self._lock:
            clients, self._all = self._all, []
        for client in clients:
            client.close()


@dataclass
class LinkCache:
    """What the network said about a URL, kept for one whole poll run (#347).

    The worker stages every tenant from the same shared listings, so without this the
    same aggregator page is resolved and the same employer posting probed once per
    tenant. ``resolved`` is keyed by ``(link, apply_link)``; ``reachable`` by URL.
    """

    resolved: dict[tuple[str, str], tuple[str | None, bool]] = field(default_factory=dict)
    reachable: dict[str, bool] = field(default_factory=dict)
    _lock: threading.Lock = field(default_factory=threading.Lock, repr=False, compare=False)
    _gates: dict[object, threading.Lock] = field(default_factory=dict, repr=False, compare=False)

    def once(self, table: dict[_K, _T], key: _K, compute: Callable[[], _T]) -> _T:
        """``table[key]``, computing it at most once even when pool threads race for it.

        A computation that raises caches nothing; the next caller tries again.
        """
        with self._lock:
            if key in table:
                return table[key]
            gate = self._gates.setdefault((id(table), key), threading.Lock())
        with gate:
            with self._lock:
                if key in table:
                    return table[key]
            value = compute()
            with self._lock:
                table[key] = value
            return value


def _host(url: str) -> str:
    return (urlsplit(url or "").hostname or "").lower()


# -- orchestration ----------------------------------------------------------


def run_poll(
    repo: LeadRepo,
    profile: Criteria,
    *,
    limit_per_source: int = 40,
    fetchers: Iterable[Fetcher] | None = None,
    listings: Iterable[Listing] | None = None,
    verify_links: bool | None = None,
) -> list[str]:
    """Fetch this tenant's enabled sources, then score and stage. Returns slug names.

    ``profile`` is the tenant's :class:`~applytrack.criteria.Criteria` (the caller
    loads it via :meth:`applytrack.db.PollRepo.load_profile`); ``repo`` is the
    tenant-scoped data access (see :class:`LeadRepo`). Network is only touched
    when ``listings`` is not supplied; pass a fixed ``listings`` iterable to run
    fully offline (tests, replay).

    When ``verify_links`` is on, each candidate's posting URL is probed as a
    browser before it becomes an entry, and unreachable/dead links are dropped.
    It defaults to on for live runs and off when ``listings`` is supplied.

    The multi-tenant worker (:mod:`applytrack.worker`) bypasses this and calls
    :func:`score_and_stage` directly over a shared, fetched-once listing set.
    """
    if verify_links is None:
        verify_links = listings is None

    if listings is None:
        if fetchers is None:
            fetchers = build_fetchers(profile)
        listings = _gather(fetchers, limit_per_source)

    return score_and_stage(repo, profile, listings, verify_links=verify_links)


def _select_for_profile(
    gathered: dict[str, list[Listing]], profile: Criteria
) -> list[Listing]:
    """Pick the listings for one tenant from a per-source gather, by enabled source.

    Lets the worker fetch each source once and route the shared buckets per tenant,
    so ``score_and_stage`` stays source-agnostic and a tenant only ever sees roles
    from the sources/boards/feeds it actually enabled."""
    out: list[Listing] = []
    for name, on in profile.sources.items():
        if on:
            out.extend(gathered.get(name, []))
    for board in profile.ats_boards:
        out.extend(gathered.get(f"{board.provider}:{board.slug}", []))
    for url in profile.rss_feeds:
        out.extend(gathered.get(f"rss:{url}", []))
    return out


@dataclass
class _Candidate:
    """A listing that passed every offline filter and waits on the network to be staged."""

    item: Listing
    slug: str
    score: int
    hits: list[str]
    judged: bool
    # Filled by _verify_candidates: the employer posting behind an aggregator listing,
    # whether it was seen live, and whether the link to be stored answered a browser.
    resolved: str | None = None
    seen_live: bool = False
    reachable: bool = True


def _verify_candidate(
    item: Listing, cache: LinkCache, clients: ThreadClients
) -> tuple[str | None, bool, bool]:
    """Resolve (for an aggregator listing) and probe one candidate, through ``cache``."""
    # An Easy Apply posting IS where it is applied to — there is no employer page
    # behind it to resolve — and LinkedIn's own API just served it, so it is not
    # probed either: linkedin.com answers an unattended GET with a 999 (#278).
    if item.source.endswith(":easy") or not item.link:
        return None, False, True
    resolved: str | None = None
    seen_live = False
    # An aggregator's listing page has no form on it: store the employer's
    # posting instead when the apply link leads there (#191).
    if is_aggregator_link(item.link):
        resolved, seen_live = cache.once(
            cache.resolved,
            (item.link, item.apply_link),
            lambda: _resolve_employer(item, clients.get()),
        )
    # The listing was live a moment ago and the employer's server will not talk to a
    # bot: the reachability check would only drop the lead.
    if resolved and not seen_live:
        return resolved, seen_live, True
    link = resolved or item.link
    reachable = cache.once(
        cache.reachable, link, lambda: is_reachable(link, client=clients.get())
    )
    return resolved, seen_live, reachable


def _verify_candidates(candidates: list[_Candidate], cache: LinkCache) -> list[_Candidate]:
    """Resolve and probe every candidate on a bounded pool; returns the ones that finished.

    Each distinct listing is checked once, and a URL an earlier tenant's staging already
    checked is not fetched again (#347). A candidate whose check raised is left out, and
    left unseen for a later run to retry.
    """
    unique: dict[tuple[str, str, bool], Listing] = {}
    for cand in candidates:
        it = cand.item
        unique.setdefault((it.link, it.apply_link, it.source.endswith(":easy")), it)
    keys = list(unique)
    clients = ThreadClients(timeout=12.0)
    try:
        outcomes = run_bounded(
            (_host(unique[k].link), partial(_verify_candidate, unique[k], cache, clients))
            for k in keys
        )
    finally:
        clients.close()
    by_key = dict(zip(keys, outcomes, strict=True))
    done: list[_Candidate] = []
    for cand in candidates:
        it = cand.item
        outcome = by_key[(it.link, it.apply_link, it.source.endswith(":easy"))]
        if isinstance(outcome, Exception):
            logger.warning(
                "link check failed for %s — %s", it.company, it.role, exc_info=outcome
            )
            continue
        cand.resolved, cand.seen_live, cand.reachable = outcome
        done.append(cand)
    return done


def score_and_stage(
    repo: LeadRepo,
    profile: Criteria,
    listings: Iterable[Listing],
    *,
    verify_links: bool = False,
    link_cache: LinkCache | None = None,
) -> list[str]:
    """Dedupe, filter, score, and stage ``listings`` for one tenant. Returns slug names.

    The dedup ledger is read from the ``seen`` table for just this batch's keys and
    seeded from the tenant's existing applications; the newly seen keys are persisted
    together at the end so a company is never re-pinged on a later run (#348). Network
    is touched only when ``verify_links`` is on: the listings that pass every offline
    filter are then resolved and probed together on a bounded pool, through
    ``link_cache`` — which the multi-tenant worker shares across every tenant of a
    run, so a URL is fetched once per run, not per tenant (#347).
    """
    batch = [it for it in listings if it.company and it.role]
    # Nothing gathered (an --ats-only pass for a tenant with no boards, every ten
    # minutes): nothing to dedupe, so no ledger, application or blacklist read (#348).
    if not batch:
        return []
    # Only the keys this batch could hit are read from the ledger, never the whole
    # history, which grows by every listing ever passed on (#348).
    url_keys, slug_keys = repo.load_seen(
        {k for it in batch for k in (_norm_url(it.link), _norm_url(it.apply_link)) if k},
        {_norm_slug(it.company, it.role) for it in batch} - {""},
    )
    # Newly seen keys are written once, together, at the end of the run (#348).
    fresh: list[tuple[str, str]] = []
    seen = Seen(url_keys, slug_keys, sink=lambda u, s: fresh.append((u, s)))
    try:
        return _score_and_stage(repo, profile, batch, seen, verify_links, link_cache)
    finally:
        if fresh:
            repo.mark_seen_many(fresh)


# Jev calls in flight at once for one tenant's batch: each can take up to its 15 s
# timeout, and a batch of 200 listings in series is most of an hour (#348).
JEV_WORKERS = 4


def _score_and_stage(
    repo: LeadRepo,
    profile: Criteria,
    listings: list[Listing],
    seen: Seen,
    verify_links: bool,
    link_cache: LinkCache | None,
) -> list[str]:
    _seed_from_existing(repo, seen)
    blacklist = Blacklist({_norm_company(c) for c in repo.blacklist_companies()})

    # The free filters first: a listing they drop never costs a paid Jev call (#348).
    scored: list[tuple[Listing, str, int, list[str]]] = []
    pending: set[tuple[str, str]] = set()
    for item in listings:
        # Blacklisted companies are dropped wholesale — never seen, never staged.
        if blacklist.has(item.company):
            continue
        slug = _norm_slug(item.company, item.role)
        if seen.has(item.link, slug) or (item.apply_link and seen.has(item.apply_link, slug)):
            continue
        # The same posting twice in one batch is judged once; staging re-checks
        # the ledger anyway, so the second copy could never be staged.
        once = ("url", item.link) if item.link else ("slug", slug)
        if once in pending:
            continue

        # We act on this listing one way or another below, so it is recorded
        # as seen *except* on a genuine write failure (handled at stage time),
        # which leaves it for a later run to retry.
        if not _passes_location(item, profile):
            seen.add(item.link, slug)
            continue

        # A dead-end aggregator's listing, or one whose Apply leads to it (a LinkedIn
        # "employer" fronting the network): nothing to apply to, so nothing to stage (#281).
        if is_dead_end_link(item.link) or is_dead_end_link(item.apply_link):
            logger.info("skipped %s — %s: dead-end aggregator", item.company, item.role)
            seen.add(item.link, slug)
            continue

        pending.add(once)
        score, hits = classify(item.role, item.description, profile.keywords)
        scored.append((item, slug, score, hits))

    # Opted in: Jev's fit replaces the keyword score, held to the same floor. It is
    # asked even with no keyword hit -- most of what it catches is a role the list
    # never names ("Data Engineer", "Backend Software Engineer") (#300). The first call
    # goes alone, so an outage still costs one timeout, not one per worker; the rest go
    # out a few at a time rather than one after another (#348).
    fits: list[int | None] = [None] * len(scored)
    judge = jev_judge_for(repo) if scored else None
    if judge is not None:
        try:
            fits[0] = judge.fit(scored[0][0], profile.keywords)
            if not judge.broken:
                outcomes = run_bounded(
                    (("jev", partial(judge.fit, item, profile.keywords))
                     for item, *_ in scored[1:]),
                    workers=JEV_WORKERS,
                    per_host=JEV_WORKERS,
                )
                fits[1:] = [None if isinstance(o, Exception) else o for o in outcomes]
        finally:
            judge.close()

    candidates: list[_Candidate] = []
    for (item, slug, score, hits), fit in zip(scored, fits, strict=True):
        if fit is not None:
            score = fit
        if (fit is None and not hits) or score < profile.min_fit_score:
            seen.add(item.link, slug)
            continue
        candidates.append(_Candidate(item, slug, score, hits, judged=fit is not None))

    if verify_links and candidates:
        candidates = _verify_candidates(
            candidates, link_cache if link_cache is not None else LinkCache()
        )
        # The employer postings behind aggregator listings were not known when the
        # ledger was read: read just those keys now (#348).
        resolved = {_norm_url(c.resolved) for c in candidates if c.resolved} - {""}
        if resolved:
            for key in repo.load_seen(resolved, set())[0]:
                seen.urls.add(key)

    added: list[str] = []
    for cand in candidates:
        item, slug = cand.item, cand.slug
        # An earlier candidate of this batch may have taken the key since it was queued.
        if seen.has(item.link, slug) or (item.apply_link and seen.has(item.apply_link, slug)):
            continue
        # The dedupe ledger is keyed on the listing's own link, whatever we store.
        listing_link = item.link
        if cand.resolved:
            logger.info(
                "resolved %s -> %s%s",
                item.link,
                cand.resolved,
                "" if cand.seen_live else " (the employer refused the probe; kept)",
            )
            if seen.has(cand.resolved, slug):
                seen.add(listing_link, slug)
                seen.add(cand.resolved, slug)
                continue
            item = replace(item, link=cand.resolved)

        # Block dead postings: don't create an entry we can't actually open.
        if not cand.reachable:
            seen.add(listing_link, slug)
            continue

        fields = _to_fields(item, profile.default_lane, cand.score, cand.hits, judged=cand.judged)
        try:
            name = repo.add_lead(fields)
        except psycopg.errors.UniqueViolation:
            # Opportunity already tracked under this slug: mark keys seen and skip
            logger.info(
                "lead already tracked for %s — %s (slug collision)",
                item.company,
                item.role,
            )
            seen.add(listing_link, slug)
            if item.link and item.link != listing_link:
                seen.add(item.link, slug)
            continue
        if not name:
            continue
        # Staged: record keys, then collect. Cover letter is drafted on demand.
        seen.add(listing_link, slug)
        if item.link and item.link != listing_link:
            seen.add(item.link, slug)
        added.append(name)

    return added


def _to_fields(
    item: Listing, lane: str, score: int, hits: list[str], *, judged: bool = False
) -> AppFields:
    snippet = item.description[:280].rstrip()
    matched = f"Matched keywords: {', '.join(hits[:6])}." if hits else ""
    if judged:
        matched = f"Fit {score}/100 by Jev. {matched}".strip()
    notes = f"**Auto-discovered** via {item.source}. {matched}\n\n{snippet}".strip()
    return AppFields(
        company=item.company,
        role=item.role,
        lane=lane,
        status="lead",
        link=item.link,
        location=item.location,
        salary=item.salary,
        source=f"auto:{item.source}",
        score=str(min(100, max(0, score))),
        notes=notes,
    )