feat: Catalia report analysis

CLAUDE.md

@@ -0,0 +1,85 @@
+# CLAUDE.md
+
+This file provides guidance to Claude Code (claude.ai/code) when working with code in this repository.
+
+## Project Overview
+
+Unity Data Services ("Catalia UDS") is a NASA Unity/Cumulus-adjacent Python service. It provides AWS Lambda functions and a FastAPI web service for data ingest, cataloging, search, and access, complying with OGC DAPA and STAC specifications, plus a subsystem ("Catalia") for archiving granules out to external DAACs (Distributed Active Archive Centers) via CNM (Cloud Notification Message).
+
+Python version is pinned to **3.10** (see `setup.py`).
+
+## Common Commands
+
+```bash
+# Install dependencies
+python3 -m pip install -r requirements.txt
+
+# Run all tests (there is no pytest.ini/tox.ini/setup.cfg — tests are plain
+# unittest.TestCase classes discovered under tests/, mirroring the
+# cumulus_lambda_functions/ package structure 1:1)
+python -m pytest tests/cumulus_lambda_functions
+
+# Run a single test file / test case / test method
+python -m pytest tests/cumulus_lambda_functions/lib/test_cql_parser.py
+python -m pytest tests/cumulus_lambda_functions/lib/test_cql_parser.py::TestCqlParser::test_01
+
+# Run the FastAPI web service locally (loads .env via python-dotenv)
+python -m cumulus_lambda_functions.catalya_uds_api.web_service
+# or, with auto-reload:
+uvicorn cumulus_lambda_functions.catalya_uds_api.web_service:app --port 8005 --reload
+
+# Run the stage-in/stage-out Docker CLI entrypoint locally
+python -m cumulus_lambda_functions.docker_entrypoint <SEARCH|DOWNLOAD|UPLOAD|CATALOG|CATALYA_COLLECTION_ARCHIVE>
+```
+
+There is no lint command configured in CI. CI (`.github/workflows/makefile.yml`) does **not** run tests — it only installs deps, strips `boto3`/`botocore`/`s3transfer` (provided by the Lambda runtime), and packages Lambda deployment zips via `ci.cd/create_aws_lambda_zip.sh`. Testing is manual/local via pytest.
+
+`tests/integration_tests/` (top-level, separate from `tests/cumulus_lambda_functions/`) contains live end-to-end tests that hit a real deployed stack and require their own `.env` (see `.env.tpl` there); these are not run in CI.
+
+## Architecture
+
+### Layering: `mdps_ds_lib` vs `cumulus_lambda_functions`
+
+This repo is the **application layer**. Nearly all low-level AWS/data plumbing (boto3 S3/SNS/SQS/DynamoDB/Lambda/Parameter Store clients, stage-in/out granule search/download/upload/catalog logic, shared `Constants`, JSON/time/file utils) lives in the external pip dependency **`mdps_ds_lib`** (`mdps-ds-lib` on PyPI). Code in `cumulus_lambda_functions/` builds on top of it — e.g. `daac_archiver/ddb_mws/*.py` wraps `mdps_ds_lib.lib.aws.no_sql_ddb.NoSqlDdb` with table-specific access patterns; `docker_entrypoint/__main__.py` just dispatches to `mdps_ds_lib.stage_in_out.*Factory` classes. When behavior seems missing from this repo, check whether it's implemented in `mdps_ds_lib` instead.
+
+### Module map (`cumulus_lambda_functions/`)
+
+- **`catalya_uds_api/`** — FastAPI app (`web_service.py`) exposing the Catalia REST API, wrapped with `Mangum` as the Lambda handler. Routers: `auth_admin_api.py` (user-group ↔ source/target collection authorization admin CRUD), `granules_archive_api.py` (trigger/track granule archiving to a DAAC), `daac_archive_config_api.py` (DAAC archive config CRUD). `granules_archive_api.py`'s archive route asynchronously re-invokes another Lambda (via `mdps_ds_lib`'s `AwsLambda`) to dodge API Gateway timeouts, unless `IS_API_IN_DOCKER=TRUE`, in which case it runs inline.
+- **`catalya_archive_trigger/`** — Lambda (`lambda_function.py`) that validates HYSDS metadata, resolves relative S3 URLs, and kicks off DAAC archiving.
+- **`daac_archiver/`** — Core DAAC-archiving business logic:
+  - `daac_archiver_catalia_2.py` — main archiver.
+  - `daac_receiver.py` — SQS/SNS CNM status-callback handler, validates against PODAAC's `cumulus_sns_schema.json`.
+  - `cnm_plugins/` — pluggable post-processing on CNM status via a factory (status updates, storage, etc.).
+  - `raw_cnm_storage/` — S3-backed storage of raw CNM messages.
+  - `services/` — MAAP API client, SFA client middleware, staging service, status update service.
+  - `ddb_mws/` — DynamoDB access classes (auth, status, DAAC handshake config, archiving traces). `catalia_auth_db.py` resolves user-group → source/target collection auth using regex + longest-common-prefix matching.
+  - `sql_mws/` — a **parallel SQLAlchemy/Postgres re-implementation** of the status table (`catalia_status_db.py`, SQLAlchemy Core, not ORM) with the same public API as its DDB counterpart, intended as a drop-in swap for analytics/ad-hoc queries that are hard to do against DynamoDB. Connects to an Aurora Postgres instance (see `tf-module/daac_delivery_analysis`) via AWS Parameter Store JSON (`URL/PORT/USERNAME/PASSWORD/DBNAME`).
+- (Deprecated, DO NOT USE) **`docker_entrypoint/`** — `__main__.py`, the CLI for the standalone stage-in/stage-out Docker image; dispatches on `argv[1]` to `mdps_ds_lib` factories.
+- **`keycloak_authorizer/`** — a placeholder API Gateway TOKEN authorizer that currently allows all requests (real Keycloak integration is pending).
+- **`lib/`** — shared internal helpers: `uds_fast_api/` (CORS config, API Gateway/Mangum auth-header extraction, STAC browser static assets), `uds_db/` (DynamoDB-backed collection/archive-index models), `authorization/` (pluggable authorizer abstraction + factory, ES-identity-pool impl), `metadata_extraction/` (ECHO metadata parsing), `lambda_logger_generator.py` (standardized Lambda logger setup, used by every handler to strip default log handlers).
+- **`mock_daac/`** — a Lambda simulating a DAAC's CNM response, for local/integration testing of the archiving pipeline.
+
+There is no dedicated `models/` directory — DynamoDB/SQL table schemas are defined inline within each `*_db.py` file in `ddb_mws/`, `sql_mws/`, and `uds_db/`.
+
+### Entry point patterns
+
+- **FastAPI + Mangum**: `catalya_uds_api/web_service.py` exposes `handler = Mangum(app=app)` for Lambda, and runs via `uvicorn` locally under `if __name__ == '__main__'`.
+- **Plain Lambda handlers**: `def lambda_handler(event, context)` (or `lambda_handler_response` in `daac_archiver`) in each `lambda_function.py`, generally triggered by SQS/SNS wrapping S3 event notifications. Each strips default log handlers via `LambdaLoggerGenerator.remove_default_handlers()` before delegating to a logic class.
+- (Deprecated, DO NOT USE) **Docker CLI**: `docker_entrypoint/__main__.py`, invoked as `python -m cumulus_lambda_functions.docker_entrypoint <VERB>`, backing the images built from `docker/Dockerfile_download_granules.public` / `.jpl` and the sample compose files under `docker/stage-in-stage-out/`.
+
+### Config / environment
+
+`python-dotenv`'s `load_dotenv()` is called once at the top of `catalya_uds_api/web_service.py` before other imports; after that, config is read via `os.environ`/`os.getenv`, often referencing shared key names from `mdps_ds_lib.lib.constants.Constants` or the locally defined `WebServiceConstants`. Lambda handlers rely on the Lambda runtime's own env vars (no `load_dotenv()` call there). `.env` at repo root is a real local-dev config file (git-tracked historically; do not put new secrets in it) — see it for the expected variable names (`ES_URL`, `SNS_TOPIC_ARN`, `DAPA_API_URL_BASE`, etc.).
+
+### Infra (`docker/`, `tf-module/`)
+
+- `docker/` — `Dockerfile.public`/`.jpl` (main service image, public vs JPL-internal registry), `Dockerfile_download_granules.public`/`.jpl` (stage-in/out image), `docker-compose-web-service*.yml` / `docker-compose-dapa.yml` for running the FastAPI service locally, `docker/stage-in-stage-out/` sample compose files per CLI verb (`dc-001-search`, `dc-002-download`, `dc-003-upload[_auxiliary]`, `dc-004-catalog`).
+- `tf-module/` — deployment order per `Deploying-Catalia-UDS.md`: `unity_vpc` → `uds_catalia_iam` → (optional) `uds_catalia_bucket` → (optional) `daac_delivery_analysis` (Aurora Postgres v2) → `uds_catalia` (Lambda + API Gateway + SNS/SQS, the main Catalia deployment). Other modules: `unity-cumulus` (broader Cumulus lambda infra), `marketplace`, `ds_img_to_ecr[_back]`, `stac_browser`, `mock_daac`, `sqs--sns-lambda-connector`.
+
+### Onboarding a new DAAC partner
+
+See `handshake-daac.md`: grant auth via the admin API, collect the DAAC's SNS topic ARN/role ARN/data version/API key, configure via the archive-config API, update the target S3 bucket policy for the DAAC's IAM role, and update `DAAC_LAMBDA_2_SNS_ROLE` in terraform so SNS accepts inbound CNM messages from that DAAC.
+
+### Other reference docs
+
+- `Deploying-Catalia-UDS.md` — full deployment runbook tying the `tf-module/*` modules together in order, including how to fetch the released Lambda zip artifact.

