# Lambdas: inputs and outputs

Reference for what each lambda in this repo listens to and what it produces. Two distinct sets live here:

- **AR sync** — CDC consumers that propagate changes from Art Relations (MSK + SQS pipeline) into the ownership service.
- **Step-function ingest** — CSV-driven pipeline that ingests sound recordings / products / participants / contributors from S3 uploads.

All trigger and queue wiring is defined in `terraform-infra` at `prod/neighbouring-rights/ownership-ingestion/` (mirrored under `qa/` and `dev/`). Source-of-truth files referenced below:

- Queues: `queue.tf` — defines all SQS FIFO queues + DLQs.
- MSK + SQS event source mappings: per-lambda `lambda-*.tf` files.
- Step Function: `step-function.tf` (IAM + state machine resource) and `ingestion_workflow.json` (state definition).
- S3 → Step Function trigger: `cloudwatch-trigger-event.tf` (EventBridge rule on `Object Created` events under the `ownership/ingestion/uploads` prefix).

GraphQL targets throughout are the ownership service (`graphql-neighbouring-rights`), accessed via Apollo (`@theorchard/lambda-apollo`).

## AR sync

CDC pipeline from Art Relations → ownership. MSK consumers transform / enrich records and forward to SQS FIFO queues; SQS consumers persist via GraphQL or write CSVs into the step-function ingest pipeline.

### MSK topics consumed

| Topic | Consumer |
|---|---|
| `cdc.artRelations.releases` | `ar-release-filter` |
| `cdc.artRelations.releaseArtist` | `ar-release-artist-filter` |
| `cdc.artRelations.trackArtist` | `ar-release-artist-filter` |
| `cdc.owsTrack.performer` | `ar-release-artist-filter` |
| `cdc.artRelations.track` | `ar-track-filter` |
| `cdc.artRelations.trackMasterRights` | `ar-rights-filter` |
| `cdc.artRelations.vendorContract` | `ar-rights-filter` |
| `cdc.artRelations.releaseTerritoryRestriction` | `ar-rights-filter` |
| `cdc.artRelations.subaccountRoyaltyCollection` | `ar-rights-filter` |
| `cdc.artRelations.subaccountRoyaltyCollectionTerritories` | `ar-rights-filter` |

MSK cluster (prod): `prod-managed-kafka-cdc-destination`.

Filtering happens in two layers and *both* can drop a record before it reaches a write path:

1. **AWS event-source filter criteria** — declared in each lambda's `event_source_mapping_filter_criteria_pattern` in `terraform-infra`. These are pre-lambda; matching records never invoke the function. Examples in use: `release_status == "in_content"` on the releases topic, `op != "c"` on the track topic.
2. **Lambda-level business logic** — vendor allowlists, registry checks, payload-shape validation, dedupe sets, etc. Lives in each filter lambda under `src/filters/`, `src/utils/`, `src/formatters.ts`, or `src/logicBuilderForMSKTopic.ts`.

If a source-row change isn't surfacing downstream, check both layers.

### SQS queues

All FIFO queues, all with DLQs (`-fifo-deadletter.fifo`).

| Queue (prod name) | Producers | Consumer |
|---|---|---|
| `prod-nr-ownership-filter-products-fifo-queue.fifo` | `ar-release-artist-filter`, `ar-rights-filter` (release-territory-restriction path), `ar-release-filter` (re-enqueue) | `ar-release-filter` |
| `prod-nr-ownership-sync-products-fifo-queue.fifo` | `ar-release-filter` | `ar-release-writer` |
| `prod-nr-ownership-sync-tracks-fifo-queue.fifo` | `ar-track-filter` | `ar-track-updater` |
| `prod-nr-ownership-sync-rights-fifo-queue.fifo` | `ar-rights-filter` | `ar-rules-writer` |

> Note: `ar-release-artist-filter` has `SQS_QUEUE_URL` set to `sqs_filter_products_queue.queue_url` in its env but its handler does not actually send to SQS — it only writes via GraphQL. Treat the env var as legacy config.

### Lambdas

