feat(pubsub-event): inital commit for the new pubsub-event feature

.github/workflows/ci.yaml

@@ -199,3 +199,21 @@ jobs:
       module: notification
       run_check_generated_files: true
       ko_build_path: "cmd/main.go"
+
+  event:
+    name: Event
+    uses: ./.github/workflows/reusable-go-ci.yaml
+    with:
+      name: event
+      module: event
+      run_check_generated_files: true
+      ko_build_path: "cmd/main.go"
+
+  pubsub:
+    name: PubSub
+    uses: ./.github/workflows/reusable-go-ci.yaml
+    with:
+      name: pubsub
+      module: pubsub
+      run_check_generated_files: true
+      ko_build_path: "cmd/main.go"

.gitignore

@@ -6,6 +6,7 @@ dist/
 **/bin/
 charts
 snapshotter-config.yaml
+testdata*
 
 # editor and IDE paraphernalia
 .idea
@@ -17,6 +18,10 @@ snapshots
 allure-report
 allure-results
 debug-results
+tmp
 
-# ignore Claude AI files
-CLAUDE.md
+# ignore AI files
+CLAUDE.md
+.grepai/
+.opencode/
+docs/IMPLEMENTATION_PLAN*

.goreleaser.yaml

@@ -136,6 +136,24 @@ builds:
       - windows # only for rover-ctl
     goarch:
       - amd64
+  - id: event
+    dir: event
+    main: cmd/main.go
+    binary: event
+    goos:
+      - linux
+    goarch:
+      - amd64
+      - arm64
+  - id: pubsub
+    dir: pubsub
+    main: cmd/main.go
+    binary: pubsub
+    goos:
+      - linux
+    goarch:
+      - amd64
+      - arm64
 archives:
   - id: rover-ctl
     ids: [rover-ctl]

api/internal/handler/apisubscription/util.go

