Feat/owned cluster

apps/console/internal/app/adapters/infra-service/infra-service.go

@@ -30,6 +30,21 @@ func (s *InfraService) EnsureGlobalVPNConnection(ctx context.Context, args ports
 	return nil
 }
 
+func (s *InfraService) GetByokClusterOwnedBy(ctx context.Context, args ports.IsClusterLabelsIn) (string, error) {
+	cl, err := s.infraClient.GetCluster(ctx, &infra.GetClusterIn{
+		UserId:      args.UserId,
+		UserName:    args.UserName,
+		UserEmail:   args.UserEmail,
+		AccountName: args.AccountName,
+		ClusterName: args.ClusterName,
+	})
+	if err != nil {
+		return "", err
+	}
+
+	return cl.OwnedBy, nil
+}
+
 var _ ports.InfraService = (*InfraService)(nil)
 
 func NewInfraService(infraClient infra.InfraClient) ports.InfraService {

apps/console/internal/app/webhook-consumer.go

@@ -3,6 +3,8 @@ package app
 import (
 	"context"
 	"encoding/json"
+	"fmt"
+
 	"github.com/kloudlite/api/apps/console/internal/domain"
 	"github.com/kloudlite/api/pkg/errors"
 	"github.com/kloudlite/api/pkg/logging"
@@ -18,25 +20,29 @@ func processWebhooks(consumer WebhookConsumer, d domain.Domain, logger logging.L
 			logger.Infof("finished processing message")
 		}()
 
-		webhook := &domain.ImageHookPayload{}
-		if err := json.Unmarshal(msg.Payload, &webhook); err != nil {
+		hook := &domain.ImageHookPayload{}
+		if err := json.Unmarshal(msg.Payload, &hook); err != nil {
 			logger.Errorf(err, "could not unmarshal into *ImageHookPayload")
 			return errors.NewE(err)
 		}
-		if webhook.Image == "" || webhook.AccountName == "" {
+		if hook.Image == "" || hook.AccountName == "" {
 			return errors.Newf("invalid webhook payload")
 		}
-		hook := &domain.ImageHookPayload{
-			Image:       webhook.Image,
-			AccountName: webhook.AccountName,
-			Meta:        webhook.Meta,
-		}
 
-		_, err := d.CreateRegistryImage(context.TODO(), hook.AccountName, hook.Image, hook.Meta)
+		_, err := d.UpsertRegistryImage(context.TODO(), hook.AccountName, hook.Image, hook.Meta)
 		if err != nil {
 			logger.Errorf(err, "could not process image hook")
 			return errors.NewE(err)
 		}
+
+		// domain.NewConsoleContext(ctx, userId repos.ID, accountName string)
+		dctx := domain.NewConsoleContext(context.TODO(), "sys-user:apply-on-error-worker", hook.AccountName)
+
+		if err := d.RolloutAppsByImage(dctx, fmt.Sprintf("%s:%s", hook.Image, hook.Image)); err != nil {
+			logger.Errorf(err, "could not rollout apps by image")
+			return errors.NewE(err)
+		}
+
 		return nil
 	}, msgTypes.ConsumeOpts{
 		OnError: func(err error) error {

apps/console/internal/domain/api.go

@@ -183,7 +183,7 @@ type Domain interface {
 	GetRegistryImageURL(ctx ConsoleContext) (*entities.RegistryImageURL, error)
 	GetRegistryImage(ctx ConsoleContext, image string) (*entities.RegistryImage, error)
 	DeleteRegistryImage(ctx ConsoleContext, image string) error
-	CreateRegistryImage(ctx context.Context, accountName string, image string, meta map[string]any) (*entities.RegistryImage, error)
+	UpsertRegistryImage(ctx context.Context, accountName string, image string, meta map[string]any) (*entities.RegistryImage, error)
 	ListRegistryImages(ctx ConsoleContext, pq repos.CursorPagination) (*repos.PaginatedRecord[*entities.RegistryImage], error)
 	SearchRegistryImages(ctx ConsoleContext, query string) ([]*entities.RegistryImage, error)
 
@@ -203,6 +203,7 @@ type Domain interface {
 	OnAppUpdateMessage(ctx ResourceContext, app entities.App, status types.ResourceStatus, opts UpdateAndDeleteOpts) error
 
 	ResyncApp(ctx ResourceContext, name string) error
+	RolloutAppsByImage(ctx ConsoleContext, imageName string) error
 
 	ListConfigs(ctx ResourceContext, search map[string]repos.MatchFilter, pq repos.CursorPagination) (*repos.PaginatedRecord[*entities.Config], error)
 	GetConfig(ctx ResourceContext, name string) (*entities.Config, error)

apps/console/internal/domain/app.go

@@ -345,3 +345,48 @@ func (d *domain) ResyncApp(ctx ResourceContext, name string) error {
 	}
 	return d.resyncK8sResource(ctx, a.EnvironmentName, a.SyncStatus.Action, &a.App, a.RecordVersion)
 }
+
+func (d *domain) listAppsByImage(ctx ConsoleContext, image string) ([]*entities.App, error) {
+	apps, err := d.appRepo.Find(ctx, repos.Query{
+		Filter: repos.Filter{
+			fields.AccountName: ctx.AccountName,
+			fmt.Sprintf("%s.image", fc.AppSpecContainers):           image,
+			fmt.Sprintf("%s.imagePullPolicy", fc.AppSpecContainers): "Always",
+		},
+		Sort: nil,
+	})
+	if err != nil {
+		return nil, errors.NewE(err)
+	}
+	return apps, nil
+}
+
+func (d *domain) RolloutAppsByImage(ctx ConsoleContext, imageName string) error {
+
+	iName, iTag := getImageNameTag(imageName)
+
+	apps, err := d.listAppsByImage(ctx, fmt.Sprintf("%s:%s", iName, iTag))
+	if err != nil {
+		return errors.NewE(err)
+	}
+
+	// for the latest
+	apps2, err := d.listAppsByImage(ctx, iName)
+	if err != nil {
+		return errors.NewE(err)
+	}
+
+	for _, app := range apps {
+		if err := d.resyncK8sResource(ctx, app.EnvironmentName, app.SyncStatus.Action, &app.App, app.RecordVersion); err != nil {
+			return errors.NewE(err)
+		}
+	}
+
+	for _, app := range apps2 {
+		if err := d.resyncK8sResource(ctx, app.EnvironmentName, app.SyncStatus.Action, &app.App, app.RecordVersion); err != nil {
+			return errors.NewE(err)
+		}
+	}
+
+	return nil
+}

apps/console/internal/domain/environment.go

@@ -18,6 +18,7 @@ import (
 	corev1 "k8s.io/api/core/v1"
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 
+	"github.com/kloudlite/api/apps/console/internal/domain/ports"
 	"github.com/kloudlite/api/apps/console/internal/entities"
 	fc "github.com/kloudlite/api/apps/console/internal/entities/field-constants"
 	"github.com/kloudlite/api/pkg/repos"
@@ -134,10 +135,30 @@ func (d *domain) findEnvironmentByTargetNs(ctx ConsoleContext, targetNs string)
 }
 
 func (d *domain) CreateEnvironment(ctx ConsoleContext, env entities.Environment) (*entities.Environment, error) {
+	if err := d.canPerformActionInAccount(ctx, iamT.CreateEnvironment); err != nil {
+		return nil, errors.NewE(err)
+	}
+
 	if strings.TrimSpace(env.ClusterName) == "" {
 		return nil, fmt.Errorf("clustername must be set while creating environments")
 	}
 
+	ownedBy, err := d.infraSvc.GetByokClusterOwnedBy(ctx, ports.IsClusterLabelsIn{
+		UserId:      string(ctx.UserId),
+		UserEmail:   ctx.UserEmail,
+		UserName:    ctx.UserName,
+		AccountName: ctx.AccountName,
+		ClusterName: env.ClusterName,
+	})
+	if err != nil {
+		return nil, errors.NewE(err)
+	}
+
+	if ownedBy != "" && ownedBy != string(ctx.UserId) {
+		return nil, fmt.Errorf("it's owned cluster, but you are not the owner")
+	}
+	env.Labels[constants.ClusterLabelOwnedBy] = string(ctx.UserId)
+
 	env.EnsureGVK()
 	if err := d.k8sClient.ValidateObject(ctx, &env.Environment); err != nil {
 		return nil, errors.NewE(err)
@@ -160,6 +181,7 @@ func (d *domain) CreateEnvironment(ctx ConsoleContext, env entities.Environment)
 		UserName:  ctx.UserName,
 		UserEmail: ctx.UserEmail,
 	}
+
 	env.LastUpdatedBy = env.CreatedBy
 
 	env.AccountName = ctx.AccountName

apps/console/internal/domain/ports/infra-service.go

@@ -4,6 +4,16 @@ import "context"
 
 type InfraService interface {
 	EnsureGlobalVPNConnection(ctx context.Context, args EnsureGlobalVPNConnectionIn) error
+	GetByokClusterOwnedBy(ctx context.Context, args IsClusterLabelsIn) (string, error)
+}
+
+type IsClusterLabelsIn struct {
+	UserId    string
+	UserEmail string
+	UserName  string
+
+	AccountName string
+	ClusterName string
 }
 
 type EnsureGlobalVPNConnectionIn struct {

apps/console/internal/domain/registry-image.go

@@ -79,7 +79,7 @@ func (d *domain) GetRegistryImageURL(ctx ConsoleContext) (*entities.RegistryImag
 	}, nil
 }
 
-func (d *domain) CreateRegistryImage(ctx context.Context, accountName string, image string, meta map[string]any) (*entities.RegistryImage, error) {
+func (d *domain) UpsertRegistryImage(ctx context.Context, accountName string, image string, meta map[string]any) (*entities.RegistryImage, error) {
 	imageName, imageTag := getImageNameTag(image)
 
 	createdImage, err := d.registryImageRepo.Upsert(ctx, repos.Filter{