| Lambda | Input | Output |
|---|---|---|
| `ar-release-filter` | MSK `cdc.artRelations.releases` + SQS `nr-ownership-filter-products` (dual handler) | SQS `nr-ownership-sync-products` (→ `ar-release-writer`) |
| `ar-release-artist-filter` | MSK `cdc.artRelations.releaseArtist`, `cdc.artRelations.trackArtist`, `cdc.owsTrack.performer` | GraphQL mutations (ownership service): updates product / sound-recording artists, contributor contributions |
| `ar-track-filter` | MSK `cdc.artRelations.track` | SQS `nr-ownership-sync-tracks` (→ `ar-track-updater`) + GraphQL queries for enrichment |
| `ar-rights-filter` | MSK 5 topics (trackMasterRights, vendorContract, releaseTerritoryRestriction, subaccountRoyaltyCollection, subaccountRoyaltyCollectionTerritories) | SQS `nr-ownership-sync-rights` (→ `ar-rules-writer`) + SQS `nr-ownership-filter-products` (→ `ar-release-filter`) for releaseTerritoryRestriction path + GraphQL queries |
| `ar-release-writer` | SQS `nr-ownership-sync-products` | **S3 CSV** uploaded to `ownership/ingestion/uploads/*.csv` (triggers the step-function ingest pipeline via EventBridge) + GraphQL mutations to update already-ingested products |
| `ar-track-updater` | SQS `nr-ownership-sync-tracks` | GraphQL mutations (track metadata, monetize rules) |
| `ar-rules-writer` | SQS `nr-ownership-sync-rights` | GraphQL mutations (release territory restrictions, track master rights, vendor contract, subaccount royalty collection ± territories) |

### Diagram

```mermaid
flowchart LR
  subgraph AR_MSK["MSK CDC topics (Art Relations source)"]
    direction TB
    T_REL["cdc.artRelations.releases"]
    T_RA["cdc.artRelations.releaseArtist"]
    T_TA["cdc.artRelations.trackArtist"]
    T_TP["cdc.owsTrack.performer"]
    T_TRK["cdc.artRelations.track"]
    T_TMR["cdc.artRelations.trackMasterRights"]
    T_VC["cdc.artRelations.vendorContract"]
    T_RTR["cdc.artRelations.releaseTerritoryRestriction"]
    T_SRC["cdc.artRelations.subaccountRoyaltyCollection"]
    T_SRCT["cdc.artRelations.subaccountRoyaltyCollectionTerritories"]
  end

  ARF["ar-release-filter"]
  ARAF["ar-release-artist-filter"]
  ATF["ar-track-filter"]
  ARiF["ar-rights-filter"]

  ARW["ar-release-writer"]
  ATU["ar-track-updater"]
  ARuW["ar-rules-writer"]

  QFP[["SQS<br/>nr-ownership-filter-products"]]
  QSP[["SQS<br/>nr-ownership-sync-products"]]
  QST[["SQS<br/>nr-ownership-sync-tracks"]]
  QSR[["SQS<br/>nr-ownership-sync-rights"]]

  S3UP[("S3<br/>ownership/ingestion/uploads")]
  OWS[("ownership service<br/>GraphQL")]

  T_REL --> ARF
  T_RA --> ARAF
  T_TA --> ARAF
  T_TP --> ARAF
  T_TRK --> ATF
  T_TMR --> ARiF
  T_VC --> ARiF
  T_RTR --> ARiF
  T_SRC --> ARiF
  T_SRCT --> ARiF

  ARAF --> OWS
  ATF --> QST
  ARiF --> QSR
  ARiF --> QFP

  QFP --> ARF
  ARF --> QSP

  QSP --> ARW
  QST --> ATU
  QSR --> ARuW

  ARW --> S3UP
  ARW --> OWS
  ATU --> OWS
  ARuW --> OWS

  S3UP -.EventBridge S3 'Object Created'.-> SF[/"Step Function:<br/>nr-ownership-ingestion"/]
```

## Step-function ingest

CSV uploads to `s3://<bucket>/ownership/ingestion/uploads/*.csv` are caught by an EventBridge rule (`cloudwatch-trigger-event.tf`) which starts the `nr-ownership-ingestion` state machine. The state machine is defined in `ingestion_workflow.json` and invokes 10 lambdas in this repo.

