feat(dsfs.CreateDataset): initial implementaiton of dsfs.CreateDataset

This commit renames SaveDataset to WriteDataset, and adds a new
exported function on the dsfs package. CreateDataset provides a clean,
canonical implementation of how to write a dataset to a cafs.

I'm excited about this implementation b/c it's a sign of our platform
coalescing into a complete base system. It's clear to me now that
while I started this journey believing that what the world needed was
'the distribed web as a database', it now seems it's smarter to explain
what qri's becoming as a 'global data version control system', and the
first "feature" built on top of this 'dvcs' is being aware of data
transformations.

dope. a.f.

Still to do: setting structure length & row fields, data validation.

Author: b5

commit.go

@@ -13,13 +13,17 @@ import (
 // to be directly analogous to the concept of a Commit Message in the
 // git version control system
 type Commit struct {
-	path    datastore.Key
-	Author  *User  `json:"author,omitempty"`
-	Kind    Kind   `json:"kind,omitempty"`
+	path   datastore.Key
+	Author *User `json:"author,omitempty"`
+	Kind   Kind  `json:"kind,omitempty"`
+	// Message is an optional
 	Message string `json:"message,omitempty"`
-	// Time this dataset was created. Required. Datasets are immutable, so no "updated"
-	Timestamp time.Time `json:"timestamp,omitempty"`
-	Title     string    `json:"title"`
+	// Signature is a base58 encoded privateKey signing of Title
+	Signature string `json:"signature,omitempty"`
+	// Time this dataset was created. Required.
+	Timestamp time.Time `json:"timestamp"`
+	// Title of the commit. Required.
+	Title string `json:"title"`
 }
 
 // NewCommitRef creates an empty struct with it's
@@ -38,6 +42,11 @@ func (cm *Commit) Path() datastore.Key {
 	return cm.path
 }
 
+// SignableBytes produces the portion of a commit message used for signing
+func (cm *Commit) SignableBytes() []byte {
+	return []byte(fmt.Sprintf("%s\n%s", cm.Timestamp.Format(time.RFC3339), cm.Title))
+}
+
 // Assign collapses all properties of a set of Commit onto one.
 // this is directly inspired by Javascript's Object.assign
 func (cm *Commit) Assign(msgs ...*Commit) {
@@ -61,6 +70,9 @@ func (cm *Commit) Assign(msgs ...*Commit) {
 		if m.Message != "" {
 			cm.Message = m.Message
 		}
+		if m.Signature != "" {
+			cm.Signature = m.Signature
+		}
 		if m.Kind.String() != "" {
 			cm.Kind = m.Kind
 		}
@@ -84,6 +96,7 @@ func (cm *Commit) MarshalJSON() ([]byte, error) {
 		Author:    cm.Author,
 		Kind:      kind,
 		Message:   cm.Message,
+		Signature: cm.Signature,
 		Timestamp: cm.Timestamp,
 		Title:     cm.Title,
 	}

commit_test.go

@@ -30,6 +30,7 @@ func TestCommitAssign(t *testing.T) {
 		Timestamp: t1,
 		Title:     "expect title",
 		Message:   "expect message",
+		Signature: "sig",
 	}
 	got := &Commit{
 		Author:  &User{ID: "maha_id", Email: "maha@example.com"},
@@ -45,6 +46,7 @@ func TestCommitAssign(t *testing.T) {
 		path:      datastore.NewKey("a"),
 		Timestamp: t1,
 		Message:   "expect message",
+		Signature: "sig",
 	})
 
 	if err := CompareCommits(expect, got); err != nil {
@@ -63,6 +65,19 @@ func TestCommitAssign(t *testing.T) {
 	}
 }
 
+func TestCommitSignableBytes(t *testing.T) {
+	expect := []byte("2001-01-01T01:01:01Z\nI'm a commit message")
+	cm := &Commit{
+		Timestamp: time.Date(2001, 01, 01, 01, 01, 01, 0, time.UTC),
+		Title:     "I'm a commit message",
+	}
+	got := cm.SignableBytes()
+
+	if !bytes.Equal(expect, got) {
+		t.Errorf("mismatch. expected:\n'%s',got:\n'%s'", string(expect), string(got))
+	}
+}
+
 func TestCommitMarshalJSON(t *testing.T) {
 	ts := time.Date(2001, 01, 01, 01, 01, 01, 0, time.UTC)
 	cases := []struct {

compare.go

@@ -239,6 +239,9 @@ func CompareCommits(a, b *Commit) error {
 	if !a.Timestamp.Equal(b.Timestamp) {
 		return fmt.Errorf("Timestamp: %s != %s", a.Timestamp, b.Timestamp)
 	}
+	if a.Signature != b.Signature {
+		return fmt.Errorf("Signature: %s != %s", a.Signature, b.Signature)
+	}
 	if a.Message != b.Message {
 		return fmt.Errorf("Message: %s != %s", a.Message, b.Message)
 	}

compare_test.go

@@ -169,6 +169,7 @@ func TestCompareCommits(t *testing.T) {
 		{&Commit{Title: "a"}, &Commit{Title: "b"}, "Title: a != b"},
 		{&Commit{Message: "a"}, &Commit{Message: "b"}, "Message: a != b"},
 		{&Commit{Kind: "a"}, &Commit{Kind: "b"}, "Kind: a != b"},
+		{&Commit{Signature: "a"}, &Commit{Signature: "b"}, "Signature: a != b"},
 	}
 
 	for i, c := range cases {

dsfs/dataset.go

@@ -3,12 +3,16 @@ package dsfs
 import (
 	"encoding/json"
 	"fmt"
+	// "io/ioutil"
+	"time"
 
 	"github.com/ipfs/go-datastore"
-	// "github.com/libp2p/go-libp2p-crypto"
+	"github.com/libp2p/go-libp2p-crypto"
+	"github.com/mr-tron/base58/base58"
 	"github.com/qri-io/cafs"
 	"github.com/qri-io/cafs/memfs"
 	"github.com/qri-io/dataset"
+	"github.com/qri-io/dataset/validate"
 )
 
 // LoadDataset reads a dataset from a cafs and dereferences structure, transform, and commitMsg if they exist,
@@ -18,17 +22,7 @@ func LoadDataset(store cafs.Filestore, path datastore.Key) (*dataset.Dataset, er
 	if err != nil {
 		return nil, fmt.Errorf("error loading dataset: %s", err.Error())
 	}
-
-	if err := DerefDatasetMetadata(store, ds); err != nil {
-		return nil, err
-	}
-	if err := DerefDatasetStructure(store, ds); err != nil {
-		return nil, err
-	}
-	if err := DerefDatasetTransform(store, ds); err != nil {
-		return nil, err
-	}
-	if err := DerefDatasetCommit(store, ds); err != nil {
+	if err := DerefDataset(store, ds); err != nil {
 		return nil, err
 	}
 
@@ -65,6 +59,23 @@ func LoadDatasetRefs(store cafs.Filestore, path datastore.Key) (*dataset.Dataset
 	return ds, nil
 }
 
+// DerefDataset attempts to fully dereference a dataset
+func DerefDataset(store cafs.Filestore, ds *dataset.Dataset) error {
+	if err := DerefDatasetMetadata(store, ds); err != nil {
+		return err
+	}
+	if err := DerefDatasetStructure(store, ds); err != nil {
+		return err
+	}
+	if err := DerefDatasetTransform(store, ds); err != nil {
+		return err
+	}
+	if err := DerefDatasetCommit(store, ds); err != nil {
+		return err
+	}
+	return nil
+}
+
 // DerefDatasetStructure derferences a dataset's structure element if required
 // should be a no-op if ds.Structure is nil or isn't a reference
 func DerefDatasetStructure(store cafs.Filestore, ds *dataset.Dataset) error {
@@ -125,111 +136,72 @@ func DerefDatasetCommit(store cafs.Filestore, ds *dataset.Dataset) error {
 	return nil
 }
 
-// CreateDatasetParams defines parmeters for the CreateDataset function
-// type CreateDatasetParams struct {
-// 	// Store is where we're going to
-// 	Store cafs.Filestore
-// 	//
-// 	Dataset  *dataset.Dataset
-// 	DataFile cafs.File
-// 	PrivKey  crypto.PrivKey
-// }
-
-// CreateDataset is the canonical method for getting a dataset pointer & it's data into a store
-// func CreateDataset(p *CreateDatasetParams) (path datastore.Key, err error) {
-// 	// TODO - need a better strategy for huge files
-// 	data, err := ioutil.ReadAll(rdr)
-// 	if err != nil {
-// 		return fmt.Errorf("error reading file: %s", err.Error())
-// 	}
-
-// 	if err = PrepareDataset(p.Store, p.Dataset, p.DataFile); err != nil {
-// 		return
-// 	}
-
-// 	// Ensure that dataset is well-formed
-// 	// format, err := detect.ExtensionDataFormat(filename)
-// 	// if err != nil {
-// 	// 	return fmt.Errorf("error detecting format extension: %s", err.Error())
-// 	// }
-// 	// if err = validate.DataFormat(format, bytes.NewReader(data)); err != nil {
-// 	// 	return fmt.Errorf("invalid data format: %s", err.Error())
-// 	// }
-
-// 	// TODO - check for errors in dataset and warn user if errors exist
-
-// 	datakey, err := store.Put(memfs.NewMemfileBytes("data."+st.Format.String(), data), false)
-// 	if err != nil {
-// 		return fmt.Errorf("error putting data file in store: %s", err.Error())
-// 	}
-
-// 	ds.Timestamp = time.Now().In(time.UTC)
-// 	if ds.Title == "" {
-// 		ds.Title = name
-// 	}
-// 	ds.Data = datakey.String()
-
-// 	if err := validate.Dataset(ds); err != nil {
-// 		return err
-// 	}
-
-// 	dskey, err := SaveDataset(store, ds, true)
-// 	if err != nil {
-// 		return fmt.Errorf("error saving dataset: %s", err.Error())
-// 	}
-// }
+// CreateDataset places a new dataset in the store. Admittedly, this isn't a simple process.
+// Store is where we're going to
+// Dataset to be saved
+// Pin the dataset if the underlying store supports the pinning interface
+func CreateDataset(store cafs.Filestore, ds *dataset.Dataset, df cafs.File, pk crypto.PrivKey, pin bool) (path datastore.Key, err error) {
+	if err = DerefDataset(store, ds); err != nil {
+		return
+	}
+	if err = validate.Dataset(ds); err != nil {
+		return
+	}
+	if err = prepareDataset(store, ds, df, pk); err != nil {
+		return
+	}
+	path, err = WriteDataset(store, ds, df, pin)
+	if err != nil {
+		err = fmt.Errorf("error writing dataset: %s", err.Error())
+	}
+	return
+}
+
+// timestamp is a function for getting commit timestamps
+// we replace this with a static function for testing purposes
+var timestamp = func() time.Time {
+	return time.Now()
+}
 
 // prepareDataset modifies a dataset in preparation for adding to a dsfs
-// func PrepareDataset(store cafs.Filestore, ds *dataset.Dataset, data cafs.File) error {
-
-// 	st, err := detect.FromReader(data.FileName(), data)
-// 	if err != nil {
-// 		return fmt.Errorf("error determining dataset schema: %s", err.Error())
-// 	}
-// 	if ds.Structure == nil {
-// 		ds.Structure = &dataset.Structure{}
-// 	}
-// 	ds.Structure.Assign(st, ds.Structure)
-
-// 	// Ensure that dataset contains valid field names
-// 	if err = validate.Structure(st); err != nil {
-// 		return fmt.Errorf("invalid structure: %s", err.Error())
-// 	}
-// 	if err := validate.DataFormat(st.Format, bytes.NewReader(data)); err != nil {
-// 		return fmt.Errorf("invalid data format: %s", err.Error())
-// 	}
-
-// 	// generate abstract form of dataset
-// 	ds.Abstract = dataset.Abstract(ds)
-
-// 	if ds.AbstractTransform != nil {
-// 		// convert abstract transform to abstract references
-// 		for name, ref := range ds.AbstractTransform.Resources {
-// 			// data, _ := ref.MarshalJSON()
-// 			// fmt.Println(string(data))
-// 			if ref.Abstract != nil {
-// 				ds.AbstractTransform.Resources[name] = ref.Abstract
-// 			} else {
-
-// 				absf, err := JSONFile(PackageFileAbstract.String(), dataset.Abstract(ref))
-// 				if err != nil {
-// 					return err
-// 				}
-// 				path, err := store.Put(absf, true)
-// 				if err != nil {
-// 					return err
-// 				}
-// 				ds.AbstractTransform.Resources[name] = dataset.NewDatasetRef(path)
-// 			}
-// 		}
-// 	}
-
-// 	return nil
-// }
-
-// SaveDataset writes a dataset to a cafs, replacing subcomponents of a dataset with hash references
-// during the write process. Directory structure is according to PackageFile naming conventions
-func SaveDataset(store cafs.Filestore, ds *dataset.Dataset, pin bool) (datastore.Key, error) {
+func prepareDataset(store cafs.Filestore, ds *dataset.Dataset, df cafs.File, privKey crypto.PrivKey) error {
+	if df == nil {
+		return fmt.Errorf("data file is required")
+	}
+
+	// TODO - need a better strategy for huge files. I think that strategy is to split
+	// the reader into multiple consumers that are all performing their task on a stream
+	// of byte slices
+	// data, err := ioutil.ReadAll(df)
+	// if err != nil {
+	// 	return fmt.Errorf("error reading file: %s", err.Error())
+	// }
+
+	// generate abstract form of dataset
+	ds.Abstract = dataset.Abstract(ds)
+
+	// datakey, err := store.Put(memfs.NewMemfileBytes("data."+ds.Structure.Format.String(), data), false)
+	// if err != nil {
+	// 	return fmt.Errorf("error putting data file in store: %s", err.Error())
+	// }
+
+	ds.Commit.Timestamp = timestamp()
+	signedBytes, err := privKey.Sign(ds.Commit.SignableBytes())
+	if err != nil {
+		return fmt.Errorf("error signing commit title: %s", err.Error())
+	}
+	ds.Commit.Signature = base58.Encode(signedBytes)
+
+	// TODO - make sure file ending matches
+	// "data."+ds.Structure.Format.String()
+	return nil
+}
+
+// WriteDataset writes a dataset to a cafs, replacing subcomponents of a dataset with path references
+// during the write process. Directory structure is according to PackageFile naming conventions.
+// This method is currently exported, but 99% of use cases should use CreateDataset instead of this
+// lower-level function
+func WriteDataset(store cafs.Filestore, ds *dataset.Dataset, dataFile cafs.File, pin bool) (datastore.Key, error) {
 	// assign to a new dataset instance to avoid clobbering input dataset
 	cp := &dataset.Dataset{}
 	cp.Assign(ds)
@@ -247,11 +219,14 @@ func SaveDataset(store cafs.Filestore, ds *dataset.Dataset, pin bool) (datastore
 	}
 
 	if ds.AbstractTransform != nil {
-		// ensure all dataset references are abstract
-		for key, r := range ds.AbstractTransform.Resources {
-			if !r.IsEmpty() {
-				return datastore.NewKey(""), fmt.Errorf("abstract transform resource '%s' is not a reference", key)
+		// convert abstract transform to abstract references
+		for name, ref := range ds.AbstractTransform.Resources {
+			absrf, err := JSONFile(fmt.Sprintf("ref_%s.json", name), dataset.Abstract(ref))
+			if err != nil {
+				return datastore.NewKey(""), fmt.Errorf("error marshaling dataset resource '%s' to json: %s", name, err.Error())
 			}
+			fileTasks++
+			adder.AddFile(absrf)
 		}
 		abstff, err := JSONFile(PackageFileAbstractTransform.String(), ds.AbstractTransform)
 		if err != nil {
@@ -279,11 +254,17 @@ func SaveDataset(store cafs.Filestore, ds *dataset.Dataset, pin bool) (datastore
 	// 	if err != nil {
 	// 		return datastore.NewKey(""), fmt.Errorf("error marshaling dataset to json: %s", err.Error())
 	// 	}
-
 	// 	fileTasks++
 	// 	adder.AddFile(dsf)
 	// 	addedDataset = true
 	// }
+	// data, err := store.Get(datastore.NewKey(ds.Data))
+	// if err != nil {
+	// 	return datastore.NewKey(""), fmt.Errorf("error getting dataset raw data: %s", err.Error())
+	// }
+
+	fileTasks++
+	adder.AddFile(dataFile)
 
 	if ds.Transform != nil {
 		// all resources must be references
@@ -348,6 +329,16 @@ func SaveDataset(store cafs.Filestore, ds *dataset.Dataset, pin bool) (datastore
 				ds.Metadata = dataset.NewMetadataRef(ao.Path)
 			case PackageFileCommit.String():
 				ds.Commit = dataset.NewCommitRef(ao.Path)
+			case dataFile.FileName():
+				ds.DataPath = ao.Path.String()
+			default:
+				if ds.AbstractTransform != nil {
+					for name := range ds.AbstractTransform.Resources {
+						if ao.Name == fmt.Sprintf("ref_%s.json", name) {
+							ds.AbstractTransform.Resources[name] = dataset.NewDatasetRef(ao.Path)
+						}
+					}
+				}
 			}
 
 			fileTasks--