"""Background scheduler: free RSS detection, enrichment of new videos, and quota-aware
recent-first / deep backfill. Runs inside the API process (single worker)."""

import contextvars
import threading
from datetime import timezone
from typing import Callable

from apscheduler.executors.pool import ThreadPoolExecutor
from apscheduler.jobstores.base import ConflictingIdError
from apscheduler.schedulers.background import BackgroundScheduler
from sqlalchemy import select

from app import audit, debugsession, jobruns, logs, logsink, progress, quota, sysconfig
from app.config import settings
from app.db import SessionLocal
from app.downloads.gc import run_download_gc
from app.models import Channel
from app.plex.sync import sync as run_plex_sync
from app.plex.tmdb import run_enrich as run_tmdb_enrich
from app.plex.tmdb import run_people_enrich
from app.plex.watch_sync import run_plex_watch_reconcile, run_plex_watch_sync
from app.notifications import create_notification
from app.runstatus import RunStatus, RunTrigger
from app.state import is_sync_paused
from app.sync.autotag import run_autotag_all
from app.sync.explore import purge_ephemeral
from app.sync.maintenance import run_maintenance
from app.sync.playlists import sync_all_playlists
from app.utils import iso
from app.sync.runner import (
    run_deep_backfill,
    run_enrich,
    run_recent_backfill,
    run_rss_poll,
    run_shorts,
    run_subscription_resync,
)

log = logs.get_logger(__name__)
_scheduler: BackgroundScheduler | None = None

# What is running RIGHT NOW, per job: the single-run claim and the live progress. In memory on
# purpose — the scheduler lives in the API's one process, and progress changes every second. What a
# run did is durable instead: its `job_runs` row (app/jobruns.py), which outlives a restart.
_activity: dict[str, dict] = {}
_activity_lock = threading.Lock()

# Set (per-thread) by a manual "run now" trigger to the id of the admin who clicked, so that
# run — and only that run, not the recurring scheduled ones — posts a completion notification
# to their inbox. Unset (None) for scheduled runs.
_manual_actor: contextvars.ContextVar[int | None] = contextvars.ContextVar(
    "manual_actor", default=None
)

# Each interval job's env/config default period (minutes). The admin can override any of these at
# runtime from the Scheduler page — a sysconfig key per job (`sysconfig.interval_key`), applied
# live; see load_intervals.
JOB_INTERVALS: dict[str, int] = {
    job: getattr(settings, attr) for job, attr in sysconfig.JOB_INTERVAL_DEFAULTS.items()
}
MIN_INTERVAL = sysconfig.MIN_INTERVAL
MAX_INTERVAL = sysconfig.MAX_INTERVAL


def load_intervals(db) -> dict[str, int]:
    """Effective per-job intervals: each job's sysconfig override, else its env/config default."""
    return {job: sysconfig.get_int(db, sysconfig.interval_key(job)) for job in JOB_INTERVALS}


def stage_interval(db, job_id: str, minutes: int) -> int:
    """Stage a job's interval override (clamped) in `db`; returns the value. The caller commits it
    with its audit row, then calls `reschedule` — the live job follows only what was stored."""
    if job_id not in JOB_INTERVALS:
        raise KeyError(job_id)
    minutes = max(MIN_INTERVAL, min(MAX_INTERVAL, int(minutes)))
    sysconfig.set_value(db, sysconfig.interval_key(job_id), minutes)
    return minutes


def reschedule(job_id: str, minutes: int) -> None:
    """Move the live interval job to `minutes` (if this process registered it)."""
    if _scheduler is not None and _scheduler.get_job(job_id) is not None:
        _scheduler.reschedule_job(job_id, trigger="interval", minutes=minutes)
        log.info("rescheduled job %s to every %s min", job_id, minutes)


def _record(name: str, **fields) -> None:
    with _activity_lock:
        _activity.setdefault(name, {}).update(fields)


def _notify_done(
    db,
    actor_id: int,
    job_id: str,
    status: RunStatus,
    summary: str | None,
    result: dict | None = None,
) -> None:
    """Post a completion notice to the triggering admin's inbox (manual runs only). Best
    effort — a notification failure must never mask the job's own outcome."""
    try:
        create_notification(
            db,
            user_id=actor_id,
            type="scheduler",
            title=f"{job_id}: {status}",  # English fallback; UI renders translated from data
            body=summary,
            data={"job_id": job_id, "status": status, "summary": summary, "result": result},
        )
    except Exception:
        db.rollback()
        log.exception("failed to write completion notification for job %s", job_id)


