app/backend/fluksio/flow/devices.py

"""Devices: the screens and boards paired with this instance.

A device is a wall panel, a knob, a camera board: something without a keyboard
that pairs with a code, holds a scoped credential and shows an ordered set of
whole dashboards. A hallway tablet and a workshop one carry different sets, the
same dashboard may hang on both, and each dashboard keeps its own canvas, draft
and version.

The definition is stored in the flow repository as ``_devices/<id>.json`` and
committed like a dashboard, so it is versioned with the dashboards it names.
What a revert must never undo lives outside git, in ``devices.json``: the nonce
a credential is held against, and the device's own report of what it has.
Whether a device is connected right now is kept in memory only.
"""

from __future__ import annotations

import ipaddress
import json
import logging
import os
import re
import threading
import time
from datetime import UTC, datetime
from pathlib import Path
from typing import Any, Literal

from pydantic import BaseModel, Field, TypeAdapter, ValidationError, field_validator

from fluksio.flow.dashboards import DashboardDef, DashboardNotFound, DashboardStore
from fluksio.flow.events import event_bus
from fluksio.flow.messages import validate_qualified
from fluksio.flow.schemas import _validate_name
from fluksio.flow.store import FlowStore, InvalidName, StaleVersion, checked

logger = logging.getLogger(__name__)

#: Sibling of the dashboards, and likewise not a flow.
DEVICE_DIR = "_devices"

#: Free text the device interprets: ``gpio.4``, ``backlight``, ``mic``.
PORT_PATTERN = re.compile(r"^[a-z0-9_.]+$")


class DevicePort(BaseModel):
    """One piece of the device's own hardware, wired to a message.

    A wall panel is a node as well as a screen: a button on its GPIO, a
    microphone, a speaker. Which ports exist is the device's business, so the
    name is free text and ``config`` is passed through to it unread.
    """

    port: str
    #: Qualified, ``flow.name``. A flow still has to declare it, as for any
    #: widget, before a publish to it lands.
    message: str
    #: ``in`` is the device publishing into fluksio, ``out`` is the device
    #: following the message.
    direction: Literal["in", "out"]
    config: dict[str, Any] = Field(default_factory=dict)

    @field_validator("port")
    @classmethod
    def _check_port(cls, value: str) -> str:
        if not PORT_PATTERN.fullmatch(value):
            raise ValueError("Use lowercase letters, digits, underscores and dots")
        return value

    _check_message = field_validator("message")(validate_qualified)


class DeviceDef(BaseModel):
    """One device, and what it shows."""

    id: str
    title: str = ""
    #: Ordered. The first one is what the device opens after pairing, and the
    #: rail follows this order. A name that no longer resolves is a dashboard
    #: someone deleted, and the device skips it.
    dashboards: list[str] = Field(default_factory=list)
    #: Bigger controls, a bigger rail and no hover states, for a screen that is
    #: touched rather than pointed at. A fact about the screen, not about any
    #: dashboard on it.
    touch: bool = False
    #: The device's own inputs and outputs. An ``in`` port may publish whether
    #: or not a dashboard is locked, since a lock covers the controls on a
    #: dashboard and not the sensors on the wall. An ``out`` port only reads.
    io: list[DevicePort] = Field(default_factory=list)
    #: Bumped on every save. A save based on an older one is refused.
    version: int = 0

    @field_validator("id")
    @classmethod
    def _check_id(cls, value: str) -> str:
        return _validate_name(value)


class DeviceState(BaseModel):
    """What this instance keeps about a device outside the flow repository."""

    #: Which generation of credential the device honours. A token names the
    #: nonce it was minted at, so bumping this refuses the screen hanging there
    #: and leaves the device and its dashboards as they are.
    nonce: int = 0
    #: What the device says it has, written by the device itself.
    report: dict[str, Any] | None = None
    reported_at: datetime | None = None


_STATES = TypeAdapter(dict[str, DeviceState])


class DeviceNotFound(KeyError):
    def __init__(self, device_id: str) -> None:
        super().__init__(device_id)
        self.id = device_id


class DeviceExists(ValueError):
    def __init__(self, device_id: str) -> None:
        super().__init__(device_id)
        self.id = device_id


