flush

class Flusher:
    """Writes the steps that report to a scratch into its ``steps`` table."""

    def __init__(self) -> None:
        """A flusher whose thread starts with the first step that has a scratch."""
        self._lock = threading.Lock()
        self._writing = threading.Lock()
        self._wake = threading.Event()
        self._closed: list[Step] = []
        self._thread: threading.Thread | None = None

    def opened(self, step: Step) -> None:
        """Wake the flusher so the index sees ``step`` open promptly."""
        if step.scratch is None:
            return
        with self._lock:
            if self._thread is None:
                self._thread = threading.Thread(target=self._run, name="cg-progress", daemon=True)
                self._thread.start()
        self._wake.set()

    def closed(self, step: Step) -> None:
        """Queue ``step``'s row for deletion and wake the flusher."""
        if step.scratch is None:
            return
        with self._lock:
            self._closed.append(step)
        self._wake.set()

    def flush(self) -> None:
        """Write every dirty open step and delete every closed one, now.

        One flush at a time: reading a step settles its dirty mark, so a caller
        arriving mid-flush would otherwise find nothing to write and return
        before the writer committed.
        """
        with self._writing:
            with self._lock:
                closed, self._closed = self._closed, []
            by_scratch: defaultdict[Scratch, tuple[list[Step], list[Step]]] = defaultdict(
                lambda: ([], [])
            )
            for step in registry.live():
                if step.scratch is not None and step.dirty:
                    by_scratch[step.scratch][0].append(step)
            for step in closed:
                if step.scratch is not None:
                    by_scratch[step.scratch][1].append(step)
            for scratch, (dirty, gone) in by_scratch.items():
                if scratch.closed:
                    continue
                try:
                    with scratch.database.transaction() as db:
                        for step in dirty:
                            steps.upsert(db, _row(step, scratch))
                        for step in gone:
                            steps.delete(db, step.id)
                except Exception:
                    logger.exception(f"progress: could not write steps to {scratch}")

    def _run(self) -> None:
        while True:
            self.flush()
            busy = any(step.scratch is not None for step in registry.live())
            self._wake.wait(INTERVAL_SECONDS if busy else None)
            self._wake.clear()