Initial implementation

.github/workflows/build_test.yml

@@ -0,0 +1,34 @@
+name: build_test
+on:
+  push:
+    branches:
+      - main
+      - dev
+  pull_request:
+
+jobs:
+  build_test:
+    runs-on: ubuntu-latest
+    strategy:
+      matrix:
+        go-version: [1.22]
+    steps:
+      - uses: actions/checkout@v4
+
+      - name: Cache Go modules
+        uses: actions/cache@v3
+        with:
+          path: ~/go/pkg/mod
+          key: ${{ runner.os }}-go-${{ hashFiles('**/go.sum') }}
+          restore-keys: |
+            ${{ runner.os }}-go-
+
+      - uses: actions/setup-go@v5
+        with:
+          go-version: ${{ matrix.go-version }}
+
+      - name: go build
+        run: make build
+
+      - name: go test
+        run: make test

.github/workflows/golangci-lint.yml

@@ -0,0 +1,27 @@
+name: golangci-lint
+on:
+  push:
+    branches:
+      - main
+      - dev
+      - release-*
+  pull_request:
+
+permissions:
+  contents: read
+  # Optional: allow write access to checks to allow the action to annotate code in the PR.
+  checks: write
+
+jobs:
+  golangci:
+    name: lint
+    runs-on: ubuntu-latest
+    steps:
+      - uses: actions/checkout@v4
+      - uses: actions/setup-go@v5
+        with:
+          go-version: stable
+      - name: golangci-lint
+        uses: golangci/golangci-lint-action@v6
+        with:
+          version: v1.61.0

.gitignore

@@ -0,0 +1,4 @@
+bin/
+*.out
+config/config.yaml
+.env

.golangci.yml

@@ -0,0 +1,54 @@
+run:
+  timeout: 5m
+
+linters:
+  disable-all: true
+  # Enable specific linter
+  # https://golangci-lint.run/usage/linters/#enabled-by-default-linters
+  enable:
+    - asciicheck
+    - bidichk
+    - durationcheck
+    - errcheck
+    - errname
+    - copyloopvar
+    - forcetypeassert
+    - goconst
+    - gofmt
+    - goimports
+    - goheader
+    - gomodguard
+    - goprintffuncname
+    - gosimple
+    - govet
+    - importas
+    - ineffassign
+    - makezero
+    - misspell
+    - nakedret
+    - nilnil
+    - promlinter
+    - staticcheck
+    - stylecheck
+    - tenv
+    - thelper
+    - tparallel
+    - typecheck
+    - thelper
+    - unconvert
+    - unused
+    - whitespace
+    - unparam
+    - revive
+    - gosec
+    - testifylint
+
+issues:
+  max-issues-per-linter: 10000
+  max-same-issues: 10000
+
+  exclude-files:
+    - ".*\\.pb\\.go"
+    - ".*\\.pb\\.gw\\.go"
+    - ".*\\.pulsar\\.go"
+    - ".*_mocks\\.go"

.mockery.yaml

@@ -0,0 +1,3 @@
+with-expecter: True 
+keeptree: True
+output: ./mocks

Dockerfile

@@ -0,0 +1,27 @@
+FROM golang:1.22-alpine AS builder
+
+ENV GO111MODULE=on \
+    CGO_ENABLED=0 \
+    GOOS=linux \
+    GOARCH=amd64
+
+WORKDIR /app
+
+COPY go.mod go.sum ./
+
+RUN go mod download
+
+COPY . .
+
+RUN go build -o allora-producer ./cmd/producer
+
+FROM alpine:3.20
+
+RUN addgroup -S appgroup && adduser -S appuser -G appgroup
+WORKDIR /home/appuser/
+
+COPY --from=builder /app/allora-producer .
+
+USER appuser
+
+CMD ["./allora-producer"]

Makefile

@@ -0,0 +1,45 @@
+# Makefile for building the Allora Producer project
+
+# Go parameters
+GOCMD=go
+GOBUILD=$(GOCMD) build
+GOCLEAN=$(GOCMD) clean
+GOTEST=$(GOCMD) test
+GOGET=$(GOCMD) get
+GOGENERATE=$(GOCMD) generate
+# Main package path
+MAIN_PACKAGE=./cmd/producer
+
+# Binary name
+BINARY_NAME=bin/allora-producer
+
+# Build the project
+build:
+	$(GOBUILD) -o $(BINARY_NAME) -v $(MAIN_PACKAGE)
+
+# Clean build files
+clean:
+	$(GOCLEAN)
+	rm -f $(BINARY_NAME)
+
+# Run tests
+test:
+	$(GOTEST) -v ./...
+
+# Get dependencies
+deps:
+	$(GOGET) -v -t -d ./...
+
+generate:
+	$(GOGENERATE) ./...
+
+# Build and run
+run: build
+	./$(BINARY_NAME)
+
+lint:
+	@echo "--> Running linter"
+	@go run github.com/golangci/golangci-lint/cmd/[email protected] run --timeout=10m --fix
+
+# Default target
+.DEFAULT_GOAL := build

README.md

