Blame for backend/app/routers/launchpad.py

"""Launchpad: the calendar, connected social accounts (OAuth / app password) and tracked short links."""

import json
import re
import secrets
import uuid
from datetime import UTC, datetime, timedelta
from typing import Literal
from urllib.parse import urlencode
from zoneinfo import ZoneInfo, ZoneInfoNotFoundError

from fastapi import APIRouter, Depends, HTTPException, Request
from fastapi.concurrency import run_in_threadpool
from fastapi.responses import RedirectResponse
from pydantic import BaseModel, Field
from sqlalchemy import select
from sqlalchemy.orm import Session

from app.auth import Ctx, decode_token_full, encode_signed, get_ctx
from app.config import settings
from app.db import get_db
from app.models import Artifact, CalendarItem, Document, EngagementEvent, SocialAccount, TrackedLink, Upload
from app.services import launchpad as lp
from app.services.artifacts import TYPE_LABELS, download_name
from app.services.crypto import decrypt, encrypt
from app.services.social import AUTO_POST, LIMITS, PLATFORM_LABELS, SocialError, bluesky, linkedin, x
from app.services.social.health import pause_account_posts, resume_account_posts
from app.services.social.media import (
    MB,
    SIZE_LIMITS,
    UPLOAD_TYPES,
    carries_media,
    check_postable,
    duration,
    is_blog2video,
    media_count,
    upload_kind,
)
from app.services.storage import storage

router = APIRouter(prefix="/api", tags=["launchpad"])

Platform = Literal["x", "linkedin", "bluesky", "substack_notes"]


# Calendar


class When(BaseModel):
    """When a post goes out: the wall time the user picked in their own time zone (local_time "2026-10-07T11:00" +
    timezone "Asia/Karachi", on a half hour), converted to UTC here; or an exact scheduled_at (Post now)."""
    scheduled_at: datetime | None = None
    local_time: str | None = Field(None, max_length=32)
    timezone: str | None = Field(None, max_length=64)

    def utc(self) -> datetime | None:
        if self.local_time:
            try:
                zone = ZoneInfo(self.timezone or "")
            except (ZoneInfoNotFoundError, ValueError) as exc:
                raise HTTPException(400, {"code": "bad_timezone", "message": "Unknown time zone."}) from exc
            try:
                wall = datetime.fromisoformat(self.local_time)
            except ValueError as exc:
                raise HTTPException(400, {"code": "bad_time", "message": "Pick a date and time."}) from exc
            if wall.tzinfo or wall.minute not in (0, 30) or wall.second or wall.microsecond:
                raise HTTPException(400, {"code": "half_hour_only", "message": "Pick a time on the hour or half hour."})
            return wall.replace(tzinfo=zone).astimezone(UTC)
        return _aware(self.scheduled_at) if self.scheduled_at else None


class ItemIn(When):
    platform: Platform
    content: str = Field("", max_length=10000)  # may be empty when the post carries images or a video
    thread: list[str] = Field(default_factory=list, max_length=25)
    artifact_id: uuid.UUID | None = None  # the attachment (a video, quote card, carousel...), never a Launch Kit
    kit_id: uuid.UUID | None = None  # the Launch Kit the post was written from (tracked, not attached)
    document_id: uuid.UUID | None = None
    social_account_id: uuid.UUID | None = None
    remind_by_email: bool = False
    draft: bool = False
    # Post now: published (or, by email, reminded) in this request, never left scheduled; no time needed
    publish_now: bool = False


class ItemPatch(When):
    artifact_id: uuid.UUID | None = None  # set or (null) clear what the post carries
    content: str | None = Field(None, max_length=10000)
    thread: list[str] | None = None
    social_account_id: uuid.UUID | None = None
    remind_by_email: bool | None = None
    status: Literal["scheduled", "draft", "paused"] | None = None


def _aware(dt: datetime) -> datetime:
    return dt if dt.tzinfo else dt.replace(tzinfo=UTC)