def _job(name: str, fn) -> None:
    # The one place both run paths converge (manual trigger AND scheduled fire), so claim the run
    # HERE, atomically: whichever reaches this first wins, the other skips. This is the real
    # single-run guard — trigger_job's pre-check is only a fast best-effort answer for the UI.
    if not _claim_running(name):
        log.info("job %s already running — skipping this trigger", name)
        return
    actor_id = _manual_actor.get()  # set only for a manual "run now" trigger
    trigger = RunTrigger.SCHEDULED if actor_id is None else RunTrigger.MANUAL
    _record(name, progress=None)  # running already claimed
    db = SessionLocal()
    try:
        try:
            run_id = jobruns.start(name, trigger, actor_id)
        except Exception:
            # No row to tell it in: the database is what failed. The log is the record.
            log.exception("job %s not started: its run could not be recorded", name)
            return
        with logs.bind(run_id=run_id):
            _run(db, name, fn, run_id, actor_id)
    finally:
        # ALWAYS release the claim, even if the body raised before its own status update — so a
        # failure can never wedge the job as permanently "running".
        _record(name, running=False, progress=None)
        db.close()


def _run(db, name: str, fn, run_id: int, actor_id: int | None) -> None:
    """The run itself, inside its `job_runs` row: the pause-skip, the work, and its outcome — on
    the row, and in the triggering admin's inbox for a manual run."""

    def sink(current, total, phase):
        _record(name, progress={"current": current, "total": total, "phase": phase})

    # Only the work is inside the try: writing its outcome is not the job, and a write that fails
    # must not turn a run that succeeded into a failed one (see _close).
    result = None
    try:
        paused = is_sync_paused(db)
        if not paused:
            with progress.bind(sink):
                result = fn(db)
    except Exception as exc:
        db.rollback()
        log.exception("job %s failed", name, extra=logs.event("scheduler.job.failed", job=name))
        err, failure = _failure(exc)
        _close(run_id, RunStatus.ERROR, error=err, failure=failure)
        if actor_id is not None:
            _notify_done(db, actor_id, name, RunStatus.ERROR, err, failure)
        return
    if paused:
        log.info("job %s skipped (sync paused)", name)
        _close(run_id, RunStatus.SKIPPED, summary="sync paused", result=_PAUSED)
        if actor_id is not None:
            _notify_done(db, actor_id, name, RunStatus.SKIPPED, "sync paused", _PAUSED)
        return
    summary, data = _finish_ok(name, run_id, result)
    if actor_id is not None:
        _notify_done(db, actor_id, name, RunStatus.OK, summary, data)


def _close(run_id: int, status: RunStatus, **outcome) -> None:
    """Write a run's outcome. Its own failure (the database) is the log's to tell, never the job's:
    the row then stays `running` until the next start's sweep closes it as interrupted."""
    try:
        jobruns.finish(run_id, status, **outcome)
    except Exception:
        log.exception("run %s: its outcome (%s) could not be recorded", run_id, status)


def _finish_ok(name: str, run_id: int, result) -> tuple[str | None, dict | None]:
    """Close a run that returned: its result on the row, as data and as the English one-liner."""
    summary = _summarize(result)
    data = _result_data(result)
    log.info("job %s -> %s", name, result)
    _close(run_id, RunStatus.OK, summary=summary, result=data)
    return summary, data


def _failure(exc: Exception) -> tuple[str, dict | None]:
    """A raised run's English error line, and the failure the job named (progress.JobFailed) as
    data, as a result is, so the client says it in the reader's language."""
    err = str(exc) or exc.__class__.__name__
    failure = {"failure": exc.reason, **exc.params} if isinstance(exc, progress.JobFailed) else None
    return err, failure


# A skipped run's result, in the same shape as a job's own `{"skipped": …}`.
_PAUSED = {"skipped": "sync paused"}


def _result_data(result) -> dict | None:
    """A job's result as the client names it in the reader's language: the dict itself (counts,
    a `skipped` / `reason` word, a nested dict per phase), `enrich`'s bare count as one. The one-line
    `_summarize` is the English fallback."""
    if isinstance(result, dict):
        return result or None
    if isinstance(result, int) and not isinstance(result, bool):
        return {"enriched": result}
    return None


