"""Bluesky (AT Protocol) client — sessions, PDS resolution, posts, and images.

FeedEcho connects to Bluesky using app passwords (the standard method for
bots and automation). App passwords are created in the Bluesky app under
Settings > Privacy & Security > App Passwords. They are scoped to app-level
actions (posting among them) and cannot change account settings or be used
to log in to the app.
"""

from __future__ import annotations

import base64
import json
import re
import unicodedata
from datetime import datetime, timedelta, timezone
from urllib.parse import quote

import httpx

from feed_parser import SSRFError, pinned_request, validate_outbound_url
from utils import (
    DestinationAuthError,
    DestinationError,
    DestinationNotFoundError,
    json_error_detail,
)

PUBLIC_API = "https://public.api.bsky.app"
PLC_DIRECTORY = "https://plc.directory"
POST_COLLECTION = "app.bsky.feed.post"
POST_RECORD_TYPE = "app.bsky.feed.post"

MAX_POST_GRAPHEMES = 300
# app.bsky.feed.post's `text` field caps at maxLength: 3000 UTF-8 bytes, a
# second, independent limit from maxGraphemes above — a post can be under
# 300 grapheme clusters and still be over 3000 bytes (e.g. ZWJ emoji
# sequences, which pack many bytes into one cluster).
MAX_POST_BYTES = 3000
MAX_ALT_GRAPHEMES = 1000
# Bluesky raised the app.bsky.embed.images limit 1 MB -> 2 MB in April 2026
# (atproto PR #4823). Images above this are downscaled/re-encoded by
# images.downscale_image before upload (scheduler._send_bluesky).
MAX_BLOB_BYTES = 2_000_000
BLUESKY_IMAGE_TYPES = {"image/jpeg", "image/png", "image/webp", "image/gif"}
# app.bsky.embed.images carries up to 4 images per post; scheduler._send_bluesky
# attaches the item's image_urls up to this cap (Mastodon's default cap is the
# same number, but the two are independent transport limits).
MAX_IMAGES = 4

# How long to trust a cached access JWT before refreshing (JWT exp takes
# precedence when decodable). Access tokens typically live ~2 hours.
DEFAULT_SESSION_TTL_SECONDS = 2 * 60 * 60


class BlueskyError(DestinationError):
    """Base error for Bluesky API interactions."""


class BlueskyAuthError(BlueskyError, DestinationAuthError):
    """Credentials rejected, session expired, or app password revoked."""


class BlueskyAccountGoneError(BlueskyError, DestinationNotFoundError):
    """The account row was deleted mid-dispatch — retrying can never help."""


def _error_detail(response) -> str:
    """Extract the PDS-provided error message from a JSON error body."""
    return json_error_detail(response, "message", "error")


def _is_token_error(detail: str) -> bool:
    """Detect expired/revoked-token responses regardless of status code.

    bsky.social historically returns ExpiredToken as a 400; treat both
    explicit 401s and ExpiredToken/InvalidToken bodies as auth failures.
    """
    return "ExpiredToken" in detail or "InvalidToken" in detail


# ── Handle normalization ─────────────────────────────────────────────────────


_HANDLE_RE = re.compile(
    r"^[a-z0-9]([a-z0-9-]*[a-z0-9])?(\.[a-z0-9]([a-z0-9-]*[a-z0-9])?)+$"
)
_HANDLE_MAX_LEN = 253


def normalize_handle(raw: str) -> str:
    """Normalize user-entered handles to lowercase bare form.

    Accepts "Name", "@Name.Bsky.Social", and "https://bsky.app/profile/x"
    style inputs (the host part is taken from URLs). Enforces the AT Protocol
    handle grammar and length limit. Raises ValueError on invalid input.
    """
    value = (raw or "").strip().lower().removeprefix("@")
    value = re.sub(r"^https?://", "", value)
    # Take the last path segment if given a profile URL.
    value = value.split("/")[-1].strip()
    value = value.rstrip(".").strip()
    if not value or len(value) > _HANDLE_MAX_LEN or not _HANDLE_RE.match(value):
        raise ValueError("Enter a valid Bluesky handle, e.g. username.bsky.social")
    return value


# ── PDS discovery ────────────────────────────────────────────────────────────


