diff --git a/.github/workflows/apply-migrations.yml b/.github/workflows/apply-migrations.yml new file mode 100644 index 00000000..d654833e --- /dev/null +++ b/.github/workflows/apply-migrations.yml @@ -0,0 +1,39 @@ +name: Apply database migrations + +on: + push: + branches: + - staging + - production + paths: + - .github/workflows/apply-migrations.yml + - web/drizzle/** + +# Serialize migration runs per environment so two merges can't apply +# migrations against the same Render database concurrently. +concurrency: + group: apply-migrations-${{ github.ref_name }} + cancel-in-progress: false + +jobs: + apply-migrations: + runs-on: ubuntu-latest + environment: ${{ github.ref_name }} + + env: + DATABASE_URL: ${{ secrets.DATABASE_URL }} + + steps: + - uses: actions/checkout@v6 + + - uses: actions/setup-node@v6 + with: + node-version: 24 + + - name: Install packages + run: npm ci + working-directory: web + + - name: Apply Drizzle migrations + run: npm run db:migrate + working-directory: web diff --git a/.gitignore b/.gitignore index 08791dcb..b8b61224 100644 --- a/.gitignore +++ b/.gitignore @@ -201,7 +201,7 @@ cython_debug/ # that can be found at https://github.com/github/gitignore/blob/main/Global/VisualStudioCode.gitignore # and can be added to the global gitignore or merged into this file. However, if you prefer, # you could uncomment the following to ignore the entire vscode folder -# .vscode/ +.vscode/ # Ruff stuff: .ruff_cache/ diff --git a/developer-docs/getting-started.md b/developer-docs/getting-started.md index 03d68430..c3bd785e 100644 --- a/developer-docs/getting-started.md +++ b/developer-docs/getting-started.md @@ -49,7 +49,6 @@ vercel env pull | `AWS_ROLE_ARN` | No | IAM role ARN for Vercel OIDC federation. Used to presign S3 URLs and SigV4-sign Lambda Function URL invocations (only needed on Vercel) | | `S3_RAW_DATA_BUCKET` | No | S3 bucket for raw data uploads (defaults to `arcadia-data-hub-raw-staging`) | | `LAMBDA_FUNCTION_URL` | No | Lambda Function URL. Required for file reprocessing and run-archive downloads. | -| `SLACK_WEBHOOK_URL` | No | Slack incoming webhook URL — when set, the web app posts a channel notification each time a new run is created | | `SLACK_BOT_TOKEN` | No | Slack bot token (`xoxb-…`) — required for personal Slack DM notifications | | `SLACK_CLIENT_ID` | No | Slack app client ID — required for the "Connect to Slack" OAuth flow on Settings > Notifications | | `SLACK_CLIENT_SECRET` | No | Slack app client secret — required for the OAuth flow | diff --git a/developer-docs/local-development.md b/developer-docs/local-development.md index 09393b4c..19da1993 100644 --- a/developer-docs/local-development.md +++ b/developer-docs/local-development.md @@ -60,7 +60,6 @@ LOCAL_S3_MIRROR=../lambda/.local-s3 Explicitly **do not** set the following — leaving them unset is what makes the relevant features short-circuit cleanly: - `LAMBDA_FUNCTION_URL` — file reprocessing and "Download all" buttons surface a 503 / "Lambda not configured" message instead of trying to invoke a Function URL. -- `SLACK_WEBHOOK_URL` — `sendSlackMessage()` in `web/lib/slack.ts` becomes a no-op with a single warn line. - `SLACK_BOT_TOKEN`, `SLACK_CLIENT_ID`, `SLACK_CLIENT_SECRET` — Slack DM/OAuth features are disabled when unset; the Settings > Notifications page renders a "Connect to Slack" button that is inert without these. - `AUTH_GOOGLE_ID` / `AUTH_GOOGLE_SECRET` — Google sign-in is unused locally; the dev Credentials provider handles auth. - `AWS_ROLE_ARN` — Vercel OIDC federation is for production. The local AWS SDK falls back to the static credentials above. @@ -132,7 +131,7 @@ Some features depend on services that aren't running in this workflow. Each one | File upload (from watcher) | `request-upload-url` returns a same-origin URL routed to `/api/local-s3/...`; `PUT` writes bytes into the mirror | Same | | Run archive ("Download all") | 503 "Archive builder is not configured" | Set `LAMBDA_FUNCTION_URL` + `S3_ARCHIVES_BUCKET` and grant `lambda:InvokeFunctionUrl` | | File reprocessing | The reprocess endpoint returns null and no Lambda is invoked | Same | -| Slack channel notifications on new runs | `console.warn` only, no HTTP call | Set `SLACK_WEBHOOK_URL` | +| Slack channel notifications on new runs | `console.warn` only, no HTTP call | Configure an incoming webhook URL in Settings > Notifications > Slack channel (admins only) | | Slack DM notifications / Connect to Slack | `console.warn` only; the "Connect to Slack" button redirects to Slack but the callback will error without credentials | Set `SLACK_BOT_TOKEN`, `SLACK_CLIENT_ID`, `SLACK_CLIENT_SECRET` | | Watcher uploads → Lambda → API loop | Not exercised end-to-end; the seed inserts the resulting rows directly. For Lambda-only smoke testing, see [Testing the Lambda end-to-end](#testing-the-lambda-end-to-end) below | Run the watcher (`reference/watcher.md`) and the Lambda (`reference/lambda.md`) end-to-end | | Sign in with Google | The button still renders but OAuth callback will 4xx without `AUTH_GOOGLE_ID` / `AUTH_GOOGLE_SECRET` | `vercel env pull` per `getting-started.md` | diff --git a/developer-docs/ops/ci-and-deployment.md b/developer-docs/ops/ci-and-deployment.md index c139092f..e4bc4a2e 100644 --- a/developer-docs/ops/ci-and-deployment.md +++ b/developer-docs/ops/ci-and-deployment.md @@ -2,7 +2,7 @@ ## GitHub Actions -Four workflows run on pushes to `staging`/`production` and on pull requests targeting those branches. A fifth (`publish-watcher.yml`) runs only on `watcher-v*` tag pushes and manual dispatch. +Four workflows run on pushes to `staging`/`production` and on pull requests targeting those branches. A fifth (`apply-migrations.yml`) runs on merges to `staging`/`production` that touch migration files, and a sixth (`publish-watcher.yml`) runs only on `watcher-v*` tag pushes and manual dispatch. ### Python lint and typecheck (`python-lint.yml`) @@ -28,6 +28,12 @@ Four workflows run on pushes to `staging`/`production` and on pull requests targ - `make fe-test-mcp` — runs in-memory MCP protocol tests (mocked data layer, no database). - `make fe-test-integration` — runs Vitest integration tests that test the API routes and MCP server over HTTP against a real database. +### Apply database migrations (`apply-migrations.yml`) + +Triggered on pushes to `staging`/`production` (i.e. PR merges) that change files under `web/drizzle/`. The single job sets `environment: ${{ github.ref_name }}` so GitHub selects that environment's secrets and protection rules, then runs `npm run db:migrate` (Drizzle) against the environment's Render database using the environment's `DATABASE_URL` secret. A per-branch `concurrency` group prevents overlapping migration runs. + +Production is gated by a required-reviewer protection rule on the `production` GitHub environment, so production migrations pause for manual approval before applying. Each environment needs a `DATABASE_URL` secret pointing at its Render connection string, and the Render database must accept connections from GitHub-hosted runners. + ### Publish watcher (`publish-watcher.yml`) Triggered on `watcher-v*` tag pushes and manual `workflow_dispatch` from `production`. The `build` job's `if:` guard refuses dispatches from any other branch so a feature branch can't accidentally publish whatever version is in its `pyproject.toml`. Builds the `data-hub-watcher` package, publishes it to PyPI via OIDC trusted publishing, and verifies the upload by installing the freshly published wheel into a clean venv. Three sequential jobs: @@ -64,7 +70,7 @@ vercel env pull Staging and production each have a dedicated PostgreSQL instance hosted on [Render](https://dashboard.render.com/project/prj-d75d0jma2pns738r4110). -Schema changes are applied with Drizzle: +Merges to `staging`/`production` that change files under `web/drizzle/` automatically apply migrations via the [`apply-migrations.yml`](#apply-database-migrations-apply-migrationsyml) workflow (production is gated on manual approval). The commands below are for local runs or manual application: ```sh cd web @@ -167,7 +173,7 @@ In your GitHub repo, go to **Settings → Environments**, create a `staging` env | `DATA_HUB_API_URL` | Base API URL for the environment | | `DATA_HUB_API_KEY` | API key for Lambda → Data Hub authentication (also used by the Lambda's archive-job PATCH callback) | -Slack notifications are sent by the **web app** (not the Lambda) when a new run is created. Configure `SLACK_WEBHOOK_URL` per environment in the Vercel dashboard alongside the other web app env vars listed below. +Slack channel notifications are sent by the **web app** (not the Lambda) when a new run is created. Workspace admins configure the incoming webhook URL in Settings > Notifications > Slack channel (stored in the `slack_channel_config` DB table). After deploying, paste the webhook URL once in that UI before removing any legacy `SLACK_WEBHOOK_URL` env var from Vercel. You'll also need the `WebAppRoleArn` and `DataHubFunctionUrl` stack outputs to configure the Vercel web app. In the Vercel dashboard (under the appropriate environment), set: diff --git a/developer-docs/reference/lambda.md b/developer-docs/reference/lambda.md index dd523894..d7b04b71 100644 --- a/developer-docs/reference/lambda.md +++ b/developer-docs/reference/lambda.md @@ -54,7 +54,7 @@ Each processor module exposes a `process_file()` function that accepts the run I ## Slack notifications -Slack notifications are sent by the **web app** (`web/lib/slack.ts`), not the Lambda. When the Lambda's `process_file` calls `POST /api/v1/instruments/:instrumentId/runs` to register a newly-detected run, that endpoint posts a single message per run to `SLACK_WEBHOOK_URL` (configured per environment in Vercel). Subsequent files for the same run do not re-notify because the upsert is idempotent on `(instrument_id, run_id)`. File-level failures remain visible in the web app via the file row's `status='failed'` and `error_message` fields. +Slack channel notifications are sent by the **web app** (`web/lib/slack.ts`), not the Lambda. When the Lambda's `process_file` calls `POST /api/v1/instruments/:instrumentId/runs` to register a newly-detected run, that endpoint posts a single message per run to the incoming webhook URL configured in Settings > Notifications > Slack channel (workspace admins only). Subsequent files for the same run do not re-notify because the upsert is idempotent on `(instrument_id, run_id)`. File-level failures remain visible in the web app via the file row's `status='failed'` and `error_message` fields. ## Adding a new instrument diff --git a/developer-docs/reference/watcher.md b/developer-docs/reference/watcher.md index 13d9e03b..bf86aa62 100644 --- a/developer-docs/reference/watcher.md +++ b/developer-docs/reference/watcher.md @@ -67,8 +67,9 @@ While running: - **File monitor** watches the directory for new/modified files using `watchdog` and waits for each file to stabilize (size + mtime unchanged for the configured stability period). Files that keep changing for longer than 5 minutes are abandoned and surface as a `stability_timeout` error event. - **Run detector** groups stable files into runs by applying the configured regex to each file's relative path. The first file for a run triggers `POST /instruments/:id/runs`; subsequent files for the same run incrementally `PATCH` only the new entries onto the manifest. Files inside the watch tree that don't match the pattern emit a `pattern_mismatch` event (throttled to one per parent directory) so misconfigured patterns surface in the dashboard. -- **Uploader** requests a presigned S3 URL from the API and uploads each file via HTTP PUT (auto mode), or polls the server's upload queue (manual mode). The watcher does not need AWS credentials. Each upload retries up to 3 times with exponential backoff (1, 2, 4 s) and is recorded locally with its SHA-256 so retries and restarts don't re-upload the same bytes. In manual mode, queue-poll failures are throttled (1st failure, then every 10th) to keep a sustained outage visible without flooding the events stream. -- **Heartbeat loop** sends periodic heartbeats (every 60 seconds) to the API. The payload includes the watcher version, instrument ID, watch directory, upload mode, per-interval activity counters, and process uptime; a final `status="stopped"` heartbeat is sent on graceful shutdown. In manual mode, the tick also polls the upload queue. +- **Uploader** requests a presigned S3 URL from the API and uploads each file via HTTP PUT (auto mode), or processes the server's upload queue (manual mode). The watcher does not need AWS credentials. Each upload retries up to 3 times with exponential backoff (1, 2, 4 s) and is recorded locally with its SHA-256 so retries and restarts don't re-upload the same bytes. In manual mode, queue-poll failures are throttled (1st failure, then every 10th) to keep a sustained outage visible without flooding the events stream. +- **Upload worker** (manual mode only) polls the server's upload queue on its own long-lived thread every 60 seconds, decoupled from the heartbeat so a slow or large upload can't delay heartbeats and make a busy watcher look offline. On shutdown it is stopped and joined before the state DB is closed. Auto mode has no worker: uploads run on the monitor's stability-checker thread via the run detector's upload callback. +- **Heartbeat loop** sends periodic heartbeats (every 60 seconds) to the API. The payload includes the watcher version, instrument ID, watch directory, upload mode, per-interval activity counters, and process uptime; a final `status="stopped"` heartbeat is sent on graceful shutdown. - **Event reporter** batches and flushes lifecycle events (started, stopped, file uploaded, errors) to the API. See [Observability](#observability) for the full taxonomy. - **Auto-updater** runs from the same heartbeat tick on every platform — not only Windows services. It polls `GET /watchers/:id/update-check` roughly hourly and applies new releases when the watcher has been idle long enough not to clobber an in-flight run. The full activity-window guard, mandatory-update behavior, and rollback flow are documented in [Upgrading the watcher](../guides/upgrading-the-watcher.md); auto-update is hard-disabled in the `preview` environment. @@ -207,12 +208,12 @@ Upgrading an existing watcher is unaffected: the environment's database already ### Upload modes - **`auto`**: Files are uploaded to S3 immediately after run detection. -- **`manual`**: Runs are reported to the API without uploading. The server decides which files to upload via a queue, polled on each heartbeat tick. Useful when uploads need human approval. +- **`manual`**: Runs are reported to the API without uploading. The server decides which files to upload via a queue, polled by the upload worker thread every 60 seconds. Useful when uploads need human approval. Queued files are resolved against the current `watch_directory` (each queue entry carries a `relative_path` anchored to the root that was active when the file was detected). Two safeguards keep a stale queue entry from erroring forever (ENG-1397): - **On `watch_directory` change**: the server reverts every pending upload request for that instrument back to `detected` (clearing `upload_requested_at`) as soon as the new config is pushed, so the queue drains immediately. The reverted files remain re-requestable detections; an operator can queue them again from their new location. -- **Per-file 3-try cap (`MAX_QUEUE_FILE_ATTEMPTS`)**: a queued file that keeps failing — missing on disk or failing to upload — is retried on at most three heartbeat polls. After that the watcher cancels the request server-side (revert to `detected`) so the file leaves the queue instead of re-erroring each tick. The attempt count resets on watcher restart, so a transient outage longer than three ticks is recovered on the next start. +- **Per-file 3-try cap (`MAX_QUEUE_FILE_ATTEMPTS`)**: a queued file that keeps failing — missing on disk or failing to upload — is retried on at most three upload-queue polls. After that the watcher cancels the request server-side (revert to `detected`) so the file leaves the queue instead of re-erroring each poll. The attempt count resets on watcher restart, so a transient outage longer than three polls is recovered on the next start. ## Local state diff --git a/uv.lock b/uv.lock index a532adb1..b0e94c01 100644 --- a/uv.lock +++ b/uv.lock @@ -416,7 +416,7 @@ requires-dist = [ [[package]] name = "data-hub-watcher" -version = "0.3.0" +version = "0.4.0" source = { editable = "watcher" } dependencies = [ { name = "click" }, diff --git a/watcher/pyproject.toml b/watcher/pyproject.toml index 242d2a00..7348515c 100644 --- a/watcher/pyproject.toml +++ b/watcher/pyproject.toml @@ -1,6 +1,6 @@ [project] name = "data-hub-watcher" -version = "0.3.0" +version = "0.4.0" description = "File-watcher agent for lab instrument PCs that ingests data into Data Hub." readme = "README.md" requires-python = ">=3.12" diff --git a/watcher/src/data_hub_watcher/constants.py b/watcher/src/data_hub_watcher/constants.py index 9c0a4abe..4c5cef79 100644 --- a/watcher/src/data_hub_watcher/constants.py +++ b/watcher/src/data_hub_watcher/constants.py @@ -71,6 +71,12 @@ def _resolve_watcher_log_dir() -> Path: SUPPORTED_ENVIRONMENTS: tuple[str, ...] = ("staging", "production", "preview") HEARTBEAT_INTERVAL_SECONDS = 60 +# Kept separate from ``HEARTBEAT_INTERVAL_SECONDS`` so the poll cadence can +# diverge now that uploads no longer ride the heartbeat tick. +UPLOAD_POLL_INTERVAL_SECONDS = 60 +# Bounded so a service stop doesn't hang on a large in-flight PUT; past this, +# shutdown stops waiting for the worker and leaves the state DB open. +UPLOAD_WORKER_STOP_TIMEOUT_SECONDS = 30 DEFAULT_STABILITY_PERIOD_SECONDS = 5 # Built-in presets for the ``init`` / ``config edit`` wizard. diff --git a/watcher/src/data_hub_watcher/runtime.py b/watcher/src/data_hub_watcher/runtime.py index 16e78cf2..3d41d206 100644 --- a/watcher/src/data_hub_watcher/runtime.py +++ b/watcher/src/data_hub_watcher/runtime.py @@ -22,6 +22,7 @@ DEFAULT_CONFIG_DIR, HEARTBEAT_INTERVAL_SECONDS, PRUNE_DAYS, + UPLOAD_WORKER_STOP_TIMEOUT_SECONDS, ) from data_hub_watcher.events import EventReporter, EventType, WatcherEvent from data_hub_watcher.heartbeat import HeartbeatLoop, WatcherCounters @@ -35,7 +36,7 @@ clear_upgrade_result, read_upgrade_result, ) -from data_hub_watcher.uploader import Uploader +from data_hub_watcher.uploader import Uploader, UploadQueueWorker logger = logging.getLogger(__name__) @@ -74,6 +75,10 @@ class WatcherRuntime: # in their stop wait so a shutdown can be triggered from any thread. shutdown_event: threading.Event = field(default_factory=threading.Event) upgrade_restart_event: threading.Event = field(default_factory=threading.Event) + # Manual mode only: the thread that polls the upload queue off the + # heartbeat. ``None`` in auto mode, where uploads run on the monitor's + # stability-checker thread via the run detector's callback. + upload_worker: UploadQueueWorker | None = None @dataclass(frozen=True) @@ -251,6 +256,11 @@ def _request_upgrade_restart(target_version: str) -> None: is_auto = inst.upload_mode == "auto" + # Shared with the manual-mode ``UploadQueueWorker`` so a shutdown can + # interrupt an in-flight upload's backoff and abort between queued files; + # unused (but harmless) in auto mode. + upload_stop_event = threading.Event() + uploader = Uploader( client=client, state_db=state_db, @@ -262,8 +272,13 @@ def _request_upgrade_restart(target_version: str) -> None: # Per-instrument knob, defaulted on the model so older configs # transparently inherit the new parallel-upload behaviour. upload_parallelism=inst.upload_parallelism, + stop_event=upload_stop_event, ) + # Manual mode polls the server queue on its own thread; auto mode uploads + # via the run detector's callback on the stability-checker thread instead. + upload_worker = None if is_auto else UploadQueueWorker(uploader, stop_event=upload_stop_event) + detector = RunDetector( pattern=inst.run_detection.pattern, instrument_id=inst.id, @@ -287,19 +302,10 @@ def _request_upgrade_restart(target_version: str) -> None: seed_baseline=seed_baseline, ) - # The heartbeat's `on_tick` hook is now multi-purpose: - # 1. In manual mode, poll the server's upload queue (uploads - # naturally inherit the heartbeat cadence). - # 2. Always: feed the in-process auto-updater so it can count - # idle ticks and run a server update-check roughly hourly. - # Each side wraps its own try/except so a failure on one side - # never blocks the other. + # Feeds the in-process auto-updater on every tick. Manual-mode upload + # polling used to run here too but moved to ``UploadQueueWorker`` so a + # slow upload can't delay a heartbeat. def _on_tick() -> None: - if not is_auto: - try: - uploader.poll_upload_queue() - except Exception: - logger.exception("Upload queue poll failed") try: updater.on_tick() except Exception: @@ -329,6 +335,7 @@ def _on_tick() -> None: config_dir=effective_config_dir, shutdown_event=shutdown_event, upgrade_restart_event=upgrade_restart_event, + upload_worker=upload_worker, ) @@ -480,6 +487,10 @@ def start_runtime(rt: WatcherRuntime, *, started_message: str) -> None: rt.detector.hydrate_from_state_db() rt.heartbeat.start() + # Start manual-mode upload polling before the (potentially long) initial + # scan so queued uploads keep draining while the scan walks the backlog. + if rt.upload_worker is not None: + rt.upload_worker.start() rt.monitor.start() @@ -551,6 +562,17 @@ def sync_config_to_api( def stop_runtime(rt: WatcherRuntime, *, stopped_message: str) -> None: """Shut everything down in reverse order and flush pending events.""" rt.monitor.stop() + # ``StateDB.close`` assumes writer threads have joined; a large upload + # outliving the join once wrote post-close. Stop the worker first, and if + # it won't stop in time, skip the close rather than race it (the OS reaps). + upload_worker_stopped = True + if rt.upload_worker is not None: + upload_worker_stopped = rt.upload_worker.stop(timeout=UPLOAD_WORKER_STOP_TIMEOUT_SECONDS) + if not upload_worker_stopped: + logger.warning( + "Upload worker still running at shutdown; leaving the state DB " + "open so the in-flight upload can finish without a closed-DB error" + ) rt.reporter.queue_event( WatcherEvent( event_type=EventType.WATCHER_STOPPED, @@ -559,4 +581,5 @@ def stop_runtime(rt: WatcherRuntime, *, stopped_message: str) -> None: ) rt.heartbeat.stop() rt.reporter.flush() - rt.state_db.close() + if upload_worker_stopped: + rt.state_db.close() diff --git a/watcher/src/data_hub_watcher/uploader.py b/watcher/src/data_hub_watcher/uploader.py index a593e87f..9ed5525b 100644 --- a/watcher/src/data_hub_watcher/uploader.py +++ b/watcher/src/data_hub_watcher/uploader.py @@ -22,6 +22,7 @@ from data_hub_watcher.api_client import ApiError, DataHubClient from data_hub_watcher.constants import ( MAX_QUEUE_FILE_ATTEMPTS, + UPLOAD_POLL_INTERVAL_SECONDS, UPLOAD_RETRY_BASE_DELAY, UPLOAD_RETRY_MAX, ) @@ -105,6 +106,7 @@ def __init__( watcher_id: str, watch_directory: Path, upload_parallelism: int = 1, + stop_event: threading.Event | None = None, ) -> None: self._client = client self._state_db = state_db @@ -116,18 +118,22 @@ def __init__( if upload_parallelism < 1: raise ValueError(f"upload_parallelism must be >= 1, got {upload_parallelism}") self._parallelism = upload_parallelism + # When set (by the owning ``UploadQueueWorker``), a shutdown can + # interrupt the retry backoff and abort between queued files; ``None`` + # for the one-shot ``upload`` CLI path, which never races a teardown. + self._stop_event = stop_event # Track consecutive upload-queue poll failures so the watcher # surfaces a ``kind=upload_queue_poll_failed`` event on the # 1st failure and every 10th repeat. The unthrottled case - # would emit one event per heartbeat tick during an outage, - # crowding out other signals on the dashboard. - # Mutated only from the heartbeat thread (manual mode), so + # would emit one event per poll during an outage, crowding out + # other signals on the dashboard. + # Mutated only from the upload worker thread (manual mode), so # not under any explicit lock. self._consecutive_queue_poll_failures = 0 # Per-file upload-queue attempt bookkeeping, keyed by server file id. # Bounds retries before giving up (see ``_process_queued_file``) and # doubles as the emit-once throttle for the missing-file error. Pruned - # each poll, so a re-requested id starts fresh. Heartbeat thread only. + # each poll, so a re-requested id starts fresh. Upload worker thread only. self._queue_attempts: dict[int, _QueueAttempt] = {} # Single ``requests.Session`` shared across every S3 PUT # (parallel or serial). Keeps TLS connections alive between @@ -230,10 +236,15 @@ def upload_files(self, run_id: str, files: list[FileInfo]) -> int: # Manual-mode: poll the server queue # ------------------------------------------------------------------ + def _stop_requested(self) -> bool: + return self._stop_event is not None and self._stop_event.is_set() + def poll_upload_queue(self) -> None: """Fetch the upload queue and process each file. - Intended to be called on heartbeat ticks in manual mode. + Driven by the manual-mode ``UploadQueueWorker`` loop on its own + thread, decoupled from the heartbeat so a slow upload can't starve + the liveness signal. """ try: queue = self._client.get_upload_queue(self._watcher_id) @@ -278,13 +289,19 @@ def poll_upload_queue(self) -> None: logger.info("Upload queue has %d file(s)", len(queue.files)) for qf in queue.files: + # Bail between files on shutdown so the worker's ``stop()`` can + # join promptly instead of draining the whole queue; the + # remaining files are picked up on the next start's poll. + if self._stop_requested(): + logger.info("Stop requested; deferring %d queued file(s)", len(queue.files)) + break self._process_queued_file(qf) def _process_queued_file(self, qf: UploadQueueFile) -> None: - """Attempt one queued file, bounding retries across heartbeat polls. + """Attempt one queued file, bounding retries across polls. - Manual-mode polling runs every heartbeat, so a file that can't be - uploaded -- missing on disk after a watch-directory change, or a + Manual-mode polling repeats on the worker's cadence, so a file that + can't be uploaded -- missing on disk after a watch-directory change, or a persistent upload error -- would otherwise re-error forever. We cap attempts at ``MAX_QUEUE_FILE_ATTEMPTS`` and then cancel the request server-side (revert to ``detected``) so it leaves the queue. The @@ -509,6 +526,7 @@ def _upload_single(self, path: Path, run_id: str) -> bool: return True last_exc: Exception | None = None + put_ok = False # Exponential backoff: 1s, 2s, 4s. Retries protect against transient # network errors common on lab-PC networks. @@ -516,6 +534,7 @@ def _upload_single(self, path: Path, run_id: str) -> bool: for attempt in range(UPLOAD_RETRY_MAX): try: self._put_to_presigned_url(presigned.upload_url, path, content_type) + put_ok = True break except Exception as exc: last_exc = exc @@ -528,8 +547,21 @@ def _upload_single(self, path: Path, run_id: str) -> bool: exc, delay, ) - time.sleep(delay) - else: + # Wait on the stop event when present so a shutdown cuts the + # backoff short; falls back to a plain sleep for the one-shot + # ``upload`` path that has no worker/event. + if self._stop_event is not None: + self._stop_event.wait(delay) + else: + time.sleep(delay) + # Abandon the remaining retries on shutdown. Returning False (not a + # hard failure event) leaves the request pending so the next start + # re-uploads it, rather than recording a spurious upload error. + if self._stop_requested(): + logger.info("Stop requested mid-upload; deferring %s", path.name) + return False + + if not put_ok: logger.error("Upload failed after %d attempts: %s", UPLOAD_RETRY_MAX, path.name) self._reporter.queue_event( WatcherEvent( @@ -583,3 +615,59 @@ def _upload_single(self, path: Path, run_id: str) -> bool: ) logger.info("Uploaded %s → s3://%s/%s", path.name, s3_bucket, s3_key) return True + + +class UploadQueueWorker: + """Polls the manual-mode upload queue on a dedicated long-lived thread. + + Uploads used to run on the heartbeat tick, so a slow or large transfer + delayed heartbeats and the dashboard flagged a busy watcher as offline. + Owning the poll loop here keeps the heartbeat free, and the shared + ``stop_event`` lets a shutdown interrupt an upload so ``stop()`` can join + before ``StateDB.close`` runs (which assumes writer threads have joined). + """ + + def __init__( + self, + uploader: Uploader, + *, + stop_event: threading.Event, + interval_seconds: int = UPLOAD_POLL_INTERVAL_SECONDS, + ) -> None: + self._uploader = uploader + self._stop_event = stop_event + self._interval = interval_seconds + self._thread: threading.Thread | None = None + + def start(self) -> None: + self._stop_event.clear() + self._thread = threading.Thread(target=self._run, daemon=True, name="upload-worker") + self._thread.start() + + def stop(self, timeout: float | None = None) -> bool: + """Signal the loop and join it. Returns whether the thread exited. + + A ``False`` return means an upload is still in flight past *timeout* + (a single S3 PUT can run up to its request timeout); the caller uses + this to avoid closing the state DB out from under the live upload. + """ + self._stop_event.set() + if self._thread is None: + return True + self._thread.join(timeout=timeout) + return not self._thread.is_alive() + + def _run(self) -> None: + # Wait first (parity with the previous heartbeat-driven cadence), + # then poll each interval until stop. + while not self._stop_event.wait(timeout=self._interval): + self._poll_once() + + def _poll_once(self) -> None: + # ``poll_upload_queue`` already handles and reports poll failures; this + # guard only keeps an unexpected error from killing the thread and + # silently ending all future polls. + try: + self._uploader.poll_upload_queue() + except Exception: + logger.exception("Upload queue poll failed") diff --git a/watcher/tests/test_runtime.py b/watcher/tests/test_runtime.py index 3b77dc35..fb3380a9 100644 --- a/watcher/tests/test_runtime.py +++ b/watcher/tests/test_runtime.py @@ -6,11 +6,13 @@ forgot to wire `on_tick` on the `HeartbeatLoop`. These tests lock in the wiring contract per `upload_mode` so any future drift fails loudly: -* auto mode -> `detector._upload_cb` is `uploader.upload_files` - and `heartbeat._on_tick` ticks the auto-updater only -* manual mode -> `detector._upload_cb` is `None` - and `heartbeat._on_tick` polls `uploader.poll_upload_queue` - *and* ticks the auto-updater +* auto mode -> `detector._upload_cb` is `uploader.upload_files`, + `heartbeat._on_tick` ticks the auto-updater only, and + `rt.upload_worker` is `None` +* manual mode -> `detector._upload_cb` is `None`, `heartbeat._on_tick` + ticks the auto-updater only (uploads now run on the + dedicated `UploadQueueWorker` thread, not the heartbeat), + and `rt.upload_worker` is set """ from __future__ import annotations @@ -29,10 +31,12 @@ from data_hub_watcher.run_detector import RunDetector from data_hub_watcher.runtime import ( ShutdownReason, + WatcherRuntime, _summarize_worker_failure, build_runtime, classify_shutdown, start_runtime, + stop_runtime, ) from data_hub_watcher.state import StateDB from data_hub_watcher.updater import Updater, write_upgrade_marker @@ -116,9 +120,20 @@ def test_heartbeat_on_tick_only_drives_updater(self, tmp_path: Path, db_path: Pa finally: rt.state_db.close() + def test_auto_mode_has_no_upload_worker(self, tmp_path: Path, db_path: Path) -> None: + cfg = _make_config(tmp_path, upload_mode="auto") + rt = build_runtime(client=MagicMock(), cfg=cfg, db_path=db_path) + + try: + # Auto-mode uploads run on the monitor's stability-checker thread + # via the detector callback, so there is no upload-queue worker. + assert rt.upload_worker is None + finally: + rt.state_db.close() + class TestBuildRuntimeManualMode: - """Manual mode: heartbeat polls the upload queue, detector does not upload.""" + """Manual mode: a dedicated worker polls the upload queue off the heartbeat.""" def test_detector_upload_callback_is_none(self, tmp_path: Path, db_path: Path) -> None: cfg = _make_config(tmp_path, upload_mode="manual") @@ -131,44 +146,37 @@ def test_detector_upload_callback_is_none(self, tmp_path: Path, db_path: Path) - finally: rt.state_db.close() - def test_heartbeat_on_tick_polls_upload_queue(self, tmp_path: Path, db_path: Path) -> None: + def test_heartbeat_on_tick_does_not_poll_upload_queue( + self, tmp_path: Path, db_path: Path + ) -> None: cfg = _make_config(tmp_path, upload_mode="manual") rt = build_runtime(client=MagicMock(), cfg=cfg, db_path=db_path) try: - # The heartbeat must call `uploader.poll_upload_queue` on every - # tick — this is the bug the runtime extraction was fixing. + # Uploads moved off the heartbeat thread onto the worker, so the + # tick must only feed the updater — a slow upload can no longer + # delay a heartbeat and make a busy watcher look offline. assert rt.heartbeat._on_tick is not None - rt.uploader.poll_upload_queue = MagicMock() # type: ignore[method-assign] rt.updater.on_tick = MagicMock(return_value=None) # type: ignore[method-assign] rt.heartbeat._on_tick() - rt.uploader.poll_upload_queue.assert_called_once_with() - # The same hook must also feed the auto-updater so its idle - # counter advances regardless of upload_mode. + rt.uploader.poll_upload_queue.assert_not_called() rt.updater.on_tick.assert_called_once_with() finally: rt.state_db.close() - def test_on_tick_swallows_poll_exceptions(self, tmp_path: Path, db_path: Path) -> None: - """Polling errors must not propagate out of the heartbeat tick, - otherwise one transient server blip kills the heartbeat thread - and the watcher goes silent until restart.""" + def test_manual_mode_builds_upload_worker_sharing_stop_event( + self, tmp_path: Path, db_path: Path + ) -> None: cfg = _make_config(tmp_path, upload_mode="manual") rt = build_runtime(client=MagicMock(), cfg=cfg, db_path=db_path) try: - rt.uploader.poll_upload_queue = MagicMock( # type: ignore[method-assign] - side_effect=RuntimeError("boom") - ) - rt.updater.on_tick = MagicMock(return_value=None) # type: ignore[method-assign] - assert rt.heartbeat._on_tick is not None - rt.heartbeat._on_tick() - rt.uploader.poll_upload_queue.assert_called_once_with() - # Updater must still tick even when the upload-queue poll - # blew up, otherwise a permanently-failing manual-mode - # poll would also disable auto-updates. - rt.updater.on_tick.assert_called_once_with() + # The worker must wrap this runtime's uploader and share its stop + # event so a shutdown can interrupt an in-flight upload. + assert rt.upload_worker is not None + assert rt.upload_worker._uploader is rt.uploader + assert rt.upload_worker._stop_event is rt.uploader._stop_event finally: rt.state_db.close() @@ -189,6 +197,62 @@ def test_on_tick_swallows_updater_exceptions(self, tmp_path: Path, db_path: Path rt.state_db.close() +class TestStopRuntimeUploadWorkerOrdering: + """`stop_runtime` must stop the upload worker before closing the state DB. + + Regression guard for the prod "Cannot operate on a closed database" race: + a still-running upload wrote to the DB after `close()` because teardown + didn't wait for the upload thread. + """ + + @staticmethod + def _runtime_with_mocks(state_db: MagicMock, upload_worker: Any) -> WatcherRuntime: + return WatcherRuntime( + state_db=state_db, + counters=MagicMock(), + reporter=MagicMock(), + uploader=MagicMock(), + detector=MagicMock(), + monitor=MagicMock(), + heartbeat=MagicMock(), + updater=MagicMock(), + config_dir=Path("/tmp"), + upload_worker=upload_worker, + ) + + def test_closes_db_when_worker_stops_cleanly(self) -> None: + state_db = MagicMock() + worker = MagicMock() + worker.stop.return_value = True + rt = self._runtime_with_mocks(state_db, worker) + + stop_runtime(rt, stopped_message="Watcher stopped") + + worker.stop.assert_called_once() + state_db.close.assert_called_once_with() + + def test_skips_close_when_worker_still_running(self) -> None: + # A large PUT outliving the join must not have the DB yanked out from + # under it; teardown leaves the connection open and lets the OS reap it. + state_db = MagicMock() + worker = MagicMock() + worker.stop.return_value = False + rt = self._runtime_with_mocks(state_db, worker) + + stop_runtime(rt, stopped_message="Watcher stopped") + + worker.stop.assert_called_once() + state_db.close.assert_not_called() + + def test_closes_db_in_auto_mode_without_worker(self) -> None: + state_db = MagicMock() + rt = self._runtime_with_mocks(state_db, None) + + stop_runtime(rt, stopped_message="Watcher stopped") + + state_db.close.assert_called_once_with() + + class TestBuildRuntimeSharedDependencies: """Cross-object wiring invariants that apply to both upload modes.""" diff --git a/watcher/tests/test_uploader.py b/watcher/tests/test_uploader.py index 6d83dc8c..43f1d08a 100644 --- a/watcher/tests/test_uploader.py +++ b/watcher/tests/test_uploader.py @@ -21,7 +21,7 @@ UploadQueueResponse, ) from data_hub_watcher.state import StateDB -from data_hub_watcher.uploader import Uploader +from data_hub_watcher.uploader import Uploader, UploadQueueWorker @pytest.fixture() @@ -742,3 +742,136 @@ def test_cancel_failure_is_retried_next_poll( # The next poll retries the cancel rather than re-erroring on upload. uploader.poll_upload_queue() assert mock_client.cancel_upload_request.call_count == 2 + + +class TestUploaderStopEvent: + """A shutdown must interrupt uploads without recording a spurious failure.""" + + def _uploader_with_stop( + self, + mock_client: MagicMock, + state_db: StateDB, + tmp_path: Path, + stop_event: threading.Event, + ) -> Uploader: + return Uploader( + client=mock_client, + state_db=state_db, + event_reporter=MagicMock(spec=EventReporter), + counters=WatcherCounters(), + instrument_id="test-instrument", + watcher_id="watcher-123", + watch_directory=tmp_path, + stop_event=stop_event, + ) + + def test_stop_during_backoff_defers_without_failure( + self, + mock_client: MagicMock, + state_db: StateDB, + tmp_file: Path, + tmp_path: Path, + ) -> None: + stop = threading.Event() + stop.set() # already stopping when the first attempt fails + up = self._uploader_with_stop(mock_client, state_db, tmp_path, stop) + mock_client.request_upload_url.return_value = PresignedUploadResponse( + upload_url="https://s3.example.com/presigned", + s3_bucket="test-bucket", + s3_key="k", + file_id=42, + expires_in=3600, + already_uploaded=False, + ) + + with patch.object( + Uploader, "_put_to_presigned_url", side_effect=ConnectionError("network down") + ) as mock_put: + result = up._upload_single(tmp_file, "RUN-001") + + # Deferred, not failed: one attempt, no success PATCH, no error event + # or counter bump, so the request stays pending for the next start. + assert result is False + assert mock_put.call_count == 1 + mock_client.mark_file_uploaded.assert_not_called() + assert up._counters.errors == 0 + cast(MagicMock, up._reporter).queue_event.assert_not_called() + + def test_poll_bails_between_files_when_stopping( + self, + mock_client: MagicMock, + state_db: StateDB, + tmp_path: Path, + ) -> None: + stop = threading.Event() + stop.set() + up = self._uploader_with_stop(mock_client, state_db, tmp_path, stop) + mock_client.get_upload_queue.return_value = UploadQueueResponse( + files=[ + UploadQueueFile( + id=1, + instrument_id="test-instrument", + run_id="R1", + filename="a.csv", + relative_path="a.csv", + ), + UploadQueueFile( + id=2, + instrument_id="test-instrument", + run_id="R1", + filename="b.csv", + relative_path="b.csv", + ), + ] + ) + + with patch.object(Uploader, "_process_queued_file") as mock_process: + up.poll_upload_queue() + + mock_process.assert_not_called() + + +class TestUploadQueueWorker: + """The worker owns the poll loop and must stop cleanly for shutdown.""" + + def test_poll_once_swallows_exceptions(self) -> None: + uploader = MagicMock() + uploader.poll_upload_queue.side_effect = RuntimeError("boom") + worker = UploadQueueWorker(uploader, stop_event=threading.Event()) + + # A poll blowup must not escape and kill the thread. + worker._poll_once() + + uploader.poll_upload_queue.assert_called_once_with() + + def test_stop_joins_idle_worker(self) -> None: + uploader = MagicMock() + worker = UploadQueueWorker(uploader, stop_event=threading.Event(), interval_seconds=60) + worker.start() + + # The loop waits on the stop event, so setting it returns the join + # immediately rather than after the 60s interval. + assert worker.stop(timeout=5) is True + + def test_stop_reports_false_when_upload_in_flight(self) -> None: + # A poll stuck mid-upload past the join timeout must report unfinished + # so teardown skips closing the state DB out from under it. + in_poll = threading.Event() + release = threading.Event() + + def blocking_poll() -> None: + in_poll.set() + release.wait(timeout=5) + + uploader = MagicMock() + uploader.poll_upload_queue.side_effect = blocking_poll + # interval=0 so the loop enters the (blocking) poll right away. + worker = UploadQueueWorker(uploader, stop_event=threading.Event(), interval_seconds=0) + worker.start() + assert in_poll.wait(timeout=5) + + try: + assert worker.stop(timeout=0.1) is False + finally: + # Let the blocked poll finish so the daemon thread exits cleanly. + release.set() diff --git a/web/.env.example b/web/.env.example index 5d271927..ccb75174 100644 --- a/web/.env.example +++ b/web/.env.example @@ -27,11 +27,6 @@ S3_ARCHIVES_BUCKET=arcadia-data-hub-archives-staging # chain (local dev). Required for run-archive downloads. LAMBDA_FUNCTION_URL= -# Slack incoming webhook URL for org-wide channel run-creation notifications. -# See: https://docs.slack.dev/messaging/sending-messages-using-incoming-webhooks/. -# Leave empty to disable. -SLACK_WEBHOOK_URL= - # Slack bot / personal DM notifications. # Create a Slack app at https://api.slack.com/apps, add a bot user with the # `chat:write` scope, install it to the workspace, and copy the bot token. diff --git a/web/app/api/v1/settings/slack-channel/route.ts b/web/app/api/v1/settings/slack-channel/route.ts new file mode 100644 index 00000000..930c177d --- /dev/null +++ b/web/app/api/v1/settings/slack-channel/route.ts @@ -0,0 +1,79 @@ +import type { NextRequest } from "next/server"; +import { requireAdmin } from "@/lib/api/auth"; +import { apiError, VALIDATION_ERROR } from "@/lib/api/errors"; +import { + getSlackChannelConfigForAdmin, + upsertSlackChannelWebhookUrl, +} from "@/lib/slack/channel-config"; +import { slackChannelWebhookPutBodySchema } from "@/lib/slack/webhook-url"; + +// Admin-only read/write of the singleton `slack_channel_config` row, +// edited via the "Slack channel" section on `/settings/notifications`. +// The webhook URL is never returned on GET — only a `configured` flag. + +interface SlackChannelResponse { + configured: boolean; + updated_at: string | null; + updated_by: { + id: string; + name: string | null; + email: string | null; + } | null; +} + +async function readCurrent(): Promise { + const config = await getSlackChannelConfigForAdmin(); + + return { + configured: config.configured, + updated_at: config.updatedAt ? config.updatedAt.toISOString() : null, + updated_by: config.updatedById + ? { + id: config.updatedById, + name: config.updatedByName, + email: config.updatedByEmail, + } + : null, + }; +} + +export async function GET() { + const authResult = await requireAdmin(); + if (authResult instanceof Response) { + return authResult; + } + + return Response.json(await readCurrent()); +} + +export async function PUT(request: NextRequest) { + const authResult = await requireAdmin(); + if (authResult instanceof Response) { + return authResult; + } + + let rawBody: unknown; + try { + rawBody = await request.json(); + } catch { + return apiError(400, VALIDATION_ERROR, "Invalid JSON body"); + } + + const parsed = slackChannelWebhookPutBodySchema.safeParse(rawBody); + if (!parsed.success) { + return apiError(400, VALIDATION_ERROR, "Invalid request body", { + issues: parsed.error.issues.map((issue) => ({ + path: issue.path.join("."), + code: issue.code, + message: issue.message, + })), + }); + } + + await upsertSlackChannelWebhookUrl( + parsed.data.webhook_url, + authResult.userId + ); + + return Response.json(await readCurrent()); +} diff --git a/web/app/globals.css b/web/app/globals.css index 1383e5e9..d3044db6 100644 --- a/web/app/globals.css +++ b/web/app/globals.css @@ -149,3 +149,16 @@ html.dark .shiki span { @apply font-sans; } } + +/* Preview banner offsets; height matches `PREVIEW_BANNER_HEIGHT` / banner `h-8`. */ +html[data-preview-deployment] { + --banner-height: 2rem; +} + +html[data-preview-deployment] body { + padding-top: var(--banner-height); +} + +html[data-preview-deployment] [data-slot="sidebar-container"] { + top: var(--banner-height); +} diff --git a/web/app/layout.tsx b/web/app/layout.tsx index c5bcde08..36a4dbab 100644 --- a/web/app/layout.tsx +++ b/web/app/layout.tsx @@ -8,6 +8,7 @@ import { NuqsAdapter } from "nuqs/adapters/next/app"; import { AppSidebar } from "@/components/app-sidebar"; import { NotificationBell } from "@/components/notifications/notification-bell"; import { NotificationsProvider } from "@/components/notifications/notifications-provider"; +import { PreviewDeploymentBanner } from "@/components/preview-deployment-banner"; import { ArchiveDownloadProvider } from "@/components/runs/archive-download-provider"; import { ThemeProvider } from "@/components/theme-provider"; import { @@ -110,6 +111,11 @@ export default async function RootLayout({ const sidebarCookie = (await cookies()).get(SIDEBAR_COOKIE_NAME)?.value; const sidebarDefaultOpen = sidebarCookie !== "false"; + // `--banner-height` is the single knob that offsets the body, the + // viewport-fixed sidebar, and the full-height auth screen for the preview + // banner. Left unset off preview, so each `var(..., 0px)` consumer is a no-op. + const isPreview = process.env.VERCEL_ENV === "preview"; + return ( @@ -126,6 +133,7 @@ export default async function RootLayout({ + {session ? ( @@ -68,6 +73,20 @@ export default async function NotificationsSettingsPage() { slackCommentsParticipatedEnabled: prefs.slackCommentsParticipatedEnabled, }} + slackChannelConfig={ + slackChannelConfig + ? { + configured: slackChannelConfig.configured, + lastUpdated: slackChannelConfig.updatedAt + ? { + at: slackChannelConfig.updatedAt.toISOString(), + byName: slackChannelConfig.updatedByName, + byEmail: slackChannelConfig.updatedByEmail, + } + : null, + } + : null + } slackConnection={ slackConn ? { diff --git a/web/components/auth/auth-screen.tsx b/web/components/auth/auth-screen.tsx index dfb1ccb1..b30b4a19 100644 --- a/web/components/auth/auth-screen.tsx +++ b/web/components/auth/auth-screen.tsx @@ -66,7 +66,7 @@ export function AuthScreen({ children, }: AuthScreenProps) { return ( -
+
diff --git a/web/components/notifications/notifications-settings-form.tsx b/web/components/notifications/notifications-settings-form.tsx index 2ccceb24..54bb8d7d 100644 --- a/web/components/notifications/notifications-settings-form.tsx +++ b/web/components/notifications/notifications-settings-form.tsx @@ -6,6 +6,7 @@ import { useRouter, useSearchParams } from "next/navigation"; import { useEffect } from "react"; import { toast } from "sonner"; import { z } from "zod"; +import { SlackChannelCard } from "@/components/notifications/slack-channel-card"; import { SlackConnectionCard, type SlackPreferences, @@ -50,6 +51,15 @@ interface SlackConnectionState { slackTeamName: string | null; } +interface SlackChannelConfigState { + configured: boolean; + lastUpdated: { + at: string; + byEmail: string | null; + byName: string | null; + } | null; +} + // The form's `perInstrument` field is a Record keyed by // instrument id. Storing it as a plain map (rather than parallel arrays) // keeps the per-row Field paths predictable: `perInstrument.${id}`. @@ -62,6 +72,9 @@ interface Props { // The page supplies both channels' prefs; the in-app subset seeds this // form, the Slack subset is handed to `SlackConnectionCard.Connected`. initialPreferences: InAppPreferences & SlackPreferences; + // Null when the viewer is not a workspace admin — the channel section is + // hidden entirely for non-admins. + slackChannelConfig: SlackChannelConfigState | null; slackConnection: SlackConnectionState; } @@ -69,6 +82,7 @@ export function NotificationsSettingsForm({ initialPreferences, initialInstruments, slackConnection, + slackChannelConfig, }: Props) { const router = useRouter(); const searchParams = useSearchParams(); @@ -379,6 +393,18 @@ export function NotificationsSettingsForm({ ) : ( )} + + {slackChannelConfig ? ( + <> + + + + ) : null}
); } diff --git a/web/components/notifications/slack-channel-card.tsx b/web/components/notifications/slack-channel-card.tsx new file mode 100644 index 00000000..2dd490ec --- /dev/null +++ b/web/components/notifications/slack-channel-card.tsx @@ -0,0 +1,318 @@ +"use client"; + +// Compound component for the org-wide Slack channel webhook section of the +// notifications settings page. Mirrors `slack-connection-card.tsx`: a +// section header above the card and an independent form so dirty state +// stays isolated from in-app and Slack DM prefs. +// +// +// + +import { useForm } from "@tanstack/react-form"; +import { Loader2 } from "lucide-react"; +import { useRouter } from "next/navigation"; +import { useEffect, useState } from "react"; +import { toast } from "sonner"; +import { Badge } from "@/components/ui/badge"; +import { Button } from "@/components/ui/button"; +import { Card, CardContent } from "@/components/ui/card"; +import { + Field, + FieldDescription, + FieldError, + FieldGroup, + FieldLabel, +} from "@/components/ui/field"; +import { Input } from "@/components/ui/input"; +import { + Tooltip, + TooltipContent, + TooltipTrigger, +} from "@/components/ui/tooltip"; +import { + slackChannelWebhookFormSchema, + slackWebhookUrlSchema, +} from "@/lib/slack/webhook-url"; +import { formatRelativeTime } from "@/lib/utils"; + +function SectionHeader({ configured }: { configured: boolean }) { + return ( +
+
+

Slack channel

+ {configured ? ( + + Configured + + ) : null} +
+

+ Post a message to a shared Slack channel whenever a new instrument run + is reported. This is separate from personal Slack DMs above — channel + notifications go to everyone in the channel. +

+
+ ); +} + +interface LastUpdated { + at: string; + byEmail: string | null; + byName: string | null; +} + +// Decoy length only — must not reflect the stored webhook URL. +const MASKED_LENGTH_MIN = 32; +const MASKED_LENGTH_RANGE = 41; + +function Form({ + configured, + lastUpdated, +}: { + configured: boolean; + lastUpdated: LastUpdated | null; +}) { + const router = useRouter(); + const [removing, setRemoving] = useState(false); + const [isReplacing, setIsReplacing] = useState(false); + const [maskedLength, setMaskedLength] = useState(MASKED_LENGTH_MIN); + + useEffect(() => { + setMaskedLength( + MASKED_LENGTH_MIN + Math.floor(Math.random() * MASKED_LENGTH_RANGE) + ); + }, []); + + useEffect(() => { + if (!configured) { + setIsReplacing(false); + } + }, [configured]); + + const form = useForm({ + defaultValues: { webhookUrl: "" }, + validators: { + onChange: slackChannelWebhookFormSchema, + onBlur: slackChannelWebhookFormSchema, + onSubmit: slackChannelWebhookFormSchema, + }, + onSubmit: async ({ value }) => { + const parsed = slackWebhookUrlSchema.safeParse(value.webhookUrl); + if (!parsed.success) { + return; + } + + const res = await fetch("/api/v1/settings/slack-channel", { + method: "PUT", + headers: { "Content-Type": "application/json" }, + body: JSON.stringify({ webhook_url: parsed.data }), + }); + + if (!res.ok) { + const body = await res.json().catch(() => null); + toast.error( + body?.error?.message ?? "Couldn't save Slack channel webhook" + ); + return; + } + + toast.success("Slack channel webhook saved"); + form.reset({ webhookUrl: "" }); + setIsReplacing(false); + router.refresh(); + }, + }); + + async function handleRemove() { + setRemoving(true); + try { + const res = await fetch("/api/v1/settings/slack-channel", { + method: "PUT", + headers: { "Content-Type": "application/json" }, + body: JSON.stringify({ webhook_url: null }), + }); + if (!res.ok) { + toast.error("Couldn't remove Slack channel webhook"); + return; + } + toast.success("Slack channel webhook removed"); + router.refresh(); + } catch { + toast.error("Couldn't remove Slack channel webhook"); + } finally { + setRemoving(false); + } + } + + return ( + + +
{ + e.preventDefault(); + e.stopPropagation(); + form.handleSubmit(); + }} + > + + + {(field) => { + const showMasked = + configured && + !isReplacing && + field.state.value.trim().length === 0; + const showFieldError = + !showMasked && + field.state.value.trim().length > 0 && + !field.state.meta.isValid; + return ( + + + Incoming webhook URL + + { + setIsReplacing(true); + field.handleChange(e.target.value); + }} + onFocus={() => { + if (showMasked) { + setIsReplacing(true); + field.handleChange(""); + } + }} + placeholder="https://hooks.slack.com/services/…" + readOnly={showMasked} + spellCheck={false} + type="password" + value={ + showMasked + ? "x".repeat(maskedLength) + : field.state.value + } + /> + + {configured ? ( + "A webhook is configured. Paste a new URL to replace it, or remove the existing webhook below." + ) : ( + <> + + Create an incoming webhook + {" "} + in your Slack workspace and paste the URL here. + + )} + + {showFieldError ? ( + + ) : null} + + ); + }} + + +
+ +
+
+ {configured ? ( + + + + + + Disable channel notifications and clear the stored webhook + URL. + + + ) : null} + {lastUpdated ? ( +

+ Last updated{" "} + + {formatRelativeTime(lastUpdated.at)} + + {lastUpdated.byName || lastUpdated.byEmail ? ( + <> + {" by "} + + {lastUpdated.byName ?? lastUpdated.byEmail} + + + ) : null} + . +

+ ) : ( +

+ No webhook configured yet. Channel notifications are disabled + until you save a URL. +

+ )} +
+ { + const trimmed = state.values.webhookUrl.trim(); + return { + canSubmit: state.canSubmit, + isSubmitting: state.isSubmitting, + isDirty: state.isDirty, + isValidUrl: slackWebhookUrlSchema.safeParse(trimmed).success, + }; + }} + > + {({ canSubmit, isSubmitting, isDirty, isValidUrl }) => ( + + )} + +
+ + + ); +} + +export const SlackChannelCard = { + SectionHeader, + Form, +}; diff --git a/web/components/preview-deployment-banner.tsx b/web/components/preview-deployment-banner.tsx new file mode 100644 index 00000000..10de8e28 --- /dev/null +++ b/web/components/preview-deployment-banner.tsx @@ -0,0 +1,37 @@ +import { TriangleAlert } from "lucide-react"; + +/** Kept in sync with the banner's `h-8` class; layout/sidebar subtract this. */ +export const PREVIEW_BANNER_HEIGHT = "2rem"; + +// `fixed` rather than in-flow so it paints above the viewport-fixed sidebar +// (`z-10`); `RootLayout` reserves space via `--banner-height` so nothing hides +// under it. `VERCEL_*` are unset off Vercel, so this only renders on previews. +export function PreviewDeploymentBanner() { + if (process.env.VERCEL_ENV !== "preview") { + return null; + } + + const branch = process.env.VERCEL_GIT_COMMIT_REF; + const productionUrl = process.env.VERCEL_PROJECT_PRODUCTION_URL; + + return ( +
+ + + This is a preview deployment + {branch ? ` for the ${branch} branch` : ""}. + + {productionUrl ? ( + + Go to production + + ) : null} +
+ ); +} diff --git a/web/components/ui/sidebar.tsx b/web/components/ui/sidebar.tsx index 21d0ae1f..29cb9040 100644 --- a/web/components/ui/sidebar.tsx +++ b/web/components/ui/sidebar.tsx @@ -133,7 +133,7 @@ function SidebarProvider({