"""Soak and chaos harness: does the engine keep its promises under load and failure?

Standalone on purpose, and never part of the test run. It drives the running
stack through the *durable* path — the webhook, which journals every trigger to
the work queue — and then takes pieces of the stack away to see what survives.

    make soak
    make soak SOAK_ARGS="--minutes 30 --rps 25"
    make soak SOAK_ARGS="--scenario redis"
    make soak SOAK_ARGS="--dry-run"

Every docker verb goes through :func:`docker`, which refuses any container that
is not part of this project's compose stack. The machine this runs on hosts
unrelated services; nothing here may ever reach them.

The engine's semantic is latest-value-wins with cascades running concurrently,
so counter equality is not a promise and is not asserted. What is asserted is
that the queue drains, that state ends on the last value sent, and that nothing
was dead-lettered.
"""

from __future__ import annotations

import argparse
import logging
import secrets as token
import subprocess
import sys
import threading
import time
from collections.abc import Callable
from concurrent.futures import ThreadPoolExecutor
from dataclasses import dataclass, field
from typing import Any

import httpx

from fluksio.core.config import settings
from fluksio.flow.metrics import FLUSH_INTERVAL_S

log = logging.getLogger("soak")

# -----------------------------------------------------------------------------
# Host safety
#
# This host runs critical services that have nothing to do with fluksio. The
# allow-list is checked first, the compose label second, and both before
# anything is executed — a typo must never become a restart of someone else's
# database.
# -----------------------------------------------------------------------------

COMPOSE_PROJECT = "fluksio-app"
REDIS = "fluksio-redis"
API = "fluksio-api"
BROKER = "fluksio-app-mosquitto-1"
DB = "fluksio-db"
#: The only containers this harness may ever name. The first three are the
#: chaos targets; the database is read and cleaned up, never stopped.
ALLOWED = frozenset({REDIS, API, BROKER, DB})


class HostSafety(RuntimeError):
    """A docker call was about to touch something outside this stack."""


def _compose_project(container: str) -> str:
    """Which compose project owns a container, or "" if there is no such container."""
    result = subprocess.run(
        [
            "docker",
            "inspect",
            "-f",
            '{{index .Config.Labels "com.docker.compose.project"}}',
            container,
        ],
        capture_output=True,
        text=True,
        timeout=30,
    )
    return result.stdout.strip() if result.returncode == 0 else ""


def docker(container: str, *argv: str, timeout: float = 120.0) -> str:
    """Run ``docker <argv>`` against a container of this stack, and nothing else."""
    if container not in ALLOWED:
        raise HostSafety(f"'{container}' is not one of {sorted(ALLOWED)} — refusing")
    if container not in argv:
        raise HostSafety(f"'{container}' is guarded but absent from {argv} — refusing")
    project = _compose_project(container)
    if project != COMPOSE_PROJECT:
        raise HostSafety(
            f"'{container}' belongs to compose project '{project}', "
            f"not '{COMPOSE_PROJECT}' — refusing"
        )
    result = subprocess.run(
        ["docker", *argv], capture_output=True, text=True, timeout=timeout
    )
    if result.returncode != 0:
        raise RuntimeError(f"docker {' '.join(argv)}: {result.stderr.strip()}")
    return result.stdout


def is_ours(container: str) -> bool:
    """Does this container exist and belong to our stack? Inspect only."""
    return _compose_project(container) == COMPOSE_PROJECT


def is_running(container: str) -> bool:
    return (
        docker(container, "inspect", "-f", "{{.State.Running}}", container).strip()
        == "true"
    )


# -----------------------------------------------------------------------------
# Thresholds
# -----------------------------------------------------------------------------

