"""Scheduled task management API."""

from __future__ import annotations

import re
import zipfile
from io import BytesIO
import mimetypes
from pathlib import Path
from typing import Any
from urllib.parse import quote

from fastapi import APIRouter, HTTPException, Query, Request
from fastapi.responses import FileResponse, Response, StreamingResponse
from pydantic import BaseModel, Field

from app.gateway.deps import (
    get_checkpointer,
    get_current_user,
    get_local_provider,
    get_optional_user_from_request,
    get_run_event_store,
    get_run_store,
    get_tag_store,
)
from deerflow.runtime import serialize_channel_values
from deerflow.runtime.scheduler import DuplicateTaskNameError, get_scheduled_task_service

router = APIRouter(prefix="/api/scheduled-tasks", tags=["scheduled-tasks"])


class ScheduledTaskCreateRequest(BaseModel):
    name: str = Field(default="Scheduled task")
    description: str | None = None
    prompt: str
    schedule: str
    timezone: str | None = None
    execution_agent_name: str | None = None
    execution_context: dict[str, Any] | None = None
    require_markdown: bool = Field(
        default=False,
        description="Require the result to be Markdown and exceed the minimum length, else mark the run failed.",
    )


class ScheduledTaskUpdateRequest(BaseModel):
    name: str | None = None
    description: str | None = None
    prompt: str | None = None
    schedule: str | None = None
    timezone: str | None = None
    enabled: bool | None = None
    execution_agent_name: str | None = None
    execution_context: dict[str, Any] | None = None
    require_markdown: bool | None = None


class ScheduledTaskDeliveryProfileUpdateRequest(BaseModel):
    element_user_id: str | None = None


class ScheduledTaskSubscribeRequest(BaseModel):
    notify_element: bool = False


class ScheduledTaskRunFileUpdateRequest(BaseModel):
    content: str = Field(..., description="New text content to write into the run file")


_MAX_RUN_FILE_EDIT_BYTES = 5 * 1024 * 1024  # 5 MB


async def _user_id(request: Request) -> str:
    user_id = await get_current_user(request)
    if not user_id:
        raise HTTPException(status_code=401, detail="Authentication required")
    return user_id


async def _is_admin(request: Request) -> bool:
    """True when the caller is an admin (or auth is disabled entirely)."""
    user = await get_optional_user_from_request(request)
    if user is None:
        return True
    return getattr(user, "system_role", None) == "admin"


async def _load_run(service, request: Request, run_id: str, user_id: str) -> dict[str, Any] | None:
    """Owner/subscriber run lookup, with an admin fallback to any run.

    Lets admins open run results for tasks they do not own (the admin
    management view links into the shared run-detail page).
    """
    run = await service.get_run(user_id, run_id)
    if run is None and await _is_admin(request):
        run = await service.admin_get_run(run_id)
    return run


async def _enrich_with_creator(tasks: list[dict[str, Any]] | dict[str, Any] | None) -> Any:
    """Attach ``creator_email`` / ``creator_user_id`` to one or many task dicts.

    Looks up the auth user once per distinct ``user_id`` in the batch so we do
    not pay N+1 round trips when rendering a long task list. Tolerates missing
    user records (deleted accounts) by leaving the email blank.
    """
    if tasks is None:
        return tasks
    items = tasks if isinstance(tasks, list) else [tasks]

    user_ids = {str(t.get("user_id")) for t in items if isinstance(t, dict) and t.get("user_id")}
    if not user_ids:
        return tasks

    try:
        provider = get_local_provider()
    except Exception:
        # Auth provider not initialised (e.g. memory-mode test harness): keep
        # the bare user_id and let the frontend render the raw id.
        provider = None

    email_by_id: dict[str, str] = {}
    if provider is not None:
        for uid in user_ids:
            try:
                user = await provider.get_user(uid)
            except Exception:
                user = None
            if user is not None:
                email_by_id[uid] = user.email

    for item in items:
        if not isinstance(item, dict):
            continue
        uid = item.get("user_id")
        if uid is None:
            continue
        uid_str = str(uid)
        item["creator_user_id"] = uid_str
        item["creator_email"] = email_by_id.get(uid_str, "")
    return tasks


