# SalesForce Sink Connector

## Description

The SalesForce Sink connector pulls data from a Kafka topic and writes data to a SalesForce Object.
The connector is based on the Apache Camel framework. More information about the connector's configuration can be found [here](https://camel.netlify.app/camel-kafka-connector/latest/connectors/camel-salesforce-kafka-sink-connector.html).


## Local testing

### SalesForce Connected App & OAuth2

To use the connector you will require a SalesForce Connected up set up, with its `CONSUMER_KEY`, `CONSUMER_SECRET` and `REFRESH_TOKEN`. Follow [this guide](https://www.notion.so/Setting-Up-Salesforce-OAuth-App-e311551e00764749bb459aa9a3db336e) to get started with this setup,
or request them from your SalesForce admin.

It is possible to create a SalesForce development test instance by signing up [here](https://developer.salesforce.com/signup), and following the guide above, but substituting all references to the SalesForce URL with the one created for you.

### Running the connector

To test this connector locally you can use the dev Kafka cluster, which is available under VPN connection.

When you are ready, copy `.env.shadow` to `.env` and fill in the values.
```sh
cp .env.shadow .env
```

and then run 
```
awsume dev
aws ecr get-login-password --region us-east-1 | docker login --username AWS --password-stdin 086679231553.dkr.ecr.us-east-1.amazonaws.com
docker-compose up --build -d
```


If all is configured correctly you should see the following logs:
```
Woohoo! Kafka Connect is up!
Generate the connector config...
Creating Kafka Connect salesforce_sink connector
Curl HTTP status code is 200
salesforce_sink connector successfully updated
salesforce_sink conenctor starte
```

and at this point you can see the connector status and its config using the following links:

- http://localhost:8083/connectors/salesforce_sink/status

- http://localhost:8083/connectors/dynamodb_sink/config


### Secrets Manager

The connector expects to find the credentials in a secret with the following structure:
`${Environment}/${SERVICE_NAME}\${SECRETS_MANAGER_SF_CREDS}`
and such secret should contain `sf_consumer_key`, `sf_consumer_secret`, and `sf_refresh_token`.

### Kafka messages

Kafka message value should be in JSON format. The connector will try to parse the message and write it to the SalesForce Object.


## Upserting into a SalesForce Object

To enable this setup, use the following environment variables:

 - **SOBJECT_NAME**: The API name of the SalesForce object.
 - **SOBJECT_ID_NAME**: The API name of the field used as upsert key.
 - **SOBJECT_OPERATION**: `upsertSObject` for upsert operations.

For **UPSERT** operations the messages must include the `CamelHeader.sObjectIdValue` header that specifies the value of the row upsert key; if a record with such key exists, the message updates an existing row in the sObject, otherwise a new record gets created.


## Only Update a SalesForce Object

To enable this setup, use the following environment variables:

 - **SOBJECT_NAME**: The API name of the SalesForce object. Eg `Subscription__c`
 - **SOBJECT_OPERATION**: `updateSObject` for update operations.

The config param `camel.sink.endpoint.sObjectIdName` is ignored in case of update operation.
For **Update** operations the messages in kafka topic must include the `CamelHeader.sObjectId` header that specifies the id of salesforce Object. If a record with such key exists, the message updates an existing row in the sObject, otherwise it will throw an error and the message will be moved to DLQ topic. Error message:
```
org.apache.camel.component.salesforce.api.NoSuchSObjectException: {errors:[{"errorCode":"NOT_FOUND","message":"The requested resource does not exist","fields":null}],statusCode:404}
```