def _summarize(result) -> str | None:
    """Compact one-line rendering of a job's return value for the dashboard."""
    if result is None:
        return None
    if isinstance(result, dict):
        return ", ".join(f"{k}={v}" for k, v in result.items()) or None
    return str(result)


def scheduler_snapshot(db) -> dict:
    """What the scheduler is doing, for the admin dashboard: each job's live state from this process
    and its last runs from `job_runs`, so a restart no longer blanks "when did it last run, and how
    did it go"; and the same for the ad-hoc runs users start on one target. `running_here` is False where the scheduler runs no interval jobs
    (SCHEDULER_ENABLED=false), so the UI can say so while still showing the DB-derived history,
    queue and quota."""
    running_here = _scheduler is not None and settings.scheduler_enabled
    next_runs: dict[str, str | None] = {}
    live_intervals: dict[str, int] = {}
    if _scheduler is not None:
        for job in _scheduler.get_jobs():
            nr = getattr(job, "next_run_time", None)
            next_runs[job.id] = nr.astimezone(timezone.utc).isoformat() if nr else None
            iv = getattr(job.trigger, "interval", None)
            if iv is not None:
                live_intervals[job.id] = round(iv.total_seconds() / 60)
    with _activity_lock:
        acts = {k: dict(v) for k, v in _activity.items()}
    recent = jobruns.recent_runs(db, [*JOB_INTERVALS, *AD_HOC_JOBS], RECENT_RUNS)
    jobs = []
    for job_id, default_interval in JOB_INTERVALS.items():
        a = acts.get(job_id, {})
        jobs.append(
            {
                "id": job_id,
                "interval_minutes": live_intervals.get(job_id, default_interval),
                "next_run": next_runs.get(job_id),
                "running": a.get("running", False),
                "progress": a.get("progress"),
                **_last_runs(recent.get(job_id, [])),
            }
        )
    # Work a user starts on one target: no interval, no "run now" (it needs its target), but its
    # runs are shown beside the jobs' — it spends the same quota. Its one in-memory guard is the
    # target's own (sync/runner), so "running" is what its rows say: two channels can backfill at
    # once, so a run still going counts even when a newer one already ended, and the row names it.
    backfills = recent.get(CHANNEL_BACKFILL, [])
    going = jobruns.running_one(db, CHANNEL_BACKFILL)
    shown = going or (backfills[0] if backfills else None)
    ad_hoc = [
        {
            "id": CHANNEL_BACKFILL,
            "running": going is not None,
            "target": shown.target if shown else None,
            "target_title": (
                db.scalar(select(Channel.title).where(Channel.id == shown.target))
                if shown is not None and shown.target
                else None
            ),
            **_last_runs(backfills),
        }
    ]
    return {
        "running_here": running_here,
        "enabled": settings.scheduler_enabled,
        "jobs": jobs,
        "ad_hoc": ad_hoc,
    }


# How many past runs a Scheduler row shows as dots; the full history is the run list.
RECENT_RUNS = 3


def _last_runs(runs: list) -> dict:
    """A row's run fields from its newest runs (newest first): how the last one went, and the few
    before it for the history dots."""
    run = runs[0] if runs else None
    # A run still going has no outcome yet: only an ended run says how it went.
    ended = run if run is not None and run.status != RunStatus.RUNNING else None
    return {
        "status": ended.status if ended else None,
        "last_started": iso(run.started_at) if run else None,
        "last_finished": iso(run.finished_at) if run else None,
        "last_result": ended.summary if ended else None,
        # The outcome as data: the result, or on a failed run the failure the job named.
        "result": _outcome_data(ended),
        "last_error": ended.error if ended else None,
        "last_trigger": run.trigger if run else None,
        "last_run_id": run.id if run else None,
        "quota_units": ended.quota_units if ended else None,
        "recent": [_run_brief(r) for r in runs],
    }


def _run_brief(run) -> dict:
    return {
        "id": run.id,
        "status": run.status,
        "trigger": run.trigger,
        "started_at": iso(run.started_at),
        "finished_at": iso(run.finished_at),
    }


def _outcome_data(run) -> dict | None:
    if run is None:
        return None
    return run.failure if run.status == RunStatus.ERROR else run.result