def resolve_pds(handle: str) -> tuple[str, str]:
    """Resolve a handle to (did, pds_url) via the public identity service.

    Handles hosted on bsky.social and custom PDSes both work: the DID
    document's #atproto_pds service entry tells us where the account's data
    lives, so we never hardcode bsky.social.
    """
    handle = normalize_handle(handle)
    validate_outbound_url(PUBLIC_API)

    try:
        response = pinned_request(
            "GET",
            f"{PUBLIC_API}/xrpc/com.atproto.identity.resolveHandle",
            timeout=15.0,
            params={"handle": handle},
        )
        response.raise_for_status()
        data = response.json()
    except (httpx.HTTPError, ValueError) as exc:
        raise BlueskyError(f"Could not resolve handle '{handle}'") from exc

    did = data.get("did")
    if not isinstance(did, str) or not did:
        raise BlueskyError(f"Handle '{handle}' did not resolve to a DID")

    if did.startswith("did:web:"):
        # did:web:example.com -> https://example.com/.well-known/did.json
        doc_url = f"https://{did.removeprefix('did:web:')}/.well-known/did.json"
        # The doc host is derived from remote identity data — SSRF-validate it
        # before fetching, like every other outbound hop.
        validate_outbound_url(doc_url)
        try:
            response = pinned_request("GET", doc_url, timeout=15.0)
            response.raise_for_status()
            doc = response.json()
        except (httpx.HTTPError, ValueError) as exc:
            raise BlueskyError(f"Could not fetch DID document for '{handle}'") from exc
    else:
        # did:plc:... -> query the PLC directory for the PDS service endpoint.
        try:
            response = pinned_request(
                "GET", f"{PLC_DIRECTORY}/{quote(did, safe='')}", timeout=15.0
            )
            response.raise_for_status()
            doc = response.json()
        except (httpx.HTTPError, ValueError) as exc:
            raise BlueskyError(f"Could not fetch DID document for '{handle}'") from exc

    # The DID document comes from a remote (for did:web, handle-controlled)
    # host, so nothing about its shape is guaranteed. Without these checks a
    # document that is a JSON array or has non-dict service entries raised
    # AttributeError, which none of the callers' except clauses catch.
    if not isinstance(doc, dict):
        raise BlueskyError(f"Invalid DID document for '{handle}'")
    pds = ""
    for service in doc.get("service") or []:
        if not isinstance(service, dict):
            continue
        endpoint = service.get("serviceEndpoint")
        if service.get("id") == "#atproto_pds" and isinstance(endpoint, str) and endpoint:
            pds = endpoint
            break
    if not pds:
        raise BlueskyError(f"No PDS endpoint found for '{handle}'")

    pds = pds.rstrip("/")
    # The PDS hostname came from a remote identity document — validate it
    # against SSRF/private-IP rules before making requests to it.
    validate_outbound_url(pds)
    return did, pds


# ── Sessions ─────────────────────────────────────────────────────────────────


def _decode_jwt_exp(jwt_str: str) -> datetime | None:
    """Return the exp claim as an aware UTC datetime, or None if undecodable."""
    try:
        payload = jwt_str.split(".")[1]
        # base64url decode with padding restored
        padded = payload + "=" * (-len(payload) % 4)
        claims = json.loads(base64.urlsafe_b64decode(padded))
        exp = claims.get("exp")
        if isinstance(exp, (int, float)):
            return datetime.fromtimestamp(exp, tz=timezone.utc)
    except (IndexError, ValueError, TypeError, json.JSONDecodeError):
        pass
    return None


def create_session(pds: str, handle: str, app_password: str) -> dict:
    """Create a session with an app password. Returns did + JWTs."""
    pds = pds.rstrip("/")
    validate_outbound_url(pds)

    try:
        response = pinned_request(
            "POST",
            f"{pds}/xrpc/com.atproto.server.createSession",
            timeout=30.0,
            json={"identifier": handle, "password": app_password},
        )
    except (httpx.RequestError, SSRFError) as exc:
        raise BlueskyError(f"Could not reach PDS {pds}") from exc

    if response.status_code in (400, 401):
        detail = _error_detail(response)
        suffix = f" ({detail})" if detail else ""
        raise BlueskyAuthError(f"Invalid handle or app password{suffix}")
    if response.status_code >= 400:
        detail = _error_detail(response)
        suffix = f" ({detail})" if detail else ""
        raise BlueskyError(f"PDS returned HTTP {response.status_code}{suffix}")

    try:
        data = response.json()
    except ValueError as exc:
        raise BlueskyError("PDS returned an invalid session response") from exc

    did = data.get("did")
    access_jwt = data.get("accessJwt")
    refresh_jwt = data.get("refreshJwt")
    if not all(isinstance(v, str) and v for v in (did, access_jwt, refresh_jwt)):
        raise BlueskyError("PDS returned an incomplete session response")
    return {"did": did, "access_jwt": access_jwt, "refresh_jwt": refresh_jwt}