# audio can't be posted; "upload" is a file the user brought from their computer
POSTABLE = ("video", "quote_card", "carousel", "summary", "launch_kit", "mind_map", "upload", "slide_deck")
# Listed on the Launchpad too, but only to download: X and LinkedIn take no audio files (check_postable refuses it)
LISTED = (*POSTABLE, "audio_overview")


def _cite_free(text: str) -> str:
    return re.sub(r"\s*\[\d+\]", "", text or "").strip()


def _prefill(a: Artifact) -> dict:
    """A starting text per platform for a post of this artifact: X as a list of posts (a thread), LinkedIn as one."""
    c = a.content_json or {}
    title = c.get("title") or TYPE_LABELS.get(a.type, a.type)
    if a.type == "upload":  # their own file: the words are theirs to write
        return {"x": [""], "linkedin": ""}
    if a.type == "launch_kit":
        posts = c.get("posts") or {}
        x_posts = posts.get("x_thread") or posts.get("x") or [title]
        return {"x": x_posts, "linkedin": "\n\n".join(posts.get("linkedin") or x_posts)}
    text = {
        "summary": _cite_free(c.get("summary") or title),
        "quote_card": f'"{c.get("quote")}"' + (f" ({c['source']})" if c.get("source") else "") if c.get("quote")
                      else title,
        "carousel": ((c.get("slides") or [{}])[0].get("heading") or title),
        "slide_deck": _deck_prefill(c) or title,
    }.get(a.type, title)
    return {"x": [text], "linkedin": text}


def _deck_prefill(c: dict) -> str:
    """A slide deck's post: its title, its subtitle, and the takeaways of its closing slide."""
    deck = c.get("deck") or {}
    slides = deck.get("slides") or []
    closing = next((s for s in reversed(slides) if s.get("layout") == "closing"), {})
    head = "\n\n".join(x for x in (deck.get("title"), deck.get("subtitle")) if x)
    takeaways = "\n".join(f"- {t}" for t in closing.get("takeaways") or [])
    return "\n\n".join(x for x in (head, takeaways) if x)


def artifact_brief(a: Artifact) -> dict:
    """What the composer and the calendar show of the attached artifact."""
    c = a.content_json or {}
    thumb = (a.storage_key if a.type == "quote_card" or upload_kind(a) == "image"
             else (c.get("slide_keys") or [a.storage_key])[0] if a.type == "carousel" else None)
    return {"id": str(a.id), "type": a.type, "type_label": TYPE_LABELS.get(a.type, a.type),
            # A deck's title starts "Slide deck: ", which its tag already says
            "title": (c.get("title") or TYPE_LABELS.get(a.type, a.type)).removeprefix("Slide deck: " if a.type == "slide_deck" else ""),
            "media": upload_kind(a) or ("video" if a.type == "video" else "audio" if a.type == "audio_overview"
                                        else "image" if media_count(a) else "text"),
            "media_count": media_count(a), "duration_s": duration(a),
            "thumb_url": storage.presign_get(thumb) if thumb else None,
            # The user's own file, to preview before posting it.
            "view_url": storage.presign_get(a.storage_key) if a.type == "upload" and a.storage_key else None,
            # An audio overview can't be posted: it is offered to download instead.
            "download_url": (storage.presign_get(a.storage_key, download_name=download_name(a))
                             if a.type == "audio_overview" and a.storage_key else None),
            # A blog2video video with scenes but no MP4 yet: listed, but it has to be rendered before it can go out.
            # Made by the video service (blog2video): it opens in the video editor.
            "editable_video": is_blog2video(a),
            "needs_render": is_blog2video(a) and not (a.content_json or {}).get("video_url"),
            # Its MP4 is being made right now: it can't be scheduled until the render finishes.
            "rendering": is_blog2video(a) and a.status == "rendering",
            "created_at": a.created_at.isoformat() if a.created_at else None, "prefill": _prefill(a)}


