From fccb1652101af5e4f004a73eb4c9a53da568de9e Mon Sep 17 00:00:00 2001 From: Wasim Amiri <7220175+wasimxyz@users.noreply.github.com> Date: Wed, 1 Jul 2026 12:45:26 -0700 Subject: [PATCH 1/4] Add staging run cleanup script and preview deployment banner (#110) * Add script to clear run/file records while preserving instruments and watchers Adds web/scripts/clear-run-file-records.ts (npm run db:clear-runs) to wipe run and file records from a remote database ahead of production cutover, keeping instruments and watchers so lab PCs stay movable between environments. Dry-run by default, requires --confirm to delete, and does not touch S3. Co-authored-by: Cursor * Document all web/scripts entries in scripts README Adds sections for seed-database, process-seeded-fixtures, and clear-run-file-records, plus a helper-modules list, so every runnable script and internal helper is covered. Co-authored-by: Cursor * chore: Ignore .vscode/ * Add a full-width preview deployment banner so reviewers know they are not on production. Uses Vercel system env vars for the branch name and a link back to production, with layout offsets so the fixed sidebar and auth pages are not covered. Co-authored-by: Cursor --------- Co-authored-by: Cursor --- .gitignore | 2 +- web/app/globals.css | 13 ++ web/app/layout.tsx | 8 ++ web/components/auth/auth-screen.tsx | 2 +- web/components/preview-deployment-banner.tsx | 37 +++++ web/components/ui/sidebar.tsx | 4 +- web/package.json | 1 + web/scripts/README.md | 76 ++++++++++ web/scripts/clear-run-file-records.ts | 137 +++++++++++++++++++ 9 files changed, 276 insertions(+), 4 deletions(-) create mode 100644 web/components/preview-deployment-banner.tsx create mode 100644 web/scripts/clear-run-file-records.ts 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/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 ? ( +
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({
); } 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/drizzle/0028_add_slack_channel_config.sql b/web/drizzle/0028_add_slack_channel_config.sql new file mode 100644 index 00000000..f6ffa6d9 --- /dev/null +++ b/web/drizzle/0028_add_slack_channel_config.sql @@ -0,0 +1,9 @@ +CREATE TABLE "slack_channel_config" ( + "id" boolean PRIMARY KEY DEFAULT true NOT NULL, + "webhook_url" text, + "updated_at" timestamp with time zone DEFAULT now() NOT NULL, + "updated_by" text, + CONSTRAINT "slack_channel_config_singleton" CHECK ("slack_channel_config"."id" = true) +); +--> statement-breakpoint +ALTER TABLE "slack_channel_config" ADD CONSTRAINT "slack_channel_config_updated_by_user_id_fk" FOREIGN KEY ("updated_by") REFERENCES "public"."user"("id") ON DELETE set null ON UPDATE no action; \ No newline at end of file diff --git a/web/drizzle/meta/0028_snapshot.json b/web/drizzle/meta/0028_snapshot.json new file mode 100644 index 00000000..c762b9f3 --- /dev/null +++ b/web/drizzle/meta/0028_snapshot.json @@ -0,0 +1,2141 @@ +{ + "id": "617a8c5b-dff4-4a2c-8e1a-fea5ecbace3b", + "prevId": "2de0e466-e06e-4b1a-97af-20c0a15c2162", + "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.archive_jobs": { + "name": "archive_jobs", + "schema": "", + "columns": { + "id": { + "name": "id", + "type": "uuid", + "primaryKey": true, + "notNull": true, + "default": "gen_random_uuid()" + }, + "instrument_run_id": { + "name": "instrument_run_id", + "type": "uuid", + "primaryKey": false, + "notNull": true + }, + "fingerprint": { + "name": "fingerprint", + "type": "text", + "primaryKey": false, + "notNull": true + }, + "archive_bucket": { + "name": "archive_bucket", + "type": "text", + "primaryKey": false, + "notNull": false + }, + "archive_key": { + "name": "archive_key", + "type": "text", + "primaryKey": false, + "notNull": false + }, + "size_bytes": { + "name": "size_bytes", + "type": "bigint", + "primaryKey": false, + "notNull": false + }, + "status": { + "name": "status", + "type": "archive_job_status", + "typeSchema": "public", + "primaryKey": false, + "notNull": true, + "default": "'pending'" + }, + "error_message": { + "name": "error_message", + "type": "text", + "primaryKey": false, + "notNull": false + }, + "created_by": { + "name": "created_by", + "type": "text", + "primaryKey": false, + "notNull": false + }, + "created_at": { + "name": "created_at", + "type": "timestamp with time zone", + "primaryKey": false, + "notNull": true, + "default": "now()" + }, + "completed_at": { + "name": "completed_at", + "type": "timestamp with time zone", + "primaryKey": false, + "notNull": false + } + }, + "indexes": { + "uq_archive_jobs_inflight": { + "name": "uq_archive_jobs_inflight", + "columns": [ + { + "expression": "instrument_run_id", + "isExpression": false, + "asc": true, + "nulls": "last" + }, + { + "expression": "fingerprint", + "isExpression": false, + "asc": true, + "nulls": "last" + } + ], + "isUnique": true, + "where": "\"archive_jobs\".\"status\" in ('pending', 'building')", + "concurrently": false, + "method": "btree", + "with": {} + }, + "idx_archive_jobs_run_fingerprint_status": { + "name": "idx_archive_jobs_run_fingerprint_status", + "columns": [ + { + "expression": "instrument_run_id", + "isExpression": false, + "asc": true, + "nulls": "last" + }, + { + "expression": "fingerprint", + "isExpression": false, + "asc": true, + "nulls": "last" + }, + { + "expression": "status", + "isExpression": false, + "asc": true, + "nulls": "last" + } + ], + "isUnique": false, + "concurrently": false, + "method": "btree", + "with": {} + } + }, + "foreignKeys": { + "archive_jobs_instrument_run_id_instrument_runs_id_fk": { + "name": "archive_jobs_instrument_run_id_instrument_runs_id_fk", + "tableFrom": "archive_jobs", + "tableTo": "instrument_runs", + "columnsFrom": ["instrument_run_id"], + "columnsTo": ["id"], + "onDelete": "cascade", + "onUpdate": "no action" + }, + "archive_jobs_created_by_user_id_fk": { + "name": "archive_jobs_created_by_user_id_fk", + "tableFrom": "archive_jobs", + "tableTo": "user", + "columnsFrom": ["created_by"], + "columnsTo": ["id"], + "onDelete": "set null", + "onUpdate": "no action" + } + }, + "compositePrimaryKeys": {}, + "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_active_instrument_run_id_filename": { + "name": "uq_files_active_instrument_run_id_filename", + "columns": [ + { + "expression": "instrument_run_id", + "isExpression": false, + "asc": true, + "nulls": "last" + }, + { + "expression": "filename", + "isExpression": false, + "asc": true, + "nulls": "last" + } + ], + "isUnique": true, + "where": "\"files\".\"deleted_at\" is 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_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_notification_subscriptions": { + "name": "instrument_notification_subscriptions", + "schema": "", + "columns": { + "user_id": { + "name": "user_id", + "type": "text", + "primaryKey": false, + "notNull": true + }, + "instrument_id": { + "name": "instrument_id", + "type": "text", + "primaryKey": false, + "notNull": true + }, + "enabled": { + "name": "enabled", + "type": "boolean", + "primaryKey": false, + "notNull": true, + "default": true + }, + "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": { + "idx_instrument_notification_subscriptions_user_id": { + "name": "idx_instrument_notification_subscriptions_user_id", + "columns": [ + { + "expression": "user_id", + "isExpression": false, + "asc": true, + "nulls": "last" + } + ], + "isUnique": false, + "concurrently": false, + "method": "btree", + "with": {} + } + }, + "foreignKeys": { + "instrument_notification_subscriptions_user_id_user_id_fk": { + "name": "instrument_notification_subscriptions_user_id_user_id_fk", + "tableFrom": "instrument_notification_subscriptions", + "tableTo": "user", + "columnsFrom": ["user_id"], + "columnsTo": ["id"], + "onDelete": "cascade", + "onUpdate": "no action" + }, + "instrument_notification_subscriptions_instrument_id_instruments_id_fk": { + "name": "instrument_notification_subscriptions_instrument_id_instruments_id_fk", + "tableFrom": "instrument_notification_subscriptions", + "tableTo": "instruments", + "columnsFrom": ["instrument_id"], + "columnsTo": ["id"], + "onDelete": "cascade", + "onUpdate": "no action" + } + }, + "compositePrimaryKeys": { + "instrument_notification_subscriptions_user_id_instrument_id_pk": { + "name": "instrument_notification_subscriptions_user_id_instrument_id_pk", + "columns": ["user_id", "instrument_id"] + } + }, + "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()" + }, + "acquired_at": { + "name": "acquired_at", + "type": "timestamp with time zone", + "primaryKey": false, + "notNull": false + }, + "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 + }, + "deleted_by": { + "name": "deleted_by", + "type": "text", + "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_active_acquired_at": { + "name": "idx_instrument_runs_active_acquired_at", + "columns": [ + { + "expression": "instrument_id", + "isExpression": false, + "asc": true, + "nulls": "last" + }, + { + "expression": "coalesce(\"acquired_at\", \"created_at\") desc", + "asc": true, + "isExpression": true, + "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" + }, + "instrument_runs_deleted_by_user_id_fk": { + "name": "instrument_runs_deleted_by_user_id_fk", + "tableFrom": "instrument_runs", + "tableTo": "user", + "columnsFrom": ["deleted_by"], + "columnsTo": ["id"], + "onDelete": "set null", + "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.notification_preferences": { + "name": "notification_preferences", + "schema": "", + "columns": { + "user_id": { + "name": "user_id", + "type": "text", + "primaryKey": true, + "notNull": true + }, + "runs_all_muted": { + "name": "runs_all_muted", + "type": "boolean", + "primaryKey": false, + "notNull": true, + "default": false + }, + "comments_attributed_enabled": { + "name": "comments_attributed_enabled", + "type": "boolean", + "primaryKey": false, + "notNull": true, + "default": true + }, + "comments_participated_enabled": { + "name": "comments_participated_enabled", + "type": "boolean", + "primaryKey": false, + "notNull": true, + "default": true + }, + "slack_runs_enabled": { + "name": "slack_runs_enabled", + "type": "boolean", + "primaryKey": false, + "notNull": true, + "default": false + }, + "slack_comments_attributed_enabled": { + "name": "slack_comments_attributed_enabled", + "type": "boolean", + "primaryKey": false, + "notNull": true, + "default": false + }, + "slack_comments_participated_enabled": { + "name": "slack_comments_participated_enabled", + "type": "boolean", + "primaryKey": false, + "notNull": true, + "default": false + }, + "updated_at": { + "name": "updated_at", + "type": "timestamp with time zone", + "primaryKey": false, + "notNull": true, + "default": "now()" + } + }, + "indexes": {}, + "foreignKeys": { + "notification_preferences_user_id_user_id_fk": { + "name": "notification_preferences_user_id_user_id_fk", + "tableFrom": "notification_preferences", + "tableTo": "user", + "columnsFrom": ["user_id"], + "columnsTo": ["id"], + "onDelete": "cascade", + "onUpdate": "no action" + } + }, + "compositePrimaryKeys": {}, + "uniqueConstraints": {}, + "policies": {}, + "checkConstraints": {}, + "isRLSEnabled": false + }, + "public.notifications": { + "name": "notifications", + "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 + }, + "type": { + "name": "type", + "type": "notification_type", + "typeSchema": "public", + "primaryKey": false, + "notNull": true + }, + "run_id": { + "name": "run_id", + "type": "uuid", + "primaryKey": false, + "notNull": true + }, + "comment_id": { + "name": "comment_id", + "type": "uuid", + "primaryKey": false, + "notNull": false + }, + "actor_user_id": { + "name": "actor_user_id", + "type": "text", + "primaryKey": false, + "notNull": false + }, + "read_at": { + "name": "read_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_notifications_user_id_created_at": { + "name": "idx_notifications_user_id_created_at", + "columns": [ + { + "expression": "user_id", + "isExpression": false, + "asc": true, + "nulls": "last" + }, + { + "expression": "created_at", + "isExpression": false, + "asc": false, + "nulls": "last" + } + ], + "isUnique": false, + "concurrently": false, + "method": "btree", + "with": {} + }, + "idx_notifications_user_id_unread": { + "name": "idx_notifications_user_id_unread", + "columns": [ + { + "expression": "user_id", + "isExpression": false, + "asc": true, + "nulls": "last" + }, + { + "expression": "created_at", + "isExpression": false, + "asc": false, + "nulls": "last" + } + ], + "isUnique": false, + "where": "\"notifications\".\"read_at\" is null", + "concurrently": false, + "method": "btree", + "with": {} + } + }, + "foreignKeys": { + "notifications_user_id_user_id_fk": { + "name": "notifications_user_id_user_id_fk", + "tableFrom": "notifications", + "tableTo": "user", + "columnsFrom": ["user_id"], + "columnsTo": ["id"], + "onDelete": "cascade", + "onUpdate": "no action" + }, + "notifications_run_id_instrument_runs_id_fk": { + "name": "notifications_run_id_instrument_runs_id_fk", + "tableFrom": "notifications", + "tableTo": "instrument_runs", + "columnsFrom": ["run_id"], + "columnsTo": ["id"], + "onDelete": "cascade", + "onUpdate": "no action" + }, + "notifications_comment_id_run_comments_id_fk": { + "name": "notifications_comment_id_run_comments_id_fk", + "tableFrom": "notifications", + "tableTo": "run_comments", + "columnsFrom": ["comment_id"], + "columnsTo": ["id"], + "onDelete": "cascade", + "onUpdate": "no action" + }, + "notifications_actor_user_id_user_id_fk": { + "name": "notifications_actor_user_id_user_id_fk", + "tableFrom": "notifications", + "tableTo": "user", + "columnsFrom": ["actor_user_id"], + "columnsTo": ["id"], + "onDelete": "set null", + "onUpdate": "no action" + } + }, + "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 + }, + "scopes": { + "name": "scopes", + "type": "text[]", + "primaryKey": false, + "notNull": true, + "default": "ARRAY['*']::text[]" + }, + "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.run_comments": { + "name": "run_comments", + "schema": "", + "columns": { + "id": { + "name": "id", + "type": "uuid", + "primaryKey": true, + "notNull": true, + "default": "gen_random_uuid()" + }, + "run_id": { + "name": "run_id", + "type": "uuid", + "primaryKey": false, + "notNull": true + }, + "user_id": { + "name": "user_id", + "type": "text", + "primaryKey": false, + "notNull": true + }, + "body": { + "name": "body", + "type": "text", + "primaryKey": false, + "notNull": true + }, + "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()" + }, + "edited_at": { + "name": "edited_at", + "type": "timestamp with time zone", + "primaryKey": false, + "notNull": false + }, + "deleted_at": { + "name": "deleted_at", + "type": "timestamp with time zone", + "primaryKey": false, + "notNull": false + } + }, + "indexes": { + "idx_run_comments_run_id_created_at": { + "name": "idx_run_comments_run_id_created_at", + "columns": [ + { + "expression": "run_id", + "isExpression": false, + "asc": true, + "nulls": "last" + }, + { + "expression": "created_at", + "isExpression": false, + "asc": false, + "nulls": "last" + } + ], + "isUnique": false, + "concurrently": false, + "method": "btree", + "with": {} + }, + "idx_run_comments_user_id": { + "name": "idx_run_comments_user_id", + "columns": [ + { + "expression": "user_id", + "isExpression": false, + "asc": true, + "nulls": "last" + } + ], + "isUnique": false, + "concurrently": false, + "method": "btree", + "with": {} + } + }, + "foreignKeys": { + "run_comments_run_id_instrument_runs_id_fk": { + "name": "run_comments_run_id_instrument_runs_id_fk", + "tableFrom": "run_comments", + "tableTo": "instrument_runs", + "columnsFrom": ["run_id"], + "columnsTo": ["id"], + "onDelete": "cascade", + "onUpdate": "no action" + }, + "run_comments_user_id_user_id_fk": { + "name": "run_comments_user_id_user_id_fk", + "tableFrom": "run_comments", + "tableTo": "user", + "columnsFrom": ["user_id"], + "columnsTo": ["id"], + "onDelete": "cascade", + "onUpdate": "no action" + } + }, + "compositePrimaryKeys": {}, + "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.slack_channel_config": { + "name": "slack_channel_config", + "schema": "", + "columns": { + "id": { + "name": "id", + "type": "boolean", + "primaryKey": true, + "notNull": true, + "default": true + }, + "webhook_url": { + "name": "webhook_url", + "type": "text", + "primaryKey": false, + "notNull": false + }, + "updated_at": { + "name": "updated_at", + "type": "timestamp with time zone", + "primaryKey": false, + "notNull": true, + "default": "now()" + }, + "updated_by": { + "name": "updated_by", + "type": "text", + "primaryKey": false, + "notNull": false + } + }, + "indexes": {}, + "foreignKeys": { + "slack_channel_config_updated_by_user_id_fk": { + "name": "slack_channel_config_updated_by_user_id_fk", + "tableFrom": "slack_channel_config", + "tableTo": "user", + "columnsFrom": ["updated_by"], + "columnsTo": ["id"], + "onDelete": "set null", + "onUpdate": "no action" + } + }, + "compositePrimaryKeys": {}, + "uniqueConstraints": {}, + "policies": {}, + "checkConstraints": { + "slack_channel_config_singleton": { + "name": "slack_channel_config_singleton", + "value": "\"slack_channel_config\".\"id\" = true" + } + }, + "isRLSEnabled": false + }, + "public.slack_connections": { + "name": "slack_connections", + "schema": "", + "columns": { + "user_id": { + "name": "user_id", + "type": "text", + "primaryKey": true, + "notNull": true + }, + "slack_user_id": { + "name": "slack_user_id", + "type": "text", + "primaryKey": false, + "notNull": true + }, + "slack_team_id": { + "name": "slack_team_id", + "type": "text", + "primaryKey": false, + "notNull": true + }, + "slack_team_name": { + "name": "slack_team_name", + "type": "text", + "primaryKey": false, + "notNull": false + }, + "connected_at": { + "name": "connected_at", + "type": "timestamp with time zone", + "primaryKey": false, + "notNull": true, + "default": "now()" + }, + "revoked_at": { + "name": "revoked_at", + "type": "timestamp with time zone", + "primaryKey": false, + "notNull": false + } + }, + "indexes": {}, + "foreignKeys": { + "slack_connections_user_id_user_id_fk": { + "name": "slack_connections_user_id_user_id_fk", + "tableFrom": "slack_connections", + "tableTo": "user", + "columnsFrom": ["user_id"], + "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 + }, + "is_admin": { + "name": "is_admin", + "type": "boolean", + "primaryKey": false, + "notNull": true, + "default": 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.watcher_release_config": { + "name": "watcher_release_config", + "schema": "", + "columns": { + "id": { + "name": "id", + "type": "boolean", + "primaryKey": true, + "notNull": true, + "default": true + }, + "latest_version": { + "name": "latest_version", + "type": "text", + "primaryKey": false, + "notNull": false + }, + "min_supported_version": { + "name": "min_supported_version", + "type": "text", + "primaryKey": false, + "notNull": false + }, + "channel": { + "name": "channel", + "type": "text", + "primaryKey": false, + "notNull": true, + "default": "'stable'" + }, + "mandatory": { + "name": "mandatory", + "type": "boolean", + "primaryKey": false, + "notNull": true, + "default": false + }, + "updated_at": { + "name": "updated_at", + "type": "timestamp with time zone", + "primaryKey": false, + "notNull": true, + "default": "now()" + }, + "updated_by": { + "name": "updated_by", + "type": "text", + "primaryKey": false, + "notNull": false + } + }, + "indexes": {}, + "foreignKeys": { + "watcher_release_config_updated_by_user_id_fk": { + "name": "watcher_release_config_updated_by_user_id_fk", + "tableFrom": "watcher_release_config", + "tableTo": "user", + "columnsFrom": ["updated_by"], + "columnsTo": ["id"], + "onDelete": "set null", + "onUpdate": "no action" + } + }, + "compositePrimaryKeys": {}, + "uniqueConstraints": {}, + "policies": {}, + "checkConstraints": { + "watcher_release_config_singleton": { + "name": "watcher_release_config_singleton", + "value": "\"watcher_release_config\".\"id\" = true" + } + }, + "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 + }, + "watcher_version": { + "name": "watcher_version", + "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": { + "uq_watchers_active_instrument_id": { + "name": "uq_watchers_active_instrument_id", + "columns": [ + { + "expression": "instrument_id", + "isExpression": false, + "asc": true, + "nulls": "last" + } + ], + "isUnique": true, + "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.archive_job_status": { + "name": "archive_job_status", + "schema": "public", + "values": ["pending", "building", "ready", "failed"] + }, + "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", + "epson_v700_scanner", + "instant_raman" + ] + }, + "public.notification_type": { + "name": "notification_type", + "schema": "public", + "values": ["run_created", "comment_attributed", "comment_participated"] + }, + "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", + "update_started", + "update_succeeded", + "update_failed" + ] + }, + "public.watcher_status": { + "name": "watcher_status", + "schema": "public", + "values": ["registered", "watching", "stopped"] + } + }, + "schemas": {}, + "sequences": {}, + "roles": {}, + "policies": {}, + "views": {}, + "_meta": { + "columns": {}, + "schemas": {}, + "tables": {} + } +} diff --git a/web/drizzle/meta/_journal.json b/web/drizzle/meta/_journal.json index c157f0ce..d24a45ff 100644 --- a/web/drizzle/meta/_journal.json +++ b/web/drizzle/meta/_journal.json @@ -197,6 +197,13 @@ "when": 1782334433553, "tag": "0027_futuristic_pyro", "breakpoints": true + }, + { + "idx": 28, + "version": "7", + "when": 1782935691822, + "tag": "0028_add_slack_channel_config", + "breakpoints": true } ] } diff --git a/web/lib/db/schema.ts b/web/lib/db/schema.ts index 252a4085..48760e3f 100644 --- a/web/lib/db/schema.ts +++ b/web/lib/db/schema.ts @@ -204,6 +204,34 @@ export const watcherReleaseConfig = pgTable( ] ); +// Singleton row holding the org-wide Slack incoming webhook URL for +// channel notifications on new runs. Edited via the admin-only "Slack +// channel" section on `/settings/notifications`. Previously sourced from +// the `SLACK_WEBHOOK_URL` env var. +// +// When the table is empty (or `webhook_url` is NULL) channel notifications +// are disabled — `sendSlackMessage` becomes a no-op. +export const slackChannelConfig = pgTable( + "slack_channel_config", + { + id: boolean("id").primaryKey().default(true), + webhookUrl: text("webhook_url"), + updatedAt: timestamp("updated_at", { + withTimezone: true, + mode: "date", + }) + .notNull() + .defaultNow() + .$onUpdate(() => new Date()), + updatedBy: text("updated_by").references(() => users.id, { + onDelete: "set null", + }), + }, + (config) => [ + check("slack_channel_config_singleton", sql`${config.id} = true`), + ] +); + export const personalAccessTokens = pgTable( "personal_access_tokens", { diff --git a/web/lib/db/seed.ts b/web/lib/db/seed.ts index 540daab2..efef3c5b 100644 --- a/web/lib/db/seed.ts +++ b/web/lib/db/seed.ts @@ -130,6 +130,23 @@ export async function seedWatcherReleaseConfig(db: Db): Promise { ); } +// Singleton row for org-wide Slack channel notifications. Integration tests +// pass the in-process capture-server URL; the dev seed leaves it unset so +// channel notifications stay disabled until an admin configures the webhook. +export async function seedSlackChannelConfig( + db: Db, + webhookUrl: string | null = null +): Promise { + await db.execute( + sql`INSERT INTO slack_channel_config (id, webhook_url) + VALUES (true, ${webhookUrl}) + ON CONFLICT (id) DO UPDATE SET + webhook_url = EXCLUDED.webhook_url, + updated_at = now(), + updated_by = NULL` + ); +} + // --------------------------------------------------------------------------- // Instruments — one row per value in `instrumentTypeEnum.enumValues` so the // dashboard exercises every instrument-type-specific UI variant. One diff --git a/web/lib/slack.ts b/web/lib/slack.ts index 2ebb6b98..74e3fd88 100644 --- a/web/lib/slack.ts +++ b/web/lib/slack.ts @@ -1,16 +1,20 @@ // Posts messages to Slack via the configured incoming webhook URL. // -// Mirrors the contract of the (now-removed) `data_hub_shared.slack` Python -// helper: if `SLACK_WEBHOOK_URL` is unset, calls become a no-op with a -// warning so local development and tests don't need a webhook configured. -// Network/HTTP failures are logged but never thrown — Slack is a notification -// side-channel and a Slack outage must not break the API request that -// triggered it. +// The webhook URL is stored in the `slack_channel_config` singleton row, +// edited via Settings > Notifications by workspace admins. If unset, calls +// become a no-op with a warning so local development and tests don't need a +// webhook configured. Network/HTTP failures are logged but never thrown — +// Slack is a notification side-channel and a Slack outage must not break the +// API request that triggered it. + +import { getSlackChannelWebhookUrl } from "@/lib/slack/channel-config"; export async function sendSlackMessage(text: string): Promise { - const webhookUrl = process.env.SLACK_WEBHOOK_URL; + const webhookUrl = await getSlackChannelWebhookUrl(); if (!webhookUrl) { - console.warn("SLACK_WEBHOOK_URL is not set, skipping Slack message."); + console.warn( + "Slack channel webhook is not configured, skipping Slack message." + ); return; } diff --git a/web/lib/slack/channel-config.ts b/web/lib/slack/channel-config.ts new file mode 100644 index 00000000..7c682cc7 --- /dev/null +++ b/web/lib/slack/channel-config.ts @@ -0,0 +1,75 @@ +import { eq } from "drizzle-orm"; +import { db } from "@/lib/db"; +import { slackChannelConfig, users } from "@/lib/db/schema"; + +export interface SlackChannelConfigForAdmin { + configured: boolean; + updatedAt: Date | null; + updatedByEmail: string | null; + updatedById: string | null; + updatedByName: string | null; +} + +export async function getSlackChannelWebhookUrl(): Promise { + const [row] = await db + .select({ webhookUrl: slackChannelConfig.webhookUrl }) + .from(slackChannelConfig); + + return row?.webhookUrl ?? null; +} + +export async function getSlackChannelConfigForAdmin(): Promise { + const [row] = await db + .select({ + webhookUrl: slackChannelConfig.webhookUrl, + updatedAt: slackChannelConfig.updatedAt, + updatedById: users.id, + updatedByName: users.name, + updatedByEmail: users.email, + }) + .from(slackChannelConfig) + .leftJoin(users, eq(users.id, slackChannelConfig.updatedBy)); + + if (!row) { + return { + configured: false, + updatedAt: null, + updatedById: null, + updatedByName: null, + updatedByEmail: null, + }; + } + + return { + configured: row.webhookUrl != null && row.webhookUrl.length > 0, + updatedAt: row.updatedAt, + updatedById: row.updatedById, + updatedByName: row.updatedByName, + updatedByEmail: row.updatedByEmail, + }; +} + +export async function upsertSlackChannelWebhookUrl( + webhookUrl: string | null, + updatedBy: string +): Promise { + const now = new Date(); + await db + .insert(slackChannelConfig) + .values({ + id: true, + webhookUrl, + updatedAt: now, + updatedBy, + }) + .onConflictDoUpdate({ + target: slackChannelConfig.id, + set: { + webhookUrl, + updatedAt: now, + updatedBy, + }, + }); + + return getSlackChannelConfigForAdmin(); +} diff --git a/web/lib/slack/webhook-url.ts b/web/lib/slack/webhook-url.ts new file mode 100644 index 00000000..95428fe0 --- /dev/null +++ b/web/lib/slack/webhook-url.ts @@ -0,0 +1,62 @@ +import { z } from "zod"; + +// Shared Slack incoming webhook URL validation for the admin settings +// route and client form. Incoming webhooks always live under this host. +export const SLACK_WEBHOOK_URL_PREFIX = "https://hooks.slack.com/services/"; + +export const SLACK_WEBHOOK_URL_REGEX = + /^https:\/\/hooks\.slack\.com\/services\/\S+$/; + +export const SLACK_WEBHOOK_URL_MESSAGE = + "Use a Slack incoming webhook URL (https://hooks.slack.com/services/…)."; + +/** Non-empty trimmed value that matches Slack's incoming webhook shape. */ +export const slackWebhookUrlSchema = z + .string() + .trim() + .min(1, SLACK_WEBHOOK_URL_MESSAGE) + .refine((url) => SLACK_WEBHOOK_URL_REGEX.test(url), { + message: SLACK_WEBHOOK_URL_MESSAGE, + }); + +export type SlackWebhookUrl = z.infer; + +export function normalizeSlackWebhookUrlInput( + v: string | null | undefined +): string | null { + if (v == null) { + return null; + } + const trimmed = v.trim(); + return trimmed.length === 0 ? null : trimmed; +} + +/** Form field: empty while masked/idle; non-empty must match {@link slackWebhookUrlSchema}. */ +export const slackWebhookUrlInputSchema = z + .string() + .refine( + (value) => value.trim().length === 0 || isValidSlackWebhookUrl(value), + { message: SLACK_WEBHOOK_URL_MESSAGE } + ); + +export const slackChannelWebhookFormSchema = z.object({ + webhookUrl: slackWebhookUrlInputSchema, +}); + +export type SlackChannelWebhookFormValues = z.infer< + typeof slackChannelWebhookFormSchema +>; + +export const slackChannelWebhookPutBodySchema = z.strictObject({ + webhook_url: z + .string() + .nullish() + .transform(normalizeSlackWebhookUrlInput) + .refine((value) => value === null || isValidSlackWebhookUrl(value), { + message: SLACK_WEBHOOK_URL_MESSAGE, + }), +}); + +export function isValidSlackWebhookUrl(url: string): boolean { + return slackWebhookUrlSchema.safeParse(url).success; +} diff --git a/web/scripts/seed-database.ts b/web/scripts/seed-database.ts index 8353b769..c2da0f6c 100644 --- a/web/scripts/seed-database.ts +++ b/web/scripts/seed-database.ts @@ -25,6 +25,7 @@ import { seedRunAttributions, seedRunComments, seedRuns, + seedSlackChannelConfig, seedTeammates, seedWatcherReleaseConfig, seedWatchers, @@ -46,6 +47,9 @@ await clearAll(db); console.log("Seeding watcher_release_config…"); await seedWatcherReleaseConfig(db); +console.log("Seeding slack_channel_config…"); +await seedSlackChannelConfig(db); + console.log("Seeding dev user…"); const { userId, email, token } = await seedDevUser(db, { email: "dev@local", diff --git a/web/tests/integration/global-setup.ts b/web/tests/integration/global-setup.ts index 3a3993c0..8187e510 100644 --- a/web/tests/integration/global-setup.ts +++ b/web/tests/integration/global-setup.ts @@ -84,7 +84,7 @@ export async function setup() { // the real Slack API. // // The capture server handles: - // POST /webhook — incoming webhook (SLACK_WEBHOOK_URL) + // POST /webhook — incoming webhook (slack_channel_config) // GET /captured — read webhook capture buffer // POST /clear — reset webhook buffer // POST /api/chat.postMessage — Web API DM capture (__TEST_SLACK_API_URL) @@ -203,12 +203,10 @@ export async function setup() { stdio: "pipe", }); - // 2a. Seed the singleton `watcher_release_config` row with stable - // defaults so the `update-check` endpoint returns deterministic - // values during integration tests. Individual tests assert against - // these exact strings (see `watchers.test.ts`). Previously this was - // done via WATCHER_* env vars; the source of truth is now the DB, - // edited via the admin-only /settings/watchers page. + // 2a. Seed singleton config rows with stable defaults for integration + // tests. Individual tests assert against these values. Previously + // watcher release used WATCHER_* env vars and Slack channel used + // SLACK_WEBHOOK_URL; the source of truth is now the DB. const seedPool = new Pool({ connectionString: databaseUrl }); try { await seedPool.query(` @@ -223,6 +221,16 @@ export async function setup() { mandatory = excluded.mandatory, updated_at = now() `); + await seedPool.query( + ` + INSERT INTO slack_channel_config (id, webhook_url) + VALUES (true, $1) + ON CONFLICT (id) DO UPDATE SET + webhook_url = excluded.webhook_url, + updated_at = now() + `, + [`${slackCaptureBaseUrl}/webhook`] + ); } finally { await seedPool.end(); } @@ -254,9 +262,8 @@ export async function setup() { process.env.S3_RAW_DATA_BUCKET ?? "test-raw-data-bucket", // Watcher release-info defaults are seeded into the // `watcher_release_config` table above; the env-var fallback is gone. - // Point Slack webhook calls at the in-process capture server defined - // above so tests can assert on the messages without hitting Slack. - SLACK_WEBHOOK_URL: `${slackCaptureBaseUrl}/webhook`, + // Slack channel webhook URL is seeded into `slack_channel_config` + // above so tests can assert on captured payloads without hitting Slack. // Stub bot token so sendSlackDm's guard passes; the WebClient is // redirected to the capture server via __TEST_SLACK_API_URL. SLACK_BOT_TOKEN: "xoxb-test-bot-token", diff --git a/web/tests/integration/helpers.ts b/web/tests/integration/helpers.ts index 456662c1..0b5f9cd8 100644 --- a/web/tests/integration/helpers.ts +++ b/web/tests/integration/helpers.ts @@ -6,6 +6,7 @@ import { clearAll, type SeedUserOptions, seedDevUser, + seedSlackChannelConfig, seedWatcherReleaseConfig, } from "@/lib/db/seed"; @@ -39,21 +40,24 @@ export async function closeTestDb() { } // TRUNCATE every `pgTable` declared in `lib/db/schema.ts`, then re-seed -// the `watcher_release_config` singleton with the deterministic baseline -// `9.9.9 / 0.1.0 / stable / false`. Tests previously hard-coded both the -// table list and the singleton SQL inline; both now live in -// `@/lib/db/seed` so adding a new table doesn't require touching this -// file. +// singleton config rows with deterministic baselines. Tests previously +// hard-coded both the table list and the singleton SQL inline; both now +// live in `@/lib/db/seed` so adding a new table doesn't require touching +// this file. // // `clearAll` uses TRUNCATE CASCADE, which ignores the `ON DELETE SET NULL` -// on `watcher_release_config.updated_by → user.id` and wipes the singleton -// regardless of whether it's in the TRUNCATE list. Re-seeding it after -// the clear keeps every test's update-check baseline identical to a -// fresh global setup. +// on singleton `updated_by → user.id` FKs and wipes the rows regardless +// of whether they're in the TRUNCATE list. Re-seeding after the clear +// keeps every test's baseline identical to a fresh global setup. export async function resetDb() { const db = getTestDb(); await clearAll(db); await seedWatcherReleaseConfig(db); + const captureBase = process.env.__TEST_SLACK_CAPTURE_URL; + await seedSlackChannelConfig( + db, + captureBase ? `${captureBase}/webhook` : null + ); } // --------------------------------------------------------------------------- @@ -128,8 +132,9 @@ export async function api( // --------------------------------------------------------------------------- // Slack webhook capture — the global setup spawns an in-process HTTP server -// that captures every payload posted to SLACK_WEBHOOK_URL. These helpers let -// individual tests inspect and reset that capture buffer. +// that captures every payload posted to the webhook URL stored in +// `slack_channel_config`. These helpers let individual tests inspect and +// reset that capture buffer. // --------------------------------------------------------------------------- function getSlackCaptureUrl(): string { @@ -148,6 +153,22 @@ export async function getCapturedSlackMessages(): Promise<{ text: string }[]> { return res.json(); } +/** Poll until at least `minCount` webhook payloads arrive or timeout. */ +export async function waitForCapturedSlackMessages( + minCount: number, + { timeoutMs = 3000, intervalMs = 50 } = {} +): Promise<{ text: string }[]> { + const deadline = Date.now() + timeoutMs; + while (Date.now() < deadline) { + const messages = await getCapturedSlackMessages(); + if (messages.length >= minCount) { + return messages; + } + await new Promise((r) => setTimeout(r, intervalMs)); + } + throw new Error(`Timed out waiting for ${minCount} Slack webhook message(s)`); +} + export async function clearCapturedSlackMessages(): Promise { const res = await fetch(`${getSlackCaptureUrl()}/clear`, { method: "POST" }); if (!res.ok) { diff --git a/web/tests/integration/instrument-runs.test.ts b/web/tests/integration/instrument-runs.test.ts index e86ee05d..e4c1dca0 100644 --- a/web/tests/integration/instrument-runs.test.ts +++ b/web/tests/integration/instrument-runs.test.ts @@ -4,10 +4,10 @@ import { api, clearCapturedSlackMessages, closeTestDb, - getCapturedSlackMessages, getTestDb, resetDb, seedTestUser, + waitForCapturedSlackMessages, } from "@/tests/integration/helpers"; describe("Instrument Runs API", () => { @@ -345,8 +345,7 @@ describe("Run creation Slack notification", () => { }); expect(duplicate.status).toBe(200); - const messages = await getCapturedSlackMessages(); - expect(messages.length).toBe(1); + const messages = await waitForCapturedSlackMessages(1); expect(messages[0].text).toContain(instrumentDisplayName); expect(messages[0].text).toContain(runId); expect(messages[0].text).toContain( diff --git a/web/tests/integration/slack-channel.test.ts b/web/tests/integration/slack-channel.test.ts new file mode 100644 index 00000000..1d48ad2d --- /dev/null +++ b/web/tests/integration/slack-channel.test.ts @@ -0,0 +1,136 @@ +import { eq } from "drizzle-orm"; +import { afterAll, beforeAll, describe, expect, it } from "vitest"; +import { instruments, slackChannelConfig } from "@/lib/db/schema"; +import { + api, + clearCapturedSlackMessages, + closeTestDb, + getCapturedSlackMessages, + getTestDb, + resetDb, + seedTestUser, + waitForCapturedSlackMessages, +} from "@/tests/integration/helpers"; + +// The `/api/v1/settings/slack-channel` surface is admin-only and +// session-only — PATs never pass the gate. The end-to-end +// "DB config → sendSlackMessage" wiring is verified by upserting the +// singleton row directly and asserting the capture server receives the +// payload on run creation. + +describe("Slack channel API admin gate", () => { + let token: string; + + beforeAll(async () => { + await resetDb(); + ({ token } = await seedTestUser({ isAdmin: true })); + }); + + afterAll(async () => { + await closeTestDb(); + }); + + it("GET /api/v1/settings/slack-channel rejects PAT auth (session required)", async () => { + const res = await api("/api/v1/settings/slack-channel", { token }); + expect(res.status).toBe(401); + }); + + it("GET /api/v1/settings/slack-channel rejects unauthenticated requests", async () => { + const res = await api("/api/v1/settings/slack-channel"); + expect(res.status).toBe(401); + }); + + it("PUT /api/v1/settings/slack-channel rejects PAT auth", async () => { + const res = await api("/api/v1/settings/slack-channel", { + method: "PUT", + token, + body: { webhook_url: "https://hooks.slack.com/services/T/B/x" }, + }); + expect(res.status).toBe(401); + }); + + it("PUT /api/v1/settings/slack-channel rejects unauthenticated requests", async () => { + const res = await api("/api/v1/settings/slack-channel", { + method: "PUT", + body: { webhook_url: "https://hooks.slack.com/services/T/B/x" }, + }); + expect(res.status).toBe(401); + }); +}); + +describe("Slack channel config flows through sendSlackMessage", () => { + let token: string; + const instrumentId = "slack-channel-config-instrument"; + const instrumentDisplayName = "Slack Channel Config Instrument"; + const captureWebhookUrl = `${process.env.__TEST_SLACK_CAPTURE_URL}/webhook`; + + beforeAll(async () => { + await resetDb(); + ({ token } = await seedTestUser()); + + const db = getTestDb(); + await db.insert(instruments).values({ + id: instrumentId, + displayName: instrumentDisplayName, + status: "active", + }); + + await db + .insert(slackChannelConfig) + .values({ + id: true, + webhookUrl: captureWebhookUrl, + }) + .onConflictDoUpdate({ + target: slackChannelConfig.id, + set: { webhookUrl: captureWebhookUrl }, + }); + }); + + afterAll(async () => { + await closeTestDb(); + }); + + it("run creation posts to the webhook URL from slack_channel_config", async () => { + await clearCapturedSlackMessages(); + const runId = "slack-channel-run-001"; + + const res = await api(`/api/v1/instruments/${instrumentId}/runs`, { + method: "POST", + token, + body: { run_id: runId, source: "lambda" }, + }); + expect(res.status).toBe(201); + + const messages = await waitForCapturedSlackMessages(1); + expect(messages[0].text).toContain(instrumentDisplayName); + expect(messages[0].text).toContain(runId); + }); + + it("sendSlackMessage is a no-op when webhook_url is null", async () => { + const db = getTestDb(); + await db + .update(slackChannelConfig) + .set({ webhookUrl: null }) + .where(eq(slackChannelConfig.id, true)); + + await clearCapturedSlackMessages(); + const runId = "slack-channel-run-disabled"; + + const res = await api(`/api/v1/instruments/${instrumentId}/runs`, { + method: "POST", + token, + body: { run_id: runId, source: "lambda" }, + }); + expect(res.status).toBe(201); + + const messages = await getCapturedSlackMessages(); + expect(messages.length).toBe(0); + + // Restore the global-setup default for other tests in the suite. + await db + .update(slackChannelConfig) + .set({ webhookUrl: captureWebhookUrl }) + .where(eq(slackChannelConfig.id, true)); + }); +}); From 6545b419c98bdb136fe15da88bb3b0f56beecdbb Mon Sep 17 00:00:00 2001 From: Wasim Amiri <7220175+wasimxyz@users.noreply.github.com> Date: Wed, 1 Jul 2026 17:18:14 -0700 Subject: [PATCH 4/4] Watcher: Move manual-mode uploads to a dedicated worker thread (#113) Manual-mode upload-queue polling ran on the heartbeat tick, so a slow or large upload delayed heartbeats and the dashboard flagged busy watchers as offline (prompting operator restarts). On stop, a long upload could also outlive the heartbeat join and write to the state DB after close, raising "Cannot operate on a closed database". Introduce `UploadQueueWorker`, a long-lived thread that owns the poll loop and shares a stop event with `Uploader` so shutdown can interrupt the retry backoff and abort between queued files. `stop_runtime` now stops the worker before closing the state DB, skipping the close if it can't stop in time, restoring the writer-threads-joined-before-close invariant. Bumps the watcher version to 0.4.0. Co-authored-by: Cursor --- developer-docs/reference/watcher.md | 9 +- uv.lock | 2 +- watcher/pyproject.toml | 2 +- watcher/src/data_hub_watcher/constants.py | 6 + watcher/src/data_hub_watcher/runtime.py | 51 +++++--- watcher/src/data_hub_watcher/uploader.py | 108 +++++++++++++++-- watcher/tests/test_runtime.py | 120 ++++++++++++++----- watcher/tests/test_uploader.py | 135 +++++++++++++++++++++- 8 files changed, 374 insertions(+), 59 deletions(-) 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()