### Docker Compose and Go Commands for Service Management

Source: https://context7.com/pancudaniel7/blockscan-ethereum-service/llms.txt

This section provides command-line instructions for managing the Docker Compose stack and running the Blockscan Ethereum Service. It includes commands to start the stack, verify Redis, list loaded functions, run the Go application, execute tests, and stop the stack.

```bash
# Start Redis with function loading
docker compose -f deployments/docker-compose.yml up -d

# Verify Redis is running
redis-cli PING

# Check loaded functions
redis-cli FUNCTION LIST

# Run the service
CONFIG_NAME=local go run ./cmd

# Run tests
go test ./test/... -v

# Stop the stack
docker compose -f deployments/docker-compose.yml down
```

--------------------------------

### Start Real-time Ethereum Block Scanner (Go)

Source: https://context7.com/pancudaniel7/blockscan-ethereum-service/llms.txt

Initiates an Ethereum block scanner using WebSocket subscriptions for real-time block ingestion. It requires a WebSocket URL and a logger. The service is designed for horizontal scalability and fault tolerance. It provides real-time block events and supports automatic reconnection.

```go
package main

import (
    "context"
    "log"
    "sync"
    "time"

    "github.com/pancudaniel7/blockscan-ethereum-service/internal/adapter/scanner"
    "github.com/pancudaniel7/blockscan-ethereum-service/internal/pkg/applog"
)

func main() {
    logger := applog.NewAppDefaultLogger()
    var wg sync.WaitGroup

    // Configure scanner for real-time subscription mode
    config := scanner.Config{
        WebSocketsURL:          "wss://mainnet.infura.io/ws/v3/YOUR_API_KEY",
        FinalizedBlocks:        false, // Use subscription mode
        FinalizedPollDelay:     10,    // Not used in subscription mode
        FinalizedConfirmations: 64,    // Not used in subscription mode
    }

    ethScanner := scanner.NewEthereumScanner(logger, &wg, config)

    // Start scanning (validates config and launches goroutine)
    if err := ethScanner.StartScanning(); err != nil {
        log.Fatalf("Failed to start scanner: %v", err)
    }

    logger.Info("Scanner running, press Ctrl+C to stop")

    // Simulate running for 60 seconds
    time.Sleep(60 * time.Second)

    // Graceful shutdown
    ethScanner.StopScanning()
    wg.Wait()
    logger.Info("Shutdown complete")
}
```

--------------------------------

### Application Bootstrap: Full Lifecycle in Go

Source: https://context7.com/pancudaniel7/blockscan-ethereum-service/llms.txt

This Go code demonstrates a complete application bootstrap process for the blockscan-ethereum-service. It covers configuration loading, logger initialization, starting the HTTP server, registering routes, initializing the Ethereum scanner, and implementing graceful shutdown. Dependencies include Fiber for the web server, Viper for configuration, and custom internal packages for infrastructure, logging, and scanning. It takes configuration settings as input and outputs a running service with health and metrics endpoints, an active Ethereum scanner, and proper shutdown handling.