@@ -96,8 +96,8 @@ func ApiVisibilityMustBeValid(ctx context.Context, apiExposure *apiapi.ApiExposu
 
 	// only same zone
 	if exposureVisibility == apiapi.VisibilityZone {
-		if apiExposure.Spec.Zone.GetName() != subZone.GetName() {
-			log.Info(fmt.Sprintf("Exposure visibility is ZONE and it doesnt match the subscription zone '%s'", subZone.GetName()))
+		if apiExposure.Spec.Zone.Equals(subZone) {
+			log.Info(fmt.Sprintf("Exposure visibility is ZONE and it does not match the subscription zone '%s'", subZone.GetName()))
 			return false, nil
 		}
 	}

common-server/internal/config/config.go

@@ -160,7 +160,7 @@ func (c *ServerConfig) BuildServer(ctx context.Context, dynamicClient dynamic.In
 	}
 
 	appCfg := server.NewAppConfig()
-	appCfg.CtxLog = &log
+	appCfg.CtxLog = log
 	s := server.NewServerWithApp(server.NewAppWithConfig(appCfg))
 	openapiBuilder := openapi.NewDocumentBuilder()
 	openapiBuilder.NewInfo(c.Openapi.Title, c.Openapi.Description, c.Openapi.Version)

common-server/pkg/client/errors.go

@@ -80,7 +80,7 @@ func HandleError(httpStatus int, msg string, okStatusCodes ...int) error {
 	}
 	var httpErr *HttpError
 	switch httpStatus {
-	case 400, 403:
+	case 400, 403, 405:
 		return BlockedErrorf("bad request error (%d): %s", httpStatus, msg)
 	case 409, 500, 502, 504:
 		httpErr = RetryableErrorf("server error (%d): %s", httpStatus, msg)

common-server/pkg/server/middleware/logging.go

@@ -34,7 +34,7 @@ func WithOutput(w io.Writer) LoggerOption {
 	}
 }
 
-const jsonFormat = `{"time":"${time}","ip":"${ip}","host":"${host}","method":"${method}","path":"${path}","status":${status},"latency":"${latency}","queryParams":"${queryParams}", "cid": "${cid}"}` + "\n"
+const jsonFormat = `{"time":"${time}","ip":"${ip}","host":"${host}","method":"${method}","path":"${path}","status":${status},"latency":"${latency}","ua":"${ua}","queryParams":"${queryParams}","cid":"${cid}"}` + "\n"
 
 var formats = map[LogFormat]string{
 	LogFormatJSON: jsonFormat,
@@ -72,7 +72,7 @@ func NewLogger(opts ...LoggerOption) fiber.Handler {
 	})
 }
 
-func NewContextLogger(log *logr.Logger) fiber.Handler {
+func NewContextLogger(log logr.Logger) fiber.Handler {
 	return func(c *fiber.Ctx) error {
 		ctx := c.UserContext()
 		cid := uuid.NewString()

common-server/pkg/server/middleware/suite_test.go

@@ -28,7 +28,7 @@ var _ = Describe("Middleware", func() {
 			var buffer = bytes.NewBuffer(nil)
 
 			app := fiber.New()
-			app.Use(middleware.NewContextLogger(&GinkgoLogr))
+			app.Use(middleware.NewContextLogger(GinkgoLogr))
 			app.Use(middleware.NewLogger(middleware.WithOutput(buffer)))
 			req := httptest.NewRequest("GET", "/", nil)
 			res, err := app.Test(req)

common-server/pkg/server/server.go

@@ -67,7 +67,7 @@ func (s *Server) RegisterController(controller Controller, opts ControllerOpts)
 
 type AppConfig struct {
 	fiber.Config
-	CtxLog        *logr.Logger
+	CtxLog        logr.Logger
 	EnableLogging bool
 	EnableMetrics bool
 	EnableCors    bool
@@ -100,7 +100,8 @@ func NewAppWithConfig(cfg AppConfig) *fiber.App {
 		EnableStackTrace: true,
 	}))
 	if cfg.EnableLogging {
-		if cfg.CtxLog != nil {
+		if cfg.CtxLog.Enabled() {
+			// TODO: in the future use the otel integration here
 			app.Use(middleware.NewContextLogger(cfg.CtxLog))
 		}
 		app.Use(middleware.NewLogger())

common-server/pkg/server/server_test.go

@@ -92,7 +92,7 @@ var _ = Describe("Server", func() {
 
 		It("should create a new app with custom config", func() {
 			appCfg := server.NewAppConfig()
-			appCfg.CtxLog = &GinkgoLogr
+			appCfg.CtxLog = GinkgoLogr
 			app := server.NewAppWithConfig(appCfg)
 			Expect(app).NotTo(BeNil())
 		})

common-server/pkg/store/inmemory/inmemory_store.go

@@ -321,20 +321,22 @@ func (s *InmemoryObjectStore[T]) CreateOrReplace(ctx context.Context, in T) erro
 		return err
 	}
 
+	// Initial creation if object does not exist
 	if problems.IsNotFound(err) {
 		obj.GetObjectKind().SetGroupVersionKind(s.gvk)
-		s.log.Info("creating object", "namespace", obj.GetNamespace(), "name", obj.GetName(), "gvk", s.gvk)
+		s.log.Info("creating object", "namespace", obj.GetNamespace(), "name", obj.GetName(), "gvk", obj.GetObjectKind().GroupVersionKind())
 
-		// check if not found
 		obj, err = s.k8sClient.Namespace(obj.GetNamespace()).Create(ctx, obj, metav1.CreateOptions{
 			FieldValidation: "Strict",
 		})
 		if err != nil {
-			return errors.Wrap(mapErrorToProblem(err), "failed to create object")
+			return errors.Wrapf(mapErrorToProblem(err), "failed to create object %v/%v", in.GetNamespace(), in.GetName())
 		}
 		return s.OnCreate(ctx, obj)
 	}
 
+	// Object exists, update it
+
 	obj.SetResourceVersion(currentObj.GetResourceVersion())
 	obj, err = s.k8sClient.Namespace(obj.GetNamespace()).Update(ctx, obj, metav1.UpdateOptions{
 		FieldValidation: "Strict",

common-server/pkg/store/noop/noop.go

@@ -0,0 +1,63 @@
+// Copyright 2025 Deutsche Telekom IT GmbH
+//
+// SPDX-License-Identifier: Apache-2.0
+
+package noop
+
+import (
+	"context"
+	"fmt"
+
+	"github.com/telekom/controlplane/common-server/pkg/problems"
+	"github.com/telekom/controlplane/common-server/pkg/store"
+	"k8s.io/apimachinery/pkg/runtime/schema"
+)
+
+// noopStore is a no-operation implementation of ObjectStore that returns
+// empty results for read operations and errors for write operations.
+// It is intended to be used when a feature is disabled, allowing callers
+// to interact with the store without special-casing feature-flag checks.
+type noopStore[T store.Object] struct {
+	gvr schema.GroupVersionResource
+	gvk schema.GroupVersionKind
+}
+
+// NewStore creates a new ObjectStore that performs no operations.
+// Read operations return not-found or empty results, and write operations
+// return an error indicating the feature is disabled.
+func NewStore[T store.Object](gvr schema.GroupVersionResource, gvk schema.GroupVersionKind) store.ObjectStore[T] {
+	return &noopStore[T]{
+		gvr: gvr,
+		gvk: gvk,
+	}
+}
+
+func (s *noopStore[T]) Info() (schema.GroupVersionResource, schema.GroupVersionKind) {
+	return s.gvr, s.gvk
+}
+
+func (s *noopStore[T]) Ready() bool {
+	return true
+}
+
+func (s *noopStore[T]) Get(_ context.Context, _, name string) (T, error) {
+	var zero T
+	return zero, problems.NotFound(name)
+}
+
+func (s *noopStore[T]) List(_ context.Context, _ store.ListOpts) (*store.ListResponse[T], error) {
+	return &store.ListResponse[T]{Items: []T{}}, nil
+}
+
+func (s *noopStore[T]) Delete(_ context.Context, _, name string) error {
+	return problems.NotFound(name)
+}
+
+func (s *noopStore[T]) CreateOrReplace(_ context.Context, _ T) error {
+	return problems.BadRequest(fmt.Sprintf("feature for %s is disabled", s.gvk.Kind))
+}
+
+func (s *noopStore[T]) Patch(_ context.Context, _, _ string, _ ...store.Patch) (T, error) {
+	var zero T
+	return zero, problems.BadRequest(fmt.Sprintf("feature for %s is disabled", s.gvk.Kind))
+}

common-server/pkg/store/noop/noop_test.go

@@ -0,0 +1,101 @@
+// Copyright 2025 Deutsche Telekom IT GmbH
+//
+// SPDX-License-Identifier: Apache-2.0
+
+package noop_test
+
+import (
+	"context"
+
+	. "github.com/onsi/ginkgo/v2"
+	. "github.com/onsi/gomega"
+	"github.com/telekom/controlplane/common-server/pkg/problems"
+	"github.com/telekom/controlplane/common-server/pkg/store"
+	"github.com/telekom/controlplane/common-server/pkg/store/noop"
+	"k8s.io/apimachinery/pkg/apis/meta/v1/unstructured"
+	"k8s.io/apimachinery/pkg/runtime/schema"
+)
+
+var _ = Describe("NoopStore", func() {
+	var (
+		s   store.ObjectStore[*unstructured.Unstructured]
+		ctx context.Context
+		gvr schema.GroupVersionResource
+		gvk schema.GroupVersionKind
+	)
+
+	BeforeEach(func() {
+		ctx = context.Background()
+		gvr = schema.GroupVersionResource{
+			Group:    "test.io",
+			Version:  "v1",
+			Resource: "things",
+		}
+		gvk = schema.GroupVersionKind{
+			Group:   "test.io",
+			Version: "v1",
+			Kind:    "Thing",
+		}
+		s = noop.NewStore[*unstructured.Unstructured](gvr, gvk)
+	})
+
+	Context("Info", func() {
+		It("should return the configured GVR and GVK", func() {
+			gotGVR, gotGVK := s.Info()
+			Expect(gotGVR).To(Equal(gvr))
+			Expect(gotGVK).To(Equal(gvk))
+		})
+	})
+
+	Context("Ready", func() {
+		It("should always return true", func() {
+			Expect(s.Ready()).To(BeTrue())
+		})
+	})
+
+	Context("Get", func() {
+		It("should return a NotFound error", func() {
+			_, err := s.Get(ctx, "default", "my-thing")
+			Expect(err).To(HaveOccurred())
+			Expect(problems.IsNotFound(err)).To(BeTrue())
+		})
+	})
+
+	Context("List", func() {
+		It("should return an empty list", func() {
+			resp, err := s.List(ctx, store.NewListOpts())
+			Expect(err).NotTo(HaveOccurred())
+			Expect(resp).NotTo(BeNil())
+			Expect(resp.Items).To(BeEmpty())
+		})
+	})
+
+	Context("Delete", func() {
+		It("should return a NotFound error", func() {
+			err := s.Delete(ctx, "default", "my-thing")
+			Expect(err).To(HaveOccurred())
+			Expect(problems.IsNotFound(err)).To(BeTrue())
+		})
+	})
+
+	Context("CreateOrReplace", func() {
+		It("should return a BadRequest error indicating the feature is disabled", func() {
+			obj := &unstructured.Unstructured{}
+			err := s.CreateOrReplace(ctx, obj)
+			Expect(err).To(HaveOccurred())
+			Expect(err.Error()).To(ContainSubstring("disabled"))
+		})
+	})
+
+	Context("Patch", func() {
+		It("should return a BadRequest error indicating the feature is disabled", func() {
+			_, err := s.Patch(ctx, "default", "my-thing", store.Patch{
+				Path:  "/spec/foo",
+				Op:    store.OpReplace,
+				Value: "bar",
+			})
+			Expect(err).To(HaveOccurred())
+			Expect(err.Error()).To(ContainSubstring("disabled"))
+		})
+	})
+})

common/pkg/client/clientutil.go

@@ -11,6 +11,8 @@ import (
 	"sigs.k8s.io/controller-runtime/pkg/controller/controllerutil"
 )
 
+// OwnedBy filters for resources owned by the given owner based on controller references.
+// It requires that the index ".metadata.controller" is registered
 func OwnedBy(owner client.Object) []client.ListOption {
 
 	ownerUID := string(owner.GetUID())