return block data when deleting it (#175)

### TL;DR

Enhanced `DeleteBlockData` to return deleted block data, enabling better handling of reorgs.

### What changed?

Modified the `DeleteBlockData` method in the storage interface to return the deleted block data:
- Updated the method signature in `IMainStorage` interface to return `([]common.BlockData, error)`
- Implemented the new functionality in `ClickHouseConnector` to collect and return deleted data
- Modified the individual delete methods (`deleteBlocks`, `deleteLogs`, etc.) to return the deleted entities
- Updated the `MemoryConnector` implementation to match the new interface
- Fixed the mock implementation to support the new signature
- Updated the caller in `ReorgHandler` to handle the new return value

### How to test?

1. Run unit tests to verify the updated interface implementations
2. Test a reorg scenario to ensure deleted block data is properly returned
3. Verify that the reorg handler correctly processes the returned data

### Why make this change?

This change improves the reorg handling process by providing access to the deleted block data, which can be useful for:
- Auditing what was removed during a reorg
- Potentially using the deleted data for other purposes
- Making the reorg process more transparent and traceable
- Setting the foundation for making delete and insert operations atomic

Author: iuwqyir

internal/orchestrator/reorg_handler.go

@@ -278,7 +278,7 @@ func (rh *ReorgHandler) handleReorg(reorgedBlockNumbers []*big.Int) error {
 		blocksToDelete = append(blocksToDelete, result.BlockNumber)
 	}
 	// TODO make delete and insert atomic
-	if err := rh.storage.MainStorage.DeleteBlockData(rh.rpc.GetChainID(), blocksToDelete); err != nil {
+	if _, err := rh.storage.MainStorage.DeleteBlockData(rh.rpc.GetChainID(), blocksToDelete); err != nil {
 		return fmt.Errorf("error deleting data for blocks %v: %w", blocksToDelete, err)
 	}
 	if err := rh.storage.MainStorage.InsertBlockData(data); err != nil {

internal/orchestrator/reorg_handler_test.go

@@ -504,7 +504,7 @@ func TestHandleReorg(t *testing.T) {
 	})
 	mockOrchestratorStorage.EXPECT().GetLastReorgCheckedBlockNumber(big.NewInt(1)).Return(big.NewInt(3), nil)
 
-	mockMainStorage.EXPECT().DeleteBlockData(big.NewInt(1), mock.Anything).Return(nil)
+	mockMainStorage.EXPECT().DeleteBlockData(big.NewInt(1), mock.Anything).Return([]common.BlockData{}, nil)
 	mockMainStorage.EXPECT().InsertBlockData(mock.Anything).Return(nil)
 
 	handler := NewReorgHandler(mockRPC, mockStorage)
@@ -613,7 +613,7 @@ func TestHandleReorgWithSingleBlockReorg(t *testing.T) {
 
 	mockMainStorage.EXPECT().DeleteBlockData(big.NewInt(1), mock.MatchedBy(func(blocks []*big.Int) bool {
 		return len(blocks) == 1
-	})).Return(nil)
+	})).Return([]common.BlockData{}, nil)
 	mockMainStorage.EXPECT().InsertBlockData(mock.MatchedBy(func(data []common.BlockData) bool {
 		return len(data) == 1
 	})).Return(nil)
@@ -681,7 +681,7 @@ func TestHandleReorgWithLatestBlockReorged(t *testing.T) {
 
 	mockMainStorage.EXPECT().DeleteBlockData(big.NewInt(1), mock.MatchedBy(func(blocks []*big.Int) bool {
 		return len(blocks) == 8
-	})).Return(nil)
+	})).Return([]common.BlockData{}, nil)
 	mockMainStorage.EXPECT().InsertBlockData(mock.MatchedBy(func(data []common.BlockData) bool {
 		return len(data) == 8
 	})).Return(nil)
@@ -745,7 +745,7 @@ func TestHandleReorgWithManyBlocks(t *testing.T) {
 
 	mockMainStorage.EXPECT().DeleteBlockData(big.NewInt(1), mock.MatchedBy(func(blocks []*big.Int) bool {
 		return len(blocks) == 5
-	})).Return(nil)
+	})).Return([]common.BlockData{}, nil)
 	mockMainStorage.EXPECT().InsertBlockData(mock.MatchedBy(func(data []common.BlockData) bool {
 		return len(data) == 5
 	})).Return(nil)

internal/storage/clickhouse.go

@@ -987,122 +987,168 @@ func (c *ClickHouseConnector) GetBlockHeadersDescending(chainId *big.Int, from *
 	return blockHeaders, nil
 }
 