def serialize_item(item: CalendarItem, account: SocialAccount | None = None, clicks: int = 0,
                   artifact: Artifact | None = None, kit: Artifact | None = None) -> dict:
    return {
        "id": str(item.id),
        "platform": item.platform,
        "platform_label": PLATFORM_LABELS.get(item.platform, item.platform),
        "scheduled_at": _aware(item.scheduled_at).isoformat(),
        "status": item.status,
        "content": item.content,
        "thread": item.thread or [],
        "artifact_id": str(item.artifact_id) if item.artifact_id else None,
        "artifact": artifact_brief(artifact) if artifact else None,
        "kit_id": str(item.kit_id) if item.kit_id else None,
        "kit_title": ((kit.content_json or {}).get("post_title") or (kit.content_json or {}).get("title")) if kit else None,
        "document_id": str(item.document_id) if item.document_id else None,
        "social_account_id": str(item.social_account_id) if item.social_account_id else None,
        "account_handle": account.handle if account else None,
        "remind_by_email": item.remind_by_email,
        "auto_post": item.platform in AUTO_POST and bool(item.social_account_id) and not item.remind_by_email,
        "external_url": item.external_url,
        "posted_at": _aware(item.posted_at).isoformat() if item.posted_at else None,
        "error": item.error,
        "metrics": item.metrics or {},
        "clicks": clicks,
    }


def _item(ctx: Ctx, item_id: uuid.UUID) -> CalendarItem:
    item = ctx.db.scalar(select(CalendarItem).where(CalendarItem.id == item_id,
                                                    CalendarItem.workspace_id == ctx.workspace.id))
    if not item:
        raise HTTPException(404, "Calendar item not found")
    return item


def _account(ctx: Ctx, account_id: uuid.UUID | None, platform: str) -> SocialAccount | None:
    if not account_id:
        return None
    account = ctx.db.scalar(select(SocialAccount).where(SocialAccount.id == account_id,
                                                        SocialAccount.workspace_id == ctx.workspace.id))
    if not account or account.platform != platform:
        raise HTTPException(400, "That account does not match the platform")
    return account


CONNECTED_ONLY = ("x", "linkedin")  # these post only through a live connection (no email fallback)


def can_post_media(a: SocialAccount) -> bool:
    return {"x": x.can_post_media, "linkedin": linkedin.can_post_media}.get(a.platform, lambda _: False)(a)


def _require_connected(platform: str, account: SocialAccount | None, artifact: Artifact | None) -> None:
    """X and LinkedIn: scheduling needs a live connection (and one allowed to post media, when media is attached)."""
    if platform not in CONNECTED_ONLY:
        return
    label = PLATFORM_LABELS[platform]
    if not account or account.status != "active":
        raise HTTPException(409, {"code": "not_connected",
                                  "message": f"Connect {label} to schedule posts there."
                                  if not account else f"Your {label} connection is down. Reconnect {label} first."})
    if artifact and media_count(artifact) and not can_post_media(account):
        raise HTTPException(409, {"code": "reconnect", "message": f"Reconnect {label} to post images and videos."})


def _require_words(posts: list[str], artifact: Artifact | None) -> None:
    """A post needs words, unless it carries images or a video (those can go out without a caption)."""
    if not any(p.strip() for p in posts) and not carries_media(artifact):
        raise HTTPException(422, {"code": "no_text", "message": "Write something to post, or attach an image or video."})


def _validate_lengths(platform: str, posts: list[str]) -> None:
    limit = LIMITS.get(platform)
    if platform in ("x", "bluesky") and limit:
        for i, p in enumerate(posts, start=1):
            if len(p) > limit:
                raise HTTPException(422, {"code": "too_long",
                                          "message": f"Post {i} is {len(p)} characters; {PLATFORM_LABELS[platform]} "
                                                     f"allows {limit}."})


