Aiven for Valkey

const { createClient } = require('redis'); // Compatible with Valkey
const fs = require('fs');

const client = createClient({
  socket: {
    host: 'valkey-service.aivencloud.com',
    port: 12345,
    tls: true,
    ca: fs.readFileSync('./ca.pem')
  },
  password: 'your-password'
});

await client.connect();

// String operations
await client.set('counter', 0);
await client.incr('counter');
const counter = await client.get('counter');
console.log(`Counter: ${counter}`);

// Hash operations
await client.hSet('product:100', {
  name: 'Widget',
  price: '19.99',
  stock: '50'
});
const product = await client.hGetAll('product:100');

// List operations
await client.lPush('notifications', 'New message');
const notification = await client.rPop('notifications');

// Set operations
await client.sAdd('online_users', 'user1', 'user2', 'user3');
const isOnline = await client.sIsMember('online_users', 'user1');

// Sorted set with timestamps
await client.zAdd('recent_posts', [
  { score: Date.now(), value: 'post1' },
  { score: Date.now() + 1000, value: 'post2' }
]);

// Pub/Sub
const subscriber = client.duplicate();
await subscriber.connect();

await subscriber.subscribe('notifications', (message) => {
  console.log('Received:', message);
});

await client.publish('notifications', 'Hello World');

// Transaction
await client.multi()
  .incr('visits')
  .incr('page_views')
  .exec();

await client.disconnect();

package main

import (
    \"context\"
    \"fmt\"
    \"github.com/valkey-io/valkey-go\"
)

func main() {
    ctx := context.Background()

    // Create client
    client, err := valkey.NewClient(valkey.ClientOption{
        InitAddress: []string{\"valkey-service.aivencloud.com:12345\"},
        Password:    \"your-password\",
        TLSConfig:   &tls.Config{},
    })
    if err != nil {
        panic(err)
    }
    defer client.Close()

    // Set and get
    err = client.Do(ctx, client.B().Set().Key(\"key\").Value(\"hello world\").Build()).Error()
    if err != nil {
        panic(err)
    }

    result, err := client.Do(ctx, client.B().Get().Key(\"key\").Build()).ToString()
    if err != nil {
        panic(err)
    }
    fmt.Println(\"The value of key is:\", result)

    // Hash operations
    client.Do(ctx, client.B().Hset().Key(\"user:1\").
        FieldValue().FieldValue(\"name\", \"John\").FieldValue(\"age\", \"30\").
        Build())

    // List operations
    client.Do(ctx, client.B().Lpush().Key(\"tasks\").Element(\"task1\").Element(\"task2\").Build())

    // Set with expiration
    client.Do(ctx, client.B().Setex().Key(\"session\").Seconds(3600).Value(\"data\").Build())
}

import redis.clients.jedis.Jedis;
import redis.clients.jedis.JedisPool;
import redis.clients.jedis.JedisPoolConfig;
import javax.net.ssl.SSLSocketFactory;

public class ValkeyExample {
    public static void main(String[] args) {
        JedisPoolConfig poolConfig = new JedisPoolConfig();
        poolConfig.setMaxTotal(10);

        JedisPool pool = new JedisPool(
            poolConfig,
            \"valkey-service.aivencloud.com\",
            12345,
            2000,
            \"your-password\",
            true // Use SSL
        );

        try (Jedis jedis = pool.getResource()) {
            // String operations
            jedis.set(\"message\", \"Hello Valkey\");
            String message = jedis.get(\"message\");
            System.out.println(message);

            // Hash operations
            jedis.hset(\"user:1\", \"name\", \"John Doe\");
            jedis.hset(\"user:1\", \"email\", \"[email protected]\");
            Map<String, String> user = jedis.hgetAll(\"user:1\");

            // List operations
            jedis.lpush(\"logs\", \"Error: Connection failed\");
            jedis.lpush(\"logs\", \"Warning: High memory usage\");
            List<String> logs = jedis.lrange(\"logs\", 0, -1);

            // Set operations
            jedis.sadd(\"tags\", \"java\", \"valkey\", \"cache\");
            Set<String> tags = jedis.smembers(\"tags\");

            // Sorted set
            jedis.zadd(\"scores\", 100, \"player1\");
            jedis.zadd(\"scores\", 200, \"player2\");
            Set<String> topPlayers = jedis.zrevrange(\"scores\", 0, 1);

            // Expiration
            jedis.setex(\"temp_key\", 60, \"temporary_data\");

            // Pipeline
            Pipeline pipeline = jedis.pipelined();
            pipeline.set(\"key1\", \"value1\");
            pipeline.set(\"key2\", \"value2\");
            pipeline.incr(\"counter\");
            pipeline.sync();

        } finally {
            pool.close();
        }
    }
}

def create_session(user_id, session_data):
    session_id = generate_session_id()
    session_key = f\"session:{session_id}\"
    
    # Store session with 24 hour expiration
    client.setex(
        session_key,
        86400,  # 24 hours
        json.dumps({
            'user_id': user_id,
            'data': session_data,
            'created_at': time.time()
        })
    )
    
    return session_id

def get_session(session_id):
    session_key = f\"session:{session_id}\"
    session_data = client.get(session_key)
    
    if session_data:
        # Refresh expiration
        client.expire(session_key, 86400)
        return json.loads(session_data)
    
    return None

def check_rate_limit(user_id, limit=100, window=60):
    key = f\"rate_limit:{user_id}\"
    
    pipe = client.pipeline()
    pipe.incr(key)
    pipe.expire(key, window)
    results = pipe.execute()
    
    request_count = results[0]
    
    return request_count <= limit

# Increment counters
client.incr('page_views:homepage')
client.incr(f'page_views:{date}')

# Add to sorted set with timestamp
client.zadd('events', {
    json.dumps(event_data): time.time()
})

# Get recent events (last 5 minutes)
recent_events = client.zrangebyscore(
    'events',
    time.time() - 300,
    time.time()
)

# Track unique visitors with HyperLogLog
client.pfadd('unique_visitors:today', user_id)
unique_count = client.pfcount('unique_visitors:today')

# Producer
def enqueue_job(queue_name, job_data):
    job = json.dumps(job_data)
    client.lpush(f\"queue:{queue_name}\", job)

# Consumer
def process_jobs(queue_name):
    while True:
        # Blocking pop (wait for job)
        job = client.brpop(f\"queue:{queue_name}\", timeout=5)
        
        if job:
            queue, job_data = job
            job_obj = json.loads(job_data)
            
            try:
                process_job(job_obj)
            except Exception as e:
                # Move to failed queue
                client.lpush(f\"queue:{queue_name}:failed\", job_data)

# Publisher
def send_notification(channel, message):
    client.publish(channel, json.dumps(message))

# Subscriber
def listen_for_notifications():
    pubsub = client.pubsub()
    pubsub.subscribe('notifications', 'alerts')
    
    for message in pubsub.listen():
        if message['type'] == 'message':
            data = json.loads(message['data'])
            handle_notification(data)