feat(blob): extend blob struct with index field

api/gateway/health.go

@@ -1,6 +1,8 @@
 package gateway
 
-import "net/http"
+import (
+	"net/http"
+)
 
 const (
 	healthEndpoint = "/status/health"

blob/blob.go

@@ -31,6 +31,9 @@ func (com Commitment) Equal(c Commitment) bool {
 }
 
 // Proof is a collection of nmt.Proofs that verifies the inclusion of the data.
+// Proof proves the WHOLE namespaced data for the particular row.
+// TODO (@vgonkivs): rework `Proof` in order to prove a particular blob.
+// https://github.com/celestiaorg/celestia-node/issues/2303
 type Proof []*nmt.Proof
 
 func (p Proof) Len() int { return len(p) }
@@ -99,6 +102,10 @@ type Blob struct {
 	// the celestia-node's namespace type
 	// this is to avoid converting to and from app's type
 	namespace share.Namespace
+
+	// index represents index of the first share in the eds.
+	// before data is being published, the index set by default to -1.
+	index int
 }
 
 // NewBlobV0 constructs a new blob from the provided Namespace and data.
@@ -127,19 +134,25 @@ func NewBlob(shareVersion uint8, namespace share.Namespace, data []byte) (*Blob,
 	if err != nil {
 		return nil, err
 	}
-	return &Blob{Blob: blob, Commitment: com, namespace: namespace}, nil
+	return &Blob{Blob: blob, Commitment: com, namespace: namespace, index: -1}, nil
 }
 
 // Namespace returns blob's namespace.
 func (b *Blob) Namespace() share.Namespace {
 	return b.namespace
 }
 
+// Index returns the index  of the first share in the eds.
+func (b *Blob) Index() int {
+	return b.index
+}
+
 type jsonBlob struct {
 	Namespace    share.Namespace `json:"namespace"`
 	Data         []byte          `json:"data"`
 	ShareVersion uint32          `json:"share_version"`
 	Commitment   Commitment      `json:"commitment"`
+	Index        int             `json:"index"`
 }
 
 func (b *Blob) MarshalJSON() ([]byte, error) {
@@ -148,6 +161,7 @@ func (b *Blob) MarshalJSON() ([]byte, error) {
 		Data:         b.Data,
 		ShareVersion: b.ShareVersion,
 		Commitment:   b.Commitment,
+		Index:        b.index,
 	}
 	return json.Marshal(blob)
 }
@@ -165,39 +179,88 @@ func (b *Blob) UnmarshalJSON(data []byte) error {
 	b.Blob.ShareVersion = blob.ShareVersion
 	b.Commitment = blob.Commitment
 	b.namespace = blob.Namespace
+	b.index = blob.Index
 	return nil
 }
 
-// buildBlobsIfExist takes shares and tries building the Blobs from them.
-// It will build blobs either until appShares will be empty or the first incomplete blob will
-// appear, so in this specific case it will return all built blobs + remaining shares.
-func buildBlobsIfExist(appShares []shares.Share) ([]*Blob, []shares.Share, error) {
-	if len(appShares) == 0 {
-		return nil, nil, errors.New("empty shares received")
+// index represents index of the blob's first share in the eds.
+// -1 by default and only retrieved onchain blobs will have it set.
+type transformer struct {
+	index  int
+	length int
+	shares []shares.Share
+}
+
+func newTransformer(index, length int) *transformer {
+	return &transformer{
+		index:  index,
+		length: length,
 	}
-	blobs := make([]*Blob, 0, len(appShares))
-	for {
-		length, err := appShares[0].SequenceLen()
-		if err != nil {
-			return nil, nil, err
-		}
+}
 
-		amount := shares.SparseSharesNeeded(length)
-		if amount > len(appShares) {
-			return blobs, appShares, nil
+// setShares sets shares until the blob is completed and returns extra shares back.
+func (b *transformer) setShares(shares []shares.Share) (shrs []shares.Share, isComplete bool) {
+	index := -1
+	for i, sh := range shares {
+		b.shares = append(b.shares, sh)
+		if len(b.shares) == b.length {
+			index = i
+			isComplete = true
+			break
 		}
+	}
 
-		b, err := parseShares(appShares[:amount])
-		if err != nil {
-			return nil, nil, err
-		}
+	if index == -1 {
+		return
+	}
 
-		// only 1 blob will be created bc we passed the exact amount of shares
-		blobs = append(blobs, b[0])
+	if index+1 >= len(shares) {
+		return shrs, true
+	}
+	return shares[index+1:], true
+}
 
-		if amount == len(appShares) {
-			return blobs, nil, nil
-		}
-		appShares = appShares[amount:]
+func (b *transformer) transform() (*Blob, error) {
+	if b.length != len(b.shares) {
+		return nil, errors.New("blob: incomplete blob")
+	}
+
+	sequence, err := shares.ParseShares(b.shares, true)
+	if err != nil {
+		return nil, err
 	}
+
+	if len(sequence) != 1 {
+		return nil, ErrBlobNotFound
+	}
+
+	data, err := sequence[0].RawData()
+	if err != nil {
+		return nil, err
+	}
+	if len(data) == 0 {
+		return nil, ErrBlobNotFound
+	}
+
+	shareVersion, err := sequence[0].Shares[0].Version()
+	if err != nil {
+		return nil, err
+	}
+
+	blob, err := NewBlob(shareVersion, sequence[0].Namespace.Bytes(), data)
+	if err != nil {
+		return nil, err
+	}
+	blob.index = b.index
+	return blob, nil
+}
+
+func (b *transformer) isEmpty() bool {
+	return b.index == 0 && b.length == 0 && len(b.shares) == 0
+}
+
+func (b *transformer) reset() {
+	b.index = 0
+	b.length = 0
+	b.shares = []shares.Share{}
 }

blob/blob_test.go

@@ -1,7 +1,7 @@
 package blob
 
 import (
-	"reflect"
+	"bytes"
 	"testing"
 
 	"github.com/stretchr/testify/assert"
@@ -14,7 +14,7 @@ import (
 )
 
 func TestBlob(t *testing.T) {
-	appBlobs, err := blobtest.GenerateV0Blobs([]int{1}, false)
+	appBlobs, err := blobtest.GenerateV0Blobs([]int{16}, false)
 	require.NoError(t, err)
 	blob, err := convertBlobs(appBlobs...)
 	require.NoError(t, err)
@@ -67,7 +67,8 @@ func TestBlob(t *testing.T) {
 
 				newBlob := &Blob{}
 				require.NoError(t, newBlob.UnmarshalJSON(data))
-				require.True(t, reflect.DeepEqual(blob[0], newBlob))
+				require.True(t, bytes.Equal(blob[0].Blob.Data, newBlob.Data))
+				require.True(t, bytes.Equal(blob[0].Commitment, newBlob.Commitment))
 			},
 		},
 	}

blob/helper.go

@@ -75,7 +75,7 @@ func BlobsToShares(blobs ...*Blob) ([]share.Share, error) {
 
 	sort.Slice(b, func(i, j int) bool {
 		val := bytes.Compare(b[i].NamespaceID, b[j].NamespaceID)
-		return val <= 0
+		return val < 0
 	})
 
 	rawShares, err := shares.SplitBlobs(b...)