def _safe_filename(value: str, default: str = "scheduled-task-result") -> str:
    name = re.sub(r'[\\/:*?"<>|\r\n]+', "-", value).strip(" .")
    return name or default


def _result_download_name(result: dict[str, Any], *, extension: str = "md") -> str:
    title = result.get("title") if isinstance(result, dict) else None
    return f"{_safe_filename(str(title or 'scheduled-task-result'))}.{extension}"


def _result_markdown(result: dict[str, Any]) -> str:
    title = str(result.get("title") or "Scheduled Task Result")
    content = str(result.get("content") or "")
    return f"# {title}\n\n{content}".strip() + "\n"


def _utf8_sig_bytes(content: str) -> bytes:
    return content.encode("utf-8-sig")


def _content_disposition(filename: str) -> str:
    quoted = quote(filename)
    return f"attachment; filename*=UTF-8''{quoted}"


def _run_files(run: dict[str, Any]) -> list:
    result = run.get("result") or {}
    return result.get("files") if isinstance(result, dict) and isinstance(result.get("files"), list) else []


def _run_file_path(run: dict[str, Any], file_index: int) -> tuple[dict[str, Any], Path]:
    files = _run_files(run)
    if file_index < 0 or file_index >= len(files):
        raise HTTPException(status_code=404, detail="Result file not found")
    item = files[file_index]
    if not isinstance(item, dict) or not isinstance(item.get("path"), str):
        raise HTTPException(status_code=404, detail="Result file not found")
    path = Path(item["path"])
    if not path.exists() or not path.is_file():
        raise HTTPException(status_code=404, detail="Result file is no longer available")
    return item, path


def _is_text_previewable(path: Path, mime_type: str | None) -> bool:
    if mime_type and (mime_type.startswith("text/") or mime_type in {"application/json", "application/xml"}):
        return True
    return path.suffix.lower() in {".md", ".markdown", ".txt", ".json", ".csv", ".xml", ".yaml", ".yml", ".log"}


def _is_text_download(path: Path, mime_type: str | None) -> bool:
    return _is_text_previewable(path, mime_type)


def _read_text_download_bytes(path: Path) -> bytes:
    try:
        content = path.read_text(encoding="utf-8-sig")
    except UnicodeDecodeError:
        content = path.read_text(encoding="utf-8", errors="replace")
    return _utf8_sig_bytes(content)


@router.get("/thread")
async def get_scheduler_thread(request: Request) -> dict:
    service = get_scheduled_task_service()
    user_id = await _user_id(request)
    return await service.ensure_scheduler_thread(user_id)


@router.get("/html-templates")
async def list_html_templates() -> dict:
    """可选的 HTML 模板清单(供「模板数据转换器」内置 agent 填充用)。

    返回 ``{"templates": [{id, name, description}]}``,前端定时任务表单在选中
    「模板数据转换器」后据此渲染「HTML 模板」下拉。开放给任意登录用户。
    """
    from deerflow.runtime.scheduler import template_fill

    return {"templates": template_fill.list_templates()}


class TemplateRenderRequest(BaseModel):
    """「模板数据转换器」对话页的调试渲染入参。"""

    template_id: str
    # 智能体回复原文(含转换后的 JSON);后端用 extract_embedded_json 抽取后填充。
    text: str | None = None
    # 或直接给已解析好的对象(优先于 text)。
    data: Any | None = None


