use resource kind topic for permissions replay

lib/replay/replay.go

@@ -23,7 +23,7 @@ import (
 	"github.com/SENERGY-Platform/permission-search/lib/configuration"
 	"github.com/SENERGY-Platform/permission-search/lib/model"
 	"github.com/SENERGY-Platform/permission-search/lib/opensearchclient"
-	"github.com/SENERGY-Platform/permission-search/lib/worker/kafka"
+	"github.com/SENERGY-Platform/permission-search/lib/rigthsproducer"
 	"github.com/opensearch-project/opensearch-go"
 	"github.com/opensearch-project/opensearch-go/opensearchutil"
 	"runtime/debug"
@@ -51,9 +51,10 @@ func ReplayPermissions(config configuration.Config, args []string) {
 	}
 	ctx, cancel := context.WithCancel(context.Background())
 	defer cancel()
-	var producer *kafka.Producer
+
+	var producer *rigthsproducer.Producer
 	if !dryrun {
-		producer, err = kafka.NewProducer(ctx, config.KafkaUrl, config.PermTopic, true)
+		producer, err = rigthsproducer.New(ctx, config)
 	}
 	if err != nil {
 		fmt.Println("ERROR:", err)
@@ -65,17 +66,12 @@ func ReplayPermissions(config configuration.Config, args []string) {
 	}
 }
 
-func ReplayPermissionsOfResourceKind(producer *kafka.Producer, client *opensearch.Client, kind string, batchSize int) {
-	for command := range GetCommands(client, kind, batchSize) {
-		msg, err := json.Marshal(command)
-		if err != nil {
-			fmt.Println("ERROR:", err)
-			debug.PrintStack()
-			return
-		}
-		fmt.Println(string(msg))
+func ReplayPermissionsOfResourceKind(producer *rigthsproducer.Producer, client *opensearch.Client, kind string, batchSize int) {
+	for entry := range GetEntries(client, kind, batchSize) {
+		rights := entry.ToResourceRights().ResourceRightsBase
+		fmt.Printf("%#v %#v %#v\n", kind, entry.Resource, rights)
 		if producer != nil {
-			err = producer.Produce(command.Resource+"_"+command.User+"_"+command.Group, msg)
+			err, _ := producer.SetResourceRights(kind, entry.Resource, rights, "")
 			if err != nil {
 				fmt.Println("ERROR:", err)
 				debug.PrintStack()
@@ -85,62 +81,6 @@ func ReplayPermissionsOfResourceKind(producer *kafka.Producer, client *opensearc
 	}
 }
 
-func GetCommands(client *opensearch.Client, kind string, batchSize int) (commands chan model.PermCommandMsg) {
-	commands = make(chan model.PermCommandMsg)
-	entries := GetEntries(client, kind, batchSize)
-	go func() {
-		defer close(commands)
-		for entry := range entries {
-			userRight := map[string]string{}
-			for _, user := range entry.ReadUsers {
-				userRight[user] = userRight[user] + "r"
-			}
-			for _, user := range entry.WriteUsers {
-				userRight[user] = userRight[user] + "w"
-			}
-			for _, user := range entry.ExecuteUsers {
-				userRight[user] = userRight[user] + "x"
-			}
-			for _, user := range entry.AdminUsers {
-				userRight[user] = userRight[user] + "a"
-			}
-			for user, right := range userRight {
-				commands <- model.PermCommandMsg{
-					Command:  "PUT",
-					Kind:     kind,
-					Resource: entry.Resource,
-					User:     user,
-					Right:    right,
-				}
-			}
-			groupRight := map[string]string{}
-			for _, group := range entry.ReadGroups {
-				groupRight[group] = groupRight[group] + "r"
-			}
-			for _, group := range entry.WriteGroups {
-				groupRight[group] = groupRight[group] + "w"
-			}
-			for _, group := range entry.ExecuteGroups {
-				groupRight[group] = groupRight[group] + "x"
-			}
-			for _, group := range entry.AdminGroups {
-				groupRight[group] = groupRight[group] + "a"
-			}
-			for group, right := range groupRight {
-				commands <- model.PermCommandMsg{
-					Command:  "PUT",
-					Kind:     kind,
-					Resource: entry.Resource,
-					Group:    group,
-					Right:    right,
-				}
-			}
-
-		}
-	}()
-	return commands
-}
-
 func GetEntries(client *opensearch.Client, kind string, batchSize int) (entries chan model.Entry) {
 	lastId := ""
 	entries = make(chan model.Entry, batchSize)