def _rss_job() -> None:
    _job("rss_poll", run_rss_poll)


def _enrich_job() -> None:
    def work(db):
        with quota.attribute(None, quota.QuotaAction.VIDEOS_ENRICH):
            return run_enrich(db)

    _job("enrich", work)


def _backfill_job() -> None:
    # Recent-first for not-yet-synced channels, then deep backfill for the rest. All
    # background spend is attributed to the system (no actor), split by action.
    def work(db):
        with quota.attribute(None, quota.QuotaAction.VIDEOS_BACKFILL_RECENT):
            recent = run_recent_backfill(db, max_channels=25)
        with quota.attribute(None, quota.QuotaAction.VIDEOS_BACKFILL_FULL):
            deep = run_deep_backfill(db, max_channels=10)
        return {"recent": recent, "deep": deep}

    _job("backfill", work)


def _autotag_job() -> None:
    _job("autotag", lambda db: run_autotag_all(db, only_missing=True))


def _shorts_job() -> None:
    _job("shorts", run_shorts)


def _subscriptions_job() -> None:
    def work(db):
        with quota.attribute(None, quota.QuotaAction.SUBSCRIPTIONS_RESYNC):
            return run_subscription_resync(db)

    _job("subscriptions", work)


def _playlist_sync_job() -> None:
    # Mirror each read-scope user's YouTube playlists; per-user quota attribution is
    # handled inside sync_all_playlists.
    _job("playlist_sync", sync_all_playlists)


def _maintenance_job() -> None:
    def work(db):
        with quota.attribute(None, quota.QuotaAction.MAINTENANCE_REVALIDATE):
            return run_maintenance(db)

    _job("maintenance", work)


def _explore_cleanup_job() -> None:
    # Reclaim un-kept ephemeral discovery content (explored channels + live-search results) after
    # their grace periods. 0 quota (pure DB), so no quota attribution.
    _job("explore_cleanup", purge_ephemeral)


def _download_gc_job() -> None:
    # Retention GC for the download center: pre-expiry warnings, TTL deletion, LRU eviction — and
    # the week-ahead warning before an API token runs out (`apitokens.warn_expiring`).
    # Pure disk/DB work (no YouTube quota).
    _job("download_gc", run_download_gc)


def _plex_sync_job() -> None:
    # Mirror the enabled Plex library sections into the local catalog. Pure metadata (no YouTube
    # quota); a no-op when the Plex module is disabled.
    _job("plex_sync", run_plex_sync)


def _plex_watch_sync_job() -> None:
    # Incremental Plex→Siftlode watch-state pull (history + on-deck) + re-push of unsynced local
    # states. Cheap; a no-op when Plex is disabled or no owner link has sync enabled.
    _job("plex_watch_sync", run_plex_watch_sync)


def _plex_watch_reconcile_job() -> None:
    # Full watch-state reconcile (whole-section rescan) — catches un-watches the incremental feed
    # can't. Heavier, so it runs far less often (daily default).
    _job("plex_watch_reconcile", run_plex_watch_reconcile)


def _tmdb_enrich_job() -> None:
    # Enrich a batch of mirrored Plex titles from TMDB (keywords/crew/budget/revenue/TMDb score/
    # recommendations/backdrops), matched by guid. External metadata (no YouTube quota); a no-op when
    # the TMDB key is unset or the Plex catalog is empty.
    _job("tmdb_enrich", run_tmdb_enrich)


def _people_enrich_job() -> None:
    # Make the people local: a batch of person bios from TMDB and a batch of portraits resized
    # through Plex, so a person page and a cast strip never wait on an outside service.
    _job("people_enrich", run_people_enrich)


def _audit_gc_job() -> None:
    # The history cleanup: audit-log rows past audit_retention_days, YouTube quota history past
    # quota_retention_days and finished runs past job_run_retention_days (0 = keep forever, each).
    # Pure DB, no quota.
    def work(db):

        reaped = audit.gc(db, sysconfig.get_int(db, "audit_retention_days"))
        quota_reaped = quota.gc(db, sysconfig.get_int(db, "quota_retention_days"))
        runs_reaped = jobruns.gc(db, sysconfig.get_int(db, "job_run_retention_days"))
        db.commit()
        return {"reaped": reaped, "quota_reaped": quota_reaped, "runs_reaped": runs_reaped}

    _job("audit_gc", work)