```go
package main

import (
    "log"
    "sync"

    "github.com/gofiber/fiber/v3"
    "github.com/pancudaniel7/blockscan-ethereum-service/internal/adapter/scanner"
    "github.com/pancudaniel7/blockscan-ethereum-service/internal/infra"
    "github.com/pancudaniel7/blockscan-ethereum-service/internal/pkg/applog"
    "github.com/spf13/viper"
)

func main() {
    // 1. Load configuration
    if err := infra.LoadConfig(); err != nil {
        log.Fatal("Failed to load config: ", err)
    }

    // 2. Initialize logger
    logger := applog.NewAppDefaultLogger()
    logger.Info("Starting Blockscan Ethereum Service")

    var wg sync.WaitGroup

    // 3. Start HTTP server
    server := infra.StartServer(logger, &wg)

    // 4. Register API routes
    server.Get("/health", healthCheckHandler)
    server.Get("/metrics", metricsHandler)

    // 5. Initialize Ethereum scanner
    scannerConfig := scanner.Config{
        WebSocketsURL:          viper.GetString("ethereum.websocket_url"),
        FinalizedBlocks:        viper.GetBool("ethereum.finalized_blocks"),
        FinalizedPollDelay:     viper.GetUint64("ethereum.finalized_poll_delay"),
        FinalizedConfirmations: viper.GetUint64("ethereum.finalized_confirmations"),
    }

    ethScanner := scanner.NewEthereumScanner(logger, &wg, scannerConfig)

    if err := ethScanner.StartScanning(); err != nil {
        log.Fatalf("Failed to start scanner: %v", err)
    }

    logger.Info("All components initialized successfully")

    // 6. Setup graceful shutdown
    shutdownCallback := func() error {
        logger.Info("Stopping Ethereum scanner...")
        ethScanner.StopScanning()

        logger.Info("Flushing buffers...")
        // Additional cleanup here

        return nil
    }

    // Blocks until SIGINT/SIGTERM
    infra.ShutdownServer(logger, &wg, server, shutdownCallback)

    logger.Info("Service stopped cleanly")
}

func healthCheckHandler(c fiber.Ctx) error {
    return c.JSON(fiber.Map{
        "status": "healthy",
        "service": "blockscan-ethereum",
    })
}

func metricsHandler(c fiber.Ctx) error {
    return c.JSON(fiber.Map{
        "blocks_scanned": 0,
        "blocks_published": 0,
        "errors": 0,
    })
}

```

--------------------------------

### Manage HTTP Server Lifecycle with Graceful Shutdown (Go)

Source: https://context7.com/pancudaniel7/blockscan-ethereum-service/llms.txt

This Go code demonstrates how to start a Fiber-based HTTP server and implement graceful shutdown using signal handling. It includes registering health and metrics endpoints and a configurable callback for cleanup tasks before shutting down. The server listens on a port defined in the configuration and gracefully handles termination signals (SIGINT, SIGTERM).

```go
package main

import (
    "log"
    "sync"

    "github.com/gofiber/fiber/v3"
    "github.com/pancudaniel7/blockscan-ethereum-service/internal/infra"
    "github.com/pancudaniel7/blockscan-ethereum-service/internal/pkg/applog"
)

func main() {
    // Load configuration
    if err := infra.LoadConfig(); err != nil {
        log.Fatal("Failed to load config: ", err)
    }

    logger := applog.NewAppDefaultLogger()
    var wg sync.WaitGroup

    // Start HTTP server (reads server.name and server.port from config)
    server := infra.StartServer(logger, &wg)

    // Register routes
    server.Get("/health", func(c fiber.Ctx) error {
        return c.JSON(fiber.Map{
            "status": "ok",
            "service": "blockscan-ethereum",
        })
    })

    server.Get("/metrics", func(c fiber.Ctx) error {
        return c.JSON(fiber.Map{
            "blocks_scanned": 12345,
            "errors": 2,
        })
    })

    // Setup graceful shutdown with optional termination callback
    callback := func() error {
        logger.Info("Executing cleanup tasks...")
        // Stop scanners, flush buffers, etc.
        return nil
    }

    // Blocks until SIGINT or SIGTERM received
    // Executes callback, then gracefully shuts down server (5s timeout)
    infra.ShutdownServer(logger, &wg, server, callback)
}
```

--------------------------------

### Start Finalized Ethereum Block Polling Scanner (Go)

Source: https://context7.com/pancudaniel7/blockscan-ethereum-service/llms.txt

Initiates an Ethereum block scanner in polling mode to ingest finalized blocks. This mode ensures block finality with a configurable number of confirmations. It requires a WebSocket URL, polling delay, and confirmation depth. The service handles reorgs and ensures downstream consistency.