@router.post("/template-fill/render")
async def render_template_fill(body: TemplateRenderRequest) -> dict:
    """调试用:把一段(智能体回复的)内容里的 JSON 填进所选 HTML 模板并返回完整 HTML。

    供「模板数据转换器」对话页的「渲染预览」按钮调试智能体效果——不调模型,只做
    抽取 → 校验 → 填充,把校验错误/缺失字段一并返回方便定位。「只展示」模板原样返回。
    开放给任意登录用户。
    """
    from deerflow.runtime.scheduler import template_fill

    spec = template_fill.get_template(body.template_id)
    if spec is None:
        raise HTTPException(status_code=404, detail=f"未知模板：{body.template_id}")

    # 「只展示」模板(no.2 等):原样返回模板 HTML。
    if not spec.convertible:
        html = template_fill.read_template_text(body.template_id)
        return {"ok": True, "display_only": True, "errors": [], "gaps": [], "html": html}

    obj = body.data
    if obj is None and body.text:
        obj = template_fill.extract_embedded_json(body.text)
    if obj is None:
        return {"ok": False, "errors": ["未从内容中识别到可用的 JSON"], "gaps": [], "html": None}

    errors = template_fill.validate_data(body.template_id, obj)
    gaps = template_fill.field_coverage_gaps(body.template_id, obj)
    try:
        html = template_fill.fill_template(body.template_id, obj)
    except template_fill.TemplateError as exc:
        return {"ok": False, "errors": [str(exc)], "gaps": gaps, "html": None}
    return {"ok": not errors, "errors": errors, "gaps": gaps, "html": html}


@router.get("")
async def list_scheduled_tasks(
    request: Request,
    search: str | None = Query(default=None, description="Case-insensitive fuzzy match on task name"),
    tag_ids: list[str] | None = Query(default=None, description="Keep tasks carrying any of these tags (OR)"),
) -> list[dict]:
    service = get_scheduled_task_service()
    user_id = await _user_id(request)
    tasks = await service.list_tasks(user_id)
    await _enrich_with_creator(tasks)

    tag_store = get_tag_store(request)
    assignments = await tag_store.list_assignments("scheduled_task", [str(t.get("task_id")) for t in tasks])
    for task in tasks:
        task["tags"] = assignments.get(str(task.get("task_id")), [])
    if search:
        needle = search.strip().lower()
        tasks = [t for t in tasks if needle in str(t.get("name", "")).lower()]
    if tag_ids:
        wanted = set(tag_ids)
        tasks = [t for t in tasks if any(tg.get("id") in wanted for tg in t.get("tags", []))]
    return tasks


@router.get("/{task_id}/runs")
async def list_scheduled_task_runs(task_id: str, request: Request, limit: int = 50) -> list[dict]:
    service = get_scheduled_task_service()
    user_id = await _user_id(request)
    runs = await service.list_runs(user_id, task_id, limit=max(1, min(limit, 200)))
    return runs


@router.get("/runs/{run_id}")
async def get_scheduled_task_run(run_id: str, request: Request) -> dict:
    service = get_scheduled_task_service()
    user_id = await _user_id(request)
    run = await _load_run(service, request, run_id, user_id)
    if run is None:
        raise HTTPException(status_code=404, detail=f"Run {run_id} not found")
    return run


@router.get("/runs/{run_id}/html")
async def get_scheduled_task_run_html(run_id: str, request: Request) -> dict:
    """Return the generated HTML document for an ``html_page`` run."""
    service = get_scheduled_task_service()
    user_id = await _user_id(request)
    run = await _load_run(service, request, run_id, user_id)
    if run is None:
        raise HTTPException(status_code=404, detail=f"Run {run_id} not found")
    page = await service.get_html_page_for_run(run_id)
    if page is None:
        raise HTTPException(status_code=404, detail="该执行记录没有 HTML 页面")
    return {
        "run_id": run_id,
        "task_id": page.get("task_id") or run.get("task_id"),
        "title": page.get("title"),
        "html": page.get("html_content"),
        "style_summary": page.get("style_summary"),
        "layout_summary": page.get("layout_summary"),
        "page_id": page.get("id"),
        "created_at": page.get("created_at"),
    }