MAX_RPS = 50
HTTP_TIMEOUT = 10.0
P95_LIMIT_MS = 500.0
ACCEPT_RATE = 0.99
#: Cascades coalesce — a value superseded before its node ran is not a loss.
#: The counter is therefore a floor, not an equality.
COUNT_FLOOR = 0.9
#: How long a quiet queue is given to prove it after the load stops.
DRAIN_S = 90.0
#: An unacknowledged entry younger than this is a cascade that is running, not
#: work that was lost. See :func:`is_quiet`.
QUIET_AGE_S = 5.0
#: Redelivery after an engine death waits on the reclaim path, not on the
#: engine: XAUTOCLAIM only takes entries idle for RECLAIM_IDLE_MS (60 s) and the
#: reaper runs every RECLAIM_INTERVAL_S (30 s) — see app/flow/executor.py. Do
#: not tighten this below their sum plus room to actually run the work.
ENGINE_DRAIN_S = 150.0
HEALTH_RECOVER_S = 90.0
REDIS_RECOVER_S = 60.0
#: A subscription is restarted with backoff 1, 5, 30, 60 s (app/flow/
#: supervision.py), so a reconnect can legitimately take half a minute.
BROKER_RECOVER_S = 60.0

# -----------------------------------------------------------------------------
# Fixtures
# -----------------------------------------------------------------------------

#: Flow names are lowercase identifiers (app/flow/schemas.py NAME_PATTERN), so
#: the marker is an underscore rather than a hyphen.
PREFIX = "soak_"
FLOW_LOAD = f"{PREFIX}load"
FLOW_CHAIN = f"{PREFIX}chain"
FLOW_MQTT = f"{PREFIX}mqtt"
#: Deletion order: whatever reads another flow's messages goes first.
FLOWS = (FLOW_MQTT, FLOW_CHAIN, FLOW_LOAD)

COUNT_SOURCE = '''"""Count what made it through, reading the total it also writes."""


def process(tick, total=0, params=None):
    return {"total": int(total) + 1}
'''


def spec(name: str, dtype: str = "int", trigger: bool = True) -> dict[str, Any]:
    return {"name": name, "dtype": dtype, "trigger": trigger}


def load_flow(secret: str) -> dict[str, Any]:
    """A webhook feeding an accumulator: the durable path, end to end."""
    return {
        "name": FLOW_LOAD,
        "title": "Soak load",
        "version": 0,
        "nodes": [
            {
                "id": "hook",
                "type": "http",
                "title": "Tick",
                "position": {"x": 40, "y": 40},
                "params": {"url": "/tick", "method": "POST", "secret": secret},
                "provides": [spec("tick")],
            },
            {
                "id": "count",
                "type": "python",
                "title": "Count",
                "position": {"x": 360, "y": 40},
                # `total` does not trigger: the node reads the message it is
                # about to write, which is how a node keeps a running total.
                "requires": [spec("tick"), spec("total", trigger=False)],
                "provides": [spec("total")],
            },
        ],
        "inputs": [{"spec": spec("total"), "initial": 0}],
    }


def chain_flow() -> dict[str, Any]:
    """A second flow reading the first one's message, so cascades cross flows."""
    return {
        "name": FLOW_CHAIN,
        "title": "Soak chain",
        "version": 0,
        "nodes": [
            {
                "id": "double",
                "type": "change",
                "title": "Double",
                "position": {"x": 40, "y": 40},
                "params": {"scale": 2.0, "offset": 0.0},
                "requires": [spec(f"{FLOW_LOAD}.total")],
                "provides": [spec("doubled", dtype="float")],
            }
        ],
    }


def mqtt_flow() -> dict[str, Any]:
    """A subscriber on the dev stack's own broker, for the broker scenario."""
    return {
        "name": FLOW_MQTT,
        "title": "Soak broker",
        "version": 0,
        "nodes": [
            {
                "id": "sub",
                "type": "mqtt",
                "title": "Broker",
                "position": {"x": 40, "y": 40},
                "params": {
                    "topic": "soak/tick",
                    "broker_host": "mosquitto",
                    "broker_port": 1883,
                    "client_id": "fluksio-soak",
                },
                "provides": [spec("tick", dtype="float")],
            }
        ],
    }


# -----------------------------------------------------------------------------
# The API
# -----------------------------------------------------------------------------