```go
package main

import (
    "log"
    "sync"
    "time"

    "github.com/pancudaniel7/blockscan-ethereum-service/internal/adapter/scanner"
    "github.com/pancudaniel7/blockscan-ethereum-service/internal/pkg/applog"
)

func main() {
    logger := applog.NewAppDefaultLogger()
    var wg sync.WaitGroup

    // Configure for finalized block polling
    config := scanner.Config{
        WebSocketsURL:          "wss://mainnet.infura.io/ws/v3/YOUR_API_KEY",
        FinalizedBlocks:        true, // Enable polling mode
        FinalizedPollDelay:     15,   // Poll every 15 seconds
        FinalizedConfirmations: 64,   // 64 blocks behind head
    }

    ethScanner := scanner.NewEthereumScanner(logger, &wg, config)

    if err := ethScanner.StartScanning(); err != nil {
        log.Fatalf("Config validation failed: %v", err)
    }

    // Scanner polls for finalized blocks at configured interval
    // Calculates finalized block number as: latest - confirmations
    // Skips already processed blocks automatically

    time.Sleep(5 * time.Minute)
    ethScanner.StopScanning()
    wg.Wait()
}
```

--------------------------------

### Docker Compose for Redis and Function Provisioner

Source: https://context7.com/pancudaniel7/blockscan-ethereum-service/llms.txt

This Docker Compose configuration sets up a Redis instance for storing blockchain data and a sidecar container for loading Lua functions. It ensures Redis is healthy before starting the provisioner, mounts local Lua scripts, and defines environment variables for Redis connection. Volumes are used for persistent data storage and read-only mounting of scripts.

```yaml
version: '3.8'

services:
  redis:
    image: redis:7.2.4
    container_name: blockscan-redis
    ports:
      - "6379:6379"
    volumes:
      - redis-data:/data
    command: redis-server --save 60 1 --loglevel warning
    healthcheck:
      test: ["CMD", "redis-cli", "ping"]
      interval: 5s
      timeout: 3s
      retries: 5

  redis-provisioner:
    image: redis:7.2.4
    container_name: redis-provisioner
    depends_on:
      redis:
        condition: service_healthy
    volumes:
      - ./redis/functions:/functions:ro
      - ./redis/load_functions.sh:/load_functions.sh:ro
    command: sh /load_functions.sh
    environment:
      REDIS_HOST: redis
      REDIS_PORT: 6379

volumes:
  redis-data:
```

--------------------------------

### Manage Configuration with Viper in Go

Source: https://context7.com/pancudaniel7/blockscan-ethereum-service/llms.txt

Shows how to load application configuration using Viper, supporting YAML files and environment variable overrides. It demonstrates accessing various configuration parameters for server, logging, and Redis.

```go
package main

import (
    "fmt"
    "log"

    "github.com/pancudaniel7/blockscan-ethereum-service/internal/infra"
    "github.com/spf13/viper"
)

func main() {
    // Load configuration (reads from configs/ directory)
    // Default: configs/local.yml
    // Override with: CONFIG_NAME=test environment variable
    if err := infra.LoadConfig(); err != nil {
        log.Fatalf("Config load failed: %v", err)
    }

    // Access configuration values
    serverName := viper.GetString("server.name")
    serverPort := viper.GetString("server.port")
    serverHost := viper.GetString("server.host")

    logLevel := viper.GetString("log.level")

    redisHost := viper.GetString("redis.host")
    redisPort := viper.GetString("redis.port")
    redisDB := viper.GetInt("redis.db")
    redisPassword := viper.GetString("redis.password")

    fmt.Printf("Server: %s at %s:%s\n", serverName, serverHost, serverPort)
    fmt.Printf("Log Level: %s\n", logLevel)
    fmt.Printf("Redis: %s:%s (DB %d)\n", redisHost, redisPort, redisDB)

    // Environment variables override with BSCAN_ prefix
    // Example: BSCAN_SERVER_PORT=9000 overrides server.port
    // Example: BSCAN_LOG_LEVEL=trace overrides log.level
}

```

