Feature: add gas_price websocket channel

cmd/api/bus/dispatcher.go

@@ -89,7 +89,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 {

cmd/api/gas/tracker.go

@@ -29,6 +29,8 @@ var (
  maxBlockSize = coreTypes.MaxDataBytesNoEvidence(1974272, 100)
 )
 
+type ComputeHandler func(ctx context.Context, gasState GasPrice) error
+
 type Tracker struct {
  state storage.IState
  stats storage.IBlockStats
@@ -39,6 +41,8 @@ type Tracker struct {
  gasState GasPrice
  q *queue
  g workerpool.Group
+
+ computeHandler ComputeHandler
 }
 
 func NewTracker(
@@ -64,6 +68,10 @@ func NewTracker(
  }
 }
 
+func (tracker *Tracker) SubscribeOnCompute(handler ComputeHandler) {
+ tracker.computeHandler = handler
+}
+
 func (tracker *Tracker) Start(ctx context.Context) {
  tracker.g.GoCtx(ctx, tracker.listen)
 }
@@ -90,6 +98,14 @@ func (tracker *Tracker) listen(ctx context.Context) {
  if err := tracker.computeMetrics(); err != nil {
  log.Err(err).Msg("compute metrics")
  }
+
+ if tracker.computeHandler != nil {
+ tracker.g.GoCtx(ctx, func(ctx context.Context) {
+ if err := tracker.computeHandler(ctx, tracker.gasState); err != nil {
+ log.Err(err).Msg("error in compute handler of gas tracker")
+ }
+ })
+ }
  }
  }
 }

cmd/api/handler/websocket/client.go

@@ -82,6 +82,8 @@ func (c *Client) ApplyFilters(msg Subscribe) error {
  c.filters.head = true
  case ChannelBlocks:
  c.filters.blocks = true
+ case ChannelGasPrice:
+ c.filters.gasPrice = true
  default:
  return errors.Wrap(ErrUnknownChannel, msg.Channel)
  }
@@ -97,6 +99,8 @@ func (c *Client) DetachFilters(msg Unsubscribe) error {
  c.filters.head = false
  case ChannelBlocks:
  c.filters.blocks = false
+ case ChannelGasPrice:
+ c.filters.gasPrice = false
  default:
  return errors.Wrap(ErrUnknownChannel, msg.Channel)
  }

cmd/api/handler/websocket/filters.go

@@ -37,7 +37,21 @@ func (f HeadFilter) Filter(c client, msg Notification[*responses.State]) bool {
  return fltrs.head
 }
 
+type GasPriceFilter struct{}
+
+func (f GasPriceFilter) Filter(c client, msg Notification[*responses.GasPrice]) bool {
+ if msg.Body == nil {
+ return false
+ }
+ fltrs := c.Filters()
+ if fltrs == nil {
+ return false
+ }
+ return fltrs.gasPrice
+}
+
 type Filters struct {
- head bool
- blocks bool
+ head bool
+ blocks bool
+ gasPrice bool
 }

cmd/api/handler/websocket/manager.go

@@ -12,6 +12,7 @@ import (
  sdkSync "github.com/dipdup-net/indexer-sdk/pkg/sync"
 
  "github.com/celenium-io/celestia-indexer/cmd/api/bus"
+ "github.com/celenium-io/celestia-indexer/cmd/api/gas"
  "github.com/celenium-io/celestia-indexer/cmd/api/handler/responses"
  "github.com/celenium-io/celestia-indexer/internal/storage"
  "github.com/gorilla/websocket"
@@ -25,8 +26,9 @@ type Manager struct {
  clients *sdkSync.Map[uint64, *Client]
  observer *bus.Observer
 
- blocks *Channel[storage.Block, *responses.Block]
- head *Channel[storage.State, *responses.State]
+ blocks *Channel[storage.Block, *responses.Block]
+ head *Channel[storage.State, *responses.State]
+ gasPrice *Channel[gas.GasPrice, *responses.GasPrice]
 
  g workerpool.Group
 }
@@ -56,6 +58,11 @@ func NewManager(observer *bus.Observer) *Manager {
  HeadFilter{},
  )
 
+ manager.gasPrice = NewChannel(
+ gasPriceProcessor,
+ GasPriceFilter{},
+ )
+
  return manager
 }
 
@@ -131,6 +138,8 @@ func (manager *Manager) AddClientToChannel(channel string, client *Client) {
  manager.head.AddClient(client)
  case ChannelBlocks:
  manager.blocks.AddClient(client)
+ case ChannelGasPrice:
+ manager.gasPrice.AddClient(client)
  default:
  log.Error().Str("channel", channel).Msg("unknown channel name")
  }