def connect(base_url: str) -> httpx.Client:
    """A logged-in client for the JSON API."""
    client = httpx.Client(base_url=base_url, timeout=HTTP_TIMEOUT)
    response = client.post(
        "/api/v1/login/access-token",
        data={
            "username": settings.FIRST_SUPERUSER,
            "password": settings.FIRST_SUPERUSER_PASSWORD,
        },
    )
    response.raise_for_status()
    client.headers["Authorization"] = f"Bearer {response.json()['access_token']}"
    return client


def health_code(client: httpx.Client) -> int:
    """The deep health check's status: 200 while the engine can do its job."""
    try:
        return client.get("/api/v1/utils/health/").status_code
    except httpx.HTTPError:
        return 0


def queue_stats(client: httpx.Client) -> dict[str, Any]:
    """Queue depth and age. Empty when the engine cannot say."""
    try:
        response = client.get("/api/v1/observability/summary")
        response.raise_for_status()
        queue: dict[str, Any] = response.json()["queue"]
    except (httpx.HTTPError, KeyError, ValueError):
        return {}
    return {} if queue.get("error") else queue


def message_value(client: httpx.Client, flow: str, message: str) -> Any:
    """The last value on a message, or None when it has none — or the API is down."""
    try:
        response = client.get(f"/api/v1/flows/{flow}/state")
        response.raise_for_status()
        entry = response.json()["values"].get(f"{flow}.{message}")
    except (httpx.HTTPError, KeyError, ValueError):
        return None
    return entry["value"] if entry else None


def node_health(client: httpx.Client, flow: str, node_id: str) -> str:
    """How a node says its connection is doing: ok, degraded or down."""
    try:
        response = client.get(f"/api/v1/flows/{flow}")
        response.raise_for_status()
    except httpx.HTTPError:
        return "unknown"
    for status in response.json()["nodes"]:
        if status["id"] == f"{flow}.{node_id}":
            return str(status["health"])
    return "unknown"


def dead_letters(client: httpx.Client) -> dict[str, dict[str, Any]]:
    """Work the engine gave up on, keyed by entry id."""
    try:
        response = client.get(
            "/api/v1/observability/dead-letter", params={"limit": 200}
        )
        response.raise_for_status()
    except httpx.HTTPError:
        return {}
    return {str(row["id"]): row for row in response.json()}


def create_flow(
    client: httpx.Client,
    definition: dict[str, Any],
    sources: dict[str, str] | None = None,
) -> None:
    """Draft, node source, publish — the sequence the editor uses."""
    name = definition["name"]
    response = client.put(f"/api/v1/flows/{name}", json=definition)
    response.raise_for_status()
    version = response.json()["definition"]["version"]

    for node_id, code in (sources or {}).items():
        client.put(
            f"/api/v1/flows/{name}/nodes/{node_id}/source", json={"code": code}
        ).raise_for_status()

    response = client.post(f"/api/v1/flows/{name}/publish", json={"version": version})
    response.raise_for_status()
    issues = response.json()["issues"]
    if issues:
        raise RuntimeError(
            f"'{name}' would not run: {[issue['message'] for issue in issues]}"
        )
    log.info("created flow '%s'", name)


def delete_flows(client: httpx.Client) -> None:
    for name in FLOWS:
        try:
            response = client.delete(f"/api/v1/flows/{name}", timeout=30.0)
            if response.status_code == 200:
                log.info("deleted flow '%s'", name)
        except httpx.HTTPError as exc:
            log.error("could not delete flow '%s': %s", name, exc)


# -----------------------------------------------------------------------------
# Waiting
# -----------------------------------------------------------------------------


def wait_for(
    predicate: Callable[[], bool], timeout: float, interval: float = 1.0
) -> float | None:
    """How many seconds until ``predicate`` held, or None if it never did."""
    started = time.monotonic()
    while True:
        if predicate():
            return time.monotonic() - started
        if time.monotonic() - started >= timeout:
            return None
        time.sleep(interval)


