feat: reorg detector add warnings and store in DB the reorg detection

reorgdetector/migrations/migrations.go

@@ -10,12 +10,19 @@ import (
 //go:embed reorgdetector0001.sql
 var mig001 string
 
+//go:embed reorgdetector0002.sql
+var mig002 string
+
 func RunMigrations(dbPath string) error {
 	migrations := []types.Migration{
 		{
 			ID:  "reorgdetector0001",
 			SQL: mig001,
 		},
+		{
+			ID:  "reorgdetector0002",
+			SQL: mig002,
+		},
 	}
 	return db.RunMigrations(dbPath, migrations)
 }

reorgdetector/migrations/reorgdetector0002.sql

@@ -0,0 +1,14 @@
+-- +migrate Down
+DROP TABLE IF EXISTS reorg_event;
+
+-- +migrate Up
+CREATE TABLE reorg_event (
+    detected_at  BIGINT NOT NULL,
+    from_block  BIGINT NOT NULL,
+    to_block    BIGINT NOT NULL,
+    subscriber_id  VARCHAR,
+    current_hash VARCHAR,
+    tracked_hash VARCHAR,
+    version string,
+    extra_data  VARCHAR
+)

reorgdetector/reorgdetector.go

@@ -176,7 +176,7 @@ func (rd *ReorgDetector) detectReorgInTrackedList(ctx context.Context) error {
 	)
 
 	subscriberIDs := rd.getSubscriberIDs()
-
+	startTime := time.Now()
 	for _, id := range subscriberIDs {
 		id := id
 
@@ -223,17 +223,35 @@ func (rd *ReorgDetector) detectReorgInTrackedList(ctx context.Context) error {
 
 					continue
 				}
-
+				event := ReorgEvent{
+					DetectedAt:   startTime,
+					FromBlock:    hdr.Num,
+					ToBlock:      headers[len(headers)-1].Num,
+					SubscriberID: id,
+					CurrentHash:  currentHeader.Hash(),
+					TrackedHash:  hdr.Hash,
+					ExtraData: struct {
+						Network        string
+						BlockTimestamp uint64
+					}{
+						Network:        rd.network.String(),
+						BlockTimestamp: currentHeader.Time,
+					},
+				}
+				if err := rd.insertReorgEvent(event); err != nil {
+					return fmt.Errorf("failed to insert reorg event: %w", err)
+				}
+				rd.log.Warnf("Reorg detected %s for subscriber %s between blocks %d and %d. currentHash: %s trackHash: %s",
+					rd.network, event.SubscriberID, event.FromBlock, event.ToBlock, event.CurrentHash, event.TrackedHash)
 				// Notify the subscriber about the reorg
 				rd.notifySubscriber(id, hdr)
-
 				// Remove the reorged block and all the following blocks from DB
-				if err := rd.removeTrackedBlockRange(id, hdr.Num, headers[len(headers)-1].Num); err != nil {
+				if err := rd.removeTrackedBlockRange(event.SubscriberID, event.FromBlock, event.ToBlock); err != nil {
 					return fmt.Errorf("error removing blocks from DB for subscriber %s between blocks %d and %d: %w",
-						id, hdr.Num, headers[len(headers)-1].Num, err)
+						event.SubscriberID, event.FromBlock, event.ToBlock, err)
 				}
 				// Remove the reorged block and all the following blocks from memory
-				hdrs.removeRange(hdr.Num, headers[len(headers)-1].Num)
+				hdrs.removeRange(event.FromBlock, event.ToBlock)
 
 				break
 			}

reorgdetector/reorgdetector_db.go

@@ -1,10 +1,14 @@
 package reorgdetector
 
 import (
+	"encoding/json"
 	"errors"
 	"fmt"
+	"time"
 
+	"github.com/agglayer/aggkit"
 	"github.com/agglayer/aggkit/db"
+	common "github.com/ethereum/go-ethereum/common"
 	"github.com/russross/meddler"
 )
 
@@ -78,3 +82,42 @@ func (rd *ReorgDetector) removeTrackedBlockRange(id string, fromBlock, toBlock u
 	)
 	return err
 }
+
+type ReorgEvent struct {
+	DetectedAt   time.Time
+	FromBlock    uint64
+	ToBlock      uint64
+	SubscriberID string
+	TrackedHash  common.Hash
+	CurrentHash  common.Hash
+	ExtraData    interface{}
+}
+
+type eventReorgRow struct {
+	DetectedAt   int64  `meddler:"detected_at"`
+	FromBlock    uint64 `meddler:"from_block"`
+	ToBlock      uint64 `meddler:"to_block"`
+	SubscriberID string `meddler:"subscriber_id"`
+	TrackedHash  string `meddler:"tracked_hash"`
+	CurrentHash  string `meddler:"current_hash"`
+	Version      string `meddler:"version"`
+	ExtraData    string `meddler:"extra_data"`
+}
+
+func (rd *ReorgDetector) insertReorgEvent(event ReorgEvent) error {
+	extra, err := json.Marshal(event.ExtraData)
+	if err != nil {
+		return fmt.Errorf("failed to marshal extra data: %w", err)
+	}
+	row := eventReorgRow{
+		DetectedAt:   event.DetectedAt.Unix(),
+		FromBlock:    event.FromBlock,
+		ToBlock:      event.ToBlock,
+		SubscriberID: event.SubscriberID,
+		TrackedHash:  event.TrackedHash.String(),
+		CurrentHash:  event.CurrentHash.String(),
+		Version:      aggkit.GetVersion().Brief(),
+		ExtraData:    string(extra),
+	}
+	return meddler.Insert(rd.db, "reorg_event", &row)
+}

version.go

@@ -52,3 +52,9 @@ func (f FullVersion) String() string {
 		f.Version, f.GitRev, f.GitBranch,
 		f.GoVersion, f.BuildDate, f.OS, f.Arch)
 }
+
+func (f FullVersion) Brief() string {
+	return fmt.Sprintf("%s - %s / %s - build:%s os:%s/%s",
+		f.Version, f.GitRev, f.GitBranch,
+		f.BuildDate, f.OS, f.Arch)
+}