pdp: add /pdp/data-sets/create-and-add client

pdp/server.go

@@ -106,6 +106,73 @@ func (s *Server) CreateDataSet(ctx context.Context, recordKeeper string, extraDa
 	}, nil
 }
 
+// CreateDataSetAndAddPieces issues POST /pdp/data-sets/create-and-add. The
+// SP submits a single addPieces(dataSetId=0, recordKeeper, ...) on chain
+// from its own wallet, atomically creating a new data set with the given
+// recordKeeper as listener and adding the supplied pieces in one tx. The
+// pieces must already be local on the SP (e.g. via prior /pdp/piece/pull
+// or /pdp/piece/uploads). extraData is the abi.encode(bytes,bytes) of the
+// EIP-712-signed CreateDataSet and AddPieces blobs (see
+// EncodeCreateDataSetAndAddPiecesExtraData). Status (and the eventual
+// dataSetId) is observed through GetDataSetCreationStatus on the returned
+// txHash.
+func (s *Server) CreateDataSetAndAddPieces(ctx context.Context, recordKeeper string, pieceCIDs []cid.Cid, extraData string) (*CreateDataSetResponse, error) {
+	pieces := make([]PieceData, len(pieceCIDs))
+	for i, c := range pieceCIDs {
+		cidStr := c.String()
+		pieces[i] = PieceData{
+			PieceCID: cidStr,
+			SubPieces: []SubPieceData{
+				{SubPieceCID: cidStr},
+			},
+		}
+	}
+
+	reqBody := CreateAndAddRequest{
+		RecordKeeper: recordKeeper,
+		Pieces:       pieces,
+		ExtraData:    extraData,
+	}
+
+	body, err := json.Marshal(reqBody)
+	if err != nil {
+		return nil, fmt.Errorf("failed to marshal create-and-add request: %w", err)
+	}
+
+	req, err := http.NewRequestWithContext(ctx, "POST", s.baseURL+"/pdp/data-sets/create-and-add", bytes.NewReader(body))
+	if err != nil {
+		return nil, fmt.Errorf("failed to create create-and-add request: %w", err)
+	}
+	req.Header.Set("Content-Type", "application/json")
+
+	resp, err := s.httpClient.Do(req)
+	if err != nil {
+		return nil, fmt.Errorf("create-and-add request failed: %w", err)
+	}
+	defer resp.Body.Close()
+
+	if resp.StatusCode != http.StatusCreated && resp.StatusCode != http.StatusAccepted {
+		respBody, _ := io.ReadAll(resp.Body)
+		return nil, fmt.Errorf("unexpected status %d: %s", resp.StatusCode, string(respBody))
+	}
+
+	location := resp.Header.Get("Location")
+	if location == "" {
+		return nil, fmt.Errorf("missing Location header")
+	}
+
+	parts := strings.Split(location, "/")
+	txHash := parts[len(parts)-1]
+	if !strings.HasPrefix(txHash, "0x") {
+		return nil, fmt.Errorf("invalid txHash in Location header: %s", txHash)
+	}
+
+	return &CreateDataSetResponse{
+		TxHash:    txHash,
+		StatusURL: s.baseURL + location,
+	}, nil
+}
+
 func (s *Server) GetDataSetCreationStatus(ctx context.Context, txHash string) (*DataSetCreationStatus, error) {
 	req, err := http.NewRequestWithContext(ctx, "GET", s.baseURL+"/pdp/data-sets/created/"+txHash, nil)
 	if err != nil {

pdp/server_test.go

@@ -14,6 +14,7 @@ import (
 
 	"github.com/ethereum/go-ethereum/common"
 	"github.com/ethereum/go-ethereum/crypto"
+	"github.com/ipfs/go-cid"
 )
 
 func testAuthHelper(t *testing.T) *AuthHelper {
@@ -165,6 +166,102 @@ func TestServer_CreateDataSet(t *testing.T) {
 	})
 }
 
+func TestServer_CreateDataSetAndAddPieces(t *testing.T) {
+	pieceCID := mustCID(t, "baga6ea4seaqao7s73y24kcutaosvacpdjgfe5pw76ooefnyqw4ynr3d2y6x2mpq")
+	recordKeeper := "0x02925630df557F957f70E112bA06e50965417CA0"
+	extraData := "0xdeadbeef"
+
+	t.Run("successful creation", func(t *testing.T) {
+		expectedTxHash := "0x1234567890abcdef1234567890abcdef1234567890abcdef1234567890abcdef"
+		var seen CreateAndAddRequest
+
+		server, _ := setupMockServer(t, http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+			if r.Method != "POST" {
+				t.Errorf("Expected POST, got %s", r.Method)
+			}
+			if r.URL.Path != "/pdp/data-sets/create-and-add" {
+				t.Errorf("Expected path /pdp/data-sets/create-and-add, got %s", r.URL.Path)
+			}
+			body, _ := io.ReadAll(r.Body)
+			if err := json.Unmarshal(body, &seen); err != nil {
+				t.Fatalf("decode body: %v", err)
+			}
+			w.Header().Set("Location", "/pdp/data-sets/created/"+expectedTxHash)
+			w.WriteHeader(http.StatusCreated)
+		}))
+
+		result, err := server.CreateDataSetAndAddPieces(
+			context.Background(),
+			recordKeeper,
+			[]cid.Cid{pieceCID},
+			extraData,
+		)
+		if err != nil {
+			t.Fatalf("CreateDataSetAndAddPieces failed: %v", err)
+		}
+		if result.TxHash != expectedTxHash {
+			t.Errorf("TxHash = %s, want %s", result.TxHash, expectedTxHash)
+		}
+		if seen.RecordKeeper != recordKeeper {
+			t.Errorf("RecordKeeper = %s, want %s", seen.RecordKeeper, recordKeeper)
+		}
+		if seen.ExtraData != extraData {
+			t.Errorf("ExtraData = %s, want %s", seen.ExtraData, extraData)
+		}
+		if len(seen.Pieces) != 1 {
+			t.Fatalf("len(Pieces) = %d, want 1", len(seen.Pieces))
+		}
+		if seen.Pieces[0].PieceCID != pieceCID.String() {
+			t.Errorf("Pieces[0].PieceCID = %s, want %s", seen.Pieces[0].PieceCID, pieceCID.String())
+		}
+		if len(seen.Pieces[0].SubPieces) != 1 || seen.Pieces[0].SubPieces[0].SubPieceCID != pieceCID.String() {
+			t.Errorf("Pieces[0].SubPieces shape mismatch: %+v", seen.Pieces[0].SubPieces)
+		}
+	})
+
+	t.Run("missing Location header", func(t *testing.T) {
+		server, _ := setupMockServer(t, http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+			w.WriteHeader(http.StatusCreated)
+		}))
+
+		_, err := server.CreateDataSetAndAddPieces(
+			context.Background(),
+			recordKeeper,
+			[]cid.Cid{pieceCID},
+			extraData,
+		)
+		if err == nil {
+			t.Error("Expected error for missing Location header, got nil")
+		}
+	})
+
+	t.Run("server error is surfaced", func(t *testing.T) {
+		server, _ := setupMockServer(t, http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+			w.WriteHeader(http.StatusBadRequest)
+			_, _ = w.Write([]byte("extraData validation failed"))
+		}))
+
+		_, err := server.CreateDataSetAndAddPieces(
+			context.Background(),
+			recordKeeper,
+			[]cid.Cid{pieceCID},
+			extraData,
+		)
+		if err == nil {
+			t.Error("Expected error for 400 response, got nil")
+		}
+	})
+}
+
+func mustCID(t *testing.T, s string) cid.Cid {
+	t.Helper()
+	c, err := cid.Decode(s)
+	if err != nil {
+		t.Fatalf("decode CID %q: %v", s, err)
+	}
+	return c
+}
+
 func TestServer_GetDataSetCreationStatus(t *testing.T) {
 	t.Run("successful status check", func(t *testing.T) {
 		txHash := "0x1234567890abcdef1234567890abcdef1234567890abcdef1234567890abcdef"