def refresh_session(pds: str, refresh_jwt: str) -> dict:
    """Refresh a session using its refresh JWT."""
    pds = pds.rstrip("/")
    validate_outbound_url(pds)

    try:
        response = pinned_request(
            "POST",
            f"{pds}/xrpc/com.atproto.server.refreshSession",
            timeout=30.0,
            headers={"Authorization": f"Bearer {refresh_jwt}"},
        )
    except (httpx.RequestError, SSRFError) as exc:
        raise BlueskyError(f"Could not reach PDS {pds}") from exc

    if response.status_code in (400, 401):
        detail = _error_detail(response)
        suffix = f" ({detail})" if detail else ""
        raise BlueskyAuthError(f"Session refresh rejected{suffix}")
    if response.status_code >= 400:
        detail = _error_detail(response)
        suffix = f" ({detail})" if detail else ""
        raise BlueskyError(f"PDS returned HTTP {response.status_code}{suffix}")

    try:
        data = response.json()
    except ValueError as exc:
        raise BlueskyError("PDS returned an invalid refresh response") from exc

    access_jwt = data.get("accessJwt")
    new_refresh = data.get("refreshJwt")
    did = data.get("did")
    if not isinstance(access_jwt, str) or not access_jwt or not isinstance(did, str):
        raise BlueskyError("PDS returned an incomplete refresh response")
    if not isinstance(new_refresh, str) or not new_refresh:
        # AT Protocol rotates refresh tokens on every refresh; a missing
        # rotated token means the old one is dead. Do not reuse it.
        raise BlueskyError("PDS refresh response omitted the rotated refresh token")
    return {"did": did, "access_jwt": access_jwt, "refresh_jwt": new_refresh}


def delete_session(pds: str, refresh_jwt: str) -> None:
    """Best-effort session cleanup (used by connection tests). Never raises."""
    pds = pds.rstrip("/")
    try:
        validate_outbound_url(pds)
        pinned_request(
            "POST",
            f"{pds}/xrpc/com.atproto.server.deleteSession",
            timeout=15.0,
            headers={"Authorization": f"Bearer {refresh_jwt}"},
        )
    except Exception:
        # SSRF validation failure or network error — cleanup is best-effort.
        pass


def session_expiry(access_jwt: str) -> str:
    """SQLite timestamp for when a cached access JWT should be refreshed."""
    exp = _decode_jwt_exp(access_jwt)
    if exp is None:
        exp = datetime.now(timezone.utc) + timedelta(seconds=DEFAULT_SESSION_TTL_SECONDS)
    else:
        # Refresh a minute early to avoid racing the actual expiry.
        exp = exp - timedelta(seconds=60)
    return exp.strftime("%Y-%m-%d %H:%M:%S")


# ── Rich text facets ─────────────────────────────────────────────────────────


_URL_RE = re.compile(r"https?://[^\s<>\"']+")
# Hashtag detection mirrors Bluesky's own richtext rules (atproto:
# packages/api/src/rich-text/util.ts TAG_REGEX): a tag starts at the
# beginning of the text or after whitespace, uses an ASCII or fullwidth
# hash, and runs to the next whitespace or zero-width character. The
# candidate's trailing punctuation is stripped, it must keep at least one
# character that is neither an ASCII digit nor punctuation (so "#123" is
# not a tag while a tag of non-ASCII digits is -- JS \d matches ASCII
# only), and it caps at 64 graphemes / 640 bytes -- the client and the
# tag lexicon limits. Cashtags ($TICKER) and mentions (@handle) are deliberately not
# detected: feeds carry foreign @user@instance handles that cannot resolve
# to Bluesky DIDs, and $TICKER content is rare in feeds.
_TAG_RE = re.compile(
    r"(^|\s)([#＃](?!\ufe0f)[^\s\u00AD\u2060\u200A\u200B\u200C\u200D\u20e2]*)"
)
_TAG_MAX_CHARS = 64
_TAG_MAX_BYTES = 640
_ASCII_DIGITS = "0123456789"
_FACET_LINK_TYPE = "app.bsky.richtext.facet#link"
_FACET_TAG_TYPE = "app.bsky.richtext.facet#tag"


def _strip_trailing_punctuation(value: str) -> str:
    """Drop trailing Unicode punctuation (atproto trim + TRAILING_PUNCTUATION)."""
    while value and unicodedata.category(value[-1]).startswith("P"):
        value = value[:-1]
    return value


