6 files changed, 65 insertions, 5 deletions
diff --git a/entity/dag/entity.go b/entity/dag/entity.go
index 2028e1b4..f8dbd53d 100644
--- a/entity/dag/entity.go
+++ b/entity/dag/entity.go
@@ -314,6 +314,9 @@ func ReadAll[EntityT entity.Interface](def Definition, wrapper func(e *Entity) E
 			return
 		}
 
+		total := int64(len(refs))
+		current := int64(1)
+
 		for _, ref := range refs {
 			e, err := read[EntityT](def, wrapper, repo, resolvers, ref)
 
@@ -322,7 +325,12 @@ func ReadAll[EntityT entity.Interface](def Definition, wrapper func(e *Entity) E
 				return
 			}
 
-			out <- entity.StreamedEntity[EntityT]{Entity: e}
+			out <- entity.StreamedEntity[EntityT]{
+				Entity:        e,
+				CurrentEntity: current,
+				TotalEntities: total,
+			}
+			current++
 		}
 	}()
 
diff --git a/entity/dag/entity_actions.go b/entity/dag/entity_actions.go
index 97a68c36..5f0abec3 100644
--- a/entity/dag/entity_actions.go
+++ b/entity/dag/entity_actions.go
@@ -258,3 +258,19 @@ func Remove(def Definition, repo repository.ClockedRepo, id entity.Id) error {
 
 	return nil
 }
+
+// RemoveAll delete all Entity matching the Definition.
+// RemoveAll is idempotent.
+func RemoveAll(def Definition, repo repository.ClockedRepo) error {
+	localIds, err := ListLocalIds(def, repo)
+	if err != nil {
+		return err
+	}
+	for _, id := range localIds {
+		err = Remove(def, repo, id)
+		if err != nil {
+			return err
+		}
+	}
+	return nil
+}
diff --git a/entity/dag/entity_actions_test.go b/entity/dag/entity_actions_test.go
index fd219644..6181614b 100644
--- a/entity/dag/entity_actions_test.go
+++ b/entity/dag/entity_actions_test.go
@@ -406,3 +406,34 @@ func TestRemove(t *testing.T) {
 	err = Remove(def, repoA, e.Id())
 	require.NoError(t, err)
 }
+
+func TestRemoveAll(t *testing.T) {
+	repoA, _, _, id1, _, resolvers, def := makeTestContextRemote(t)
+
+	var ids []entity.Id
+
+	for i := 0; i < 10; i++ {
+		e := New(def)
+		e.Append(newOp1(id1, "foo"))
+		require.NoError(t, e.Commit(repoA))
+		ids = append(ids, e.Id())
+	}
+
+	_, err := Push(def, repoA, "remote")
+	require.NoError(t, err)
+
+	err = RemoveAll(def, repoA)
+	require.NoError(t, err)
+
+	for _, id := range ids {
+		_, err = Read(def, wrapper, repoA, resolvers, id)
+		require.Error(t, err)
+
+		_, err = readRemote(def, wrapper, repoA, resolvers, "remote", id)
+		require.Error(t, err)
+	}
+
+	// Remove is idempotent
+	err = RemoveAll(def, repoA)
+	require.NoError(t, err)
+}
diff --git a/entity/id.go b/entity/id.go
index 49398da8..0949bf92 100644
--- a/entity/id.go
+++ b/entity/id.go
@@ -11,7 +11,7 @@ import (
 
 // sha-256
 const idLength = 64
-const humanIdLength = 7
+const HumanIdLength = 7
 
 const UnsetId = Id("unset")
 
@@ -34,7 +34,7 @@ func (i Id) String() string {
 
 // Human return the identifier, shortened for human consumption
 func (i Id) Human() string {
-	format := fmt.Sprintf("%%.%ds", humanIdLength)
+	format := fmt.Sprintf("%%.%ds", HumanIdLength)
 	return fmt.Sprintf(format, i)
 }
 
diff --git a/entity/id_interleaved.go b/entity/id_interleaved.go
index 28c59a42..7ae6d72e 100644
--- a/entity/id_interleaved.go
+++ b/entity/id_interleaved.go
@@ -22,7 +22,7 @@ func (ci CombinedId) String() string {
 
 // Human return the identifier, shortened for human consumption
 func (ci CombinedId) Human() string {
-	format := fmt.Sprintf("%%.%ds", humanIdLength)
+	format := fmt.Sprintf("%%.%ds", HumanIdLength)
 	return fmt.Sprintf(format, ci)
 }
 
diff --git a/entity/streamed.go b/entity/streamed.go
index 789224a3..33124ef0 100644
--- a/entity/streamed.go
+++ b/entity/streamed.go
@@ -1,6 +1,11 @@
 package entity
 
 type StreamedEntity[EntityT Interface] struct {
-	Entity EntityT
 	Err    error
+	Entity EntityT
+
+	// CurrentEntity is the index of the current entity being streamed, to express progress.
+	CurrentEntity int64
+	// TotalEntities is the total count of expected entities, if known.
+	TotalEntities int64
 }