changefeed for rethinkdb

server/handlers/changefeed.go

@@ -43,7 +43,12 @@ func Changefeed(ctx context.Context, w http.ResponseWriter, r *http.Request) err
 
 func changefeed(logger ctxu.Logger, store storage.MetaStore, gun, changeID string, records int64) ([]byte, error) {
 	changes, err := store.GetChanges(changeID, int(records), gun)
-	if err != nil {
+	switch err.(type) {
+	case nil:
+		// no error to return
+	case storage.ErrBadQuery:
+		return nil, errors.ErrInvalidParams.WithDetail(err)
+	default:
 		logger.Errorf("%d GET could not retrieve records: %s", http.StatusInternalServerError, err.Error())
 		return nil, errors.ErrUnknown.WithDetail(err)
 	}

server/handlers/default.go

@@ -2,6 +2,8 @@ package handlers
 
 import (
 	"bytes"
+	"crypto/sha256"
+	"encoding/hex"
 	"encoding/json"
 	"io"
 	"net/http"
@@ -114,9 +116,24 @@ func atomicUpdateHandler(ctx context.Context, w http.ResponseWriter, r *http.Req
 		logger.Errorf("500 POST error applying update request: %v", err)
 		return errors.ErrUpdating.WithDetail(nil)
 	}
+
+	logTS(logger, gun.String(), updates)
+
 	return nil
 }
 
+// logTS logs the timestamp update at Info level
+func logTS(logger ctxu.Logger, gun string, updates []storage.MetaUpdate) {
+	for _, update := range updates {
+		if update.Role == data.CanonicalTimestampRole {
+			checksumBin := sha256.Sum256(update.Data)
+			checksum := hex.EncodeToString(checksumBin[:])
+			logger.Infof("updated %s to timestamp version %d, checksum %s", gun, update.Version, checksum)
+			break
+		}
+	}
+}
+
 // GetHandler returns the json for a specified role and GUN.
 func GetHandler(ctx context.Context, w http.ResponseWriter, r *http.Request) error {
 	defer r.Body.Close()
@@ -174,6 +191,7 @@ func DeleteHandler(ctx context.Context, w http.ResponseWriter, r *http.Request)
 		logger.Error("500 DELETE repository")
 		return errors.ErrUnknown.WithDetail(err)
 	}
+	logger.Infof("trust data deleted for %s", gun)
 	return nil
 }
 

server/storage/errors.go

@@ -40,3 +40,13 @@ type ErrNoKey struct {
 func (err ErrNoKey) Error() string {
 	return fmt.Sprintf("Error, no timestamp key found for %s", err.gun)
 }
+
+// ErrBadQuery is used when the parameters provided cannot be appropriately
+// coerced.
+type ErrBadQuery struct {
+	msg string
+}
+
+func (err ErrBadQuery) Error() string {
+	return fmt.Sprintf("did not recognize parameters: %s", err.msg)
+}

server/storage/memory.go

@@ -85,7 +85,7 @@ func (st *MemStorage) UpdateCurrent(gun data.GUN, update MetaUpdate) error {
 // the MemStorage. Behaviour is undefined otherwise
 func (st *MemStorage) writeChange(gun data.GUN, version int, checksum string) {
 	c := Change{
-		ID:        uint(len(st.changes) + 1),
+		ID:        strconv.Itoa(len(st.changes) + 1),
 		GUN:       gun.String(),
 		Version:   version,
 		SHA256:    checksum,
@@ -200,7 +200,7 @@ func (st *MemStorage) Delete(gun data.GUN) error {
 	}
 	delete(st.checksums, gun.String())
 	c := Change{
-		ID:        uint(len(st.changes) + 1),
+		ID:        strconv.Itoa(len(st.changes) + 1),
 		GUN:       gun.String(),
 		Category:  changeCategoryDeletion,
 		CreatedAt: time.Now(),
@@ -224,7 +224,7 @@ func (st *MemStorage) GetChanges(changeID string, records int, filterName string
 	} else {
 		id, err = strconv.ParseInt(changeID, 10, 32)
 		if err != nil {
-			return nil, err
+			return nil, ErrBadQuery{msg: fmt.Sprintf("change ID expected to be integer, provided ID was: %d", changeID)}
 		}
 	}
 	var (

server/storage/mysql_test.go

@@ -43,7 +43,7 @@ func init() {
 
 			// drop all tables, if they exist
 			gormDB.DropTable(&TUFFile{})
-			gormDB.DropTable(&Change{})
+			gormDB.DropTable(&SQLChange{})
 		}
 		cleanup1()
 		dbStore := SetupSQLDB(t, "mysql", dburl)

server/storage/postgresql_test.go

@@ -44,7 +44,7 @@ func init() {
 
 			// drop all tables, if they exist
 			gormDB.DropTable(&TUFFile{})
-			gormDB.DropTable(&Change{})
+			gormDB.DropTable(&SQLChange{})
 		}
 		cleanup1()
 		dbStore := SetupSQLDB(t, notary.PostgresBackend, dburl)

server/storage/rethink_realdb_test.go

@@ -36,6 +36,7 @@ func rethinkDBSetup(t *testing.T) (RethinkDB, func()) {
 	cleanup()
 	require.NoError(t, rethinkdb.SetupDB(session, dbName, []rethinkdb.Table{
 		TUFFilesRethinkTable,
+		ChangeRethinkTable,
 	}))
 	return NewRethinkDBStorage(dbName, "", "", session), cleanup
 }
@@ -169,3 +170,10 @@ func TestRethinkTUFMetaStoreGetCurrent(t *testing.T) {
 
 	testTUFMetaStoreGetCurrent(t, dbStore)
 }
+
+func TestRethinkDBGetChanges(t *testing.T) {
+	dbStore, cleanup := rethinkDBSetup(t)
+	defer cleanup()
+
+	testGetChanges(t, dbStore)
+}