# SPIKE PP-1411: CDC Lambda Design and Deployment

[https://theorchard.atlassian.net/browse/PP-1411](https://theorchard.atlassian.net/browse/PP-1411)

[PP-1411 Tickets](https://www.notion.so/PP-1411-Tickets-31e97177520f8078a411ee99e0c756c8?pvs=21) 

## Background

The [PP-1411 spike](https://www.notion.so/SPIKE-PP-1411-Hydrate-Workstation-Roles-for-Authorization-Checks-31897177520f8197920ecbc1c4e0ab1f?pvs=21) selected Option 2 (Kafka CDC → DynamoDB) to hydrate workstation roles into `pp_identity`. This doc covers the two remaining open questions before tickets are cut: the CDC event shape and Lambda logic, and how to deploy the Lambda given a cross-account MSK constraint.

---

## CDC Topic

We consume `cdc.musicGraphV5.profile`, which emits an event for every `Profile` node change across **all** profile types. The Lambda discards non-`LabelProfile` events in-process — consistent with the [PP-1107](https://www.notion.so/SPIKE-PP-1107-ows-pdp-is-the-source-of-truth-for-ows-permissions-rap-admin-role-26b97177520f808cbeccfb9cea4a1c68?pvs=21) finding that custom topic filters are unreliable.

- **Topic**: `cdc.musicGraphV5.profile`
- **Operations**: `CREATE`, `UPDATE`, `DELETE`
- **Serialization**: plain JSON (no schema registry)
- **Trigger condition**: `profileType == "LabelProfile"` and the workstation-role subset changed

### ows-permissions DELETE vs SOFT-DELETE

ows-permissions never hard-deletes a `LabelProfile` node. [detach_label_profile_roles (user_update.py)](https://github.com/theorchard/ows-permissions/blob/d9c30db45cde2d7ac57f7c6e3985bc7fb6a2a854/permissions/logic/user_update.py#L266-L276) always calls `create_or_update_label_profile_with_tenant_relationship`,
which issues a `MERGE … ON MATCH SET roles = $roles`. When all roles are removed, only the `HAS_ACCESS_TO` relationship is soft-deleted; the node remains. Role removal always arrives as `operation: UPDATE` with a smaller or empty `roles` array — the existing `_diff_roles` logic handles this correctly. (See [Role diff](https://www.notion.so/Role-diff-31e97177520f80e08bf7c051ddd312ca?pvs=21)) 

**References**:

- detach_label_profile_roles (user_update.py)
    - [https://github.com/theorchard/ows-permissions/blob/d9c30db45cde2d7ac57f7c6e3985bc7fb6a2a854/permissions/logic/user_update.py#L266-L276](https://github.com/theorchard/ows-permissions/blob/d9c30db45cde2d7ac57f7c6e3985bc7fb6a2a854/permissions/logic/user_update.py#L266-L276)
- create_or_update_label_profile_with_tenant_relationship (profile.py)
    - [https://github.com/theorchard/ows-permissions/blob/d9c30db45cde2d7ac57f7c6e3985bc7fb6a2a854/permissions/models/profile.py#L1327](https://github.com/theorchard/ows-permissions/blob/d9c30db45cde2d7ac57f7c6e3985bc7fb6a2a854/permissions/models/profile.py#L1327)

#### Consuming DELETED_HAS_ACCESS_TO  topic

Follow-up from these threads:

- ‣
- ‣

`ows-permissions` may not update the `roles` list in the Profile node for revoke-one and revoke-all requests. In this case, the `cdc.musicGraphV5.profile` topic won’t have a CDC event to consume. We may need to consume the `DELETED_HAS_ACCESS_TO` topic to handle these revoke use cases.

When a `DELETED_HAS_ACCESS_TO` edge is created:

- If ows-pdp gets the `deactivate_one` or `deactivate_all` request from graphql-user prior to this CDC event, then there’s no work to do.
- If the lambda handles the `DELETED_HAS_ACCESS_TO` CDC event prior to ows-pdp getting the `deactivate_one` or `deactivate_all` request from graphql-user, the consumer can detach the workstation roles. PP needs to think through race-conditions.

When a `DELETED_HAS_ACCESS_TO` edge is deleted:

- Is this possible?
- If yes, ows-pdp will have hard-deleted the identity/tenant row. The consumer may need to fetch the profile to update the ows-pdp datastore with the current state of the Profile roles.

 

### Event encoding

Neo4j CDC serializes every property as a typed-value envelope. The `type` key names the active field; all others are `null`. A string property `profileType = "LabelProfile"` arrives as `{"type": "S", "S": "LabelProfile", "I64": null, ...}`.

```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]
```

### Relevant fields (trimmed)

The Lambda only reads four fields: `operation`, `keys`, and the `roles` and `profileType` properties in `state.before` / `state.after`. This example shows a subset of the full event message.

```json
{
  "event": {
    "operation": "UPDATE",
    "keys": {
      "Profile": [
        { "profileType": { "type": "S", "S": "LabelProfile" }, "uuid": null },
        { "profileType": null, "uuid": { "type": "S", "S": "<profile-uuid>" } }
      ]
    },
    "state": {
      "before": {
        "properties": {
          "profileType": { "type": "S", "S": "LabelProfile" },
          "roles": { "type": "LS", "LS": ["administrator"] }
        }
      },
      "after": {
        "properties": {
          "profileType": { "type": "S", "S": "LabelProfile" },
          "roles": { "type": "LS", "LS": ["administrator", "catalog"] }
        }
      }
    }
  }
}
```

---

## Lambda Design

The Lambda follows the `EventSourceMessage` + `JSONDeserializer` pattern used by other KDH consumers — no Avro or schema registry needed. Each invocation processes a batch: filter, resolve identity, diff roles, write.

![image.png](SPIKE%20PP-1411%20CDC%20Lambda%20Design%20and%20Deployment/image.png)

### Per-record steps

1. **Deserialize** — `JSONDeserializer` over `EventSourceMessage` records.
2. **Filter** — skip if `profileType != "LabelProfile"` (check `after` for CREATE/UPDATE, `before` for DELETE).
3. **Skip no-op** — skip if the workstation-role subset (`administrator`, `catalog`) is identical in before and after.
4. **Extract** — pull `profile_uuid` from `keys`; pull before/after role lists from `state`.
5. **Lookup** — `POST /lookup/profiles/identity/uuids/` → `identity_uuid`, `vendor_uuid`.
6. **Diff** — compare before vs. after roles to produce `roles_to_attach` / `roles_to_detach`.
7. **Write** — call `OwsPdpClient.attach_detach_roles_by_identity_tenant`.

### Operation handling

| **Operation** | **`state.before`** | **`state.after`** | **Lambda action** |
| --- | --- | --- | --- |
| `CREATE` | `null` | present | Attach all workstation roles in after state |
| `UPDATE` | present | present | Attach/detach the diff only |
| `DELETE` | present | `null` | Detach all workstation roles from before state |

> `DELETE` is included for completeness. The ows-permissions write path never hard-deletes a `LabelProfile` node — role removal always arrives as `UPDATE` with a smaller or empty `roles` array.
> 

### 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):
       """`half` is the `after` or `before` half of the kafka event message"""
        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},
    )
```

### Skeleton

```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)
```

---

## Deployment: Cross-account Challenge

`cdc.musicGraphV5.profile` lives on the CDC MSK cluster in the shared prod account (`437795906767`). The PP Lambda would normally live in the PP account (`591204808501` QA / `031099521156` PROD). 

The Lambda's native MSK ESM supports cross-account access for **provisioned** clusters via multi-VPC connectivity (AWS PrivateLink) — the CDC cluster is provisioned, so this is technically available (AWS docs: [msk-cross-account](https://docs.aws.amazon.com/lambda/latest/dg/msk-cross-account.html), [triggering-aws-lambda-function-from-a-cross-account](https://aws.amazon.com/blogs/compute/triggering-aws-lambda-function-from-a-cross-account-amazon-managed-streaming-for-apache-kafka/)). However, multi-VPC connectivity is **not currently enabled** on the cluster; enabling it requires action from the KDH team. See the `aws cli` output below.

- Output from `aws kafka describe-cluster qa-managed-kafka-cdc-destination` - See `ConnectivityInfo.ClientAuthentication` .
    
    ```jsx
    % aws kafka describe-cluster --cluster-arn "arn:aws:kafka:us-east-1:437795906767:cluster/qa-managed-kafka-cdc-destination/9d840e30-6e78-4bf7-8ab2-8d788cff806d-6"
    {
        "ClusterInfo": {
            "BrokerNodeGroupInfo": {
                "BrokerAZDistribution": "DEFAULT",
                "ClientSubnets": [
                    "subnet-043d235081c966332",
                    "subnet-067a6b45b079d7296",
                    "subnet-5366ef24"
                ],
                "InstanceType": "kafka.m5.large",
                "SecurityGroups": [
                    "sg-060638c709467512e"
                ],
                "StorageInfo": {
                    "EbsStorageInfo": {
                        "VolumeSize": 2396
                    }
                },
                "ConnectivityInfo": {
                    "PublicAccess": {
                        "Type": "DISABLED"
                    },
                    "VpcConnectivity": {
                        "ClientAuthentication": {
                            "Sasl": {
                                "Scram": {
                                    "Enabled": false
                                },
                                "Iam": {
                                    "Enabled": false
                                }
                            },
                            "Tls": {
                                "Enabled": false
                            }
                        }
                    }
                },
                "ZoneIds": [
                    "use1-az4",
                    "use1-az6",
                    "use1-az1"
                ]
            },
            "ClusterArn": "arn:aws:kafka:us-east-1:437795906767:cluster/qa-managed-kafka-cdc-destination/9d840e30-6e78-4bf7-8ab2-8d788cff806d-6",
            "ClusterName": "qa-managed-kafka-cdc-destination",
            "CreationTime": "2020-12-16T03:54:32.795Z",
            "CurrentBrokerSoftwareInfo": {
                "ConfigurationArn": "arn:aws:kafka:us-east-1:437795906767:configuration/qa-managed-kafka-cdc-destination-39x-configuration/0b6cab2e-ab1a-43da-bccf-32d2f5662ae1-11",
                "ConfigurationRevision": 1,
                "KafkaVersion": "3.9.x"
            },
            "CurrentVersion": "K3JI3C11GUW6OM",
            "EncryptionInfo": {
                "EncryptionAtRest": {
                    "DataVolumeKMSKeyId": "arn:aws:kms:us-east-1:437795906767:key/b282def5-fe2f-4bde-9d4d-72b5fbe662fc"
                },
                "EncryptionInTransit": {
                    "ClientBroker": "TLS",
                    "InCluster": true
                }
            },
            "EnhancedMonitoring": "PER_TOPIC_PER_BROKER",
            "OpenMonitoring": {
                "Prometheus": {
                    "JmxExporter": {
                        "EnabledInBroker": false
                    },
                    "NodeExporter": {
                        "EnabledInBroker": false
                    }
                }
            },
            "LoggingInfo": {
                "BrokerLogs": {
                    "CloudWatchLogs": {
                        "Enabled": true,
                        "LogGroup": "qa-managed-kafka-cdc-destination"
                    }
                }
            },
            "NumberOfBrokerNodes": 6,
            "State": "ACTIVE",
            "Tags": {
                "application_family": "kafka-data-highway",
                "environment": "qa",
                "terraform_github_repository": "terraform-infra",
                "terraformed": "true",
                "service_name": "managed-kafka-cdc-destination",
                "project": "kafka-data-highway",
                "terraform_github_path": "qa/kafka-infra/kafka-cluster"
            },
            "ZookeeperConnectString": "z-2.qa-managed-kafka-cdc-d.2kgc64.c6.kafka.us-east-1.amazonaws.com:2181,z-3.qa-managed-kafka-cdc-d.2kgc64.c6.kafka.us-east-1.amazonaws.com:2181,z-1.qa-managed-kafka-cdc-d.2kgc64.c6.kafka.us-east-1.amazonaws.com:2181",
            "ZookeeperConnectStringTls": "z-2.qa-managed-kafka-cdc-d.2kgc64.c6.kafka.us-east-1.amazonaws.com:2182,z-3.qa-managed-kafka-cdc-d.2kgc64.c6.kafka.us-east-1.amazonaws.com:2182,z-1.qa-managed-kafka-cdc-d.2kgc64.c6.kafka.us-east-1.amazonaws.com:2182",
            "StorageMode": "LOCAL",
            "CustomerActionStatus": "NONE"
        }
    }
    
    ```
    

### Option A: Self-managed Kafka Lambda in PP account

The Lambda uses `kafka_event_enabled` with CDC cluster broker addresses via `kafka_bootstrap_servers`. Lambda lives in the PP account and stays aligned with the multi-account direction.

This pattern has a working precedent: [lambda-abacus/sync-account](https://github.com/theorchard/terraform-infra/blob/7aaa9ffecb47217cfb792a975bdf3dc48599557f/accounting/uat/lambda-abacus/sync-account.tf#L32) (Accounting UAT account `989790945997`) reads `event.accounts` from the [uat-account artist-diy MSK cluster](https://github.com/theorchard/terraform-infra/blob/0ac5df87ba77ee855d7cb1ace1aba34544134678/uat/kafka-infra/kafka-cluster-artist-diy/main.tf#L49-L52) (`437795906767`) using the same `kafka_event_enabled` + broker-address approach. KDH confirmed this is all that's needed.

From `terraform-lambda`

https://github.com/theorchard/terraform-lambda/blob/953e28f573cb3664e1bf2381fabf153f49dc6ad2/variables.tf#L424

```jsx
variable "kafka_event_enabled" {
  type        = bool
  description = <<EOF
  Whether or not to create an event source mapping for Kafka. If the Kafka cluster is an MSK cluster in the same AWS account, use msk_event_enabled instead.
  Use this for self-managed Kafka clusters or MSK clusters in other AWS accounts. For cross-account triggers, the MSK cluster security group must allow ingress
  from the Lambda subnets and security group.
EOF
  default     = false
}
```

Cross-account requirements:

- Auth: TLS in transit (port `9094`); network-only via SG — no SASL/SCRAM or MSK IAM required
- PP adds the PP account's private subnet prefix list to `kafka_allowed_custom_prefix_list_names` on the CDC cluster (same pattern as `lambda-abacus`; no MSK resource-based policy needed)

![cross-account-esm.png](SPIKE%20PP-1411%20CDC%20Lambda%20Design%20and%20Deployment/cross-account-esm.png)

According to terraform-lambda, `kafka_event_enabled` works for cross-account MSK: the Lambda treats the CDC cluster like an external Kafka cluster and polls it over the network. See [AWS docs — self-managed Kafka](https://docs.aws.amazon.com/lambda/latest/dg/with-kafka.html).

### Option B: Lambda in prod account (same account as MSK)

Deploy the Lambda in the prod account (`437795906767`) alongside the CDC MSK cluster. Uses `msk_event_enabled = true` — identical to `lambda-store-api` ([and others](https://github.com/search?q=repo%3Atheorchard%2Fterraform-infra+%2Fmsk_event_enabled.*%3D.*true%2F&type=code)). No cross-account MSK complexity.

HTTP calls to ows-pdp go cross-account (prod → QA/PROD PP account), but:

- **This path is already established.** Almost all microservices live in the prod account and reach ows-pdp via existing devops CIDR/SG cross-account networking.
- A VPC-attached Lambda gets an ENI with a VPC IP — the same SG rules that apply to ECS/EC2 services should apply. **Needs devops confirmation** that Lambda ENIs are not excluded.

Tradeoff: a new PP Lambda in the shared prod account goes against the multi-account direction PP is adopting. Ownership and deployment live outside PP account CI.

---

## Terraform

Both options use the same `terraform-lambda` module shape. The key difference is the Kafka trigger block.

- **Option A**: `permissions-platform/prod/lambda-pp-cdc-workstation-roles/`
- **Option B**: `prod/lambda-pp-cdc-workstation-roles/`

```hcl
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

  # Option A: self-managed (PP account, cross-account broker access)
  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
  }

  # Option B: native MSK ESM (prod account, same-account cluster — like lambda-store-api)
  # msk_event_enabled                      = true
  # kafka_topics                           = ["cdc.musicGraphV5.profile"]
  # event_source_mapping_msk_cluster_name  = var.cdc_msk_cluster_name
  # event_source_mapping_msk_cluster_uuid  = var.cdc_msk_cluster_uuid
  # event_source_mapping_batch_size        = 100
  # event_source_mapping_starting_position = "LATEST"

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

For Option B, the IAM policy also needs MSK read actions against the CDC cluster (same as
`lambda-store-api`). For Option A, no MSK IAM actions are needed — `kafka_event_enabled` uses direct broker access, not the AWS MSK API.

---

## Comparison

|  | **Option A (PP account, self-managed)** | **Option B (prod account, native MSK ESM)** |
| --- | --- | --- |
| MSK connectivity | Cross-account — TGW + resource policy required | Same account — no extra setup |
| ows-pdp connectivity | VPC-internal | Cross-account via existing CIDR/SG rules |
| Kafka trigger | `kafka_event_enabled` | `msk_event_enabled` |
| Account ownership | PP account CI | Shared prod account |
| Multi-account alignment | Yes | No — new Lambda in shared prod |

---

## Open Questions

1. **Which account?** — ~~Option B (prod account) is simpler for MSK access and follows proven
cross-account networking to ows-pdp.~~ Option A stays in the multi-account direction. Decision needed before cutting tickets.
2. **Lambda ENI + cross-account SG rules (Option B)** — Confirm with devops that the existing
CIDR/SG rules covering prod-account services to ows-pdp also apply to Lambda ENIs. This would be the first Lambda (vs ECS/EC2) using that path.
3. **~~Option A: consumer auth~~** — **Resolved**: TLS in transit (port `9094`), network-only via SG.
No SASL/SCRAM or MSK IAM required. Confirmed by `lambda-abacus/sync-account`, which uses the same
cross-account `kafka_event_enabled` pattern against the prod-account MSK cluster.
4. **~~QA environment for Option B~~** — **Resolved**: both the QA and prod CDC MSK clusters live
in the prod account (`437795906767`). A QA Lambda deployed there targets
`qa-managed-kafka-cdc-destination`; prod targets `prod-managed-kafka-cdc-destination`.
5. **~~Missing `roles` key~~** — **Resolved**: shouldn't happen in practice, but `_extract_roles`
logs a warning if `roles` is absent and returns an empty set.
6. **Event volume / batching** — The Lambda receives up to `batch_size` records per invocation. The current skeleton calls `POST /lookup/profiles/identity/uuids/` once per record. Batching means collecting all `profile_uuid`s from the filtered records first, then making one lookup call for the whole invocation. Whether this is necessary depends on burst rate. Confirm expected volume with the KDH team before deciding — per-record is simpler if traffic is low.

```python
def handler(event, context):
    deserializer = JSONDeserializer()
    pending = []
    for _, msk_message in EventSourceMessage(event):
        record = deserializer.deserialize(msk_message.value)
        item = _filter_and_extract(record)  # returns None if skip
        if item:
            pending.append(item)

    if not pending:
        return

    # One lookup call for the entire batch
    profile_uuids = [item["profile_uuid"] for item in pending]
    identity_map = _lookup_identities(profile_uuids)  # {profile_uuid: (identity_uuid, vendor_uuid)}

    for item in pending:
        ids = identity_map.get(item["profile_uuid"])
        if not ids:
            logger.warning(f"No identity found for profile {item['profile_uuid']}, skipping.")
            continue
        identity_uuid, vendor_uuid = ids
        _write_roles(identity_uuid, vendor_uuid, item["roles_to_attach"], item["roles_to_detach"])
```

---

## References

- [Using Lambda with Amazon MSK](https://docs.aws.amazon.com/lambda/latest/dg/with-msk.html)
- [Creating cross-account event source mappings in Lambda](https://docs.aws.amazon.com/lambda/latest/dg/msk-cross-account.html) — cross-account native ESM requires provisioned cluster + multi-VPC connectivity (PrivateLink); not supported for serverless clusters
- [Triggering AWS Lambda from a cross-account Amazon MSK](https://aws.amazon.com/blogs/compute/triggering-aws-lambda-function-from-a-cross-account-amazon-managed-streaming-for-apache-kafka/) — AWS blog walkthrough for the PrivateLink path