# PP-1411 — Workstation Roles CDC Lambda

**Epic**

https://theorchard.atlassian.net/browse/PP-1442

**Spike references**:

- [SPIKE PP-1411: Hydrate Workstation Roles for Authorization
  Checks](https://www.notion.so/SPIKE-PP-1411-Hydrate-Workstation-Roles-for-Authorization-Checks-31897177520f8197920ecbc1c4e0ab1f)
- [SPIKE PP-1411: CDC Lambda Design and Deployment](https://www.notion.so/SPIKE-PP-1411-CDC-Lambda-Design-and-Deployment-31e97177520f80348649fee0d98883e2)

---

## Open Questions

1. ~~**Deployment option**~~ — **Resolved**: Option A (PP account, `kafka_event_enabled`). Confirmed by KDH team.
2. ~~**Lambda ENI + cross-account SG rules (Option B)**~~ — **Resolved**: Option B not chosen; question is moot.
3. ~~**Consumer auth for CDC MSK cluster**~~ — **Resolved**: TLS in transit (port `9094`), network-only via SG.
   No SASL/SCRAM or MSK IAM required. Confirmed by KDH team; same pattern as `lambda-abacus/sync-account`.

---

## Tickets

### Ticket Summary

| Ticket     | Title                                                    | Service         | Blocked by                            |
|------------|----------------------------------------------------------|-----------------|---------------------------------------|
| PP-1443 (01)  | Create `lambda-pp-cdc` GitHub repo                       | infra           | —                                        |
| PP-1444 (02a) | `permissions/models/profile.py` — Neo4j query function   | ows-permissions | —                                        |
| PP-1446 (02b) | `permissions/logic/profile.py` — logic wrapper           | ows-permissions | PP-1444                                  |
| PP-1447 (02c) | `POST /lookup/profiles/identity/uuids/` handler + schema | ows-permissions | PP-1446                                  |
| PP-1448 (03)  | CDC Lambda handler implementation                        | lambda-pp-cdc   | PP-1443, PP-1447                         |
| PP-1453 (07)  | Lambda: Report consumer lag metric to Datadog                | lambda-pp-cdc   | PP-1448                                  |
| PP-1454 (08)  | Lambda: Emit per-role attach/detach metrics to Datadog       | lambda-pp-cdc   | PP-1448, PP-1453                         |
| PP-1452 (04a) | KDH: Add PP prefix list to CDC MSK clusters              | terraform-infra | —                                        |
| PP-1449 (04b) | Terraform module `lambda-pp-cdc`                         | terraform-infra | PP-1448, PP-1452                         |
| PP-1445 (05)  | Optimistic locking on `DynamoDbConnector.update_item`    | ows-pdp         | —                                        |
| PP-1450 (06)  | Backfill existing workstation roles into DynamoDB        | ows-pdp         | PP-1448, PP-1449, PP-1445                |
| PP-1485 (09)  | ows-pdp: Lambda Machine Identity Cerbos Policy           | ows-pdp         | —                                        |
| PP-1486 (10)  | lambda-pp-cdc: DELETED_HAS_ACCESS_TO Handler             | lambda-pp-cdc   | PP-1448, PP-1485 (09)                    |

### Dependency Graph

```
01 (GitHub repo)
 └── 03 (Lambda handler)
          ├── 04b (Terraform) ◄── 04a (KDH: CDC MSK prefix list)
          │    └── 06 (Backfill)
          │                      ▲
          ├── 07 (consumer lag metric)
          ├── 08 (per-role metrics) — leaf node; blocked by 03 and 07
          └── 10 (soft delete handler) ◄── 09 (machine identity)

02a (ows-permissions: model)     │
 └── 02b (logic wrapper)         │
      └── 02c (handler + schema) │
           └── 03 (Lambda) ──────┘

04a (KDH: CDC MSK prefix list) — independent; must ship before 04b
05 (optimistic locking) — independent; must ship before 06
07 (consumer lag metric) — leaf node; ships after 03
09 (machine identity) — independent; must ship before 10
```

- Tickets **01**, **02a**, **04a**, **05**, and **09** are independent starting points.
- Ticket **04b** is blocked by Ticket 03 and Ticket 04a (KDH prefix list).
- Ticket **06** (backfill) is the final integration step — requires 03, 04b, and 05.
- Ticket **07** (consumer lag) is a leaf — blocked only by 03, no downstream dependencies.
- Ticket **08** (per-role metrics) is a leaf — blocked by 03 and 07, no downstream dependencies.
- Ticket **09** (machine identity) is independent; must ship before 10.
- Ticket **10** (soft delete handler) is blocked by 03 and 09.

---

### Ticket PP-1443 (01) — Infra: Create `lambda-pp-cdc` GitHub Repo

**Service**: GitHub / infra | **Blocked by**: — | **Blocks**: PP-1448

#### Summary

Create a new GitHub repository `lambda-pp-cdc` to house the CDC Lambda code.
Bootstrap the repo using [cookiecutter-python-lambda](https://github.com/theorchard/cookiecutter-python-lambda) to get
the standard Dockerfile, CI pipeline, and Python project structure.

#### Acceptance Criteria

- Repository exists at `github.com/theorchard/lambda-pp-cdc`.
- Repo scaffolded from `cookiecutter-python-lambda` (Dockerfile, CI pipeline, project structure in place).
- CI pipeline builds and pushes a container image to ECR on merge to main.
- Local development setup documented in README (virtualenv, test runner).

---

### Ticket PP-1444 (02a) — ows-permissions: Model Layer — Neo4j Query for Profile Identity Lookup

**Service**: ows-permissions | **Blocked by**: — | **Blocks**: PP-1446

#### Summary

Add a Neo4j query function to the permissions model layer that resolves a batch of `profile_uuid`s
to their `identity_uuid` and `vendor_uuid`. This is a pure data-layer function with no HTTP concerns.

#### Acceptance Criteria

- `get_identity_uuids_for_profile_uuids(profile_uuids, session)` returns a list of result dicts with `profile_uuid`,
  `identity_uuid`, `vendor_uuid`, and `roles` (the raw Neo4j `LabelProfile.roles` array).
- Profile UUIDs with no match are omitted from the result.
- Unit tests cover: matched UUID, unmatched UUID omitted, empty input, profile with empty roles array.

#### Implementation Details

**`permissions/models/profile.py`** — new Neo4j query function

```python
def get_identity_uuids_for_profile_uuids(profile_uuids: list[str], session) -> list[dict]:
# MATCH (i:Identity)-[:HAS_PROFILE]-(p:Profile)-[:HAS_ACCESS_TO]-(v:Vendor)
# WHERE p.uuid IN $profile_uuids
# RETURN i.id AS identity_uuid, p.uuid AS profile_uuid, v.uuid AS vendor_uuid, p.roles AS roles
# Returns raw Neo4j results — ordering is handled by the logic layer
```

---

### Ticket PP-1446 (02b) — ows-permissions: Logic Layer — Profile Identity Lookup Wrapper

**Service**: ows-permissions | **Blocked by**: PP-1444 | **Blocks**: PP-1447

#### Summary

Add a logic-layer wrapper in the permissions logic module that calls the model function from
Ticket 02a, applies the dataloader ordering pattern, and returns a response dict. All business
logic lives here — the handler (Ticket 02c) does nothing but call this function.

#### Acceptance Criteria

- `lookup_identity_uuids_by_profile_uuids(uuids)` returns `{"profiles": [...]}` with results in the same order as the
  input `uuids` list (dataloader pattern).
- Each entry includes `profile_uuid`, `identity_uuid`, `vendor_uuid`, and `roles`.
- A UUID present in the input but absent from Neo4j is omitted from `profiles`.
- Unit tests cover: all UUIDs matched in order, partial match with correct ordering, empty input, profile with empty roles array.

#### Implementation Details

**`permissions/logic/profile.py`** — logic wrapper

```python
def lookup_identity_uuids_by_profile_uuids(uuids: list[str]) -> dict:
    rows = models.profile.get_identity_uuids_for_profile_uuids(uuids, session)
    # Build a dict keyed by profile_uuid for O(1) lookup
    by_uuid = {r["profile_uuid"]: r for r in rows}
    # Iterate input order to preserve dataloader ordering; omit unmatched UUIDs
    profiles = [by_uuid[u] for u in uuids if u in by_uuid]
    return {"profiles": profiles}
```

---

### Ticket PP-1447 (02c) — ows-permissions: Handler + Schema — `POST /lookup/profiles/identity/uuids/`

**Service**: ows-permissions | **Blocked by**: PP-1446 | **Blocks**: PP-1448

#### Summary

Add the marshmallow request schema and route handler for the new PIP endpoint. The Lambda calls
this endpoint to resolve `profile_uuid` → `identity_uuid` + `vendor_uuid` before writing roles to
DynamoDB. The handler delegates entirely to the logic layer; all validation and ordering happen in
Tickets 02a–02b.

#### Acceptance Criteria

- `POST /lookup/profiles/identity/uuids/` exists and is reachable.
- Request body is validated via marshmallow schema before reaching the handler.
- Response `profiles` list matches input UUID order (dataloader pattern); unmatched UUIDs are omitted.
- Each profile entry includes `profile_uuid`, `identity_uuid`, `vendor_uuid`, and `roles`.
- No access rule checks (PIP endpoint).
- Unit tests cover: valid request, empty list, invalid input, profile with empty roles array.
- Integration tests in `tests/integration/api/test_profile_lookups.py` cover: matched UUIDs returned in order with roles,
  unmatched UUID omitted, empty list request.

#### Implementation Details

**`permissions/validations/schemas/lookup.py`** — request and response schemas

```python
from marshmallow import fields

from permissions.validations import ma


class LookupProfileIdentityUuids(ma.Schema):
    uuids = fields.List(fields.UUID(required=True), required=True)


class ProfileIdentitySchema(ma.Schema):
    profile_uuid = fields.UUID(required=True)
    identity_uuid = fields.UUID(required=True)
    vendor_uuid = fields.UUID(required=True)
    roles = fields.List(fields.Str(), load_default=list)


class LookupProfileIdentityResponse(ma.Schema):
    profiles = fields.List(fields.Nested(ProfileIdentitySchema), required=True)
```

**`permissions/handlers/handlers.py`** — new route

```python
@app.route('/lookup/profiles/identity/uuids/', methods=['POST'])
@validate_request_data(LookupProfileIdentityUuids())
def lookup_profiles_identity_uuids(deserialize_schema):
    """NOTE: No access rule checks — PIP endpoint for Permissions Platform."""
    return flaskify(profile.lookup_identity_uuids_by_profile_uuids(deserialize_schema['uuids']))
```

**Request / response shape**

```json
{
  "uuids": [
    "profile-uuid-1",
    "profile-uuid-2"
  ]
}

{
  "profiles": [
    {
      "profile_uuid": "profile-uuid-1",
      "identity_uuid": "...",
      "vendor_uuid": "...",
      "roles": ["administrator", "catalog"]
    }
  ]
}
```

---

### Ticket PP-1448 (03) — Lambda: CDC Handler Implementation

**Service**: lambda-pp-cdc | **Blocked by**: PP-1443, PP-1447 | **Blocks**: PP-1449

#### Summary

Implement the Lambda handler that consumes `cdc.musicGraphV5.profile` events, filters for
`LabelProfile` changes, diffs workstation roles, and calls `OwsPdpClient.attach_detach_roles_by_identity_tenant`.
Uses the `EventSourceMessage` + `JSONDeserializer` KDH consumer pattern.

#### Acceptance Criteria

- `CREATE` events attach all workstation roles present in after state.
- `UPDATE` events attach/detach only the diff between before and after.
- `DELETE` events detach all workstation roles from before state (handled for completeness; should not occur in
  practice).
- Non-`LabelProfile` events are silently skipped.
- No-op events (workstation-role subset unchanged) are silently skipped.
- Missing `roles` key in CDC payload logs a warning and returns empty set (does not raise).
- No identity found for a `profile_uuid` logs a warning and skips the record (does not raise).
- Unit tests cover all operation types, the no-op skip, the missing-roles warning, and the missing-identity warning.

#### Implementation Details

**`handler.py`** — entry point

```python
def handler(event, context):
    deserializer = JSONDeserializer()
    for _, msk_message in EventSourceMessage(event):
        record = deserializer.deserialize(msk_message.value)
        _process_record(record)


def _process_record(record: dict):
    evt = record["event"]
    state = evt["state"]

    props_for_type = (state["after"] or state["before"])["properties"]
    if extract_value(props_for_type["profileType"]) != "LabelProfile":
        return

    roles_to_attach, roles_to_detach = _diff_roles(state)
    if not roles_to_attach and not roles_to_detach:
        return

    profile_uuid = _extract_profile_uuid(evt["keys"])
    identity_uuid, vendor_uuid = _lookup_identity(profile_uuid)
    if not identity_uuid:
        logger.warning(f"No identity found for profile {profile_uuid}, skipping.")
        return

    _write_roles(identity_uuid, vendor_uuid, roles_to_attach, roles_to_detach)
```

**`handler.py`** — role diff

```python
KNOWN_ROLES = {"administrator", "catalog"}
ROLE_MAP = {"administrator": "workstation_admin", "catalog": "workstation_catalog"}


def _diff_roles(state: dict) -> tuple[set, set]:
    """Return (roles_to_attach, roles_to_detach) as PDP role names."""

    def _extract_roles(half, side: str):
        if not half:
            return set()
        roles_val = half["properties"].get("roles")
        if roles_val is None:
            logger.warning(f"LabelProfile CDC event missing 'roles' key in {side} state")
            return set()
        return set(extract_value(roles_val)) & KNOWN_ROLES

    after_roles = _extract_roles(state["after"], "after")
    before_roles = _extract_roles(state["before"], "before")
    return (
        {ROLE_MAP[r] for r in after_roles - before_roles},
        {ROLE_MAP[r] for r in before_roles - after_roles},
    )
```

**`handler.py`** — CDC typed-value helper

```python
def extract_value(typed_val: dict):
    """Unwrap a Neo4j CDC typed-value envelope.

    Ex: { "type": "S", "S": "LabelProfile" }
    """
    type_key = typed_val["type"]
    return typed_val[type_key]
```

**Role mapping**

| Neo4j role      | DynamoDB / PDP role   |
|-----------------|-----------------------|
| `administrator` | `workstation_admin`   |
| `catalog`       | `workstation_catalog` |

#### Notes

- Confirm with KDH team whether per-record `POST /lookup/profiles/identity/uuids/` calls are acceptable at steady-state
  volume, or whether batching within the invocation is required. See open question 6 in the spike doc.

---

### Ticket PP-1453 (07) — Lambda: Report Consumer Lag Metric to Datadog

**Service**: lambda-pp-cdc | **Blocked by**: PP-1448 | **Blocks**: —

#### Summary

Add consumer lag instrumentation to the CDC Lambda handler. Every invocation computes the elapsed time
between each message's `metadata.txCommitTime` and `datetime.now(UTC)`, then emits the per-invocation
maximum as the Datadog gauge `pp.cdc.workstation_roles.consumer_lag_seconds`. Lag is measured for all
messages — including non-`LabelProfile` records — to reflect true topic lag.

#### Acceptance Criteria

- `_compute_lag_seconds(record)` parses `metadata.txCommitTime` and returns `(now - txCommitTime).total_seconds()`.
- `handler` collects lag samples from all deserialized messages (before the `LabelProfile` filter) and calls `_report_consumer_lag(max(lag_samples))`.
- `_report_consumer_lag` emits `pp.cdc.workstation_roles.consumer_lag_seconds` as a Datadog gauge tagged with `environment` and `service`.
- `_report_consumer_lag` is a no-op when `ENVIRONMENT` is `dev` or `test`.
- Unit tests cover: lag computation from a known timestamp, no-op in dev/test, max value is used across a batch.

#### Implementation Details

**`handler.py`** — additions and modified `handler` entry point

```python
from datetime import datetime, timezone
from dateutil.parser import isoparse
import datadog
import os

METRIC_CONSUMER_LAG = "pp.cdc.workstation_roles.consumer_lag_seconds"


def handler(event, context):
    datadog.initialize(api_key=os.environ["DD_API_KEY"])

    deserializer = JSONDeserializer()
    lag_samples = []

    for _, msk_message in EventSourceMessage(event):
        record = deserializer.deserialize(msk_message.value)
        lag_samples.append(_compute_lag_seconds(record))
        _process_record(record)

    if lag_samples:
        _report_consumer_lag(max(lag_samples))


def _compute_lag_seconds(record: dict) -> float:
    tx_commit_str = extract_value(record["metadata"]["txCommitTime"])  # "2026-03-03T23:03:52.383Z"
    tx_commit_time = isoparse(tx_commit_str)
    return (datetime.now(timezone.utc) - tx_commit_time).total_seconds()


def _report_consumer_lag(lag_seconds: float) -> None:
    env = os.environ.get("ENVIRONMENT", "")
    if env.lower() in ("dev", "test"):
        return
    datadog.api.Metric.send(
        metric=METRIC_CONSUMER_LAG,
        points=lag_seconds,
        type="gauge",
        tags=[
            f"environment:{env}",
            "service:lambda-pp-cdc-workstation-roles",
        ],
    )
```

`DD_API_KEY` is provided via the Lambda environment (wired in PP-1449).

---

### Ticket PP-1454 (08) — Lambda: Emit Per-Role Attach/Detach Metrics to Datadog

**Service**: lambda-pp-cdc | **Blocked by**: PP-1448, PP-1453 | **Blocks**: —

#### Summary

Add per-role count metrics to the CDC Lambda handler to support a workstation-roles throughput
dashboard. Every invocation accumulates `workstation_admin` and `workstation_catalog` attach/detach
counts across all records, then emits them as Datadog `count` metrics tagged with `role`. Dashboard
totals are derived by summing across the `role` tag — no separate total metric is emitted.

#### Acceptance Criteria

- `_process_record` (from PP-1448) is updated to return `tuple[Counter[str], Counter[str]]` —
  `(role_attach_counts, role_detach_counts)`, keyed by PDP role name.
- Returns `(Counter(), Counter())` for skipped records (non-LabelProfile, no-op diff, missing identity).
- `handler` accumulates per-role counters across all records and calls
  `_report_roles_processed(role_attach_counts, role_detach_counts)`.
- `_report_roles_processed` emits `pp.cdc.workstation_roles.roles_attached` and
  `pp.cdc.workstation_roles.roles_detached` as Datadog `count` metrics tagged with `environment`,
  `service`, and `role:<role-name>` for each role present in the counters.
- `_report_roles_processed` is a no-op when `ENVIRONMENT` is `dev` or `test`.
- Unit tests cover: correct per-role counts from `_process_record` for attach-only, detach-only,
  mixed (`workstation_admin` attach + `workstation_catalog` detach), and no-op; no-op in dev/test;
  correct accumulated counters across a multi-record batch.

#### Implementation Details

**`handler.py`** — new metric constants (alongside existing `METRIC_CONSUMER_LAG`):

```python
METRIC_ROLES_ATTACHED = "pp.cdc.workstation_roles.roles_attached"
METRIC_ROLES_DETACHED = "pp.cdc.workstation_roles.roles_detached"
```

**`handler.py`** — modified `handler` entry point (extends PP-1453 version):

```python
from collections import Counter

def handler(event, context):
    datadog.initialize(api_key=os.environ["DD_API_KEY"])

    deserializer = JSONDeserializer()
    lag_samples = []
    role_attach_counts: Counter[str] = Counter()
    role_detach_counts: Counter[str] = Counter()

    for _, msk_message in EventSourceMessage(event):
        record = deserializer.deserialize(msk_message.value)
        lag_samples.append(_compute_lag_seconds(record))
        attached, detached = _process_record(record)
        role_attach_counts += attached
        role_detach_counts += detached

    if lag_samples:
        _report_consumer_lag(max(lag_samples))
        _report_roles_processed(role_attach_counts, role_detach_counts)
```

**`handler.py`** — new `_report_roles_processed`:

```python
def _report_roles_processed(
    attached: Counter[str], detached: Counter[str]
) -> None:
    env = os.environ.get("ENVIRONMENT", "")
    if env.lower() in ("dev", "test"):
        return
    base_tags = [f"environment:{env}", "service:lambda-pp-cdc-workstation-roles"]
    for role, count in attached.items():
        datadog.api.Metric.send(
            metric=METRIC_ROLES_ATTACHED, points=count, type="count",
            tags=base_tags + [f"role:{role}"],
        )
    for role, count in detached.items():
        datadog.api.Metric.send(
            metric=METRIC_ROLES_DETACHED, points=count, type="count",
            tags=base_tags + [f"role:{role}"],
        )
```

**Dashboard note**: totals across `workstation_admin` + `workstation_catalog` are computed in
Datadog by summing `pp.cdc.workstation_roles.roles_attached{*}` (or `roles_detached`) without
filtering on `role`.

#### Notes

- `_process_record` return-type change (PP-1448 → `tuple[Counter[str], Counter[str]]`) is a
  prerequisite; coordinate with the PP-1448 implementer.
- PP-1453 is listed as a blocker to avoid merge conflicts on `handler.py`; both tickets modify
  the same function.

---

### Ticket PP-1452 (04a) — terraform-infra (KDH): Add PP Prefix List to CDC MSK Clusters

**Service**: terraform-infra (KDH-owned) | **Blocked by**: — | **Blocks**: PP-1449

#### Summary

Add the PP account's private subnet prefix list to `kafka_allowed_custom_prefix_list_names` on the
QA and prod CDC MSK clusters. This allows the `lambda-pp-cdc` Lambda (running in the PP account)
to reach the CDC brokers over TLS — the same pattern used by `lambda-abacus/sync-account`.

#### Acceptance Criteria

- `qa-permissions-platform-private-subnet-prefix-list` added to the QA CDC cluster.
- `prod-permissions-platform-private-subnet-prefix-list` added to the prod CDC cluster.
- Changes applied via Atlantis; no manual console changes.

#### Implementation Details

**`qa/kafka-infra/kafka-cluster/main.tf`**

```hcl
kafka_allowed_custom_prefix_list_names = [
  "shared-orcd-atlantis-private-subnet-prefix-list",
  "shared-orcd-private-subnet-prefix-list",
  "${var.environment}-fansifter-private-subnet-prefix-list",
  "${var.environment}-songwhip-private-subnet-prefix-list",
  "${var.environment}-accounting-private-subnet-prefix-list",
  "${var.environment}-permissions-platform-private-subnet-prefix-list", # lambda-pp-cdc
]
```

**`prod/kafka-infra/kafka-cluster/main.tf`**

```hcl
kafka_allowed_custom_prefix_list_names = [
  "shared-orcd-atlantis-private-subnet-prefix-list",
  "${var.environment}-songwhip-private-subnet-prefix-list",
  "${var.environment}-permissions-platform-private-subnet-prefix-list", # lambda-pp-cdc
]
```

---

### Ticket PP-1449 (04b) — terraform-infra: `lambda-pp-cdc` TF Module

**Service**: terraform-infra | **Blocked by**: PP-1448, PP-1452 | **Blocks**: PP-1450

#### Summary

Add a new TF module for `lambda-pp-cdc` in the PP account using `kafka_event_enabled`. The Lambda
lives at `permissions-platform/{env}/lambda-pp-cdc/` and connects to the CDC MSK cluster via direct
broker addresses (port `9094`, TLS). Access is gated by the prefix list added in Ticket 04a.

#### Acceptance Criteria

- Lambda deploys to QA and PROD environments under `permissions-platform/`.
- Lambda is VPC-attached with access to ows-permissions and ows-pdp.
- Kafka trigger uses `kafka_event_enabled = true` with CDC broker addresses at port `9094`.
- IAM execution role scoped to Secrets Manager only (no MSK IAM actions needed in self-managed mode).
- Lambda environment includes `DD_API_KEY` sourced from Secrets Manager (`{env}/datadog/DD_API_KEY`).
- Datadog monitor `{env}-lambda-pp-cdc-workstation-roles-consumer-lag` exists with warning=120s / critical=300s thresholds.

#### Implementation Details

**`permissions-platform/{env}/lambda-pp-cdc/main.tf`**

```hcl
data "aws_secretsmanager_secret_version" "datadog_api" {
  secret_id = "${var.environment}/datadog/DD_API_KEY"
}

module "lambda_pp_cdc_workstation_roles" {
  source              = "git@github.com:theorchard/terraform-lambda.git?ref=<version>"
  environment         = var.environment
  application_family  = "permissions-platform"
  lambda_name         = "lambda-pp-cdc-workstation-roles"
  use_container_image = true
  vpc_enabled         = true
  vpc_id              = module.vpc_info.vpc_id
  vpc_subnet_ids      = module.vpc_info.default_private_subnet_ids

  kafka_event_enabled                    = true
  kafka_bootstrap_servers                = var.cdc_kafka_bootstrap_servers  # broker:9094 (TLS)
  kafka_topics                           = ["cdc.musicGraphV5.profile"]
  event_source_mapping_batch_size        = 100
  event_source_mapping_starting_position = "LATEST"

  lambda_function_environment_variables = {
    ENVIRONMENT = var.environment
    DD_API_KEY  = data.aws_secretsmanager_secret_version.datadog_api.secret_string
  }

  iam_managed_policy_attachments = [aws_iam_policy.workstation_roles_lambda_policy.arn]
}
```

**`permissions-platform/{env}/lambda-pp-cdc/datadog_monitor.tf`** — new file

```hcl
data "aws_secretsmanager_secret_version" "datadog_app_key" {
  secret_id = "${var.environment}/datadog/DD_APP_KEY"
}

provider "datadog" {
  api_key = data.aws_secretsmanager_secret_version.datadog_api.secret_string
  app_key = data.aws_secretsmanager_secret_version.datadog_app_key.secret_string
}

resource "datadog_monitor" "lambda_pp_cdc_consumer_lag_monitor" {
  name    = "${var.environment}-lambda-pp-cdc-workstation-roles-consumer-lag"
  type    = "query alert"
  message = "CDC consumer lag > 5 minutes. Notify: ${var.notification_endpoints}"

  query = "max(last_30m):max:pp.cdc.workstation_roles.consumer_lag_seconds{environment:${var.environment}} >= 300"

  monitor_thresholds {
    warning  = 120   # 2 min
    critical = 300   # 5 min
  }

  include_tags        = true
  notify_no_data      = false
  notify_audit        = false
  priority            = 2
  renotify_interval   = 60
  timeout_h           = 1
  require_full_window = false

  tags = [
    "environment:${var.environment}",
    "service_name:lambda-pp-cdc-workstation-roles",
    "application_family:${var.application_family}",
    "team:permissions-platform",
  ]
}
```

**`permissions-platform/{env}/lambda-pp-cdc/versions.tf`** — add Datadog provider

```hcl
required_providers {
  # ... existing aws provider ...
  datadog = {
    source  = "datadog/datadog"
    version = ">= 3.40.0"
  }
}
```

#### Notes

- When Ticket 10 (soft delete handler) ships, add `cdc.musicGraphV5.deletedHasAccessTo` to `kafka_topics` alongside `cdc.musicGraphV5.profile`. The same lambda handles both topics; the handler routes on `evt["type"]`.
- Both QA and prod CDC MSK clusters live in the prod account (`437795906767`): `qa-managed-kafka-cdc-destination` and
  `prod-managed-kafka-cdc-destination`.
- `cdc_kafka_bootstrap_servers` values (TLS, port `9094`):
  - **QA**: `b-3.qa-managed-kafka-cdc-d.2kgc64.c6.kafka.us-east-1.amazonaws.com:9094,b-4.qa-managed-kafka-cdc-d.2kgc64.c6.kafka.us-east-1.amazonaws.com:9094,b-5.qa-managed-kafka-cdc-d.2kgc64.c6.kafka.us-east-1.amazonaws.com:9094`
  - **PROD**: `b-3.prod-managed-kafka-cdc.40dsdw.c6.kafka.us-east-1.amazonaws.com:9094,b-2.prod-managed-kafka-cdc.40dsdw.c6.kafka.us-east-1.amazonaws.com:9094,b-1.prod-managed-kafka-cdc.40dsdw.c6.kafka.us-east-1.amazonaws.com:9094`

---

### Ticket PP-1445 (05) — ows-pdp: Optimistic Locking on `DynamoDbConnector.update_item`

**Service**: ows-pdp | **Blocked by**: — | **Blocks**: PP-1450

#### Summary

`DynamoDbConnector.update_item` currently issues a blind `SET` with no `ConditionExpression`.
The CDC Lambda, Settings/SEAT, and the backfill can all write the same `pp_identity` row
concurrently, causing lost updates. Add optimistic locking on the existing `version` field:
increment on each write, retry on `ConditionalCheckFailedException`.

#### Acceptance Criteria

- `update_item` adds a `ConditionExpression` asserting the current `version` value before writing.
- On `ConditionalCheckFailedException`, the caller retries with a re-fetched item (max retries configurable).
- Existing callers (Settings/SEAT attach-detach paths) are covered by the same locking.
- Unit tests cover: successful write, first-attempt conflict + retry succeeds, max retries exceeded raises.

#### Notes

- The `IdentityTenant` schema at `pdp/fastapi/schemas/identity.py:131` already carries a `version` field — no schema
  changes needed.
- Must ship before the backfill (Ticket 06) runs against production data.

---

### Ticket PP-1450 (06) — ows-pdp: Backfill Existing Workstation Roles into DynamoDB

**Service**: ows-pdp | **Blocked by**: PP-1448, PP-1449, PP-1445 | **Blocks**: —

#### Summary

One-time backfill script to populate `workstation_admin` / `workstation_catalog` roles for all
identities that currently hold those roles in Neo4j but predate the CDC Lambda. Runs after the
Lambda and optimistic locking are deployed.

#### Acceptance Criteria

- Script reads all `LabelProfile` nodes with `administrator` or `catalog` roles from Neo4j.
- For each, resolves `identity_uuid` + `vendor_uuid` and calls `attach_detach_roles_by_identity_tenant`.
- Idempotent — safe to re-run; attaching an already-present role is a no-op.
- Dry-run mode logs intended writes without executing them.
- Progress logged at reasonable intervals (e.g. every 100 identities).

#### Implementation Details

**Scope query** — count of profiles to backfill (run before executing):

```cypher
MATCH (p:Profile {profileType: 'LabelProfile'})
WHERE any(role IN p.roles WHERE role IN ['administrator', 'catalog'])
MATCH (i:Identity)-[:HAS_PROFILE]->(p)-[:HAS_ACCESS_TO]->(v:Vendor)
UNWIND [role IN p.roles WHERE role IN ['administrator', 'catalog']] AS neo4j_role
RETURN
  sum(CASE WHEN neo4j_role = 'administrator' THEN 1 ELSE 0 END) AS administrator_count,
  sum(CASE WHEN neo4j_role = 'catalog' THEN 1 ELSE 0 END)       AS catalog_count,
  count(*)                                                        AS total_rows
```

**Counts as of 2026-03-09**

| Env  | `administrator_count` | `catalog_count` | `total_rows` |
|------|-----------------------|-----------------|--------------|
| QA   | 80,742                | 28,115          | 108,857      |
| Prod | 80,442                | 27,758          | 108,200      |

**NOTE**: `total_rows` counts (profile × vendor × role) tuples, not unique profiles, so the number will exceed the profile 
count if any profiles have multiple vendor relationships.


**CSV export query** — produces `identity_uuid,tenant_uuid,tenant_type,role,operation`:

```cypher
MATCH (p:Profile {profileType: 'LabelProfile'})
WHERE any(role IN p.roles WHERE role IN ['administrator', 'catalog'])
MATCH (i:Identity)-[:HAS_PROFILE]->(p)-[:HAS_ACCESS_TO]->(v:Vendor)
UNWIND [role IN p.roles WHERE role IN ['administrator', 'catalog']] AS neo4j_role
RETURN
  i.id                                                           AS identity_uuid,
  v.uuid                                                           AS tenant_uuid,
  'account'                                                        AS tenant_type,
  CASE neo4j_role
    WHEN 'administrator' THEN 'workstation_admin'
    WHEN 'catalog'       THEN 'workstation_catalog'
  END                                                              AS role,
  'attach'                                                         AS operation
```

Three files are needed — create them in the same directory:

**`.env`** — credentials, never commit:

```sh
NEO4J_USERNAME=<your-username>
NEO4J_PASSWORD='<your-password-here>'
```

**`export_query.cypher`** — the query above, verbatim.

**`workstation_roles_export.sh`** — the export script:

```sh
#!/usr/bin/env bash
set -euo pipefail

# Set to dev, qa, or prod
ENV=qa

case "$ENV" in
  dev)  NEO4J_HOST="neo4j+s://dev-neo4j-cluster.dev.theorchard.io" ;;
  qa)   NEO4J_HOST="neo4j+s://qa-neo4j-cluster.theorchard.io" ;;
  prod) NEO4J_HOST="neo4j+s://prod-neo4j-cluster.theorchard.io" ;;
  *)    echo "Unknown ENV: $ENV"; exit 1 ;;
esac

source .env

echo "Exporting workstation roles from $NEO4J_HOST"
{
  echo "identity_uuid,tenant_uuid,tenant_type,role,operation"
  cypher-shell -a "$NEO4J_HOST" -u "$NEO4J_USERNAME" -p "$NEO4J_PASSWORD" --format plain \
    "$(cat export_query.cypher)" \
    | tail -n +2 \
    | sed 's/"//g; s/, /,/g'
} > "workstation_roles_backfill_${ENV}.csv"
```

Run with:

```sh
chmod +x workstation_roles_export.sh && ./workstation_roles_export.sh
```

#### Notes

- Run against QA first to validate before prod.
- Optimistic locking (Ticket 05) must be in place before running against prod.

---

### Ticket PP-1485 (09) — ows-pdp: Lambda Machine Identity Cerbos Policy

**Service**: ows-pdp | **Blocked by**: — | **Blocks**: PP-1486 (10)

#### Summary

Create a Cerbos principal policy for the CDC lambda's machine identity, granting `list_tenants` and
`attach_and_detach_role` on the `identity` resource. Register the machine identity as a member of both
parent companies so `filter_for_identity_tenants` returns results for users under either company.

#### Acceptance Criteria

- New policy file at `cerbos/policies/machines/pp/<lambda-machine-name>.yml` grants `list_tenants`
  and `attach_and_detach_role` on the `identity` resource.
- `list_tenants` is unconditional (no self-restriction, unlike `pdp_integration_test_machine.yml`).
- `attach_and_detach_role` is conditioned on `P.id != R.attr.identity_uuid`.
- Machine identity is registered as a member of parent companies sme
  (`f1594122-7f99-4916-b103-08b0444c7b46`) and theorchard (`955a1bbd-b623-4ea1-ab5f-8d6620c442fb`).
- Cerbos policy unit tests cover: `list_tenants` allowed for arbitrary identity,
  `attach_and_detach_role` allowed for other user, `attach_and_detach_role` denied for self.

#### Implementation Details

**`cerbos/policies/machines/pp/<lambda-machine-name>.yml`** — new file, modelled on `pdp_integration_test_machine.yml`

```yaml
apiVersion: api.cerbos.dev/v1
principalPolicy:
  principal: <lambda-machine-identity-uuid>
  version: default
  rules:
    - resource: identity
      actions:
      - action: list_tenants
        effect: EFFECT_ALLOW
      - action: attach_and_detach_role
        effect: EFFECT_ALLOW
        condition:
          match:
            expr: P.id != R.attr.identity_uuid
```

#### Notes

- The lambda machine identity UUID must be provisioned before this policy can be wired up.
- See [Soft Delete Follow-up](https://www.notion.so/33697177520f81968744c66e04558641) for full design rationale.

---

### Ticket PP-1486 (10) — lambda-pp-cdc: DELETED_HAS_ACCESS_TO Handler

**Service**: lambda-pp-cdc | **Blocked by**: PP-1448, PP-1485 (09) | **Blocks**: —

#### Summary

Extend the CDC lambda to consume `cdc.musicGraphV5.deletedHasAccessTo` events alongside the existing
`cdc.musicGraphV5.profile` topic. Route on `evt["type"]`: a `DELETED_HAS_ACCESS_TO` CREATE (soft
delete) calls `ows-permissions` to resolve profile → identity + vendor, then blindly detaches all
known workstation roles; a DELETE (re-activation) does the same lookup and attaches all current roles
as a full sync.

#### Acceptance Criteria

- `cdc.musicGraphV5.deletedHasAccessTo` is added to `kafka_topics` in PP-1449 (TF module).
- `_process_record` routes on `evt["type"]`: `DELETED_HAS_ACCESS_TO` events go to the new handler;
  all other events continue to the existing `_diff_roles` path.
- **Soft delete (CREATE)**: calls `POST /lookup/profiles/identity/uuids/` on ows-permissions to
  resolve profile → identity + vendor, then calls `attach_detach_roles_by_identity_tenant` to detach
  all known workstation roles. Does not query ows-pdp for existing roles — detaching a role that is
  not attached is a no-op in ows-pdp.
- **Re-activation (DELETE)**: calls `POST /lookup/profiles/identity/uuids/` on ows-permissions to
  resolve profile → identity + vendor + current roles in a single round trip, then attaches all
  workstation roles as a full sync (not a diff — ows-pdp may be empty due to prior soft delete).
- Detach and attach are scoped to the `(identity_uuid, vendor_uuid)` pair resolved via the `ows-permissions` lookup — `vendor_uuid` comes from the profile record, not from the CDC event's `end` node.
- No identity found logs a warning and skips (does not raise).
- Unit tests cover: soft delete (all known roles detached), re-activation full sync, missing identity
  warning, unrelated event type routed to existing path.

#### Implementation Details

**`handler.py`** — routing branch in `_process_record`

```python
def _process_record(record: dict):
    evt = record["event"]

    if evt["type"] == "DELETED_HAS_ACCESS_TO":
        return _process_deleted_has_access_to(evt)

    # existing LabelProfile UPDATE path ...
```

**`handler.py`** — new soft delete / re-activation handler

```python
def _process_deleted_has_access_to(evt: dict):
    operation = evt["operation"]  # "CREATE" or "DELETE"
    profile_uuid = _extract_uuid_from_node(evt["start"])

    response = ows_permissions_client.lookup_profiles_identity_uuids([profile_uuid])
    profiles = response.get("profiles", [])
    if not profiles:
        logger.warning(f"No identity found for profile {profile_uuid}, skipping.")
        return Counter(), Counter()

    result = profiles[0]  # endpoint returns profiles in input order
    identity_uuid = result["identity_uuid"]
    vendor_uuid = result["vendor_uuid"]

    if operation == "CREATE":
        # soft delete — blindly detach all known workstation roles (no-op if not attached)
        return _detach_all_known_roles(identity_uuid, vendor_uuid)
    elif operation == "DELETE":
        # re-activation — full sync from ows-permissions
        return _attach_all_roles(identity_uuid, vendor_uuid, result["roles"])
```

#### Notes

- Both the soft delete and re-activation paths resolve identity + vendor via
  `POST /lookup/profiles/identity/uuids/` on ows-permissions — no GET to ows-pdp is needed.
- Detaching all known workstation roles on soft delete is safe because ows-pdp treats detach of a
  non-existent role as a no-op.
- Full sync on re-activation is required because `_diff_roles` cannot be used: ows-pdp may be empty
  while ows-permissions still carries the role array from before the soft delete.
- See [Soft Delete Follow-up](https://www.notion.so/33697177520f81968744c66e04558641) for full
  design rationale including race condition analysis.