def is_quiet(stats: dict[str, Any]) -> bool:
    """Is nothing *waiting* in the queue?

    The queue belongs to the whole stack, not to this run, so "pending is zero"
    is the wrong question on a machine where other flows keep triggering. What
    is asked instead is whether anything has been sitting there: an item that
    was claimed a moment ago belongs to a cascade that is simply running, while
    work the engine has lost hold of only gets older. ``oldest_pending_s`` is
    the age of the *oldest* unacknowledged entry, so a stuck item always shows
    even while newer ones come and go.

    ``delayed`` is deliberately not part of it: a rate-limited output keeps a
    flush item parked there for as long as the flow runs, and no soak fixture
    declares an interval, so none of that backlog is this run's.
    """
    return bool(stats) and (
        not stats.get("pending")
        or float(stats.get("oldest_pending_s") or 0) < QUIET_AGE_S
    )


def settle(client: httpx.Client, timeout: float) -> tuple[bool, dict[str, Any]]:
    """Wait for the work queue to go quiet, two clear polls in a row.

    Two, because a freshly journaled item sits in the stream unclaimed for up to
    the consumer's block window without ever showing up as pending.
    """
    quiet = 0
    stats: dict[str, Any] = {}
    deadline = time.monotonic() + timeout
    while True:
        stats = queue_stats(client)
        quiet = quiet + 1 if is_quiet(stats) else 0
        if quiet >= 2:
            return True, stats
        if time.monotonic() >= deadline:
            return False, stats
        time.sleep(1.5)


def seconds(value: float | None) -> str:
    return f"{value:.1f}s" if value is not None else "never"


# -----------------------------------------------------------------------------
# The load
# -----------------------------------------------------------------------------


class Driver:
    """Fires the webhook at a fixed rate from a small pool of threads."""

    def __init__(self, url: str, rps: int) -> None:
        self.url = url
        self.sent = 0
        self.accepted = 0
        self.failed = 0
        self.skipped = 0
        self.tick = 0
        self.latencies: list[float] = []
        self._interval = 1.0 / rps
        self._client = httpx.Client(timeout=HTTP_TIMEOUT)
        self._pool = ThreadPoolExecutor(
            max_workers=min(32, rps + 4), thread_name_prefix="soak"
        )
        # Bounds what a stalled backend can make this script hold: a request
        # with no free slot is skipped rather than queued forever.
        self._slots = threading.Semaphore(4 * rps)
        self._lock = threading.Lock()
        self._stop = threading.Event()
        self._pacer = threading.Thread(target=self._pace, name="soak-pacer")

    def start(self) -> None:
        self._pacer.start()

    def stop(self) -> None:
        self._stop.set()
        self._pacer.join()
        self._pool.shutdown(wait=True)

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

    def post(self, tick: int) -> bool:
        """One webhook call, timed. Returns whether it was accepted."""
        started = time.perf_counter()
        try:
            accepted = self._client.post(self.url, json={"tick": tick}).is_success
        except httpx.HTTPError:
            accepted = False
        elapsed_ms = (time.perf_counter() - started) * 1000
        with self._lock:
            self.latencies.append(elapsed_ms)
            if accepted:
                self.accepted += 1
            else:
                self.failed += 1
        return accepted

    def sentinel(self) -> int | None:
        """One last tick, sent on its own — what state has to end on."""
        with self._lock:
            self.tick += 1
            self.sent += 1
            tick = self.tick
        return tick if self.post(tick) else None

    def _fire(self, tick: int) -> None:
        try:
            self.post(tick)
        finally:
            self._slots.release()

    def _pace(self) -> None:
        due = time.monotonic()
        while not self._stop.is_set():
            due += self._interval
            if self._slots.acquire(blocking=False):
                with self._lock:
                    self.tick += 1
                    self.sent += 1
                    tick = self.tick
                self._pool.submit(self._fire, tick)
            else:
                with self._lock:
                    self.skipped += 1
            self._stop.wait(max(0.0, due - time.monotonic()))


def percentile(values: list[float], fraction: float) -> float:
    if not values:
        return 0.0
    ordered = sorted(values)
    return ordered[min(len(ordered) - 1, int(fraction * len(ordered)))]


# -----------------------------------------------------------------------------
# Reporting
# -----------------------------------------------------------------------------


