Capture Data Mutation in Couchbase by using cURL Function with the Eventing Service

Couchbase supports you to define your own functions to navigate any mutation that happens in a bucket to a Rest API you want with the Event Service features. In this article, we will look at how to capture data mutations (i.e. create, update, and delete/expiry) that happens in a source bucket by using the cURL function.

With an Eventing function, we can listen to the mutations of a bucket we specified, navigate it to wherever we want, and process it before send it to an API. According to the official document, with Couchbase Server 6.5, they improve the usability of the Eventing Service by including new enhancements. Also, we will mention some of them.

Create an API

Before defining the function, we create an API with Node.js and Express framework. It will be a very basic API it just has 1 endpoint that listens to the Eventing function and shows the updated document whenever it is triggered. To create a new Node.js project we execute the following commands and fill the fields as you want;

npm init -y

Then we install express and body-parser and add the dependencies to the project via the following command;

npm install express body-parser -save

When creating the Node.js project, we defined the main project as the index.js file so we create it in our project as the following. In your case, you should give the name that you defined at the npm init step. Here, we just define an endpoint and it just writes the body object. The application is working at 3000 port defined.

const Express = require("express");
const BodyParser = require("body-parser");
var app = Express();
app.use(BodyParser.json());
app.use(BodyParser.urlencoded({extended: true}));
app.post("/notify", (req, res) => {
    console.log(req.body);
    res.send("OK");
});
var server = app.listen(3000, () => {
    console.log("listening...");
});

We finished creating API, now, we can continue with creating buckets and defining the Eventing function on the Couchbase.

Create buckets

Before creating the buckets, we make sure about Eventing Service is active in the cluster we will use. We can easily check it from Dashboard. Among the services seen in the picture below, the box near at the Eventing service should be green. If not we should enable the Eventing service.

dashboard

We start with creating a bucket to listen to each mutation via function. As you see in the following picture, we create 2 different buckets.

• country-sample : where the actual record kept in
• country-sample-metadata : where the metadata information kept in

buckets

Add Function

To add a new function, we go to Eventing in the menu on the left side and click the ADD FUNCTION button at the top. In the modal opened, we;

• select the Source Bucket and Metadata Bucket that we have already created
• define the Function Name
• bind the host and port that we execute the API to a URL

add function

Then we click the Next: Add Code button to define the function. In function editor, the onUpdate function (defined what happens when a mutation occurs on a document) and the onDelete function (defined what happens when a document is deleted) seen. We just define inside of the onUpdate function to evaluate the result as the following;

the onUpdate method

Firstly, we create a request object that has the path and body information. On the API side, we defined our endpoint as /notify so we declare it as the path parameter. Since we bind the URL alias, we can use URL alias instead of host and port for the path parameters. The onUpdate function accepts the doc and meta parameters.

• The doc keeps the document that just mutated
• The meta keeps the metadata of the document that just mutated

In the body parameter, we will send the id and content of the document in the id and value parameters. Since the id of the document is kept in metadata, we take it from the meta parameter. On the other hand, the value is directly set with the doc parameter. Then we use the curl method to define how we send it to API. We use POST type because we defined /notify endpoint works on POST in API above.

One of the enhancements that we mentioned is the “Ability to access HTTP request and response headers”. Therefore we can check the response that returns from API and it can be handled however we want. In here, we just send a log if the status of the response is not equal to 200.

Then we click the Save button and turn back to the Eventing page. Now, we can see the notify function just created.

the function undeployed

As you see, it is not deployed so we click the Deploy button and we make sure that it is deployed successfully. If the bar at the left turns green, it means that the function was deployed successfully.

the function deployed

Interact with the API

We started our API project with the following command and it will be ready to collect all mutations from the Country bucket whenever it happens.

node index.js

If the application starts properly, we will see the listening... message in the terminal.

Insert Query

Therefore, we go to the Query tab at the left panel and execute the following command;

INSERT INTO `country-sample` (KEY, VALUE)
VALUES (“1”, {
    “zone”: “GMT+3”,
    “name”: “Turkey”
});

We see the new document in the terminal

the insert log

Update Query

Execute the following command;

UPDATE `country-sample`
SET zone = “GMT+2”
WHERE name = “Turkey”;

We see the updated document in the terminal

the update log

Until this point, we implement a function in the Eventing Service and navigate each mutation that happens in the bucket to the API we developed. Beyond this point, the “Additional authentication types supported” is another enhancement that comes with the new version.

Implement basic authentication

Firstly, we install express-basic-auth and add the dependency to the project via the following command;

npm install express-basic-auth — save

Then we update the index.js file as the following code. We define the username and password as be and password at theBasicAuth part. It shows the auth => text and the mutation log instead of just the mutation log that comes from the Couchbase.

const Express = require("express");
const BodyParser = require("body-parser");
const BasicAuth = require(“express-basic-auth”);
var app = Express();
app.use(BodyParser.json());
app.use(BodyParser.urlencoded({extended: true}));
app.use(BasicAuth({users: { ‘be’: ‘password’}}));
app.post("/notify-auth", (req, res) => {
    console.log("auth => ");
    console.log(req.body);
    res.send("OK");
});
var server = app.listen(3000, () => {
    console.log("listening...");
});

Then, we start our API project with the following command again and it will be ready to collect all mutations from the Country bucket whenever it happens but this time it should pass the authentication mechanism.

node index.js

If the application starts properly, we will see the listening... message again in the terminal.

Provide Authentication for the Function

In this step, we add basic authentication while making the URL binding for the function. We add the notify-auth function and content of the function exactly the same as the notify function we defined.

The main difference is selecting basic auth and set username and password while we bind the URL alias. Since we have already defined them in the API, the username is set to be and the password is set to password here.

add auth function

Then we will undeploy the notify function and deploy the notify-auth function.

deploy auth function

Update Query

Execute the following command;

UPDATE `country-sample`
SET zone = “GMT+3”
WHERE name = “Turkey”;

We see the updated document in the terminal, but this time, as we expected, the auth => text was printed before the mutation log.

In this article, we looked at how to capture data change with Eventing Service in Couchbase and how to navigate it with a cURL request to an API. The API project, which contains all code we implemented at the steps above, is available here.

couchbase-samples/couchbase-eventing at main · burakcanekici/couchbase-samples