feat(namespace): delete (#487)

internal/ingest/kafkaingest/namespace.go

@@ -23,11 +23,7 @@ type NamespaceHandler struct {
 
 // CreateNamespace implements the namespace handler interface.
 func (h NamespaceHandler) CreateNamespace(ctx context.Context, namespace string) error {
-	if namespace == "" {
-		return fmt.Errorf("namespace is empty")
-	}
-
-	topic := fmt.Sprintf(h.NamespacedTopicTemplate, namespace)
+	topic := h.getTopicName(namespace)
 	result, err := h.AdminClient.CreateTopics(ctx, []kafka.TopicSpecification{
 		{
 			Topic:         topic,
@@ -50,3 +46,24 @@ func (h NamespaceHandler) CreateNamespace(ctx context.Context, namespace string)
 
 	return nil
 }
+
+// DeleteNamespace implements the namespace handler interface.
+func (h NamespaceHandler) DeleteNamespace(ctx context.Context, namespace string) error {
+	topic := h.getTopicName(namespace)
+	result, err := h.AdminClient.DeleteTopics(ctx, []string{topic})
+	if err != nil {
+		return err
+	}
+	for _, r := range result {
+		if r.Error.Code() != kafka.ErrNoError {
+			return r.Error
+		}
+	}
+
+	return nil
+}
+
+func (h NamespaceHandler) getTopicName(namespace string) string {
+	topic := fmt.Sprintf(h.NamespacedTopicTemplate, namespace)
+	return topic
+}

internal/namespace/namespace.go

@@ -37,15 +37,31 @@ func NewManager(config ManagerConfig) (*Manager, error) {
 // The behavior for trying to create a namespace that already exists is unspecified at the moment.
 type Handler interface {
 	CreateNamespace(ctx context.Context, name string) error
+	DeleteNamespace(ctx context.Context, name string) error
 }
 
 // CreateNamespace orchestrates namespace creation across different components.
 func (m Manager) CreateNamespace(ctx context.Context, name string) error {
-	// TODO: validate name
+	if name == "" {
+		return errors.New("cannot create empty namespace")
+	}
 
 	return m.createNamespace(ctx, name)
 }
 
+// DeleteNamespace orchestrates namespace creation across different components.
+func (m Manager) DeleteNamespace(ctx context.Context, name string) error {
+	if name == "" {
+		return errors.New("cannot delete empty namespace")
+	}
+
+	if name == m.config.DefaultNamespace {
+		return errors.New("cannot delete default namespace")
+	}
+
+	return m.deleteNamespace(ctx, name)
+}
+
 // CreateDefaultNamespace orchestrates the creation of a default namespace.
 //
 // The concept of a default namespace is implementation specific.
@@ -74,3 +90,17 @@ func (m Manager) createNamespace(ctx context.Context, name string) error {
 
 	return errors.Join(errs...)
 }
+
+// TODO: introduce some resiliency (eg. retries or rollbacks in case a component fails to delete a namespace).
+func (m Manager) deleteNamespace(ctx context.Context, name string) error {
+	var errs []error
+
+	for _, handler := range m.config.Handlers {
+		err := handler.DeleteNamespace(ctx, name)
+		if err != nil {
+			errs = append(errs, err)
+		}
+	}
+
+	return errors.Join(errs...)
+}

internal/namespace/namespace_test.go

@@ -30,6 +30,15 @@ func (h *fakeHandler) CreateNamespace(_ context.Context, name string) error {
 	return nil
 }
 
+func (h *fakeHandler) DeleteNamespace(_ context.Context, name string) error {
+	h.mu.Lock()
+	defer h.mu.Unlock()
+
+	delete(h.namespaces, name)
+
+	return nil
+}
+
 func TestManager_CreateNamespce(t *testing.T) {
 	handler := newFakeHandler()
 
@@ -61,3 +70,25 @@ func TestManager_CreateDefaultNamespce(t *testing.T) {
 
 	assert.True(t, handler.namespaces["default"])
 }
+
+func TestManager_DeleteNamespce(t *testing.T) {
+	handler := newFakeHandler()
+
+	manager, err := NewManager(ManagerConfig{
+		Handlers:         []Handler{handler},
+		DefaultNamespace: "default",
+	})
+	require.NoError(t, err)
+
+	const namespace = "my-namespace"
+
+	err = manager.CreateNamespace(context.Background(), namespace)
+	require.NoError(t, err)
+
+	assert.True(t, handler.namespaces[namespace])
+
+	err = manager.DeleteNamespace(context.Background(), namespace)
+	require.NoError(t, err)
+
+	assert.False(t, handler.namespaces[namespace])
+}

internal/streaming/clickhouse_connector/connector.go

@@ -133,10 +133,6 @@ func (c *ClickhouseConnector) ListMeterSubjects(ctx context.Context, namespace s
 }
 
 func (c *ClickhouseConnector) CreateNamespace(ctx context.Context, namespace string) error {
-	if namespace == "" {
-		return fmt.Errorf("namespace is required")
-	}
-
 	err := c.createEventsTable(ctx, namespace)
 	if err != nil {
 		return fmt.Errorf("create namespace in clickhouse: %w", err)
@@ -145,6 +141,38 @@ func (c *ClickhouseConnector) CreateNamespace(ctx context.Context, namespace str
 	return nil
 }
 
+func (c *ClickhouseConnector) DeleteNamespace(ctx context.Context, namespace string) error {
+	err := c.deleteNamespace(ctx, namespace)
+	if err != nil {
+		return fmt.Errorf("delete namespace in clickhouse: %w", err)
+	}
+	return nil
+}
+
+// DeleteNamespace deletes the namespace related resources from Clickhouse
+// We don't delete the events table as it it reused between namespaces
+// We only delete the materialized views for the meters
+func (c *ClickhouseConnector) deleteNamespace(ctx context.Context, namespace string) error {
+	// Retrieve meters belonging to the namespace
+	meters, err := c.config.Meters.ListMeters(ctx, namespace)
+	if err != nil {
+		return fmt.Errorf("failed to list meters: %w", err)
+	}
+
+	for _, meter := range meters {
+		err := c.deleteMeterView(ctx, namespace, meter.Slug)
+		if err != nil {
+			// If the meter view does not exist, we ignore the error
+			if _, ok := err.(*models.MeterNotFoundError); ok {
+				return nil
+			}
+			return fmt.Errorf("delete meter view: %w", err)
+		}
+	}
+
+	return nil
+}
+
 func (c *ClickhouseConnector) createEventsTable(ctx context.Context, namespace string) error {
 	table := createEventsTable{
 		Database: c.config.Database,