feat(firehose): use same label builder for create/edit operations

internal/server/v1/firehose/crud.go

@@ -64,16 +64,7 @@ func (api *firehoseAPI) handleCreate(w http.ResponseWriter, r *http.Request) {
 		return
 	}
 
-	def.Labels = cloneAndMergeMaps(def.Labels, map[string]string{
-		labelTitle:       *def.Title,
-		labelGroup:       groupID,
-		labelTeam:        groupSlug,
-		labelStream:      *def.Configs.StreamName,
-		labelDescription: def.Description,
-		labelSinkType:    def.Configs.EnvVars[configSinkType],
-		labelTopic:       def.Configs.EnvVars[configSourceKafkaTopic],
-		labelKubeCluster: *def.Configs.KubeCluster,
-	})
+	def.Labels = cloneAndMergeMaps(def.Labels, api.buildLabels(def, groupSlug))
 
 	prj, err := project.GetProject(ctx, def.Project, api.Shield)
 	if err != nil {
@@ -185,7 +176,8 @@ func (api *firehoseAPI) handleList(w http.ResponseWriter, r *http.Request) {
 	topicName := q.Get("topic_name")
 	kubeCluster := q.Get("kube_cluster")
 	sinkTypes := sinkTypeSet(q.Get("sink_type"))
-	var arr []models.Firehose
+
+	arr := []models.Firehose{}
 	for _, res := range rpcResp.GetResources() {
 		def, err := mapEntropyResourceToFirehose(res)
 		if err != nil {
@@ -253,20 +245,15 @@ func (api *firehoseAPI) handleUpdate(w http.ResponseWriter, r *http.Request) {
 		utils.WriteErr(w, err)
 		return
 	}
-	labels := cloneAndMergeMaps(existingFirehose.Labels, map[string]string{
-		labelGroup: groupID,
-		labelTeam:  groupSlug,
-	})
-	if updates.Description != "" {
-		labels[labelDescription] = existingFirehose.Description
-	}
 
 	err = api.buildEnvVars(r.Context(), &existingFirehose, reqCtx.UserID, false)
 	if err != nil {
 		utils.WriteErr(w, fmt.Errorf("error building env vars: %w", err))
 		return
 	}
 
+	labels := cloneAndMergeMaps(existingFirehose.Labels, api.buildLabels(existingFirehose, groupSlug))
+
 	cfgStruct, err := makeConfigStruct(existingFirehose.Configs)
 	if err != nil {
 		utils.WriteErr(w, err)
@@ -324,19 +311,16 @@ func (api *firehoseAPI) handlePartialUpdate(w http.ResponseWriter, r *http.Reque
 		return
 	}
 
-	labels := existing.Labels
-	if req.Description != "" {
-		labels[labelDescription] = req.Description
-	}
+	var groupSlug string
 	if req.Group != "" {
 		groupID := req.Group
-		groupSlug, err := api.getGroupSlug(ctx, groupID)
+		groupSlug, err = api.getGroupSlug(ctx, groupID)
 		if err != nil {
 			utils.WriteErr(w, err)
 			return
 		}
-		labels[labelGroup] = groupID
-		labels[labelTeam] = groupSlug
+	} else {
+		groupSlug = existing.Labels[labelGroup]
 	}
 
 	if req.Configs.Stopped != nil {
@@ -349,7 +333,6 @@ func (api *firehoseAPI) handlePartialUpdate(w http.ResponseWriter, r *http.Reque
 
 	if req.Configs.StreamName != "" {
 		existing.Configs.StreamName = &req.Configs.StreamName
-		labels[labelStream] = req.Configs.StreamName
 	}
 
 	if req.Configs.Replicas > 0 {
@@ -372,14 +355,6 @@ func (api *firehoseAPI) handlePartialUpdate(w http.ResponseWriter, r *http.Reque
 		}
 	}
 
-	if existing.Configs.EnvVars["SINK_TYPE"] != req.Configs.EnvVars["SINK_TYPE"] {
-		labels[labelSinkType] = req.Configs.EnvVars["SINK_TYPE"]
-	}
-
-	if existing.Configs.EnvVars[configSourceKafkaTopic] != req.Configs.EnvVars[configSourceKafkaTopic] {
-		labels[labelTopic] = req.Configs.EnvVars[configSourceKafkaTopic]
-	}
-
 	existing.Configs.EnvVars = cloneAndMergeMaps(
 		existing.Configs.EnvVars,
 		req.Configs.EnvVars,
@@ -393,6 +368,8 @@ func (api *firehoseAPI) handlePartialUpdate(w http.ResponseWriter, r *http.Reque
 		return
 	}
 
+	labels := cloneAndMergeMaps(existing.Labels, api.buildLabels(existing, groupSlug))
+
 	cfgStruct, err := makeConfigStruct(existing.Configs)
 	if err != nil {
 		utils.WriteErr(w, err)
@@ -429,6 +406,19 @@ func (api *firehoseAPI) handlePartialUpdate(w http.ResponseWriter, r *http.Reque
 	utils.WriteJSON(w, http.StatusOK, updatedFirehose)
 }
 
+func (*firehoseAPI) buildLabels(firehose models.Firehose, groupSlug string) map[string]string {
+	return map[string]string{
+		labelTitle:       *firehose.Title,
+		labelGroup:       string(*firehose.Group),
+		labelTeam:        groupSlug,
+		labelStream:      *firehose.Configs.StreamName,
+		labelDescription: firehose.Description,
+		labelSinkType:    firehose.Configs.EnvVars[configSinkType],
+		labelTopic:       firehose.Configs.EnvVars[configSourceKafkaTopic],
+		labelKubeCluster: *firehose.Configs.KubeCluster,
+	}
+}
+
 func (api *firehoseAPI) handleGetHistory(w http.ResponseWriter, r *http.Request) {
 	urn := chi.URLParam(r, pathParamURN)