-func (c *ClickHouseConnector) DeleteBlockData(chainId *big.Int, blockNumbers []*big.Int) error {
+func (c *ClickHouseConnector) DeleteBlockData(chainId *big.Int, blockNumbers []*big.Int) ([]common.BlockData, error) {
 	var deleteErr error
 	var deleteErrMutex sync.Mutex
 	var wg sync.WaitGroup
 	wg.Add(4)
 
+	// Create a map to store block data that will be deleted
+	deletedBlockDataByNumber := make(map[*big.Int]common.BlockData)
 	go func() {
 		defer wg.Done()
-		if err := c.deleteBlocks(chainId, blockNumbers); err != nil {
+		deletedBlocks, err := c.deleteBlocks(chainId, blockNumbers)
+		if err != nil {
 			deleteErrMutex.Lock()
 			deleteErr = fmt.Errorf("error deleting blocks: %v", err)
 			deleteErrMutex.Unlock()
 		}
+		for _, block := range deletedBlocks {
+			data := deletedBlockDataByNumber[block.Number]
+			data.Block = block
+			deletedBlockDataByNumber[block.Number] = data
+		}
 	}()
 
 	go func() {
 		defer wg.Done()
-		if err := c.deleteLogs(chainId, blockNumbers); err != nil {
+		deletedLogs, err := c.deleteLogs(chainId, blockNumbers)
+		if err != nil {
 			deleteErrMutex.Lock()
 			deleteErr = fmt.Errorf("error deleting logs: %v", err)
 			deleteErrMutex.Unlock()
 		}
+		for _, log := range deletedLogs {
+			data := deletedBlockDataByNumber[log.BlockNumber]
+			data.Logs = append(data.Logs, log)
+			deletedBlockDataByNumber[log.BlockNumber] = data
+		}
 	}()
 
 	go func() {
 		defer wg.Done()
-		if err := c.deleteTransactions(chainId, blockNumbers); err != nil {
+		deletedTransactions, err := c.deleteTransactions(chainId, blockNumbers)
+		if err != nil {
 			deleteErrMutex.Lock()
 			deleteErr = fmt.Errorf("error deleting transactions: %v", err)
 			deleteErrMutex.Unlock()
 		}
+		for _, tx := range deletedTransactions {
+			data := deletedBlockDataByNumber[tx.BlockNumber]
+			data.Transactions = append(data.Transactions, tx)
+			deletedBlockDataByNumber[tx.BlockNumber] = data
+		}
 	}()
 
 	go func() {
 		defer wg.Done()
-		if err := c.deleteTraces(chainId, blockNumbers); err != nil {
+		deletedTraces, err := c.deleteTraces(chainId, blockNumbers)
+		if err != nil {
 			deleteErrMutex.Lock()
 			deleteErr = fmt.Errorf("error deleting traces: %v", err)
 			deleteErrMutex.Unlock()
 		}
+		for _, trace := range deletedTraces {
+			data := deletedBlockDataByNumber[trace.BlockNumber]
+			data.Traces = append(data.Traces, trace)
+			deletedBlockDataByNumber[trace.BlockNumber] = data
+		}
 	}()
 
 	wg.Wait()
 
 	if deleteErr != nil {
-		return deleteErr
+		return nil, deleteErr
 	}
-	return nil
+	deletedBlockData := make([]common.BlockData, 0, len(deletedBlockDataByNumber))
+	for _, data := range deletedBlockDataByNumber {
+		deletedBlockData = append(deletedBlockData, data)
+	}
+	return deletedBlockData, nil
 }
 
-func (c *ClickHouseConnector) deleteBlocks(chainId *big.Int, blockNumbers []*big.Int) error {
+func (c *ClickHouseConnector) deleteBlocks(chainId *big.Int, blockNumbers []*big.Int) ([]common.Block, error) {
 	blocksQueryResult, err := c.GetBlocks(QueryFilter{
 		ChainId:             chainId,
 		BlockNumbers:        blockNumbers,
 		ForceConsistentData: true,
 	}, "*")
 	if err != nil {
-		return err
+		return nil, err
 	}
 	if len(blocksQueryResult.Data) == 0 {
-		return nil // No blocks to delete
+		return nil, nil // No blocks to delete
 	}
-	return c.insertBlocks(blocksQueryResult.Data, InsertOptions{
+	err = c.insertBlocks(blocksQueryResult.Data, InsertOptions{
 		AsDeleted: true,
 	})
+	if err != nil {
+		return nil, err
+	}
+	return blocksQueryResult.Data, nil
 }
 
-func (c *ClickHouseConnector) deleteLogs(chainId *big.Int, blockNumbers []*big.Int) error {
+func (c *ClickHouseConnector) deleteLogs(chainId *big.Int, blockNumbers []*big.Int) ([]common.Log, error) {
 	logsQueryResult, err := c.GetLogs(QueryFilter{
 		ChainId:             chainId,
 		BlockNumbers:        blockNumbers,
 		ForceConsistentData: true,
 	}, "*")
 	if err != nil {
-		return err
+		return nil, err
 	}
 	if len(logsQueryResult.Data) == 0 {
-		return nil // No logs to delete
+		return nil, nil // No logs to delete
 	}
-	return c.insertLogs(logsQueryResult.Data, InsertOptions{
+	err = c.insertLogs(logsQueryResult.Data, InsertOptions{
 		AsDeleted: true,
 	})
+	if err != nil {
+		return nil, err
+	}
+	return logsQueryResult.Data, nil
 }
 
-func (c *ClickHouseConnector) deleteTransactions(chainId *big.Int, blockNumbers []*big.Int) error {
+func (c *ClickHouseConnector) deleteTransactions(chainId *big.Int, blockNumbers []*big.Int) ([]common.Transaction, error) {
 	txsQueryResult, err := c.GetTransactions(QueryFilter{
 		ChainId:             chainId,
 		BlockNumbers:        blockNumbers,
 		ForceConsistentData: true,
 	}, "*")
 	if err != nil {
-		return err
+		return nil, err
 	}
 	if len(txsQueryResult.Data) == 0 {
-		return nil // No transactions to delete
+		return nil, nil // No transactions to delete
 	}
-	return c.insertTransactions(txsQueryResult.Data, InsertOptions{
+	err = c.insertTransactions(txsQueryResult.Data, InsertOptions{
 		AsDeleted: true,
 	})
+	if err != nil {
+		return nil, err
+	}
+	return txsQueryResult.Data, nil
 }
 
-func (c *ClickHouseConnector) deleteTraces(chainId *big.Int, blockNumbers []*big.Int) error {
+func (c *ClickHouseConnector) deleteTraces(chainId *big.Int, blockNumbers []*big.Int) ([]common.Trace, error) {
 	tracesQueryResult, err := c.GetTraces(QueryFilter{
 		ChainId:             chainId,
 		BlockNumbers:        blockNumbers,
 		ForceConsistentData: true,
 	}, "*")
 	if err != nil {
-		return err
+		return nil, err
 	}
 	if len(tracesQueryResult.Data) == 0 {
-		return nil // No traces to delete
+		return nil, nil // No traces to delete
 	}
-	return c.insertTraces(tracesQueryResult.Data, InsertOptions{
+	err = c.insertTraces(tracesQueryResult.Data, InsertOptions{
 		AsDeleted: true,
 	})
+	if err != nil {
+		return nil, err
+	}
+	return tracesQueryResult.Data, nil
 }
 
 // TODO make this atomic

internal/storage/connector.go

@@ -97,7 +97,7 @@ type IMainStorage interface {
 	 * Get block headers ordered from latest to oldest.
 	 */
 	GetBlockHeadersDescending(chainId *big.Int, from *big.Int, to *big.Int) (blockHeaders []common.BlockHeader, err error)
-	DeleteBlockData(chainId *big.Int, blockNumbers []*big.Int) error
+	DeleteBlockData(chainId *big.Int, blockNumbers []*big.Int) ([]common.BlockData, error)
 
 	GetTokenBalances(qf BalancesQueryFilter, fields ...string) (QueryResult[common.TokenBalance], error)
 	GetTokenTransfers(qf TransfersQueryFilter, fields ...string) (QueryResult[common.TokenTransfer], error)

internal/storage/memory.go

@@ -435,7 +435,7 @@ func (m *MemoryConnector) InsertBlockData(data []common.BlockData) error {
 	return nil
 }
 
-func (m *MemoryConnector) DeleteBlockData(chainId *big.Int, blockNumbers []*big.Int) error {
+func (m *MemoryConnector) DeleteBlockData(chainId *big.Int, blockNumbers []*big.Int) ([]common.BlockData, error) {
 	blockNumbersToCheck := getBlockNumbersToCheck(QueryFilter{BlockNumbers: blockNumbers})
 	for _, key := range m.cache.Keys() {
 		prefixes := []string{fmt.Sprintf("block:%s:", chainId.String()), fmt.Sprintf("log:%s:", chainId.String()), fmt.Sprintf("transaction:%s:", chainId.String()), fmt.Sprintf("trace:%s:", chainId.String())}
@@ -444,7 +444,7 @@ func (m *MemoryConnector) DeleteBlockData(chainId *big.Int, blockNumbers []*big.
 			m.cache.Remove(key)
 		}
 	}
-	return nil
+	return []common.BlockData{}, nil // TODO implement
 }
 
 func (m *MemoryConnector) GetBlockHeadersDescending(chainId *big.Int, from *big.Int, to *big.Int) ([]common.BlockHeader, error) {