def _has_tag_body_char(tag: str) -> bool:
    r"""Whether the tag keeps a character that is neither digit nor punctuation.

    atproto's TAG_REGEX requires one such character inside the body; it is
    what makes "#123" render as plain text rather than a tag. Only ASCII
    digits are excluded (JS \d is ASCII-only), so a tag of non-ASCII digits
    like "#٢٠٢٦" is a tag to the official client and stays one here.

    `tag` is always derived from a _TAG_RE match with trailing punctuation
    stripped, so it can never contain whitespace or the zero-width
    characters _TAG_RE's body class already excludes -- only digits and
    punctuation need checking here.
    """
    for ch in tag:
        if ch in _ASCII_DIGITS:
            continue
        if unicodedata.category(ch).startswith("P"):
            continue
        return True
    return False


def build_facets(
    text: str, extra_links: list[tuple[int, int, str]] | None = None
) -> list[dict]:
    """Find URLs and hashtags in post text and build richtext facets.

    Byte offsets are relative to the UTF-8 encoding of the text, as required
    by app.bsky.richtext.facet. Link and tag detection follow Bluesky's own
    richtext rules: trailing punctuation is trimmed from each URL; a tag must
    start the text or follow whitespace, and keeps at least one non-digit,
    non-punctuation character. Truncation artifacts are dropped ("…" inside a
    match), and a tag candidate overlapping a URL range is skipped so facets
    never overlap.

    extra_links: optional (char_start, char_end, uri) spans recovered from
    rich rendering ({{ content_html }} anchors, see template_engine
    .render_template_rich). They take precedence: URL and tag detection skip
    any candidate overlapping an extra span, and spans beyond the text
    (truncated away) are dropped.
    """
    encoded = text.encode("utf-8")

    # (char_start, char_end, facet) — overlap checks run in character space;
    # byte offsets are derived from character positions per final match.
    found: list[tuple[int, int, dict]] = []

    # Seed with rich-render link spans. Offsets computed from the text
    # itself are always valid decodes, so no truncation guard is needed;
    # spans past the end (content truncated away) are dropped here.
    extra_spans: list[tuple[int, int]] = []
    for char_start, char_end, uri in extra_links or []:
        if not (0 <= char_start < char_end <= len(text)) or not uri:
            continue
        # Overlapping extra spans would produce overlapping facets, which
        # Bluesky's record validation rejects. The converter cannot emit
        # them (anchors never nest), so treat overlap as a caller bug and
        # keep the first span.
        if any(char_start < end and start < char_end for start, end in extra_spans):
            continue
        byte_start = len(text[:char_start].encode("utf-8"))
        byte_end = len(text[:char_end].encode("utf-8"))
        found.append(
            (
                char_start,
                char_end,
                {
                    "index": {"byteStart": byte_start, "byteEnd": byte_end},
                    "features": [
                        {"$type": _FACET_LINK_TYPE, "uri": uri},
                    ],
                },
            )
        )
        extra_spans.append((char_start, char_end))

    for match in _URL_RE.finditer(text):
        uri = match.group(0)
        # If the truncation ellipsis is inside the match, the URL was cut off
        # mid-way — drop it instead of linkifying a broken/partial URL.
        if "…" in uri:
            continue
        # Strip trailing punctuation that is unlikely to be part of the URL.
        while uri and uri[-1] in ".,;:!?'\"()[]{}<>":
            uri = uri[:-1]
        if not uri:
            continue

        # Locate the trimmed URI's byte range inside the encoded text.
        char_start = match.start()
        byte_start = len(text[:char_start].encode("utf-8"))
        byte_end = byte_start + len(uri.encode("utf-8"))

        # If the text was truncated mid-URL, the byte range won't decode back
        # to the full URI — drop the facet instead of linkifying a broken URL.
        if byte_end > len(encoded) or encoded[byte_start:byte_end].decode("utf-8") != uri:
            continue

        # A rich-render anchor already covers this range — keep its facet.
        span_end = char_start + len(uri)
        if any(char_start < end and start < span_end for start, end in extra_spans):
            continue

        found.append(
            (
                char_start,
                char_start + len(uri),
                {
                    "index": {"byteStart": byte_start, "byteEnd": byte_end},
                    "features": [
                        {"$type": _FACET_LINK_TYPE, "uri": uri},
                    ],
                },
            )
        )

    link_spans = [(start, end) for start, end, _ in found]

    for match in _TAG_RE.finditer(text):
        leading = match.group(1)
        candidate = match.group(2)
        # Same truncation guard as URLs: an ellipsis inside the candidate
        # means the tag was sliced mid-word.
        if "…" in candidate:
            continue
        tag = _strip_trailing_punctuation(candidate[1:])
        if not tag:
            continue
        # Tag cap: 64 graphemes (the lexicon limit). len(tag) is the cheap
        # pre-check the client also uses -- code points are always >=
        # graphemes, so cluster counting only runs for over-cap raw lengths.
        # The lexicon separately caps the property at 640 UTF-8 bytes; an
        # oversized tag facet would fail record validation and kill the
        # whole post, so guard that too.
        if (
            len(tag) > _TAG_MAX_CHARS
            and len(_grapheme_clusters(tag)) > _TAG_MAX_CHARS
        ):
            continue
        if len(tag.encode("utf-8")) > _TAG_MAX_BYTES:
            continue
        if not _has_tag_body_char(tag):
            continue

        char_start = match.start() + len(leading)  # position of the hash
        char_end = char_start + 1 + len(tag)
        if any(char_start < end and start < char_end for start, end in link_spans):
            continue

        byte_start = len(text[:char_start].encode("utf-8"))
        byte_end = len(text[:char_end].encode("utf-8"))
        found.append(
            (
                char_start,
                char_end,
                {
                    "index": {"byteStart": byte_start, "byteEnd": byte_end},
                    "features": [
                        {"$type": _FACET_TAG_TYPE, "tag": tag},
                    ],
                },
            )
        )

    found.sort(key=lambda entry: entry[0])
    return [facet for _, _, facet in found]