--------------------------------

### Implement Structured Logging with Configurable Levels (Go)

Source: https://context7.com/pancudaniel7/blockscan-ethereum-service/llms.txt

This Go code showcases a structured logging interface for application-wide use. It supports multiple log levels (trace, debug, info, warn, error, fatal) configurable via settings. The logger facilitates both simple message logging and structured logging with key-value pairs, aiding in detailed diagnostics and troubleshooting. It relies on the internal `applog` package.

```go
package main

import (
    "fmt"

    "github.com/pancudaniel7/blockscan-ethereum-service/internal/pkg/applog"
)

func main() {
    // Creates logger with level from config (log.level)
    // Supports: trace, debug, info, warn, error
    logger := applog.NewAppDefaultLogger()

    // Trace: Most verbose, for detailed diagnostics
    logger.Trace("Scanned block #12345678 with 250 transactions")

    // Debug: Development and troubleshooting
    logger.Debug("Connected to Redis at localhost:6379")

    // Info: General informational messages
    logger.Info("Ethereum scanner started in subscription mode")

    // Warn: Warning conditions
    logger.Warn("Block fetch took 5s, exceeding threshold")

    // Error: Error conditions that don't stop execution
    logger.Error("Failed to publish to Kafka: connection timeout")

    // Fatal: Logs error and exits with code 1
    // logger.Fatal("Critical configuration error, cannot continue")

    // Structured logging with key-value pairs
    blockNum := 12345678
    txCount := 250
    logger.Info(fmt.Sprintf("Processed block #%d with %d txs", blockNum, txCount))
}
```

--------------------------------

### Implement Exponential Backoff Retry Pattern in Go

Source: https://context7.com/pancudaniel7/blockscan-ethereum-service/llms.txt

Demonstrates how to use a context-aware retry mechanism with exponential backoff and jitter for making resilient operations. It showcases default settings, custom configurations, infinite retries with custom predicates, and handling context cancellation.

```go
package main

import (
    "context"
    "errors"
    "fmt"
    "time"

    "github.com/pancudaniel7/blockscan-ethereum-service/internal/pkg/pattern"
)

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

    // Example 1: Retry with default settings (5 attempts, 100ms initial, 2x multiplier)
    err := pattern.Retry(ctx, func(attempt int) error {
        fmt.Printf("Attempt %d: Connecting to database...\n", attempt)
        // Simulate flaky connection
        if attempt < 3 {
            return errors.New("connection refused")
        }
        return nil
    })

    if err != nil {
        fmt.Printf("Failed after retries: %v\n", err)
    }

    // Example 2: Custom retry configuration
    err = pattern.Retry(ctx, func(attempt int) error {
        fmt.Printf("Attempt %d: Fetching block from node...\n", attempt)
        return fetchBlockFromNode()
    },
        pattern.WithAttempts(10),                        // 10 attempts max
        pattern.WithInitialDelay(500*time.Millisecond), // Start at 500ms
        pattern.WithMaxDelay(30*time.Second),            // Cap at 30s
        pattern.WithMultiplier(2.0),                     // Double each time
        pattern.WithJitter(0.3),                         // 30% jitter
    )

    // Example 3: Infinite retries with custom predicate
    shouldRetry := func(err error) bool {
        // Don't retry on validation errors
        return !errors.Is(err, errInvalidInput)
    }

    err = pattern.Retry(ctx, func(attempt int) error {
        return processBlock(attempt)
    },
        pattern.WithInfiniteAttempts(),
        pattern.WithShouldRetry(shouldRetry),
    )

    // Example 4: Context cancellation
    timeoutCtx, cancel := context.WithTimeout(ctx, 5*time.Second)
    defer cancel()

    err = pattern.Retry(timeoutCtx, func(attempt int) error {
        return longRunningOperation()
    })

    if errors.Is(err, context.DeadlineExceeded) {
        fmt.Println("Operation timed out after 5 seconds")
    }
}

var errInvalidInput = errors.New("invalid input")

func fetchBlockFromNode() error { return nil }
func processBlock(attempt int) error { return nil }
func longRunningOperation() error { time.Sleep(10 * time.Second); return nil }

```

