Event

model/log_dedup.go

@@ -0,0 +1,111 @@
+package model
+
+// LogKey defines the composite key for counting logs
+// All fields together form the unique identifier for a log entry
+type LogKey struct {
+	BlockId    uint64 // Block identifier
+	TxIndex    uint   // Transaction index within block
+	ContractId uint64 // Contract identifier
+	TopicId    uint64 // Topic identifier for log classification
+	Param1     uint64 // First parameter value
+	Param2     uint64 // Second parameter value
+	Param3     uint64 // Third parameter value
+}
+
+type LogEntry struct {
+	Key   *LogKey
+	Value *Log
+}
+
+// LogCounter manages in-memory counting of log entries
+// Uses composite key for grouping and counting duplicate log entries
+type LogCounter struct {
+	// count map stores the occurrence count for each unique log combination
+	// Key is the composite LogKey, value is the count of occurrences
+	count map[LogKey]uint
+}
+
+// NewLogCounter creates and initializes a new LogCounter instance
+// Returns a ready-to-use counter with initialized internal map
+func NewLogCounter() *LogCounter {
+	return &LogCounter{
+		count: make(map[LogKey]uint),
+	}
+}
+
+// IncrementAndGet increases the count for the given log and returns the new count
+// Creates a composite key from log fields and updates the counter
+func (lc *LogCounter) IncrementAndGet(log *Log) (uint, *LogKey) {
+	key := LogKey{
+		BlockId:    log.BlockId,
+		TxIndex:    log.TxIndex,
+		ContractId: log.ContractId,
+		TopicId:    log.TopicId,
+		Param1:     log.Param1,
+		Param2:     log.Param2,
+		Param3:     log.Param3,
+	}
+
+	// Increment count for this key
+	lc.count[key]++
+	return lc.count[key], &key
+}
+
+// GetCount retrieves the current count for the given log without modifying it
+// Returns 0 if the log combination has not been seen before
+func (lc *LogCounter) GetCount(key *LogKey) uint {
+	return lc.count[*key]
+}
+
+// Reset clears all counting data from the counter
+// Re-initializes the internal map to empty state
+func (lc *LogCounter) Reset() {
+	lc.count = make(map[LogKey]uint)
+}
+
+// GetTotalCount returns the total number of unique log combinations counted
+// This represents the size of the internal map
+func (lc *LogCounter) GetTotalCount() int {
+	return len(lc.count)
+}
+
+// ProcessAndCountLogs processes a slice of logs and returns a new slice with Count fields populated
+// Each log's Count field is set based on its occurrence order in the sequence
+func (lc *LogCounter) ProcessAndCountLogs(logs []*Log) []*Log {
+	result := make([]*Log, len(logs))
+
+	for i, log := range logs {
+		// Create a copy of the log to avoid modifying the original
+		newLog := *log
+		// Set the Count field based on occurrence
+		newLog.Count, _ = lc.IncrementAndGet(log)
+		result[i] = &newLog
+	}
+
+	return result
+}
+
+// GetStats returns statistics about the counted logs
+// Includes total unique combinations and maximum count value
+func (lc *LogCounter) GetStats() (uniqueCombinations int, maxCount uint) {
+	uniqueCombinations = len(lc.count)
+
+	for _, count := range lc.count {
+		if count > maxCount {
+			maxCount = count
+		}
+	}
+
+	return uniqueCombinations, maxCount
+}
+
+// GetAllCounts returns a copy of all counts for debugging or analysis
+// Useful for testing or monitoring purposes
+func (lc *LogCounter) GetAllCounts() map[LogKey]uint {
+	// Create a copy to prevent external modification
+	counts := make(map[LogKey]uint, len(lc.count))
+	for key, value := range lc.count {
+		counts[key] = value
+	}
+	return counts
+}

model/lru_cache.go