# ── Grapheme-aware truncation ────────────────────────────────────────────────


_ZWJ = "\u200d"
_VARIATION_SELECTORS = {"\ufe0e", "\ufe0f"}
_SKIN_TONE_RANGE = range(0x1F3FB, 0x1F400)
_EMOJI_TAG_RANGE = range(0xE0020, 0xE0080)

# Python's unicodedata does not expose Extended_Pictographic (it is not a
# general category), and the `regex` package that provides
# \p{Extended_Pictographic} is not a project dependency (checked
# requirements.txt / pyproject.toml). This is a range-based approximation
# covering the Unicode blocks that hold the overwhelming majority of
# real-world emoji, used only to gate GB11 below. It is NOT a complete
# UAX #29 implementation: a few rare Extended_Pictographic code points
# outside these ranges will be missed (under-glue -- same failure mode the
# old code already had for other boundaries, harmless for the 300/3000
# caps, just an occasional visually-split emoji), but ordinary letters,
# digits, and punctuation are reliably excluded, which is what fixes the
# bug this guards against.
_EXTENDED_PICTOGRAPHIC_RANGES = (
    (0x00A9, 0x00A9),    # copyright
    (0x00AE, 0x00AE),    # registered
    (0x203C, 0x203C),    # double exclamation
    (0x2049, 0x2049),    # exclamation question
    (0x2122, 0x2122),    # trademark
    (0x2139, 0x2139),    # information
    (0x2194, 0x21AA),    # arrows used as emoji
    (0x231A, 0x231B),    # watch, hourglass
    (0x2328, 0x2328),    # keyboard
    (0x23CF, 0x23CF),
    (0x23E9, 0x23FA),    # playback controls / clocks
    (0x24C2, 0x24C2),
    (0x25AA, 0x25FE),    # geometric shapes used as emoji
    (0x2600, 0x27BF),    # misc symbols + dingbats (most classic emoji)
    (0x2934, 0x2935),
    (0x2B00, 0x2BFF),    # misc symbols and arrows (stars, etc.)
    (0x3030, 0x3030),
    (0x303D, 0x303D),
    (0x3297, 0x3297),
    (0x3299, 0x3299),
    (0x1F000, 0x1FAFF),  # mahjong/dominoes through symbols & pictographs ext-A
    (0x1FB00, 0x1FBFF),  # legacy computing symbols (some emoji-adjacent)
)


def _is_regional_indicator(ch: str) -> bool:
    return 0x1F1E6 <= ord(ch) <= 0x1F1FF


def _is_extended_pictographic(ch: str) -> bool:
    """Approximate check for Unicode's Extended_Pictographic property.

    Restricts the ZWJ forward-glue (GB11) below to actual emoji
    continuations -- see _EXTENDED_PICTOGRAPHIC_RANGES for why this is a
    range-based stand-in rather than a full property lookup.
    """
    cp = ord(ch)
    return any(lo <= cp <= hi for lo, hi in _EXTENDED_PICTOGRAPHIC_RANGES)


def _trailing_ri_count(cluster: str) -> int:
    n = 0
    for ch in reversed(cluster):
        if _is_regional_indicator(ch):
            n += 1
        else:
            break
    return n