def _enrich(ctx: Ctx, items: list[CalendarItem]) -> list[dict]:
    accounts = {a.id: a for a in ctx.db.scalars(
        select(SocialAccount).where(SocialAccount.workspace_id == ctx.workspace.id))}
    clicks = lp.clicks_by_item(ctx.db, [i.id for i in items])
    ids = {i.artifact_id for i in items if i.artifact_id} | {i.kit_id for i in items if i.kit_id}
    artifacts = {a.id: a for a in ctx.db.scalars(select(Artifact).where(Artifact.id.in_(ids)))} if ids else {}
    return [serialize_item(i, accounts.get(i.social_account_id), clicks.get(i.id, 0), artifacts.get(i.artifact_id),
                           artifacts.get(i.kit_id)) for i in items]


def _kit(ctx: Ctx, kit_id: uuid.UUID) -> Artifact:
    """This workspace's Launch Kit (what a post was written from)."""
    kit = ctx.db.scalar(select(Artifact).where(Artifact.id == kit_id, Artifact.workspace_id == ctx.workspace.id,
                                               Artifact.type == "launch_kit"))
    if not kit:
        raise HTTPException(404, "Launch Kit not found")
    return kit


def _postable(ctx: Ctx, artifact_id: uuid.UUID, platform: str) -> Artifact:
    """This workspace's artifact, checked for the platform (audio, unfinished, too long for X: 400)."""
    a = ctx.db.scalar(select(Artifact).where(Artifact.id == artifact_id, Artifact.workspace_id == ctx.workspace.id))
    if not a:
        raise HTTPException(404, "Artifact not found")
    check_postable(a, platform)
    return a


@router.get("/calendar")
def list_items(ctx: Ctx = Depends(get_ctx), start: datetime | None = None, end: datetime | None = None,
               status: str | None = None, limit: int = 500):
    q = select(CalendarItem).where(CalendarItem.workspace_id == ctx.workspace.id)
    if start:
        q = q.where(CalendarItem.scheduled_at >= start)
    if end:
        q = q.where(CalendarItem.scheduled_at < end)
    if status:
        q = q.where(CalendarItem.status.in_(status.split(",")))
    items = list(ctx.db.scalars(q.order_by(CalendarItem.scheduled_at).limit(min(limit, 1000))).all())
    return _enrich(ctx, items)


@router.post("/calendar")
def create_item(body: ItemIn, ctx: Ctx = Depends(get_ctx)):
    when = datetime.now(UTC) if body.publish_now else body.utc()
    if when is None:
        raise HTTPException(422, {"code": "no_time", "message": "Pick when it goes out."})
    account = _account(ctx, body.social_account_id, body.platform)
    _validate_lengths(body.platform, [body.content, *body.thread])
    artifact = _postable(ctx, body.artifact_id, body.platform) if body.artifact_id else None
    _require_words([body.content, *body.thread], artifact)
    kit = _kit(ctx, body.kit_id) if body.kit_id else None
    _require_connected(body.platform, account, artifact)
    if body.document_id and not ctx.db.scalar(select(Document.id).where(Document.id == body.document_id,
                                                                        Document.workspace_id == ctx.workspace.id)):
        raise HTTPException(404, "Document not found")
    item = CalendarItem(
        workspace_id=ctx.workspace.id, platform=body.platform, scheduled_at=when,
        content=body.content, thread=[t for t in body.thread if t.strip()], artifact_id=body.artifact_id,
        kit_id=kit.id if kit else None,
        document_id=body.document_id, social_account_id=account.id if account else None,
        remind_by_email=body.remind_by_email or body.platform not in AUTO_POST or not account,
        status="draft" if body.draft else "publishing" if body.publish_now else "scheduled",
    )
    ctx.db.add(item)
    ctx.db.commit()
    if body.publish_now:
        _go_now(ctx, item)
    return _enrich(ctx, [item])[0]