def _event_type_for_checkpoint_message(message: dict) -> str:
    """Map a serialized LangChain message to the same ``event_type`` strings
    the run-event store uses, so the frontend's existing role labels keep
    working for checkpoint-derived synthetic rows."""
    raw_type = message.get("type") or message.get("role") or ""
    if raw_type in {"human", "user"}:
        return "llm.human.input"
    if raw_type in {"ai", "assistant"}:
        return "llm.ai.response"
    if raw_type == "tool":
        return "llm.tool.result"
    if raw_type == "system":
        return "llm.system"
    return f"llm.{raw_type or 'message'}"


async def _checkpoint_messages_for_run(
    request: Request, agent_run: dict, *, limit: int
) -> list[dict]:
    """Return a synthetic list of message rows pulled from the LangGraph
    checkpoint for an in-flight run.

    The event store is updated by the streaming pipeline and may stay empty
    until the first message tuple is flushed — meanwhile the checkpoint
    already carries the current state (input + accumulating AI deltas). We
    project that into the same row shape the frontend uses for the event
    store, so the run-detail page can show progress immediately."""
    try:
        checkpointer = get_checkpointer(request)
    except Exception:
        return []
    thread_id = agent_run.get("thread_id")
    if not thread_id:
        return []
    config = {"configurable": {"thread_id": thread_id, "checkpoint_ns": ""}}
    try:
        tup = await checkpointer.aget_tuple(config)
    except Exception:
        return []
    if tup is None:
        return []
    checkpoint = getattr(tup, "checkpoint", {}) or {}
    channel_values = checkpoint.get("channel_values") or {}
    try:
        values = serialize_channel_values(channel_values)
    except Exception:
        return []
    messages = values.get("messages") or []

    rows: list[dict] = []
    for index, message in enumerate(messages[-limit:]):
        if not isinstance(message, dict):
            continue
        rows.append(
            {
                "seq": index + 1,
                "event_type": _event_type_for_checkpoint_message(message),
                "category": "message",
                "content": message,
                "metadata": {"source": "checkpoint"},
                "created_at": None,
            }
        )
    return rows


@router.get("/runs/{run_id}/messages")
async def get_scheduled_task_run_messages(run_id: str, request: Request, limit: int = 500) -> dict:
    """Agent messages produced by a scheduled run's underlying agent run.

    Resolution order (most authoritative first):

    1. Event-store rows for the underlying agent run — these are the per-tuple
       streaming events the worker writes as the agent runs.
    2. LangGraph checkpoint snapshot — the canonical conversation state. Used
       as a fallback when the event store is still empty (early in a run) or
       when only partial events have been flushed; we project the snapshot
       into the same row shape the frontend already renders so the user can
       see the prompt + current AI output while the task is mid-flight.
    3. Synthetic human-input row built from ``task.prompt`` — shown even
       before the agent run is registered, so a `pending` task at minimum
       displays what was queued.
    """
    service = get_scheduled_task_service()
    user_id = await _user_id(request)
    run = await _load_run(service, request, run_id, user_id)
    if run is None:
        raise HTTPException(status_code=404, detail=f"Run {run_id} not found")
    capped = max(1, min(limit, 1000))

    # Always available: task prompt as the seeded human input. Lets the
    # ``execution process`` tab show *something* even before the agent run is
    # registered, which is the user-visible bug we're fixing.
    fallback_rows: list[dict] = []
    task_id = run.get("task_id")
    if task_id:
        task = await service.store.get_task_any(task_id)
        prompt = (task or {}).get("prompt")
        if prompt:
            fallback_rows.append(
                {
                    "seq": 0,
                    "event_type": "llm.human.input",
                    "category": "message",
                    # Frontend treats human-input content as a plain string.
                    "content": str(prompt),
                    "metadata": {"source": "task_prompt"},
                    "created_at": run.get("created_at") or run.get("scheduled_for"),
                }
            )

    agent_run_id = run.get("agent_run_id")
    if not agent_run_id:
        return {"messages": fallback_rows, "has_more": False}

    # Each scheduled run executes on its own throwaway runtime thread — NOT the
    # run row's ``scheduler_thread_id`` — so resolve the real thread from the
    # run store before querying its message events.
    run_store = get_run_store(request)
    agent_run = await run_store.get(agent_run_id, user_id=None)
    if agent_run is None:
        return {"messages": fallback_rows, "has_more": False}

    event_store = get_run_event_store(request)
    rows = await event_store.list_messages_by_run(
        agent_run["thread_id"],
        agent_run_id,
        limit=capped + 1,
        user_id=None,
    )

    # Event store has no rows yet (run just started, or events still buffering).
    # Fall back to the live checkpoint so the user immediately sees the prompt
    # + whatever AI output has been streamed so far.
    if not rows:
        checkpoint_rows = await _checkpoint_messages_for_run(request, agent_run, limit=capped)
        if checkpoint_rows:
            return {"messages": checkpoint_rows[:capped], "has_more": False}
        return {"messages": fallback_rows, "has_more": False}

    has_more = len(rows) > capped
    return {"messages": rows[:capped] if has_more else rows, "has_more": has_more}