def _grapheme_clusters(text: str) -> list[str]:
    """Split text into grapheme clusters using a conservative UAX #29 subset.

    A new grapheme starts at any character that is not a combining mark,
    zero-width joiner, variation selector, skin-tone modifier, or emoji tag
    character. A ZWJ always glues to the cluster before it (GB9). A
    character *after* a ZWJ only glues onto that cluster when it is itself
    emoji/Extended_Pictographic (an approximation of GB11) -- a ZWJ used for
    non-emoji purposes (e.g. some Indic-script ligatures) no longer swallows
    whatever ordinary character happens to follow it. Regional-indicator
    pairs (flags) merge so a flag never splits. The remaining gaps
    over-merge, which can only under-count graphemes, so truncation never
    exceeds platform limits (that reasoning does NOT extend to the ZWJ
    forward-glue case, which is why it is now gated on Extended_Pictographic
    instead of firing unconditionally).
    """
    clusters: list[str] = []
    for ch in text:
        cat = unicodedata.category(ch)
        is_continuation = (
            cat in ("Mn", "Mc", "Me")
            or ch in _VARIATION_SELECTORS
            or ord(ch) in _SKIN_TONE_RANGE
            or ord(ch) in _EMOJI_TAG_RANGE
        )
        if clusters:
            is_continuation = (
                is_continuation
                or ch == _ZWJ
                or (
                    clusters[-1][-1] == _ZWJ
                    and _is_extended_pictographic(ch)
                )
                or (
                    _is_regional_indicator(ch)
                    and _trailing_ri_count(clusters[-1]) == 1
                )
            )
        if clusters and is_continuation:
            clusters[-1] += ch
        else:
            clusters.append(ch)
    return clusters


def truncate_graphemes(
    text: str,
    max_graphemes: int = MAX_POST_GRAPHEMES,
    max_bytes: int | None = None,
    *,
    preserve_links: bool = True,
) -> str:
    """Truncate to max_graphemes grapheme clusters, appending an ellipsis.

    When max_bytes is given, the result is also capped at that many UTF-8
    bytes -- app.bsky.feed.post's `text` enforces maxGraphemes AND maxLength
    (bytes) as two independent limits, and a handful of grapheme clusters
    can still add up to well over 3000 bytes (e.g. dense ZWJ emoji
    sequences). Byte truncation drops whole grapheme clusters from the end
    rather than slicing raw bytes, so it never splits a UTF-8 multi-byte
    sequence.

    When preserve_links=True (default), if a trailing URL would be severed
    or dropped by right-side truncation, preceding text is truncated instead
    so the link remains intact and clickable.
    """
    clusters = _grapheme_clusters(text)
    fits_graphemes = len(clusters) <= max_graphemes
    fits_bytes = max_bytes is None or len(text.encode("utf-8")) <= max_bytes
    if fits_graphemes and fits_bytes:
        return text

    def _assemble_slice(cl: list[str], n: int) -> str:
        head = "".join(cl[:n]).rstrip()
        # Don't leave a dangling ZWJ at the cut point -- it renders as a
        # broken sequence in most clients.
        while head.endswith(_ZWJ):
            head = head[:-1]
        return head + "…"

    if preserve_links:
        urls = list(_URL_RE.finditer(text))
        if urls:
            last_url = urls[-1]
            u_start = last_url.start()
            u_end = last_url.end()

            char_idx = 0
            u_start_cluster = None
            u_end_cluster = None
            for i, c in enumerate(clusters):
                if u_start_cluster is None and char_idx <= u_start < char_idx + len(c):
                    u_start_cluster = i
                char_idx += len(c)
                if char_idx >= u_end and u_end_cluster is None:
                    u_end_cluster = i + 1

            # Determine where naive right-side truncation would cut
            naive_n = min(max_graphemes - 1, len(clusters)) if not fits_graphemes else len(clusters)
            if max_bytes is not None:
                while naive_n > 0 and len(_assemble_slice(clusters, naive_n).encode("utf-8")) > max_bytes:
                    naive_n -= 1

            naive_head = "".join(clusters[:naive_n]).rstrip()
            needs_preservation = False
            if u_start_cluster is not None and u_start_cluster > 0:
                if u_end_cluster is not None and u_end_cluster >= naive_n:
                    needs_preservation = True
                elif naive_head.endswith(last_url.group(0)):
                    needs_preservation = True

            if needs_preservation and u_start_cluster is not None:
                tail_clusters = clusters[u_start_cluster:]
                tail_str = "".join(tail_clusters).rstrip()

                # If tail with suffix exceeds limit, try just the URL itself
                if len(_grapheme_clusters(tail_str)) > max_graphemes and u_end_cluster is not None:
                    tail_clusters = clusters[u_start_cluster:u_end_cluster]
                    tail_str = "".join(tail_clusters).rstrip()

                tail_graphemes = len(_grapheme_clusters(tail_str))
                if tail_graphemes <= max_graphemes:
                    head_clusters = clusters[:u_start_cluster]
                    head_budget = max_graphemes - tail_graphemes - 2
                    if head_budget > 0:
                        head_str = _assemble_slice(head_clusters, head_budget)
                        result = head_str.rstrip() + " " + tail_str.lstrip()
                    elif tail_graphemes <= max_graphemes - 2:
                        result = "… " + tail_str.lstrip()
                    elif tail_graphemes <= max_graphemes - 1:
                        result = "…" + tail_str.lstrip()
                    else:
                        result = tail_str.strip()

                    if max_bytes is not None:
                        while len(result.encode("utf-8")) > max_bytes and head_budget > 0:
                            head_budget -= 1
                            if head_budget > 0:
                                head_str = _assemble_slice(head_clusters, head_budget)
                                result = head_str.rstrip() + " " + tail_str.lstrip()
                            elif tail_graphemes <= max_graphemes - 2:
                                result = "… " + tail_str.lstrip()
                            elif tail_graphemes <= max_graphemes - 1:
                                result = "…" + tail_str.lstrip()
                            else:
                                result = tail_str.strip()

                    res_clusters = _grapheme_clusters(result)
                    if len(res_clusters) <= max_graphemes and (
                        max_bytes is None or len(result.encode("utf-8")) <= max_bytes
                    ):
                        return result.rstrip()

    n = min(max_graphemes - 1, len(clusters)) if not fits_graphemes else len(clusters)
    result = _assemble_slice(clusters, n)
    if max_bytes is not None:
        while n > 0 and len(result.encode("utf-8")) > max_bytes:
            n -= 1
            result = _assemble_slice(clusters, n)
    return result