Each non-entry step receives `LambdaInput & AWSStepFunctionContext` (where `LambdaInput` is the previous step's return value, optionally fanned-out by a `Map` state) and returns a result that becomes the next step's input. Deduplicators write deduped JSON back to S3; writers persist via GraphQL.

Any state failure is caught and routed to `Finalizer` with the error captured in `$.originalFileError`, so a single failed step never silently drops a pipeline run.

### Lambdas

| Lambda | Input | Output |
|---|---|---|
| `participant-deduplicator` | EventBridge S3 PutObject (entry — receives the raw upload event after `ContextAdder` merges execution name) | S3 JSON (deduped participants); returns `{ originalFile, dedupedFile }` |
| `participant-writer` | Step Function (output of dedup) | GraphQL mutations (`NrParticipant`) |
| `product-deduplicator` | Step Function (`originalFile`) | S3 JSON (deduped products); returns `{ originalFile, dedupedFile }` |
| `product-writer` | Step Function (`Map` over deduped products, max concurrency 10) | GraphQL mutations (`NrProduct`) |
| `sr-deduplicator` | Step Function (`originalFile`) | S3 JSON (deduped sound recordings); returns `{ originalFile, dedupedFile }` |
| `sr-writer` | Step Function (`Map` over deduped SRs, max concurrency 5) | GraphQL mutations (`NrTrack`, `NrSoundRecording`); returns `{ soundRecordingUuid, isrc, labelRights, … }` |
| `rules-writer` | Step Function (chained immediately after `sr-writer` within the same Map item) | GraphQL mutations (creates ownership rules for the sound recording) |
| `contributor-deduplicator` | Step Function (`originalFile`) | S3 JSON (one file per contributor); returns `{ originalFile, dedupedFilePath }` (a prefix) |
| `contributor-writer` | Step Function (`Map` over `s3:listObjectsV2` of the prefix, max concurrency 10) | GraphQL mutations (`NrContributor` + contributions, in chunks of 50) |
| `finalizer` | Step Function (terminal) | Kafka reporter message: `finished` (or propagates `originalFileError` from a failed branch) |

### Diagram

```mermaid
flowchart TD
  S3[("S3 PutObject<br/>ownership/ingestion/uploads/*.csv")] -.EventBridge.-> CA["ContextAdder<br/>(injects ExecutionName)"]
  CA --> PD["participant-deduplicator"]
  PD --> PW["participant-writer"]
  PW --> ProdD["product-deduplicator"]
  ProdD --> MP{{"MapProducts<br/>max concurrency 10"}}
  MP --> ProdW["product-writer"]
  ProdW --> SRD["sr-deduplicator"]
  SRD --> MSR{{"MapSoundRecordings<br/>max concurrency 5"}}
  MSR --> SRW["sr-writer"]
  SRW --> RuW["rules-writer"]
  RuW --> CD["contributor-deduplicator"]
  CD --> MC{{"MapContributors<br/>max concurrency 10"}}
  MC --> CW["contributor-writer"]
  CW --> Fin["finalizer"]

  PD -. any error .-> Fin
  PW -. any error .-> Fin
  ProdD -. any error .-> Fin
  MP -. any error .-> Fin
  SRD -. any error .-> Fin
  MSR -. any error .-> Fin
  CD -. any error .-> Fin
  MC -. any error .-> Fin
```

## Upstream change impact map

For someone making changes to the Art Relations data model, here is what to look at first. Each row maps a change in an AR source table/topic to the lambdas (and downstream effects) that consume it.

| AR change | First-hit consumer | Downstream effects |
|---|---|---|
| `releases` table schema or `release_status` semantics | `ar-release-filter` | → SQS `sync-products` → `ar-release-writer` → CSV upload → **entire step-function ingest pipeline** runs over the changed releases. |
| `releaseArtist` table | `ar-release-artist-filter` | GraphQL update to product `primaryArtists` / `featuredArtists` / `productDisplayArtistName` on the ownership service. No SQS hop. |
| `trackArtist` table | `ar-release-artist-filter` | GraphQL update to sound-recording `primaryArtists` / `featuredArtists` / `displayArtistName`. (This is the highest-volume topic — has its own DD offset-lag monitor at 800k.) |
| `track` table | `ar-track-filter` | → SQS `sync-tracks` → `ar-track-updater` → GraphQL update to track metadata or end-date monetize rules on delete. |
| `trackMasterRights` | `ar-rights-filter` | → SQS `sync-rights` → `ar-rules-writer` → GraphQL upserts of track-master-rights ownership rules. |
| `vendorContract` | `ar-rights-filter` | → SQS `sync-rights` → `ar-rules-writer` → GraphQL upserts of vendor-contract ownership rules. |
| `releaseTerritoryRestriction` | `ar-rights-filter` | Splits across two queues: → SQS `sync-rights` → `ar-rules-writer` (rule writes), and → SQS `filter-products` → `ar-release-filter` (re-syncs affected products). |
| `subaccountRoyaltyCollection` | `ar-rights-filter` | → SQS `sync-rights` → `ar-rules-writer`. |
| `subaccountRoyaltyCollectionTerritories` | `ar-rights-filter` | → SQS `sync-rights` → `ar-rules-writer`. (Highest-volume of the rights topics — has its own DD offset-lag monitor at 2k.) |
| `owsTrack.performer` (not strictly AR — OWS track table) | `ar-release-artist-filter` | GraphQL update to contributor contributions by ISRC. |

Watch points for upstream schema/semantic changes:
- The MSK consumers parse Debezium-style payloads (`payload.before` / `payload.after` / `payload.op`). A column rename or type change in the source table will surface as a parse failure in the *filter* lambda — see DD alerts on offset lag and Sentry for the relevant filter.
- Field-level mappings are in each filter's `formatters.ts` / `filters/` / `logicBuilderForMSKTopic.ts`. Those are the files to grep when an upstream column changes name.
- `ar-release-filter`'s release_status filter (`"in_content"`) and `ar-track-filter`'s op filter (`!= "c"`) live in the terraform `event_source_mapping_filter_criteria_pattern` — changes to those enum values upstream need a terraform PR, not a code PR.