class DeviceStore:
    """Device definitions in the flow repository, their nonces beside it.

    Shares the flow store's write lock and commit, as the dashboards do, so a
    device save cannot interleave with a flow or dashboard save.
    """

    def __init__(self, flows: FlowStore, state_file: Path) -> None:
        self.flows = flows
        self.root = flows.root / DEVICE_DIR
        self.state_file = state_file
        # Read-modify-write of a device's version.
        self._lock = threading.Lock()
        # Read-modify-write of the nonce file. An unpair interleaved with
        # anything else writing it must not be undone by that write.
        self._state_lock = threading.Lock()

    def _file(self, device_id: str) -> Path:
        return self.root / f"{checked(device_id)}.json"

    def _write(self, device: DeviceDef) -> None:
        path = self._file(device.id)
        path.parent.mkdir(parents=True, exist_ok=True)
        path.write_text(device.model_dump_json(indent=2) + "\n")

    def list(self) -> list[DeviceDef]:
        found = []
        for path in sorted(self.root.glob("*.json")):
            try:
                found.append(DeviceDef.model_validate_json(path.read_text()))
            except (OSError, ValidationError):
                logger.warning("Skipping unreadable device %s", path.name)
        return found

    def find(self, device_id: str) -> DeviceDef | None:
        """The device by that id, or None if it is gone.

        Read from disk on every call: this is what makes deleting a device
        revoke its credential, so it has to see the current file.
        """
        try:
            path = self._file(device_id)
        except InvalidName:
            return None
        if not path.exists():
            return None
        try:
            return DeviceDef.model_validate_json(path.read_text())
        except (OSError, ValidationError):
            return None

    def create(self, device: DeviceDef) -> DeviceDef:
        with self._lock, self.flows._write_lock:
            if self._file(device.id).exists():
                raise DeviceExists(device.id)
            saved = device.model_copy(update={"version": 1})
            self._write(saved)
            self.flows._commit(f"Add device '{device.id}'")
            return saved

    def write(self, device: DeviceDef, base_version: int) -> DeviceDef:
        """Save a device, refusing a save someone else has moved past."""
        with self._lock, self.flows._write_lock:
            current = self.find(device.id)
            if current is None:
                raise DeviceNotFound(device.id)
            if base_version != current.version:
                raise StaleVersion(device.id, current.version)
            saved = device.model_copy(update={"version": current.version + 1})
            self._write(saved)
            self.flows._commit(f"Save device '{device.id}'")
            return saved

    def delete(self, device_id: str) -> None:
        """Remove the device, and refuse every credential minted for it.

        The nonce is bumped rather than dropped: a device created again under
        the same id, or brought back by a revert, must not take the old
        screen's credential back with it.
        """
        with self.flows._write_lock:
            path = self._file(device_id)
            if not path.exists():
                raise DeviceNotFound(device_id)
            path.unlink()
            self.flows._commit(f"Delete device '{device_id}'")
        self.bump(device_id)
        presence.forget(device_id)

    # -- outside git -------------------------------------------------------

    def _read_states(self) -> dict[str, DeviceState]:
        """Raises on a file that does not parse, rather than reading it as empty.

        Empty would put every nonce back to 0, which honours again every
        credential an unpair refused.
        """
        if not self.state_file.exists():
            return {}
        return _STATES.validate_json(self.state_file.read_bytes())

    def _write_states(self, states: dict[str, DeviceState]) -> None:
        # Through a temporary file, so a crash mid-write cannot leave a torn
        # file behind.
        self.state_file.parent.mkdir(parents=True, exist_ok=True)
        tmp = self.state_file.with_name(self.state_file.name + ".tmp")
        tmp.write_bytes(_STATES.dump_json(states, indent=2))
        os.replace(tmp, self.state_file)

    def state(self, device_id: str) -> DeviceState:
        return self._read_states().get(device_id, DeviceState())

    def nonce(self, device_id: str) -> int:
        return self.state(device_id).nonce

    def bump(self, device_id: str) -> int:
        """Refuse the credential this device holds now. Returns the new nonce."""
        with self._state_lock:
            states = self._read_states()
            state = states.setdefault(device_id, DeviceState())
            state.nonce += 1
            self._write_states(states)
            return state.nonce

    # -- the panels file of an older instance -------------------------------

    def migrate(self, panels_file: Path) -> None:
        """Move ``panels.json`` into the store and the nonce file, once.

        Each panel becomes ``_devices/<id>.json`` at version 1, in one commit,
        and its nonce goes to ``devices.json`` (the higher one wins, should
        both have it). The file is then renamed ``.migrated``. Running it again
        after a crash part-way skips the devices already written.
        """
        if not panels_file.exists():
            return
        try:
            panels = json.loads(panels_file.read_text()).get("panels", [])
        except (OSError, ValueError, AttributeError) as exc:
            logger.error("Could not read %s to migrate it: %s", panels_file, exc)
            return

        devices: list[tuple[DeviceDef, int]] = []
        for raw in panels:
            try:
                device = DeviceDef.model_validate({**raw, "version": 1})
            except (ValidationError, TypeError) as exc:
                logger.error("Not migrating panel %r: %s", raw, exc)
                continue
            devices.append((device, int(raw.get("nonce") or 0)))

        with self._lock, self.flows._write_lock:
            written = [d.id for d, _ in devices if not self._file(d.id).exists()]
            for device, _ in devices:
                if device.id in written:
                    self._write(device)
            if written:
                self.flows._commit("Migrate panels to devices")
        with self._state_lock:
            states = self._read_states()
            for device, nonce in devices:
                state = states.setdefault(device.id, DeviceState())
                state.nonce = max(state.nonce, nonce)
            self._write_states(states)
        panels_file.replace(panels_file.with_name(panels_file.name + ".migrated"))
        logger.warning(
            "Moved %d panels from %s into devices (%s)",
            len(devices),
            panels_file,
            ", ".join(written) or "all were there already",
        )