Deploying-Catalia-UDS.md

@@ -0,0 +1,52 @@
+## Steps to run terraform modules to deploy Catalia UDS Resources
+
+### Preparing AWS credentials
+1. This terraform is setup to run in Science Cloud http://aws.sciencecloud.nasa.gov/ There may be some changes needed for other cloud. 
+2. Retrieve Access Keys for `Project-Power-User`, and set it as `saml-pub-smce`
+
+```
+~/.aws/credentials
+[saml-pub-smce]
+aws_access_key_id = xxx
+aws_secret_access_key = xxx
+aws_session_token = xxx
+
+~/.aws/config
+[profile saml-pub-smce]
+output = json
+region = us-west-2
+sso_session = saml-pub-smce
+sso_account_id = <account-id>
+sso_role_name = Project-Power-User
+```
+3. Get elevated credential via /tf-module/uds_catalia/smce_keys_assume_deployment.sh. It is looking for `saml-pub-smce` profile, and output is in `mdps-temp-creds-assumed`. 
+4. Check ~/.aws/credentials and ~/.aws/config files and verify `mdps-temp-creds-assumed` is created and values are valid. 
+5. export the above profile as default. 
+
+### Setting up VPC. 
+1. VPC should be setup by SA or a project admin. 
+2. If missing, there is a unity VPC terraform which can be used to setup a VPC. `/tf-module/unity_vpc`
+
+### Setting up IAM
+1. Using VPC above, `/tf-module/uds_catalia_iam` sets up most of IAM roles needed for lambdas, and others. 
+2. *NOTE*: In Variables, there is a `prefix`. The `prefix` value must match between this module and the following modules such has `/tf-module/uds_catalia`
+### Setting up Bucket (Optional)
+1. This is a legacy bucket where initial workflow requires data to be staged in UDS bucket. 
+2. If needed, `/tf-module/uds_catalia_bucket` can be used to setup a bucket. 
+
+### Deploying Aurora V2 (Optional)
+1. This should be replaced with Postgres DB from MAAP. 
+2. IF needed, `/tf-module/daac_delivery_analysis` can be used to deploy a minimal Aurora DB. 
+3. It will create a parameter store with the following JSON `{\"DBNAME\":\"xxx\",\"PASSWORD\":\"xxx\",\"PORT\":xxx,\"URL\":\"xxx\",\"USERNAME\":\"xxx\"}`
+
+### Manually creating Postgres Connection String in Parameter Store
+1. If Aurora V2 is not created, there needs to be a parameter store with the following JSON `{\"DBNAME\":\"xxx\",\"PASSWORD\":\"xxx\",\"PORT\":xxx,\"URL\":\"xxx\",\"USERNAME\":\"xxx\"}`
+``
+### Setting up Catalia DAAC Delivery
+1. Deploy main resources `/tf-module/uds_catalia` which includes Lambda, API Gateway, SNS/SQS pipeline. 
+2. Lambda zip file needs to be downloaded from https://github.com/unity-sds/unity-data-services/releases where each release has a `cumulus_lambda_functions-<version>-*.zip` file. It is re-used for all lambda functions.
+3. The zip file must be renamed and placed at the `/tf-module/uds_catalia/build/cumulus_lambda_functions_deployment.zip`
+3. `CATALYA_RDS_CREDS_PARAM_PATH` is retrieved from `Deploying Aurora V2 (Optional)` or `Manually creating Postgres Connection String in Parameter Store`
+4. `DAAC_LAMBDA_2_SNS_ROLE` needs to be DAAC role. If unknown, put dummy value at this moment, and update it when needed. 
+
+