def _log_gc_job() -> None:
    # Prune log_events rows older than log_retention_days — the limit the privacy policy states, so
    # this runs hourly rather than daily. Pure DB, no quota.
    def work(db):

        days = sysconfig.get_int(db, "log_retention_days")
        reaped = logsink.gc(db, days)
        # Debug sessions go with the rows they bundle; an open one past its cap is closed.
        closed = debugsession.sweep(db)
        sessions = debugsession.gc(db, days)
        db.commit()
        return {"reaped": reaped, "sessions_closed": closed, "sessions_reaped": sessions}

    _job("log_gc", work)


# job_id -> wrapper. The single source of truth for which jobs exist and how to run one,
# shared by start_scheduler (recurring registration) and trigger_job (manual "run now").
JOB_FUNCS: dict[str, Callable[[], None]] = {
    "rss_poll": _rss_job,
    "enrich": _enrich_job,
    "backfill": _backfill_job,
    "autotag": _autotag_job,
    "shorts": _shorts_job,
    "subscriptions": _subscriptions_job,
    "playlist_sync": _playlist_sync_job,
    "maintenance": _maintenance_job,
    "explore_cleanup": _explore_cleanup_job,
    "download_gc": _download_gc_job,
    "plex_sync": _plex_sync_job,
    "plex_watch_sync": _plex_watch_sync_job,
    "plex_watch_reconcile": _plex_watch_reconcile_job,
    "tmdb_enrich": _tmdb_enrich_job,
    "people_enrich": _people_enrich_job,
    "audit_gc": _audit_gc_job,
    "log_gc": _log_gc_job,
}


def _is_running(job_id: str) -> bool:
    with _activity_lock:
        return bool(_activity.get(job_id, {}).get("running"))


def _claim_running(job_id: str) -> bool:
    """Atomically claim a job as running: True if it was idle (and marks it running now), False if
    it was already running. Holding the lock across BOTH the check and the set is what makes the
    single-run guard race-free — called once at the top of _job so a manual trigger and a scheduled
    fire (or two rapid triggers) converge here and exactly one wins. Released in _job's finally."""
    with _activity_lock:
        if _activity.get(job_id, {}).get("running"):
            return False
        _activity.setdefault(job_id, {})["running"] = True
        return True


def running_job_ids() -> set[str]:
    """Ids of jobs executing right now (any trigger). Lets non-admin surfaces — e.g. the
    header's sync indicator — reflect real activity instead of just pending-work counts."""
    with _activity_lock:
        return {jid for jid, a in _activity.items() if a.get("running")}


def _as_actor(fn: Callable[[], None], actor_id: int | None) -> Callable[[], None]:
    """`fn` run as `actor_id`'s manual run. The actor is RESET afterwards: the executor's pool
    threads are reused and keep their context from task to task, so an actor left set would make
    the next scheduled run that thread picks up look like this admin's manual run."""

    def run() -> None:
        token = _manual_actor.set(actor_id)
        try:
            fn()
        finally:
            _manual_actor.reset(token)

    return run


def trigger_job(job_id: str, actor_id: int | None = None) -> str:
    """Run a job now, independent of its interval schedule: a one-shot job on the same scheduler,
    so a manual run shares the scheduled runs' executor and shutdown — no thread of its own. The
    wrapper handles its own DB session, run row and pause-skip. Returns "started",
    "already_running", or raises KeyError for an unknown job.

    `actor_id` (the admin who clicked "run now") is set in a contextvar inside the run's thread, so
    the run is recorded as theirs and posts a completion notification to their inbox."""
    fn = JOB_FUNCS.get(job_id)
    if fn is None:
        raise KeyError(job_id)
    # Best-effort fast answer for the UI; the authoritative single-run guard is _job's atomic claim
    # (so a manual trigger and a scheduled fire can't both run). A race here just means the loser's
    # _job claims, sees it's taken, and skips — no double run.
    if _is_running(job_id):
        return "already_running"
    sched = _scheduler
    if sched is None:
        # Only between shutdown and exit: start_scheduler runs at every API startup.
        raise RuntimeError("the scheduler is not running")

    try:
        # One id per job: a second click before the first one-shot has even started is the same
        # run, not a queued second one. misfire_grace_time=None: a busy pool delays it, never drops it.
        sched.add_job(_as_actor(fn, actor_id), "date", id=f"{job_id}@now", misfire_grace_time=None)
    # silent-ok: not a failure — the answer is the 409 the admin sees ("already running").
    except ConflictingIdError:
        return "already_running"
    return "started"