def _go_now(ctx: Ctx, item: CalendarItem) -> None:
    """Publish an item claimed for it ("publishing") now: with images or a video through a job (uploading takes a
    while; the page follows its status), anything else right here."""
    if lp.has_media(ctx.db, item):
        lp.queue_publish(ctx.db, item)
    else:
        lp.publish_item(ctx.db, item)


@router.patch("/calendar/{item_id}")
def update_item(item_id: uuid.UUID, body: ItemPatch, ctx: Ctx = Depends(get_ctx)):
    item = _item(ctx, item_id)
    if item.status in ("posted", "publishing"):
        raise HTTPException(409, "This post already went out.")
    when = body.utc()
    if when is not None:
        item.scheduled_at = when
    if body.content is not None:
        item.content = body.content
    if body.thread is not None:
        item.thread = [t for t in body.thread if t.strip()]
    if "artifact_id" in body.model_fields_set:
        item.artifact_id = _postable(ctx, body.artifact_id, item.platform).id if body.artifact_id else None
    if "social_account_id" in body.model_fields_set:
        item.social_account_id = (_account(ctx, body.social_account_id, item.platform).id
                                  if body.social_account_id else None)
    if body.remind_by_email is not None:
        item.remind_by_email = body.remind_by_email
    if not item.social_account_id or item.platform not in AUTO_POST:
        item.remind_by_email = True
    account = ctx.db.get(SocialAccount, item.social_account_id) if item.social_account_id else None
    artifact = ctx.db.get(Artifact, item.artifact_id) if item.artifact_id else None
    if body.status == "paused":
        item.status = "paused"  # held by hand: never attempted until resumed
    elif body.status is not None or item.status in ("failed", "reminded", "paused"):
        # Going out (again), rescheduled: only through a live connection, and only at a time still ahead (a failed or
        # reminded post's old time has passed, and it would otherwise go out at the very next run).
        _require_connected(item.platform, account, artifact)
        if (body.status or "scheduled") == "scheduled" and _aware(item.scheduled_at) <= datetime.now(UTC):
            raise HTTPException(409, {"code": "missed", "message": "Pick a time that is still ahead."})
        item.status = body.status or "scheduled"
    elif item.platform in CONNECTED_ONLY and body.model_fields_set & {"social_account_id", "artifact_id"}:
        _require_connected(item.platform, account, artifact)
    item.error = None
    item.attempts = 0
    _require_words([item.content or "", *(item.thread or [])], artifact)  # e.g. the caption emptied, or the video removed
    _validate_lengths(item.platform, [item.content, *(item.thread or [])])
    ctx.db.commit()
    return _enrich(ctx, [item])[0]


@router.delete("/calendar/{item_id}")
def delete_item(item_id: uuid.UUID, ctx: Ctx = Depends(get_ctx)):
    item = _item(ctx, item_id)
    ctx.db.delete(item)
    ctx.db.commit()
    return {"ok": True}


@router.post("/calendar/{item_id}/publish")
def publish_now(item_id: uuid.UUID, ctx: Ctx = Depends(get_ctx)):
    """Publish a scheduled post now. A plain (threadpool) endpoint: its database calls and an inline publish must
    never hold up the event loop, and with it every other request."""
    item = _item(ctx, item_id)
    if item.status in ("posted", "publishing"):
        raise HTTPException(409, "This post already went out.")
    item.status = "publishing"
    item.attempts = 0
    ctx.db.commit()
    _go_now(ctx, item)
    return _enrich(ctx, [item])[0]


class UploadIn(BaseModel):
    upload_id: uuid.UUID
    duration_s: float | None = Field(None, ge=0, le=24 * 3600)  # a video's length, read by the browser