@dataclass
class Report:
    """What one scenario measured, and which of its promises held."""

    name: str
    stats: list[str] = field(default_factory=list)
    checks: list[tuple[str, bool]] = field(default_factory=list)
    notes: list[str] = field(default_factory=list)
    skipped: str = ""

    def stat(self, line: str) -> None:
        self.stats.append(line)

    def note(self, line: str) -> None:
        self.notes.append(line)

    def check(self, description: str, ok: bool) -> None:
        self.checks.append((description, bool(ok)))

    @property
    def ok(self) -> bool:
        return all(ok for _, ok in self.checks)

    def show(self) -> None:
        log.info("")
        log.info("── %s %s", self.name, "─" * max(3, 56 - len(self.name)))
        if self.skipped:
            log.info("   SKIP  %s", self.skipped)
            return
        for line in self.stats:
            log.info("   %s", line)
        for description, ok in self.checks:
            log.info("   %s  %s", "pass" if ok else "FAIL", description)
        for line in self.notes:
            log.info("   note  %s", line)


def report_load(report: Report, driver: Driver) -> None:
    report.stat(
        f"posts sent {driver.sent}  accepted {driver.accepted}  "
        f"failed {driver.failed}  skipped {driver.skipped}"
    )
    report.stat(
        f"latency p50 {percentile(driver.latencies, 0.5):.0f} ms  "
        f"p95 {percentile(driver.latencies, 0.95):.0f} ms"
    )


def finish_load(
    client: httpx.Client,
    driver: Driver,
    report: Report,
    drain_s: float,
    dead_before: dict[str, dict[str, Any]],
) -> None:
    """Stop the load and check how the engine ended up: drained, fresh, nothing lost."""
    driver.stop()
    report_load(report, driver)

    drained, _ = settle(client, drain_s)
    report.check(f"queue drains within {drain_s:.0f}s of the load stopping", drained)

    # Sent alone, after the drain: with four cascades running at once, only a
    # tick that had the queue to itself is guaranteed to be the last one
    # written. Latest-value-wins says nothing about which of two racing
    # cascades lands second.
    tick = driver.sentinel()
    report.check("the last webhook was accepted", tick is not None)

    drained, stats = settle(client, drain_s)
    report.check("nothing left waiting in the queue at the end", drained)
    report.stat(
        f"queue (shared with the rest of the stack) pending {stats.get('pending', '?')}"
        f"  oldest {stats.get('oldest_pending_s', '?')}s"
        f"  delayed {stats.get('delayed', '?')}  parked {stats.get('parked', '?')}"
    )

    value = message_value(client, FLOW_LOAD, "tick")
    total = message_value(client, FLOW_LOAD, "total")
    report.stat(f"state tick {value}  total {total}")
    report.check(
        f"state ends on the last tick sent (sent {tick}, holds {value})",
        tick is not None and value == tick,
    )

    new_dead = [
        row for key, row in dead_letters(client).items() if key not in dead_before
    ]
    ours = [row for row in new_dead if str(row.get("flow", "")).startswith(PREFIX)]
    report.stat(f"dead letters: {len(ours)} for soak flows, {len(new_dead)} in total")
    report.check("nothing was dead-lettered", not ours)

    report.check("engine healthy at the end", health_code(client) == 200)
    driver.close()


# -----------------------------------------------------------------------------
# Scenarios
# -----------------------------------------------------------------------------


def scenario_load(client: httpx.Client, args: argparse.Namespace, url: str) -> Report:
    """(a) Steady load down the durable path."""
    report = Report("load")
    dead_before = dead_letters(client)
    before = message_value(client, FLOW_LOAD, "total") or 0

    driver = Driver(url, args.rps)
    log.info("load: %d rps for %.0fs", args.rps, args.seconds)
    driver.start()
    time.sleep(args.seconds)
    finish_load(client, driver, report, DRAIN_S, dead_before)

    rate = driver.accepted / driver.sent if driver.sent else 0.0
    report.check(
        f"at least {ACCEPT_RATE:.0%} of posts accepted ({rate:.1%})",
        rate >= ACCEPT_RATE,
    )
    p95 = percentile(driver.latencies, 0.95)
    report.check(f"p95 under {P95_LIMIT_MS:.0f} ms ({p95:.0f} ms)", p95 < P95_LIMIT_MS)

    total = message_value(client, FLOW_LOAD, "total")
    counted = (total or 0) - before
    report.check(
        f"counter kept up with the load ({counted} of {driver.accepted} accepted)",
        total is not None and counted >= COUNT_FLOOR * driver.accepted,
    )
    doubled = message_value(client, FLOW_CHAIN, "doubled")
    report.check(
        f"the downstream flow saw the final total ({doubled} vs {total})",
        total is not None and doubled == total * 2,
    )
    report.show()
    return report


