create registry upon ecr provisioning'

cmd/app/main.go

@@ -98,7 +98,7 @@ func main() {
 
 	errorChan := make(chan error)
 
-	go prov.GlobalStreamListener(redis, repo.AWSInfra, errorChan)
+	go prov.GlobalStreamListener(redis, *repo, errorChan)
 
 	if err := s.ListenAndServe(); err != nil && err != http.ErrServerClosed {
 		log.Fatal("Server startup failed", err)

internal/kubernetes/provisioner/global_stream.go

@@ -2,6 +2,7 @@ package provisioner
 
 import (
 	"context"
+	"encoding/json"
 	"fmt"
 
 	"github.com/porter-dev/porter/internal/repository"
@@ -72,11 +73,11 @@ type ResourceCRUDHandler interface {
 	OnCreate(id uint) error
 }
 
-// GlobalStreamListener performs an XREADGROUP operation on a given stream
-// and sends a GlobalStreamMessage to the msgChan
+// GlobalStreamListener performs an XREADGROUP operation on a given stream and
+// updates models in the database as necessary
 func GlobalStreamListener(
 	client *redis.Client,
-	infraRepo repository.AWSInfraRepository,
+	repo repository.Repository,
 	errorChan chan error,
 ) {
 	for {
@@ -98,10 +99,10 @@ func GlobalStreamListener(
 		// parse messages from the global stream
 		for _, msg := range xstreams[0].Messages {
 			// parse the id to identify the infra
-			infraID, err := models.GetInfraIDFromWorkspaceID(fmt.Sprintf("%v", msg.Values["id"]))
+			kind, projID, infraID, err := models.ParseWorkspaceID(fmt.Sprintf("%v", msg.Values["id"]))
 
 			if fmt.Sprintf("%v", msg.Values["status"]) == "created" {
-				infra, err := infraRepo.ReadAWSInfra(infraID)
+				infra, err := repo.AWSInfra.ReadAWSInfra(infraID)
 
 				if err != nil {
 					continue
@@ -109,11 +110,30 @@ func GlobalStreamListener(
 
 				infra.Status = models.StatusCreated
 
-				infra, err = infraRepo.UpdateAWSInfra(infra)
+				infra, err = repo.AWSInfra.UpdateAWSInfra(infra)
 
 				if err != nil {
 					continue
 				}
+
+				// create ECR/EKS
+				if kind == string(models.AWSInfraECR) {
+					reg := &models.Registry{
+						ProjectID:        projID,
+						AWSIntegrationID: infra.AWSIntegrationID,
+					}
+
+					// parse raw data into ECR type
+					bytes, _ := msg.Values["data"].([]byte)
+
+					json.Unmarshal(bytes, reg)
+
+					reg, err := repo.Registry.CreateRegistry(reg)
+
+					if err != nil {
+						continue
+					}
+				}
 			}
 
 			// acknowledge the message as read

internal/models/infra.go

@@ -73,19 +73,25 @@ func (ai *AWSInfra) GetWorkspaceID() string {
 	return fmt.Sprintf("%s-%d-%d", ai.Kind, ai.ProjectID, ai.ID)
 }
 
-// GetInfraIDFromWorkspaceID returns the infra id given a workspace id
-func GetInfraIDFromWorkspaceID(workspaceID string) (uint, error) {
+// ParseWorkspaceID returns the (kind, projectID, infraID)
+func ParseWorkspaceID(workspaceID string) (string, uint, uint, error) {
 	strArr := strings.Split(workspaceID, "-")
 
 	if len(strArr) != 3 {
-		return 0, fmt.Errorf("workspace id improperly formatted")
+		return "", 0, 0, fmt.Errorf("workspace id improperly formatted")
 	}
 
-	u, err := strconv.ParseUint(strArr[2], 10, 64)
+	projID, err := strconv.ParseUint(strArr[1], 10, 64)
 
 	if err != nil {
-		return 0, err
+		return "", 0, 0, err
 	}
 
-	return uint(u), nil
+	infraID, err := strconv.ParseUint(strArr[2], 10, 64)
+
+	if err != nil {
+		return "", 0, 0, err
+	}
+
+	return strArr[0], uint(projID), uint(infraID), nil
 }