# Option 3 — CDC Replica: Local copy of `vendor_restricted_features` via Kafka ## Overview Instead of calling ows-account at request time to fetch feature data, PDP maintains its own read-only replica of the `vendor_restricted_features` table (and the `features` table) in an RDS Aurora database owned by our team. The replica is kept in sync via CDC events published to the `cdc.artRelations.vendorRestrictedFeatures` Kafka topic and consumed by a Lambda function. At request time, PDP queries its local RDS directly — no cross-service HTTP call to ows-account. --- ## Architecture ``` art_relations (MySQL, owned by ows-account) vendor_restricted_features table │ │ Debezium CDC ▼ Kafka topic: cdc.artRelations.vendorRestrictedFeatures │ │ Event Source Mapping ▼ AWS Lambda (python-kafka-utils consumer) │ │ INSERT / UPDATE / DELETE ▼ RDS Aurora (ows-pdp team) vendor_restricted_features replica table features snapshot table (see below) │ │ SQL query ▼ ows-pdp (PDP service) ``` --- ## Data Model ### Tables to replicate #### `vendor_restricted_features` (CDC-maintained) Mirrors the source table in art_relations. CDC events keep this in sync in near-real-time. ```sql CREATE TABLE vendor_restricted_features_replica ( vendor_id INT NOT NULL, vendor_uuid VARCHAR(36) NOT NULL, -- denormalized from vendor table for PDP's UUID-based lookups feature_id INT NOT NULL, PRIMARY KEY (vendor_id, feature_id), INDEX idx_vendor_uuid (vendor_uuid) ); ``` > **Note on `vendor_uuid`**: The source `vendor_restricted_features` table stores `vendor_id` > (integer), but PDP works with UUIDs. The CDC event from `art_relations` may include the > `vendor_uuid` via a join in the Debezium connector config, or the Lambda can resolve it from > the `vendor` table CDC (if available). Alternatively, seed the table from an initial snapshot > that includes `vendor_uuid`, then rely on the vendor_id in CDC events for subsequent updates > (joining against a locally maintained `vendor` mapping table). > > **Simpler alternative**: Store only `(vendor_id, feature_id)` and maintain a separate > `vendor_id_to_uuid` mapping (seeded at startup from ows-account, refreshed periodically). #### `features_snapshot` (periodically refreshed or CDC-maintained) The `vendor_restricted_features` table records *restrictions* (features a vendor does NOT have). To compute enabled features, PDP needs to know the full set of feature IDs. The `features` table in art_relations is small and changes rarely. Options: - **Periodic refresh**: Lambda or PDP startup job fetches all feature IDs from ows-account's existing `/vendor//features` or a new `/features/` endpoint. - **Second CDC topic**: If `cdc.artRelations.features` is available or can be added. - **Hardcoded constant**: Since feature IDs rarely change, PDP could hold a config-managed list and receive an alert when a new feature is added. For simplicity, we recommend a **periodic refresh** (e.g., on PDP startup + hourly): ```sql CREATE TABLE features_snapshot ( feature_id INT PRIMARY KEY ); ``` ### Computing enabled features at query time ```sql -- Enabled feature IDs for a given vendor_uuid SELECT f.feature_id FROM features_snapshot f WHERE NOT EXISTS ( SELECT 1 FROM vendor_restricted_features_replica r WHERE r.vendor_uuid = :vendor_uuid AND r.feature_id = f.feature_id ); ``` Or equivalently (potentially faster with an index): ```sql SELECT f.feature_id FROM features_snapshot f LEFT JOIN vendor_restricted_features_replica r ON r.vendor_uuid = :vendor_uuid AND r.feature_id = f.feature_id WHERE r.feature_id IS NULL; ``` --- ## Lambda Consumer ### Kafka topic: `cdc.artRelations.vendorRestrictedFeatures` Messages are Debezium-formatted CDC events. Each message contains: - `op`: `c` (create/insert), `u` (update), `d` (delete), `r` (read/snapshot) - `before`: row state before the change (null for inserts) - `after`: row state after the change (null for deletes) ### Lambda handler (using python-kafka-utils) ```python from kafka_utils.consumer.deserializer.string import StringDeserializer # or Avro if schema registered from kafka_utils.consumer.source.mapping import EventSourceMessage from kafka_utils.consumer.message.debezium import DebeziumMessage, DebeziumMessageException import boto3 import pymysql def lambda_handler(event, context): deserializer = ... # Avro or JSON deserializer matching the topic's schema message = EventSourceMessage(event) conn = get_rds_connection() # connection to our RDS replica for _batch_key, msk_message in message: if not msk_message.value: continue # tombstone / delete marker at topic level db_msg = DebeziumMessage(msk_message.value, allowed_event_ops=['c', 'u', 'd', 'r']) op = db_msg.get_field_values('op') # 'c', 'u', 'd', 'r' after = db_msg.get_field_values('after') before = db_msg.get_field_values('before') with conn.cursor() as cur: if op in ('c', 'r'): # insert or snapshot read cur.execute( "INSERT IGNORE INTO vendor_restricted_features_replica " "(vendor_id, feature_id) VALUES (%s, %s)", (after['vendor_id'], after['feature_id']) ) elif op == 'u': # vendor_restricted_features likely has no non-PK columns to update, # but handle defensively cur.execute( "DELETE FROM vendor_restricted_features_replica " "WHERE vendor_id = %s AND feature_id = %s", (before['vendor_id'], before['feature_id']) ) cur.execute( "INSERT INTO vendor_restricted_features_replica " "(vendor_id, feature_id) VALUES (%s, %s)", (after['vendor_id'], after['feature_id']) ) elif op == 'd': cur.execute( "DELETE FROM vendor_restricted_features_replica " "WHERE vendor_id = %s AND feature_id = %s", (before['vendor_id'], before['feature_id']) ) conn.commit() ``` > **Error handling**: If the Lambda raises an exception, the MSK Event Source Mapping will retry > the failed batch until it succeeds or the Kafka message expires. Design the handler to be > idempotent (use `INSERT IGNORE` / `ON DUPLICATE KEY UPDATE`). --- ## RDS Infrastructure (terraform-rds) Create an Aurora Serverless v2 cluster in the ows-pdp AWS account using `terraform-rds`: ```hcl provider "aws" { region = var.aws_region } provider "aws" { alias = "networking" region = var.aws_region profile = "networking" } module "pdp_feature_controls_db" { source = "git@github.com:theorchard/terraform-rds.git//?ref=" providers = { aws.dns = aws.networking } environment = var.environment service_name = "ows-pdp-feature-controls" vpc_id = var.vpc_id rds_db_subnet_ids = var.rds_subnet_ids rds_engine = "aurora-mysql" rds_engine_version = "8.0" rds_engine_mode = "provisioned" # or "serverless" for lower-traffic environments rds_cluster_instance_class = "db.r8g.large" # Graviton, as required by terraform-rds >= 5.0 enabled_cloudwatch_logs_exports = ["audit", "error", "slowquery"] } ``` The Lambda's IAM role needs `rds-connect` permissions (or a DB user via Secrets Manager). --- ## PDP Changes ### New connector: `pdp/connectors/feature_controls_db.py` ```python class FeatureControlsDB: """Read-only client for the local vendor_restricted_features replica.""" def __init__(self, engine): self._engine = engine # SQLAlchemy async engine async def get_enabled_feature_ids_for_vendor_uuid( self, vendor_uuid: str ) -> list[int]: """Return enabled feature IDs for a vendor UUID.""" async with self._engine.connect() as conn: result = await conn.execute( text(""" SELECT f.feature_id FROM features_snapshot f LEFT JOIN vendor_restricted_features_replica r ON r.vendor_uuid = :vendor_uuid AND r.feature_id = f.feature_id WHERE r.feature_id IS NULL """), {"vendor_uuid": vendor_uuid}, ) return [row[0] for row in result] async def get_enabled_feature_ids_for_vendor_uuids( self, vendor_uuids: list[str] ) -> dict[str, list[int]]: """Return enabled feature IDs for multiple vendor UUIDs.""" # ... batched version of the above ``` ### Datasources initialization (`pdp/fastapi/datasources.py`) Add the RDS engine and `FeatureControlsDB` client to the lifespan startup, similar to how `RedisConnector` is initialized today. ### Hydration integration The hydration function `_hydrate_resources_with_feature_controls_as_needed` (described in Option 2 / the main plan doc) would accept a `FeatureControlsDB` instead of `OwsAccountClient` for the feature lookup step. Everything else stays the same. For subaccount tenants, PDP still calls `lookup_subaccounts_by_uuids` on ows-account to resolve `vendor_uuid` (this is already done by `MultiTenantProxy` for hierarchy hydration), then queries the local DB with the resolved UUID. --- ## Bootstrapping / Initial Load On first deployment the CDC topic will be empty (as noted: "currently empty"). The replica tables need to be seeded before CDC events start flowing: 1. Run a one-time migration job or Lambda that calls ows-account's `/vendor//features` endpoint (or a new bulk endpoint) to populate `vendor_restricted_features_replica` and `features_snapshot`. 2. Once seeded, the Lambda consumer keeps the replica in sync via CDC. > **Coordinate with the data pipelines team**: they need to configure the Debezium connector for > `art_relations.vendor_restricted_features` and confirm the message schema (Avro vs JSON, > field names) before the Lambda can be deployed. --- ## Pros and Cons ### Pros - **No cross-service HTTP call at authorization time** — feature data is local, sub-millisecond query latency. No ows-account dependency in the request hot path. - **Team ownership** — ows-pdp controls its own data store; no need to coordinate schema or fetch-flag changes with ows-account for future additions. - **Resilience** — PDP remains operational if ows-account has an outage; stale feature data is served rather than returning errors. - **Scalability** — RDS scales independently; no risk of overwhelming ows-account with per-request feature lookups at high traffic. - **Auditability** — CDC log provides a complete history of all changes to feature restrictions. ### Cons - **Eventual consistency** — there is a lag (typically seconds to low minutes) between a change in art_relations and its appearance in PDP's replica. During this window, authorization decisions may be based on stale data. - **Operational complexity** — new infrastructure to own: Lambda, RDS cluster, IAM roles, Secrets Manager, CloudWatch alarms, dead-letter queue for failed events. - **Bootstrapping coordination** — requires the data pipelines team to activate the CDC topic and confirm the schema before the Lambda is useful. - **`vendor_uuid` gap** — the source table stores `vendor_id` (integer); resolving to UUID requires either a join in the Debezium connector or a separately maintained `vendor` mapping table. This adds non-trivial complexity. - **Two data sources for feature computation** — enabled features = all features − restricted; keeping `features_snapshot` current is a separate concern. - **Higher upfront cost** — more work to stand up than Options 1 or 2 before a single feature lookup works in production. --- ## Required Changes ### New infrastructure 1. **RDS Aurora cluster** — provisioned via `terraform-rds` module, owned by ows-pdp team. 2. **AWS Lambda** — Python, using `python-kafka-utils`, deployed via existing Lambda pipeline. 3. **MSK Event Source Mapping** — connect Lambda to `cdc.artRelations.vendorRestrictedFeatures`. 4. **IAM roles / Secrets Manager** — Lambda needs RDS credentials; PDP service needs read-only DB credentials. 5. **Dead-letter queue (DLQ)** — for failed Lambda invocations. ### New code (ows-pdp) 1. `pdp/connectors/feature_controls_db.py` — `FeatureControlsDB` client. 2. `pdp/connectors/feature_controls_lambda/handler.py` — Lambda handler for CDC events. 3. `pdp/fastapi/datasources.py` — initialize `FeatureControlsDB` in lifespan. 4. `pdp/logic/cerbos.py` — update `_hydrate_resources_with_feature_controls_as_needed` to accept `FeatureControlsDB` and query it. 5. `pdp/fastapi/routers/identity.py` — inject `FeatureControlsDB` dependency. 6. Migration / seed script — one-time bootstrap of the replica tables. ### ows-account changes - None required for the steady-state feature lookup path. - A bulk seed endpoint (e.g., `GET /features/all/` and `GET /vendors/restricted-features/bulk/`) would help with bootstrapping, but could be avoided by seeding directly from the art_relations database if ows-pdp has read access. ### Coordination required - **Data pipelines team**: activate Debezium connector for `art_relations.vendor_restricted_features` on `cdc.artRelations.vendorRestrictedFeatures`; confirm schema (Avro/JSON, field names including whether `vendor_uuid` is included or only `vendor_id`). --- ## Comparison with Options 1 and 2 | | Option 1 (Fetch flag) | Option 2 (Policy-driven) | **Option 3 (CDC replica)** | |---|---|---|---| | New ows-account endpoint | No (extends existing) | Yes | No (for steady state) | | Extra HTTP call at request time | No (bundled with hierarchy) | Yes (separate feature call) | **No (local DB query)** | | Subaccount resolution | In ows-account | In PDP | In PDP (same as Option 2) | | Fetches only when needed | Depends on flag usage | Yes (policy-driven) | Yes (policy-driven) | | Latency impact | Low | Low-medium | **Lowest** | | ows-account changes | Low-medium | Medium | **None (steady state)** | | ows-pdp changes | Medium | High | **High + infra** | | Data freshness | Real-time | Real-time | **Eventually consistent** | | Resilience to ows-account outage | Low | Low | **High** | | Operational complexity | Low | Medium | **High** | | Upfront implementation cost | Low-medium | Medium | **High** | --- ## Open Questions 1. **CDC topic schema**: Will Debezium events include `vendor_uuid`, or only `vendor_id`? If only `vendor_id`, how do we maintain the `vendor_id → vendor_uuid` mapping? 2. **`features_snapshot` refresh**: How do we keep the full feature list current? Periodic ows-account call, second CDC topic, or a managed constant? 3. **Consistency SLA**: Is eventual consistency (seconds to minutes of lag) acceptable for authorization decisions? For example, if a feature restriction is added, PDP might briefly allow an action that should now be denied. 4. **Bootstrapping access**: Does ows-pdp have (or can it get) direct read access to the art_relations database for the initial seed, or must it go through ows-account APIs? 5. **Lambda deployment**: Does the team have an existing pattern for deploying MSK-triggered Lambdas, or is this new infrastructure to stand up from scratch?