@router.delete("/runs/{run_id}")
async def delete_scheduled_task_run(run_id: str, request: Request) -> dict:
    service = get_scheduled_task_service()
    user_id = await _user_id(request)
    try:
        deleted = await service.delete_run(user_id, run_id)
    except PermissionError as exc:
        # Owner-only for regular users; admins may still delete any run.
        if await _is_admin(request):
            deleted = await service.admin_delete_run(run_id)
        else:
            raise HTTPException(status_code=403, detail=str(exc))
    # A run the caller does not own returns deleted=False — fall back to an
    # admin-scoped delete so admins can clean up other users' runs.
    if not deleted and await _is_admin(request):
        deleted = await service.admin_delete_run(run_id)
    if not deleted:
        raise HTTPException(status_code=404, detail=f"Run {run_id} not found")
    return {"success": True}


@router.get("/runs/{run_id}/files/{file_index}")
async def get_scheduled_task_run_file(run_id: str, file_index: int, request: Request) -> dict:
    service = get_scheduled_task_service()
    user_id = await _user_id(request)
    run = await _load_run(service, request, run_id, user_id)
    if run is None:
        raise HTTPException(status_code=404, detail=f"Run {run_id} not found")
    item, path = _run_file_path(run, file_index)
    mime_type = item.get("mime_type") or mimetypes.guess_type(path.name)[0]
    previewable = _is_text_previewable(path, mime_type)
    content = None
    if previewable:
        try:
            content = path.read_text(encoding="utf-8")
        except UnicodeDecodeError:
            content = path.read_text(encoding="utf-8", errors="replace")
    return {
        "index": file_index,
        "name": item.get("name") or path.name,
        "mime_type": mime_type,
        "kind": item.get("kind"),
        "size": path.stat().st_size,
        "previewable": previewable,
        "content": content,
    }


@router.put("/runs/{run_id}/files/{file_index}")
async def update_scheduled_task_run_file(
    run_id: str,
    file_index: int,
    body: ScheduledTaskRunFileUpdateRequest,
    request: Request,
) -> dict:
    if len(body.content.encode("utf-8")) > _MAX_RUN_FILE_EDIT_BYTES:
        raise HTTPException(status_code=413, detail="Edited content exceeds the 5 MB limit")
    service = get_scheduled_task_service()
    user_id = await _user_id(request)
    try:
        return await service.update_run_file(user_id, run_id, file_index, body.content)
    except PermissionError as exc:
        raise HTTPException(status_code=403, detail=str(exc))
    except FileNotFoundError as exc:
        raise HTTPException(status_code=404, detail=str(exc))
    except ValueError as exc:
        raise HTTPException(status_code=400, detail=str(exc))