--------------------------------

### Bash Script to Load Redis Lua Functions

Source: https://context7.com/pancudaniel7/blockscan-ethereum-service/llms.txt

This shell script loads Lua functions into Redis. It waits for Redis to be ready, then uses `redis-cli` to load the Lua script located in the mounted `/functions` volume. It includes error handling and a verification step using `FCALL_RO`.

```bash
#!/bin/sh
set -e

echo "Waiting for Redis to be ready..."
sleep 2

echo "Loading Redis Lua functions..."
redis-cli -h ${REDIS_HOST} -p ${REDIS_PORT} FUNCTION LOAD REPLACE "$(cat /functions/add_block.lua)"

echo "Redis functions loaded successfully"
redis-cli -h ${REDIS_HOST} -p ${REDIS_PORT} FCALL_RO add_block 0 || echo "Function verification failed"
```

--------------------------------

### Integration Test: Block Deduplication and Stream Processing with Redis in Go

Source: https://context7.com/pancudaniel7/blockscan-ethereum-service/llms.txt

This Go integration test verifies the functionality of block deduplication and stream processing using Redis. It tests the `add_block` Lua function, checks for correct key persistence, stream entry creation, and consumer group acknowledgment. Dependencies include the Redis Go client and testify assertion library. It inputs Redis connection details and block data, and outputs verification of Redis operations and stream processing.

```go
package test

import (
    "context"
    "testing"

    "github.com/pancudaniel7/blockscan-ethereum-service/test/util"
    "github.com/stretchr/testify/assert"
    "github.com/stretchr/testify/require"
)

func TestAddBlockFunctionPersistsAndAcknowledgesStreamEntry(t *testing.T) {
    // Load test configuration
    require.NoError(t, util.InitConfig())

    // Create Redis client from config
    rdb, err := util.NewRedisClientFromConfig()
    require.NoError(t, err)
    defer rdb.Close()

    ctx := context.Background()

    // Clean database before test
    require.NoError(t, util.FlushDB(rdb, ctx))

    // Test parameters
    setKey := "block:123"
    streamKey := "blocks"
    ttl := "60000"  // 60 seconds
    id := "*"       // auto-generate stream ID
    xAddArgs := []string{"number", "123"}

    // Call add_block Lua function
    res, err := util.CallAddBlockFunction(rdb, ctx, setKey, streamKey, ttl, id, xAddArgs...)
    require.NoError(t, err)
    require.Len(t, res, 2)

    // Verify dedup key was set
    setResult := res[0].(int64)
    assert.Equal(t, int64(1), setResult, "SETNX should succeed")

    // Verify stream ID returned
    streamID := res[1].(string)
    assert.NotEmpty(t, streamID, "Stream ID should be returned")

    // Verify dedup key exists
    val, err := util.GetValue(rdb, ctx, setKey)
    require.NoError(t, err)
    assert.Equal(t, "1", val)

    // Verify stream entry
    streamEntries, err := rdb.XRange(ctx, streamKey, "-", "+").Result()
    require.NoError(t, err)
    require.Len(t, streamEntries, 1)
    assert.Equal(t, "123", streamEntries[0].Values["number"])

    // Test consumer group
    const (
        consumerGroup = "test-group"
        consumerName  = "test-consumer"
    )

    require.NoError(t, util.CreateConsumerGroup(rdb, ctx, streamKey, consumerGroup))

    // Read from consumer group
    streams, err := util.ReadFromGroup(rdb, ctx, streamKey, consumerGroup, consumerName, 1)
    require.NoError(t, err)
    require.Len(t, streams, 1)
    require.Len(t, streams[0].Messages, 1)
    assert.Equal(t, streamID, streams[0].Messages[0].ID)

    // Acknowledge message
    ackCount, err := util.XAck(rdb, ctx, streamKey, consumerGroup, streamID)
    require.NoError(t, err)
    assert.Equal(t, int64(1), ackCount)

    // Cleanup
    delCount, err := rdb.Del(ctx, setKey, streamKey).Result()
    require.NoError(t, err)
    assert.Equal(t, int64(2), delCount)
}

```

