e2db: fix Delete issues

- Scan object while it is being deleted to delete entries in
secondary/unique indexes for its current values. This requires that the
object be retrieved from storage prior to deletion.
- Add trailing slash to table key for prefixing.
- Add test cases.

Author: Kyle Travis

pkg/e2db/db_test.go

@@ -31,6 +31,7 @@ func init() {
 		RequiredClusterSize: 1,
 		HealthCheckInterval: 1 * time.Second,
 		HealthCheckTimeout:  5 * time.Second,
+		EtcdLogLevel:        zapcore.WarnLevel,
 	})
 	if err != nil {
 		log.Fatal(err)
@@ -217,72 +218,3 @@ func TestUpdate(t *testing.T) {
 		t.Errorf("e2db: after Update differs: (-want +got)\n%s", diff)
 	}
 }
-
-func TestDeletePrimaryIndex(t *testing.T) {
-	resetTable(t)
-	roles := db.Table(&Role{})
-
-	n, err := roles.Delete("ID", 10)
-	if err != nil {
-		t.Fatalf("unexpected error deleting non-existant key: %v", err)
-	}
-	if n != 0 {
-		t.Fatalf("expected zero rows affected when deleting non-existant key, got %d", n)
-	}
-
-	n, err = roles.Delete("ID", 1)
-	if err != nil {
-		t.Fatalf("unexpected error deleting by ID: %v", err)
-	}
-	if n != 1 {
-		t.Fatalf("expected one row affected when deleting by ID, got %d", n)
-	}
-
-	var r []*Role
-	if err := roles.All(&r); err != nil {
-		t.Fatal(err)
-	}
-	expected := []*Role{
-		{ID: 2, Name: "admin", Description: "administrator"},
-		{ID: 3, Name: "superadmin", Description: "administrator"},
-		{ID: 4, Name: "smoot", Description: "administrator"},
-	}
-	if diff := cmp.Diff(expected, r); diff != "" {
-		t.Errorf("e2db: after Update differs: (-want +got)\n%s", diff)
-	}
-}
-
-func TestDeleteSecondaryIndex(t *testing.T) {
-	resetTable(t)
-	roles := db.Table(&Role{})
-
-	n, err := roles.Delete("Name", "n/a")
-	if err != nil {
-		t.Fatalf("unexpected error deleting non-existant key: %v", err)
-	}
-	if n != 0 {
-		t.Fatalf("expected zero rows affected when deleting non-existent key, got %d", n)
-	}
-
-	n, err = roles.Delete("Name", "smoot")
-	if err != nil {
-		t.Fatalf("unexpected error deleting by Name: %v", err)
-	}
-	// NOTE(chris): includes value and index
-	if n != 2 {
-		t.Fatalf("expected one row affected when deleting by Name, got %d", n)
-	}
-
-	var r []*Role
-	if err := roles.All(&r); err != nil {
-		t.Fatal(err)
-	}
-	expected := []*Role{
-		{ID: 1, Name: "user", Description: "user"},
-		{ID: 2, Name: "admin", Description: "administrator"},
-		{ID: 3, Name: "superadmin", Description: "administrator"},
-	}
-	if diff := cmp.Diff(expected, r); diff != "" {
-		t.Errorf("e2db: after Update differs: (-want +got)\n%s", diff)
-	}
-}

pkg/e2db/key/key.go

