By the end of this tutorial, you\u2019ll have a robust, end-to-end solution that reacts to document mutations or TTL-based expirations in Couchbase and archives them efficiently in S3\u2014without any manual intervention.<\/p>\n
Architecture overview<\/h2>\n
Here\u2019s what the architecture looks like:<\/p>\n
![\"Architecture\"](\"https:\/\/www.couchbase.com\/blog\/wp-content\/uploads\/sites\/1\/2025\/07\/arch.jpg\")
<\/center> <\/p>\nFlow:<\/strong><\/p>\n This setup is decoupled, scalable, and requires zero polling.<\/p>\n Couchbase Eventing<\/strong> provides a native way to trigger business logic in response to document mutations (creates, updates, deletes) or expirations.<\/p>\n With Eventing, we can:<\/p>\n Here\u2019s a simplified example of the Couchbase Eventing function we use for archiving documents. The function implements logic to handle two primary scenarios:<\/p>\n In both cases, the document is sent to an external API Gateway for archiving. If the API response status is either Note:<\/strong> For performance reasons, we recommend commenting out all the Here is how we defined the settings and bindings, while creating the eventing function.<\/p>\n Hit Hit The Lambda function consumes the SNS message and archives the full JSON to an S3 bucket, organized by date.<\/p>\n This results in an S3 object like:<\/p>\n SNS (Simple Notification Service)<\/strong> allows multiple services to receive the same message. Here, it passes the archive request to a Lambda function.<\/p>\n Steps:<\/strong><\/p>\n Ensure API Gateway is allowed to publish to the SNS topic via a trust and access policy:<\/p>\n API Gateway acts as our public endpoint that receives the archive requests and forwards them to SNS.<\/p>\n Key steps:<\/strong><\/p>\n From the options provided select Give an <\/p>\n On the next Hit Congratulations, you have just deployed an API Gateway<\/strong> that can POST your JSON document to the SNS Topic<\/strong>, which ultimately triggers the Lambda<\/strong> to write it to S3<\/strong>.<\/p>\n You can test your pipeline in two ways:<\/p>\n When you hit the After triggering, check that a corresponding object is created in your S3 bucket.<\/p>\n To test the setup from the Capella<\/strong>, insert a document with a TTL value of 120 seconds<\/strong>.<\/p>\n Execute the above SQL command from query workbench and then wait for the document to appear in the configured S3 bucket<\/strong> approximately 60 seconds before its expiration<\/strong>, as the Eventing function sets a timer to trigger one minute prior to the TTL<\/strong>.<\/p>\n Here are some common issues and how to fix them:<\/p>\n In this blog post, we built a robust, real-time document archival pipeline using:<\/p>\n This architecture is fully serverless, scales effortlessly, and is a cost-effective way to offload historical data for retention, compliance, or analysis.<\/p>\n To help you dive deeper and expand your knowledge on the technologies used in this pipeline, here are some valuable resources:<\/p>\n In modern data-driven applications, retaining historical documents is essential for compliance, auditing, and cost optimization. However, keeping all data indefinitely in your primary operational database is often unsustainable and expensive. In this blog post, I\u2019ll walk you through building a […]<\/p>\n","protected":false},"author":33279,"featured_media":17296,"comment_status":"open","ping_status":"open","sticky":false,"template":"","format":"standard","meta":{"_acf_changed":false,"inline_featured_image":false,"footnotes":""},"categories":[1821,2225,1816],"tags":[],"ppma_author":[9090],"class_list":["post-17248","post","type-post","status-publish","format-standard","has-post-thumbnail","hentry","category-couchbase-architecture","category-cloud","category-couchbase-server"],"acf":[],"yoast_head":"\n\n
\n
archive: true<\/code> flag).<\/li>\n
\nWhy Couchbase Eventing for archival?<\/h2>\n
\n
\n
archive === true<\/code> and\/or type === logs<\/code>)<\/li>\nWriting the Couchbase Eventing function<\/h2>\n
\n
\n
expiration<\/code> set, we register a timer that fires 60 seconds before the TTL. Once the timer expires, the DocTimerCallback<\/code> function is invoked, which then calls the publish(doc, meta)<\/code> function to archive the document.<\/li>\narchive: true<\/code>, the function immediately calls publish(doc, meta)<\/code> to archive the document.<\/li>\n<\/ol>\n<\/li>\n<\/ol>\n200<\/code> or 302<\/code>, the document is explicitly deleted from the source bucket, completing the archive workflow. This provides a flexible mechanism to archive documents either on demand or via TTL-based automation.<\/p>\nfunction OnUpdate(doc, meta) {\r\n if (doc.archive && doc.archive === true){\r\n log('Archiving document with ID:', meta.id);\r\n var status = publish(doc, meta);\r\n if (status === true) {\r\n delete src[meta.id];\r\n } else {\r\n log('Publish failed, document will not be deleted:', meta.id);\r\n }\r\n } else if (meta.expiration > 0){\r\n var nMinsPrior = new Date((meta.expiration - 60) * 1000);\r\n var currentTime = new Date().getTime();\r\n log('Time difference (ms): ', currentTime - nMinsPrior);\r\n if (currentTime > nMinsPrior) {\r\n log('Within 1 minute of TTL expiry, archiving:', meta.id);\r\n var publishStatus = publish(doc, meta);\r\n } else {\r\n log('Timer set for future archiving:', meta.id);\r\n createTimer(DocTimerCallback, nMinsPrior, meta.id, meta.id);\r\n }\r\n } else {\r\n log('No archiving conditions met for:', meta.id);\r\n return;\r\n }\r\n}\r\n\r\nfunction DocTimerCallback(context) {\r\n var doc = src[context];\r\n if (doc) {\r\n var meta = { id: context };\r\n var publishStatus = publish(doc, meta);\r\n } else {\r\n log('Timer callback failed: document not found for:', context);\r\n }\r\n}\r\n\r\nfunction publish(doc, meta) {\r\n try {\r\n var request = {\r\n path: 'archive',\r\n headers: {\r\n 'Content-Type': 'application\/json'\r\n },\r\n body: { ...doc, id: meta.id }\r\n };\r\n log(\"Sending request:\", request);\r\n var response = curl('POST', archive2S3, request);\r\n if (response.status === 200 || response.status === 302) {\r\n log(\"Publish success for:\", meta.id, \" Response body:\", response.body);\r\n return true;\r\n } else {\r\n log(\"Publish failed with status:\", response.status, \" Request body:\", request);\r\n return false;\r\n }\r\n } catch (e) {\r\n log(\"Exception during publish:\", e);\r\n return false;\r\n }\r\n}\r\n<\/pre>\nlog()<\/code> statements shown above. These logs were included primarily for debugging and development purposes. Excessive logging in production environments can impact performance and increase log storage costs.<\/p><\/blockquote>\n![\"\"](\"https:\/\/www.couchbase.com\/blog\/wp-content\/uploads\/sites\/1\/2025\/07\/settings.png\")
<\/p>\nNext<\/code> to create bindings. This is where we will bind the endpoint of our API Gateway, to an alias archive2S3<\/code> and also uses source bucket as alias src<\/code>. Make a note that we used Read\/Write permission for our source bucket as we would like data to be purged out from there.<\/p>\n![\"\"](\"https:\/\/www.couchbase.com\/blog\/wp-content\/uploads\/sites\/1\/2025\/07\/bindings-new.png\")
<\/p>\nNext<\/code> again and copy\/paste the JS function from above into the window and Save<\/code>. At this point your function is saved but not deployed. Hit three dots and select Deploy<\/code> option to run the eventing function. This is how it would look once function is running.<\/p>\n![\"\"](\"https:\/\/www.couchbase.com\/blog\/wp-content\/uploads\/sites\/1\/2025\/07\/deployed.png\")
<\/p>\n
\nBuilding the Lambda Function to Archive to S3<\/h2>\n
Example Lambda Code<\/h3>\n
import boto3\r\nimport json\r\nfrom datetime import datetime\r\n\r\ns3 = boto3.client('s3')\r\nbucket_name = 'your-s3-archive-bucket'\r\n\r\ndef lambda_handler(event, context):\r\n for record in event['Records']:\r\n msg = json.loads(record['Sns']['Message'])\r\n doc_id = msg['id']\r\n content = msg\r\n now = datetime.utcnow()\r\n folder = f\"{now.year}\/{now.month}\/{now.day}\"\r\n key = f\"{folder}\/{doc_id}.json\"\r\n s3.put_object(\r\n Bucket=bucket_name,\r\n Key=key,\r\n Body=json.dumps(content),\r\n ContentType='application\/json'\r\n )\r\n\r\n return {\r\n 'statusCode': 200,\r\n 'body': 'Archived successfully.'\r\n }\r\n<\/code><\/pre>\ns3:\/\/your-s3-archive-bucket\/2025\/6\/29\/log123.json<\/code><\/pre>\n
\nUsing SNS to decouple the archival trigger<\/h2>\n
\n
\n
ArchiveTriggerTopic<\/code><\/li>\n
\n![\"\"](\"https:\/\/www.couchbase.com\/blog\/wp-content\/uploads\/sites\/1\/2025\/07\/SNS-Lambda.png\")
<\/center>
\nFixing permissions<\/strong><\/p>\nTrust Policy for
snsAccess<\/code> Role:<\/h3>\n{\r\n \"Effect\": \"Allow\",\r\n \"Principal\": {\r\n \"Service\": \"apigateway.amazonaws.com\"\r\n },\r\n \"Action\": \"sts:AssumeRole\"\r\n}\r\n<\/code><\/pre>\nAccess Policy for the Role:<\/h3>\n
{\r\n \"Effect\": \"Allow\",\r\n \"Action\": \"sns:Publish\",\r\n \"Resource\": \"arn:aws:sns:us-east-1:account-id:ArchiveTriggerTopic\"\r\n}\r\n<\/code><\/pre>\n
\nSetting up API Gateway to accept archive requests<\/h2>\n
\n
\n
REST API<\/code> as it provides integration with the SNS<\/code>\u00a0service.<\/p>\n![\"\"](\"https:\/\/www.couchbase.com\/blog\/wp-content\/uploads\/sites\/1\/2025\/07\/API-GW-1.png\")
<\/p>\nAPI name<\/code> and select API endpoint type<\/code> as Regional<\/code>. Hit Create API<\/code> button.<\/p>\n![\"\"](\"https:\/\/www.couchbase.com\/blog\/wp-content\/uploads\/sites\/1\/2025\/07\/API-GW-2.png\")
<\/p>\n\n
\n
POST \/archive<\/code> route.<\/li>\n<\/ul>\n<\/li>\n<\/ul>\nResources<\/code> page, create a resource by hitting Create resource<\/code> button on the left pane.<\/p>\n![\"\"](\"https:\/\/www.couchbase.com\/blog\/wp-content\/uploads\/sites\/1\/2025\/07\/API-GW-3.png\")
<\/center>Give the Resource name<\/code>. I am calling my resource name to be archive<\/code>.<\/p>\n![\"\"](\"https:\/\/www.couchbase.com\/blog\/wp-content\/uploads\/sites\/1\/2025\/07\/API-GW-4.png\")
<\/p>\nCreate resource<\/code> button. On the next page, under the Methods<\/code> pane hit Create method<\/code> button. This will allow us to map bunch of settings to our POST method and details about our SNS<\/strong> service like what AWS Region<\/code> it is running, ARN of the IAM role<\/code> that has the required permission to publish to the SNS topic.<\/p>\n![\"\"](\"https:\/\/www.couchbase.com\/blog\/wp-content\/uploads\/sites\/1\/2025\/07\/API-GW-6.png\")
<\/center>If you scroll down the page, you’ll find the URL Query String Parameters<\/strong> section. Here, we’ll map the ARN of our SNS topic using the parameter name TopicArn<\/code><\/strong>. Additionally, we’ll map Message<\/code><\/strong> to method.request.body<\/code><\/strong>, which will contain the full payload of our JSON document.<\/p>\n![\"\"](\"https:\/\/www.couchbase.com\/blog\/wp-content\/uploads\/sites\/1\/2025\/07\/API-GW-9.png\")
<\/center>Finally, hit the Save<\/code> button.<\/p>\n![\"\"](\"https:\/\/www.couchbase.com\/blog\/wp-content\/uploads\/sites\/1\/2025\/07\/API-Method-redacted.png\")
<\/p>\n
\nTesting the end-to-end flow<\/h2>\n
First test the API POST method<\/h3>\n
\n
\n
Test<\/code> tab and submit a simple JSON, with id<\/code> field as must.<\/li>\n<\/ul>\n<\/li>\n<\/ul>\n![\"\"](\"https:\/\/www.couchbase.com\/blog\/wp-content\/uploads\/sites\/1\/2025\/07\/API-Test-button-redacted.png\")
<\/p>\nTest<\/code> button make sure the log trace shows no error and response status is 200. At this point we have our API endpoint working fine. Next we will test this service from curl.<\/p>\nFrom curl or Postman<\/h3>\n
curl -X POST https:\/\/your-api-gateway-url\/archive \\\r\n -H \"Content-Type: application\/json\" \\\r\n -d '{\r\n \"id\": \"hotel::10025\",\r\n \"type\": \"Hotel\",\r\n \"message\": \"Archiving via curl\",\r\n \"archive\": true\r\n }'\r\n<\/code><\/pre>\n![\"\"](\"https:\/\/www.couchbase.com\/blog\/wp-content\/uploads\/sites\/1\/2025\/07\/s3-bucket.png\")
<\/p>\nFrom Capella using Query Workbench<\/h3>\n
UPSERT INTO bulk.data.source (KEY, VALUE)\r\nVALUES (\"test::001\", {\"type\": \"test\", \"field\": \"value\"}, {\"expiration\": 2*60});\r\n<\/code><\/pre>\n![\"\"](\"https:\/\/www.couchbase.com\/blog\/wp-content\/uploads\/sites\/1\/2025\/07\/s3-eventing.png\")
<\/center><\/p>\n
\nTroubleshooting<\/h2>\n
ValidationError<\/strong>: message must not be null<\/strong><\/h3>\n
\n
\n
Message<\/code> field sent to SNS is empty.<\/li>\nAPI Gateway does not have permission to assume the role<\/strong><\/h3>\n
\n
\n
apigateway.amazonaws.com<\/code> service to assume it.<\/li>\n<\/ul>\n<\/li>\n<\/ul>\nWrong Content-Type in API request<\/strong><\/h3>\n
\n
\n
application\/json<\/code>.<\/li>\nSNS receives escaped or malformed JSON<\/strong><\/h3>\n
\n
\n
$util.escapeJavaScript($input.body)<\/code> in the mapping template.<\/li>\nCloudWatch logs to inspect Lambda<\/strong><\/h3>\n
\n
\n
![\"\"](\"https:\/\/www.couchbase.com\/blog\/wp-content\/uploads\/sites\/1\/2025\/07\/cloudwatch.png\")
<\/p>\nEnhancements & best practices<\/h2>\n
\n
\n
\nConclusion<\/h2>\n
\n
\n
Resources<\/h2>\n
\n
\n