@@ -0,0 +1,235 @@
+package model
+
+import (
+	"container/list"
+	"sync"
+	"time"
+)
+
+// LRUCache implements a thread-safe LRU cache with generic support
+type LRUCache[K comparable, V any] struct {
+	mu       sync.RWMutex
+	capacity int                 // Maximum number of items
+	size     int                 // Current number of items
+	cache    map[K]*list.Element // Map for O(1) lookups
+	list     *list.List          // Doubly linked list for LRU ordering
+	ttl      time.Duration       // Time-to-live for cache entries (0 = no expiry)
+	stats    Stats               // Cache statistics
+}
+
+// cacheEntry represents an item in the cache
+type cacheEntry[K comparable, V any] struct {
+	key       K
+	value     V
+	expiresAt time.Time // When this entry expires
+}
+
+// Stats holds cache statistics
+type Stats struct {
+	Hits        int64 // Number of cache hits
+	Misses      int64 // Number of cache misses
+	Evictions   int64 // Number of items evicted
+	Inserts     int64 // Number of items inserted
+	Expirations int64 // Number of items expired
+}
+
+// NewLRUCache creates a new LRU cache with the specified capacity and TTL
+// If ttl is 0, entries never expire
+func NewLRUCache[K comparable, V any](capacity int, ttl time.Duration) *LRUCache[K, V] {
+	if capacity <= 0 {
+		capacity = 100 // Default capacity
+	}
+
+	return &LRUCache[K, V]{
+		capacity: capacity,
+		cache:    make(map[K]*list.Element),
+		list:     list.New(),
+		ttl:      ttl,
+		stats:    Stats{},
+	}
+}
+
+// Get retrieves a value from the cache by key
+// Returns the value and true if found, otherwise zero value and false
+func (c *LRUCache[K, V]) Get(key K) (V, bool) {
+	c.mu.RLock()
+	element, exists := c.cache[key]
+	c.mu.RUnlock()
+
+	if !exists {
+		c.mu.Lock()
+		c.stats.Misses++
+		c.mu.Unlock()
+		var zero V
+		return zero, false
+	}
+
+	c.mu.Lock()
+	defer c.mu.Unlock()
+
+	entry := element.Value.(*cacheEntry[K, V])
+
+	// Check if entry has expired
+	if c.ttl > 0 && time.Now().After(entry.expiresAt) {
+		c.removeElement(element)
+		c.stats.Misses++
+		c.stats.Expirations++
+		var zero V
+		return zero, false
+	}
+
+	// Move to front (most recently used)
+	c.list.MoveToFront(element)
+	c.stats.Hits++
+
+	return entry.value, true
+}
+
+// Put adds or updates a value in the cache
+func (c *LRUCache[K, V]) Put(key K, value V) {
+	c.mu.Lock()
+	defer c.mu.Unlock()
+
+	// Check if key already exists
+	if element, exists := c.cache[key]; exists {
+		// Update existing entry
+		entry := element.Value.(*cacheEntry[K, V])
+		entry.value = value
+		if c.ttl > 0 {
+			entry.expiresAt = time.Now().Add(c.ttl)
+		}
+		c.list.MoveToFront(element)
+		return
+	}
+
+	// Create new entry
+	entry := &cacheEntry[K, V]{
+		key:   key,
+		value: value,
+	}
+	if c.ttl > 0 {
+		entry.expiresAt = time.Now().Add(c.ttl)
+	}
+
+	// Add to cache
+	element := c.list.PushFront(entry)
+	c.cache[key] = element
+	c.size++
+	c.stats.Inserts++
+
+	// Evict if over capacity
+	if c.size > c.capacity {
+		c.evictOldest()
+	}
+}
+
+// Remove deletes a key from the cache
+func (c *LRUCache[K, V]) Remove(key K) bool {
+	c.mu.Lock()
+	defer c.mu.Unlock()
+
+	element, exists := c.cache[key]
+	if !exists {
+		return false
+	}
+
+	c.removeElement(element)
+	return true
+}
+
+// removeElement removes an element from the cache
+func (c *LRUCache[K, V]) removeElement(element *list.Element) {
+	entry := element.Value.(*cacheEntry[K, V])
+	c.list.Remove(element)
+	delete(c.cache, entry.key)
+	c.size--
+}
+
+// evictOldest removes the least recently used item
+func (c *LRUCache[K, V]) evictOldest() {
+	element := c.list.Back()
+	if element != nil {
+		c.removeElement(element)
+		c.stats.Evictions++
+	}
+}
+
+// Clear removes all items from the cache
+func (c *LRUCache[K, V]) Clear() {
+	c.mu.Lock()
+	defer c.mu.Unlock()
+
+	c.cache = make(map[K]*list.Element)
+	c.list.Init()
+	c.size = 0
+}
+
+// Len returns the current number of items in the cache
+func (c *LRUCache[K, V]) Len() int {
+	c.mu.RLock()
+	defer c.mu.RUnlock()
+	return c.size
+}
+
+// Capacity returns the cache capacity
+func (c *LRUCache[K, V]) Capacity() int {
+	return c.capacity
+}
+
+// Stats returns current cache statistics
+func (c *LRUCache[K, V]) Stats() Stats {
+	c.mu.RLock()
+	defer c.mu.RUnlock()
+	return c.stats
+}
+
+// ResetStats resets all statistics counters
+func (c *LRUCache[K, V]) ResetStats() {
+	c.mu.Lock()
+	defer c.mu.Unlock()
+	c.stats = Stats{}
+}
+
+// Keys returns all keys in the cache (from most to least recently used)
+func (c *LRUCache[K, V]) Keys() []K {
+	c.mu.RLock()
+	defer c.mu.RUnlock()
+
+	keys := make([]K, 0, c.size)
+	for element := c.list.Front(); element != nil; element = element.Next() {
+		entry := element.Value.(*cacheEntry[K, V])
+		keys = append(keys, entry.key)
+	}
+	return keys
+}
+
+// CleanupExpired removes all expired entries from the cache
+func (c *LRUCache[K, V]) CleanupExpired() int {
+	if c.ttl <= 0 {
+		return 0
+	}
+
+	c.mu.Lock()
+	defer c.mu.Unlock()
+
+	expiredCount := 0
+	now := time.Now()
+
+	// Collect expired elements
+	var expiredElements []*list.Element
+	for element := c.list.Back(); element != nil; element = element.Prev() {
+		entry := element.Value.(*cacheEntry[K, V])
+		if now.After(entry.expiresAt) {
+			expiredElements = append(expiredElements, element)
+		}
+	}
+
+	// Remove expired elements
+	for _, element := range expiredElements {
+		c.removeElement(element)
+		expiredCount++
+	}
+
+	c.stats.Expirations += int64(expiredCount)
+	return expiredCount
+}

