Migrate listener observed block from redis to db

api/listener/controller.go

@@ -34,6 +34,11 @@ type ListenerInsertModel struct {
 	Chain     string `db:"chain_name" json:"chain" validate:"required"`
 }
 
+type ObservedBlockModel struct {
+	BlockKey	string 	`db:"block_key" json:"blockKey" validate:"required"`
+	BlockNumber	int64 	`db:"block_number" json:"blockNumber" validate:"blockNumberValidator"`
+}
+
 func insert(c *fiber.Ctx) error {
 	payload := new(ListenerInsertModel)
 	if err := c.BodyParser(payload); err != nil {
@@ -151,3 +156,33 @@ func deleteById(c *fiber.Ctx) error {
 
 	return c.JSON(result)
 }
+
+func upsertObservedBlock(c *fiber.Ctx) error {
+	payload := new(ObservedBlockModel)
+	if err := c.BodyParser(payload); err != nil {
+		return err
+	}
+
+	validate := validator.New()
+	validate.RegisterValidation("blockNumberValidator", func(fl validator.FieldLevel) bool {
+		return fl.Field().Int() >= 0
+	})
+	if err := validate.Struct(payload); err != nil {
+		return err
+	}
+
+	result, err := utils.QueryRow[ObservedBlockModel](c, UpsertObservedBlock, map[string]any{"block_key": payload.BlockKey, "block_number": payload.BlockNumber})
+	if err != nil {
+		return err
+	}
+	return c.JSON(result)
+}
+
+func getObservedBlock(c *fiber.Ctx) error {
+	blockKey := c.Query("blockKey")
+	result, err := utils.QueryRow[ObservedBlockModel](c, GetObservedBlock, map[string]any{"block_key": blockKey})
+	if err != nil {
+		return err
+	}
+	return c.JSON(result)
+}

api/listener/queries.go

@@ -42,6 +42,20 @@ const (
 		(SELECT name from chains WHERE chain_id = listeners.chain_id) AS chain_name,
 		(SELECT name from services WHERE service_id = listeners.service_id) AS service_name;
 		`
+
+	GetObservedBlock = `
+		SELECT * FROM observed_blocks
+		WHERE block_key = @block_key
+		LIMIT 1;
+	`
+
+	UpsertObservedBlock = `
+		INSERT INTO observed_blocks (block_key, block_number)
+		VALUES (@block_key, @block_number)
+		ON CONFLICT (block_key) 
+		DO UPDATE SET block_number = @block_number
+		RETURNING *;
+	`
 )
 
 func GenerateGetListenerQuery(params GetListenerQueryParams) string {

api/listener/route.go

@@ -9,6 +9,8 @@ func Routes(router fiber.Router) {
 
 	listener.Post("", insert)
 	listener.Get("", get)
+	listener.Post("/observed-block", upsertObservedBlock)
+	listener.Get("/observed-block", getObservedBlock)
 	listener.Get("/:id", getById)
 	listener.Patch("/:id", updateById)
 	listener.Delete("/:id", deleteById)

api/migrations/000002_add_observed_block.down.sql

@@ -0,0 +1 @@
+DROP TABLE IF EXISTS "observed_blocks";

api/migrations/000002_add_observed_block.up.sql

@@ -0,0 +1,5 @@
+CREATE TABLE IF NOT EXISTS "observed_blocks" (
+    block_key TEXT NOT NULL,
+    block_number BIGINT NOT NULL,
+    CONSTRAINT "observed_blocks_key" UNIQUE ("block_key")
+)

core/src/errors.ts

@@ -73,5 +73,7 @@ export enum OraklErrorCode {
   AxiosCanceledByUser,
   AxiosNotSupported,
   AxiosInvalidUrl,
-  FailedToConnectAPI
+  FailedToConnectAPI,
+  UpsertObservedBlockFailed,
+  GetObservedBlockFailed
 }

core/src/listener/api.ts

@@ -4,6 +4,7 @@ import { OraklError, OraklErrorCode } from '../errors'
 import { ORAKL_NETWORK_API_URL } from '../settings'
 import { IListenerRawConfig } from '../types'
 import { buildUrl } from '../utils'
+import { IObservedBlock } from './types'
 
 const FILE_NAME = import.meta.url
 
@@ -58,3 +59,54 @@ export async function getListener({
     throw new OraklError(OraklErrorCode.GetListenerRequestFailed)
   }
 }
+
+/**
+ * Get observed block number from the Orakl Network API for a given contract address
+ *
+ * @param {string} blockKey
+ * @param {pino.Logger} logger
+ * @return {Promise<IObservedBlock>}
+ * @exception {OraklErrorCode.GetObservedBlockFailed}
+ */
+export async function getObservedBlock({
+  blockKey,
+  logger
+}: {
+  blockKey: string
+  logger?: Logger
+}): Promise<IObservedBlock> {
+  try {
+    const endpoint = buildUrl(ORAKL_NETWORK_API_URL, `listener/observed-block?blockKey=${blockKey}`)
+    return (await axios.get(endpoint))?.data
+  } catch (e) {
+    logger?.error({ name: 'getObservedBlock', file: FILE_NAME, ...e }, 'error')
+    throw new OraklError(OraklErrorCode.GetObservedBlockFailed)
+  }
+}
+
+/**
+ * Upsert listener observed block number to the Orakl Network API for a given contract address
+ *
+ * @param {string} blockKey
+ * @param {number} blockNumber
+ * @param {pino.Logger} logger
+ * @return {Promise<IObservedBlock>}
+ * @exception {OraklErrorCode.UpsertObservedBlockFailed}
+ */
+export async function upsertObservedBlock({
+  blockKey,
+  blockNumber,
+  logger
+}: {
+  blockKey: string
+  blockNumber: number
+  logger?: Logger
+}): Promise<IObservedBlock> {
+  try {
+    const endpoint = buildUrl(ORAKL_NETWORK_API_URL, 'listener/observed-block')
+    return (await axios.post(endpoint, { blockKey, blockNumber }))?.data
+  } catch (e) {
+    logger?.error({ name: 'upsertObservedBlock', file: FILE_NAME, ...e }, 'error')
+    throw new OraklError(OraklErrorCode.UpsertObservedBlockFailed)
+  }
+}

core/src/listener/data-feed-L2.ts

@@ -10,7 +10,6 @@ import {
   L2_DATA_FEED_SERVICE_NAME,
   L2_LISTENER_DATA_FEED_HISTORY_QUEUE_NAME,
   L2_LISTENER_DATA_FEED_LATEST_QUEUE_NAME,
-  L2_LISTENER_DATA_FEED_PROCESS_EVENT_QUEUE_NAME,
   L2_WORKER_AGGREGATOR_QUEUE_NAME
 } from '../settings'
 import { IAnswerUpdated, IDataFeedListenerWorkerL2, IListenerConfig } from '../types'
@@ -30,7 +29,6 @@ export async function buildListener(
   const eventName = 'AnswerUpdated'
   const latestQueueName = L2_LISTENER_DATA_FEED_LATEST_QUEUE_NAME
   const historyQueueName = L2_LISTENER_DATA_FEED_HISTORY_QUEUE_NAME
-  const processEventQueueName = L2_LISTENER_DATA_FEED_PROCESS_EVENT_QUEUE_NAME
   const workerQueueName = L2_WORKER_AGGREGATOR_QUEUE_NAME
   const abi = Aggregator__factory.abi
   const iface = new ethers.utils.Interface(abi)
@@ -44,11 +42,9 @@ export async function buildListener(
     eventName,
     latestQueueName,
     historyQueueName,
-    processEventQueueName,
     workerQueueName,
     processFn: await processEvent({ iface, logger }),
     redisClient,
-    listenerInitType: 'latest',
     logger
   })
 }

core/src/listener/data-feed.ts

@@ -57,11 +57,9 @@ export async function buildListener(
     eventName,
     latestQueueName,
     historyQueueName,
-    processEventQueueName,
     workerQueueName,
     processFn: await processEvent({ iface, logger }),
     redisClient,
-    listenerInitType: 'latest',
     logger
   })
 }