@router.post("/launchpad/uploads")
def register_upload(body: UploadIn, ctx: Ctx = Depends(get_ctx)):
    """A file the user uploaded from their computer (through /api/storage/uploads), made postable: it becomes an
    `upload` artifact, attached and posted like anything Notestack made, and offered again in the picker later."""
    upload = ctx.db.scalar(select(Upload).where(Upload.id == body.upload_id, Upload.workspace_id == ctx.workspace.id))
    if not upload:
        raise HTTPException(404, "Upload not found")
    if upload.status != "complete":
        raise HTTPException(409, "This file hasn't finished uploading.")
    kind = UPLOAD_TYPES.get(upload.content_type)
    if not kind:
        raise HTTPException(415, "LinkedIn takes JPG, PNG or GIF images and MP4 videos.")
    low, high = SIZE_LIMITS[("linkedin", kind)]
    size = upload.size_bytes or 0
    if size > high:
        raise HTTPException(413, f"This {kind} is {size / MB:.0f} MB; LinkedIn takes up to {high // MB} MB.")
    if size < low:
        raise HTTPException(400, f"This {kind} is too small for LinkedIn (at least {low // 1024} KB).")
    title = upload.filename.rsplit(".", 1)[0] if "." in upload.filename else upload.filename
    a = Artifact(workspace_id=ctx.workspace.id, type="upload", status="ready", storage_key=upload.key,
                 content_json={"title": title[:200] or "Upload", "filename": upload.filename,
                               "content_type": upload.content_type, "media": kind, "size_bytes": size,
                               **({"duration_s": body.duration_s} if body.duration_s else {})})
    ctx.db.add(a)
    ctx.db.commit()
    return artifact_brief(a)


@router.get("/launchpad/postable")
def postable(ctx: Ctx = Depends(get_ctx), q: str = "", limit: int = 60):
    """What can be attached to a post: finished videos, quote cards, carousels, summaries, launch kits, mind maps,
    slide decks, newest first, and audio overviews (listed to download: they can't be posted). Videos made but not
    rendered yet are listed too, flagged needs_render (they can only be posted once rendered), and so are videos being
    rendered, flagged rendering."""
    rows = ctx.db.scalars(select(Artifact).where(Artifact.workspace_id == ctx.workspace.id,
                                                 Artifact.type.in_(LISTED))
                          .order_by(Artifact.created_at.desc()).limit(400))
    needle = q.strip().lower()
    out = []
    for a in rows:
        # A blog2video video counts once blog2video made it (ready), rendered or not; anything else once it is ready.
        # A blog2video video mid-render stays listed (flagged rendering).
        ready = a.status == "ready" or (is_blog2video(a) and (bool((a.content_json or {}).get("video_url"))
                                                              or a.status == "rendering"))
        title = ((a.content_json or {}).get("title") or "").lower()
        if ready and (not needle or needle in title):
            out.append(artifact_brief(a))
        if len(out) >= min(limit, 200):
            break
    # What each one already has coming up (scheduled, still ahead), soonest first: the list offers to reschedule it
    upcoming = ctx.db.scalars(select(CalendarItem).where(
        CalendarItem.workspace_id == ctx.workspace.id, CalendarItem.status == "scheduled",
        CalendarItem.scheduled_at > datetime.now(UTC),
        CalendarItem.artifact_id.in_([uuid.UUID(b["id"]) for b in out])).order_by(CalendarItem.scheduled_at))
    by_artifact: dict[str, list[dict]] = {}
    for it in upcoming:
        by_artifact.setdefault(str(it.artifact_id), []).append(
            {"id": str(it.id), "scheduled_at": _aware(it.scheduled_at).isoformat(),
             "platform_label": PLATFORM_LABELS.get(it.platform, it.platform)})
    for b in out:
        b["scheduled"] = by_artifact.get(b["id"], [])
    return out


# Social accounts


def serialize_account(a: SocialAccount) -> dict:
    return {"id": str(a.id), "platform": a.platform, "platform_label": PLATFORM_LABELS[a.platform],
            "handle": a.handle, "status": a.status, "avatar_url": a.avatar_url,
            "can_post_media": can_post_media(a),
            # No refresh token and the connection ends within a week (LinkedIn's 60 day tokens).
            "expires_soon": bool(a.expires_at and not a.refresh_token
                                 and _aware(a.expires_at) - datetime.now(UTC) < timedelta(days=7)),
            "expires_at": _aware(a.expires_at).isoformat() if a.expires_at else None,
            # Down (expired, revoked, disconnected), or unable to post images and videos.
            "needs_reconnect": a.status != "active" or (a.platform in CONNECTED_ONLY and not can_post_media(a)),
            "connected_at": a.created_at.isoformat() if a.created_at else None}