@@ -0,0 +1,120 @@
+# Allora Chain Producer
+
+This repository implements a service to monitor Allora Chain blocks, parse specific transactions and events, and send them to Kafka server topics.
+
+## Features
+
+- Monitors Allora Chain blocks in real-time
+- Parses configurable transactions and events
+- Sends parsed data to specified Kafka topics
+- Offers flexible configuration for data processing and topic associations
+- Optimized for efficient processing and near real-time data handling
+
+## Configuration
+
+Transactions, events, and their associations with Kafka topics are configurable.
+
+The configuration is managed via the `config.yaml` file, which supports overriding through environment variables. We use [Viper](https://github.com/spf13/viper) to help loading the configurations.
+
+### Configuring Transactions and Events
+
+Events and transactions can be configured at the `filter_event` and `filter_transaction` sections of the configuration file. Example:
+
+```yaml
+filter_event:
+  types:
+    - "emissions.v4.EventNetworkLossSet"
+    - "emissions.v4.EventRewardsSettled"
+
+filter_transaction:
+  types:
+    - "emissions.v4.RemoveStakeRequest"
+    - "emissions.v4.InsertReputerPayloadRequest"
+```
+
+You must use the fully qualified name of the protobuf type.
+
+### Configuring association with Kafka topics
+
+Events and Transactions can be associated with Kafta topics at the kafka_topic_router section:
+
+```yaml
+kafka_topic_router:
+  - name: "domain-event.input.allora.staking"
+    types:
+      - "emissions.v4.AddStakeRequest"
+      - "emissions.v4.RemoveStakeRequest"
+  - name: "domain-event.input.allora.delegation"
+    types:
+      - "emissions.v3.DelegateStakeRemovalInfo"
+```
+
+`name` refers to the topic name and `types` is a list of fully qualified names of the types.
+
+### Other configurations
+
+Postgres, Kafka and Allora RPC configurations can also be performed at the config file:
+
+```yaml
+database:
+  url: "postgres://devuser:devpass@localhost:5432/dev-indexer?sslmode=disable"
+
+kafka:
+  seeds:
+    - "localhost:9092"
+  user: "user"
+  password: "pass"
+
+allora:
+  rpc: "https://allora-rpc.testnet.allora.network"
+  timeout: 10
+
+```
+
+### Overriding configurations using environment variables
+
+To override configurations using environment variables, you:
+1. Join the different levels of the yaml config using underscore (`_`).
+2. Change it to uppercase
+3. Add this prefix: `ALLORA_`
+
+Example:
+
+```yaml
+database:
+  url: "postgres://devuser:devpass@localhost:5432/dev-indexer?sslmode=disable"
+```
+
+Becomes:
+
+```yaml
+ALLORA_DATABASE_URL: "postgres://devuser:devpass@localhost:5432/dev-indexer?sslmode=disable"
+```
+
+## Setup and Usage
+
+[Add instructions on how to set up and run the service]
+
+## Project Structure
+
+- `cmd/`: Contains the main entry point for the application (producer).
+- `app/`: Implements the business logic, including the use cases for monitoring transactions and events.
+- `app/domain/`: Defines core abstractions such as interfaces for repositories and services.
+- `infra/`: Handles interactions with external infrastructure like Kafka, PostgreSQL, and the Allora blockchain.
+- `util/`: Contains utility functions like logging execution time.
+- `config/`: Manages configuration loading and validation using Viper.
+
+## Technologies
+
+- Go 1.22+
+- Kafka (Franz-Go)
+- Postgres (pgx)
+- Cosmos SDK Go client
+
+## Testing
+
+1. Unit tests:
+```bash
+make test
+```
+

app/app.go

@@ -0,0 +1,63 @@
+package app
+
+import (
+	"context"
+	"fmt"
+	"sync"
+
+	"github.com/allora-network/allora-producer/app/domain"
+	"github.com/rs/zerolog/log"
+)
+
+type App struct {
+	monitorEvents       domain.MonitorEventsUseCase
+	monitorTransactions domain.MonitorTransactionsUseCase
+}
+
+func NewApp(monitorEvents domain.MonitorEventsUseCase, monitorTransactions domain.MonitorTransactionsUseCase) *App {
+	return &App{monitorEvents: monitorEvents, monitorTransactions: monitorTransactions}
+}
+
+func (a *App) Run(ctx context.Context) error {
+	var wg sync.WaitGroup
+	errChan := make(chan error, 2)
+
+	// Start the monitorEvents use case
+	wg.Add(1)
+	go func() {
+		defer wg.Done()
+		log.Info().Msg("Starting monitorEvents use case")
+		if err := a.monitorEvents.Execute(ctx); err != nil {
+			errChan <- fmt.Errorf("failed to start monitorEvents use case: %w", err)
+		}
+	}()
+
+	// Start the monitorTransactions use case
+	wg.Add(1)
+	go func() {
+		defer wg.Done()
+		log.Info().Msg("Starting monitorTransactions use case")
+		if err := a.monitorTransactions.Execute(ctx); err != nil {
+			errChan <- fmt.Errorf("failed to start monitorTransactions use case: %w", err)
+		}
+	}()
+
+	// Wait for both use cases to complete
+	wg.Wait()
+	close(errChan)
+
+	log.Info().Msg("All use cases completed")
+
+	// Check for errors
+	var errs []error
+	for err := range errChan {
+		if err != nil {
+			errs = append(errs, err)
+		}
+	}
+	if len(errs) > 0 {
+		return fmt.Errorf("multiple errors occurred: %v", errs)
+	}
+
+	return nil
+}