Feature: sudo flags in address

cmd/api/bus/dispatcher.go

@@ -87,7 +87,7 @@ func (d *Dispatcher) listen(ctx context.Context) {
 				return
 			}
 			if notification == nil {
-				log.Warn().Str("channel", notification.Channel).Msg("nil notification")
+				log.Warn().Msg("nil notification")
 				continue
 			}
 			if err := d.handleNotification(ctx, notification); err != nil {
@@ -107,7 +107,9 @@ func (d *Dispatcher) handleNotification(ctx context.Context, notification *pq.No
 
 		return d.handleBlock(ctx, id)
 	case storage.ChannelHead:
-		return d.handleHead(ctx, notification.Extra)
+		return d.handleHead(notification.Extra)
+	case storage.ChannelConstant:
+		return d.handleConstant(notification.Extra)
 	default:
 		return errors.Errorf("unknown channel name: %s", notification.Channel)
 	}
@@ -126,7 +128,7 @@ func (d *Dispatcher) handleBlock(ctx context.Context, id uint64) error {
 	return nil
 }
 
-func (d *Dispatcher) handleHead(ctx context.Context, msg string) error {
+func (d *Dispatcher) handleHead(msg string) error {
 	var state storage.State
 	if err := json.Unmarshal([]byte(msg), &state); err != nil {
 		return err
@@ -139,3 +141,17 @@ func (d *Dispatcher) handleHead(ctx context.Context, msg string) error {
 	d.mx.RUnlock()
 	return nil
 }
+
+func (d *Dispatcher) handleConstant(msg string) error {
+	var c storage.Constant
+	if err := json.Unmarshal([]byte(msg), &c); err != nil {
+		return err
+	}
+
+	d.mx.RLock()
+	for i := range d.observers {
+		d.observers[i].notifyConstants(&c)
+	}
+	d.mx.RUnlock()
+	return nil
+}

cmd/api/bus/observer.go

@@ -9,11 +9,13 @@ import (
 )
 
 type Observer struct {
-	blocks chan *storage.Block
-	head   chan *storage.State
+	blocks    chan *storage.Block
+	head      chan *storage.State
+	constants chan *storage.Constant
 
-	listenHead   bool
-	listenBlocks bool
+	listenHead      bool
+	listenBlocks    bool
+	listenConstants bool
 
 	g workerpool.Group
 }
@@ -24,9 +26,10 @@ func NewObserver(channels ...string) *Observer {
 	}
 
 	observer := &Observer{
-		blocks: make(chan *storage.Block, 1024),
-		head:   make(chan *storage.State, 1024),
-		g:      workerpool.NewGroup(),
+		blocks:    make(chan *storage.Block, 1024),
+		head:      make(chan *storage.State, 1024),
+		constants: make(chan *storage.Constant, 1024),
+		g:         workerpool.NewGroup(),
 	}
 
 	for i := range channels {
@@ -35,6 +38,8 @@ func NewObserver(channels ...string) *Observer {
 			observer.listenBlocks = true
 		case storage.ChannelHead:
 			observer.listenHead = true
+		case storage.ChannelConstant:
+			observer.listenConstants = true
 		}
 	}
 
@@ -45,6 +50,7 @@ func (observer Observer) Close() error {
 	observer.g.Wait()
 	close(observer.blocks)
 	close(observer.head)
+	close(observer.constants)
 	return nil
 }
 
@@ -58,6 +64,11 @@ func (observer Observer) notifyState(state *storage.State) {
 		observer.head <- state
 	}
 }
+func (observer Observer) notifyConstants(constant *storage.Constant) {
+	if observer.listenConstants {
+		observer.constants <- constant
+	}
+}
 
 func (observer Observer) Blocks() <-chan *storage.Block {
 	return observer.blocks
@@ -66,3 +77,7 @@ func (observer Observer) Blocks() <-chan *storage.Block {
 func (observer Observer) Head() <-chan *storage.State {
 	return observer.head
 }
+
+func (observer Observer) Constants() <-chan *storage.Constant {
+	return observer.constants
+}

cmd/api/cache/constants.go

@@ -0,0 +1,88 @@
+// SPDX-FileCopyrightText: 2024 PK Lab AG <[email protected]>
+// SPDX-License-Identifier: MIT
+
+package cache
+
+import (
+	"context"
+	"sync"
+
+	"github.com/celenium-io/astria-indexer/cmd/api/bus"
+	"github.com/celenium-io/astria-indexer/internal/storage"
+	"github.com/celenium-io/astria-indexer/internal/storage/types"
+)
+
+type ConstantsCache struct {
+	data     map[string]map[string]string
+	observer *bus.Observer
+
+	wg *sync.WaitGroup
+	mx *sync.RWMutex
+}
+
+func NewConstantsCache(observer *bus.Observer) *ConstantsCache {
+	return &ConstantsCache{
+		data:     make(map[string]map[string]string),
+		observer: observer,
+		wg:       new(sync.WaitGroup),
+		mx:       new(sync.RWMutex),
+	}
+}
+
+func (c *ConstantsCache) Start(ctx context.Context, repo storage.IConstant) error {
+	constants, err := repo.All(ctx)
+	if err != nil {
+		return err
+	}
+
+	for i := range constants {
+		c.addConstant(&constants[i])
+	}
+
+	c.wg.Add(1)
+	go c.listen(ctx)
+
+	return nil
+}
+
+func (c *ConstantsCache) listen(ctx context.Context) {
+	defer c.wg.Done()
+
+	for {
+		select {
+		case <-ctx.Done():
+			return
+		case constant := <-c.observer.Constants():
+			c.addConstant(constant)
+		}
+	}
+}
+
+func (c *ConstantsCache) addConstant(constant *storage.Constant) {
+	c.mx.Lock()
+	{
+		module := string(constant.Module)
+		if _, ok := c.data[module]; !ok {
+			c.data[module] = make(map[string]string)
+		}
+		c.data[module][constant.Name] = constant.Value
+	}
+	c.mx.Unlock()
+}
+
+func (c *ConstantsCache) Get(module types.ModuleName, name string) (string, bool) {
+	c.mx.RLock()
+	defer c.mx.RUnlock()
+
+	if m, ok := c.data[string(module)]; ok {
+		val, ok := m[name]
+		return val, ok
+	}
+
+	return "", false
+}
+
+func (c *ConstantsCache) Close() error {
+	c.wg.Wait()
+	return nil
+}