feat: add slot status table and start tracking missed slots

.gitignore

@@ -1,4 +1,5 @@
 build
 .envrc
 
-*.txt
+*.txt
+.env

internal/data/sql/queries/metrics.sql

@@ -208,3 +208,14 @@ WITH upserted AS (
 )
 SELECT EXISTS (SELECT 1 FROM upserted) AS did_upsert;
 
+-- name: CreateSlotStatus :exec
+INSERT INTO slot_status (
+	slot,
+	status
+)
+VALUES ($1, $2);
+
+-- name: QueryLatestSlotStatus :one
+SELECT slot, status FROM slot_status
+ORDER BY slot DESC
+LIMIT 1;

internal/metrics/tx_mapper_db.go

@@ -206,6 +206,42 @@ func (tm *TxMapperDB) AddBlock(
 	return err
 }
 
+func (tm *TxMapperDB) AddSlotStatus(
+	ctx context.Context,
+	slot int64,
+	status data.SlotStatusVal,
+) error {
+	latest, err := tm.dbQuery.QueryLatestSlotStatus(ctx)
+	if err != nil {
+		if errors.Is(err, pgx.ErrNoRows) {
+			return tm.dbQuery.CreateSlotStatus(ctx, data.CreateSlotStatusParams{
+				Slot:   slot,
+				Status: status,
+			})
+		}
+		return err
+	}
+
+	if slot <= latest.Slot {
+		return nil
+	}
+
+	for s := latest.Slot + 1; s < slot; s++ {
+		err := tm.dbQuery.CreateSlotStatus(ctx, data.CreateSlotStatusParams{
+			Slot:   s,
+			Status: data.SlotStatusValMissed,
+		})
+		if err != nil {
+			return err
+		}
+	}
+
+	return tm.dbQuery.CreateSlotStatus(ctx, data.CreateSlotStatusParams{
+		Slot:   slot,
+		Status: status,
+	})
+}
+
 func (tm *TxMapperDB) QueryBlockNumberFromValidatorRegistryEventsSyncedUntil(ctx context.Context) (int64, error) {
 	data, err := tm.dbQuery.QueryValidatorRegistryEventsSyncedUntil(ctx)
 	if err != nil {

internal/metrics/types.go

@@ -61,6 +61,7 @@ type TxMapper interface {
 		ctx context.Context,
 		b *data.Block,
 	) error
+	AddSlotStatus(ctx context.Context, slot int64, status data.SlotStatusVal) error
 	QueryBlockNumberFromValidatorRegistryEventsSyncedUntil(ctx context.Context) (int64, error)
 	AddValidatorRegistryEvent(ctx context.Context, tx pgx.Tx, vr *validatorRegistryBindings.ValidatorregistryUpdated) error
 	UpdateValidatorStatus(ctx context.Context) error

internal/watcher/blocks.go

@@ -147,16 +147,25 @@ func (bw *BlocksWatcher) insertBlock(ctx context.Context, header *types.Header)
 		bw.mostRecentBlock = header.Number.Uint64()
 	}
 
+	slot := int64(utils.GetSlotForBlock(header.Time, GenesisTimestamp, SlotDuration))
+
 	err := bw.txMapper.AddBlock(ctx, &data.Block{
 		BlockHash:      header.Hash().Bytes(),
 		BlockNumber:    header.Number.Int64(),
 		BlockTimestamp: int64(header.Time),
-		Slot:           int64(utils.GetSlotForBlock(header.Time, GenesisTimestamp, SlotDuration)),
+		Slot:           slot,
 	})
 	if err != nil {
 		log.Err(err).Msg("err adding block")
+		return err
 	}
-	return err
+
+	err = bw.txMapper.AddSlotStatus(ctx, slot, data.SlotStatusValProposed)
+	if err != nil {
+		log.Err(err).Msg("err adding slot status")
+		return err
+	}
+	return nil
 }
 
 func (bw *BlocksWatcher) clearOldBlocks(latestHeader *types.Header) {

internal/watcher/blocks_test.go

@@ -127,6 +127,10 @@ func (m *mockTxMapper) AddBlock(ctx context.Context, b *data.Block) error {
 	panic("unexpected call to AddBlock")
 }
 
+func (m *mockTxMapper) AddSlotStatus(ctx context.Context, slot int64, status data.SlotStatusVal) error {
+	panic("unexpected call to AddSlotStatus")
+}
+
 func (m *mockTxMapper) QueryBlockNumberFromValidatorRegistryEventsSyncedUntil(ctx context.Context) (int64, error) {
 	panic("unexpected call to QueryBlockNumberFromValidatorRegistryEventsSyncedUntil")
 }

migrations/20260116055945_add_slot_status_table.sql

@@ -0,0 +1,21 @@
+-- +goose Up
+-- +goose StatementBegin
+CREATE TYPE slot_status_val AS ENUM 
+(
+    'proposed', 
+    'missed'
+);
+
+CREATE TABLE IF NOT EXISTS slot_status
+(
+    slot                BIGINT PRIMARY KEY,
+    status              slot_status_val NOT NULL,
+    created_at          TIMESTAMP WITH TIME ZONE DEFAULT NOW()  NOT NULL
+);
+-- +goose StatementEnd
+
+-- +goose Down
+-- +goose StatementBegin
+DROP TABLE slot_status;
+DROP TYPE slot_status_val;
+-- +goose StatementEnd