feat: Celestia DA Layer Client implementation (#399)

* Celestia Node RPC Client
* celestia DA implementation
* mock celestia-node RPC server

Co-authored-by: Ismail Khoffi <Ismail.Khoffi@gmail.com>

Author: tzdybal

da/celestia/celestia.go

@@ -0,0 +1,147 @@
+package celestia
+
+import (
+	"context"
+	"encoding/json"
+	"time"
+
+	"github.com/gogo/protobuf/proto"
+
+	"github.com/celestiaorg/optimint/da"
+	"github.com/celestiaorg/optimint/libs/cnrc"
+	"github.com/celestiaorg/optimint/log"
+	"github.com/celestiaorg/optimint/store"
+	"github.com/celestiaorg/optimint/types"
+	pb "github.com/celestiaorg/optimint/types/pb/optimint"
+)
+
+// DataAvailabilityLayerClient use celestia-node public API.
+type DataAvailabilityLayerClient struct {
+	client *cnrc.Client
+
+	config Config
+	logger log.Logger
+}
+
+var _ da.DataAvailabilityLayerClient = &DataAvailabilityLayerClient{}
+var _ da.BlockRetriever = &DataAvailabilityLayerClient{}
+
+type Config struct {
+	BaseURL     string        `json:"base_url"`
+	Timeout     time.Duration `json:"timeout"`
+	GasLimit    uint64        `json:"gas_limit"`
+	NamespaceID [8]byte       `json:"namespace_id"`
+}
+
+func (c *DataAvailabilityLayerClient) Init(config []byte, kvStore store.KVStore, logger log.Logger) error {
+	c.logger = logger
+
+	if len(config) > 0 {
+		return json.Unmarshal(config, &c.config)
+	}
+
+	return nil
+}
+
+func (c *DataAvailabilityLayerClient) Start() error {
+	c.logger.Info("starting Celestia Data Availability Layer Client", "baseURL", c.config.BaseURL)
+	var err error
+	c.client, err = cnrc.NewClient(c.config.BaseURL, cnrc.WithTimeout(c.config.Timeout))
+	return err
+}
+
+func (c *DataAvailabilityLayerClient) Stop() error {
+	c.logger.Info("stopping Celestia Data Availability Layer Client")
+	return nil
+}
+
+func (c *DataAvailabilityLayerClient) SubmitBlock(block *types.Block) da.ResultSubmitBlock {
+	blob, err := block.MarshalBinary()
+	if err != nil {
+		return da.ResultSubmitBlock{
+			DAResult: da.DAResult{
+				Code:    da.StatusError,
+				Message: err.Error(),
+			},
+		}
+	}
+
+	txResponse, err := c.client.SubmitPFD(context.TODO(), c.config.NamespaceID, blob, c.config.GasLimit)
+
+	if err != nil {
+		return da.ResultSubmitBlock{
+			DAResult: da.DAResult{
+				Code:    da.StatusError,
+				Message: err.Error(),
+			},
+		}
+	}
+
+	return da.ResultSubmitBlock{
+		DAResult: da.DAResult{
+			Code:     da.StatusSuccess,
+			Message:  "tx hash: " + txResponse.TxHash,
+			DAHeight: uint64(txResponse.Height),
+		},
+	}
+}
+
+func (c *DataAvailabilityLayerClient) CheckBlockAvailability(dataLayerHeight uint64) da.ResultCheckBlock {
+	shares, err := c.client.NamespacedShares(context.TODO(), c.config.NamespaceID, dataLayerHeight)
+	if err != nil {
+		return da.ResultCheckBlock{
+			DAResult: da.DAResult{
+				Code:    da.StatusError,
+				Message: err.Error(),
+			},
+		}
+	}
+
+	return da.ResultCheckBlock{
+		DAResult: da.DAResult{
+			Code:     da.StatusSuccess,
+			DAHeight: dataLayerHeight,
+		},
+		DataAvailable: len(shares) > 0,
+	}
+}
+
+func (c *DataAvailabilityLayerClient) RetrieveBlocks(dataLayerHeight uint64) da.ResultRetrieveBlocks {
+	data, err := c.client.NamespacedData(context.TODO(), c.config.NamespaceID, dataLayerHeight)
+	if err != nil {
+		return da.ResultRetrieveBlocks{
+			DAResult: da.DAResult{
+				Code:    da.StatusError,
+				Message: err.Error(),
+			},
+		}
+	}
+
+	blocks := make([]*types.Block, len(data))
+	for i, msg := range data {
+		var block pb.Block
+		err = proto.Unmarshal(msg, &block)
+		if err != nil {
+			c.logger.Error("failed to unmarshal block", "daHeight", dataLayerHeight, "position", i, "error", err)
+			continue
+		}
+		blocks[i] = new(types.Block)
+		err := blocks[i].FromProto(&block)
+		if err != nil {
+			return da.ResultRetrieveBlocks{
+				DAResult: da.DAResult{
+					Code:    da.StatusError,
+					Message: err.Error(),
+				},
+			}
+		}
+	}
+
+	return da.ResultRetrieveBlocks{
+		DAResult: da.DAResult{
+			Code:     da.StatusSuccess,
+			DAHeight: dataLayerHeight,
+		},
+		Blocks: blocks,
+	}
+}

da/celestia/mock/messages.go

@@ -0,0 +1,106 @@
+package mock
+
+import (
+	"bytes"
+	"encoding/binary"
+)
+
+// This code is extracted from celestia-app. It's here to build shares from messages (serialized blocks).
+// TODO(tzdybal): if we stop using `/namespaced_shares` we can get rid of this file.
+
+const (
+	ShareSize     = 256
+	NamespaceSize = 8
+	MsgShareSize  = ShareSize - NamespaceSize
+)
+
+// splitMessage breaks the data in a message into the minimum number of
+// namespaced shares
+func splitMessage(rawData []byte, nid []byte) []NamespacedShare {
+	shares := make([]NamespacedShare, 0)
+	firstRawShare := append(append(
+		make([]byte, 0, ShareSize),
+		nid...),
+		rawData[:MsgShareSize]...,
+	)
+	shares = append(shares, NamespacedShare{firstRawShare, nid})
+	rawData = rawData[MsgShareSize:]
+	for len(rawData) > 0 {
+		shareSizeOrLen := min(MsgShareSize, len(rawData))
+		rawShare := append(append(
+			make([]byte, 0, ShareSize),
+			nid...),
+			rawData[:shareSizeOrLen]...,
+		)
+		paddedShare := zeroPadIfNecessary(rawShare, ShareSize)
+		share := NamespacedShare{paddedShare, nid}
+		shares = append(shares, share)
+		rawData = rawData[shareSizeOrLen:]
+	}
+	return shares
+}
+
+// Share contains the raw share data without the corresponding namespace.
+type Share []byte
+
+// NamespacedShare extends a Share with the corresponding namespace.
+type NamespacedShare struct {
+	Share
+	ID []byte
+}
+
+func min(a, b int) int {
+	if a <= b {
+		return a
+	}
+	return b
+}
+
+func zeroPadIfNecessary(share []byte, width int) []byte {
+	oldLen := len(share)
+	if oldLen < width {
+		missingBytes := width - oldLen
+		padByte := []byte{0}
+		padding := bytes.Repeat(padByte, missingBytes)
+		share = append(share, padding...)
+		return share
+	}
+	return share
+}
+
+// marshalDelimited marshals the raw data (excluding the namespace) of this
+// message and prefixes it with the length of that encoding.
+func marshalDelimited(data []byte) ([]byte, error) {
+	lenBuf := make([]byte, binary.MaxVarintLen64)
+	length := uint64(len(data))
+	n := binary.PutUvarint(lenBuf, length)
+	return append(lenBuf[:n], data...), nil
+}
+
+// appendToShares appends raw data as shares.
+// Used to build shares from blocks/messages.
+func appendToShares(shares []NamespacedShare, nid []byte, rawData []byte) []NamespacedShare {
+	if len(rawData) <= MsgShareSize {
+		rawShare := append(append(
+			make([]byte, 0, len(nid)+len(rawData)),
+			nid...),
+			rawData...,
+		)
+		paddedShare := zeroPadIfNecessary(rawShare, ShareSize)
+		share := NamespacedShare{paddedShare, nid}
+		shares = append(shares, share)
+	} else { // len(rawData) > MsgShareSize
+		shares = append(shares, splitMessage(rawData, nid)...)
+	}
+	return shares
+}
+
+type namespacedSharesResponse struct {
+	Shares []Share `json:"shares"`
+	Height uint64  `json:"height"`
+}
+
+type namespacedDataResponse struct {
+	Data   [][]byte `json:"data"`
+	Height uint64   `json:"height"`
+}