def scenario_redis(client: httpx.Client, args: argparse.Namespace, url: str) -> Report:
    """(b) The state and journal backend goes away mid-load."""
    report = Report("redis")
    dead_before = dead_letters(client)

    driver = Driver(url, args.rps)
    log.info("redis: %d rps, taking %s away halfway through", args.rps, REDIS)
    driver.start()
    time.sleep(args.seconds / 2)

    before = message_value(client, FLOW_LOAD, "total")
    # Past redis' appendfsync-everysec window, so what was just read is on disk
    # and its survival is a real claim about the append-only file.
    time.sleep(1.5)

    docker(REDIS, "stop", REDIS)
    degraded = wait_for(lambda: health_code(client) != 200, 30.0, 0.5)
    report.check("health reports trouble while redis is gone", degraded is not None)

    docker(REDIS, "start", REDIS)
    recovered = wait_for(lambda: health_code(client) == 200, REDIS_RECOVER_S, 1.0)
    report.check(
        f"health back to 200 within {REDIS_RECOVER_S:.0f}s of redis returning",
        recovered is not None,
    )
    report.stat(
        f"degraded after {seconds(degraded)}, healthy again {seconds(recovered)} "
        "after the restart"
    )

    after = message_value(client, FLOW_LOAD, "total")
    report.check(
        f"state journaled before the outage survived it ({before} → {after})",
        before is not None and after is not None and after >= before,
    )

    time.sleep(args.seconds / 2)
    finish_load(client, driver, report, DRAIN_S, dead_before)
    report.show()
    return report


def scenario_engine(client: httpx.Client, args: argparse.Namespace, url: str) -> Report:
    """(c) The engine is killed mid-cascade and has to pick the work back up."""
    report = Report("engine")
    dead_before = dead_letters(client)

    driver = Driver(url, args.rps)
    log.info("engine: %d rps, killing %s halfway through", args.rps, API)
    driver.start()
    time.sleep(args.seconds / 2)

    at_kill = message_value(client, FLOW_LOAD, "total")
    # Read after the total, so work already accepted but not yet counted makes
    # the claim below easier rather than impossible.
    accepted_at_kill = driver.accepted

    docker(API, "kill", API)
    time.sleep(2.0)
    # The dev stack sets restart:"no" for the backend, so nothing else brings
    # it back — and it must never be left down.
    docker(API, "start", API)
    recovered = wait_for(lambda: health_code(client) == 200, HEALTH_RECOVER_S, 2.0)
    report.stat(f"engine answered healthy again {seconds(recovered)} after restart")
    report.check(
        f"health back to 200 within {HEALTH_RECOVER_S:.0f}s of the restart",
        recovered is not None,
    )
    if recovered is None:
        log.error("engine health after restart: %s", health_code(client))

    time.sleep(args.seconds / 2)
    finish_load(client, driver, report, ENGINE_DRAIN_S, dead_before)

    accepted_after = driver.accepted - accepted_at_kill
    after = message_value(client, FLOW_LOAD, "total")
    floor = (at_kill or 0) + COUNT_FLOOR * accepted_after
    report.check(
        f"counter carried on across the kill ({at_kill} → {after}, "
        f"{accepted_after} accepted after it)",
        at_kill is not None and after is not None and after >= floor,
    )
    report.note(
        "alert delivery is not asserted — that needs a channel on the receiving end"
    )
    report.show()
    return report