@router.get("/runs/{run_id}/download")
async def download_scheduled_task_run(run_id: str, request: Request):
    service = get_scheduled_task_service()
    user_id = await _user_id(request)
    run = await _load_run(service, request, run_id, user_id)
    if run is None:
        raise HTTPException(status_code=404, detail=f"Run {run_id} not found")

    result = run.get("result") or {}
    if not isinstance(result, dict):
        result = {}
    files = result.get("files") if isinstance(result.get("files"), list) else []
    accessible_files: list[tuple[str, Path]] = []
    for item in files:
        if not isinstance(item, dict):
            continue
        path_value = item.get("path")
        if not isinstance(path_value, str) or not path_value:
            continue
        path = Path(path_value)
        if path.exists() and path.is_file():
            accessible_files.append((str(item.get("name") or path.name), path))

    if not accessible_files:
        body = _utf8_sig_bytes(_result_markdown(result))
        return Response(
            content=body,
            media_type="text/markdown; charset=utf-8",
            headers={"Content-Disposition": _content_disposition(_result_download_name(result))},
        )

    archive = BytesIO()
    with zipfile.ZipFile(archive, mode="w", compression=zipfile.ZIP_DEFLATED) as zf:
        zf.writestr("result.md", _utf8_sig_bytes(_result_markdown(result)))
        for name, path in accessible_files:
            arcname = _safe_filename(name, path.name)
            mime_type = mimetypes.guess_type(path.name)[0]
            if _is_text_download(path, mime_type):
                zf.writestr(arcname, _read_text_download_bytes(path))
            else:
                zf.write(path, arcname=arcname)
    archive.seek(0)
    return StreamingResponse(
        archive,
        media_type="application/zip",
        headers={"Content-Disposition": _content_disposition(_result_download_name(result, extension="zip"))},
    )


@router.get("/runs/{run_id}/files/{file_index}/download")
async def download_scheduled_task_run_file(run_id: str, file_index: int, request: Request):
    service = get_scheduled_task_service()
    user_id = await _user_id(request)
    run = await _load_run(service, request, run_id, user_id)
    if run is None:
        raise HTTPException(status_code=404, detail=f"Run {run_id} not found")
    item, path = _run_file_path(run, file_index)
    filename = _safe_filename(str(item.get("name") or path.name), path.name)
    mime_type = item.get("mime_type") or mimetypes.guess_type(path.name)[0]
    if _is_text_download(path, mime_type):
        return Response(
            content=_read_text_download_bytes(path),
            media_type=f"{mime_type or 'text/plain'}; charset=utf-8",
            headers={"Content-Disposition": _content_disposition(filename)},
        )
    return FileResponse(path, filename=filename, media_type=item.get("mime_type") or "application/octet-stream")


@router.get("/delivery-profile")
async def get_delivery_profile(request: Request) -> dict:
    service = get_scheduled_task_service()
    user_id = await _user_id(request)
    return await service.get_delivery_profile(user_id)


@router.put("/delivery-profile")
async def update_delivery_profile(body: ScheduledTaskDeliveryProfileUpdateRequest, request: Request) -> dict:
    service = get_scheduled_task_service()
    user_id = await _user_id(request)
    return await service.update_delivery_profile(user_id, body.model_dump(exclude_unset=True))


@router.post("")
async def create_scheduled_task(body: ScheduledTaskCreateRequest, request: Request) -> dict:
    service = get_scheduled_task_service()
    user_id = await _user_id(request)
    try:
        task = await service.create_task(
            user_id=user_id,
            name=body.name,
            description=body.description,
            prompt=body.prompt,
            schedule=body.schedule,
            timezone=body.timezone,
            execution_agent_name=body.execution_agent_name,
            execution_context=body.execution_context,
            require_markdown=body.require_markdown,
        )
    except DuplicateTaskNameError:
        raise HTTPException(status_code=409, detail="已存在同名定时任务,请换一个名称")
    await _enrich_with_creator(task)
    return task