class Presence:
    """Which devices are connected now, and when each was last heard from.

    In memory: a restart forgets it, and a device that is still there
    reconnects within seconds. One process serves the API, so one dictionary
    is the whole answer.
    """

    def __init__(self) -> None:
        self._sockets: dict[str, int] = {}
        self._seen: dict[str, float] = {}
        self._lock = threading.Lock()

    def seen(self, device_id: str) -> None:
        self._seen[device_id] = time.time()

    def opened(self, device_id: str) -> None:
        with self._lock:
            count = self._sockets.get(device_id, 0) + 1
            self._sockets[device_id] = count
            self._seen[device_id] = time.time()
        if count == 1:
            self._announce(device_id)

    def closed(self, device_id: str) -> None:
        with self._lock:
            count = max(0, self._sockets.get(device_id, 0) - 1)
            if count:
                self._sockets[device_id] = count
            else:
                self._sockets.pop(device_id, None)
            self._seen[device_id] = time.time()
        if not count:
            self._announce(device_id)

    def forget(self, device_id: str) -> None:
        with self._lock:
            self._sockets.pop(device_id, None)
            self._seen.pop(device_id, None)

    def online(self, device_id: str) -> bool:
        return self._sockets.get(device_id, 0) > 0

    def last_seen(self, device_id: str) -> datetime | None:
        seen = self._seen.get(device_id)
        return datetime.fromtimestamp(seen, UTC) if seen else None

    def _announce(self, device_id: str) -> None:
        last = self.last_seen(device_id)
        event_bus.publish(
            {
                "type": "device_status",
                "device": device_id,
                "online": self.online(device_id),
                "last_seen": last.isoformat() if last else None,
                "ts": time.time(),
            }
        )


presence = Presence()


def is_lan(address: str) -> bool:
    """Whether a request came from a private network rather than the internet."""
    try:
        ip = ipaddress.ip_address(address)
    except ValueError:
        return False
    return ip.is_private or ip.is_loopback


# -- what a device's credential reaches --------------------------------------


def dashboards_for(
    device: DeviceDef | None, store: DashboardStore
) -> list[DashboardDef]:
    """The published documents this device shows, in rail order.

    Published, since that is what a device draws. A name that no longer
    resolves is skipped, and a device that is gone shows nothing.

    Handed back whole rather than walked here, because what a device may do
    with a message depends on the document it came from: a dashboard that says
    it is locked entitles a screen to read it and not to touch it.
    """
    if device is None:
        return []
    found: list[DashboardDef] = []
    for name in device.dashboards:
        try:
            found.append(store.read(name))
        except DashboardNotFound:
            continue
    return found


def messages_of(defn: DashboardDef) -> set[str]:
    """Every message one dashboard reads or writes.

    A dashboard's own bound settings count, not only its widgets': the theme a
    panel is driven to is a message no tile on it draws.
    """
    names = set(defn.setting_messages)
    for widget in defn.widgets:
        names.update(widget.messages)
        if widget.target:
            names.add(widget.target)
    return names


def requests_of(defn: DashboardDef) -> set[str]:
    """The publishes this dashboard makes in order to read.

    A querying chart asks a flow for the series it draws by publishing a
    request, so that publish is how the tile reads. Every other message a
    dashboard sends comes from a control, which is what locking it turns off,
    so this is what a locked dashboard may still send.
    """
    return {w.target for w in defn.widgets if w.type == "chart" and w.target}


def targets_of(defn: DashboardDef) -> set[str]:
    """The messages this dashboard's inputs publish to.

    What a screen showing it may write, a narrower set than what it may read.
    """
    return {w.target for w in defn.widgets if w.target}


def io_in(device: DeviceDef | None) -> set[str]:
    """The messages this device's own hardware publishes to."""
    if device is None:
        return set()
    return {port.message for port in device.io if port.direction == "in"}


def messages_for(device: DeviceDef | None, store: DashboardStore) -> set[str]:
    """Every message this device's dashboards and its own I/O read or write.

    Empty for a device that is gone, which is the same answer as "nothing".
    """
    own = {port.message for port in device.io} if device else set()
    return own | {
        name for defn in dashboards_for(device, store) for name in messages_of(defn)
    }