--------------------------------

### Processing Redis Streams with Consumer Groups in Go

Source: https://context7.com/pancudaniel7/blockscan-ethereum-service/llms.txt

This Go code demonstrates how to consume messages from a Redis stream using consumer groups for distributed processing. It includes creating a consumer group if it doesn't exist, reading messages in batches, processing them, and acknowledging their receipt. It also shows how to optionally delete processed messages.

```go
package main

import (
    "context"
    "fmt"
    "log"

    "github.com/go-redis/redis/v8"
)

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

    rdb := redis.NewClient(&redis.Options{
        Addr: "localhost:6379",
        DB:   0,
    })
    defer rdb.Close()

    streamKey := "blocks"
    groupName := "block-processors"
    consumerName := "processor-1"

    // Create consumer group (idempotent)
    err := rdb.XGroupCreateMkStream(ctx, streamKey, groupName, "0").Err()
    if err != nil && err.Error() != "BUSYGROUP Consumer Group name already exists" {
        log.Fatalf("Failed to create group: %v", err)
    }

    // Read from consumer group
    streams, err := rdb.XReadGroup(ctx, &redis.XReadGroupArgs{
        Group:    groupName,
        Consumer: consumerName,
        Streams:  []string{streamKey, ">"}, // ">" means new messages
        Count:    10,                       // batch size
        Block:    0,                        // blocking read
    }).Result()

    if err != nil {
        log.Fatalf("XReadGroup failed: %v", err)
    }

    for _, stream := range streams {
        for _, msg := range stream.Messages {
            blockNum := msg.Values["number"]
            blockHash := msg.Values["hash"]

            fmt.Printf("Processing block %s (hash: %s)\n", blockNum, blockHash)

            // Process block...

            // Acknowledge message
            ackCount, err := rdb.XAck(ctx, streamKey, groupName, msg.ID).Result()
            if err != nil {
                log.Printf("ACK failed: %v", err)
                continue
            }

            if ackCount == 1 {
                // Optionally delete processed message
                rdb.XDel(ctx, streamKey, msg.ID)
            }
        }
    }
}
```

--------------------------------

### Define Application Error Types with Chaining (Go)

Source: https://context7.com/pancudaniel7/blockscan-ethereum-service/llms.txt

This Go code defines a set of custom, type-safe error types for the application, including `InvalidArgErr`, `NotFoundErr`, `AlreadyExistsErr`, `NotAuthorizedErr`, `InternalErr`, `BlockScanErr`, `BlockLockErr`, and `BlockStreamErr`. Each error can optionally include a cause for chaining and provides methods for retrieving an error code and message. This enables robust error handling and inspection using `errors.As`.