@router.patch("/{task_id}")
async def update_scheduled_task(task_id: str, body: ScheduledTaskUpdateRequest, request: Request) -> dict:
    service = get_scheduled_task_service()
    user_id = await _user_id(request)
    try:
        task = await service.update_task(user_id, task_id, body.model_dump(exclude_unset=True))
    except DuplicateTaskNameError:
        raise HTTPException(status_code=409, detail="已存在同名定时任务,请换一个名称")
    if task is None:
        raise HTTPException(status_code=404, detail=f"Task {task_id} not found")
    await _enrich_with_creator(task)
    return task


@router.delete("/{task_id}")
async def delete_scheduled_task(task_id: str, request: Request) -> dict:
    service = get_scheduled_task_service()
    user_id = await _user_id(request)
    deleted = await service.delete_task(user_id, task_id)
    if not deleted:
        raise HTTPException(status_code=404, detail=f"Task {task_id} not found")
    try:
        await get_tag_store(request).unassign_all("scheduled_task", task_id)
    except Exception:  # noqa: BLE001 — tag cleanup must not block deletion
        pass
    return {"success": True}


@router.post("/{task_id}/pause")
async def pause_scheduled_task(task_id: str, request: Request) -> dict:
    service = get_scheduled_task_service()
    user_id = await _user_id(request)
    task = await service.pause_task(user_id, task_id)
    if task is None:
        raise HTTPException(status_code=404, detail=f"Task {task_id} not found")
    return task


@router.post("/{task_id}/resume")
async def resume_scheduled_task(task_id: str, request: Request) -> dict:
    service = get_scheduled_task_service()
    user_id = await _user_id(request)
    task = await service.resume_task(user_id, task_id)
    if task is None:
        raise HTTPException(status_code=404, detail=f"Task {task_id} not found")
    return task


@router.post("/{task_id}/trigger")
async def trigger_scheduled_task(task_id: str, request: Request) -> dict:
    service = get_scheduled_task_service()
    user_id = await _user_id(request)
    task = await service.trigger_task(user_id, task_id)
    if task is None:
        raise HTTPException(status_code=404, detail=f"Task {task_id} not found")
    return {"success": True, "task": task}


# --- Publish / subscribe ---


@router.get("/published")
async def list_published_scheduled_tasks(request: Request, limit: int = 50, offset: int = 0) -> list[dict]:
    service = get_scheduled_task_service()
    await _user_id(request)
    tasks = await service.list_published(limit=max(1, min(limit, 200)), offset=max(0, offset))
    # published list uses owner_user_id; resolve it into a creator_email here too
    # so the discovery view can show the author's account.
    for task in tasks:
        if isinstance(task, dict) and task.get("owner_user_id"):
            task["user_id"] = task["owner_user_id"]
    await _enrich_with_creator(tasks)
    for task in tasks:
        if isinstance(task, dict):
            task.pop("user_id", None)
    return tasks


@router.post("/{task_id}/publish")
async def publish_scheduled_task(task_id: str, request: Request) -> dict:
    service = get_scheduled_task_service()
    user_id = await _user_id(request)
    task = await service.publish_task(user_id, task_id)
    if task is None:
        raise HTTPException(status_code=404, detail=f"Task {task_id} not found")
    return task


@router.post("/{task_id}/unpublish")
async def unpublish_scheduled_task(task_id: str, request: Request) -> dict:
    service = get_scheduled_task_service()
    user_id = await _user_id(request)
    task = await service.unpublish_task(user_id, task_id)
    if task is None:
        raise HTTPException(status_code=404, detail=f"Task {task_id} not found")
    return task


@router.post("/{task_id}/subscribe")
async def subscribe_scheduled_task(task_id: str, body: ScheduledTaskSubscribeRequest, request: Request) -> dict:
    service = get_scheduled_task_service()
    user_id = await _user_id(request)
    sub = await service.subscribe_task(user_id, task_id, notify_element=body.notify_element)
    if sub is None:
        raise HTTPException(status_code=404, detail="Task is not available for subscription")
    return sub


