More admin data delete endpoints (#361)

file/handler/files.go

@@ -10,6 +10,8 @@ import (
 	log "github.com/micro/micro/v3/service/logger"
 	"github.com/micro/micro/v3/service/store"
 	file "github.com/micro/services/file/proto"
+	pauth "github.com/micro/services/pkg/auth"
+	adminpb "github.com/micro/services/pkg/service/proto"
 	"github.com/micro/services/pkg/tenant"
 )
 
@@ -150,3 +152,32 @@ func (e *File) List(ctx context.Context, req *file.ListRequest, rsp *file.ListRe
 
 	return nil
 }
+
+func (e *File) DeleteData(ctx context.Context, request *adminpb.DeleteDataRequest, response *adminpb.DeleteDataResponse) error {
+	method := "admin.DeleteData"
+	_, err := pauth.VerifyMicroAdmin(ctx, method)
+	if err != nil {
+		return err
+	}
+
+	if len(request.TenantId) == 0 {
+		return errors.BadRequest(method, "Missing tenant ID")
+	}
+
+	path := filepath.Join("file", request.TenantId)
+
+	// read all the files for the project
+	records, err := store.List(store.ListPrefix(path))
+	if err != nil {
+		return err
+	}
+
+	for _, file := range records {
+		if err := store.Delete(file); err != nil {
+			return err
+		}
+	}
+	log.Infof("Deleted %d records for %s", len(records), request.TenantId)
+
+	return nil
+}

file/main.go

@@ -3,6 +3,7 @@ package main
 import (
 	"github.com/micro/services/file/handler"
 	pb "github.com/micro/services/file/proto"
+	admin "github.com/micro/services/pkg/service/proto"
 	"github.com/micro/services/pkg/tracing"
 
 	"github.com/micro/micro/v3/service"
@@ -16,8 +17,10 @@ func main() {
 		service.Version("latest"),
 	)
 
+	h := handler.NewFile()
 	// Register handler
-	pb.RegisterFileHandler(srv.Server(), handler.NewFile())
+	pb.RegisterFileHandler(srv.Server(), h)
+	admin.RegisterAdminHandler(srv.Server(), h)
 
 	traceCloser := tracing.SetupOpentracing("file")
 	defer traceCloser.Close()

location/domain/domain.go

@@ -111,7 +111,6 @@ func Search(ctx context.Context, typ string, entity *Entity, radius float64, num
 
 	// get the index
 	index := getIndex(ctx)
-
 	points := index.KNearest(entity, numEntities, geo.Meters(radius), func(p geo.Point) bool {
 		e, ok := p.(*Entity)
 		if !ok || e.Type != typ {
@@ -132,3 +131,11 @@ func Search(ctx context.Context, typ string, entity *Entity, radius float64, num
 
 	return entities
 }
+
+func DeleteIndex(tenantID string) error {
+	mtx.Lock()
+	defer mtx.Unlock()
+	delete(indexes, tenantID)
+
+	return nil
+}

location/handler/handler.go

@@ -6,9 +6,12 @@ import (
 
 	"github.com/micro/micro/v3/service"
 	"github.com/micro/micro/v3/service/errors"
+	"github.com/micro/micro/v3/service/logger"
 	"github.com/micro/services/location/domain"
 	loc "github.com/micro/services/location/proto"
 	"github.com/micro/services/location/subscriber"
+	pauth "github.com/micro/services/pkg/auth"
+	adminpb "github.com/micro/services/pkg/service/proto"
 )
 
 type Location struct{}
@@ -70,3 +73,21 @@ func (l *Location) Search(ctx context.Context, req *loc.SearchRequest, rsp *loc.
 
 	return nil
 }
+
+func (l *Location) DeleteData(ctx context.Context, request *adminpb.DeleteDataRequest, response *adminpb.DeleteDataResponse) error {
+	method := "admin.DeleteData"
+	_, err := pauth.VerifyMicroAdmin(ctx, method)
+	if err != nil {
+		return err
+	}
+
+	if len(request.TenantId) == 0 {
+		return errors.BadRequest(method, "Missing tenant ID")
+	}
+
+	if err := domain.DeleteIndex(request.TenantId); err != nil {
+		return err
+	}
+	logger.Infof("Deleted index for %s", request.TenantId)
+	return nil
+}

location/main.go

@@ -6,6 +6,7 @@ import (
 	"github.com/micro/micro/v3/service"
 	"github.com/micro/services/location/handler"
 	pb "github.com/micro/services/location/proto"
+	admin "github.com/micro/services/pkg/service/proto"
 	"github.com/micro/services/pkg/tracing"
 )
 
@@ -14,7 +15,9 @@ func main() {
 		service.Name("location"),
 	)
 
-	pb.RegisterLocationHandler(location.Server(), new(handler.Location))
+	h := new(handler.Location)
+	pb.RegisterLocationHandler(location.Server(), h)
+	admin.RegisterAdminHandler(location.Server(), h)
 
 	// TODO reinstate me
 	//service.Subscribe(subscriber.Topic, new(subscriber.Location))

notes/handler/notes.go

@@ -9,9 +9,12 @@ import (
 	"github.com/google/uuid"
 	"github.com/micro/micro/v3/service/client"
 	"github.com/micro/micro/v3/service/errors"
+	"github.com/micro/micro/v3/service/logger"
 	"github.com/micro/micro/v3/service/store"
 	streamPb "github.com/micro/services/mq/proto"
 	pb "github.com/micro/services/notes/proto"
+	pauth "github.com/micro/services/pkg/auth"
+	adminpb "github.com/micro/services/pkg/service/proto"
 	"github.com/micro/services/pkg/tenant"
 	"google.golang.org/protobuf/types/known/structpb"
 )
@@ -292,3 +295,29 @@ func (h *Notes) List(ctx context.Context, req *pb.ListRequest, rsp *pb.ListRespo
 
 	return nil
 }
+
+func (h *Notes) DeleteData(ctx context.Context, request *adminpb.DeleteDataRequest, response *adminpb.DeleteDataResponse) error {
+	method := "admin.DeleteData"
+	_, err := pauth.VerifyMicroAdmin(ctx, method)
+	if err != nil {
+		return err
+	}
+
+	if len(request.TenantId) == 0 {
+		return errors.BadRequest(method, "Missing tenant ID")
+	}
+
+	keys, err := store.List(store.ListPrefix(request.TenantId))
+	if err != nil {
+		return err
+	}
+
+	for _, k := range keys {
+		if err := store.Delete(k); err != nil {
+			return err
+		}
+	}
+
+	logger.Infof("Deleted %d keys for %s", len(keys), request.TenantId)
+	return nil
+}

notes/main.go

@@ -5,6 +5,7 @@ import (
 	log "github.com/micro/micro/v3/service/logger"
 	"github.com/micro/services/notes/handler"
 	pb "github.com/micro/services/notes/proto"
+	admin "github.com/micro/services/pkg/service/proto"
 )
 
 func main() {
@@ -17,8 +18,10 @@ func main() {
 	// Initialise service
 	srv.Init()
 
+	h := handler.New(srv.Client())
 	// Register Handler
-	pb.RegisterNotesHandler(srv.Server(), handler.New(srv.Client()))
+	pb.RegisterNotesHandler(srv.Server(), h)
+	admin.RegisterAdminHandler(srv.Server(), h)
 
 	// Run service
 	if err := srv.Run(); err != nil {

rss/handler/crawl.go

@@ -51,6 +51,7 @@ func (e *crawl) FetchAll() {
 		return
 	}
 
+	currList := map[string]bool{}
 	for _, v := range records {
 		feed := pb.Feed{}
 		if err := json.Unmarshal(v.Value, &feed); err != nil {
@@ -62,6 +63,22 @@ func (e *crawl) FetchAll() {
 		if err != nil {
 			log.Errorf("Error saving post: %v", err)
 		}
+		currList[feed.Url] = true
+	}
+
+	// prune anything that has been deleted
+	rssSync.Lock()
+	defer rssSync.Unlock()
+	for url, _ := range rssFeeds {
+		if currList[url] {
+			continue
+		}
+		// this isn't in the current list. delete from store any entries
+		keys, _ := store.List(store.ListPrefix(generateEntryKey(url, "")))
+		for _, k := range keys {
+			store.Delete(k)
+		}
+		delete(rssFeeds, url)
 	}
 }
 

rss/handler/rss.go

@@ -5,10 +5,13 @@ import (
 	"encoding/json"
 	"fmt"
 	"hash/fnv"
+	"strings"
 
 	"github.com/micro/micro/v3/service/errors"
 	log "github.com/micro/micro/v3/service/logger"
 	"github.com/micro/micro/v3/service/store"
+	pauth "github.com/micro/services/pkg/auth"
+	adminpb "github.com/micro/services/pkg/service/proto"
 
 	"github.com/micro/services/pkg/tenant"
 	pb "github.com/micro/services/rss/proto"
@@ -175,3 +178,32 @@ func (e *Rss) Remove(ctx context.Context, req *pb.RemoveRequest, rsp *pb.RemoveR
 
 	return e.store.Delete(generateFeedKey(ctx, req.Name))
 }
+
+func (e *Rss) DeleteData(ctx context.Context, request *adminpb.DeleteDataRequest, response *adminpb.DeleteDataResponse) error {
+	method := "admin.DeleteData"
+	_, err := pauth.VerifyMicroAdmin(ctx, method)
+	if err != nil {
+		return err
+	}
+
+	if len(request.TenantId) == 0 {
+		return errors.BadRequest(method, "Missing tenant ID")
+	}
+	split := strings.Split(request.TenantId, "/")
+	tctx := tenant.NewContext(split[1], split[0], split[1])
+
+	prefix := generateFeedKey(tctx, "")
+	records, err := e.store.Read(prefix, store.ReadPrefix())
+	if err != nil {
+		return err
+	}
+
+	for _, val := range records {
+		if err := e.store.Delete(val.Key); err != nil {
+			return err
+		}
+	}
+	log.Infof("Delete %d records for %s", len(records), request.TenantId)
+	return nil
+
+}

rss/main.go

@@ -6,6 +6,7 @@ import (
 	"github.com/micro/micro/v3/service"
 	"github.com/micro/micro/v3/service/logger"
 	"github.com/micro/micro/v3/service/store"
+	admin "github.com/micro/services/pkg/service/proto"
 
 	"github.com/micro/services/pkg/tracing"
 	"github.com/micro/services/rss/handler"
@@ -33,6 +34,7 @@ func main() {
 
 	// Register handler
 	pb.RegisterRssHandler(srv.Server(), rss)
+	admin.RegisterAdminHandler(srv.Server(), rss)
 	traceCloser := tracing.SetupOpentracing("rss")
 	defer traceCloser.Close()