Valkey Message Queue with ElastiCache and Next.js

Valkey Message Queue with Next.js + AWS ElastiCache

This is an example of a Next.js application using AWS ElastiCache for implementing a reliable message queue with streams. The template demonstrates a contact form processor where messages are queued, consumed, and acknowledged using Valkey's streaming capabilities.

How to Use

This template demonstrates the code pattern for implementing message queues with Valkey streams. It's designed to work with AWS ElastiCache in production environments.

Local Development

Execute create-next-app with pnpm to bootstrap the example:

1
pnpm create next-app --example https://github.com/vercel/examples/tree/main/solutions/aws-message-queue-elasticache

Run Valkey locally:

Using Docker:

1
docker run -d -p 6379:6379 valkey/valkey:latest

Or install Valkey directly following the official installation guide.

Configure environment:

Create an .env.local file:

1
VALKEY_ENDPOINT=localhost:6379

Start the development server:

1
pnpm dev

Visit http://localhost:3000 to see the application.

Production Deployment with AWS ElastiCache

AWS ElastiCache clusters run within a VPC (private network), which requires network connectivity setup for production deployments on Vercel.

Networking Requirements

For Vercel Enterprise customers, connectivity to AWS ElastiCache is available through Vercel Secure Compute, which enables private network access between Vercel Functions and AWS VPC resources.

High-level setup steps:

AWS Side:

1. Create an ElastiCache for Valkey cluster (version 7.0+) in your AWS VPC
2. Configure security groups to allow traffic from Vercel's CIDR block
3. Set up VPC peering or AWS PrivateLink based on your architecture
4. Note the cluster endpoint URL

Vercel Side:

1. Contact Vercel to enable Secure Compute for your Enterprise account
2. Coordinate with Vercel to receive your dedicated CIDR block
3. Add the ElastiCache endpoint to your project environment variables:

1
VALKEY_ENDPOINT=your-cluster.cache.amazonaws.com:6379

1. Deploy your application

For detailed networking configuration, refer to the Vercel Secure Compute documentation.

How It Works

⚠️ Note: This demo does not include authentication. In production, add authentication middleware to protect the API endpoints from unauthorized access.

This template demonstrates a reliable serverless message queue workflow:

1. Producer: A visitor submits a contact form, and the message is immediately written to a Valkey stream
2. Consumer: A reviewer opens the processing view, which reads the next unhandled message from the consumer group
3. Acknowledgment: When the reviewer confirms, the app acknowledges the message and removes it from the pending list

Key Features:

• Single consumer group prevents message duplication
• Message acknowledgment ensures reliable processing
• Refreshing the page won't cause duplicate processing
• Demonstrates how ElastiCache Streams support reliable serverless workflows

API Endpoints

The application provides a single API route (/api/messages) with three HTTP methods demonstrating the message queue pattern:

Message Operations

• POST /api/messages - Add a new message to the queue (contact form submission)
• GET /api/messages - Read the next unprocessed message from the consumer group
• DELETE /api/messages?messageId=<id> - Acknowledge and remove a message from the pending list

Testing

Important Notes:

• Consumer groups track which messages have been delivered. Once a message is consumed (via GET), it moves to the Pending Entries List (PEL) and won't appear in subsequent GET requests until acknowledged.
• The streamMessageId returned by GET is Valkey's unique stream entry ID and must be used for the DELETE operation.
• Always complete the full flow: POST → GET → DELETE.

Complete Message Flow Example

1. Produce Message (Add to Queue)

1
curl -X POST http://localhost:3000/api/messages \
2
  -H "Content-Type: application/json" \
3
  -d '{
4
    "name": "Test User",
5
    "email": "test@example.com",
6
    "message": "Hello from local dev!"
7
  }'

Response:

1
{
2
  "streamMessageId": "1764009314892-0",
3
  "timestamp": "2024-11-24T18:35:14.890Z"
4
}

2. Consume Message (Read from Queue)

1
curl http://localhost:3000/api/messages

Response with message:

1
{
2
  "message": {
3
    "streamMessageId": "1764009314892-0",
4
    "name": "Test User",
5
    "email": "test@example.com",
6
    "message": "Hello from local dev!",
7
    "timestamp": "2024-11-24T18:35:14.890Z",
8
    "claimed": true
9
  }
10
}

Response when queue is empty:

1
{ "message": null }

Note: The claimed field indicates whether this message was recovered from the Pending Entries List (a previously delivered but unacknowledged message). Messages idle for more than 60 seconds are automatically reclaimed.

3. Acknowledge Message (Mark as Processed)

Critical: Use the streamMessageId from step 2 for the DELETE operation:

1
curl -X DELETE "http://localhost:3000/api/messages?messageId=1764009314892-0"

Response:

1
{ "success": true }

User Interface Flow

This template consists of two UI views that demonstrate the complete message queue workflow:

Submitting Messages (/)

The home page features a contact form where visitors can submit their name, email, and message. Upon submission, the message is immediately added to the Valkey stream and a confirmation is displayed.

Processing Messages (/process)

The processing view allows reviewers to consume and acknowledge messages from the queue. The page loads the messages on page load and displays:

• Message details (name, email, message content)
• Timestamp when the message was submitted
• A warning banner if the message was recovered from the pending entries list

Clicking Acknowledge confirms the message and removes it from the queue. A success message appears with a Next Message button to load the next message in the queue.

When the queue is empty, a "No message to process" indicator appears.

Troubleshooting

Message Recovery: If you GET a message but don't DELETE (acknowledge) it, the message stays in the Pending Entries List. After 60 seconds of idle time, subsequent GET requests will automatically reclaim that message (indicated by "claimed": true in the response). This is a reliability feature that handles consumer failures.

For clean testing, if you want to reset and start fresh:

1
# Access your Valkey instance
2
docker exec -it <container_id> valkey-cli
3

4
# Delete the entire stream (removes consumer group too)
5
DEL contact-messages
6

7
# Or just delete the consumer group
8
XGROUP DESTROY contact-messages contact-processors
9

10
# The consumer group will be recreated automatically on next GET

Checking Pending Messages: To see what's currently in the Pending Entries List:

1
# Access Valkey
2
docker exec -it <container_id> valkey-cli
3

4
# View pending messages
5
XPENDING contact-messages contact-processors
6

7
# View all messages in the stream
8
XRANGE contact-messages - +