@@ -27,7 +27,7 @@ func ID(model, key string) string {
 }
 
 func Table(model string) string {
-	return join(model)
+	return join(model) + "/"
 }
 
 func TableDef(model string) string {

pkg/e2db/model.go

@@ -82,6 +82,8 @@ func (f *FieldDef) indexKey(tableName string, value string) (string, error) {
 type ModelDef struct {
 	Name   string
 	Fields map[string]*FieldDef
+
+	t reflect.Type
 }
 
 func NewModelDef(t reflect.Type) *ModelDef {
@@ -94,6 +96,7 @@ func NewModelDef(t reflect.Type) *ModelDef {
 	m := &ModelDef{
 		Name:   t.Name(),
 		Fields: make(map[string]*FieldDef),
+		t:      t,
 	}
 	for i := 0; i < t.NumField(); i++ {
 		ft := t.Field(i)
@@ -116,6 +119,14 @@ func NewModelDef(t reflect.Type) *ModelDef {
 	return m
 }
 
+func (m *ModelDef) New() *reflect.Value {
+	if m.t == nil {
+		return nil
+	}
+	v := reflect.New(m.t)
+	return &v
+}
+
 type Field struct {
 	*FieldDef
 	value reflect.Value

pkg/e2db/query.go

@@ -255,6 +255,14 @@ func (q *query) Find(fieldName string, data interface{}, to interface{}) error {
 	}
 	k := toString(data)
 	if v.Type().Kind() == reflect.Slice {
+		if f.isPrimaryKey() {
+			item := reflect.New(v.Type().Elem())
+			if err := q.findOneByPrimaryKey(key.ID(q.t.meta.Name, k), reflect.Indirect(item)); err != nil {
+				return err
+			}
+			v.Set(reflect.Append(v, item.Elem()))
+			return nil
+		}
 		return q.findManyByIndex(key.Indexes(q.t.meta.Name, f.Name, k), v)
 	}
 	switch f.Type() {

pkg/e2db/tx.go

@@ -2,6 +2,8 @@ package e2db
 
 import (
 	"context"
+	"fmt"
+	"path/filepath"
 	"reflect"
 	"strings"
 	"time"
@@ -12,6 +14,7 @@ import (
 	"github.com/pkg/errors"
 	"go.etcd.io/etcd/clientv3"
 	"go.etcd.io/etcd/etcdserver/etcdserverpb"
+	"go.uber.org/zap"
 )
 
 var (
@@ -196,36 +199,102 @@ func (tx *Tx) Update(iface interface{}) error {
 	return err
 }
 
+// getIndexesByPrimaryKey returns all index keys for the provided primary key
+// based on the stored value.
+func (tx *Tx) getIndexesByPrimaryKey(pk string) ([]string, error) {
+	val := tx.meta.New()
+	if val == nil {
+		return nil, errors.Errorf("underlying type is uninitialized: %s", tx.meta.Name)
+	}
+	v := reflect.Indirect(reflect.ValueOf(val.Interface()))
+	if err := newQuery(tx.Table).findOneByPrimaryKey(pk, v); err != nil {
+		if errors.Cause(err) == ErrNoRows {
+			return nil, nil
+		}
+		return nil, err
+	}
+	keys := []string{pk}
+	_, id := filepath.Split(pk)
+	for n, f := range tx.meta.Fields {
+		switch f.Type() {
+		case UniqueIndex:
+			keys = append(keys, key.Unique(tx.meta.Name, n, toString(v.FieldByName(n).Interface())))
+		case SecondaryIndex:
+			keys = append(keys, key.Index(tx.meta.Name, n, toString(v.FieldByName(n).Interface()), id))
+		}
+	}
+	return keys, nil
+}
+
+func deleteOps(keys []string) (ops []clientv3.Op) {
+	for _, k := range keys {
+		ops = append(ops, clientv3.OpDelete(k))
+	}
+	return
+}
+
 func (tx *Tx) Delete(fieldName string, data interface{}) (int64, error) {
+	var n int64
+	st := time.Now()
+	defer func() {
+		log.Debug("tx.Delete",
+			zap.String("key", fmt.Sprintf("%s/%v", tx.meta.Name, fieldName)),
+			zap.String("q", toString(data)),
+			zap.Int64("n", n),
+			zap.Duration("elapsed", time.Now().Sub(st)),
+		)
+	}()
 	f, ok := tx.meta.Fields[fieldName]
 	if !ok {
 		return 0, errors.Errorf("invalid field name: %#v", fieldName)
 	}
 	k := toString(data)
-	if f.isPrimaryKey() {
-		resp, err := tx.db.client.Delete(context.TODO(), key.ID(tx.meta.Name, k))
+	pks := make([]string, 0)
+
+	// get the primary key of the item(s) being deleted
+	switch f.Type() {
+	case PrimaryKey:
+		pks = append(pks, key.ID(tx.meta.Name, k))
+	case UniqueIndex:
+		b, err := tx.db.client.Get(key.Unique(tx.meta.Name, fieldName, k))
 		if err != nil {
+			if errors.Cause(err) == client.ErrKeyNotFound {
+				return 0, nil
+			}
 			return 0, err
 		}
-		return resp.Deleted, nil
-	}
-	kvs, err := tx.db.client.Prefix(key.Indexes(tx.meta.Name, fieldName, k))
-	if err != nil {
-		if errors.Cause(err) == client.ErrKeyNotFound {
-			return 0, nil
+		pks = append(pks, string(b))
+	case SecondaryIndex:
+		x := key.Indexes(tx.meta.Name, fieldName, k)
+		kvs, err := tx.db.client.Prefix(x)
+		if err != nil {
+			if errors.Cause(err) == client.ErrKeyNotFound {
+				return 0, nil
+			}
+			return 0, err
 		}
-		return 0, err
+		for _, kv := range kvs {
+			pks = append(pks, string(kv.Value))
+		}
+	default:
+		return 0, errors.Wrapf(ErrNotIndexed, "cannot delete %#v", fieldName)
 	}
+
 	ops := make([]clientv3.Op, 0)
-	for _, kv := range kvs {
-		ops = append(ops, clientv3.OpDelete(string(kv.Key)))
-		ops = append(ops, clientv3.OpDelete(string(kv.Value)))
+	for _, pk := range pks {
+		keys, err := tx.getIndexesByPrimaryKey(pk)
+		if err != nil {
+			return 0, err
+		}
+		if len(keys) > 0 {
+			n++
+		}
+		ops = append(ops, deleteOps(keys)...)
 	}
-	resp, err := tx.batchOps(ops...)
-	if err != nil {
+	if _, err := tx.batchOps(ops...); err != nil {
 		return 0, err
 	}
-	return resp.Deleted, nil
+	return n, nil
 }
 
 func (tx *Tx) DeleteAll() error {

pkg/e2db/tx_test.go

@@ -3,6 +3,7 @@ package e2db
 import (
 	"testing"
 
+	"github.com/google/go-cmp/cmp"
 	"github.com/pkg/errors"
 )
 
@@ -24,3 +25,96 @@ func TestTxInsert(t *testing.T) {
 		t.Fatal(err)
 	}
 }
+
+func TestTxDelete(t *testing.T) {
+	cases := []struct {
+		name   string
+		field  string
+		value  interface{}
+		filter func(*Role) bool
+		n      int64
+		err    error
+	}{
+		{
+			name:   "delete absent key",
+			field:  "ID",
+			value:  20,
+			filter: func(r *Role) bool { return r.ID == 20 },
+		},
+		{
+			name:   "delete primary key",
+			field:  "ID",
+			value:  3,
+			n:      1,
+			filter: func(r *Role) bool { return r.ID == 3 },
+		},
+		{
+			name:   "delete secondary index",
+			field:  "Description",
+			value:  "administrator",
+			n:      3,
+			filter: func(r *Role) bool { return r.Description == "administrator" },
+		},
+		{
+			name:   "delete unique index",
+			field:  "Name",
+			value:  "smoot",
+			n:      1,
+			filter: func(r *Role) bool { return r.Name == "smoot" },
+		},
+		{
+			name:   "delete secondary index not found",
+			field:  "Description",
+			value:  "dinosaur",
+			filter: func(r *Role) bool { return r.Description == "dinosaur" },
+		},
+		{
+			name:   "delete unique index not found",
+			field:  "Name",
+			value:  "smootest",
+			filter: func(r *Role) bool { return r.Name == "smootest" },
+		},
+		{
+			name:  "delete non-index",
+			field: "NotIndexed",
+			value: "something",
+			err:   ErrNotIndexed,
+		},
+		// TODO(ktravis): make field/value a map to test multiple deletions in sequence
+	}
+	for _, c := range cases {
+		t.Run(c.name, func(t *testing.T) {
+			resetTable(t)
+			roles := db.Table(&Role{})
+			n, err := roles.Delete(c.field, c.value)
+			if errors.Cause(err) != c.err {
+				t.Fatalf("expected error %v, got %v", c.err, err)
+			}
+			if c.err == nil {
+				if err != nil {
+					t.Fatal(err)
+				}
+			} else {
+				return
+			}
+			if n != c.n {
+				t.Fatalf("expected delete count %d, got %d", c.n, n)
+			}
+
+			expected := append([]*Role{}, newRoles...)
+			for i := 0; i < len(expected); i++ {
+				if c.filter(expected[i]) {
+					expected = append(expected[:i], expected[i+1:]...)
+					i--
+				}
+			}
+			var result []*Role
+			if err := roles.All(&result); err != nil {
+				t.Fatal(err)
+			}
+			if diff := cmp.Diff(expected, result); diff != "" {
+				t.Fatal("results did not match expected", diff)
+			}
+		})
+	}
+}