model/model_db.go

@@ -9,6 +9,7 @@ import (
 
 var EvmTables = []any{
 	&Address{}, &Block{}, &Tx{}, &Trace{}, &ContractLifecycle{},
+	&Topic{}, &LogParam{}, &Log{},
 }
 
 type Model struct {
@@ -69,6 +70,29 @@ type ContractLifecycle struct {
 	Value               decimal.Decimal `gorm:"type:decimal(36,18);not null" json:"value"`
 }
 
+type Topic struct {
+	Model
+	Topic string `gorm:"size:66;not null;unique" json:"topic"`
+}
+type LogParam struct {
+	Model
+	Hex string `gorm:"size:66;not null;unique" json:"topic"`
+}
+
+type Log struct {
+	Model
+	BlockId    uint64 `gorm:"not null" json:"blockId"`
+	TxIndex    uint   `gorm:"not null" json:"txIndex"`
+	TxId       uint64 `gorm:"not null" json:"txId"`
+	ContractId uint64 `gorm:"not null" json:"contractId"`
+	TopicId    uint64 `gorm:"not null" json:"topicId"`
+	// refer to LogParam
+	Param1 uint64 `gorm:"not null" json:"param1"`
+	Param2 uint64 `gorm:"not null" json:"param2"`
+	Param3 uint64 `gorm:"not null" json:"param3"`
+	Count  uint   `gorm:"not null" json:"count"`
+}
+
 const (
 	ContractLifecycleCreate  = "create"
 	ContractLifecycleDestroy = "destroy"