diff --git a/.github/workflows/python-test.yml b/.github/workflows/python-test.yml index aa6cf376..28e802e4 100644 --- a/.github/workflows/python-test.yml +++ b/.github/workflows/python-test.yml @@ -9,6 +9,7 @@ on: - .github/workflows/python-test.yml - lambda/** - packages/** + - watcher/** - pyproject.toml - uv.lock - Makefile @@ -20,6 +21,7 @@ on: - .github/workflows/python-test.yml - lambda/** - packages/** + - watcher/** - pyproject.toml - uv.lock - Makefile diff --git a/docs/guides/installing-a-watcher.md b/docs/guides/installing-a-watcher.md index 11689f13..728877bd 100644 --- a/docs/guides/installing-a-watcher.md +++ b/docs/guides/installing-a-watcher.md @@ -41,7 +41,7 @@ The wizard will walk you through: 1. **Environment** — choose `staging` (for testing), `production`, or `preview` (for testing against a Vercel preview deployment). If you choose `preview`, you'll be prompted for the deployment's API base URL (e.g. `https://data-hub-git-my-branch.vercel.app/api/v1`). -2. **API key** — paste the personal access token. The key is saved to `~/.data-hub/.env`. You can also set the `DATA_HUB_API_KEY` environment variable before running `init` to skip this prompt. +2. **API key** — paste the personal access token. The key is saved to `~/.data-hub/.env.` (e.g. `~/.data-hub/.env.staging`), so each environment keeps its own key and you can switch between them by re-running `init` without re-entering credentials. You can also set the `DATA_HUB_API_KEY` environment variable before running `init` to skip this prompt. 3. **Instrument** — select an existing instrument from the list, or register a new one by choosing the last option. New instruments start as `pending` and must be activated by an admin in the web app before the watcher can start. @@ -59,7 +59,7 @@ The wizard will walk you through: - **`auto`** — files are uploaded to S3 immediately after detection. - **`manual`** — files are reported to the server but not uploaded until an admin approves them via the upload queue. -The wizard saves configuration to `~/.data-hub/config.yaml`, the API key to `~/.data-hub/.env`, and syncs the config to the server. +The wizard saves configuration to `~/.data-hub/config.yaml`, the API key to `~/.data-hub/.env.`, and syncs the config to the server. ## Starting the watcher diff --git a/docs/watcher.md b/docs/watcher.md index 9d720bb0..83ffe457 100644 --- a/docs/watcher.md +++ b/docs/watcher.md @@ -37,11 +37,11 @@ uv run data-hub-watcher watch Interactive setup wizard that: 1. Prompts for the environment (`staging`, `production`, or `preview`). Choosing `preview` also prompts for a custom API base URL. -2. Prompts for an API key (or reads `DATA_HUB_API_KEY` from the environment). The key is saved to `~/.data-hub/.env`. +2. Prompts for an API key (or reads `DATA_HUB_API_KEY` from the environment). The key is saved to a per-environment file at `~/.data-hub/.env.` (e.g. `.env.staging`), so switching between environments later doesn't require re-entering it. 3. Fetches existing instruments from the API or registers a new one. 4. Prompts for the watch directory, file patterns, run detection pattern, stability period, and upload mode. 5. Registers the watcher with the API. -6. Saves the config to `~/.data-hub/config.yaml`, the API key to `~/.data-hub/.env`, and syncs the config to the API. +6. Saves the config to `~/.data-hub/config.yaml`, the API key to `~/.data-hub/.env.`, and syncs the config to the API. ### `watch` @@ -105,7 +105,7 @@ Manage the watcher as a Windows service: ## Configuration -The config file lives at `~/.data-hub/config.yaml` by default. Override with `--config` or the `DATA_HUB_CONFIG_PATH` environment variable. The API key is stored separately in `~/.data-hub/.env`. +The config file lives at `~/.data-hub/config.yaml` by default. Override with `--config` or the `DATA_HUB_CONFIG_PATH` environment variable. The API key is stored separately in `~/.data-hub/.env.` (e.g. `.env.staging`, `.env.production`, or `.env.preview`); the legacy `~/.data-hub/.env` is also loaded for backwards compatibility, with the per-environment file taking precedence. ### Config file format diff --git a/pyproject.toml b/pyproject.toml index 7e7ea12b..f7672de2 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -6,7 +6,7 @@ dev = [ "pre-commit>=4.2.0", "psycopg2-binary>=2.9.10", "pyright>=1.1.408", - "pytest>=8.3.5", + "pytest>=9.0.3", "ruff>=0.11.11", ] diff --git a/uv.lock b/uv.lock index ddeffd5c..cbef4dd8 100644 --- a/uv.lock +++ b/uv.lock @@ -25,7 +25,7 @@ dev = [ { name = "pre-commit", specifier = ">=4.2.0" }, { name = "psycopg2-binary", specifier = ">=2.9.10" }, { name = "pyright", specifier = ">=1.1.408" }, - { name = "pytest", specifier = ">=8.3.5" }, + { name = "pytest", specifier = ">=9.0.3" }, { name = "ruff", specifier = ">=0.11.11" }, ] @@ -1212,7 +1212,7 @@ wheels = [ [[package]] name = "pytest" -version = "9.0.2" +version = "9.0.3" source = { registry = "https://pypi.org/simple" } dependencies = [ { name = "colorama", marker = "sys_platform == 'win32'" }, @@ -1221,9 +1221,9 @@ dependencies = [ { name = "pluggy" }, { name = "pygments" }, ] -sdist = { url = "https://files.pythonhosted.org/packages/d1/db/7ef3487e0fb0049ddb5ce41d3a49c235bf9ad299b6a25d5780a89f19230f/pytest-9.0.2.tar.gz", hash = "sha256:75186651a92bd89611d1d9fc20f0b4345fd827c41ccd5c299a868a05d70edf11", size = 1568901, upload-time = "2025-12-06T21:30:51.014Z" } +sdist = { url = "https://files.pythonhosted.org/packages/7d/0d/549bd94f1a0a402dc8cf64563a117c0f3765662e2e668477624baeec44d5/pytest-9.0.3.tar.gz", hash = "sha256:b86ada508af81d19edeb213c681b1d48246c1a91d304c6c81a427674c17eb91c", size = 1572165, upload-time = "2026-04-07T17:16:18.027Z" } wheels = [ - { url = "https://files.pythonhosted.org/packages/3b/ab/b3226f0bd7cdcf710fbede2b3548584366da3b19b5021e74f5bde2a8fa3f/pytest-9.0.2-py3-none-any.whl", hash = "sha256:711ffd45bf766d5264d487b917733b453d917afd2b0ad65223959f59089f875b", size = 374801, upload-time = "2025-12-06T21:30:49.154Z" }, + { url = "https://files.pythonhosted.org/packages/d4/24/a372aaf5c9b7208e7112038812994107bc65a84cd00e0354a88c2c77a617/pytest-9.0.3-py3-none-any.whl", hash = "sha256:2c5efc453d45394fdd706ade797c0a81091eccd1d6e4bccfcd476e2b8e0ab5d9", size = 375249, upload-time = "2026-04-07T17:16:16.13Z" }, ] [[package]] diff --git a/watcher/src/data_hub_watcher/api_client.py b/watcher/src/data_hub_watcher/api_client.py index 52932a2c..b9381c22 100644 --- a/watcher/src/data_hub_watcher/api_client.py +++ b/watcher/src/data_hub_watcher/api_client.py @@ -1,6 +1,7 @@ from __future__ import annotations import logging import os +from datetime import datetime, timezone from typing import Any import requests @@ -203,12 +204,17 @@ def request_upload_url( filename: str, content_type: str | None = None, size_bytes: int | None = None, + file_created_at_ts: float | None = None, ) -> PresignedUploadResponse: payload: dict[str, Any] = {"filename": filename} if content_type: payload["content_type"] = content_type if size_bytes is not None: payload["size_bytes"] = size_bytes + if file_created_at_ts: + payload["file_created_at"] = datetime.fromtimestamp( + file_created_at_ts, tz=timezone.utc + ).isoformat() resp = self._request( "POST", f"/instruments/{instrument_id}/runs/{run_id}/request-upload-url", diff --git a/watcher/src/data_hub_watcher/cli.py b/watcher/src/data_hub_watcher/cli.py index 0cd692e3..f609df0c 100644 --- a/watcher/src/data_hub_watcher/cli.py +++ b/watcher/src/data_hub_watcher/cli.py @@ -21,6 +21,8 @@ DEFAULT_STABILITY_PERIOD_SECONDS, RUN_DETECTION_PRESETS, STATE_DB_FILENAME, + SUPPORTED_ENVIRONMENTS, + env_file_path, load_env, resolve_config_path, save_api_key, @@ -76,6 +78,49 @@ def _resolve_path(ctx: click.Context) -> Path: return resolve_config_path(ctx.obj.get("config_path")) +API_KEY_PREFIX = "dhub_" + +# Invisible characters that some Windows clipboards (Outlook, Teams, Word, etc.) +# silently inject when an operator copies an API key. Stripping them here +# avoids 401s caused by a hash mismatch on the server. +_INVISIBLE_CHARS = ( + "\u00a0", # non-breaking space + "\u200b", # zero-width space + "\u200c", # zero-width non-joiner + "\u200d", # zero-width joiner + "\ufeff", # BOM / zero-width no-break space +) + + +def _clean_api_key(value: str) -> str: + """Normalize and validate an API key entered by the operator. + + Pasting into a hidden ``click.prompt`` on Windows frequently introduces + stray whitespace (CR, LF, NBSP) or zero-width characters from rich-text + clipboards. We strip those defensively and then verify the value still + looks like a Data Hub PAT before any network call so the operator sees a + clear error instead of a confusing 401. + """ + cleaned = value + for ch in _INVISIBLE_CHARS: + cleaned = cleaned.replace(ch, "") + cleaned = cleaned.strip() + + if not cleaned: + raise click.ClickException("API key is empty.") + if any(c.isspace() for c in cleaned): + raise click.ClickException( + "API key contains whitespace. Re-copy the key — your clipboard " + "may have included a line break or non-breaking space." + ) + if not cleaned.startswith(API_KEY_PREFIX): + raise click.ClickException( + f"API key must start with '{API_KEY_PREFIX}'. Re-copy the key from " + "the Data Hub UI; the value may have been truncated on paste." + ) + return cleaned + + def _make_client( environment: str, api_key: str | None = None, api_base_url: str | None = None ) -> DataHubClient: @@ -92,6 +137,9 @@ def _load_and_client(ctx: click.Context) -> tuple[WatcherConfig, DataHubClient, """Load config and build a matching API client. Returns (config, client, path).""" path = _resolve_path(ctx) cfg = load_config(path) + # Overlay the env-specific file (e.g. ``.env.staging``) so the API key + # picked up by ``DataHubClient`` always matches the configured environment. + load_env(cfg.environment) client = _make_client(cfg.environment, api_base_url=cfg.api_base_url) return cfg, client, path @@ -115,8 +163,16 @@ def _setup_file_logging() -> None: @cli.command() +@click.option( + "--show-key", + is_flag=True, + help=( + "Echo the API key as it is typed/pasted. Useful on Windows terminals " + "where hidden input is unreliable for paste." + ), +) @click.pass_context -def init(ctx: click.Context) -> None: +def init(ctx: click.Context, show_key: bool) -> None: """Interactive setup wizard + API registration.""" path = _resolve_path(ctx) if path.exists(): @@ -126,7 +182,7 @@ def init(ctx: click.Context) -> None: # 1. Environment environment = click.prompt( "Environment", - type=click.Choice(["staging", "production", "preview"], case_sensitive=False), + type=click.Choice(list(SUPPORTED_ENVIRONMENTS), case_sensitive=False), ) api_base_url: str | None = None @@ -136,10 +192,24 @@ def init(ctx: click.Context) -> None: ) api_base_url = raw_url.rstrip("/") - # 2. API key - api_key = os.environ.get("DATA_HUB_API_KEY", "") - if not api_key: - api_key = click.prompt("DATA_HUB_API_KEY", hide_input=True) + # 2. API key — overlay any existing per-environment env file so the user + # doesn't have to re-enter a key they've already saved for this target. + load_env(environment) + existing_key = os.environ.get("DATA_HUB_API_KEY", "") + env_specific_path = env_file_path(environment) + hide_input = not show_key + if existing_key and env_specific_path.exists(): + click.echo(f"Found saved API key for {environment} at {env_specific_path}.") + if click.confirm("Use the saved key?", default=True): + api_key = existing_key + else: + api_key = click.prompt("DATA_HUB_API_KEY", hide_input=hide_input) + elif existing_key: + api_key = existing_key + else: + api_key = click.prompt("DATA_HUB_API_KEY", hide_input=hide_input) + + api_key = _clean_api_key(api_key) client = _make_client(environment, api_key=api_key, api_base_url=api_base_url) @@ -153,7 +223,7 @@ def init(ctx: click.Context) -> None: "The API key was not saved. Please re-run init with a valid key." ) from exc - env_path = save_api_key(api_key) + env_path = save_api_key(api_key, environment) click.echo(f"API key saved to {env_path}") if instruments: @@ -779,29 +849,27 @@ def _windows_only() -> None: "env_path_override", type=click.Path(dir_okay=False), default=None, - help="Path to the .env file. Defaults to ~/.data-hub/.env.", + help="Path to the .env file. Defaults to ~/.data-hub/.env..", ) @click.pass_context def service_install(ctx: click.Context, env_path_override: str | None) -> None: """Install the watcher as a Windows service.""" _windows_only() path = _resolve_path(ctx) - load_config(path) + cfg = load_config(path) from data_hub_watcher.service import install_service if env_path_override is not None: env_path = Path(env_path_override).resolve() else: - from data_hub_watcher.constants import DEFAULT_CONFIG_DIR, ENV_FILENAME - - env_path = (DEFAULT_CONFIG_DIR / ENV_FILENAME).resolve() + env_path = env_file_path(cfg.environment).resolve() if not env_path.exists(): click.echo( click.style( f"⚠ Warning: {env_path} does not exist. " - "Run 'data-hub-watcher login' first or pass --env-path.", + "Run 'data-hub-watcher init' first or pass --env-path.", fg="yellow", ), err=True, diff --git a/watcher/src/data_hub_watcher/constants.py b/watcher/src/data_hub_watcher/constants.py index aeeebfbc..13fe09a7 100644 --- a/watcher/src/data_hub_watcher/constants.py +++ b/watcher/src/data_hub_watcher/constants.py @@ -13,6 +13,7 @@ DEFAULT_CONFIG_DIR = Path("~/.data-hub").expanduser() DEFAULT_CONFIG_FILENAME = "config.yaml" ENV_FILENAME = ".env" +SUPPORTED_ENVIRONMENTS: tuple[str, ...] = ("staging", "production", "preview") HEARTBEAT_INTERVAL_SECONDS = 60 DEFAULT_STABILITY_PERIOD_SECONDS = 5 @@ -68,24 +69,47 @@ CONFIG_PATH_ENV_VAR = "DATA_HUB_CONFIG_PATH" -def load_env() -> None: - """Load ``~/.data-hub/.env`` into the process environment. +def env_file_path(environment: str | None = None) -> Path: + """Return the env file path for *environment* (or the base file if ``None``). - Existing environment variables take precedence (``override=False``), - so an explicit ``DATA_HUB_API_KEY`` export still wins. + Examples: + ``env_file_path()`` -> ``~/.data-hub/.env`` + ``env_file_path("staging")`` -> ``~/.data-hub/.env.staging`` """ - env_path = DEFAULT_CONFIG_DIR / ENV_FILENAME - load_dotenv(env_path) + if environment: + return DEFAULT_CONFIG_DIR / f"{ENV_FILENAME}.{environment}" + return DEFAULT_CONFIG_DIR / ENV_FILENAME -def save_api_key(api_key: str) -> Path: - """Persist *api_key* to ``~/.data-hub/.env`` and return the file path. +def load_env(environment: str | None = None) -> None: + """Load env files from ``~/.data-hub/`` into the process environment. + + Always loads the base ``.env`` first. If *environment* is provided, + overlays ``.env.`` on top so its values take precedence. + Existing process-level environment variables (e.g. an explicit + ``DATA_HUB_API_KEY`` export) still win over the base file but are + overridden by the env-specific file when one is supplied — this lets + operators switch environments simply by changing the config without + re-exporting their key. + """ + load_dotenv(env_file_path()) + if environment: + load_dotenv(env_file_path(environment), override=True) + + +def save_api_key(api_key: str, environment: str | None = None) -> Path: + """Persist *api_key* to the env file for *environment* and return its path. + + When *environment* is provided the key is written to + ``~/.data-hub/.env.`` so each deployment target keeps its + own credentials. Without *environment* the legacy ``~/.data-hub/.env`` + file is used. Preserves any other variables already present in the file and single-quotes the value to guard against special characters (``#``, ``=``, whitespace) that would confuse dotenv parsers. """ - env_path = DEFAULT_CONFIG_DIR / ENV_FILENAME + env_path = env_file_path(environment) env_path.parent.mkdir(parents=True, exist_ok=True) key_line = f"DATA_HUB_API_KEY='{api_key}'\n" diff --git a/watcher/src/data_hub_watcher/run_detector.py b/watcher/src/data_hub_watcher/run_detector.py index 4b59e8bd..17af3aff 100644 --- a/watcher/src/data_hub_watcher/run_detector.py +++ b/watcher/src/data_hub_watcher/run_detector.py @@ -8,9 +8,11 @@ from __future__ import annotations import logging +import os import re from collections.abc import Callable from dataclasses import dataclass, field +from datetime import datetime, timezone from pathlib import Path from data_hub_watcher.api_client import ApiError, DataHubClient @@ -21,6 +23,17 @@ logger = logging.getLogger(__name__) +def file_created_at(st: os.stat_result) -> float: + """Best-effort on-disk creation time for *st*. + + Prefers `st_birthtime` (macOS, Windows, BSD), which is the true + creation time. Falls back to `st_mtime` on platforms that don't + expose birthtime (most Linux filesystems through the legacy stat + interface) — this is the closest universally-available approximation. + """ + return getattr(st, "st_birthtime", None) or st.st_mtime + + @dataclass class FileInfo: """Metadata about a single detected file within a run.""" @@ -32,6 +45,11 @@ class FileInfo: # future restart's initial scan can cheaply stat-match against it # (see `StateDB.has_detected_stat_match`) and skip re-reporting. mtime: float = 0.0 + # On-disk creation time (st_birthtime when available, else mtime). + # Sent to the API and persisted in the state DB so it survives + # restarts and reflects the original creation time even after the + # file has been re-stat'd or partially modified. + file_created_at: float = 0.0 @dataclass @@ -118,7 +136,13 @@ def on_stable_file(self, path: Path) -> None: logger.warning("File disappeared before run detection: %s", path) return - info = FileInfo(path=path, filename=path.name, size_bytes=st.st_size, mtime=st.st_mtime) + info = FileInfo( + path=path, + filename=path.name, + size_bytes=st.st_size, + mtime=st.st_mtime, + file_created_at=file_created_at(st), + ) run = self._runs.get(run_id) if run is None: @@ -161,11 +185,16 @@ def _file_payload(self, info: FileInfo) -> dict[str, object]: relative_path = info.path.relative_to(self._watch_dir).as_posix() except ValueError: relative_path = info.filename - return { + payload: dict[str, object] = { "relative_path": relative_path, "filename": info.filename, "size_bytes": info.size_bytes, } + if info.file_created_at: + payload["file_created_at"] = datetime.fromtimestamp( + info.file_created_at, tz=timezone.utc + ).isoformat() + return payload def _relative_path(self, info: FileInfo) -> str: try: @@ -186,6 +215,7 @@ def _persist_detected_files(self, run: RunState) -> None: info.filename, info.size_bytes, info.mtime, + info.file_created_at or None, ) for info in run.files ] @@ -283,6 +313,7 @@ def hydrate_from_state_db(self) -> None: filename=rec.filename, size_bytes=rec.size_bytes, mtime=rec.mtime, + file_created_at=rec.file_created_at or 0.0, ) for rec in records ] diff --git a/watcher/src/data_hub_watcher/service.py b/watcher/src/data_hub_watcher/service.py index 297cd1fb..0829a8c4 100644 --- a/watcher/src/data_hub_watcher/service.py +++ b/watcher/src/data_hub_watcher/service.py @@ -17,14 +17,12 @@ from __future__ import annotations import logging import sys +import threading from pathlib import Path -from typing import TYPE_CHECKING, Any +from typing import Any from data_hub_watcher.constants import SERVICE_NAME -if TYPE_CHECKING: - pass - logger = logging.getLogger(__name__) SERVICE_DISPLAY_NAME = "Data Hub Watcher" @@ -34,6 +32,14 @@ _REG_CONFIG_PATH = "ConfigPath" _REG_ENV_PATH = "EnvPath" +# Windows services that must be running before the watcher can usefully +# contact the Data Hub API. Declaring these as dependencies makes the SCM +# wait for the TCP/IP stack and DNS resolver to be ready before it tries +# to start the watcher. Combined with delayed-auto-start (below), this +# avoids the classic "service starts at boot before the network is up, +# fails its API health check, and stays stopped" failure mode. +_SERVICE_DEPENDENCIES: list[str] = ["Tcpip", "Dnscache"] + def install_service(config_path: Path, env_path: Path) -> None: """Install the watcher as a Windows service with automatic start and recovery. @@ -46,6 +52,12 @@ def install_service(config_path: Path, env_path: Path) -> None: *config_path* and *env_path* are persisted to the service's registry key so that ``SvcDoRun`` can locate them regardless of which Windows user account the service runs under (typically Local System). + + The service is registered with ``delayedstart=True`` and a dependency + on the TCP/IP and DNS-client services so it does not start until the + network stack is up after a reboot. Without this, lab PCs frequently + boot the service before any NIC has DHCP-leased an address, the + initial API call fails, and the service exits. """ import win32service as ws # type: ignore[import-untyped] import win32serviceutil # type: ignore[import-untyped] @@ -57,9 +69,11 @@ def install_service(config_path: Path, env_path: Path) -> None: serviceName=SERVICE_NAME, displayName=SERVICE_DISPLAY_NAME, startType=ws.SERVICE_AUTO_START, + serviceDeps=_SERVICE_DEPENDENCIES, exeName=sys.executable, exeArgs="-m data_hub_watcher.service", description=SERVICE_DESCRIPTION, + delayedstart=True, ) _store_paths_in_registry(config_path, env_path) @@ -136,6 +150,13 @@ def _configure_recovery() -> None: After two consecutive failures the service stops retrying to avoid a crash loop (e.g. due to a persistent config or credential issue). The failure counter resets after 24 h of healthy uptime. + + We additionally set ``SERVICE_CONFIG_FAILURE_ACTIONS_FLAG`` so that + these recovery actions fire when ``SvcDoRun`` exits with a non-zero + code -- not only when the process actually crashes. Our startup + sequence reports controlled failures by raising ``SystemExit(1)``, + so without this flag the SCM would treat them as graceful stops and + never restart the service. """ import win32service as ws # type: ignore[import-untyped] @@ -159,6 +180,11 @@ def _configure_recovery() -> None: "Actions": actions, }, ) + ws.ChangeServiceConfig2( + hs, + ws.SERVICE_CONFIG_FAILURE_ACTIONS_FLAG, + {"fFailureActionsOnNonCrashFailures": True}, + ) finally: ws.CloseServiceHandle(hs) finally: @@ -238,6 +264,122 @@ def query_service_status() -> dict[str, Any]: ws.CloseServiceHandle(hscm) +def _run_service_loop(stop_event: threading.Event, sm: Any) -> None: + """Run the watcher service loop until *stop_event* is set. + + This is the testable body of ``DataHubWatcherService.SvcDoRun``. + Extracted as a top-level function so unit tests can exercise the + full startup sequence (registry read, env loading, API health + check, checksum sync, runtime build/start/stop) on any platform + by injecting a mock *sm* (servicemanager) and patching the + dependencies it pulls in. ``SvcDoRun`` itself becomes a thin + wrapper that imports ``servicemanager`` lazily and delegates here. + + Any controlled failure exits the process with a non-zero status + via ``raise SystemExit(1)``. Combined with the + ``SERVICE_CONFIG_FAILURE_ACTIONS_FLAG`` set by + ``_configure_recovery``, the SCM treats such exits as service + failures and runs the configured restart actions. This is what + allows the service to recover from a transient API error at boot + -- the most common reason a freshly-rebooted lab PC fails to + bring the watcher back up. + """ + import platform + + from dotenv import load_dotenv + + from data_hub_watcher.api_client import ApiError, DataHubClient + from data_hub_watcher.config_io import config_checksum, load_config + from data_hub_watcher.constants import ( + API_URLS, + STATE_DB_FILENAME, + env_file_path, + ) + from data_hub_watcher.runtime import build_runtime, start_runtime, stop_runtime + + sm.LogInfoMsg(f"{SERVICE_DISPLAY_NAME} starting") + + try: + path, env_path = _read_paths_from_registry() + except Exception as exc: + sm.LogErrorMsg( + f"Cannot read config/env paths from registry: {exc}. " + "Re-run 'data-hub-watcher service install'." + ) + raise SystemExit(1) from exc + + # Mirror the CLI's ``load_env`` semantics: load the base + # ``~/.data-hub/.env`` first (for any shared, non-secret values + # an operator may keep there), then overlay the registered env + # file (typically ``.env.``, or a custom path + # supplied via ``service install --env-path``) so its values win. + # Without the base-file load, an operator that splits shared + # config from per-environment secrets would see the service + # silently miss the shared half. + base_env = env_file_path() + if base_env != env_path and base_env.exists(): + load_dotenv(base_env) + load_dotenv(env_path, override=True) + cfg = load_config(path) + inst = cfg.instrument + + if cfg.environment == "preview": + # WatcherConfig's model validator guarantees api_base_url is + # set whenever environment is "preview"; the assertion is here + # to make that invariant visible to pyright. + assert cfg.api_base_url is not None + base_url = cfg.api_base_url + else: + base_url = API_URLS[cfg.environment] + client = DataHubClient(base_url) + + # Step 1: Check instrument status (mirrors CLI watch startup) + try: + detail = client.get_instrument(inst.id) + except ApiError as exc: + sm.LogErrorMsg(f"Cannot reach API during startup: {exc.message}") + raise SystemExit(1) from exc + + if detail.status == "pending": + sm.LogErrorMsg( + f"Instrument {inst.id!r} is still pending activation. " + "Service cannot start until the instrument is activated." + ) + raise SystemExit(1) + + # Step 2: Sync config checksum + if cfg.watcher_id: + local_cs = config_checksum(path) + try: + remote = client.get_config_checksum(cfg.watcher_id) + if remote is None or remote.config_checksum != local_cs: + yaml_content = path.read_text(encoding="utf-8") + client.push_config(cfg.watcher_id, yaml_content, local_cs) + sm.LogInfoMsg("Config synced to Data Hub") + except ApiError as exc: + sm.LogWarningMsg(f"Could not sync config to API: {exc.message}") + + # build_runtime asserts cfg.watcher_id is set; surface that as + # a service-manager error rather than a hard crash so operators + # see a clear message in the Windows event log. + if not cfg.watcher_id: + sm.LogErrorMsg("No watcher_id in config. Run 'data-hub-watcher init' first.") + raise SystemExit(1) + + db_path = path.parent / STATE_DB_FILENAME + rt = build_runtime(client=client, cfg=cfg, db_path=db_path) + + start_runtime(rt, started_message=f"Service started on {platform.node()}") + + sm.LogInfoMsg(f"{SERVICE_DISPLAY_NAME} is running") + + stop_event.wait() + + stop_runtime(rt, stopped_message="Service stopped") + + sm.LogInfoMsg(f"{SERVICE_DISPLAY_NAME} stopped") + + def _create_service_class() -> type | None: """Dynamically create the ServiceFramework subclass on Windows only. @@ -258,95 +400,12 @@ class DataHubWatcherService(win32serviceutil.ServiceFramework): # type: ignore[ def __init__(self, args: list[str]) -> None: super().__init__(args) - import threading - self._stop_event = threading.Event() def SvcDoRun(self) -> None: - import platform - import servicemanager # type: ignore[import-untyped] - from dotenv import load_dotenv - from data_hub_watcher.api_client import ApiError, DataHubClient - from data_hub_watcher.config_io import config_checksum, load_config - from data_hub_watcher.constants import ( - API_URLS, - STATE_DB_FILENAME, - ) - from data_hub_watcher.runtime import ( - build_runtime, - start_runtime, - stop_runtime, - ) - - servicemanager.LogInfoMsg(f"{SERVICE_DISPLAY_NAME} starting") - - try: - path, env_path = _read_paths_from_registry() - except Exception as exc: - servicemanager.LogErrorMsg( - f"Cannot read config/env paths from registry: {exc}. " - "Re-run 'data-hub-watcher service install'." - ) - return - - load_dotenv(env_path) - cfg = load_config(path) - inst = cfg.instrument - - if cfg.environment == "preview": - base_url = cfg.api_base_url - else: - base_url = API_URLS[cfg.environment] - client = DataHubClient(base_url) - - # Step 1: Check instrument status (mirrors CLI watch startup) - try: - detail = client.get_instrument(inst.id) - if detail.status == "pending": - servicemanager.LogErrorMsg( - f"Instrument {inst.id!r} is still pending activation. " - "Service cannot start until the instrument is activated." - ) - return - except ApiError as exc: - servicemanager.LogErrorMsg(f"Cannot reach API during startup: {exc.message}") - return - - # Step 2: Sync config checksum - if cfg.watcher_id: - local_cs = config_checksum(path) - try: - remote = client.get_config_checksum(cfg.watcher_id) - if remote is None or remote.config_checksum != local_cs: - yaml_content = path.read_text(encoding="utf-8") - client.push_config(cfg.watcher_id, yaml_content, local_cs) - servicemanager.LogInfoMsg("Config synced to Data Hub") - except ApiError as exc: - servicemanager.LogWarningMsg(f"Could not sync config to API: {exc.message}") - - # build_runtime asserts cfg.watcher_id is set; surface that as - # a service-manager error rather than a hard crash so operators - # see a clear message in the Windows event log. - if not cfg.watcher_id: - servicemanager.LogErrorMsg( - "No watcher_id in config. Run 'data-hub-watcher init' first." - ) - return - - db_path = path.parent / STATE_DB_FILENAME - rt = build_runtime(client=client, cfg=cfg, db_path=db_path) - - start_runtime(rt, started_message=f"Service started on {platform.node()}") - - servicemanager.LogInfoMsg(f"{SERVICE_DISPLAY_NAME} is running") - - self._stop_event.wait() - - stop_runtime(rt, stopped_message="Service stopped") - - servicemanager.LogInfoMsg(f"{SERVICE_DISPLAY_NAME} stopped") + _run_service_loop(self._stop_event, servicemanager) def SvcStop(self) -> None: self.ReportServiceStatus(ws.SERVICE_STOP_PENDING) diff --git a/watcher/src/data_hub_watcher/state.py b/watcher/src/data_hub_watcher/state.py index 18da6c49..740702d7 100644 --- a/watcher/src/data_hub_watcher/state.py +++ b/watcher/src/data_hub_watcher/state.py @@ -32,6 +32,9 @@ class DetectedFileRecord: filename: str size_bytes: int mtime: float + # On-disk creation time at the moment the file was reported. NULL for + # rows persisted before this column was added (legacy state DBs). + file_created_at: float | None = None class StateDB: @@ -107,6 +110,12 @@ def _create_tables(self) -> None: "CREATE INDEX IF NOT EXISTS idx_uploaded_files_stat " "ON uploaded_files (relative_path, size_bytes, mtime)" ) + # detected_files predates the file_created_at column; add it as + # nullable so legacy rows survive the migration. New rows always + # populate it (record_detected_files passes the value through). + detected_cols = {row[1] for row in self._conn.execute("PRAGMA table_info(detected_files)")} + if "file_created_at" not in detected_cols: + self._conn.execute("ALTER TABLE detected_files ADD COLUMN file_created_at REAL") self._conn.commit() # ------------------------------------------------------------------ @@ -186,26 +195,27 @@ def record_upload( def record_detected_files( self, run_id: str, - files: Iterable[tuple[str, str, int, float]], + files: Iterable[tuple[str, str, int, float, float | None]], ) -> None: """Persist the file manifest for a reported run. *files* is an iterable of `(relative_path, filename, size_bytes, - mtime)` tuples. Rows are upserted so repeated calls for the same - run (e.g. as more files stabilise and PATCHes are issued) keep - the table consistent with the in-memory `RunState`. + mtime, file_created_at)` tuples. Rows are upserted so repeated + calls for the same run (e.g. as more files stabilise and PATCHes + are issued) keep the table consistent with the in-memory + `RunState`. *file_created_at* may be `None` for legacy callers. """ rows = [ - (run_id, rel_path, filename, size_bytes, mtime) - for rel_path, filename, size_bytes, mtime in files + (run_id, rel_path, filename, size_bytes, mtime, file_created_at) + for rel_path, filename, size_bytes, mtime, file_created_at in files ] if not rows: return with self._lock: self._conn.executemany( "INSERT OR REPLACE INTO detected_files " - "(run_id, relative_path, filename, size_bytes, mtime) " - "VALUES (?, ?, ?, ?, ?)", + "(run_id, relative_path, filename, size_bytes, mtime, file_created_at) " + "VALUES (?, ?, ?, ?, ?, ?)", rows, ) self._conn.commit() @@ -234,7 +244,7 @@ def get_detected_files_for_run(self, run_id: str) -> list[DetectedFileRecord]: """Return the persisted file manifest for *run_id*, ordered by path.""" with self._lock: cur = self._conn.execute( - "SELECT relative_path, filename, size_bytes, mtime " + "SELECT relative_path, filename, size_bytes, mtime, file_created_at " "FROM detected_files WHERE run_id = ? ORDER BY relative_path", (run_id,), ) @@ -245,6 +255,7 @@ def get_detected_files_for_run(self, run_id: str) -> list[DetectedFileRecord]: filename=row[1], size_bytes=row[2], mtime=row[3], + file_created_at=row[4], ) for row in rows ] diff --git a/watcher/src/data_hub_watcher/uploader.py b/watcher/src/data_hub_watcher/uploader.py index 118718b4..70394aee 100644 --- a/watcher/src/data_hub_watcher/uploader.py +++ b/watcher/src/data_hub_watcher/uploader.py @@ -22,7 +22,7 @@ ) from data_hub_watcher.events import EventReporter, EventType, WatcherEvent from data_hub_watcher.heartbeat import WatcherCounters -from data_hub_watcher.run_detector import FileInfo +from data_hub_watcher.run_detector import FileInfo, file_created_at from data_hub_watcher.state import StateDB from data_hub_watcher.util import file_sha256 @@ -188,6 +188,7 @@ def _upload_single(self, path: Path, run_id: str) -> bool: path.name, content_type=content_type, size_bytes=stat.st_size, + file_created_at_ts=file_created_at(stat), ) except ApiError as exc: logger.error("Failed to get presigned URL for %s: %s", path.name, exc.message) diff --git a/watcher/tests/test_monitor_initial_scan.py b/watcher/tests/test_monitor_initial_scan.py index 95ffd431..ebd36a86 100644 --- a/watcher/tests/test_monitor_initial_scan.py +++ b/watcher/tests/test_monitor_initial_scan.py @@ -239,7 +239,7 @@ def test_skips_file_recorded_in_detected_files( st = f.stat() state_db.record_detected_files( "run-42", - [("run-42.nd2", "run-42.nd2", st.st_size, st.st_mtime)], + [("run-42.nd2", "run-42.nd2", st.st_size, st.st_mtime, st.st_mtime)], ) monitor = _make_monitor(watch_dir, state_db) @@ -266,7 +266,15 @@ def test_skips_union_of_uploaded_and_detected(self, state_db: StateDB, watch_dir st_det = detected.stat() state_db.record_detected_files( "run-1", - [("detected.nd2", "detected.nd2", st_det.st_size, st_det.st_mtime)], + [ + ( + "detected.nd2", + "detected.nd2", + st_det.st_size, + st_det.st_mtime, + st_det.st_mtime, + ) + ], ) monitor = _make_monitor(watch_dir, state_db) @@ -317,21 +325,27 @@ def test_record_and_lookup_roundtrip(self, state_db: StateDB) -> None: state_db.record_detected_files( "run-1", [ - ("a/one.nd2", "one.nd2", 1024, 1_700_000_000.0), - ("a/two.nd2", "two.nd2", 2048, 1_700_000_001.0), + ("a/one.nd2", "one.nd2", 1024, 1_700_000_000.0, 1_699_999_900.0), + ("a/two.nd2", "two.nd2", 2048, 1_700_000_001.0, None), ], ) records = state_db.get_detected_files_for_run("run-1") - assert [(r.relative_path, r.filename, r.size_bytes, r.mtime) for r in records] == [ - ("a/one.nd2", "one.nd2", 1024, 1_700_000_000.0), - ("a/two.nd2", "two.nd2", 2048, 1_700_000_001.0), + assert [ + (r.relative_path, r.filename, r.size_bytes, r.mtime, r.file_created_at) for r in records + ] == [ + ("a/one.nd2", "one.nd2", 1024, 1_700_000_000.0, 1_699_999_900.0), + ("a/two.nd2", "two.nd2", 2048, 1_700_000_001.0, None), ] def test_record_is_idempotent_upsert(self, state_db: StateDB) -> None: """Re-recording the same (run_id, relative_path) must upsert, not dup.""" - state_db.record_detected_files("run-1", [("a.nd2", "a.nd2", 1024, 1_700_000_000.0)]) - state_db.record_detected_files("run-1", [("a.nd2", "a.nd2", 2048, 1_700_000_050.0)]) + state_db.record_detected_files( + "run-1", [("a.nd2", "a.nd2", 1024, 1_700_000_000.0, 1_700_000_000.0)] + ) + state_db.record_detected_files( + "run-1", [("a.nd2", "a.nd2", 2048, 1_700_000_050.0, 1_700_000_000.0)] + ) records = state_db.get_detected_files_for_run("run-1") assert len(records) == 1 @@ -344,17 +358,21 @@ def test_record_empty_iterable_is_noop(self, state_db: StateDB) -> None: assert state_db.get_reported_run_ids_with_files() == [] def test_has_detected_stat_match_hit(self, state_db: StateDB) -> None: - state_db.record_detected_files("run-1", [("x.nd2", "x.nd2", 1024, 1_700_000_000.0)]) + state_db.record_detected_files( + "run-1", [("x.nd2", "x.nd2", 1024, 1_700_000_000.0, 1_700_000_000.0)] + ) assert state_db.has_detected_stat_match("x.nd2", 1024, 1_700_000_000.0) is True def test_has_detected_stat_match_respects_mtime_tolerance(self, state_db: StateDB) -> None: - state_db.record_detected_files("run-1", [("x.nd2", "x.nd2", 1024, 1_700_000_000.0)]) + state_db.record_detected_files( + "run-1", [("x.nd2", "x.nd2", 1024, 1_700_000_000.0, 1_700_000_000.0)] + ) assert state_db.has_detected_stat_match("x.nd2", 1024, 1_700_000_000.5) is True assert state_db.has_detected_stat_match("x.nd2", 1024, 1_700_000_100.0) is False def test_has_detected_stat_match_miss_on_different_path(self, state_db: StateDB) -> None: state_db.record_detected_files( - "run-a", [("run-a/out.nd2", "out.nd2", 1024, 1_700_000_000.0)] + "run-a", [("run-a/out.nd2", "out.nd2", 1024, 1_700_000_000.0, 1_700_000_000.0)] ) assert state_db.has_detected_stat_match("run-b/out.nd2", 1024, 1_700_000_000.0) is False @@ -362,11 +380,13 @@ def test_get_reported_run_ids_with_files_distinct_and_scoped(self, state_db: Sta state_db.record_detected_files( "run-1", [ - ("run-1/a.nd2", "a.nd2", 1, 1_700_000_000.0), - ("run-1/b.nd2", "b.nd2", 2, 1_700_000_001.0), + ("run-1/a.nd2", "a.nd2", 1, 1_700_000_000.0, 1_700_000_000.0), + ("run-1/b.nd2", "b.nd2", 2, 1_700_000_001.0, 1_700_000_001.0), ], ) - state_db.record_detected_files("run-2", [("run-2/a.nd2", "a.nd2", 3, 1_700_000_002.0)]) + state_db.record_detected_files( + "run-2", [("run-2/a.nd2", "a.nd2", 3, 1_700_000_002.0, 1_700_000_002.0)] + ) state_db.record_run_reported("run-legacy") ids = state_db.get_reported_run_ids_with_files() diff --git a/watcher/tests/test_run_detector_hydration.py b/watcher/tests/test_run_detector_hydration.py index 2f87b340..1460b905 100644 --- a/watcher/tests/test_run_detector_hydration.py +++ b/watcher/tests/test_run_detector_hydration.py @@ -14,7 +14,7 @@ import pytest -from data_hub_watcher.run_detector import RunDetector +from data_hub_watcher.run_detector import RunDetector, file_created_at from data_hub_watcher.state import StateDB @@ -50,13 +50,39 @@ def _make_detector( ) +class TestFileCreatedAtHelper: + """Unit tests for the platform-portable `file_created_at` helper.""" + + def test_prefers_st_birthtime_when_present(self) -> None: + st = MagicMock(st_birthtime=1_700_000_100.0, st_mtime=1_700_000_200.0) + assert file_created_at(st) == 1_700_000_100.0 + + def test_falls_back_to_st_mtime_when_birthtime_missing(self) -> None: + # Linux stat results don't expose st_birthtime; spec= constrains the + # mock to exactly the attributes a real stat_result would have. + import os + + spec_attrs = [a for a in dir(os.stat_result) if a.startswith("st_")] + # Force-remove st_birthtime from the spec so getattr() returns None. + if "st_birthtime" in spec_attrs: + spec_attrs.remove("st_birthtime") + st = MagicMock(spec=spec_attrs) + st.st_mtime = 1_700_000_500.0 + assert file_created_at(st) == 1_700_000_500.0 + + def test_falls_back_when_st_birthtime_is_zero(self) -> None: + """A zero birthtime (some FUSE / network filesystems) should fall through.""" + st = MagicMock(st_birthtime=0.0, st_mtime=1_700_000_777.0) + assert file_created_at(st) == 1_700_000_777.0 + + class TestHydrateFromStateDb: def test_populates_runs_with_reported_true(self, state_db: StateDB, watch_dir: Path) -> None: state_db.record_detected_files( "run-alpha", [ - ("run-alpha/a.nd2", "a.nd2", 1024, 1_700_000_000.0), - ("run-alpha/b.nd2", "b.nd2", 2048, 1_700_000_001.0), + ("run-alpha/a.nd2", "a.nd2", 1024, 1_700_000_000.0, 1_699_999_900.0), + ("run-alpha/b.nd2", "b.nd2", 2048, 1_700_000_001.0, None), ], ) @@ -74,6 +100,10 @@ def test_populates_runs_with_reported_true(self, state_db: StateDB, watch_dir: P ] assert run.files[0].size_bytes == 1024 assert run.files[0].mtime == pytest.approx(1_700_000_000.0) + assert run.files[0].file_created_at == pytest.approx(1_699_999_900.0) + # Legacy rows with NULL file_created_at hydrate to 0.0 so the wire + # payload omits the field rather than emitting a bogus epoch time. + assert run.files[1].file_created_at == 0.0 def test_skips_runs_without_detected_files(self, state_db: StateDB, watch_dir: Path) -> None: """Legacy runs (in `runs` but not `detected_files`) must not hydrate. @@ -90,7 +120,9 @@ def test_skips_runs_without_detected_files(self, state_db: StateDB, watch_dir: P assert detector._runs == {} def test_hydration_is_idempotent(self, state_db: StateDB, watch_dir: Path) -> None: - state_db.record_detected_files("run-1", [("run-1/a.nd2", "a.nd2", 10, 1_700_000_000.0)]) + state_db.record_detected_files( + "run-1", [("run-1/a.nd2", "a.nd2", 10, 1_700_000_000.0, 1_700_000_000.0)] + ) detector = _make_detector(watch_dir, state_db) detector.hydrate_from_state_db() @@ -114,7 +146,15 @@ def test_new_file_patches_instead_of_posting(self, state_db: StateDB, watch_dir: st = existing.stat() state_db.record_detected_files( "run-42", - [("run-42/existing.nd2", "existing.nd2", st.st_size, st.st_mtime)], + [ + ( + "run-42/existing.nd2", + "existing.nd2", + st.st_size, + st.st_mtime, + st.st_mtime, + ) + ], ) client = MagicMock() @@ -142,7 +182,9 @@ def test_update_run_persists_new_manifest(self, state_db: StateDB, watch_dir: Pa existing = watch_dir / "run-7" / "a.nd2" existing.write_bytes(b"x" * 1024) st = existing.stat() - state_db.record_detected_files("run-7", [("run-7/a.nd2", "a.nd2", st.st_size, st.st_mtime)]) + state_db.record_detected_files( + "run-7", [("run-7/a.nd2", "a.nd2", st.st_size, st.st_mtime, st.st_mtime)] + ) client = MagicMock() detector = _make_detector(watch_dir, state_db, client=client) @@ -173,6 +215,33 @@ def test_successful_post_records_detected_files( assert [r.relative_path for r in records] == ["run-new/first.nd2"] assert records[0].filename == "first.nd2" assert records[0].size_bytes == 42 + # On-disk creation time is captured and persisted alongside the + # rest of the manifest (st_birthtime where supported, mtime + # fallback otherwise — both produce a positive float here). + assert records[0].file_created_at is not None + assert records[0].file_created_at > 0 + + def test_post_payload_includes_file_created_at( + self, state_db: StateDB, watch_dir: Path + ) -> None: + """The wire payload sent to POST /runs must include file_created_at as ISO 8601.""" + client = MagicMock() + client.report_run.return_value = MagicMock(id="api-run-id") + detector = _make_detector(watch_dir, state_db, client=client) + + (watch_dir / "run-iso").mkdir() + f = watch_dir / "run-iso" / "data.nd2" + f.write_bytes(b"q" * 8) + detector.on_stable_file(f) + + client.report_run.assert_called_once() + _, payload = client.report_run.call_args.args + detected_files = payload["detected_files"] + assert len(detected_files) == 1 + # ISO 8601 UTC string with timezone offset (e.g. "+00:00"). + iso = detected_files[0]["file_created_at"] + assert isinstance(iso, str) + assert iso.endswith("+00:00") def test_manifest_not_recorded_if_post_fails(self, state_db: StateDB, watch_dir: Path) -> None: from data_hub_watcher.api_client import ApiError diff --git a/watcher/tests/test_service.py b/watcher/tests/test_service.py new file mode 100644 index 00000000..ebd4e3ab --- /dev/null +++ b/watcher/tests/test_service.py @@ -0,0 +1,671 @@ +"""Unit tests for the Windows service module. + +The service module imports ``win32service``, ``win32serviceutil``, +``winreg``, and ``servicemanager`` lazily inside each function so it can +be imported on any platform. These tests exploit that by injecting +``MagicMock`` stand-ins via ``sys.modules`` *before* importing +``data_hub_watcher.service``, then exercise: + +* the SCM-facing helpers (``install_service``, ``uninstall_service``, + ``start_service``, ``stop_service``, ``_configure_recovery``, + ``query_service_status``) by asserting on the arguments passed to the + fake win32 modules; and +* the testable startup body ``_run_service_loop`` by patching the + side-effecting collaborators (config loader, ``DataHubClient``, + ``build_runtime`` / ``start_runtime`` / ``stop_runtime``) and a fake + ``servicemanager`` object. + +This locks in regressions of the kind that have historically shipped to +production lab PCs (delayed-start + non-crash failure flags, network +dependencies, manual-mode wiring) without needing a Windows runner. +""" + +from __future__ import annotations +import importlib +import sys +import threading +from collections.abc import Iterator +from pathlib import Path +from types import ModuleType +from typing import Any +from unittest.mock import MagicMock + +import pytest + +from data_hub_watcher.api_client import ApiError +from data_hub_watcher.models import ( + InstrumentConfig, + InstrumentDetailResponse, + RunDetectionConfig, + WatcherConfig, +) + +# Real winreg.REG_SZ and win32service.SERVICE_RUNNING values. Using the +# real numeric values lets ``query_service_status`` look them up in its +# state_map keyed on those constants without having to special-case the +# fakes. +_REG_SZ = 1 +_SERVICE_RUNNING = 4 +_SERVICE_STOPPED = 1 + + +def _make_win32_fakes() -> dict[str, ModuleType]: + """Build ``sys.modules`` stand-ins for the four win32 modules. + + Each returned mock pre-populates the constants the service module + actually reads (e.g. ``ws.SERVICE_AUTO_START``, ``winreg.REG_SZ``) + with concrete values so callers can assert on them. + """ + winreg = MagicMock(name="winreg") + winreg.HKEY_LOCAL_MACHINE = "HKLM" + winreg.KEY_SET_VALUE = "KEY_SET_VALUE" + winreg.KEY_QUERY_VALUE = "KEY_QUERY_VALUE" + winreg.REG_SZ = _REG_SZ + + ws = MagicMock(name="win32service") + ws.SERVICE_AUTO_START = "SERVICE_AUTO_START" + ws.SC_MANAGER_ALL_ACCESS = "SC_MANAGER_ALL_ACCESS" + ws.SC_MANAGER_CONNECT = "SC_MANAGER_CONNECT" + ws.SERVICE_ALL_ACCESS = "SERVICE_ALL_ACCESS" + ws.SERVICE_QUERY_STATUS = "SERVICE_QUERY_STATUS" + ws.SERVICE_CONFIG_FAILURE_ACTIONS = "SERVICE_CONFIG_FAILURE_ACTIONS" + ws.SERVICE_CONFIG_FAILURE_ACTIONS_FLAG = "SERVICE_CONFIG_FAILURE_ACTIONS_FLAG" + ws.SERVICE_STOPPED = _SERVICE_STOPPED + ws.SERVICE_START_PENDING = 2 + ws.SERVICE_STOP_PENDING = 3 + ws.SERVICE_RUNNING = _SERVICE_RUNNING + ws.SERVICE_CONTINUE_PENDING = 5 + ws.SERVICE_PAUSE_PENDING = 6 + ws.SERVICE_PAUSED = 7 + + win32serviceutil = MagicMock(name="win32serviceutil") + servicemanager = MagicMock(name="servicemanager") + + return { + "winreg": winreg, + "win32service": ws, + "win32serviceutil": win32serviceutil, + "servicemanager": servicemanager, + } + + +@pytest.fixture +def service_module(monkeypatch: pytest.MonkeyPatch) -> Iterator[ModuleType]: + """Reload ``data_hub_watcher.service`` with fresh win32 fakes. + + Each test gets its own set of fakes so call counts and recorded + arguments are isolated. The fakes are injected into ``sys.modules`` + before the reload so that the lazy ``import`` statements inside + each function pick them up. + """ + fakes = _make_win32_fakes() + for name, mod in fakes.items(): + monkeypatch.setitem(sys.modules, name, mod) + + import data_hub_watcher.service as svc + + reloaded = importlib.reload(svc) + yield reloaded + # Reload once more on teardown with the real (or absent) win32 + # modules popped so other tests don't see stale state. monkeypatch + # automatically restores sys.modules. + importlib.reload(svc) + + +# --- Registry round-trip + helpers ------------------------------------------ + + +class TestRegistryRoundTrip: + """``_store_paths_in_registry`` + ``_read_paths_from_registry`` form a pair.""" + + def test_store_then_read_returns_same_paths(self, service_module: ModuleType) -> None: + winreg = sys.modules["winreg"] + config_path = Path("C:/data-hub/config.yaml") + env_path = Path("C:/data-hub/.env.staging") + + # Build a fake registry key that records writes and replays them + # for the corresponding read. + stored: dict[str, str] = {} + fake_key = MagicMock(name="reg_key") + winreg.OpenKey.return_value = fake_key + + def _set(key: Any, name: str, _reserved: int, _type: int, value: str) -> None: + stored[name] = value + + def _query(key: Any, name: str) -> tuple[str, int]: + return stored[name], _REG_SZ + + winreg.SetValueEx.side_effect = _set + winreg.QueryValueEx.side_effect = _query + + service_module._store_paths_in_registry(config_path, env_path) + got_cfg, got_env = service_module._read_paths_from_registry() + + assert got_cfg == config_path + assert got_env == env_path + + def test_store_writes_under_hklm_with_set_value_access( + self, service_module: ModuleType + ) -> None: + winreg = sys.modules["winreg"] + winreg.OpenKey.return_value = MagicMock() + + service_module._store_paths_in_registry(Path("c.yaml"), Path("e.env")) + + winreg.OpenKey.assert_called_once_with( + "HKLM", + service_module._REG_KEY, + 0, + "KEY_SET_VALUE", + ) + names_written = [call.args[1] for call in winreg.SetValueEx.call_args_list] + assert names_written == [ + service_module._REG_CONFIG_PATH, + service_module._REG_ENV_PATH, + ] + # Both writes use the string registry value type. + for call in winreg.SetValueEx.call_args_list: + assert call.args[3] == _REG_SZ + + def test_delete_paths_silently_ignores_missing_key(self, service_module: ModuleType) -> None: + winreg = sys.modules["winreg"] + winreg.OpenKey.side_effect = OSError("key not found") + + # Must not raise even if the key doesn't exist. + service_module._delete_paths_from_registry() + + winreg.DeleteValue.assert_not_called() + + def test_delete_paths_removes_both_values(self, service_module: ModuleType) -> None: + winreg = sys.modules["winreg"] + winreg.OpenKey.return_value = MagicMock() + + service_module._delete_paths_from_registry() + + deleted_names = [call.args[1] for call in winreg.DeleteValue.call_args_list] + assert deleted_names == [ + service_module._REG_CONFIG_PATH, + service_module._REG_ENV_PATH, + ] + + +# --- install_service argument shape ----------------------------------------- + + +class TestInstallService: + """``install_service`` must register the service with very specific kwargs. + + The exact shape is what makes a freshly-rebooted lab PC bring the + watcher back up reliably (delayed start + network deps + recovery). + Regressions here have repeatedly broken production. + """ + + def test_install_passes_expected_kwargs_to_win32serviceutil( + self, service_module: ModuleType + ) -> None: + win32serviceutil = sys.modules["win32serviceutil"] + sys.modules["winreg"].OpenKey.return_value = MagicMock() + + service_module.install_service(Path("c.yaml"), Path("e.env")) + + win32serviceutil.InstallService.assert_called_once() + kwargs = win32serviceutil.InstallService.call_args.kwargs + assert kwargs["pythonClassString"] == ("data_hub_watcher.service.DataHubWatcherService") + assert kwargs["serviceName"] == service_module.SERVICE_NAME + assert kwargs["displayName"] == service_module.SERVICE_DISPLAY_NAME + assert kwargs["startType"] == "SERVICE_AUTO_START" + assert kwargs["serviceDeps"] == ["Tcpip", "Dnscache"] + assert kwargs["exeArgs"] == "-m data_hub_watcher.service" + assert kwargs["delayedstart"] is True + assert kwargs["description"] == service_module.SERVICE_DESCRIPTION + + def test_install_persists_paths_and_configures_recovery( + self, + service_module: ModuleType, + monkeypatch: pytest.MonkeyPatch, + ) -> None: + sys.modules["winreg"].OpenKey.return_value = MagicMock() + store_calls: list[tuple[Path, Path]] = [] + recovery_calls: list[None] = [] + + monkeypatch.setattr( + service_module, + "_store_paths_in_registry", + lambda p, e: store_calls.append((p, e)), + ) + monkeypatch.setattr( + service_module, + "_configure_recovery", + lambda: recovery_calls.append(None), + ) + + service_module.install_service(Path("c.yaml"), Path("e.env")) + + assert store_calls == [(Path("c.yaml"), Path("e.env"))] + assert recovery_calls == [None] + + +# --- _configure_recovery actions + non-crash failure flag -------------------- + + +class TestConfigureRecovery: + """Lock in the recovery contract added in commit 1712c70. + + The watcher's two-restart-then-stop policy + the + ``fFailureActionsOnNonCrashFailures`` flag are what make a lab PC + survive a transient API outage at boot. Without the flag, the SCM + treats our ``SystemExit(1)`` as a graceful stop. + """ + + def test_configure_recovery_sets_two_restarts_then_no_action( + self, service_module: ModuleType + ) -> None: + ws = sys.modules["win32service"] + fake_scm = MagicMock(name="scm_handle") + fake_svc = MagicMock(name="service_handle") + ws.OpenSCManager.return_value = fake_scm + ws.OpenService.return_value = fake_svc + + service_module._configure_recovery() + + ws.OpenSCManager.assert_called_once_with(None, None, "SC_MANAGER_ALL_ACCESS") + ws.OpenService.assert_called_once_with( + fake_scm, service_module.SERVICE_NAME, "SERVICE_ALL_ACCESS" + ) + + # Two ChangeServiceConfig2 calls: one for the actions, one for the flag. + assert ws.ChangeServiceConfig2.call_count == 2 + actions_call, flag_call = ws.ChangeServiceConfig2.call_args_list + + # First call: failure-actions list. + assert actions_call.args[1] == "SERVICE_CONFIG_FAILURE_ACTIONS" + actions_payload = actions_call.args[2] + assert actions_payload["ResetPeriod"] == 86400 + assert actions_payload["Actions"] == [ + (1, 60_000), + (1, 120_000), + (0, 0), + ] + + # Second call: the non-crash failure flag must be enabled. This + # is the bit that makes the SCM honour our SystemExit(1) as a + # failure and trigger the actions above. + assert flag_call.args[1] == "SERVICE_CONFIG_FAILURE_ACTIONS_FLAG" + assert flag_call.args[2] == {"fFailureActionsOnNonCrashFailures": True} + + def test_configure_recovery_closes_handles(self, service_module: ModuleType) -> None: + ws = sys.modules["win32service"] + fake_scm = MagicMock() + fake_svc = MagicMock() + ws.OpenSCManager.return_value = fake_scm + ws.OpenService.return_value = fake_svc + + service_module._configure_recovery() + + # Both handles must be closed even on the happy path. + closed = [call.args[0] for call in ws.CloseServiceHandle.call_args_list] + assert fake_svc in closed + assert fake_scm in closed + + +# --- uninstall / start / stop ------------------------------------------------ + + +class TestServiceLifecycle: + def test_uninstall_stops_clears_registry_and_removes( + self, + service_module: ModuleType, + monkeypatch: pytest.MonkeyPatch, + ) -> None: + win32serviceutil = sys.modules["win32serviceutil"] + delete_calls: list[None] = [] + monkeypatch.setattr( + service_module, + "_delete_paths_from_registry", + lambda: delete_calls.append(None), + ) + + service_module.uninstall_service() + + win32serviceutil.StopService.assert_called_once_with(service_module.SERVICE_NAME) + assert delete_calls == [None] + win32serviceutil.RemoveService.assert_called_once_with(service_module.SERVICE_NAME) + + def test_uninstall_swallows_stop_errors( + self, + service_module: ModuleType, + monkeypatch: pytest.MonkeyPatch, + ) -> None: + win32serviceutil = sys.modules["win32serviceutil"] + win32serviceutil.StopService.side_effect = RuntimeError("not running") + monkeypatch.setattr(service_module, "_delete_paths_from_registry", lambda: None) + + # Even if StopService raises (e.g. service already stopped or + # doesn't exist), uninstall must still proceed to RemoveService. + service_module.uninstall_service() + + win32serviceutil.RemoveService.assert_called_once() + + def test_start_service_delegates(self, service_module: ModuleType) -> None: + win32serviceutil = sys.modules["win32serviceutil"] + service_module.start_service() + win32serviceutil.StartService.assert_called_once_with(service_module.SERVICE_NAME) + + def test_stop_service_delegates(self, service_module: ModuleType) -> None: + win32serviceutil = sys.modules["win32serviceutil"] + service_module.stop_service() + win32serviceutil.StopService.assert_called_once_with(service_module.SERVICE_NAME) + + +# --- query_service_status ---------------------------------------------------- + + +class TestQueryServiceStatus: + def test_running_state_returns_pid(self, service_module: ModuleType) -> None: + ws = sys.modules["win32service"] + ws.OpenSCManager.return_value = MagicMock() + ws.OpenService.return_value = MagicMock() + ws.QueryServiceStatusEx.return_value = { + "CurrentState": _SERVICE_RUNNING, + "ProcessId": 4321, + } + + result = service_module.query_service_status() + + assert result == { + "service_name": service_module.SERVICE_NAME, + "state": "running", + "pid": 4321, + } + + def test_stopped_state_returns_none_pid(self, service_module: ModuleType) -> None: + ws = sys.modules["win32service"] + ws.OpenSCManager.return_value = MagicMock() + ws.OpenService.return_value = MagicMock() + # Real services often report a stale pid even when stopped; the + # code must zero it out so callers don't show a phantom process. + ws.QueryServiceStatusEx.return_value = { + "CurrentState": _SERVICE_STOPPED, + "ProcessId": 9999, + } + + result = service_module.query_service_status() + + assert result["state"] == "stopped" + assert result["pid"] is None + + +# --- _run_service_loop ------------------------------------------------------- + + +def _make_config( + tmp_path: Path, + *, + watcher_id: str | None = "w-test", + environment: str = "staging", +) -> WatcherConfig: + """Build a minimal valid `WatcherConfig` rooted in *tmp_path*. + + Mirrors the helper in ``test_runtime.py`` so the two suites share + the same fixture shape. + """ + watch_dir = tmp_path / "data" + watch_dir.mkdir() + (watch_dir / "RUN001_sample.csv").write_text("a,b\n1,2\n") + instrument = InstrumentConfig( + id="test-instrument", + watch_directory=watch_dir, + file_patterns=["*.csv"], + upload_mode="auto", + run_detection=RunDetectionConfig(pattern=r"^([^_]+)", recursive=False), + ) + return WatcherConfig( + version=1, + environment=environment, # type: ignore[arg-type] + watcher_id=watcher_id, + instrument=instrument, + ) + + +def _make_instrument_detail(status: str = "active") -> InstrumentDetailResponse: + return InstrumentDetailResponse( + id="test-instrument", + display_name="Test Instrument", + status=status, # type: ignore[arg-type] + run_count=0, + watcher_count=1, + ) + + +class _LoopHarness: + """Patches the collaborators of ``_run_service_loop``. + + Keeps the patches in one place so each test only needs to override + the one thing it cares about (failing API call, missing watcher_id, + etc.). Returns a populated ``servicemanager`` mock so tests can + assert which log functions were called. + """ + + def __init__( + self, + service_module: ModuleType, + monkeypatch: pytest.MonkeyPatch, + tmp_path: Path, + ) -> None: + self.svc = service_module + self.monkeypatch = monkeypatch + self.tmp_path = tmp_path + self.config_path = tmp_path / "config.yaml" + self.env_path = tmp_path / ".env.staging" + self.config_path.write_text("# placeholder, content is mocked away\n") + self.env_path.write_text("") + self.cfg = _make_config(tmp_path) + self.client = MagicMock(name="DataHubClient") + self.client.get_instrument.return_value = _make_instrument_detail("active") + self.client.get_config_checksum.return_value = None + self.runtime = MagicMock(name="WatcherRuntime") + self.start_calls: list[Any] = [] + self.stop_calls: list[Any] = [] + self.sm = MagicMock(name="servicemanager") + + # Patch the registry read at the service-module level so the + # winreg fake doesn't need to participate. + monkeypatch.setattr( + service_module, + "_read_paths_from_registry", + lambda: (self.config_path, self.env_path), + ) + + # Patch source modules of the lazy imports inside _run_service_loop. + from data_hub_watcher import api_client, config_io, constants, runtime + + monkeypatch.setattr(api_client, "DataHubClient", lambda *_a, **_kw: self.client) + monkeypatch.setattr(config_io, "load_config", lambda _p: self.cfg) + monkeypatch.setattr(config_io, "config_checksum", lambda _p: "deadbeef") + # env_file_path is called with no argument in _run_service_loop; + # return a path that's not on disk so the "if base_env != env_path + # and base_env.exists()" branch short-circuits. + monkeypatch.setattr( + constants, + "env_file_path", + lambda environment=None: tmp_path / ".env.nonexistent", + ) + monkeypatch.setattr( + runtime, + "build_runtime", + lambda **_kw: self.runtime, + ) + monkeypatch.setattr( + runtime, + "start_runtime", + lambda rt, started_message="": self.start_calls.append((rt, started_message)), + ) + monkeypatch.setattr( + runtime, + "stop_runtime", + lambda rt, stopped_message="": self.stop_calls.append((rt, stopped_message)), + ) + + # load_dotenv is imported directly inside the function from + # `dotenv`, so patch it on the dotenv module. + import dotenv + + self.dotenv_calls: list[Any] = [] + monkeypatch.setattr( + dotenv, + "load_dotenv", + lambda *args, **kwargs: self.dotenv_calls.append((args, kwargs)), + ) + + +@pytest.fixture +def harness( + service_module: ModuleType, + monkeypatch: pytest.MonkeyPatch, + tmp_path: Path, +) -> _LoopHarness: + return _LoopHarness(service_module, monkeypatch, tmp_path) + + +class TestRunServiceLoopHappyPath: + def test_full_startup_through_stop(self, harness: _LoopHarness) -> None: + stop_event = threading.Event() + # Pre-set the stop event so loop returns immediately after start. + stop_event.set() + + harness.svc._run_service_loop(stop_event, harness.sm) + + # The full sequence ran in order: + harness.client.get_instrument.assert_called_once_with("test-instrument") + # Checksum sync was attempted (remote was None -> push_config). + harness.client.get_config_checksum.assert_called_once_with("w-test") + harness.client.push_config.assert_called_once() + assert len(harness.start_calls) == 1 + assert len(harness.stop_calls) == 1 + # Sanity-check the start/stop messages reach the runtime so the + # API event log shows the boot/shutdown. + assert "Service started" in harness.start_calls[0][1] + assert harness.stop_calls[0][1] == "Service stopped" + # No error logs. + harness.sm.LogErrorMsg.assert_not_called() + + def test_overlays_env_files_in_order(self, harness: _LoopHarness) -> None: + # base_env defaults to a non-existent path in the harness, so + # only the registered env_path should be loaded. Make the base + # exist to verify the overlay behaviour. + from data_hub_watcher import constants + + base_env = harness.tmp_path / ".env.base" + base_env.write_text("") + harness.monkeypatch.setattr(constants, "env_file_path", lambda environment=None: base_env) + + stop_event = threading.Event() + stop_event.set() + harness.svc._run_service_loop(stop_event, harness.sm) + + # Two load_dotenv calls: base first (no override), then registered + # env_path with override=True. + assert len(harness.dotenv_calls) == 2 + first_args, first_kwargs = harness.dotenv_calls[0] + second_args, second_kwargs = harness.dotenv_calls[1] + assert first_args == (base_env,) + assert "override" not in first_kwargs or first_kwargs["override"] is False + assert second_args == (harness.env_path,) + assert second_kwargs == {"override": True} + + +class TestRunServiceLoopFailures: + def test_registry_read_failure_exits_with_error_log(self, harness: _LoopHarness) -> None: + def boom() -> tuple[Path, Path]: + raise OSError("registry key missing") + + harness.monkeypatch.setattr(harness.svc, "_read_paths_from_registry", boom) + + with pytest.raises(SystemExit) as excinfo: + harness.svc._run_service_loop(threading.Event(), harness.sm) + + assert excinfo.value.code == 1 + harness.sm.LogErrorMsg.assert_called_once() + msg = harness.sm.LogErrorMsg.call_args.args[0] + assert "registry" in msg.lower() + assert "service install" in msg + # No runtime should have been built. + assert harness.start_calls == [] + + def test_api_unreachable_at_startup_exits(self, harness: _LoopHarness) -> None: + # This is the boot-before-network failure mode that the + # SERVICE_CONFIG_FAILURE_ACTIONS_FLAG exists to recover from. + harness.client.get_instrument.side_effect = ApiError("connection refused", status_code=0) + + with pytest.raises(SystemExit) as excinfo: + harness.svc._run_service_loop(threading.Event(), harness.sm) + + assert excinfo.value.code == 1 + harness.sm.LogErrorMsg.assert_called_once() + assert "API" in harness.sm.LogErrorMsg.call_args.args[0] + assert harness.start_calls == [] + + def test_pending_instrument_exits(self, harness: _LoopHarness) -> None: + harness.client.get_instrument.return_value = _make_instrument_detail("pending") + + with pytest.raises(SystemExit) as excinfo: + harness.svc._run_service_loop(threading.Event(), harness.sm) + + assert excinfo.value.code == 1 + harness.sm.LogErrorMsg.assert_called_once() + assert "pending" in harness.sm.LogErrorMsg.call_args.args[0] + assert harness.start_calls == [] + + def test_missing_watcher_id_exits_after_status_check(self, harness: _LoopHarness) -> None: + # Rebuild config without a watcher_id. The startup must reach + # the explicit guard (not crash inside build_runtime's assert) + # so operators get a clear log line. + from data_hub_watcher import config_io + + extra = harness.tmp_path / "extra" + extra.mkdir() + cfg = _make_config(extra, watcher_id=None) + harness.monkeypatch.setattr(config_io, "load_config", lambda _p: cfg) + + with pytest.raises(SystemExit) as excinfo: + harness.svc._run_service_loop(threading.Event(), harness.sm) + + assert excinfo.value.code == 1 + harness.sm.LogErrorMsg.assert_called_once() + assert "watcher_id" in harness.sm.LogErrorMsg.call_args.args[0] + assert harness.start_calls == [] + + +class TestRunServiceLoopChecksumSync: + def test_checksum_api_error_is_a_warning_not_a_fatal(self, harness: _LoopHarness) -> None: + # If the API is reachable enough to answer get_instrument but + # the checksum endpoint blips, the service must still come up. + # This is an explicit design choice: don't punish operators for + # transient sync failures, the next heartbeat will retry. + harness.client.get_config_checksum.side_effect = ApiError("transient 500", status_code=500) + + stop_event = threading.Event() + stop_event.set() + harness.svc._run_service_loop(stop_event, harness.sm) + + harness.sm.LogWarningMsg.assert_called_once() + assert "sync" in harness.sm.LogWarningMsg.call_args.args[0].lower() + assert len(harness.start_calls) == 1 + harness.sm.LogErrorMsg.assert_not_called() + + def test_matching_checksum_skips_push(self, harness: _LoopHarness) -> None: + # When the remote checksum already matches, push_config must + # NOT be called -- otherwise every reboot would no-op rewrite + # the config and flood the audit log. + from data_hub_watcher.models import ConfigChecksumResponse + + harness.client.get_config_checksum.return_value = ConfigChecksumResponse( + config_checksum="deadbeef" + ) + + stop_event = threading.Event() + stop_event.set() + harness.svc._run_service_loop(stop_event, harness.sm) + + harness.client.push_config.assert_not_called() + assert len(harness.start_calls) == 1 diff --git a/watcher/tests/test_uploader.py b/watcher/tests/test_uploader.py index 94fef93a..ebe093e6 100644 --- a/watcher/tests/test_uploader.py +++ b/watcher/tests/test_uploader.py @@ -88,6 +88,12 @@ def test_successful_upload( rel_path = tmp_file.relative_to(uploader._watch_dir).as_posix() assert state_db.has_stat_match(rel_path, st.st_size, st.st_mtime) is True + # request_upload_url is invoked with the on-disk creation time so + # the API can persist files.file_created_at. + kwargs = mock_client.request_upload_url.call_args.kwargs + assert "file_created_at_ts" in kwargs + assert kwargs["file_created_at_ts"] > 0 + def test_already_uploaded_skips( self, uploader: Uploader, diff --git a/web-app/app/api/v1/files/[fileId]/route.ts b/web-app/app/api/v1/files/[fileId]/route.ts index 5116f12f..ed9586c3 100644 --- a/web-app/app/api/v1/files/[fileId]/route.ts +++ b/web-app/app/api/v1/files/[fileId]/route.ts @@ -170,6 +170,7 @@ export async function PATCH(request: NextRequest, { params }: RouteContext) { uploaded_at: updated.uploadedAt, processed_at: updated.processedAt, created_at: updated.createdAt, + file_created_at: updated.fileCreatedAt, }); } diff --git a/web-app/app/api/v1/instruments/[instrumentId]/runs/[runId]/files/route.ts b/web-app/app/api/v1/instruments/[instrumentId]/runs/[runId]/files/route.ts index 67bed785..ed76f07e 100644 --- a/web-app/app/api/v1/instruments/[instrumentId]/runs/[runId]/files/route.ts +++ b/web-app/app/api/v1/instruments/[instrumentId]/runs/[runId]/files/route.ts @@ -127,5 +127,6 @@ function formatFileResponse(f: typeof files.$inferSelect) { uploaded_at: f.uploadedAt, processed_at: f.processedAt, created_at: f.createdAt, + file_created_at: f.fileCreatedAt, }; } diff --git a/web-app/app/api/v1/instruments/[instrumentId]/runs/[runId]/request-upload-url/route.ts b/web-app/app/api/v1/instruments/[instrumentId]/runs/[runId]/request-upload-url/route.ts index 56ae8164..01ffdbd1 100644 --- a/web-app/app/api/v1/instruments/[instrumentId]/runs/[runId]/request-upload-url/route.ts +++ b/web-app/app/api/v1/instruments/[instrumentId]/runs/[runId]/request-upload-url/route.ts @@ -74,6 +74,10 @@ export async function POST(request: NextRequest, { params }: RouteContext) { typeof body.content_type === "string" ? body.content_type : undefined; const sizeBytes = typeof body.size_bytes === "number" ? body.size_bytes : undefined; + const fileCreatedAt = + typeof body.file_created_at === "string" + ? new Date(body.file_created_at) + : null; // Look up existing file record by run + filename. The file may have been // created by report_run with a full relative_path (e.g. "EXP-001/data.csv"), @@ -113,6 +117,15 @@ export async function POST(request: NextRequest, { params }: RouteContext) { if (existingFile) { // File exists in detected / upload_requested — reuse the record. fileId = existingFile.id; + // Backfill file_created_at if the watcher started reporting it after + // the row was first inserted (e.g. a queue-mode upload following an + // earlier detected_files report that predated this column). + if (fileCreatedAt && !existingFile.fileCreatedAt) { + await db + .update(files) + .set({ fileCreatedAt }) + .where(eq(files.id, existingFile.id)); + } } else { const now = new Date(); const [inserted] = await db @@ -125,6 +138,7 @@ export async function POST(request: NextRequest, { params }: RouteContext) { sizeBytes: sizeBytes ?? null, status: "detected", detectedAt: now, + fileCreatedAt, }) .returning({ id: files.id }); diff --git a/web-app/app/api/v1/instruments/[instrumentId]/runs/[runId]/route.ts b/web-app/app/api/v1/instruments/[instrumentId]/runs/[runId]/route.ts index 433c433b..efc0fe76 100644 --- a/web-app/app/api/v1/instruments/[instrumentId]/runs/[runId]/route.ts +++ b/web-app/app/api/v1/instruments/[instrumentId]/runs/[runId]/route.ts @@ -70,6 +70,7 @@ export async function GET(request: NextRequest, { params }: RouteContext) { ? await getPresignedDownloadUrl(f.s3Bucket, f.s3Key) : null, created_at: f.createdAt, + file_created_at: f.fileCreatedAt, })) ); @@ -156,6 +157,7 @@ export async function PATCH(request: NextRequest, { params }: RouteContext) { relative_path: string; filename: string; size_bytes?: number; + file_created_at?: string; }) => ({ instrumentRunId: run.id, relativePath: f.relative_path, @@ -163,6 +165,10 @@ export async function PATCH(request: NextRequest, { params }: RouteContext) { sizeBytes: f.size_bytes ?? null, status: "detected" as const, detectedAt: now, + fileCreatedAt: + typeof f.file_created_at === "string" + ? new Date(f.file_created_at) + : null, }) ); diff --git a/web-app/app/api/v1/instruments/[instrumentId]/runs/route.ts b/web-app/app/api/v1/instruments/[instrumentId]/runs/route.ts index d0e7f4e1..fb60e9be 100644 --- a/web-app/app/api/v1/instruments/[instrumentId]/runs/route.ts +++ b/web-app/app/api/v1/instruments/[instrumentId]/runs/route.ts @@ -141,6 +141,7 @@ export async function POST(request: NextRequest, { params }: RouteContext) { relative_path: string; filename: string; size_bytes?: number; + file_created_at?: string; }) => ({ instrumentRunId: run.id, relativePath: f.relative_path, @@ -148,6 +149,10 @@ export async function POST(request: NextRequest, { params }: RouteContext) { sizeBytes: f.size_bytes ?? null, status: "detected" as const, detectedAt: now, + fileCreatedAt: + typeof f.file_created_at === "string" + ? new Date(f.file_created_at) + : null, }) ); diff --git a/web-app/components/runs/run-files-table.tsx b/web-app/components/runs/run-files-table.tsx index 1298d75f..45ba8e55 100644 --- a/web-app/components/runs/run-files-table.tsx +++ b/web-app/components/runs/run-files-table.tsx @@ -186,7 +186,11 @@ function FileInfoCells({ file }: { file: RunFile }) { {formatBytes(file.sizeBytes)} - {file.createdAt ? formatDateTime(file.createdAt) : "—"} + {file.fileCreatedAt + ? formatDateTime(file.fileCreatedAt) + : file.createdAt + ? formatDateTime(file.createdAt) + : "—"} diff --git a/web-app/components/watchers/status-badge.ts b/web-app/components/watchers/status-badge.ts deleted file mode 100644 index 5d833f8d..00000000 --- a/web-app/components/watchers/status-badge.ts +++ /dev/null @@ -1,14 +0,0 @@ -import type { EffectiveStatus } from "@/lib/api/watchers"; - -export const statusBadge: Record< - EffectiveStatus, - { - label: string; - variant: "default" | "outline" | "secondary" | "destructive"; - } -> = { - watching: { label: "Watching", variant: "default" }, - stale: { label: "Stale", variant: "destructive" }, - stopped: { label: "Stopped", variant: "secondary" }, - registered: { label: "Registered", variant: "outline" }, -}; diff --git a/web-app/components/watchers/watcher-header.tsx b/web-app/components/watchers/watcher-header.tsx index ff82e6b3..87319fb2 100644 --- a/web-app/components/watchers/watcher-header.tsx +++ b/web-app/components/watchers/watcher-header.tsx @@ -8,14 +8,13 @@ import { BreadcrumbSeparator, } from "@/components/ui/breadcrumb"; import { DeregisterDialog } from "@/components/watchers/deregister-dialog"; -import { statusBadge } from "@/components/watchers/status-badge"; +import { WatcherStatusBadge } from "@/components/watchers/watcher-status-badge"; import type { WatcherDetail } from "@/lib/api/watchers"; import { formatDate } from "@/lib/date"; import { formatRelativeTime } from "@/lib/utils"; import Link from "next/link"; export function WatcherHeader({ watcher }: { watcher: WatcherDetail }) { - const sb = statusBadge[watcher.effectiveStatus]; const isDeregistered = !!watcher.deletedAt; return ( @@ -57,9 +56,10 @@ export function WatcherHeader({ watcher }: { watcher: WatcherDetail }) {

{watcher.hostname ?? "Unnamed Watcher"}

- - {sb.label} - + {isDeregistered && ( Deregistered diff --git a/web-app/components/watchers/watcher-status-badge.tsx b/web-app/components/watchers/watcher-status-badge.tsx index 8912ccb6..528cf584 100644 --- a/web-app/components/watchers/watcher-status-badge.tsx +++ b/web-app/components/watchers/watcher-status-badge.tsx @@ -7,60 +7,103 @@ import { TooltipTrigger, } from "@/components/ui/tooltip"; import type { WatcherOnlineStatus } from "@/components/watchers/watcher-online-status"; +import type { EffectiveStatus } from "@/lib/api/watchers"; import { cn, formatRelativeTime } from "@/lib/utils"; -import { Radio, WifiOff } from "lucide-react"; +import { Clock, Power, Radio, WifiOff } from "lucide-react"; + +/** + * Union of every status this badge can render: + * - `WatcherOnlineStatus` is the instrument-level aggregate (any watcher + * online vs none) used in the instruments table and instrument header. + * - `EffectiveStatus` is the per-watcher state used in the watchers table + * and watcher detail header. + * + * `watching` and `online` collapse to the same green "Online" treatment; + * `stale` and `offline` collapse to the same destructive treatment but with + * different labels ("Unresponsive" vs "Offline") since the aggregate badge + * can't tell why an instrument's watchers are silent, while the per-watcher + * badge can. + */ +export type WatcherBadgeStatus = WatcherOnlineStatus | EffectiveStatus; + +const ONLINE_CLASSNAME = + "border-transparent bg-green-500/10 text-green-700 dark:bg-green-500/15 dark:text-green-400"; + +const OFFLINE_CLASSNAME = + "border-transparent bg-destructive/10 text-destructive dark:bg-destructive/20"; + +const MUTED_FILLED_CLASSNAME = + "border-transparent bg-muted text-muted-foreground"; + +const MUTED_OUTLINED_CLASSNAME = "border-border text-muted-foreground"; const STATUS_CONFIG: Record< - WatcherOnlineStatus, + WatcherBadgeStatus, { label: string; Icon: typeof Radio; className: string; } > = { - online: { - label: "Online", - Icon: Radio, - className: - "border-transparent bg-green-500/10 text-green-700 dark:bg-green-500/15 dark:text-green-400", - }, - offline: { - label: "Offline", - Icon: WifiOff, - className: - "border-transparent bg-destructive/10 text-destructive dark:bg-destructive/20", - }, + // Instrument-level aggregate + online: { label: "Online", Icon: Radio, className: ONLINE_CLASSNAME }, + offline: { label: "Offline", Icon: WifiOff, className: OFFLINE_CLASSNAME }, no_watcher: { label: "No Watcher", Icon: WifiOff, - className: "border-border text-muted-foreground", + className: MUTED_OUTLINED_CLASSNAME, + }, + // Per-watcher + watching: { label: "Online", Icon: Radio, className: ONLINE_CLASSNAME }, + stale: { + label: "Unresponsive", + Icon: WifiOff, + className: OFFLINE_CLASSNAME, + }, + // Distinct from `stale`: someone (or the host) shut the watcher down + // gracefully, so it's expected to be silent and shouldn't read as an alarm. + stopped: { label: "Stopped", Icon: Power, className: MUTED_FILLED_CLASSNAME }, + // Registered but hasn't sent its first heartbeat yet — transient. + registered: { + label: "Registered", + Icon: Clock, + className: MUTED_OUTLINED_CLASSNAME, }, }; +// Statuses where a "last online at" tooltip makes sense — i.e. the watcher +// *should* be reporting but isn't. Excludes `stopped` (intentional) and +// `registered` (no heartbeat history yet). +const TOOLTIP_STATUSES = new Set(["offline", "stale"]); + export function WatcherStatusBadge({ status, lastOnlineAt, verbose = false, className, }: { - status: WatcherOnlineStatus; + status: WatcherBadgeStatus; /** - * Most recent watcher heartbeat for the instrument. When the badge is in - * `offline` state and this is provided, it's surfaced via tooltip so users - * can tell at a glance how long the instrument has been silent. + * Most recent watcher heartbeat. When the badge is in an "unexpected + * silence" state (`offline` / `stale`) and this is provided, it's + * surfaced via tooltip so users can tell at a glance how long the + * watcher has been silent. */ lastOnlineAt?: Date | null; /** - * When true, prefixes the label with "Watcher" (e.g. "Watcher Online"). - * Useful in headers where the badge stands alone outside a status column. - * The `no_watcher` label ("No Watcher") is never prefixed. + * When true, prefixes the aggregate online/offline labels with "Watcher" + * (e.g. "Watcher Online"). Useful in headers where the badge stands alone + * outside a status column. Has no effect on per-watcher labels or on + * `no_watcher`, where prefixing would be redundant or wrong. */ verbose?: boolean; className?: string; }) { const { label, Icon, className: variantClassName } = STATUS_CONFIG[status]; const fullLabel = - verbose && status !== "no_watcher" ? `Watcher ${label}` : label; + verbose && (status === "online" || status === "offline") + ? `Watcher ${label}` + : label; const badge = ( @@ -69,7 +112,7 @@ export function WatcherStatusBadge({ ); - if (status !== "offline" || !lastOnlineAt) { + if (!TOOLTIP_STATUSES.has(status) || !lastOnlineAt) { return badge; } diff --git a/web-app/components/watchers/watchers-table.tsx b/web-app/components/watchers/watchers-table.tsx index 856add91..ad9d44ab 100644 --- a/web-app/components/watchers/watchers-table.tsx +++ b/web-app/components/watchers/watchers-table.tsx @@ -1,5 +1,4 @@ import { ClickableRow } from "@/components/instruments/runs-table/clickable-row"; -import { Badge } from "@/components/ui/badge"; import { Table, TableBody, @@ -9,7 +8,7 @@ import { TableRow, } from "@/components/ui/table"; import { DeregisterDialog } from "@/components/watchers/deregister-dialog"; -import { statusBadge } from "@/components/watchers/status-badge"; +import { WatcherStatusBadge } from "@/components/watchers/watcher-status-badge"; import type { WatcherListItem } from "@/lib/api/watchers"; import { cn, formatRelativeTime } from "@/lib/utils"; import { SearchX } from "lucide-react"; @@ -49,7 +48,6 @@ export function WatchersTable({ {data.map((row) => { - const sb = statusBadge[row.effectiveStatus]; return ( - - {sb.label} - + diff --git a/web-app/drizzle/0009_clammy_tyger_tiger.sql b/web-app/drizzle/0009_clammy_tyger_tiger.sql new file mode 100644 index 00000000..c8ea81fd --- /dev/null +++ b/web-app/drizzle/0009_clammy_tyger_tiger.sql @@ -0,0 +1 @@ +ALTER TABLE "files" ADD COLUMN "file_created_at" timestamp with time zone; \ No newline at end of file diff --git a/web-app/drizzle/meta/0009_snapshot.json b/web-app/drizzle/meta/0009_snapshot.json new file mode 100644 index 00000000..05f4ac2a --- /dev/null +++ b/web-app/drizzle/meta/0009_snapshot.json @@ -0,0 +1,1368 @@ +{ + "id": "eef230a9-9ce9-499c-a0cd-f920941ab529", + "prevId": "c2b6ce60-ef03-4243-b22b-989c7dc2e1bd", + "version": "7", + "dialect": "postgresql", + "tables": { + "public.account": { + "name": "account", + "schema": "", + "columns": { + "userId": { + "name": "userId", + "type": "text", + "primaryKey": false, + "notNull": true + }, + "type": { + "name": "type", + "type": "text", + "primaryKey": false, + "notNull": true + }, + "provider": { + "name": "provider", + "type": "text", + "primaryKey": false, + "notNull": true + }, + "providerAccountId": { + "name": "providerAccountId", + "type": "text", + "primaryKey": false, + "notNull": true + }, + "refresh_token": { + "name": "refresh_token", + "type": "text", + "primaryKey": false, + "notNull": false + }, + "access_token": { + "name": "access_token", + "type": "text", + "primaryKey": false, + "notNull": false + }, + "expires_at": { + "name": "expires_at", + "type": "integer", + "primaryKey": false, + "notNull": false + }, + "token_type": { + "name": "token_type", + "type": "text", + "primaryKey": false, + "notNull": false + }, + "scope": { + "name": "scope", + "type": "text", + "primaryKey": false, + "notNull": false + }, + "id_token": { + "name": "id_token", + "type": "text", + "primaryKey": false, + "notNull": false + }, + "session_state": { + "name": "session_state", + "type": "text", + "primaryKey": false, + "notNull": false + } + }, + "indexes": { + "idx_accounts_user_id": { + "name": "idx_accounts_user_id", + "columns": [ + { + "expression": "userId", + "isExpression": false, + "asc": true, + "nulls": "last" + } + ], + "isUnique": false, + "concurrently": false, + "method": "btree", + "with": {} + } + }, + "foreignKeys": { + "account_userId_user_id_fk": { + "name": "account_userId_user_id_fk", + "tableFrom": "account", + "tableTo": "user", + "columnsFrom": [ + "userId" + ], + "columnsTo": [ + "id" + ], + "onDelete": "cascade", + "onUpdate": "no action" + } + }, + "compositePrimaryKeys": { + "account_provider_providerAccountId_pk": { + "name": "account_provider_providerAccountId_pk", + "columns": [ + "provider", + "providerAccountId" + ] + } + }, + "uniqueConstraints": {}, + "policies": {}, + "checkConstraints": {}, + "isRLSEnabled": false + }, + "public.files": { + "name": "files", + "schema": "", + "columns": { + "id": { + "name": "id", + "type": "bigserial", + "primaryKey": true, + "notNull": true + }, + "instrument_run_id": { + "name": "instrument_run_id", + "type": "uuid", + "primaryKey": false, + "notNull": true + }, + "relative_path": { + "name": "relative_path", + "type": "text", + "primaryKey": false, + "notNull": false + }, + "s3_bucket": { + "name": "s3_bucket", + "type": "text", + "primaryKey": false, + "notNull": false + }, + "s3_key": { + "name": "s3_key", + "type": "text", + "primaryKey": false, + "notNull": false + }, + "filename": { + "name": "filename", + "type": "text", + "primaryKey": false, + "notNull": true + }, + "content_type": { + "name": "content_type", + "type": "text", + "primaryKey": false, + "notNull": false + }, + "size_bytes": { + "name": "size_bytes", + "type": "bigint", + "primaryKey": false, + "notNull": false + }, + "category": { + "name": "category", + "type": "file_category", + "typeSchema": "public", + "primaryKey": false, + "notNull": true, + "default": "'raw'" + }, + "status": { + "name": "status", + "type": "file_status", + "typeSchema": "public", + "primaryKey": false, + "notNull": true, + "default": "'detected'" + }, + "metadata": { + "name": "metadata", + "type": "jsonb", + "primaryKey": false, + "notNull": true, + "default": "'{}'::jsonb" + }, + "error_message": { + "name": "error_message", + "type": "text", + "primaryKey": false, + "notNull": false + }, + "detected_at": { + "name": "detected_at", + "type": "timestamp with time zone", + "primaryKey": false, + "notNull": false + }, + "upload_requested_at": { + "name": "upload_requested_at", + "type": "timestamp with time zone", + "primaryKey": false, + "notNull": false + }, + "uploaded_at": { + "name": "uploaded_at", + "type": "timestamp with time zone", + "primaryKey": false, + "notNull": false + }, + "processed_at": { + "name": "processed_at", + "type": "timestamp with time zone", + "primaryKey": false, + "notNull": false + }, + "file_created_at": { + "name": "file_created_at", + "type": "timestamp with time zone", + "primaryKey": false, + "notNull": false + }, + "created_at": { + "name": "created_at", + "type": "timestamp with time zone", + "primaryKey": false, + "notNull": true, + "default": "now()" + }, + "deleted_at": { + "name": "deleted_at", + "type": "timestamp with time zone", + "primaryKey": false, + "notNull": false + } + }, + "indexes": { + "uq_files_instrument_run_id_relative_path": { + "name": "uq_files_instrument_run_id_relative_path", + "columns": [ + { + "expression": "instrument_run_id", + "isExpression": false, + "asc": true, + "nulls": "last" + }, + { + "expression": "relative_path", + "isExpression": false, + "asc": true, + "nulls": "last" + } + ], + "isUnique": true, + "where": "\"files\".\"relative_path\" is not null", + "concurrently": false, + "method": "btree", + "with": {} + }, + "uq_files_s3_key": { + "name": "uq_files_s3_key", + "columns": [ + { + "expression": "s3_key", + "isExpression": false, + "asc": true, + "nulls": "last" + } + ], + "isUnique": true, + "where": "\"files\".\"s3_key\" is not null", + "concurrently": false, + "method": "btree", + "with": {} + }, + "idx_files_instrument_run_id": { + "name": "idx_files_instrument_run_id", + "columns": [ + { + "expression": "instrument_run_id", + "isExpression": false, + "asc": true, + "nulls": "last" + } + ], + "isUnique": false, + "concurrently": false, + "method": "btree", + "with": {} + }, + "idx_files_status_instrument_run_id": { + "name": "idx_files_status_instrument_run_id", + "columns": [ + { + "expression": "status", + "isExpression": false, + "asc": true, + "nulls": "last" + }, + { + "expression": "instrument_run_id", + "isExpression": false, + "asc": true, + "nulls": "last" + } + ], + "isUnique": false, + "concurrently": false, + "method": "btree", + "with": {} + }, + "idx_files_active": { + "name": "idx_files_active", + "columns": [ + { + "expression": "instrument_run_id", + "isExpression": false, + "asc": true, + "nulls": "last" + } + ], + "isUnique": false, + "where": "\"files\".\"deleted_at\" is null", + "concurrently": false, + "method": "btree", + "with": {} + }, + "idx_files_upload_queue": { + "name": "idx_files_upload_queue", + "columns": [ + { + "expression": "upload_requested_at", + "isExpression": false, + "asc": true, + "nulls": "last" + } + ], + "isUnique": false, + "where": "\"files\".\"upload_requested_at\" is not null and \"files\".\"uploaded_at\" is null and \"files\".\"deleted_at\" is null", + "concurrently": false, + "method": "btree", + "with": {} + }, + "idx_files_metadata_gin": { + "name": "idx_files_metadata_gin", + "columns": [ + { + "expression": "metadata", + "isExpression": false, + "asc": true, + "nulls": "last" + } + ], + "isUnique": false, + "concurrently": false, + "method": "gin", + "with": {} + } + }, + "foreignKeys": { + "files_instrument_run_id_instrument_runs_id_fk": { + "name": "files_instrument_run_id_instrument_runs_id_fk", + "tableFrom": "files", + "tableTo": "instrument_runs", + "columnsFrom": [ + "instrument_run_id" + ], + "columnsTo": [ + "id" + ], + "onDelete": "no action", + "onUpdate": "no action" + } + }, + "compositePrimaryKeys": {}, + "uniqueConstraints": {}, + "policies": {}, + "checkConstraints": {}, + "isRLSEnabled": false + }, + "public.instrument_runs": { + "name": "instrument_runs", + "schema": "", + "columns": { + "id": { + "name": "id", + "type": "uuid", + "primaryKey": true, + "notNull": true, + "default": "gen_random_uuid()" + }, + "instrument_id": { + "name": "instrument_id", + "type": "text", + "primaryKey": false, + "notNull": true + }, + "run_id": { + "name": "run_id", + "type": "text", + "primaryKey": false, + "notNull": true + }, + "source": { + "name": "source", + "type": "instrument_run_source", + "typeSchema": "public", + "primaryKey": false, + "notNull": true, + "default": "'lambda'" + }, + "watcher_id": { + "name": "watcher_id", + "type": "uuid", + "primaryKey": false, + "notNull": false + }, + "metadata": { + "name": "metadata", + "type": "jsonb", + "primaryKey": false, + "notNull": true, + "default": "'{}'::jsonb" + }, + "created_at": { + "name": "created_at", + "type": "timestamp with time zone", + "primaryKey": false, + "notNull": true, + "default": "now()" + }, + "updated_at": { + "name": "updated_at", + "type": "timestamp with time zone", + "primaryKey": false, + "notNull": true, + "default": "now()" + }, + "deleted_at": { + "name": "deleted_at", + "type": "timestamp with time zone", + "primaryKey": false, + "notNull": false + }, + "files_purged_at": { + "name": "files_purged_at", + "type": "timestamp with time zone", + "primaryKey": false, + "notNull": false + } + }, + "indexes": { + "idx_instrument_runs_instrument_id_created_at": { + "name": "idx_instrument_runs_instrument_id_created_at", + "columns": [ + { + "expression": "instrument_id", + "isExpression": false, + "asc": true, + "nulls": "last" + }, + { + "expression": "created_at", + "isExpression": false, + "asc": false, + "nulls": "last" + } + ], + "isUnique": false, + "concurrently": false, + "method": "btree", + "with": {} + }, + "idx_instrument_runs_active": { + "name": "idx_instrument_runs_active", + "columns": [ + { + "expression": "instrument_id", + "isExpression": false, + "asc": true, + "nulls": "last" + }, + { + "expression": "created_at", + "isExpression": false, + "asc": false, + "nulls": "last" + } + ], + "isUnique": false, + "where": "\"instrument_runs\".\"deleted_at\" is null", + "concurrently": false, + "method": "btree", + "with": {} + }, + "idx_instrument_runs_metadata_gin": { + "name": "idx_instrument_runs_metadata_gin", + "columns": [ + { + "expression": "metadata", + "isExpression": false, + "asc": true, + "nulls": "last" + } + ], + "isUnique": false, + "concurrently": false, + "method": "gin", + "with": {} + } + }, + "foreignKeys": { + "instrument_runs_instrument_id_instruments_id_fk": { + "name": "instrument_runs_instrument_id_instruments_id_fk", + "tableFrom": "instrument_runs", + "tableTo": "instruments", + "columnsFrom": [ + "instrument_id" + ], + "columnsTo": [ + "id" + ], + "onDelete": "no action", + "onUpdate": "no action" + }, + "instrument_runs_watcher_id_watchers_id_fk": { + "name": "instrument_runs_watcher_id_watchers_id_fk", + "tableFrom": "instrument_runs", + "tableTo": "watchers", + "columnsFrom": [ + "watcher_id" + ], + "columnsTo": [ + "id" + ], + "onDelete": "no action", + "onUpdate": "no action" + } + }, + "compositePrimaryKeys": {}, + "uniqueConstraints": { + "uq_instrument_runs_instrument_id_run_id": { + "name": "uq_instrument_runs_instrument_id_run_id", + "nullsNotDistinct": false, + "columns": [ + "instrument_id", + "run_id" + ] + } + }, + "policies": {}, + "checkConstraints": {}, + "isRLSEnabled": false + }, + "public.instruments": { + "name": "instruments", + "schema": "", + "columns": { + "id": { + "name": "id", + "type": "text", + "primaryKey": true, + "notNull": true + }, + "display_name": { + "name": "display_name", + "type": "text", + "primaryKey": false, + "notNull": true + }, + "status": { + "name": "status", + "type": "instrument_status", + "typeSchema": "public", + "primaryKey": false, + "notNull": true, + "default": "'active'" + }, + "instrument_type": { + "name": "instrument_type", + "type": "instrument_type", + "typeSchema": "public", + "primaryKey": false, + "notNull": true, + "default": "'generic'" + }, + "created_at": { + "name": "created_at", + "type": "timestamp with time zone", + "primaryKey": false, + "notNull": true, + "default": "now()" + }, + "updated_at": { + "name": "updated_at", + "type": "timestamp with time zone", + "primaryKey": false, + "notNull": true, + "default": "now()" + } + }, + "indexes": {}, + "foreignKeys": {}, + "compositePrimaryKeys": {}, + "uniqueConstraints": {}, + "policies": {}, + "checkConstraints": {}, + "isRLSEnabled": false + }, + "public.personal_access_tokens": { + "name": "personal_access_tokens", + "schema": "", + "columns": { + "id": { + "name": "id", + "type": "uuid", + "primaryKey": true, + "notNull": true, + "default": "gen_random_uuid()" + }, + "user_id": { + "name": "user_id", + "type": "text", + "primaryKey": false, + "notNull": true + }, + "name": { + "name": "name", + "type": "text", + "primaryKey": false, + "notNull": true + }, + "token_hash": { + "name": "token_hash", + "type": "text", + "primaryKey": false, + "notNull": true + }, + "token_prefix": { + "name": "token_prefix", + "type": "text", + "primaryKey": false, + "notNull": true + }, + "last_used_at": { + "name": "last_used_at", + "type": "timestamp with time zone", + "primaryKey": false, + "notNull": false + }, + "expires_at": { + "name": "expires_at", + "type": "timestamp with time zone", + "primaryKey": false, + "notNull": false + }, + "created_at": { + "name": "created_at", + "type": "timestamp with time zone", + "primaryKey": false, + "notNull": true, + "default": "now()" + } + }, + "indexes": { + "idx_personal_access_tokens_user_id": { + "name": "idx_personal_access_tokens_user_id", + "columns": [ + { + "expression": "user_id", + "isExpression": false, + "asc": true, + "nulls": "last" + } + ], + "isUnique": false, + "concurrently": false, + "method": "btree", + "with": {} + } + }, + "foreignKeys": { + "personal_access_tokens_user_id_user_id_fk": { + "name": "personal_access_tokens_user_id_user_id_fk", + "tableFrom": "personal_access_tokens", + "tableTo": "user", + "columnsFrom": [ + "user_id" + ], + "columnsTo": [ + "id" + ], + "onDelete": "cascade", + "onUpdate": "no action" + } + }, + "compositePrimaryKeys": {}, + "uniqueConstraints": { + "personal_access_tokens_token_hash_unique": { + "name": "personal_access_tokens_token_hash_unique", + "nullsNotDistinct": false, + "columns": [ + "token_hash" + ] + } + }, + "policies": {}, + "checkConstraints": {}, + "isRLSEnabled": false + }, + "public.run_attributions": { + "name": "run_attributions", + "schema": "", + "columns": { + "run_id": { + "name": "run_id", + "type": "uuid", + "primaryKey": false, + "notNull": true + }, + "user_id": { + "name": "user_id", + "type": "text", + "primaryKey": false, + "notNull": true + }, + "created_at": { + "name": "created_at", + "type": "timestamp with time zone", + "primaryKey": false, + "notNull": true, + "default": "now()" + } + }, + "indexes": { + "idx_run_attributions_run_id": { + "name": "idx_run_attributions_run_id", + "columns": [ + { + "expression": "run_id", + "isExpression": false, + "asc": true, + "nulls": "last" + } + ], + "isUnique": false, + "concurrently": false, + "method": "btree", + "with": {} + }, + "idx_run_attributions_user_id": { + "name": "idx_run_attributions_user_id", + "columns": [ + { + "expression": "user_id", + "isExpression": false, + "asc": true, + "nulls": "last" + } + ], + "isUnique": false, + "concurrently": false, + "method": "btree", + "with": {} + } + }, + "foreignKeys": { + "run_attributions_run_id_instrument_runs_id_fk": { + "name": "run_attributions_run_id_instrument_runs_id_fk", + "tableFrom": "run_attributions", + "tableTo": "instrument_runs", + "columnsFrom": [ + "run_id" + ], + "columnsTo": [ + "id" + ], + "onDelete": "cascade", + "onUpdate": "no action" + }, + "run_attributions_user_id_user_id_fk": { + "name": "run_attributions_user_id_user_id_fk", + "tableFrom": "run_attributions", + "tableTo": "user", + "columnsFrom": [ + "user_id" + ], + "columnsTo": [ + "id" + ], + "onDelete": "cascade", + "onUpdate": "no action" + } + }, + "compositePrimaryKeys": { + "run_attributions_run_id_user_id_pk": { + "name": "run_attributions_run_id_user_id_pk", + "columns": [ + "run_id", + "user_id" + ] + } + }, + "uniqueConstraints": {}, + "policies": {}, + "checkConstraints": {}, + "isRLSEnabled": false + }, + "public.session": { + "name": "session", + "schema": "", + "columns": { + "sessionToken": { + "name": "sessionToken", + "type": "text", + "primaryKey": true, + "notNull": true + }, + "userId": { + "name": "userId", + "type": "text", + "primaryKey": false, + "notNull": true + }, + "expires": { + "name": "expires", + "type": "timestamp", + "primaryKey": false, + "notNull": true + } + }, + "indexes": { + "idx_sessions_user_id": { + "name": "idx_sessions_user_id", + "columns": [ + { + "expression": "userId", + "isExpression": false, + "asc": true, + "nulls": "last" + } + ], + "isUnique": false, + "concurrently": false, + "method": "btree", + "with": {} + } + }, + "foreignKeys": { + "session_userId_user_id_fk": { + "name": "session_userId_user_id_fk", + "tableFrom": "session", + "tableTo": "user", + "columnsFrom": [ + "userId" + ], + "columnsTo": [ + "id" + ], + "onDelete": "cascade", + "onUpdate": "no action" + } + }, + "compositePrimaryKeys": {}, + "uniqueConstraints": {}, + "policies": {}, + "checkConstraints": {}, + "isRLSEnabled": false + }, + "public.user": { + "name": "user", + "schema": "", + "columns": { + "id": { + "name": "id", + "type": "text", + "primaryKey": true, + "notNull": true + }, + "name": { + "name": "name", + "type": "text", + "primaryKey": false, + "notNull": false + }, + "email": { + "name": "email", + "type": "text", + "primaryKey": false, + "notNull": false + }, + "emailVerified": { + "name": "emailVerified", + "type": "timestamp", + "primaryKey": false, + "notNull": false + }, + "image": { + "name": "image", + "type": "text", + "primaryKey": false, + "notNull": false + } + }, + "indexes": {}, + "foreignKeys": {}, + "compositePrimaryKeys": {}, + "uniqueConstraints": { + "user_email_unique": { + "name": "user_email_unique", + "nullsNotDistinct": false, + "columns": [ + "email" + ] + } + }, + "policies": {}, + "checkConstraints": {}, + "isRLSEnabled": false + }, + "public.watcher_events": { + "name": "watcher_events", + "schema": "", + "columns": { + "id": { + "name": "id", + "type": "bigserial", + "primaryKey": true, + "notNull": true + }, + "watcher_id": { + "name": "watcher_id", + "type": "uuid", + "primaryKey": false, + "notNull": true + }, + "event_type": { + "name": "event_type", + "type": "watcher_event_type", + "typeSchema": "public", + "primaryKey": false, + "notNull": true + }, + "message": { + "name": "message", + "type": "text", + "primaryKey": false, + "notNull": true + }, + "details": { + "name": "details", + "type": "jsonb", + "primaryKey": false, + "notNull": false + }, + "timestamp": { + "name": "timestamp", + "type": "timestamp with time zone", + "primaryKey": false, + "notNull": true + }, + "created_at": { + "name": "created_at", + "type": "timestamp with time zone", + "primaryKey": false, + "notNull": true, + "default": "now()" + } + }, + "indexes": { + "idx_watcher_events_watcher_id_timestamp": { + "name": "idx_watcher_events_watcher_id_timestamp", + "columns": [ + { + "expression": "watcher_id", + "isExpression": false, + "asc": true, + "nulls": "last" + }, + { + "expression": "timestamp", + "isExpression": false, + "asc": false, + "nulls": "last" + } + ], + "isUnique": false, + "concurrently": false, + "method": "btree", + "with": {} + }, + "idx_watcher_events_watcher_id_event_type": { + "name": "idx_watcher_events_watcher_id_event_type", + "columns": [ + { + "expression": "watcher_id", + "isExpression": false, + "asc": true, + "nulls": "last" + }, + { + "expression": "event_type", + "isExpression": false, + "asc": true, + "nulls": "last" + } + ], + "isUnique": false, + "concurrently": false, + "method": "btree", + "with": {} + } + }, + "foreignKeys": { + "watcher_events_watcher_id_watchers_id_fk": { + "name": "watcher_events_watcher_id_watchers_id_fk", + "tableFrom": "watcher_events", + "tableTo": "watchers", + "columnsFrom": [ + "watcher_id" + ], + "columnsTo": [ + "id" + ], + "onDelete": "no action", + "onUpdate": "no action" + } + }, + "compositePrimaryKeys": {}, + "uniqueConstraints": {}, + "policies": {}, + "checkConstraints": {}, + "isRLSEnabled": false + }, + "public.watcher_heartbeats": { + "name": "watcher_heartbeats", + "schema": "", + "columns": { + "id": { + "name": "id", + "type": "bigserial", + "primaryKey": true, + "notNull": true + }, + "watcher_id": { + "name": "watcher_id", + "type": "uuid", + "primaryKey": false, + "notNull": true + }, + "timestamp": { + "name": "timestamp", + "type": "timestamp with time zone", + "primaryKey": false, + "notNull": true + }, + "status": { + "name": "status", + "type": "text", + "primaryKey": false, + "notNull": true + }, + "upload_mode": { + "name": "upload_mode", + "type": "upload_mode", + "typeSchema": "public", + "primaryKey": false, + "notNull": false + }, + "files_uploaded_since_last": { + "name": "files_uploaded_since_last", + "type": "integer", + "primaryKey": false, + "notNull": false, + "default": 0 + }, + "runs_reported_since_last": { + "name": "runs_reported_since_last", + "type": "integer", + "primaryKey": false, + "notNull": false, + "default": 0 + }, + "errors_since_last": { + "name": "errors_since_last", + "type": "integer", + "primaryKey": false, + "notNull": false, + "default": 0 + }, + "uptime_seconds": { + "name": "uptime_seconds", + "type": "integer", + "primaryKey": false, + "notNull": false + }, + "created_at": { + "name": "created_at", + "type": "timestamp with time zone", + "primaryKey": false, + "notNull": true, + "default": "now()" + } + }, + "indexes": { + "idx_watcher_heartbeats_watcher_id_timestamp": { + "name": "idx_watcher_heartbeats_watcher_id_timestamp", + "columns": [ + { + "expression": "watcher_id", + "isExpression": false, + "asc": true, + "nulls": "last" + }, + { + "expression": "timestamp", + "isExpression": false, + "asc": false, + "nulls": "last" + } + ], + "isUnique": false, + "concurrently": false, + "method": "btree", + "with": {} + } + }, + "foreignKeys": { + "watcher_heartbeats_watcher_id_watchers_id_fk": { + "name": "watcher_heartbeats_watcher_id_watchers_id_fk", + "tableFrom": "watcher_heartbeats", + "tableTo": "watchers", + "columnsFrom": [ + "watcher_id" + ], + "columnsTo": [ + "id" + ], + "onDelete": "no action", + "onUpdate": "no action" + } + }, + "compositePrimaryKeys": {}, + "uniqueConstraints": {}, + "policies": {}, + "checkConstraints": {}, + "isRLSEnabled": false + }, + "public.watchers": { + "name": "watchers", + "schema": "", + "columns": { + "id": { + "name": "id", + "type": "uuid", + "primaryKey": true, + "notNull": true, + "default": "gen_random_uuid()" + }, + "instrument_id": { + "name": "instrument_id", + "type": "text", + "primaryKey": false, + "notNull": true + }, + "hostname": { + "name": "hostname", + "type": "text", + "primaryKey": false, + "notNull": false + }, + "os_info": { + "name": "os_info", + "type": "text", + "primaryKey": false, + "notNull": false + }, + "config_checksum": { + "name": "config_checksum", + "type": "text", + "primaryKey": false, + "notNull": false + }, + "config_yaml": { + "name": "config_yaml", + "type": "text", + "primaryKey": false, + "notNull": false + }, + "last_heartbeat_at": { + "name": "last_heartbeat_at", + "type": "timestamp with time zone", + "primaryKey": false, + "notNull": false + }, + "status": { + "name": "status", + "type": "watcher_status", + "typeSchema": "public", + "primaryKey": false, + "notNull": true, + "default": "'registered'" + }, + "created_at": { + "name": "created_at", + "type": "timestamp with time zone", + "primaryKey": false, + "notNull": true, + "default": "now()" + }, + "updated_at": { + "name": "updated_at", + "type": "timestamp with time zone", + "primaryKey": false, + "notNull": true, + "default": "now()" + }, + "deleted_at": { + "name": "deleted_at", + "type": "timestamp with time zone", + "primaryKey": false, + "notNull": false + } + }, + "indexes": { + "idx_watchers_instrument_id": { + "name": "idx_watchers_instrument_id", + "columns": [ + { + "expression": "instrument_id", + "isExpression": false, + "asc": true, + "nulls": "last" + } + ], + "isUnique": false, + "where": "\"watchers\".\"deleted_at\" is null", + "concurrently": false, + "method": "btree", + "with": {} + } + }, + "foreignKeys": { + "watchers_instrument_id_instruments_id_fk": { + "name": "watchers_instrument_id_instruments_id_fk", + "tableFrom": "watchers", + "tableTo": "instruments", + "columnsFrom": [ + "instrument_id" + ], + "columnsTo": [ + "id" + ], + "onDelete": "no action", + "onUpdate": "no action" + } + }, + "compositePrimaryKeys": {}, + "uniqueConstraints": {}, + "policies": {}, + "checkConstraints": {}, + "isRLSEnabled": false + } + }, + "enums": { + "public.file_category": { + "name": "file_category", + "schema": "public", + "values": [ + "raw", + "processed" + ] + }, + "public.file_status": { + "name": "file_status", + "schema": "public", + "values": [ + "detected", + "upload_requested", + "uploaded", + "processing", + "completed", + "failed" + ] + }, + "public.instrument_run_source": { + "name": "instrument_run_source", + "schema": "public", + "values": [ + "lambda", + "watcher" + ] + }, + "public.instrument_status": { + "name": "instrument_status", + "schema": "public", + "values": [ + "pending", + "active", + "inactive" + ] + }, + "public.instrument_type": { + "name": "instrument_type", + "schema": "public", + "values": [ + "generic", + "plate_reader", + "gel_doc", + "qpcr", + "tape_station", + "hina_microscope" + ] + }, + "public.upload_mode": { + "name": "upload_mode", + "schema": "public", + "values": [ + "auto", + "manual" + ] + }, + "public.watcher_event_type": { + "name": "watcher_event_type", + "schema": "public", + "values": [ + "watcher_started", + "watcher_stopped", + "file_uploaded", + "upload_failed", + "run_reported", + "config_synced", + "error" + ] + }, + "public.watcher_status": { + "name": "watcher_status", + "schema": "public", + "values": [ + "registered", + "watching", + "stopped" + ] + } + }, + "schemas": {}, + "sequences": {}, + "roles": {}, + "policies": {}, + "views": {}, + "_meta": { + "columns": {}, + "schemas": {}, + "tables": {} + } +} \ No newline at end of file diff --git a/web-app/drizzle/meta/_journal.json b/web-app/drizzle/meta/_journal.json index 18af47d4..ad5e207d 100644 --- a/web-app/drizzle/meta/_journal.json +++ b/web-app/drizzle/meta/_journal.json @@ -64,6 +64,13 @@ "when": 1776986418967, "tag": "0008_chilly_princess_powerful", "breakpoints": true + }, + { + "idx": 9, + "version": "7", + "when": 1777404985691, + "tag": "0009_clammy_tyger_tiger", + "breakpoints": true } ] } \ No newline at end of file diff --git a/web-app/lib/db/schema.ts b/web-app/lib/db/schema.ts index 61644494..6accb0c9 100644 --- a/web-app/lib/db/schema.ts +++ b/web-app/lib/db/schema.ts @@ -441,6 +441,13 @@ export const files = pgTable( withTimezone: true, mode: "date", }), + // On-disk creation time of the file, reported by the watcher + // (st_birthtime where available, else st_mtime). NULL for + // Lambda-created files and rows predating this column. + fileCreatedAt: timestamp("file_created_at", { + withTimezone: true, + mode: "date", + }), createdAt: timestamp("created_at", { withTimezone: true, mode: "date",