feat: add DPS provenance to generated STAC items

README.md

@@ -21,6 +21,32 @@ User STAC catalog configuration:
 - `USER_STAC_CATALOG_TRANSACTIONS_AUTH_MODE=basic` enables catalog write routes and selects the supported auth mode. Catalog write routes require catalogs to stay enabled.
 - `USER_STAC_CATALOG_TRANSACTIONS_AUTH_SECRET_ARN` can point at an existing auth secret.
 
+## DPS-generated STAC items
+
+The DPS item generator assigns unregistered items to collections named
+`{username}__{algorithm_name}__{algorithm_version}`. Authorized user-supplied
+collection IDs are preserved. Generated items include the filterable
+`maap-dps:algorithm_name`, `processing:version`, `maap-dps:username`,
+and `maap-dps:tag` properties, the MAAP DPS STAC extension, and a `dps-metadata`
+asset containing the source `.met.json` file. The generator also overwrites the
+STAC Common Metadata `created` property with the UTC publication time shared by
+all Items generated from that catalog.
+
+To merge legacy tag-specific DPS collections into these tag-free IDs, preview
+then apply the database migration:
+
+```bash
+./scripts/migrate_dps_collection_ids.py --dry-run
+./scripts/migrate_dps_collection_ids.py --apply
+```
+
+It recognizes four-part IDs (`username__algorithm__version__tag`), merges their
+items into the corresponding three-part ID, and adds the DPS metadata fields
+from the legacy ID. Collections containing an item-ID collision after merging
+are reported and left unchanged. For a deployed database, follow the
+[RDS connection guide](#connect-to-rds-through-an-ssm-tunnel) below and the
+RDS usage instructions in the migration script's docstring.
+
 Collection-only STAC transactions can still be enabled with:
 
 - `USER_STAC_COLLECTION_TRANSACTIONS_AUTH_MODE=basic`
@@ -92,6 +118,130 @@ This has three consequences :
 2. In addition, because these APIs _also_ sometimes need access to the internet, a NAT gateway must in addition be deployed in that VPC.
 3. For direct, administrative connections to the database, one _must_ go through an instance placed in the same VPC as the database.
 
+### Connect to RDS through an SSM tunnel
+
+For administrative database access, use the existing PgBouncer EC2 instance
+as an SSM network relay and run your database client locally. Forward to the
+**RDS endpoint**, not the PgBouncer service, to bypass connection pooling. RDS stays
+private, and you do not need to install dependencies on the EC2 instance or
+open inbound ports.
+
+You need the AWS CLI, `jq`, `curl`, and a PostgreSQL client such as `psql`
+on your workstation. Also
+[install the Session Manager plugin](https://docs.aws.amazon.com/systems-manager/latest/userguide/session-manager-working-with-install-plugin.html);
+it is a separate installation from the AWS CLI.
+Configure your AWS profile and region first. Your AWS
+identity needs permission to read the stack resources, SSM parameter, and
+database secret (including KMS decryption if applicable), and start sessions
+using `AWS-StartPortForwardingSessionToRemoteHost`. The EC2 instance must be
+SSM-managed with SSM Agent 3.1.1374.0 or later.
+
+#### Find the database secret
+
+Choose the deployment you want to connect to:
+
+| Database | Stack name | SSM parameter type |
+| --- | --- | --- |
+| User STAC (including DPS outputs) | `MAAP-STAC-<stage>-userSTAC` | `internal` |
+| Public STAC | `MAAP-STAC-<stage>-pgSTAC` | `public` |
+
+The examples use userSTAC. Confirm your account and stage, then list the
+secrets belonging to that CDK deployment:
+
+```bash
+aws sts get-caller-identity
+STAGE=test  # change as appropriate
+STACK="MAAP-STAC-${STAGE}-userSTAC"  # userSTAC or pgSTAC
+
+aws cloudformation list-stack-resources \
+  --stack-name "$STACK" \
+  --query 'StackResourceSummaries[?ResourceType==`AWS::SecretsManager::Secret`].[LogicalResourceId,PhysicalResourceId]' \
+  --output table
+```
+
+You can also find these under **CloudFormation → stack → Resources**. 
+
+Select the database secret whose ID contains `pgstacdbbootstrappersecret`, not the
+STAC HTTP basic-auth secret. CloudFormation gives you the secret's identifier; retrieve its value
+from Secrets Manager. In the same terminal:
+
+```bash
+SECRET_ID='<database secret physical ID (not arn) from the table>'
+DB_SECRET=$(aws secretsmanager get-secret-value \
+  --secret-id "$SECRET_ID" --query SecretString --output text)
+
+export PGHOST=$(jq -er '.host' <<< "$DB_SECRET")
+export PGDATABASE=$(jq -er '.dbname' <<< "$DB_SECRET")
+export PGUSER=$(jq -er '.username' <<< "$DB_SECRET")
+export PGPASSWORD=$(jq -er '.password' <<< "$DB_SECRET")
+unset DB_SECRET
+```
+
+Check that these commands succeed and that `PGHOST` matches the selected RDS
+endpoint. Do not print the secret or run these commands with shell tracing
+(`set -x`) enabled.
+
+#### Start the tunnel
+
+In a second terminal with the same AWS profile and region, retrieve the RDS
+endpoint from the same secret and start the session. Variables set in the first
+terminal are not available in this terminal:
+
+```bash
+STAGE=test  # use the same stage as above
+TYPE=internal  # use public for the pgSTAC stack
+SECRET_ID='<same database secret physical ID from the table>'
+RDS_HOST=$(aws secretsmanager get-secret-value \
+  --secret-id "$SECRET_ID" --query SecretString --output text | jq -er '.host')
+INSTANCE_ID=$(aws ssm get-parameter \
+  --name "/maap-eoapi/$STAGE/$TYPE/pgbouncer-instance-id" \
+  --query Parameter.Value --output text)
+
+aws ssm start-session \
+  --target "$INSTANCE_ID" \
+  --document-name AWS-StartPortForwardingSessionToRemoteHost \
+  --parameters "{\"host\":[\"$RDS_HOST\"],\"portNumber\":[\"5432\"],\"localPortNumber\":[\"15432\"]}"
+```
+
+Leave this terminal open while you use the database. The EC2 host needs
+network access to RDS on port 5432, as it does for normal PgBouncer traffic.
+
+#### Connect with a local client
+
+Back in the first terminal, download the
+[AWS RDS CA bundle](https://docs.aws.amazon.com/AmazonRDS/latest/UserGuide/UsingWithRDS.SSL.html)
+and configure TLS. `PGHOSTADDR` sends the connection through localhost while
+`PGHOST` retains the RDS hostname for certificate verification:
+
+```bash
+curl --fail --show-error --silent \
+  https://truststore.pki.rds.amazonaws.com/global/global-bundle.pem \
+  --output /tmp/maap-rds-global-bundle.pem
+
+export PGHOSTADDR=127.0.0.1
+export PGPORT=15432
+export PGSSLMODE=verify-full
+export PGSSLROOTCERT=/tmp/maap-rds-global-bundle.pem
+
+psql -c 'SELECT current_database(), current_user;'
+psql
+```
+
+`psql` and other libpq-based clients, including psycopg, can use these `PG*`
+environment variables. A client's explicit connection string can override
+them; check the tool's connection options before running commands.
+
+Use `\q` to leave `psql`. When finished, clear the connection variables and
+close the SSM session in the second terminal:
+
+```bash
+unset PGPASSWORD PGHOST PGHOSTADDR PGPORT PGDATABASE PGUSER PGSSLMODE PGSSLROOTCERT
+```
+
+Before destructive operations, confirm the target database and ensure you have
+a recoverable backup. For work expected to run for hours, prefer a durable
+in-VPC execution environment over a workstation tunnel.
+
 ## Ingestion
 
 The term "ingestion" refers to the process of cataloging data in the STAC catalog associated with this deployment.

cdk/constructs/DpsStacItemGenerator/runtime/Dockerfile

@@ -10,7 +10,9 @@ WORKDIR /asset
 COPY runtime/pyproject.toml pyproject.toml
 COPY runtime/src/dps_stac_item_generator/ dps_stac_item_generator/
 
-RUN uv export --no-dev --no-editable -o requirements.txt && \
+RUN dnf install -y git && \
+  dnf clean all && \
+  uv export --no-dev --no-editable -o requirements.txt && \
   uv pip install --target /asset -r requirements.txt
 
 CMD ["dps_stac_item_generator.handler.handler"]

cdk/constructs/DpsStacItemGenerator/runtime/pyproject.toml

@@ -9,9 +9,10 @@ requires-python = ">=3.12"
 dependencies = [
     "obstore>=0.7.0",
     "pydantic>=2.11.0",
-    "pystac[validation]>=1.13.0",
     "stac-pydantic>=3.2.0",
     "python-slugify==8.0.4",
+    "pystac-ext-maap-dps",
+    "pystac[validation]>=1.15.2",
 ]
 
 [dependency-groups]
@@ -29,3 +30,6 @@ build-backend = "hatchling.build"
 [tool.pytest.ini_options]
 addopts = "-vv --ignore=cdk.out --no-header --tb=native"
 pythonpath = "."
+
+[tool.uv.sources]
+pystac-ext-maap-dps = { git = "https://github.com/MAAP-Project/maap-dps-stac-extension.git", rev = "v0.1.0" }

cdk/constructs/DpsStacItemGenerator/runtime/src/dps_stac_item_generator/item.py

@@ -3,21 +3,23 @@
 import logging
 import re
 from collections.abc import Generator
+from datetime import datetime, timezone
 from typing import Any
 from urllib.parse import urlparse
 
 import obstore
 import pystac
 from obstore.store import from_url
-from pystac import Link
+from pystac import Asset, Link
+from pystac.extensions.maap_dps import MaapDpsExtension
 from pystac.stac_io import DefaultStacIO, StacIO
 from slugify import slugify
 from stac_pydantic.item import Item
 
 logger = logging.getLogger()
 logger.setLevel(logging.INFO)
 
-COLLECTION_ID_FORMAT = "{username}__{algorithm_name}__{algorithm_version}__{tag}"
+COLLECTION_ID_FORMAT = "{username}__{algorithm_name}__{algorithm_version}"
 
 
 class ObstoreStacIO(DefaultStacIO):
@@ -62,18 +64,24 @@ def get_dps_output_prefix(s3_key) -> str | None:
     return None
 
 
-def load_met_json(bucket: str, job_output_prefix: str) -> dict[str, str] | None:
-    """Load the .met.json file that gets uploaded with DPS job outputs"""
+def load_met_json(
+    bucket: str, job_output_prefix: str
+) -> tuple[dict[str, str], str] | None:
+    """Load DPS metadata and return its discovered object key."""
     store = from_url(f"s3://{bucket}/{job_output_prefix}")
     stream = obstore.list(store, chunk_size=10)
     for list_result in stream:
         for result in list_result:
-            if result["path"].endswith("met.json"):
-                return json.loads(
-                    obstore.get(store, result["path"])
-                    .bytes()
-                    .to_bytes()
-                    .decode("utf-8")
+            met_json_key = result["path"]
+            if met_json_key.endswith("met.json"):
+                return (
+                    json.loads(
+                        obstore.get(store, met_json_key)
+                        .bytes()
+                        .to_bytes()
+                        .decode("utf-8")
+                    ),
+                    met_json_key,
                 )
     return None
 
@@ -129,24 +137,26 @@ def get_stac_items(
 
     s3_key_parsed = urlparse(catalog_json_key)
 
-    job_metadata = load_met_json(s3_key_parsed.netloc, job_output_prefix)
-    if not job_metadata:
+    met_json = load_met_json(s3_key_parsed.netloc, job_output_prefix)
+    if not met_json:
         raise ValueError(
             "could not locate the .met.json file "
             f"with the DPS job outputs in {job_output_prefix}"
         )
 
+    job_metadata, met_json_key = met_json
     deterministic_collection_id = slugify(
         COLLECTION_ID_FORMAT.format(**job_metadata), regex_pattern=r"[/\?#%& ]+"
     )
     username = job_metadata.get("username", "")
+    met_json_href = f"s3://{s3_key_parsed.netloc}/{met_json_key.lstrip('/')}"
+    processing_time = datetime.now(timezone.utc).isoformat().replace("+00:00", "Z")
 
     catalog = pystac.Catalog.from_file(catalog_json_key)
     catalog.make_all_asset_hrefs_absolute()
 
     for item in catalog.get_all_items():
-        item_dict = item.to_dict()
-        item_collection_id = item_dict.get("collection")
+        item_collection_id = item.collection_id
 
         if item_collection_id and is_authorized(username, item_collection_id, registry):
             logger.info(
@@ -155,6 +165,29 @@ def get_stac_items(
                 username,
             )
         else:
-            item_dict["collection"] = deterministic_collection_id
+            item.collection_id = deterministic_collection_id
+
+        item.stac_extensions[:] = list(dict.fromkeys(item.stac_extensions))
+        MaapDpsExtension.ext(item, add_if_missing=True).apply(
+            algorithm_name=job_metadata["algorithm_name"],
+            processing_version=job_metadata["algorithm_version"],
+            username=job_metadata["username"],
+            tag=job_metadata["tag"],
+        )
+        item.properties["created"] = processing_time
+        item.add_asset(
+            "dps-metadata",
+            Asset(
+                href=met_json_href,
+                media_type="application/json",
+                roles=["metadata"],
+                title="DPS job metadata",
+            ),
+        )
+        item.links = [
+            link
+            for link in item.links
+            if not (link.rel == "via" and link.href == met_json_href)
+        ]
 
-        yield Item(**item_dict)
+        yield Item(**item.to_dict())