processor.go

package main

import (
	"context"
	"fmt"
	"strings"
	"sync"
	"time"

	"github.com/confluentinc/confluent-kafka-go/v2/kafka"
	"github.com/hashicorp/golang-lru/v2/expirable"
	"golang.org/x/sync/errgroup"
)

type ProcessorConfig struct {
	Buffer  int
	Workers int
}

type Listener interface {
	enqueue(message *kafka.Message)
}

type dedupCache struct {
	cache *expirable.LRU[string, bool]
	mu    sync.Mutex
}

type processFn func(context.Context, *kafka.Message, int, *dedupCache) error

type Processor struct {
	queue     chan (*kafka.Message)
	wg        errgroup.Group
	config    ProcessorConfig
	processFn processFn
	stat      *Statistics
	dedup     *dedupCache
}

func newProcessor(config *Config) (*Processor, error) {
	processor := &Processor{
		queue:  make(chan *kafka.Message, config.Processor.Buffer),
		config: config.Processor,
		stat:   newStatistics(),
		dedup: &dedupCache{
			cache: expirable.NewLRU[string, bool](100000, nil,
				time.Second*time.Duration(240)),
		},
	}

	processFn, err := selectProcessFn(processor, config.Consumer.Topic)
	if err != nil {
		return nil, err
	}

	processor.processFn = processFn
	return processor, nil
}

func selectProcessFn(p *Processor, topic string) (processFn, error) {
	parts := strings.SplitN(topic, ".", 2)
	if len(parts) < 2 {
		return nil, fmt.Errorf("invalid topic: %s", topic)
	}
	chain, suffix := parts[0], parts[1]
	suffix = strings.TrimPrefix(suffix, "broadcasted.")

	switch chain {
	case "solana":
		switch suffix {
		case "dextrades.proto":
			return p.solanaDexTradesHandler, nil
		case "tokens.proto":
			return p.solanaTokensHandler, nil
		case "transactions.proto":
			return p.solanaTransactionsHandler, nil
		}
	case "tron":
		switch suffix {
		case "dextrades.proto":
			return p.tronDexTradesHandler, nil
		case "tokens.proto":
			return p.tronTokensHandler, nil
		case "transactions.proto":
			return p.tronTransactionsHandler, nil
		}
	default:
		switch suffix {
		case "dextrades.proto":
			return p.evmDexTradesHandler, nil
		case "tokens.proto":
			return p.evmTokensHandler, nil
		case "transactions.proto":
			return p.evmTransactionsHandler, nil
		case "dexpools.proto":
			return p.evmDexPoolsHandler, nil
		case "predictions.proto":
			return p.evmPredictionsHandler, nil
		}
	}
	return nil, fmt.Errorf("unsupported topic: %s", topic)
}

func (processor *Processor) enqueue(message *kafka.Message) {
	processor.queue <- message
}

func (processor *Processor) start(ctx context.Context) {
	counter := 0
	for i := 0; i < processor.config.Workers; i++ {
		processor.wg.Go(func() error {
			i := i
			fmt.Println("Starting worker ", i)
			for {
				select {
				case <-ctx.Done():
					fmt.Println("Done, exiting processor loop worker ", i)
					return nil
				case message := <-processor.queue:
					err := processor.processFn(ctx, message, i, processor.dedup)
					if err != nil {
						fmt.Println("Error processing message", err)
					}
					counter++
					if counter%100 == 0 {
						processor.stat.report()
					}
				}
			}
			return nil
		})
	}
}

func (processor *Processor) close() {
	fmt.Println("Shutting down processor...")
	processor.wg.Wait()
	fmt.Println("Processor stopped")
	processor.stat.report()
}

func (dedup *dedupCache) isDuplicated(key string) bool {
	dedup.mu.Lock()
	defer dedup.mu.Unlock()

	if dedup.cache.Contains(key) {
		return true
	}

	dedup.cache.Add(key, true)
	return false
}