# ── Posts and images ─────────────────────────────────────────────────────────


def upload_blob(
    pds: str, access_jwt: str, image_bytes: bytes, content_type: str
) -> dict | None:
    """Upload an image blob.

    Enforces Bluesky's image constraints (type allowlist, 1 MB cap). Returns
    the blob reference dict, or None on transport failure. Raises
    BlueskyAuthError on expired/revoked sessions and BlueskyError on other
    API rejections, so callers can distinguish "skip the image" from
    "refresh the session".
    """
    pds = pds.rstrip("/")
    validate_outbound_url(pds)

    if content_type not in BLUESKY_IMAGE_TYPES:
        raise BlueskyError(f"Unsupported image type for Bluesky: {content_type}")
    if len(image_bytes) > MAX_BLOB_BYTES:
        raise BlueskyError("Image exceeds the Bluesky blob limit")

    try:
        response = pinned_request(
            "POST",
            f"{pds}/xrpc/com.atproto.repo.uploadBlob",
            timeout=60.0,
            headers={
                "Authorization": f"Bearer {access_jwt}",
                "Content-Type": content_type,
            },
            content=image_bytes,
        )
    except (httpx.RequestError, SSRFError):
        return None

    detail = _error_detail(response)
    if response.status_code == 401 or _is_token_error(detail):
        suffix = f" ({detail})" if detail else ""
        raise BlueskyAuthError(f"Blob upload rejected: session expired{suffix}")
    if response.status_code >= 400:
        suffix = f" ({detail})" if detail else ""
        raise BlueskyError(f"Blob upload failed (HTTP {response.status_code}){suffix}")

    try:
        blob = response.json().get("blob")
    except ValueError:
        return None
    if not isinstance(blob, dict) or "ref" not in blob:
        return None
    return blob


