From e4e797b97b58af3b6eb74dde86827f625e2d79a7 Mon Sep 17 00:00:00 2001 From: Wasim Sandhu Date: Thu, 7 May 2026 17:00:27 -0700 Subject: [PATCH] Slack: Move run-creation notifications from Lambda to web app The Lambda previously called slack.send_message once per file processed, which produced multiple "Finished preprocessing run X" messages per run even though the wording implied one. Notifications now fire from POST /api/v1/instruments/:instrumentId/runs when the upsert actually inserts a new row, so each run triggers exactly one Slack message regardless of how many files arrive (lambda or watcher). Adds web-app/lib/slack.ts (no-op when SLACK_WEBHOOK_URL is unset, swallows errors so a Slack outage cannot fail the API request) and an integration test using an in-process capture server. Removes data_hub_shared.slack, the SLACK_WEBHOOK_URL config field, the SlackWebhookUrl SAM parameter / Makefile / deploy-lambda.yml plumbing, the dead result_url return path from each per-instrument process_file, and the now-unused DATA_HUB_WEB_URL constant. Operators must set SLACK_WEBHOOK_URL in the Vercel environment (it is no longer needed in GitHub environment secrets or infra/.env.*). Co-authored-by: Cursor --- .github/workflows/deploy-lambda.yml | 1 - Makefile | 1 - README.md | 2 +- docs/architecture.md | 4 +- docs/ci-and-deployment.md | 6 +- docs/getting-started.md | 2 +- docs/guides/adding-an-instrument.md | 9 ++- docs/lambda.md | 20 +++--- docs/shared-library.md | 15 ----- infra/.env.example | 1 - infra/template.yaml | 6 -- lambda/.env.example | 6 +- .../agilent_4150_tapestation/process_file.py | 8 +-- .../data_hub_lambda/akta_fplc/process_file.py | 8 +-- .../azure_600_gel_doc/process_file.py | 8 +-- .../azure_cielo_qpcr/process_file.py | 8 +-- lambda/src/data_hub_lambda/constants.py | 4 -- .../epson_v700_scanner/process_file.py | 8 +-- lambda/src/data_hub_lambda/handler.py | 52 ++++----------- .../hina_microscope/process_file.py | 8 +-- .../spectramax_plate_reader/process_file.py | 8 +-- .../epson_v700_scanner/test_process_file.py | 4 +- lambda/tests/integration/conftest.py | 20 +----- lambda/tests/integration/test_lambda_api.py | 63 +----------------- packages/shared/src/data_hub_shared/config.py | 4 -- packages/shared/src/data_hub_shared/slack.py | 29 --------- web-app/.env.example | 5 ++ .../instruments/[instrumentId]/runs/route.ts | 18 ++++- web-app/lib/slack.ts | 32 +++++++++ web-app/tests/integration/global-setup.ts | 58 +++++++++++++++++ web-app/tests/integration/helpers.ts | 28 ++++++++ .../tests/integration/instrument-runs.test.ts | 65 ++++++++++++++++++- 32 files changed, 254 insertions(+), 257 deletions(-) delete mode 100644 lambda/src/data_hub_lambda/constants.py delete mode 100644 packages/shared/src/data_hub_shared/slack.py create mode 100644 web-app/lib/slack.ts diff --git a/.github/workflows/deploy-lambda.yml b/.github/workflows/deploy-lambda.yml index d9dd9633..20525f7c 100644 --- a/.github/workflows/deploy-lambda.yml +++ b/.github/workflows/deploy-lambda.yml @@ -65,7 +65,6 @@ jobs: "EcrImageUri=${IMAGE_URI}" \ "DataHubApiUrl=${{ secrets.DATA_HUB_API_URL }}" \ "DataHubApiKey=${{ secrets.DATA_HUB_API_KEY }}" \ - "SlackWebhookUrl=${{ secrets.SLACK_WEBHOOK_URL }}" \ "Environment=${{ github.ref_name }}" \ "GitHubOidcProviderArn=${{ secrets.GH_OIDC_PROVIDER_ARN }}" \ "VercelOidcProviderArn=${{ secrets.VERCEL_OIDC_PROVIDER_ARN }}" \ diff --git a/Makefile b/Makefile index b412070f..80eff460 100644 --- a/Makefile +++ b/Makefile @@ -146,7 +146,6 @@ endif "EcrImageUri=$(ECR_IMAGE_URI)" \ "DataHubApiUrl=$(DATA_HUB_API_URL)" \ "DataHubApiKey=$(DATA_HUB_API_KEY)" \ - "SlackWebhookUrl=$(SLACK_WEBHOOK_URL)" \ "GitHubOidcProviderArn=$(GITHUB_OIDC_PROVIDER_ARN)" \ "VercelOidcProviderArn=$(VERCEL_OIDC_PROVIDER_ARN)" \ "LambdaInvokeToken=$(LAMBDA_INVOKE_TOKEN)" diff --git a/README.md b/README.md index 1d53fc35..a86ec32c 100644 --- a/README.md +++ b/README.md @@ -19,7 +19,7 @@ flowchart LR | `web-app/` | Next.js web application and REST API (Vercel) | [API reference](docs/api.md) | | `lambda/` | AWS Lambda function for instrument data processing | [Lambda docs](docs/lambda.md) | | `watcher/` | CLI agent for lab instrument PCs | [Watcher docs](docs/watcher.md) | -| `packages/shared/` | Shared Python library (S3, enums, Slack, test infra) | [Shared library](docs/shared-library.md) | +| `packages/shared/` | Shared Python library (S3, enums, test infra) | [Shared library](docs/shared-library.md) | | `docs/` | Project documentation | — | ## Quick start diff --git a/docs/architecture.md b/docs/architecture.md index 2c6c9b7d..688a031a 100644 --- a/docs/architecture.md +++ b/docs/architecture.md @@ -27,7 +27,7 @@ flowchart LR | `web-app/` | `data-hub-web-app` | Next.js web application, REST API, and MCP server. Deployed on Vercel. | | `lambda/` | `data-hub-lambda` | AWS Lambda function triggered by S3 uploads. Runs instrument-specific processing pipelines. | | `watcher/` | `data-hub-watcher` | CLI agent installed on lab instrument PCs. Detects new files, uploads them to S3, and reports status to the API. | -| `packages/shared/` | `data-hub-shared` | Shared Python library providing S3 utilities, instrument enums, Slack integration, and test infrastructure. | +| `packages/shared/` | `data-hub-shared` | Shared Python library providing S3 utilities, instrument enums, and test infrastructure. | ## Data flow @@ -39,7 +39,7 @@ flowchart LR 4. It uploads raw files to **S3** at the key `{instrument_id}/{run_id}/{filename}`. 5. The S3 upload triggers the **Lambda** function. 6. Lambda downloads the file and dispatches to the appropriate instrument processor for preprocessing (e.g., extracting metadata). -7. Lambda creates/updates the run and files via the **API** and sends a **Slack** notification. +7. Lambda creates/updates the run and files via the **API**. The API sends a **Slack** notification once per newly-created run. 8. Users view the run in the **web dashboard**. ### Manual upload (manual mode) diff --git a/docs/ci-and-deployment.md b/docs/ci-and-deployment.md index b12edf44..1bbd5ac3 100644 --- a/docs/ci-and-deployment.md +++ b/docs/ci-and-deployment.md @@ -133,7 +133,6 @@ cp infra/.env.example infra/.env.staging ECR_IMAGE_URI= DATA_HUB_API_URL=https://data-hub-env-staging-arcadia-science.vercel.app/api/v1 DATA_HUB_API_KEY= -SLACK_WEBHOOK_URL= GITHUB_OIDC_PROVIDER_ARN= VERCEL_OIDC_PROVIDER_ARN= LAMBDA_INVOKE_TOKEN= @@ -169,9 +168,10 @@ In your GitHub repo, go to **Settings → Environments**, create a `staging` env | `SAM_S3_BUCKET` | SAM CLI managed S3 bucket name (see `sam deploy` output, e.g. `aws-sam-cli-managed-default-samclisourcebucket-*`) | | `DATA_HUB_API_URL` | Base API URL for the environment | | `DATA_HUB_API_KEY` | API key for Lambda → Data Hub authentication | -| `SLACK_WEBHOOK_URL` | Slack incoming webhook URL | | `LAMBDA_INVOKE_TOKEN` | Shared secret for web app → Lambda Function URL authentication | +Slack notifications are sent by the **web app** (not the Lambda) when a new run is created. Configure `SLACK_WEBHOOK_URL` per environment in the Vercel dashboard alongside the other web-app env vars listed below. + You'll also need the `WebAppRoleArn` and `DataHubFunctionUrl` stack outputs to configure the Vercel web app. In the Vercel dashboard (under the appropriate environment), set: | Vercel env var | Value | @@ -190,7 +190,7 @@ On pushes to `staging` or `production`, the **Deploy Lambda** workflow: 2. Builds and pushes the Docker image to ECR. 3. Runs `sam deploy` to update the CloudFormation stack. -Secrets (`DATA_HUB_API_KEY`, `SLACK_WEBHOOK_URL`, etc.) are stored in GitHub environment secrets scoped to each environment. +Secrets (`DATA_HUB_API_KEY`, `LAMBDA_INVOKE_TOKEN`, etc.) are stored in GitHub environment secrets scoped to each environment. > **Note:** The CI deploy role has intentionally narrow permissions — enough to push a new container image, update the existing CloudFormation stack, and modify the data buckets' S3 event notifications (so new instrument triggers roll out via CI), but _not_ enough to create the stack from scratch or to add/remove S3 buckets or Lambda functions. Initial stack creation and structural infrastructure changes must be performed by an admin with broader AWS permissions. Once the stack exists, routine image-update deploys and new-trigger rollouts through CI work without issue. diff --git a/docs/getting-started.md b/docs/getting-started.md index 4e894367..a45cb3c5 100644 --- a/docs/getting-started.md +++ b/docs/getting-started.md @@ -48,6 +48,7 @@ vercel env pull | `AWS_REGION` | No | AWS region for S3 presigned URLs (defaults to `us-west-1`) | | `AWS_ROLE_ARN` | No | IAM role ARN for Vercel OIDC federation to S3 (only needed on Vercel) | | `S3_RAW_DATA_BUCKET` | No | S3 bucket for raw data uploads (defaults to `arcadia-data-hub-raw-staging`) | +| `SLACK_WEBHOOK_URL` | No | Slack incoming webhook URL — when set, the web app posts a notification each time a new run is created | ### Lambda / shared library @@ -60,7 +61,6 @@ These are set in the Lambda runtime environment: | `AWS_SECRET_ACCESS_KEY` | No | AWS credentials | | `AWS_S3_RAW_DATA_BUCKET` | No | S3 bucket for raw data | | `AWS_S3_PROCESSED_DATA_BUCKET` | No | S3 bucket for processed data | -| `SLACK_WEBHOOK_URL` | No | Slack webhook for notifications | ### Watcher diff --git a/docs/guides/adding-an-instrument.md b/docs/guides/adding-an-instrument.md index b171ad47..336bb545 100644 --- a/docs/guides/adding-an-instrument.md +++ b/docs/guides/adding-an-instrument.md @@ -76,8 +76,8 @@ lambda/src/data_hub_lambda/bio_rad_cfx96/ The processor must expose a `process_file` function: ```python -def process_file(run_id: str, filename: str) -> str: - """Preprocess a file and return the URL to the run in the web dashboard.""" +def process_file(run_id: str, filename: str) -> None: + """Preprocess a file, reporting progress through the Data Hub API.""" ... ``` @@ -90,7 +90,6 @@ A typical processor: 5. Performs instrument-specific preprocessing (parsing, metadata extraction, etc.). 6. Optionally uploads processed artifacts (CSV, images) to the S3 processed bucket and registers them via `client.create_file(..., category="processed")`. This is the pattern used by the SpectraMax plate reader (processed CSV) and Azure 600 Gel Doc (contrast-enhanced PNG). 7. Updates the raw file status to `completed`. -8. Returns the web app URL for the run. See any existing processor (e.g., `lambda/src/data_hub_lambda/azure_cielo_qpcr/process_file.py` for simple metadata extraction, or `lambda/src/data_hub_lambda/spectramax_plate_reader/process_file.py` for the processed-artifact pattern) for complete examples. @@ -100,7 +99,7 @@ Add an `elif` branch in the `lambda_handler` function in `lambda/src/data_hub_la ```python elif instrument_id == Instrument.BIO_RAD_CFX96.value: - result_url = bio_rad_cfx96.process_file( + bio_rad_cfx96.process_file( run_id=event_info.run_id, filename=event_info.filename, ) @@ -139,4 +138,4 @@ Even without Step 4, you get a fully functional instrument in Data Hub: - Files are downloadable via pre-signed S3 URLs. - Watcher health monitoring (heartbeats, events) works in the dashboard. -Lambda preprocessing adds automated metadata extraction and Slack notifications on top of that. +Lambda preprocessing adds automated metadata extraction on top of that. (Slack notifications fire from the web app whenever a new run is created, regardless of whether a Lambda processor exists for the instrument.) diff --git a/docs/lambda.md b/docs/lambda.md index d5d1cb4f..583ec390 100644 --- a/docs/lambda.md +++ b/docs/lambda.md @@ -1,6 +1,6 @@ # Lambda -The Data Hub Lambda function preprocesses raw instrument data uploaded to S3 and builds run-archive zips on demand for the web app's "Download all" actions. It can be triggered automatically by S3 events or manually via Function URL invocations from the web app. It runs an instrument-specific preprocessing pipeline (or the archive builder) and reports results back through the API and Slack. +The Data Hub Lambda function preprocesses raw instrument data uploaded to S3 and builds run-archive zips on demand for the web app's "Download all" actions. It can be triggered automatically by S3 events or manually via Function URL invocations from the web app. It runs an instrument-specific preprocessing pipeline (or the archive builder) and reports results back through the API. ## How it works @@ -12,8 +12,8 @@ The Lambda has three invocation paths: 2. The handler parses the S3 key to extract the instrument ID, run ID, and filename. The expected key layout is `{instrument_id}/{run_id}/{filename}`. 3. It dispatches to the appropriate instrument processor based on the instrument ID. 4. The processor downloads the raw file from S3, preprocesses it (e.g., extracting metadata), and creates/updates the run and files via the Data Hub API. -5. On success, a Slack message is sent with a link to view the run in the web dashboard. -6. On failure, a Slack message is sent with a link to the CloudWatch logs. + +Slack notifications are sent by the **web app**, not the Lambda — see [Slack notifications](#slack-notifications) below. ### Function URL (manual reprocessing) @@ -23,7 +23,7 @@ When a file fails processing (or needs to be re-run), users can trigger reproces 2. The web app's `POST /api/v1/files/:fileId/reprocess` endpoint transitions the file to `processing` status, clears any previous error, and sends a POST request to the Lambda Function URL. 3. The request includes an `Authorization: Bearer ` header and a JSON body containing a synthetic S3 event payload. 4. The Lambda handler detects the Function URL invocation (via `requestContext` in the event), verifies the Bearer token against the `LAMBDA_INVOKE_TOKEN` environment variable using constant-time comparison, and parses the S3 event from the request body. -5. From here, processing follows the same dispatch logic as the S3 trigger path (steps 2–6 above). +5. From here, processing follows the same dispatch logic as the S3 trigger path (steps 2–4 above). ### Function URL (archive build) @@ -49,7 +49,11 @@ See [Run archives](run-archives.md) for the full flow, S3 bucket layout, cache s | SpectraMax iD3 Plate Reader | `spectramax_plate_reader` | `spectramax-id3-plate-reader` | | SpectraMax iD5 Plate Reader | `spectramax_plate_reader` | `spectramax-id5-plate-reader` | -Each processor module exposes a `process_file()` function that accepts the run ID and filename (and instrument ID for SpectraMax readers) and returns a URL to the run in the web dashboard. +Each processor module exposes a `process_file()` function that accepts the run ID and filename (and instrument ID for SpectraMax readers) and reports progress back through the Data Hub API. + +## Slack notifications + +Slack notifications are sent by the **web app** (`web-app/lib/slack.ts`), not the Lambda. When the Lambda's `process_file` calls `POST /api/v1/instruments/:instrumentId/runs` to register a newly-detected run, that endpoint posts a single message per run to `SLACK_WEBHOOK_URL` (configured per environment in Vercel). Subsequent files for the same run do not re-notify because the upsert is idempotent on `(instrument_id, run_id)`. File-level failures remain visible in the web app via the file row's `status='failed'` and `error_message` fields. ## Adding a new instrument @@ -58,8 +62,8 @@ Each processor module exposes a `process_file()` function that accepts the run I 2. **Create a processor module.** Add a new module under `lambda/src/data_hub_lambda/` (e.g., `new_instrument.py`). It must expose: ```python - def process_file(run_id: str, filename: str) -> str: - """Process a file and return the URL to the run in the web dashboard.""" + def process_file(run_id: str, filename: str) -> None: + """Process a file, reporting progress via the Data Hub API.""" ... ``` @@ -121,4 +125,4 @@ The Lambda function depends on a scientific Python stack: - `pydantic` — data validation - `requests` — HTTP client for the Data Hub API - `aws-lambda-typing` — type stubs for Lambda events/context -- `data-hub-shared` — shared utilities (S3, Slack, enums) +- `data-hub-shared` — shared utilities (S3, enums) diff --git a/docs/shared-library.md b/docs/shared-library.md index defcd414..b6b4abff 100644 --- a/docs/shared-library.md +++ b/docs/shared-library.md @@ -41,7 +41,6 @@ Shared environment-based configuration. Instantiated as a module-level singleton from data_hub_shared.config import config config.AWS_S3_RAW_DATA_BUCKET -config.SLACK_WEBHOOK_URL ``` | Attribute | Source env var | Default | @@ -53,7 +52,6 @@ config.SLACK_WEBHOOK_URL | `AWS_SESSION_TOKEN` | `AWS_SESSION_TOKEN` | `None` | | `AWS_S3_RAW_DATA_BUCKET` | `AWS_S3_RAW_DATA_BUCKET` | `None` | | `AWS_S3_PROCESSED_DATA_BUCKET` | `AWS_S3_PROCESSED_DATA_BUCKET` | `None` | -| `SLACK_WEBHOOK_URL` | `SLACK_WEBHOOK_URL` | `None` | ### `s3_utils` @@ -69,18 +67,6 @@ Boto3-based S3 utilities. Callers can pass an explicit `s3_client` or let the mo | `upload_folder(local_path, s3_uri_prefix)` | Upload all files in a directory | | `get_content_type(file_path)` | Guess MIME type for a file | -### `slack` - -Posts messages to a Slack channel via a webhook URL: - -```python -from data_hub_shared import slack - -slack.send_message("Hello from Data Hub!") -``` - -If `SLACK_WEBHOOK_URL` is not set, messages are silently skipped with a warning log. - ### `logger` Provides a `get_named_logger(name)` helper for consistent logging setup. @@ -98,5 +84,4 @@ Integration test infrastructure used by both Lambda and watcher test suites. Pro ## Dependencies - `boto3` — AWS SDK for S3 operations -- `requests` — HTTP client for Slack webhooks - `psycopg2-binary` — PostgreSQL driver (dev dependency, used by `testing.py`) diff --git a/infra/.env.example b/infra/.env.example index b78133c8..f534b782 100644 --- a/infra/.env.example +++ b/infra/.env.example @@ -4,7 +4,6 @@ ECR_IMAGE_URI= DATA_HUB_API_URL= DATA_HUB_API_KEY= -SLACK_WEBHOOK_URL= GITHUB_OIDC_PROVIDER_ARN= VERCEL_OIDC_PROVIDER_ARN= LAMBDA_INVOKE_TOKEN= diff --git a/infra/template.yaml b/infra/template.yaml index f35ad6ad..ddeb3b12 100644 --- a/infra/template.yaml +++ b/infra/template.yaml @@ -32,11 +32,6 @@ Parameters: NoEcho: true Description: API key for Lambda to authenticate with the Data Hub API - SlackWebhookUrl: - Type: String - NoEcho: true - Description: Slack incoming webhook URL for Lambda notifications - LambdaInvokeToken: Type: String NoEcho: true @@ -223,7 +218,6 @@ Resources: AWS_S3_ARCHIVES_BUCKET: !Sub "arcadia-data-hub-archives-${Environment}" DATA_HUB_API_URL: !Ref DataHubApiUrl DATA_HUB_API_KEY: !Ref DataHubApiKey - SLACK_WEBHOOK_URL: !Ref SlackWebhookUrl LAMBDA_INVOKE_TOKEN: !Ref LambdaInvokeToken # ---- S3 → Lambda event wiring ---- diff --git a/lambda/.env.example b/lambda/.env.example index d950ad1b..725d4343 100644 --- a/lambda/.env.example +++ b/lambda/.env.example @@ -13,8 +13,4 @@ AWS_S3_PROCESSED_DATA_BUCKET= # - access to push images to the ECR repository # - access to update the Lambda function code AWS_ACCESS_KEY_ID= -AWS_SECRET_ACCESS_KEY= - -# This is the Slack webhook URL for the channel to send messages to on workflow completion. -# See: https://docs.slack.dev/messaging/sending-messages-using-incoming-webhooks/. -SLACK_WEBHOOK_URL= \ No newline at end of file +AWS_SECRET_ACCESS_KEY= \ No newline at end of file diff --git a/lambda/src/data_hub_lambda/agilent_4150_tapestation/process_file.py b/lambda/src/data_hub_lambda/agilent_4150_tapestation/process_file.py index b744d531..9dd9283e 100644 --- a/lambda/src/data_hub_lambda/agilent_4150_tapestation/process_file.py +++ b/lambda/src/data_hub_lambda/agilent_4150_tapestation/process_file.py @@ -3,7 +3,6 @@ from data_hub_lambda.agilent_4150_tapestation.utils import parse_tape_type from data_hub_lambda.api_client import get_client -from data_hub_lambda.constants import DATA_HUB_WEB_URL from data_hub_shared import s3_utils from data_hub_shared.config import config from data_hub_shared.enums import Instrument @@ -13,15 +12,12 @@ INSTRUMENT_ID = Instrument.AGILENT_4150_TAPESTATION.value -def process_file(run_id: str, filename: str) -> str: +def process_file(run_id: str, filename: str) -> None: """Process a single Agilent 4150 TapeStation file through the Data Hub API. Args: run_id: The run ID (`YYYY-MM-DD - HH-MM-SS` prefix). filename: The original filename (e.g. `2026-02-18 - 18-00-04-gDNA_peakTable.csv`). - - Returns: - The web app URL for the instrument run. """ logger.info("Processing Agilent 4150 TapeStation file: %s (run: %s)", filename, run_id) @@ -62,5 +58,3 @@ def process_file(run_id: str, filename: str) -> str: logger.error("Error processing file: %s", e) client.update_file(file_id, status="failed", error_message=str(e)) raise - - return f"{DATA_HUB_WEB_URL}/instruments/{INSTRUMENT_ID}/runs/{run_id}" diff --git a/lambda/src/data_hub_lambda/akta_fplc/process_file.py b/lambda/src/data_hub_lambda/akta_fplc/process_file.py index 73145855..0b7a67df 100644 --- a/lambda/src/data_hub_lambda/akta_fplc/process_file.py +++ b/lambda/src/data_hub_lambda/akta_fplc/process_file.py @@ -2,7 +2,6 @@ import logging from data_hub_lambda.api_client import get_client -from data_hub_lambda.constants import DATA_HUB_WEB_URL from data_hub_shared import s3_utils from data_hub_shared.config import config from data_hub_shared.enums import Instrument @@ -12,15 +11,12 @@ INSTRUMENT_ID = Instrument.AKTA_FPLC.value -def process_file(run_id: str, filename: str) -> str: +def process_file(run_id: str, filename: str) -> None: """Process a single Akta FPLC file through the Data Hub API. Args: run_id: The run ID (filename stem). filename: The original filename (e.g. `2025-09-23_test.pdf`). - - Returns: - The web app URL for the instrument run. """ logger.info("Processing Akta FPLC file: %s (run: %s)", filename, run_id) @@ -54,5 +50,3 @@ def process_file(run_id: str, filename: str) -> str: logger.error("Error processing file: %s", e) client.update_file(file_id, status="failed", error_message=str(e)) raise - - return f"{DATA_HUB_WEB_URL}/instruments/{INSTRUMENT_ID}/runs/{run_id}" diff --git a/lambda/src/data_hub_lambda/azure_600_gel_doc/process_file.py b/lambda/src/data_hub_lambda/azure_600_gel_doc/process_file.py index f4fd7e82..3b6af421 100644 --- a/lambda/src/data_hub_lambda/azure_600_gel_doc/process_file.py +++ b/lambda/src/data_hub_lambda/azure_600_gel_doc/process_file.py @@ -4,7 +4,6 @@ from data_hub_lambda.api_client import get_client from data_hub_lambda.azure_600_gel_doc.image_processing import TIFFProcessor from data_hub_lambda.azure_600_gel_doc.parse_metadata import parse_metadata -from data_hub_lambda.constants import DATA_HUB_WEB_URL from data_hub_shared import s3_utils from data_hub_shared.config import config from data_hub_shared.enums import Instrument @@ -14,7 +13,7 @@ INSTRUMENT_ID = Instrument.AZURE_600_GEL_DOC.value -def process_file(run_id: str, filename: str) -> str: +def process_file(run_id: str, filename: str) -> None: """Process a single Azure 600 Gel Doc file through the Data Hub API. Downloads the raw TIFF, runs it through the image processing pipeline @@ -24,9 +23,6 @@ def process_file(run_id: str, filename: str) -> str: Args: run_id: The run ID (filename stem). filename: The original filename (e.g. `26.04.01_16.51.59.tif`). - - Returns: - The web app URL for the instrument run. """ logger.info("Processing Azure 600 Gel Doc file: %s (run: %s)", filename, run_id) @@ -87,5 +83,3 @@ def process_file(run_id: str, filename: str) -> str: logger.error("Error processing file: %s", e) client.update_file(file_id, status="failed", error_message=str(e)) raise - - return f"{DATA_HUB_WEB_URL}/instruments/{INSTRUMENT_ID}/runs/{run_id}" diff --git a/lambda/src/data_hub_lambda/azure_cielo_qpcr/process_file.py b/lambda/src/data_hub_lambda/azure_cielo_qpcr/process_file.py index 127d8804..8f7a1b61 100644 --- a/lambda/src/data_hub_lambda/azure_cielo_qpcr/process_file.py +++ b/lambda/src/data_hub_lambda/azure_cielo_qpcr/process_file.py @@ -3,7 +3,6 @@ from data_hub_lambda.api_client import get_client from data_hub_lambda.azure_cielo_qpcr.parse_dye_channels import parse_dye_channels -from data_hub_lambda.constants import DATA_HUB_WEB_URL from data_hub_shared import s3_utils from data_hub_shared.config import config from data_hub_shared.enums import Instrument @@ -13,7 +12,7 @@ INSTRUMENT_ID = Instrument.AZURE_CIELO_QPCR.value -def process_file(run_id: str, filename: str) -> str: +def process_file(run_id: str, filename: str) -> None: """Process a single Azure Cielo qPCR file through the Data Hub API. For Cq Values CSV files, the unique dye channel names are extracted from @@ -22,9 +21,6 @@ def process_file(run_id: str, filename: str) -> str: Args: run_id: The run ID (`Experiment_YYYYMMDD` prefix). filename: The original filename (e.g. `Experiment_20260101_CqValues.csv`). - - Returns: - The web app URL for the instrument run. """ logger.info("Processing Azure Cielo qPCR file: %s (run: %s)", filename, run_id) @@ -65,5 +61,3 @@ def process_file(run_id: str, filename: str) -> str: logger.error("Error processing file: %s", e) client.update_file(file_id, status="failed", error_message=str(e)) raise - - return f"{DATA_HUB_WEB_URL}/instruments/{INSTRUMENT_ID}/runs/{run_id}" diff --git a/lambda/src/data_hub_lambda/constants.py b/lambda/src/data_hub_lambda/constants.py deleted file mode 100644 index 600aff78..00000000 --- a/lambda/src/data_hub_lambda/constants.py +++ /dev/null @@ -1,4 +0,0 @@ -from __future__ import annotations - -# Base URL for building web app links included in Slack notifications. -DATA_HUB_WEB_URL = "https://data-hub.arcadiascience.com" diff --git a/lambda/src/data_hub_lambda/epson_v700_scanner/process_file.py b/lambda/src/data_hub_lambda/epson_v700_scanner/process_file.py index ae95ba33..ba157e67 100644 --- a/lambda/src/data_hub_lambda/epson_v700_scanner/process_file.py +++ b/lambda/src/data_hub_lambda/epson_v700_scanner/process_file.py @@ -2,7 +2,6 @@ import logging from data_hub_lambda.api_client import get_client -from data_hub_lambda.constants import DATA_HUB_WEB_URL from data_hub_lambda.epson_v700_scanner.image_processing import TIFFToJPEGConverter from data_hub_shared import s3_utils from data_hub_shared.config import config @@ -13,7 +12,7 @@ INSTRUMENT_ID = Instrument.EPSON_V700_SCANNER.value -def process_file(run_id: str, filename: str) -> str: +def process_file(run_id: str, filename: str) -> None: """Process a single Epson V700 Scanner file through the Data Hub API. Downloads the raw TIFF, resizes it to a web-friendly JPEG, uploads the @@ -23,9 +22,6 @@ def process_file(run_id: str, filename: str) -> str: Args: run_id: The run ID. filename: The original filename (e.g. ``scan_001.tif``). - - Returns: - The web app URL for the instrument run. """ logger.info("Processing Epson V700 Scanner file: %s (run: %s)", filename, run_id) @@ -86,5 +82,3 @@ def process_file(run_id: str, filename: str) -> str: logger.error("Error processing file: %s", e) client.update_file(file_id, status="failed", error_message=str(e)) raise - - return f"{DATA_HUB_WEB_URL}/instruments/{INSTRUMENT_ID}/runs/{run_id}" diff --git a/lambda/src/data_hub_lambda/handler.py b/lambda/src/data_hub_lambda/handler.py index e78c7e12..7967f18f 100644 --- a/lambda/src/data_hub_lambda/handler.py +++ b/lambda/src/data_hub_lambda/handler.py @@ -7,7 +7,7 @@ from dataclasses import dataclass from pprint import pformat from typing import Any -from urllib.parse import quote, unquote_plus +from urllib.parse import unquote_plus from aws_lambda_typing.context import Context from aws_lambda_typing.events.s3 import S3Event @@ -22,8 +22,6 @@ hina_microscope, spectramax_plate_reader, ) -from data_hub_shared import slack -from data_hub_shared.constants import INSTRUMENT_ID_TO_NAME_MAP from data_hub_shared.enums import Instrument from data_hub_shared.logger import get_named_logger @@ -86,22 +84,6 @@ def parse_s3_event(event: S3Event) -> S3EventInfo: ) -# ------------------------------------------------------------------ -# Helpers -# ------------------------------------------------------------------ - - -def get_cloudwatch_logs_url(context: Context) -> str: - """Generates the CloudWatch logs URL for the current Lambda execution.""" - arn_parts = context.invoked_function_arn.split(":") - region = arn_parts[3] if len(arn_parts) > 3 else "us-east-1" - log_group_encoded = quote(context.log_group_name, safe="") - log_stream_encoded = quote(context.log_stream_name, safe="") - base_url = f"https://{region}.console.aws.amazon.com/cloudwatch/home" - logs_path = f"log-groups/log-group/{log_group_encoded}/log-events/{log_stream_encoded}" - return f"{base_url}?region={region}#logsV2:{logs_path}" - - # ------------------------------------------------------------------ # Function URL helpers # ------------------------------------------------------------------ @@ -358,7 +340,6 @@ def lambda_handler(event: dict[str, Any], context: Context) -> dict[str, Any] | run_id = event_info.run_id logger.info("Instrument ID: '%s'", instrument_id) logger.info("Run ID: '%s'", run_id) - instrument_name = INSTRUMENT_ID_TO_NAME_MAP[instrument_id] # Pre-cleanup: if the previous invocation on this warm container was # SIGKILL'd (e.g. OOM), the `finally` block below didn't run and stale @@ -369,37 +350,37 @@ def lambda_handler(event: dict[str, Any], context: Context) -> dict[str, Any] | logger.info("Processing file %s...", event_info.filename) if instrument_id == Instrument.AKTA_FPLC.value: - result_url = akta_fplc.process_file( + akta_fplc.process_file( run_id=event_info.run_id, filename=event_info.filename, ) elif instrument_id == Instrument.AGILENT_4150_TAPESTATION.value: - result_url = agilent_4150_tapestation.process_file( + agilent_4150_tapestation.process_file( run_id=event_info.run_id, filename=event_info.filename, ) elif instrument_id == Instrument.AZURE_600_GEL_DOC.value: - result_url = azure_600_gel_doc.process_file( + azure_600_gel_doc.process_file( run_id=event_info.run_id, filename=event_info.filename, ) elif instrument_id == Instrument.AZURE_CIELO_QPCR.value: - result_url = azure_cielo_qpcr.process_file( + azure_cielo_qpcr.process_file( run_id=event_info.run_id, filename=event_info.filename, ) elif instrument_id == Instrument.EPSON_V700_SCANNER.value: - result_url = epson_v700_scanner.process_file( + epson_v700_scanner.process_file( run_id=event_info.run_id, filename=event_info.filename, ) elif instrument_id == Instrument.HINA_MICROSCOPE.value: - result_url = hina_microscope.process_file( + hina_microscope.process_file( run_id=event_info.run_id, filename=event_info.filename, ) @@ -408,7 +389,7 @@ def lambda_handler(event: dict[str, Any], context: Context) -> dict[str, Any] | Instrument.SPECTRAMAX_ID3_PLATE_READER.value, Instrument.SPECTRAMAX_ID5_PLATE_READER.value, ): - result_url = spectramax_plate_reader.process_file( + spectramax_plate_reader.process_file( instrument_id=event_info.instrument_id, # pyright: ignore[reportArgumentType] run_id=event_info.run_id, filename=event_info.filename, @@ -418,18 +399,13 @@ def lambda_handler(event: dict[str, Any], context: Context) -> dict[str, Any] | logger.error("Unsupported instrument: %s", instrument_id) return None - slack.send_message( - f"*{instrument_name}*\n" - f"Finished preprocessing run `{run_id}`!\n" - f"<{result_url}|View in Data Hub>" - ) except Exception: + # Per-file failure is already PATCHed back to the web app's file row + # (status='failed', error_message=...) by each instrument's + # `process_file`, so the failure remains visible in the UI without a + # Slack notification here. logger.exception("Failed to preprocess run %s.", run_id) - logs_url = get_cloudwatch_logs_url(context) - slack.send_message( - f"*{instrument_name}*\n" - f"Failed to preprocess run `{run_id}`!\n" - f"<{logs_url}|View CloudWatch logs>" - ) finally: _cleanup_tmp() + + return None diff --git a/lambda/src/data_hub_lambda/hina_microscope/process_file.py b/lambda/src/data_hub_lambda/hina_microscope/process_file.py index 37cf2fec..b027c71f 100644 --- a/lambda/src/data_hub_lambda/hina_microscope/process_file.py +++ b/lambda/src/data_hub_lambda/hina_microscope/process_file.py @@ -2,7 +2,6 @@ import logging from data_hub_lambda.api_client import get_client -from data_hub_lambda.constants import DATA_HUB_WEB_URL from data_hub_lambda.hina_microscope.image_processing import ND2Processor from data_hub_lambda.hina_microscope.parse_metadata import parse_metadata from data_hub_shared import s3_utils @@ -14,7 +13,7 @@ INSTRUMENT_ID = Instrument.HINA_MICROSCOPE.value -def process_file(run_id: str, filename: str) -> str: +def process_file(run_id: str, filename: str) -> None: """Process a single Hina microscope ND2 file through the Data Hub API. Downloads the raw ND2, runs it through the image processing pipeline to @@ -27,9 +26,6 @@ def process_file(run_id: str, filename: str) -> str: Args: run_id: The run ID (grouping key for files in a single imaging session). filename: The original filename (e.g. `well_A1_xy01.nd2`). - - Returns: - The web app URL for the instrument run. """ logger.info("Processing Hina microscope file: %s (run: %s)", filename, run_id) @@ -96,5 +92,3 @@ def process_file(run_id: str, filename: str) -> str: logger.error("Error processing file: %s", e) client.update_file(file_id, status="failed", error_message=str(e)) raise - - return f"{DATA_HUB_WEB_URL}/instruments/{INSTRUMENT_ID}/runs/{run_id}" diff --git a/lambda/src/data_hub_lambda/spectramax_plate_reader/process_file.py b/lambda/src/data_hub_lambda/spectramax_plate_reader/process_file.py index d864f95b..a3a877d2 100644 --- a/lambda/src/data_hub_lambda/spectramax_plate_reader/process_file.py +++ b/lambda/src/data_hub_lambda/spectramax_plate_reader/process_file.py @@ -3,7 +3,6 @@ from typing import Literal from data_hub_lambda.api_client import get_client -from data_hub_lambda.constants import DATA_HUB_WEB_URL from data_hub_lambda.spectramax_plate_reader.utils import parse_metadata, parse_raw_well_data from data_hub_shared import s3_utils from data_hub_shared.config import config @@ -13,16 +12,13 @@ InstrumentType = Literal["spectramax-id3-plate-reader", "spectramax-id5-plate-reader"] -def process_file(instrument_id: InstrumentType, run_id: str, filename: str) -> str: +def process_file(instrument_id: InstrumentType, run_id: str, filename: str) -> None: """Process a single SpectraMax plate reader file through the Data Hub API. Args: instrument_id: The instrument ID (iD3 or iD5). run_id: The run ID (filename stem). filename: The original filename (e.g. `033126_CM_Od750.xls`). - - Returns: - The web app URL for the instrument run. """ logger.info("Processing SpectraMax file: %s (run: %s)", filename, run_id) @@ -87,5 +83,3 @@ def process_file(instrument_id: InstrumentType, run_id: str, filename: str) -> s logger.error("Error processing file: %s", e) client.update_file(file_id, status="failed", error_message=str(e)) raise - - return f"{DATA_HUB_WEB_URL}/instruments/{instrument_id}/runs/{run_id}" diff --git a/lambda/tests/epson_v700_scanner/test_process_file.py b/lambda/tests/epson_v700_scanner/test_process_file.py index 8231e146..cd08522d 100644 --- a/lambda/tests/epson_v700_scanner/test_process_file.py +++ b/lambda/tests/epson_v700_scanner/test_process_file.py @@ -103,14 +103,12 @@ def test_creates_run_and_files( ): from data_hub_lambda.epson_v700_scanner.process_file import process_file - result_url = process_file(run_id="run-xyz", filename="scan.tif") + process_file(run_id="run-xyz", filename="scan.tif") client.ensure_run.assert_called_once() assert client.create_file.call_count == 2 s3_mock.upload_file.assert_called_once() client.update_run.assert_called_once() - assert "epson-v700-scanner" in result_url - assert "run-xyz" in result_url def test_uploads_jpg_and_registers_processed_file( self, diff --git a/lambda/tests/integration/conftest.py b/lambda/tests/integration/conftest.py index d96dd26d..3d4fc968 100644 --- a/lambda/tests/integration/conftest.py +++ b/lambda/tests/integration/conftest.py @@ -2,7 +2,7 @@ These fixtures spin up a real Next.js server backed by Postgres so that `lambda_handler` -> `process_file` -> `DataHubClient` exercises the -full HTTP path with only S3 downloads and Slack mocked. +full HTTP path with only S3 downloads mocked. """ from __future__ import annotations @@ -270,18 +270,6 @@ def mock_s3_upload() -> Generator[MagicMock, None, None]: yield mock -# --------------------------------------------------------------------------- -# Step 3c — Slack mock -# --------------------------------------------------------------------------- - - -@pytest.fixture(autouse=True) -def mock_slack() -> Generator[MagicMock, None, None]: - """Patch `slack.send_message` as a no-op (captures calls for assertions).""" - with patch("data_hub_shared.slack.send_message") as mock: - yield mock - - # --------------------------------------------------------------------------- # Step 5 — mock Lambda context # --------------------------------------------------------------------------- @@ -289,11 +277,7 @@ def mock_slack() -> Generator[MagicMock, None, None]: @pytest.fixture() def mock_context() -> MagicMock: - """Lightweight mock for `aws_lambda_typing.context.Context`. - - Provides the attributes accessed by `get_cloudwatch_logs_url` so it - doesn't crash on the failure path. - """ + """Lightweight mock for `aws_lambda_typing.context.Context`.""" from aws_lambda_typing.context import Context ctx = MagicMock(spec=Context) diff --git a/lambda/tests/integration/test_lambda_api.py b/lambda/tests/integration/test_lambda_api.py index 1ab80cb3..eb19e6c3 100644 --- a/lambda/tests/integration/test_lambda_api.py +++ b/lambda/tests/integration/test_lambda_api.py @@ -1,8 +1,8 @@ """Integration tests: Lambda -> API -> Postgres. Each test constructs a realistic S3 event, calls `lambda_handler` with -only S3 and Slack mocked, then verifies the outcome via API GETs and -direct DB queries. +only S3 mocked, then verifies the outcome via API GETs and direct DB +queries. """ from __future__ import annotations @@ -55,7 +55,6 @@ def test_csv_completes_with_dye_channels( make_s3_event: Callable[..., dict[str, Any]], s3_fixture_files: dict[str, Path], mock_context: MagicMock, - mock_slack: MagicMock, ) -> None: # Register the real fixture CSV so the patched S3 download can find it. run_id = "Experiment_20260101" @@ -90,11 +89,6 @@ def test_csv_completes_with_dye_channels( # qPCR files don't produce processed CSV files (unlike plate readers). assert all(f["category"] == "raw" for f in run["files"]) - mock_slack.assert_called_once() - slack_msg = mock_slack.call_args[0][0] - assert "Experiment_20260101" in slack_msg - assert "View in Data Hub" in slack_msg - # ------------------------------------------------------------------ # Test 4b: SpectraMax plate reader — happy path with processed CSV @@ -143,7 +137,6 @@ def test_xls_completes_with_processed_csv( make_s3_event: Callable[..., dict[str, Any]], s3_fixture_files: dict[str, Path], mock_context: MagicMock, - mock_slack: MagicMock, mock_s3_upload: MagicMock, fixture_file: str, run_id: str, @@ -185,11 +178,6 @@ def test_xls_completes_with_processed_csv( f"s3://test-processed-bucket/spectramax-id3-plate-reader/{run_id}/{run_id}_raw_well_data.csv" ) - mock_slack.assert_called_once() - slack_msg = mock_slack.call_args[0][0] - assert run_id in slack_msg - assert "View in Data Hub" in slack_msg - # ------------------------------------------------------------------ # Test 4b': Azure 600 Gel Doc — happy path with processed image @@ -203,7 +191,6 @@ def test_tif_completes_with_metadata_and_processed_image( make_s3_event: Callable[..., dict[str, Any]], s3_fixture_files: dict[str, Path], mock_context: MagicMock, - mock_slack: MagicMock, mock_s3_upload: MagicMock, ) -> None: # Register the real fixture TIFF so the patched S3 download can find it. @@ -254,12 +241,6 @@ def test_tif_completes_with_metadata_and_processed_image( upload_dest = mock_s3_upload.call_args[0][1] assert upload_dest == f"s3://test-processed-bucket/azure-600-gel-doc/{run_id}/{run_id}.png" - # Verify the Slack notification. - mock_slack.assert_called_once() - slack_msg = mock_slack.call_args[0][0] - assert run_id in slack_msg - assert "View in Data Hub" in slack_msg - # ------------------------------------------------------------------ # Test 4c: Malformed file — failure path @@ -273,7 +254,6 @@ def test_malformed_csv_marks_file_as_failed( make_s3_event: Callable[..., dict[str, Any]], s3_fixture_files: dict[str, Path], mock_context: MagicMock, - mock_slack: MagicMock, tmp_path: Path, ) -> None: # Create a CSV whose headers don't match the expected qPCR format. @@ -303,12 +283,6 @@ def test_malformed_csv_marks_file_as_failed( assert file["status"] == "failed" assert file["error_message"] - # Verify the Slack notification. - mock_slack.assert_called_once() - slack_msg = mock_slack.call_args[0][0] - assert run_id in slack_msg - assert "View CloudWatch logs" in slack_msg - # ------------------------------------------------------------------ # Test 4d: Idempotent run creation @@ -322,7 +296,6 @@ def test_duplicate_event_creates_single_run( make_s3_event: Callable[..., dict[str, Any]], s3_fixture_files: dict[str, Path], mock_context: MagicMock, - mock_slack: MagicMock, ) -> None: run_id = "Experiment_20260301" filename = f"{run_id}_CqValues.csv" @@ -362,12 +335,6 @@ def test_duplicate_event_creates_single_run( assert run["run_id"] == run_id assert run["source"] == "lambda" - assert mock_slack.call_count == 2 - for call in mock_slack.call_args_list: - slack_msg = call[0][0] - assert run_id in slack_msg - assert "View in Data Hub" in slack_msg - # ------------------------------------------------------------------ # Test 4e: File reprocessing @@ -381,7 +348,6 @@ def test_duplicate_event_creates_single_file( make_s3_event: Callable[..., dict[str, Any]], s3_fixture_files: dict[str, Path], mock_context: MagicMock, - mock_slack: MagicMock, ) -> None: """Firing the same S3 event twice must not create a duplicate file row. @@ -427,7 +393,6 @@ def test_reprocess_clears_failed_state( make_s3_event: Callable[..., dict[str, Any]], s3_fixture_files: dict[str, Path], mock_context: MagicMock, - mock_slack: MagicMock, tmp_path: Path, ) -> None: """A failed file can be reprocessed successfully on a retry. @@ -468,18 +433,12 @@ def test_reprocess_clears_failed_state( assert run["files"][0]["status"] == "completed" assert run["files"][0]["error_message"] is None - # Verify the Slack notifications. - assert mock_slack.call_count == 2 - assert "View CloudWatch logs" in mock_slack.call_args_list[0][0][0] - assert "View in Data Hub" in mock_slack.call_args_list[1][0][0] - def test_reprocess_does_not_duplicate_processed_csv( self, integration_env: IntegrationEnv, make_s3_event: Callable[..., dict[str, Any]], s3_fixture_files: dict[str, Path], mock_context: MagicMock, - mock_slack: MagicMock, mock_s3_upload: MagicMock, ) -> None: """Reprocessing a plate reader file must not duplicate the CSV file record. @@ -508,12 +467,6 @@ def test_reprocess_does_not_duplicate_processed_csv( assert len(processed_files) == 1 assert processed_files[0]["filename"] == f"{run_id}_raw_well_data.csv" - assert mock_slack.call_count == 2 - for call in mock_slack.call_args_list: - slack_msg = call[0][0] - assert run_id in slack_msg - assert "View in Data Hub" in slack_msg - # ------------------------------------------------------------------ # Test 4f: Function URL invocation @@ -527,7 +480,6 @@ def test_happy_path_processes_file( make_function_url_event: Callable[..., dict[str, Any]], s3_fixture_files: dict[str, Path], mock_context: MagicMock, - mock_slack: MagicMock, ) -> None: """A Function URL event with a valid token processes the file identically to a direct S3 trigger.""" @@ -551,14 +503,10 @@ def test_happy_path_processes_file( assert len(run["files"]) == 1 assert run["files"][0]["status"] == "completed" - mock_slack.assert_called_once() - assert "View in Data Hub" in mock_slack.call_args[0][0] - def test_wrong_token_returns_401( self, make_function_url_event: Callable[..., dict[str, Any]], mock_context: MagicMock, - mock_slack: MagicMock, ) -> None: """A Function URL event with an incorrect Bearer token is rejected.""" event = make_function_url_event( @@ -570,13 +518,11 @@ def test_wrong_token_returns_401( result = lambda_handler(event, mock_context) assert result == {"statusCode": 401, "body": "Unauthorized"} - mock_slack.assert_not_called() def test_missing_auth_header_returns_401( self, make_function_url_event: Callable[..., dict[str, Any]], mock_context: MagicMock, - mock_slack: MagicMock, ) -> None: """A Function URL event with no Authorization header is rejected.""" event = make_function_url_event( @@ -588,13 +534,11 @@ def test_missing_auth_header_returns_401( result = lambda_handler(event, mock_context) assert result == {"statusCode": 401, "body": "Unauthorized"} - mock_slack.assert_not_called() def test_unconfigured_token_returns_401( self, make_function_url_event: Callable[..., dict[str, Any]], mock_context: MagicMock, - mock_slack: MagicMock, monkeypatch: pytest.MonkeyPatch, ) -> None: """When LAMBDA_INVOKE_TOKEN is not set, all Function URL requests @@ -610,13 +554,11 @@ def test_unconfigured_token_returns_401( result = lambda_handler(event, mock_context) assert result == {"statusCode": 401, "body": "Unauthorized"} - mock_slack.assert_not_called() def test_invalid_json_body_returns_401( self, make_function_url_event: Callable[..., dict[str, Any]], mock_context: MagicMock, - mock_slack: MagicMock, ) -> None: """A Function URL event with a valid token but non-JSON body is rejected (the handler cannot parse the S3 event payload).""" @@ -629,4 +571,3 @@ def test_invalid_json_body_returns_401( result = lambda_handler(event, mock_context) assert result == {"statusCode": 401, "body": "Unauthorized"} - mock_slack.assert_not_called() diff --git a/packages/shared/src/data_hub_shared/config.py b/packages/shared/src/data_hub_shared/config.py index 5e027391..417b8742 100644 --- a/packages/shared/src/data_hub_shared/config.py +++ b/packages/shared/src/data_hub_shared/config.py @@ -22,8 +22,6 @@ class Config: AWS_S3_RAW_DATA_BUCKET: str | None AWS_S3_PROCESSED_DATA_BUCKET: str | None - SLACK_WEBHOOK_URL: str | None - def __init__(self) -> None: self.LOCAL_DATA_DIRPATH = Path(os.getenv("LOCAL_DATA_DIRPATH") or "/tmp/data") self.LOCAL_RAW_DATA_DIRPATH = self.LOCAL_DATA_DIRPATH / "raw-data" @@ -37,7 +35,5 @@ def __init__(self) -> None: self.AWS_S3_RAW_DATA_BUCKET = os.getenv("AWS_S3_RAW_DATA_BUCKET") self.AWS_S3_PROCESSED_DATA_BUCKET = os.getenv("AWS_S3_PROCESSED_DATA_BUCKET") - self.SLACK_WEBHOOK_URL = os.getenv("SLACK_WEBHOOK_URL") - config = Config() diff --git a/packages/shared/src/data_hub_shared/slack.py b/packages/shared/src/data_hub_shared/slack.py deleted file mode 100644 index 2d51b47e..00000000 --- a/packages/shared/src/data_hub_shared/slack.py +++ /dev/null @@ -1,29 +0,0 @@ -from __future__ import annotations -import logging - -import requests - -from data_hub_shared.config import config - -logger = logging.getLogger(__name__) - - -def send_message(message: str) -> None: - """Sends a message to the Slack channel via the configured webhook URL. - - Args: - message: The message text to send. - """ - if not config.SLACK_WEBHOOK_URL: - logger.warning("`SLACK_WEBHOOK_URL` is not set, skipping message.") - return - - payload = {"text": message} - - logger.info("Sending message to Slack: %s", message) - try: - response = requests.post(config.SLACK_WEBHOOK_URL, json=payload) - response.raise_for_status() - logger.info("Message sent to Slack.") - except requests.exceptions.HTTPError: - logger.exception("Failed to send message to Slack: %s", message) diff --git a/web-app/.env.example b/web-app/.env.example index bee03618..00e44015 100644 --- a/web-app/.env.example +++ b/web-app/.env.example @@ -22,6 +22,11 @@ S3_ARCHIVES_BUCKET=arcadia-data-hub-archives-staging LAMBDA_FUNCTION_URL= LAMBDA_INVOKE_TOKEN= +# Slack incoming webhook URL for run-creation notifications. +# See: https://docs.slack.dev/messaging/sending-messages-using-incoming-webhooks/. +# Leave empty to disable Slack notifications. +SLACK_WEBHOOK_URL= + # Watcher release metadata served by GET /api/v1/watchers/:id/update-check. # Leave WATCHER_LATEST_VERSION unset to disable self-updates (the endpoint # returns latest_version: null and clients skip the upgrade). Set diff --git a/web-app/app/api/v1/instruments/[instrumentId]/runs/route.ts b/web-app/app/api/v1/instruments/[instrumentId]/runs/route.ts index 4516bf07..1be66f77 100644 --- a/web-app/app/api/v1/instruments/[instrumentId]/runs/route.ts +++ b/web-app/app/api/v1/instruments/[instrumentId]/runs/route.ts @@ -9,6 +9,7 @@ import { buildRunListQuery } from "@/lib/api/instrument-runs"; import { parseIntParam } from "@/lib/api/validators"; import { db } from "@/lib/db"; import { files, instrumentRuns, instruments, watchers } from "@/lib/db/schema"; +import { sendSlackMessage } from "@/lib/slack"; import { and, eq, isNull } from "drizzle-orm"; import type { NextRequest } from "next/server"; @@ -36,7 +37,7 @@ export async function POST(request: NextRequest, { params }: RouteContext) { const { instrumentId } = await params; const [instrument] = await db - .select({ id: instruments.id }) + .select({ id: instruments.id, displayName: instruments.displayName }) .from(instruments) .where(eq(instruments.id, instrumentId)) .limit(1); @@ -161,6 +162,21 @@ export async function POST(request: NextRequest, { params }: RouteContext) { await db.insert(files).values(fileValues).onConflictDoNothing(); } + // One Slack notification per newly-created run. The upsert above + // (`onConflictDoNothing`) gives us the "first time only" guarantee, so + // both the lambda-auto-create and watcher-report paths fire at most once + // per (instrument_id, run_id). Awaited so the message is delivered before + // the serverless function instance is recycled; `sendSlackMessage` + // swallows its own errors so a Slack outage cannot fail the API request. + if (isNew) { + const origin = new URL(request.url).origin; + await sendSlackMessage( + `*${instrument.displayName}*\n` + + `New run \`${runId}\` created (source: ${source}).\n` + + `<${origin}/instruments/${instrumentId}/runs/${encodeURIComponent(runId)}|View in Data Hub>` + ); + } + return Response.json( { id: run.id, diff --git a/web-app/lib/slack.ts b/web-app/lib/slack.ts new file mode 100644 index 00000000..2ebb6b98 --- /dev/null +++ b/web-app/lib/slack.ts @@ -0,0 +1,32 @@ +// 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. + +export async function sendSlackMessage(text: string): Promise { + const webhookUrl = process.env.SLACK_WEBHOOK_URL; + if (!webhookUrl) { + console.warn("SLACK_WEBHOOK_URL is not set, skipping Slack message."); + return; + } + + try { + const response = await fetch(webhookUrl, { + method: "POST", + headers: { "content-type": "application/json" }, + body: JSON.stringify({ text }), + }); + if (!response.ok) { + const body = await response.text().catch(() => ""); + console.error( + `Slack webhook returned ${response.status}: ${body.slice(0, 200)}` + ); + } + } catch (err) { + console.error("Failed to POST to Slack webhook:", err); + } +} diff --git a/web-app/tests/integration/global-setup.ts b/web-app/tests/integration/global-setup.ts index 1e87870f..6be3ab3c 100644 --- a/web-app/tests/integration/global-setup.ts +++ b/web-app/tests/integration/global-setup.ts @@ -1,5 +1,6 @@ import type { ChildProcess } from "node:child_process"; import { execSync, spawn } from "node:child_process"; +import http from "node:http"; import net from "node:net"; import postgres from "postgres"; @@ -9,6 +10,7 @@ const TEST_DB = "data_hub_test"; const PG_URL = `postgres://postgres:postgres@127.0.0.1:5432`; let serverProcess: ChildProcess | null = null; +let slackCaptureServer: http.Server | null = null; // Bind to port 0, let the OS assign a free port, then immediately release it. // This avoids hardcoding a port that might collide with other services. @@ -60,6 +62,52 @@ export async function setup() { const databaseUrl = `${PG_URL}/${TEST_DB}`; + // Stand up an in-process HTTP capture server so tests can assert on + // outgoing Slack webhook calls without depending on the real Slack API. + // The Next.js server (spawned below) is configured with a SLACK_WEBHOOK_URL + // that points at this capture endpoint; tests inspect captured payloads + // via `getCapturedSlackMessages()` in helpers.ts. + const captured: { text: string }[] = []; + const slackPort = await getFreePort(); + slackCaptureServer = http.createServer((req, res) => { + if (req.method === "POST" && req.url === "/webhook") { + let raw = ""; + req.on("data", (chunk) => { + raw += chunk; + }); + req.on("end", () => { + try { + const parsed = JSON.parse(raw); + if (typeof parsed.text === "string") { + captured.push({ text: parsed.text }); + } + } catch { + // ignore non-JSON bodies; never thrown by sendSlackMessage + } + res.writeHead(200, { "content-type": "text/plain" }); + res.end("ok"); + }); + return; + } + if (req.method === "GET" && req.url === "/captured") { + res.writeHead(200, { "content-type": "application/json" }); + res.end(JSON.stringify(captured)); + return; + } + if (req.method === "POST" && req.url === "/clear") { + captured.length = 0; + res.writeHead(200, { "content-type": "text/plain" }); + res.end("ok"); + return; + } + res.writeHead(404); + res.end(); + }); + await new Promise((resolve) => + slackCaptureServer!.listen(slackPort, "127.0.0.1", resolve) + ); + const slackCaptureBaseUrl = `http://127.0.0.1:${slackPort}`; + // 2. Push schema via drizzle-kit. --force skips the interactive confirmation // prompt that drizzle-kit shows when it detects destructive changes. execSync("npx drizzle-kit push --force", { @@ -109,6 +157,9 @@ export async function setup() { // `download-archive/route.ts` still trigger. LAMBDA_INVOKE_TOKEN: process.env.LAMBDA_INVOKE_TOKEN ?? "test-lambda-invoke-token", + // 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`, }; // Strip the Lambda Function URL so "not configured" test cases work // regardless of the developer's local .env. Tests that need a stubbed @@ -143,11 +194,18 @@ export async function setup() { process.env.__TEST_DATABASE_URL = databaseUrl; process.env.__TEST_LAMBDA_INVOKE_TOKEN = serverEnv.LAMBDA_INVOKE_TOKEN ?? "test-lambda-invoke-token"; + process.env.__TEST_SLACK_CAPTURE_URL = slackCaptureBaseUrl; return async () => { if (serverProcess) { serverProcess.kill("SIGTERM"); serverProcess = null; } + if (slackCaptureServer) { + await new Promise((resolve, reject) => + slackCaptureServer!.close((err) => (err ? reject(err) : resolve())) + ); + slackCaptureServer = null; + } }; } diff --git a/web-app/tests/integration/helpers.ts b/web-app/tests/integration/helpers.ts index 0a323a4f..357496d9 100644 --- a/web-app/tests/integration/helpers.ts +++ b/web-app/tests/integration/helpers.ts @@ -146,3 +146,31 @@ export async function api( body: body !== undefined ? JSON.stringify(body) : undefined, }); } + +// --------------------------------------------------------------------------- +// 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. +// --------------------------------------------------------------------------- + +function getSlackCaptureUrl(): string { + const url = process.env.__TEST_SLACK_CAPTURE_URL; + if (!url) + throw new Error("__TEST_SLACK_CAPTURE_URL not set — global setup failed?"); + return url; +} + +export async function getCapturedSlackMessages(): Promise<{ text: string }[]> { + const res = await fetch(`${getSlackCaptureUrl()}/captured`); + if (!res.ok) { + throw new Error(`Slack capture /captured returned ${res.status}`); + } + return res.json(); +} + +export async function clearCapturedSlackMessages(): Promise { + const res = await fetch(`${getSlackCaptureUrl()}/clear`, { method: "POST" }); + if (!res.ok) { + throw new Error(`Slack capture /clear returned ${res.status}`); + } +} diff --git a/web-app/tests/integration/instrument-runs.test.ts b/web-app/tests/integration/instrument-runs.test.ts index 1105a15f..578f4ae1 100644 --- a/web-app/tests/integration/instrument-runs.test.ts +++ b/web-app/tests/integration/instrument-runs.test.ts @@ -1,12 +1,14 @@ import { instruments } from "@/lib/db/schema"; import { api, + clearCapturedSlackMessages, closeTestDb, + getCapturedSlackMessages, getTestDb, resetDb, seedTestUser, } from "@/tests/integration/helpers"; -import { afterAll, beforeAll, describe, expect, it } from "vitest"; +import { afterAll, beforeAll, beforeEach, describe, expect, it } from "vitest"; describe("Instrument Runs API", () => { let token: string; @@ -290,3 +292,64 @@ describe("Instrument Runs API", () => { expect(res.status).toBe(409); }); }); + +// --------------------------------------------------------------------------- +// Slack notification on run creation +// +// The web app fires one Slack message per newly-created run. The +// `onConflictDoNothing` upsert in the route guarantees "first time only", so +// duplicate POSTs (which return 200) must not re-notify. We use a separate +// describe block so the capture buffer is isolated from the rest of the suite. +// --------------------------------------------------------------------------- + +describe("Run creation Slack notification", () => { + let token: string; + const instrumentId = "slack-notif-instrument"; + const instrumentDisplayName = "Slack Notification Instrument"; + + beforeAll(async () => { + ({ token } = await seedTestUser()); + const db = getTestDb(); + await db.insert(instruments).values({ + id: instrumentId, + displayName: instrumentDisplayName, + status: "active", + }); + }); + + beforeEach(async () => { + await clearCapturedSlackMessages(); + }); + + afterAll(async () => { + await closeTestDb(); + }); + + it("POST sends a Slack message on first-time run creation only", async () => { + const runId = "slack-run-001"; + + const first = await api(`/api/v1/instruments/${instrumentId}/runs`, { + method: "POST", + token, + body: { run_id: runId, source: "lambda" }, + }); + expect(first.status).toBe(201); + + const duplicate = await api(`/api/v1/instruments/${instrumentId}/runs`, { + method: "POST", + token, + body: { run_id: runId, source: "lambda" }, + }); + expect(duplicate.status).toBe(200); + + const messages = await getCapturedSlackMessages(); + expect(messages.length).toBe(1); + expect(messages[0].text).toContain(instrumentDisplayName); + expect(messages[0].text).toContain(runId); + expect(messages[0].text).toContain("source: lambda"); + expect(messages[0].text).toContain( + `/instruments/${instrumentId}/runs/${runId}` + ); + expect(messages[0].text).toContain("View in Data Hub"); + }); +});