# lambda-kinesis-to-neo4j

## Description

The goal of the lambda is processing of AWS Kinesis messages.

Currently it's used for a synchronization between Art Relation database and Neo4j graph.
The whole data flow consists of the following parts:
1. Art Relations is the source database (MySQL)
2. Maxwell's Daemon is a change data capture tool which listens for changes in MySQL db.
3. AWS Kinesis is a streaming service where daemon pushes the messages
4. This lambda is the main layer which contains the logic of data processing.
5. Neo4j graph is the target database.

If you're adding a new handler to this lambda, make sure the table you intend as the trigger is actually being watched by Maxwell's Daemon by checking the [include list in terraform-infra](https://github.com/theorchard/terraform-infra/blob/3eeeb1539dd5f71c7589cca0353931a217e94bc6/qa/maxwells/maxwells-neo4j-kinesis/variables.tf#L51-L64).

**Important:** After adding a table name in the terraform-infra configuration, you need to deploy the changes to Maxwell itself via the [orchard-maxwells-daemon pipeline](https://pipeline.theorchard.io/job/theorchard/job/orchard-maxwells-daemon/job/master/).


## Setup
* Install [docker](https://docs.docker.com/get-docker/)
* Install [uv](https://docs.astral.sh/uv/getting-started/installation/) (Python 3.13 + dependency management)
* Clone this repository
* Navigate to `lambda/kinesis-to-neo4j`

## Running

### Start Container
```
$ docker compose up --build -d function
```

### Execute Function
Use the HTTP client of your choice. The body of the request is the `event` passed into the function.

```
curl --request POST \
  --url http://localhost:9000/2015-03-31/functions/function/invocations \
  --header 'Content-Type: application/json' \
  --data @tests/sample_event.json
```

Here is the description of Kinesis record event https://docs.aws.amazon.com/lambda/latest/dg/with-kinesis.html and example:

```python
{'Records': [{'kinesis': {'kinesisSchemaVersion': '1.0', 'partitionKey': '60203983', 'sequenceNumber': '49604206089665496894785157698505685508862225287779188738', 'data': 'eyJkYXRhYmFzZSI6ImFydF9yZWxhdGlvbnMiLCJ0YWJsZSI6InRyYWNrX3dyaXRlciIsInR5cGUiOiJpbnNlcnQiLCJ0cyI6MTU4MjI4MzU3NCwieGlkIjo0NTM1ODUsImNvbW1pdCI6dHJ1ZSwiZGF0YSI6eyJ0cmFja193cml0ZXJfaWQiOjYwMjAzOTgzLCJ3cml0ZXJfbmFtZSI6InRlc3QtcGFydGljaXBhbnQtdHJhY2stYXJ0aXN0IiwidXBjIjoxOTM0ODM4MDUxMDgsImNkIjoxLCJ0cmFja19pZCI6MSwidW5pcXVlX3RyYWNrX2lkIjozMDY0MzM3NCwiYXJ0aXN0X2luZm9faWQiOm51bGwsImxhYmVsX3BhcnRpY2lwYW50X2lkIjpudWxsfX0=', 'approximateArrivalTimestamp': 1582283574.462}, 'eventSource': 'aws:kinesis', 'eventVersion': '1.0', 'eventID': 'shardId-000000000000:49604206089665496894785157698505685508862225287779188738', 'eventName': 'aws:kinesis:record', 'invokeIdentityArn': 'arn:aws:iam::103233932089:role/lambda-dev-neo4j-kinesis-consumer', 'awsRegion': 'us-east-1', 'eventSourceARN': 'arn:aws:kinesis:us-east-1:103233932089:stream/dev-neo4j-stream'}]}
```

The value of the `data` field is a base64 encoded string of a JSON object created by [Maxwell's daemon](https://maxwells-daemon.io/) representing the mysql db change.
Example:

```json
{
    "database": "art_relations",
    "table": "vendor_service_tier",
    "type": "update",
    "data": {
        "vendor_id": 77418,
        "service_tier_uuid": "5f2bd4fc-df94-4f35-97d3-ef23f8573279"
    },
    "old": {
        "vendor_id": 77418,
        "service_tier_uuid": "1ed7aac0-ceb6-4c09-9166-afda8f349316"
    },
    "ts": 1724686151
}
```

### Load Changes
The `src` directory and `config.py` are mounted into the container via the `docker-compose.yaml` configuration. This means you don't need to re-build the entire container to see changes. **However** the container will need to be restarted. This process should be as quick as restarting uwsgi when making changes to a web application.
```
$ docker compose restart function
```

## Linting and Testing

### Run Container
```
$ docker compose run --rm --build lint-and-test
```

Linting runs: `yamllint` → `ruff check` → `ruff format --check` → `mypy`, followed by `pytest`.

### Set pytest Options
Sometimes you may want to customize how tests run using [pytest options](https://docs.pytest.org/en/stable/usage.html) when doing development locally, only running a subset of tests using `-k` for example.

```
$ docker compose run --rm -e TEST_ARGS="-s -k some_test" lint-and-test
```

### Set pytest-cov Options
By default test coverage is calculated for files in `./src` and displayed in a table after tests run.

```
$ docker compose run --rm -e COV_REPORT=html lint-and-test
```

This writes HTML coverage output to `htmlcov/` in the lambda directory.

### Skip Linting
When developing locally, you might not want to worry about linting when working to make tests pass.

```
$ docker compose run --rm -e SKIP_LINT=1 lint-and-test
```

### Capture Exit Code
Unless the container itself crashes, or is killed, the exit code will be `0`. When the exit code of `lint-and-test.sh` script matters, a pull-request builder for example, run tests with the following flags.
```
$ docker compose up --exit-code-from lint-and-test --abort-on-container-exit --build lint-and-test
```

## Updating Dependencies

Dependencies are managed with [uv](https://docs.astral.sh/uv/). To add or update a dependency, edit `pyproject.toml` then regenerate the lockfile:

```
$ docker compose run --rm lockfile
```

Or locally (requires access to `pypi.theorchard.io`):
```
$ uv lock
```

Local `uv` commands require Python `3.13` to be available on your machine.

### Datadog Configuration

In order to get traces and metrics (in addition to logs) from the [Datadog Lambda Library](https://docs.datadoghq.com/serverless/installation/python/?tab=containerimage), you'll need to have the following environment variables:

    DD_ENV (dev, qa, or prod)
    DD_SERVICE (the name of your service)
    DD_API_KEY_SECRET_ARN or DD_API_KEY (depending on how you are accessing the api key from secrets manager)
    DD_LAMBDA_HANDLER (your lambda's original handler, e.g. "src.app.handler")
    DD_TRACE_ENABLED (set to true)
These can be set via terraform, for example [here](https://github.com/theorchard/terraform-infra/blob/master/qa/lambda-sound-recordings/lambda_sr_add_version.tf#L84-L91).