def scenario_broker(client: httpx.Client) -> Report:
    """(e) The MQTT broker restarts under a live subscription."""
    report = Report("broker")
    if not is_ours(BROKER):
        report.skipped = f"no {BROKER} container — the broker lives in the dev stack"
        report.show()
        return report

    def subscribed() -> bool:
        return node_health(client, FLOW_MQTT, "sub") == "ok"

    def publish(value: str) -> None:
        docker(BROKER, "exec", BROKER, "mosquitto_pub", "-t", "soak/tick", "-m", value)

    report.check("the subscriber connects", wait_for(subscribed, 60.0, 2.0) is not None)
    publish("1")
    landed = wait_for(
        lambda: message_value(client, FLOW_MQTT, "tick") == 1.0, 30.0, 1.0
    )
    report.check("a published value reaches the flow", landed is not None)

    docker(BROKER, "stop", BROKER)
    down = wait_for(lambda: node_health(client, FLOW_MQTT, "sub") == "down", 60.0, 1.0)
    report.check(
        "the subscriber reports down while the broker is gone", down is not None
    )

    docker(BROKER, "start", BROKER)
    back = wait_for(subscribed, BROKER_RECOVER_S, 2.0)
    report.check(
        f"the subscriber is ok again within {BROKER_RECOVER_S:.0f}s", back is not None
    )
    report.stat(f"reported down after {seconds(down)}, ok again {seconds(back)} later")

    publish("2")
    landed = wait_for(
        lambda: message_value(client, FLOW_MQTT, "tick") == 2.0, 30.0, 1.0
    )
    report.check("a value published after recovery lands", landed is not None)
    report.show()
    return report


# -----------------------------------------------------------------------------
# Cleanup
# -----------------------------------------------------------------------------

#: Mirrors FlowController.forget_flow: the messages themselves, and the
#: engine's bookkeeping about them, which is keyed by message name too.
REDIS_PATTERNS = (f"pipeline:{PREFIX}*", f"pipeline:*:{PREFIX}*")


def clear_redis() -> None:
    """Belt and braces: deleting a flow clears its keys through the engine,
    which is no help when the engine was the thing that died."""
    try:
        for pattern in REDIS_PATTERNS:
            keys = docker(
                REDIS, "exec", REDIS, "redis-cli", "--scan", "--pattern", pattern
            ).split()
            for start in range(0, len(keys), 200):
                docker(
                    REDIS, "exec", REDIS, "redis-cli", "del", *keys[start : start + 200]
                )
            if keys:
                log.info("deleted %d redis key(s) matching %s", len(keys), pattern)
    except Exception as exc:
        log.error("redis keys may be left behind: %s", exc)


def clear_observability() -> None:
    """The collector wrote rollups, runs and events for the soak flows."""
    names = "', '".join(FLOWS)
    statement = "; ".join(
        f"DELETE FROM {table} WHERE flow IN ('{names}')"
        for table in ("metric_minute", "flow_run", "engine_event")
    )
    try:
        docker(
            DB,
            "exec",
            DB,
            "psql",
            "-q",
            "-U",
            settings.POSTGRES_USER,
            "-d",
            settings.POSTGRES_DB,
            "-c",
            statement,
        )
        log.info("deleted observability rows for the soak flows")
    except Exception as exc:
        log.error("observability rows may be left behind: %s", exc)


def cleanup(client: httpx.Client | None) -> None:
    """Undo everything this run created, whatever went wrong on the way."""
    log.info("")
    log.info("cleaning up")
    for container in (REDIS, API, BROKER):
        try:
            if is_ours(container) and not is_running(container):
                log.info("starting %s", container)
                docker(container, "start", container)
        except Exception as exc:
            log.error("could not start %s: %s", container, exc)

    if client is not None:
        wait_for(lambda: health_code(client) == 200, HEALTH_RECOVER_S, 2.0)
        delete_flows(client)

    clear_redis()
    # Deleting a flow publishes an audit event of its own, and the metrics
    # collector buffers what it hears for up to FLUSH_INTERVAL_S before writing
    # it. Clearing the tables any sooner races that flush and leaves the rows
    # this very cleanup caused.
    time.sleep(FLUSH_INTERVAL_S + 5.0)
    clear_observability()