@router.patch("/{task_id}/subscribe")
async def update_scheduled_task_subscription(task_id: str, body: ScheduledTaskSubscribeRequest, request: Request) -> dict:
    service = get_scheduled_task_service()
    user_id = await _user_id(request)
    sub = await service.subscribe_task(user_id, task_id, notify_element=body.notify_element)
    if sub is None:
        raise HTTPException(status_code=404, detail="Subscription not found")
    return sub


@router.delete("/{task_id}/subscribe")
async def delete_scheduled_task_subscription(task_id: str, request: Request) -> dict:
    service = get_scheduled_task_service()
    user_id = await _user_id(request)
    deleted = await service.unsubscribe_task(user_id, task_id)
    if not deleted:
        raise HTTPException(status_code=404, detail="Subscription not found")
    return {"success": True}


# --- Admin (cross-user management) ---


async def _require_admin(request: Request) -> None:
    """Reject non-admins. Lets the call through when auth is disabled."""
    user = await get_optional_user_from_request(request)
    if user is None:
        return
    if getattr(user, "system_role", None) != "admin":
        raise HTTPException(status_code=403, detail="定时任务管理仅限管理员")


@router.get("/admin/all")
async def admin_list_scheduled_tasks(
    request: Request,
    search: str | None = Query(default=None, description="Case-insensitive fuzzy match on task name"),
) -> list[dict]:
    """List every scheduled task across all users. Admin-only."""
    await _require_admin(request)
    service = get_scheduled_task_service()
    tasks = await service.admin_list_tasks()
    await _enrich_with_creator(tasks)

    tag_store = get_tag_store(request)
    assignments = await tag_store.list_assignments("scheduled_task", [str(t.get("task_id")) for t in tasks])
    for task in tasks:
        task["tags"] = assignments.get(str(task.get("task_id")), [])
    if search:
        needle = search.strip().lower()
        tasks = [t for t in tasks if needle in str(t.get("name", "")).lower()]
    return tasks


@router.get("/admin/{task_id}/runs")
async def admin_list_scheduled_task_runs(task_id: str, request: Request, limit: int = 50) -> list[dict]:
    """List runs for any task regardless of owner. Admin-only."""
    await _require_admin(request)
    service = get_scheduled_task_service()
    return await service.admin_list_runs(task_id, limit=max(1, min(limit, 200)))


@router.patch("/admin/{task_id}")
async def admin_update_scheduled_task(task_id: str, body: ScheduledTaskUpdateRequest, request: Request) -> dict:
    """Update any task regardless of owner. Admin-only."""
    await _require_admin(request)
    service = get_scheduled_task_service()
    try:
        task = await service.admin_update_task(task_id, body.model_dump(exclude_unset=True))
    except DuplicateTaskNameError:
        raise HTTPException(status_code=409, detail="已存在同名定时任务,请换一个名称")
    if task is None:
        raise HTTPException(status_code=404, detail=f"Task {task_id} not found")
    await _enrich_with_creator(task)
    return task


@router.delete("/admin/{task_id}")
async def admin_delete_scheduled_task(task_id: str, request: Request) -> dict:
    """Delete any task regardless of owner. Admin-only."""
    await _require_admin(request)
    service = get_scheduled_task_service()
    deleted = await service.admin_delete_task(task_id)
    if not deleted:
        raise HTTPException(status_code=404, detail=f"Task {task_id} not found")
    try:
        await get_tag_store(request).unassign_all("scheduled_task", task_id)
    except Exception:  # noqa: BLE001 — tag cleanup must not block deletion
        pass
    return {"success": True}


@router.post("/admin/{task_id}/trigger")
async def admin_trigger_scheduled_task(task_id: str, request: Request) -> dict:
    """Trigger an immediate run of any task regardless of owner. Admin-only."""
    await _require_admin(request)
    service = get_scheduled_task_service()
    task = await service.admin_trigger_task(task_id)
    if task is None:
        raise HTTPException(status_code=404, detail=f"Task {task_id} not found")
    return {"success": True, "task": task}