```go
package main

import (
    "errors"
    "fmt"

    "github.com/pancudaniel7/blockscan-ethereum-service/internal/pkg/apperr"
)

func main() {
    // Example 1: Invalid argument error
    err := validateConfig("")
    if err != nil {
        fmt.Printf("Error: %v\n", err)
        // Output: [INVALID_ARGUMENT] websocket URL is required: empty URL provided
    }

    // Example 2: Block scan error with cause
    err = scanBlock()
    if err != nil {
        var scanErr *apperr.BlockScanErr
        if errors.As(err, &scanErr) {
            fmt.Printf("Scan failed - Code: %s, Message: %s\n",
                scanErr.Code(), scanErr.Message())
            fmt.Printf("Root cause: %v\n", errors.Unwrap(err))
        }
    }

    // Example 3: All error types available
    examples := []apperr.BaseError{
        &apperr.InvalidArgErr{Msg: "port must be between 1-65535"},
        &apperr.NotFoundErr{Msg: "block not found", Cause: errors.New("RPC returned null")},
        &apperr.AlreadyExistsErr{Msg: "consumer group already exists"},
        &apperr.NotAuthorizedErr{Msg: "invalid API key"},
        &apperr.InternalErr{Msg: "unexpected panic", Cause: errors.New("nil pointer")},
        &apperr.BlockScanErr{Msg: "subscription failed", Cause: errors.New("websocket closed")},
        &apperr.BlockLockErr{Msg: "failed to acquire lock"},
        &apperr.BlockStreamErr{Msg: "kafka publish timeout"},
    }

    for _, e := range examples {
        fmt.Printf("[%s] %s\n", e.Code(), e.Message())
    }
}

func validateConfig(url string) error {
    if url == "" {
        return &apperr.InvalidArgErr{
            Msg: "websocket URL is required",
            Cause: errors.New("empty URL provided"),
        }
    }
    return nil
}

func scanBlock() error {
    // Simulate scan failure
    rootErr := errors.New("connection refused")
    return &apperr.BlockScanErr{
        Msg: "failed to connect to Ethereum node",
        Cause: rootErr,
    }
}
```

--------------------------------

### Calling Redis Lua Function from Go

Source: https://context7.com/pancudaniel7/blockscan-ethereum-service/llms.txt

Demonstrates how to call the 'add_block' Lua function from a Go application using the go-redis client. It passes the necessary keys (deduplication key and stream key) and arguments (TTL, stream ID, and block fields) to the function. The result indicates success or failure and provides the stream ID if successful.

```go
package main

import (
    "context"
    "fmt"
    "log"

    "github.com/go-redis/redis/v8"
)

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

    rdb := redis.NewClient(&redis.Options{
        Addr: "localhost:6379",
        DB:   0,
    })
    defer rdb.Close()

    // Call add_block Lua function
    // Keys: [1] dedup key, [2] stream key
    // Args: [1] ttl_ms, [2] stream_id, [3+] stream fields
    result, err := rdb.Do(ctx,
        "FCALL", "add_block", 2,
        "block:12345678",           // dedup key
        "blocks",                   // stream name
        "300000",                   // 5 min TTL in ms
        "*",                        // auto-generate stream ID
        "number", "12345678",       // block number
        "hash", "0xabc123...",      // block hash
        "txCount", "250",           // transaction count
    ).Result()

    if err != nil {
        log.Fatalf("FCALL failed: %v", err)
    }

    arr := result.([]interface{})
    success := arr[0].(int64)

    if success == 1 {
        streamID := arr[1].(string)
        fmt.Printf("Block added to stream: %s\n", streamID)
    } else {
        fmt.Println("Block already exists (duplicate)")
    }
}
```

--------------------------------

### Atomic Block Deduplication with Redis Lua Function

Source: https://context7.com/pancudaniel7/blockscan-ethereum-service/llms.txt

Implements an atomic block deduplication and stream insertion mechanism using a Redis Lua function. It first checks for an existing key with a TTL to prevent duplicates and then adds the block to a Redis stream. This function requires a TTL in milliseconds and a block ID as arguments.

```lua
local function upk(t, i, j)
    i = i or 1; j = j or #t
    if i > j then return end
    return t[i], upk(t, i + 1, j)
end

redis.register_function{
    function_name='add_block',
    callback=function(keys, args)
        local ttl = tonumber(args[1])
        local id  = args[2]
        if not ttl or not id then return {0, false} end
        if redis.call('SET', keys[1], '1', 'NX', 'PX', ttl) then
            local msg_id = redis.call('XADD', keys[2], id, upk(args, 3))
            return {1, msg_id}
        end
        return {0, false}
    end
}
```

=== COMPLETE CONTENT === This response contains all available snippets from this library. No additional content exists. Do not make further requests.