import os
import time
from datetime import UTC, datetime

import pytest
from sqlmodel import Session, col, select

from fluksio.core.db import engine as db_engine
from fluksio.flow.artifacts import ArtifactStore
from fluksio.flow.runs import new_run_id, sweep_artifacts
from fluksio.flow.state import MemoryState
from fluksio.models import Run, RunArtifact


@pytest.fixture
def store(tmp_path) -> ArtifactStore:
    return ArtifactStore(tmp_path / "artifacts")


def _run_row(status: str, digest: str = "") -> str:
    """A run, and optionally the artifact it recorded. Returns its id."""
    run_id = new_run_id()
    with Session(db_engine) as session:
        session.add(
            Run(id=run_id, flow="f", status=status, created_at=datetime.now(UTC))
        )
        if digest:
            session.add(
                RunArtifact(
                    run_id=run_id,
                    name="f.out",
                    node="n",
                    digest=digest,
                    size=3,
                )
            )
        session.commit()
    return run_id


def _forget(run_id: str) -> None:
    with Session(db_engine) as session:
        rows = session.exec(
            select(RunArtifact).where(col(RunArtifact.run_id) == run_id)
        ).all()
        for row in rows:
            session.delete(row)
        session.delete(session.get(Run, run_id))
        session.commit()


def test_the_grace_window_spares_a_fresh_artifact(store):
    ref = store.put([b"new"])
    assert store.collect(set(), grace_s=3600) == 0
    assert store.path(ref["digest"]) is not None
    assert store.collect(set(), grace_s=0) == 1
    assert store.path(ref["digest"]) is None


def test_a_sweep_keeps_what_a_run_recorded_and_what_a_message_holds(store):
    recorded = store.put([b"kept by a run"])
    held = store.put([b"kept by a message"])
    nested = store.put([b"kept inside a payload"])
    orphan = store.put([b"referred to by nothing"])

    run_id = _run_row("ok", recorded["digest"])
    state = MemoryState()
    state.update(
        {
            "cam.frame": held,
            # A reference inside a json payload counts as much as a bare one.
            "cam.report": {"latest": {"clip": nested}},
            "cam.count": 3,
        }
    )

    try:
        assert sweep_artifacts(store, state) == 1
    finally:
        _forget(run_id)

    assert store.path(orphan["digest"]) is None
    for kept in (recorded, held, nested):
        assert store.path(kept["digest"]) is not None


def test_a_sweep_stands_aside_while_a_run_is_in_flight(store):
    orphan = store.put([b"mid-run checkpoint"])
    run_id = _run_row("running")
    try:
        assert sweep_artifacts(store, MemoryState()) == 0
    finally:
        _forget(run_id)
    assert store.path(orphan["digest"]) is not None


def test_a_streamed_chunk_falls_out_once_the_message_moves_on(store):
    """What makes a media stream affordable: only the current frame is held."""
    state = MemoryState()
    first = store.put([b"frame one"], media_type="image/png")
    state.update({"cam.frame": first})
    assert sweep_artifacts(store, state) == 0

    second = store.put([b"frame two"], media_type="image/png")
    state.update({"cam.frame": second})
    # Old enough to be swept: the sweep only spares what grace covers.
    os.utime(store.path(first["digest"]), (time.time() - 10, time.time() - 10))
    assert sweep_artifacts(store, state, grace_s=5) == 1
    assert store.path(first["digest"]) is None
    assert store.path(second["digest"]) is not None