TODO.md

@@ -0,0 +1,9 @@
+### Future tasks
+
+
+1. Update Terraform policies to accept multiple DAACs
+2. Create Terraform or script to update S3 bucket policy for DAACs to accept files
+3. Add additional endpoint to return all statuses of a granule by asking for granule + collection similar to current endpoint asking for operation-id
+4. Consider the possibility of 1 source collection being sent to different DAACs and DAAC having the same collection ID on their side. This will currently fail.
+5. Verify if 1 CNM message can accept 1 file only or a complete set of data, metadata, qc files (3 files)
+6. Use SNS Batch Send to alleviate pressure on SNS calls, but this will complicate database entries for audit, and complicate failure handling. 

cumulus_lambda_functions/catalya_uds_api/auth_admin_api.py

@@ -1,10 +1,13 @@
+import json
 from typing import Union
 
 from cumulus_lambda_functions.daac_archiver.ddb_mws.catalia_auth_db import CataliaAuthDb
+from cumulus_lambda_functions.daac_archiver.sql_mws.catalia_status_db import CataliaStatusDb
 from cumulus_lambda_functions.lib.lambda_logger_generator import LambdaLoggerGenerator
 from cumulus_lambda_functions.lib.uds_fast_api.fast_api_utils import FastApiUtils
 from cumulus_lambda_functions.lib.uds_fast_api.web_service_constants import WebServiceConstants
 from fastapi import APIRouter, HTTPException, Request