@router.get("/social/accounts")
def list_accounts(ctx: Ctx = Depends(get_ctx)):
    accounts = ctx.db.scalars(select(SocialAccount).where(SocialAccount.workspace_id == ctx.workspace.id)
                              .order_by(SocialAccount.created_at)).all()
    return {
        "accounts": [serialize_account(a) for a in accounts],
        "available": {"x": x.configured(), "linkedin": linkedin.configured(), "bluesky": True},
        "redirect_uris": {"x": x.redirect_uri(), "linkedin": linkedin.redirect_uri()},
    }


@router.delete("/social/accounts/{account_id}")
def disconnect(account_id: uuid.UUID, ctx: Ctx = Depends(get_ctx)):
    account = ctx.db.scalar(select(SocialAccount).where(SocialAccount.id == account_id,
                                                        SocialAccount.workspace_id == ctx.workspace.id))
    if not account:
        raise HTTPException(404, "Account not found")
    # Kept (its posts stay linked to it), tokens gone, posts paused: reconnecting the same account resumes them.
    account.status = "revoked"
    account.access_token = ""  # no usable token kept
    account.refresh_token = None
    label = PLATFORM_LABELS[account.platform]
    pause_account_posts(ctx.db, account, f"{label} disconnected. Reconnect {label} to resume.")
    ctx.db.commit()
    return {"ok": True}


def _upsert_account(db: Session, workspace_id: uuid.UUID, platform: str, info: dict, secret: str,
                    refresh: str | None = None, expires_in: int | None = None, scopes: str | None = None) -> None:
    account = db.scalar(select(SocialAccount).where(SocialAccount.workspace_id == workspace_id,
                                                    SocialAccount.platform == platform,
                                                    SocialAccount.external_id == info["external_id"]))
    if not account:
        account = SocialAccount(workspace_id=workspace_id, platform=platform, external_id=info["external_id"],
                                handle=info["handle"], access_token="")
        db.add(account)
    account.handle = info["handle"]
    account.avatar_url = info.get("avatar_url")
    account.access_token = encrypt(secret)
    account.refresh_token = encrypt(refresh) if refresh else account.refresh_token
    account.expires_at = datetime.now(UTC) + timedelta(seconds=int(expires_in)) if expires_in else None
    account.scopes = scopes
    account.status = "active"
    resume_account_posts(db, account)  # its paused posts still ahead go back to scheduled
    db.commit()


class BlueskyIn(BaseModel):
    handle: str = Field(min_length=3, max_length=200)
    app_password: str = Field(min_length=8, max_length=100)


@router.post("/social/bluesky")
async def connect_bluesky(body: BlueskyIn, ctx: Ctx = Depends(get_ctx)):
    try:
        session = await run_in_threadpool(bluesky.create_session, body.handle, body.app_password)
    except SocialError as exc:
        raise HTTPException(400, {"code": "bluesky_auth", "message": str(exc)}) from exc
    _upsert_account(ctx.db, ctx.workspace.id, "bluesky",
                    {"external_id": session["did"], "handle": session.get("handle", body.handle)}, body.app_password)
    return list_accounts(ctx)


# OAuth (X, LinkedIn). The signed state carries the workspace and PKCE verifier; the callback hands a
# short lived encrypted ticket back to the SPA, which completes the link as the signed in user.

STATE_TTL = timedelta(minutes=10)


def _launchpad_redirect(**params: str) -> RedirectResponse:
    return RedirectResponse(f"{settings.frontend_url.rstrip('/')}/app/launchpad?{urlencode(params)}", 302)


