# PP-1411 Follow-up: CDC Topic and Lambda Design

Team selected **Option 2 (Kafka CDC → DynamoDB)**. This doc details the CDC topic shape and Lambda design before tickets
are cut.

---

## CDC Topic

We consume `cdc.musicGraphV5.profile`, which emits an event for every `Profile` node change across **all** profile
types. The topic is not filtered by `profileType`, so the Lambda discards non-`LabelProfile` events in-process. This is
consistent with the PP-1107 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

### Event value encoding

Neo4j CDC serializes every property as a typed-value envelope. The `type` key names the active field; all others are
`null`. For example, 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."""
    return typed_val[typed_val["type"]]
```

### TypedDict schema

Type hints only — zero runtime cost. Covers the four fields the Lambda reads.

```python
from typing import TypedDict

class TypedValue(TypedDict):
    type: str  # "S", "I64", "LS", etc.
    S: str | None
    I64: int | None
    LS: list[str] | None


class ProfileProperties(TypedDict):
    profileType: TypedValue
    roles: TypedValue  # absent if never set


class StateHalf(TypedDict):
    properties: ProfileProperties


class EventState(TypedDict):
    before: StateHalf | None
    after: StateHalf | None


class ProfileKey(TypedDict):
    profileType: TypedValue | None
    uuid: TypedValue | None


class CdcEvent(TypedDict):
    operation: str
    keys: dict[str, list[ProfileKey]]
    state: EventState
```

### Extracting `profile_uuid`

The `keys` array holds two entries: one keyed by `profileType`/`profileId`, one by `uuid`. Pull the second:

```python
profile_uuid = next(
    extract_value(k["uuid"])
    for k in event["event"]["keys"]["Profile"]
    if k["uuid"] is not None
)
```

### Relevant fields (trimmed)

Full events are verbose. The Lambda only reads four fields:

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

<!-- DIAGRAM: cdc-event-fields — annotated CDC event showing which four fields the Lambda reads -->

---

## Lambda Design

The Lambda follows the same `EventSourceMessage` + `JSONDeserializer` pattern used by other KDH consumers in this
codebase (no Avro/schema registry needed). Each invocation processes a batch; the Lambda filters, resolves identity,
diffs roles, and calls the ows-pdp attach/detach API.

<!-- DIAGRAM: lambda-flow — flowchart: deserialize → filter → extract → lookup → diff → write -->

### 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 present in after state |
| `UPDATE`  | present        | present       | Attach/detach the diff only                         |
| `DELETE`  | present        | `null`        | Detach all workstation roles from before state      |

> **Note**: `DELETE` is included for completeness. The ows-permissions write path never hard-deletes a `LabelProfile`
> node, so this operation will not occur in practice via that path.

### 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):
        if not half:
            return set()
        roles_val = half["properties"].get("roles")
        if not roles_val:
            return set()
        return set(extract_value(roles_val)) & KNOWN_ROLES

    after_roles = _extract_roles(state["after"])
    before_roles = _extract_roles(state["before"])

    return (
        {ROLE_MAP[r] for r in after_roles - before_roles},
        {ROLE_MAP[r] for r in before_roles - after_roles},
    )
```

### Skeleton

```python
from kafka_utils.consumer.deserializer.simple_json import JSONDeserializer
from kafka_utils.consumer.source.mapping import EventSourceMessage
from lambdacommon.common_config import logger


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

    # Resolve which half to check for profileType (DELETE has no after)
    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  # workstation roles unchanged

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

---

## Open Questions

1. ~~**DELETE vs. soft-delete**~~ — **Resolved**: ows-permissions never hard-deletes a `LabelProfile` node.
   `detach_label_profile_roles` (user_update.py) 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. No second topic needed.

2. **Missing `roles` key** — If a `LabelProfile` has never had roles set, the `roles` property may be absent from the
   CDC payload entirely (not an empty `LS`). The `_diff_roles` skeleton above handles this via `.get("roles")`, but this
   needs a real test against production data to confirm.

3. **Event volume / batching** — What is the steady-state rate of `LabelProfile` updates on this topic? If the
   ows-permissions lookup is called per-record, high burst traffic could overload it. The endpoint already supports
   batching (`POST /lookup/profiles/identity/uuids/` accepts a list) — confirm whether the Lambda should batch records
   within an invocation before calling.