+from mdps_ds_lib.lib.aws.aws_param_store import AwsParamStore
 
 LOGGER = LambdaLoggerGenerator.get_logger(__name__, LambdaLoggerGenerator.get_level_from_env())
 
@@ -248,3 +251,35 @@ async def list_auth_mappings(request: Request, tenant: Union[str, None]=None, ve
     if query_result['statusCode'] == 200:
         return query_result['body']
     raise HTTPException(status_code=query_result['statusCode'], detail=query_result['body'])
+
+@router.post("/status-table")
+@router.post("/status-table/")
+async def build_status_table(request: Request):
+    """
+    Ensures the status db table (CATALYA_STATUS_DB) exists in the RDS Postgres
+    database, creating it -- along with its indexes/unique constraint -- via
+    CataliaStatusDb.create_table_if_missing() if it doesn't. Meant to be called once
+    on startup/deployment so the archiving Lambdas never hit a missing-table error
+    at runtime.
+    """
+    LOGGER.debug('started build_status_table')
+    auth_info = FastApiUtils.get_authorization_info(request)
+    auth_crud = AuthCrud(auth_info, {})
+    is_admin_result = auth_crud.is_admin()
+    if is_admin_result['statusCode'] != 200:
+        raise HTTPException(status_code=is_admin_result['statusCode'], detail=is_admin_result['body'])
+
+    required_env = ['CATALYA_RDS_CREDS', 'CATALYA_STATUS_DB']
+    if not all([k in os.environ for k in required_env]):
+        LOGGER.error(f'one or more missing env: {required_env}')
+        raise HTTPException(status_code=500, detail=f'one or more missing env: {required_env}')
+
+    table_name = os.getenv('CATALYA_STATUS_DB')
+    db_config = json.loads(AwsParamStore().get_param(os.getenv('CATALYA_RDS_CREDS')))
+    status_db = CataliaStatusDb(table_name, db_config)
+
+    already_existed = status_db.table_exists()
+    if not already_existed:
+        status_db.create_table_if_missing()
+        LOGGER.info(f'created status db table: {table_name}')
+    return {'table': table_name, 'already_existed': already_existed, 'created': not already_existed}

cumulus_lambda_functions/catalya_uds_api/granules_archive_api.py

@@ -5,7 +5,8 @@
 
 
 from cumulus_lambda_functions.daac_archiver.daac_archiver_catalia_2 import DaacArchiverCatalia
-from cumulus_lambda_functions.daac_archiver.ddb_mws.catalia_status_db import CataliaStatusDb
+from cumulus_lambda_functions.daac_archiver.services.status_update_svc import StatusUpdateSvc
+from cumulus_lambda_functions.daac_archiver.sql_mws.catalia_status_db import CataliaStatusDb
 from cumulus_lambda_functions.lib.lambda_logger_generator import LambdaLoggerGenerator
 from cumulus_lambda_functions.lib.uds_fast_api.internal_ddb_connector import InternalDDBConnector
 from cumulus_lambda_functions.lib.uds_fast_api.web_service_constants import WebServiceConstants
@@ -319,12 +320,63 @@ async def archive_entire_collection_actual(request: Request, collection_id: str)
     return {'message': 'archive initiated'}
 
 
+@router.get("/report")
+@router.get("/report/")
+async def get_archive_report(request: Request, collection: str, target_collection: str,
+                              start_datetime: Optional[str] = None, end_datetime: Optional[str] = None):
+    """
+    Summary status counts (submit/response success & failure, plus the full
+    per-status breakdown) for a given collection + target_collection pair.
+
+    collection & target_collection are mandatory. start_datetime/end_datetime are
+    optional (either, both, or neither may be given), covering all-time / since /
+    until / between reporting windows. Values may be epoch-millis ints or RFC3339
+    strings (same formats CataliaStatusDb.search() already accepts).
+
+    Status names come from StatusUpdateSvc.archival_status_schema's enum instead of
+    being hardcoded here, so this stays in sync if that list changes.
+    """
+    LOGGER.debug(f'started get_archive_report for collection={collection}, target_collection={target_collection}, '
+                 f'start_datetime={start_datetime}, end_datetime={end_datetime}')
+    uds_api_creds = json.loads(AwsParamStore().get_param(os.getenv('CATALYA_RDS_CREDS', 'NA')))
+    status_db = CataliaStatusDb(os.getenv('CATALYA_STATUS_DB'), uds_api_creds)
+
+    status_counts = status_db.count_by_status(collection, target_collection, start_datetime, end_datetime)
+
+    known_statuses = StatusUpdateSvc.archival_status_schema['properties']['status']['enum']
+    full_counts = {status: status_counts.get(status, 0) for status in known_statuses}
+
+    def find_status(stage_keyword: str, result_keyword: str) -> Optional[str]:
+        matches = [s for s in known_statuses if stage_keyword in s and result_keyword in s]
+        return matches[0] if matches else None
+
+    submit_success_status = find_status('submit', 'success')
+    submit_failed_status = find_status('submit', 'failed')
+    response_success_status = find_status('receive', 'success')
+    response_failed_status = find_status('receive', 'failed')
+
+    return {
+        'collection': collection,
+        'target_collection': target_collection,
+        'start_datetime': start_datetime,
+        'end_datetime': end_datetime,
+        'status_counts': full_counts,
+        'summary': {
+            'submit_success': full_counts.get(submit_success_status, 0),
+            'submit_failed': full_counts.get(submit_failed_status, 0),
+            'response_success': full_counts.get(response_success_status, 0),
+            'response_failed': full_counts.get(response_failed_status, 0),
+        },
+    }
+
+
 @router.get("/{operation_id}")
 @router.get("/{operation_id}/")
 async def get_archive_status(request: Request, operation_id: str):
     LOGGER.debug(f'started get_archive_status with operation_id: {operation_id}')
-    status_ddb = CataliaStatusDb(os.getenv('CATALYA_STATUS_DB', None))
-    existing_statuses = status_ddb.get(operation_id)
+    uds_api_creds = json.loads(AwsParamStore().get_param(os.getenv('CATALYA_RDS_CREDS', 'NA')))
+    status_db = CataliaStatusDb(os.getenv('CATALYA_STATUS_DB'), uds_api_creds)
+    existing_statuses = status_db.get(operation_id)
     if len(existing_statuses) < 1:
         raise HTTPException(status_code=404, detail=f'STATUS DB does not have any entry for {operation_id}')
-    return {'status_list': existing_statuses}
+    return {'status_list': existing_statuses}