From fa32346860a616367efae19aac1878d8a6f13331 Mon Sep 17 00:00:00 2001
From: mpl <mathieu.lonjaret@gmail.com>
Date: Thu, 28 Nov 2013 17:46:36 +0100
Subject: [PATCH] pkg/index: check for deletions in EdgesTo

Also introduced and used kvEdgeBackward

http://camlistore.org/issue/191

Change-Id: Ifdca9f2453ec0c7db55afc40cf0b5f1441b83158
---
 pkg/index/index.go           | 66 +++++++++++++++++++++++-------------
 pkg/index/indextest/tests.go | 40 ++++++++++++++++++++--
 pkg/types/camtypes/search.go |  1 +
 3 files changed, 80 insertions(+), 27 deletions(-)

diff --git a/pkg/index/index.go b/pkg/index/index.go
index dc0e6a076..d992cf321 100644
--- a/pkg/index/index.go
+++ b/pkg/index/index.go
@@ -952,43 +952,61 @@ func (x *Index) GetImageInfo(fileRef blob.Ref) (camtypes.ImageInfo, error) {
 func (x *Index) EdgesTo(ref blob.Ref, opts *camtypes.EdgesToOpts) (edges []*camtypes.Edge, err error) {
 	it := x.queryPrefix(keyEdgeBackward, ref)
 	defer closeIterator(it, &err)
-	permanodeParents := map[string]blob.Ref{} // blobref key => blobref set
+	permanodeParents := make(map[string]*camtypes.Edge)
 	for it.Next() {
-		keyPart := strings.Split(it.Key(), "|")[1:]
-		if len(keyPart) < 2 {
-			continue
-		}
-		parent := keyPart[1]
-		parentRef, ok := blob.Parse(parent)
+		edge, ok := kvEdgeBackward(it.Key(), it.Value())
 		if !ok {
 			continue
 		}
-		valPart := strings.Split(it.Value(), "|")
-		if len(valPart) < 2 {
+		if x.IsDeleted(edge.From) {
 			continue
 		}
-		parentType, parentName := valPart[0], valPart[1]
-		if parentType == "permanode" {
-			permanodeParents[parent] = parentRef
+		if x.IsDeleted(edge.BlobRef) {
+			continue
+		}
+		edge.To = ref
+		if edge.FromType == "permanode" {
+			permanodeParents[edge.From.String()] = edge
 		} else {
-			edges = append(edges, &camtypes.Edge{
-				From:      parentRef,
-				FromType:  parentType,
-				FromTitle: parentName,
-				To:        ref,
-			})
+			edges = append(edges, edge)
 		}
 	}
-	for _, parentRef := range permanodeParents {
-		edges = append(edges, &camtypes.Edge{
-			From:     parentRef,
-			FromType: "permanode",
-			To:       ref,
-		})
+	for _, e := range permanodeParents {
+		edges = append(edges, e)
 	}
 	return edges, nil
 }
 
+func kvEdgeBackward(k, v string) (edge *camtypes.Edge, ok bool) {
+	// TODO(bradfitz): garbage
+	keyPart := strings.Split(k, "|")
+	valPart := strings.Split(v, "|")
+	if len(keyPart) != 4 || len(valPart) != 2 {
+		// TODO(mpl): use glog
+		log.Printf("bogus keyEdgeBackward index entry: %q = %q", k, v)
+		return
+	}
+	if keyPart[0] != "edgeback" {
+		return
+	}
+	parentRef, ok := blob.Parse(keyPart[2])
+	if !ok {
+		log.Printf("bogus parent in keyEdgeBackward index entry: %q", keyPart[2])
+		return
+	}
+	blobRef, ok := blob.Parse(keyPart[3])
+	if !ok {
+		log.Printf("bogus blobref in keyEdgeBackward index entry: %q", keyPart[3])
+		return
+	}
+	return &camtypes.Edge{
+		From:      parentRef,
+		FromType:  valPart[0],
+		FromTitle: valPart[1],
+		BlobRef:   blobRef,
+	}, true
+}
+
 // GetDirMembers sends on dest the children of the static directory dir.
 func (x *Index) GetDirMembers(dir blob.Ref, dest chan<- blob.Ref, limit int) (err error) {
 	defer close(dest)
diff --git a/pkg/index/indextest/tests.go b/pkg/index/indextest/tests.go
index c9c5d4064..a1127b541 100644
--- a/pkg/index/indextest/tests.go
+++ b/pkg/index/indextest/tests.go
@@ -825,16 +825,15 @@ func EdgesTo(t *testing.T, initIdx func() *index.Index) {
 	idx := initIdx()
 	id := NewIndexDeps(idx)
 	id.Fataler = t
+	defer id.DumpIndex(t)
 
 	// pn1 ---member---> pn2
 	pn1 := id.NewPermanode()
 	pn2 := id.NewPermanode()
-	id.AddAttribute(pn1, "camliMember", pn2.String())
+	claim1 := id.AddAttribute(pn1, "camliMember", pn2.String())
 
 	t.Logf("edge %s --> %s", pn1, pn2)
 
-	id.DumpIndex(t)
-
 	// Look for pn1
 	{
 		edges, err := idx.EdgesTo(pn2, nil)
@@ -853,6 +852,41 @@ func EdgesTo(t *testing.T, initIdx func() *index.Index) {
 			t.Errorf("Wrong edge.\n GOT: %v\nWANT: %v", got, want)
 		}
 	}
+
+	// Delete claim -> break edge relationship.
+	del1 := id.Delete(claim1)
+	t.Logf("del claim %q deletes claim %q, breaks link between p1 and p2", del1, claim1)
+	// test that we can't find anymore pn1 from pn2
+	{
+		edges, err := idx.EdgesTo(pn2, nil)
+		if err != nil {
+			t.Fatal(err)
+		}
+		if len(edges) != 0 {
+			t.Fatalf("num edges = %d; want 0", len(edges))
+		}
+	}
+
+	// Undelete, should restore the link.
+	del2 := id.Delete(del1)
+	t.Logf("del claim %q deletes del claim %q, restores link between p1 and p2", del2, del1)
+	{
+		edges, err := idx.EdgesTo(pn2, nil)
+		if err != nil {
+			t.Fatal(err)
+		}
+		if len(edges) != 1 {
+			t.Fatalf("num edges = %d; want 1", len(edges))
+		}
+		wantEdge := &camtypes.Edge{
+			From:     pn1,
+			To:       pn2,
+			FromType: "permanode",
+		}
+		if got, want := edges[0].String(), wantEdge.String(); got != want {
+			t.Errorf("Wrong edge.\n GOT: %v\nWANT: %v", got, want)
+		}
+	}
 }
 
 func Delete(t *testing.T, initIdx func() *index.Index) {
diff --git a/pkg/types/camtypes/search.go b/pkg/types/camtypes/search.go
index 7ec97f0fb..49ffd471f 100644
--- a/pkg/types/camtypes/search.go
+++ b/pkg/types/camtypes/search.go
@@ -157,6 +157,7 @@ type Edge struct {
 	FromType  string // "permanode", "directory", etc
 	FromTitle string // name of source permanode or directory
 	To        blob.Ref
+	BlobRef   blob.Ref // the blob responsible for the edge relationship
 }
 
 func (e *Edge) String() string {