@router.post("/social/{platform}/start")
def oauth_start(platform: Literal["x", "linkedin"], ctx: Ctx = Depends(get_ctx)):
    """Returns the provider URL. The SPA calls this with its bearer token, then navigates to `url`."""
    mod = x if platform == "x" else linkedin
    if not mod.configured():
        raise HTTPException(503, {"code": "not_configured",
                                  "message": f"Set {platform.upper()}_CLIENT_ID and {platform.upper()}_CLIENT_SECRET."})
    nonce = secrets.token_urlsafe(24)
    payload = {"typ": "social_state", "nonce": nonce, "ws": str(ctx.workspace.id), "platform": platform}
    if platform == "x":
        verifier, challenge = x.pkce_pair()
        payload["verifier"] = verifier
        state = encode_signed(payload, STATE_TTL)
        url = x.authorize_url(state, challenge)
    else:
        state = encode_signed(payload, STATE_TTL)
        url = linkedin.authorize_url(state)
    return {"url": url, "nonce": nonce}


@router.get("/social/{platform}/callback")
def oauth_callback(platform: Literal["x", "linkedin"], code: str | None = None, state: str | None = None,
                   error: str | None = None, db: Session = Depends(get_db)):
    label = PLATFORM_LABELS[platform]
    if error or not code or not state:
        return _launchpad_redirect(error=f"{label} connection was cancelled.")
    try:
        saved = decode_token_full(state, expected_type="social_state")
    except HTTPException:
        return _launchpad_redirect(error=f"{label} connection expired. Try again.")
    if saved.get("platform") != platform:
        return _launchpad_redirect(error=f"{label} connection failed.")
    try:
        if platform == "x":
            info = x.exchange_code(code, saved["verifier"])
        else:
            info = linkedin.exchange_code(code)
    except Exception:
        return _launchpad_redirect(error=f"{label} connection failed. Try again.")
    # Do not link here: this request carries no session. The signed in SPA finishes the link, and only
    # for the workspace that started it, so a forwarded authorize link cannot attach someone's account
    # to another workspace.
    ticket = encode_signed({"typ": "social_link", "ws": saved["ws"], "platform": platform,
                            "data": encrypt(json.dumps(info))}, timedelta(minutes=10))
    return _launchpad_redirect(link=ticket, platform=platform)


class LinkIn(BaseModel):
    ticket: str


@router.post("/social/complete")
def oauth_complete(body: LinkIn, ctx: Ctx = Depends(get_ctx)):
    try:
        saved = decode_token_full(body.ticket, expected_type="social_link")
    except HTTPException as exc:
        raise HTTPException(400, {"code": "expired", "message": "That connection expired. Try again."}) from exc
    if saved.get("ws") != str(ctx.workspace.id):
        raise HTTPException(403, {"code": "wrong_workspace", "message": "This connection was started elsewhere."})
    info = json.loads(decrypt(saved["data"]) or "{}")
    if not info.get("access_token"):
        raise HTTPException(400, {"code": "expired", "message": "That connection expired. Try again."})
    _upsert_account(ctx.db, ctx.workspace.id, saved["platform"], info, info["access_token"],
                    info.get("refresh_token"), info.get("expires_in"), info.get("scope"))
    return list_accounts(ctx)


# Tracked links (public)

links_router = APIRouter(tags=["links"])


@links_router.get("/l/{slug}")
def follow_link(slug: str, request: Request, db: Session = Depends(get_db)):
    link = db.scalar(select(TrackedLink).where(TrackedLink.slug == slug))
    if not link:
        raise HTTPException(404, "Link not found")
    db.add(EngagementEvent(workspace_id=link.workspace_id, tracked_link_id=link.id, kind="click",
                           metadata_json={"referer": request.headers.get("referer", "")[:300],
                                          "ua": request.headers.get("user-agent", "")[:200]}))
    db.commit()
    return RedirectResponse(link.target_url, status_code=302)