def trigger_all(actor_id: int | None = None) -> list[str]:
    """Trigger every job that isn't already running; returns the ids actually started."""
    started = []
    for job_id in JOB_FUNCS:
        if trigger_job(job_id, actor_id) == "started":
            started.append(job_id)
    return started


# Work a user starts on one target outside the job list — today the "get full history" backfill of
# one channel. It runs on the same scheduler as a recorded run (trigger `ad_hoc`), so what it did and
# spent is in `job_runs` like any job's, not a background task only the log ever heard of.
CHANNEL_BACKFILL = "channel_backfill"
# Every ad-hoc job id: the page names each one (a frontend contract test reads this tuple).
AD_HOC_JOBS = (CHANNEL_BACKFILL,)


def run_ad_hoc(job: str, fn: Callable[[], dict], *, actor_id: int, target: str) -> None:
    """Run `fn` now as `actor_id`'s run of `job` on `target`, a one-shot job on the scheduler. Its
    own exclusion (if any) lives in `fn`, which says so in its result; one id per job and target, so
    a second request before the first has even started is that same run."""
    if job not in AD_HOC_JOBS:
        raise KeyError(job)  # an unnamed job would show on the page as its bare id
    sched = _scheduler
    if sched is None:
        # Only between shutdown and exit: start_scheduler runs at every API startup.
        raise RuntimeError("the scheduler is not running")
    try:
        sched.add_job(
            _ad_hoc,
            "date",
            args=(job, fn, actor_id, target),
            id=f"{job}@{target}",
            misfire_grace_time=None,
        )
    # silent-ok: not a failure — the run for this target is already queued, and it does this work.
    except ConflictingIdError:
        log.info("ad-hoc %s for %s already queued", job, target)


def _ad_hoc(job: str, fn: Callable[[], dict], actor_id: int, target: str) -> None:
    try:
        run_id = jobruns.start(job, RunTrigger.AD_HOC, actor_id, target)
    except Exception:
        # No row to tell it in: the database is what failed. The log is the record.
        log.exception("ad-hoc %s for %s not started: its run could not be recorded", job, target)
        return
    with logs.bind(run_id=run_id):
        # As in _run: only the work is inside the try.
        try:
            result = fn()
        except Exception as exc:
            log.exception("job %s failed", job, extra=logs.event("scheduler.job.failed", job=job))
            err, failure = _failure(exc)
            _close(run_id, RunStatus.ERROR, error=err, failure=failure)
            return
        _finish_ok(job, run_id, result)


# Room for every job at once (a "run all") beside a few ad-hoc runs: APScheduler's default pool of
# 10 would queue a scheduled fire behind long manual runs.
_POOL_SIZE = len(JOB_FUNCS) + 8


def start_scheduler() -> None:
    """Start this process's scheduler. It always starts, so "run now" works on every install; only
    with SCHEDULER_ENABLED does it also register the recurring interval jobs. Before anything can
    run, every run a previous process left open is closed as interrupted — one API process serves a
    database (app/jobruns.py), so an open run here is a dead process's."""
    global _scheduler
    if _scheduler is not None:
        return
    left = jobruns.interrupt_leftovers()
    if left:
        log.warning("%s run(s) left running by the previous process marked interrupted", left)
    scheduler = BackgroundScheduler(
        timezone="UTC", executors={"default": ThreadPoolExecutor(_POOL_SIZE)}
    )
    if settings.scheduler_enabled:
        # Effective intervals = env defaults overlaid with any admin overrides from the DB.
        db = SessionLocal()
        try:
            iv = load_intervals(db)
        finally:
            db.close()
        for job_id, fn in JOB_FUNCS.items():
            scheduler.add_job(fn, "interval", minutes=iv[job_id], id=job_id)
    scheduler.start()
    _scheduler = scheduler
    log.info("scheduler started (interval jobs %s)", "on" if settings.scheduler_enabled else "off")


def shutdown_scheduler() -> None:
    global _scheduler
    if _scheduler is not None:
        _scheduler.shutdown(wait=False)
        _scheduler = None