@@ -142,7 +151,13 @@ func (manager *Manager) RemoveClientFromChannel(channel string, client *Client)
  manager.head.RemoveClient(client.id)
  case ChannelBlocks:
  manager.blocks.RemoveClient(client.id)
+ case ChannelGasPrice:
+ manager.gasPrice.RemoveClient(client.id)
  default:
  log.Error().Str("channel", channel).Msg("unknown channel name")
  }
 }
+
+func (manager *Manager) GasTrackerHandler(_ context.Context, state gas.GasPrice) error {
+ return manager.gasPrice.processMessage(state)
+}

cmd/api/handler/websocket/messages.go

@@ -16,8 +16,9 @@ const (
 
 // channels
 const (
- ChannelHead = "head"
- ChannelBlocks = "blocks"
+ ChannelHead = "head"
+ ChannelBlocks = "blocks"
+ ChannelGasPrice = "gas_price"
 )
 
 type Message struct {
@@ -40,7 +41,7 @@ type TransactionFilters struct {
 }
 
 type INotification interface {
- *responses.Block | *responses.State
+ *responses.Block | *responses.State | *responses.GasPrice
 }
 
 type Notification[T INotification] struct {
@@ -61,3 +62,10 @@ func NewStateNotification(state responses.State) Notification[*responses.State]
  Body: &state,
  }
 }
+
+func NewGasPriceNotification(value responses.GasPrice) Notification[*responses.GasPrice] {
+ return Notification[*responses.GasPrice]{
+ Channel: ChannelGasPrice,
+ Body: &value,
+ }
+}

cmd/api/handler/websocket/processors.go

@@ -4,6 +4,7 @@
 package websocket
 
 import (
+ "github.com/celenium-io/celestia-indexer/cmd/api/gas"
  "github.com/celenium-io/celestia-indexer/cmd/api/handler/responses"
  "github.com/celenium-io/celestia-indexer/internal/storage"
 )
@@ -17,3 +18,11 @@ func headProcessor(state storage.State) Notification[*responses.State] {
  response := responses.NewState(state)
  return NewStateNotification(response)
 }
+
+func gasPriceProcessor(data gas.GasPrice) Notification[*responses.GasPrice] {
+ return NewGasPriceNotification(responses.GasPrice{
+ Slow: data.Slow,
+ Median: data.Median,
+ Fast: data.Fast,
+ })
+}

cmd/api/init.go

@@ -555,6 +555,9 @@ var (
 func initWebsocket(ctx context.Context, group *echo.Group) {
  observer := dispatcher.Observe(storage.ChannelHead, storage.ChannelBlock)
  wsManager = websocket.NewManager(observer)
+ if gasTracker != nil {
+ gasTracker.SubscribeOnCompute(wsManager.GasTrackerHandler)
+ }
  wsManager.Start(ctx)
  group.GET("/ws", wsManager.Handle)
 }

cmd/api/markdown/websocket.md

@@ -27,7 +27,7 @@ To receive updates from websocket API send `subscribe` request to server.
 }
 ```
 
-Now 2 channels are supported:
+Now 3 channels are supported:
 
 * `head` - receive information about indexer state. Channel does not have any filters. Subscribe message should looks like:
 
@@ -55,6 +55,19 @@ Notification body of `responses.State` type will be sent to the channel.
 
 Notification body of `responses.Block` type will be sent to the channel.
 
+* `gas_price` - receive information about current gas price. Channel does not have any filters. Subscribe message should looks like:
+
+```json
+{
+ "method": "subscribe",
+ "body": {
+ "channel": "gas_price"
+ }
+}
+```
+
+Notification body of `responses.GasPrice` type will be sent to the channel.
+
 
 ### Unsubscribe
 

pkg/indexer/storage/storage.go

@@ -335,10 +335,10 @@ func (module *Module) processBlockInTransaction(ctx context.Context, tx storage.
 }
 
 func (module *Module) notify(ctx context.Context, state storage.State, block storage.Block) error {
- if time.Since(block.Time) > time.Hour {
- // do not notify all about events if initial indexing is in progress
- return nil
- }
+ // if time.Since(block.Time) > time.Hour {
+ //  // do not notify all about events if initial indexing is in progress
+ //  return nil
+ // }
 
  rawState, err := jsoniter.MarshalToString(state)
  if err != nil {