# -----------------------------------------------------------------------------
# Entry point
# -----------------------------------------------------------------------------


def parse_args(argv: list[str] | None) -> argparse.Namespace:
    parser = argparse.ArgumentParser(
        description="Soak and chaos harness for the running fluksio stack.",
    )
    parser.add_argument(
        "--minutes", type=float, default=2.0, help="Load time per scenario (default 2)."
    )
    parser.add_argument(
        "--rps",
        type=int,
        default=10,
        help=f"Webhooks per second, capped at {MAX_RPS} (default 10).",
    )
    parser.add_argument(
        "--scenario",
        choices=("all", "load", "redis", "engine", "broker"),
        default="all",
        help="Which scenario to run (default all).",
    )
    parser.add_argument(
        "--base-url",
        default="http://api.localhost",
        help="Where the API is (default http://api.localhost).",
    )
    parser.add_argument(
        "--dry-run",
        action="store_true",
        help="Report what the harness can see and change nothing.",
    )
    args = parser.parse_args(argv)
    args.rps = max(1, min(MAX_RPS, args.rps))
    args.seconds = max(10.0, args.minutes * 60)
    return args


def dry_run(client: httpx.Client, args: argparse.Namespace) -> None:
    """Everything the harness needs, read-only: login, flows, health, containers."""
    log.info("logged in as %s at %s", settings.FIRST_SUPERUSER, args.base_url)
    flows = client.get("/api/v1/flows/").json()["data"]
    log.info("flows: %s", ", ".join(flow["name"] for flow in flows) or "none")
    response = client.get("/api/v1/utils/health/")
    body = response.json()
    log.info("health %s: %s %s", response.status_code, body["status"], body["problems"])
    log.info("queue: %s", body["queue"])
    for container in sorted(ALLOWED):
        ours = is_ours(container)
        log.info(
            "%s: ours=%s running=%s",
            container,
            ours,
            is_running(container) if ours else "-",
        )


def main(argv: list[str] | None = None) -> int:
    args = parse_args(argv)
    logging.basicConfig(level=logging.INFO, format="%(message)s", stream=sys.stdout)
    logging.getLogger("httpx").setLevel(logging.WARNING)

    client = connect(args.base_url)
    if args.dry_run:
        dry_run(client, args)
        return 0

    # The webhook is mounted without authentication, so its secret is the only
    # thing guarding it — even for a fixture that lives for a few minutes.
    secret = token.token_hex(16)
    hook = f"{args.base_url.rstrip('/')}/hooks/{FLOW_LOAD}/tick/{secret}"
    wanted = ("load", "redis", "engine") if args.scenario == "all" else (args.scenario,)
    reports: list[Report] = []
    failure = ""

    try:
        # A previous run that died before its cleanup would otherwise collide.
        delete_flows(client)
        if args.scenario != "broker":
            create_flow(client, load_flow(secret), {"count": COUNT_SOURCE})
            create_flow(client, chain_flow())
        if args.scenario in ("all", "broker") and is_ours(BROKER):
            create_flow(client, mqtt_flow())

        if "load" in wanted:
            reports.append(scenario_load(client, args, hook))
        if "redis" in wanted:
            reports.append(scenario_redis(client, args, hook))
        if "engine" in wanted:
            reports.append(scenario_engine(client, args, hook))
        if args.scenario in ("all", "broker"):
            reports.append(scenario_broker(client))
    except KeyboardInterrupt:
        failure = "interrupted"
    except Exception as exc:
        failure = f"{type(exc).__name__}: {exc}"
        log.exception("the run stopped early")
    finally:
        cleanup(client)

    checks = sum(len(report.checks) for report in reports)
    ok = bool(reports) and not failure and all(report.ok for report in reports)
    log.info("")
    log.info(
        "SOAK %s — %d check(s) across %d scenario(s)%s",
        "PASS" if ok else "FAIL",
        checks,
        len(reports),
        f" — {failure}" if failure else "",
    )
    return 0 if ok else 1


if __name__ == "__main__":
    sys.exit(main())