From 3ffda89f0c3f00cd7336ab49c7d9fc492820bd8d Mon Sep 17 00:00:00 2001
From: Daniel Mancia <21249320+dmanc@users.noreply.github.com>
Date: Wed, 26 Jun 2024 14:44:37 -0700
Subject: [PATCH] Get blob metadata by account id

---
 .../common/blobstore/blob_metadata_store.go   | 29 +++++++++++++++++--
 1 file changed, 27 insertions(+), 2 deletions(-)

diff --git a/disperser/common/blobstore/blob_metadata_store.go b/disperser/common/blobstore/blob_metadata_store.go
index a731d418b..15950d9dd 100644
--- a/disperser/common/blobstore/blob_metadata_store.go
+++ b/disperser/common/blobstore/blob_metadata_store.go
@@ -16,8 +16,9 @@ import (
 )
 
 const (
-	statusIndexName = "StatusIndex"
-	batchIndexName  = "BatchIndex"
+	statusIndexName               = "StatusIndex"
+	batchIndexName                = "BatchIndex"
+	accountIDRequestedAtIndexName = "AccountID-RequestedAt-Index"
 )
 
 // BlobMetadataStore is a blob metadata storage backed by DynamoDB
@@ -166,6 +167,30 @@ func (s *BlobMetadataStore) GetBlobMetadataByStatusWithPagination(ctx context.Co
 	return metadata, exclusiveStartKey, nil
 }
 
+// GetBlobMetadataByAccountIdRequestedAt returns all the metadata with the given accountID after the given requestedAt timestamp
+func (s *BlobMetadataStore) GetBlobMetadataByAccountIdRequestedAt(ctx context.Context, accountID string, requestedAt uint64) ([]*disperser.BlobMetadata, error) {
+	items, err := s.dynamoDBClient.QueryIndex(ctx, s.tableName, accountIDRequestedAtIndexName, "AccountID = :account_id AND RequestedAt > :requested_at", commondynamodb.ExpresseionValues{
+		":account_id": &types.AttributeValueMemberS{
+			Value: accountID,
+		},
+		":requested_at": &types.AttributeValueMemberN{
+			Value: strconv.Itoa(int(requestedAt)),
+		}})
+	if err != nil {
+		return nil, err
+	}
+
+	metadata := make([]*disperser.BlobMetadata, len(items))
+	for i, item := range items {
+		metadata[i], err = UnmarshalBlobMetadata(item)
+		if err != nil {
+			return nil, err
+		}
+	}
+
+	return metadata, nil
+}
+
 func (s *BlobMetadataStore) GetAllBlobMetadataByBatch(ctx context.Context, batchHeaderHash [32]byte) ([]*disperser.BlobMetadata, error) {
 	items, err := s.dynamoDBClient.QueryIndex(ctx, s.tableName, batchIndexName, "BatchHeaderHash = :batch_header_hash", commondynamodb.ExpresseionValues{
 		":batch_header_hash": &types.AttributeValueMemberB{