def create_post(
    pds: str,
    access_jwt: str,
    repo: str,
    text: str,
    facets: list[dict] | None = None,
    embed: dict | None = None,
) -> dict:
    """Create a post record. Returns {uri, cid}. Raises BlueskyAuthError on 401."""
    pds = pds.rstrip("/")
    validate_outbound_url(pds)

    record: dict = {
        "$type": POST_RECORD_TYPE,
        "text": text,
        "createdAt": datetime.now(timezone.utc).strftime("%Y-%m-%dT%H:%M:%SZ"),
    }
    if facets:
        record["facets"] = facets
    if embed:
        record["embed"] = embed

    payload = {"repo": repo, "collection": POST_COLLECTION, "record": record}

    try:
        response = pinned_request(
            "POST",
            f"{pds}/xrpc/com.atproto.repo.createRecord",
            timeout=30.0,
            headers={"Authorization": f"Bearer {access_jwt}"},
            json=payload,
        )
    except (httpx.RequestError, SSRFError) as exc:
        raise BlueskyError(f"Could not reach PDS {pds}") from exc

    detail = _error_detail(response)
    if response.status_code == 401 or _is_token_error(detail):
        # 401 (or a 400 carrying ExpiredToken/InvalidToken) is a session
        # problem. A plain 400 is a record validation error, not auth.
        suffix = f" ({detail})" if detail else ""
        raise BlueskyAuthError(f"Post rejected (HTTP {response.status_code}){suffix}")
    if response.status_code >= 400:
        suffix = f" ({detail})" if detail else ""
        raise BlueskyError(f"Post rejected (HTTP {response.status_code}){suffix}")

    try:
        data = response.json()
    except ValueError as exc:
        raise BlueskyError("PDS returned an invalid post response") from exc
    if not isinstance(data.get("uri"), str) or not isinstance(data.get("cid"), str):
        raise BlueskyError("PDS returned an incomplete post response")
    return {"uri": data["uri"], "cid": data["cid"]}


def build_image_embed(image_entries: list[dict]) -> dict:
    """Wrap uploaded blobs in an app.bsky.embed.images embed.

    Takes a list of {"blob", "alt"} entries (scheduler._send_bluesky builds
    them) and drops anything beyond MAX_IMAGES defensively — the per-post
    cap belongs to this transport, not the caller. The lexicon requires at
    least one image, so an empty list raises ValueError instead of
    returning a payload the PDS would reject.
    """
    if not image_entries:
        raise ValueError("build_image_embed requires at least one image entry")
    images = [
        {
            "alt": truncate_graphemes(
                (entry.get("alt") or "").strip(), MAX_ALT_GRAPHEMES
            ),
            "image": entry["blob"],
        }
        for entry in image_entries[:MAX_IMAGES]
    ]
    return {"$type": "app.bsky.embed.images", "images": images}


EXTERNAL_TITLE_MAX_CHARS = 200
EXTERNAL_DESCRIPTION_MAX_CHARS = 500
# The external thumb carries its own lexicon cap (maxSize 1000000), which
# stayed at 1 MB when app.bsky.embed.images was raised to 2 MB in April
# 2026 — a 1-2 MB thumb here would upload fine but get the whole record
# rejected with a 400 on createRecord.
EXTERNAL_THUMB_MAX_BYTES = 1_000_000


def build_external_embed(
    uri: str, title: str, description: str = "", thumb_blob: dict | None = None
) -> dict:
    """Build an app.bsky.embed.external link card.

    The card Bluesky renders when a post carries a link: the linked page's
    Open Graph title/description, optionally an uploaded thumbnail blob.
    Truncation is deliberately conservative — well inside the lexicon's
    byte limits for any plausible title, so the PDS never rejects the
    record over an oversized card. A missing thumb just renders a
    text-only card, matching how the official client handles pages
    without og:image.
    """
    external: dict = {
        "uri": uri,
        "title": truncate_graphemes((title or "").strip(), EXTERNAL_TITLE_MAX_CHARS),
        "description": truncate_graphemes(
            (description or "").strip(), EXTERNAL_DESCRIPTION_MAX_CHARS
        ),
    }
    if thumb_blob:
        external["thumb"] = thumb_blob
    return {"$type": "app.bsky.embed.external", "external": external}


def first_link_uri(facets: list[dict] | None) -> str | None:
    """Return the uri of the first link facet, or None.

    build_facets seeds rich-render link spans before bare-URL detection,
    so first-wins here gives the same precedence the rendered post shows:
    a {{ content_html }} anchor beats a trailing bare permalink.
    """
    for facet in facets or []:
        for feature in facet.get("features", []):
            if (
                isinstance(feature, dict)
                and feature.get("$type") == _FACET_LINK_TYPE
                and feature.get("uri")
            ):
                return feature["uri"]
    return None


# ── Connection testing ───────────────────────────────────────────────────────


def test_connection(handle: str, app_password: str) -> tuple[bool, str]:
    """Resolve the handle and verify the app password works."""
    try:
        handle = normalize_handle(handle)
        did, pds = resolve_pds(handle)
        session = create_session(pds, handle, app_password)
        try:
            return True, f"Connected as @{handle} ({session['did'][:20]}…)"
        finally:
            # Don't accumulate server-side sessions from repeated tests —
            # bsky.social rate-limits createSession aggressively.
            delete_session(pds, session["refresh_jwt"])
    except ValueError as e:
        return False, str(e)
    except BlueskyAuthError:
        return False, "Invalid handle or app password"
    except BlueskyError as e:
        return False, str(e)