Address Copilot review comments on backend flow handlers

- parseFlowStreamFilter: trim whitespace from each comma-separated
  value and skip empty elements to avoid passing blank strings to the
  gRPC filter; extract a splitTrimmed helper.
- protoFlowToAPI: guard GetStartTs/GetEndTs for nil before calling
  AsTime() to avoid a nil-pointer dereference on malformed records.
- handler.go errCh select case: add a comment clarifying that !ok
  means the gRPC goroutine exited cleanly (EOF / ctx cancelled) and
  buffered flows were already delivered before errCh closed (LIFO
  defers ensure flowsCh is closed after errCh).

Signed-off-by: Anlan He <anlan.he@broadcom.com>
Co-authored-by: Cursor <cursoragent@cursor.com>

Author: heanlan

pkg/handlers/flowstream/grpc.go

@@ -187,10 +187,14 @@ func ipBytesToString(b []byte) string {
 func protoFlowToAPI(pb *flowpb.Flow) apisv1.Flow {
 	f := apisv1.Flow{
 		ID:        pb.GetId(),
-		StartTs:   pb.GetStartTs().AsTime().Format(time.RFC3339Nano),
-		EndTs:     pb.GetEndTs().AsTime().Format(time.RFC3339Nano),
 		EndReason: apisv1.FlowEndReason(pb.GetEndReason()),
 	}
+	if ts := pb.GetStartTs(); ts != nil {
+		f.StartTs = ts.AsTime().Format(time.RFC3339Nano)
+	}
+	if ts := pb.GetEndTs(); ts != nil {
+		f.EndTs = ts.AsTime().Format(time.RFC3339Nano)
+	}
 
 	if ip := pb.GetIp(); ip != nil {
 		f.IP = apisv1.FlowIP{

pkg/handlers/flowstream/handler.go

@@ -63,17 +63,30 @@ func NewSSEHandler(logger logr.Logger, handler FlowStreamSubscriber) *SSEHandler
 	}
 }
 
+// splitTrimmed splits s by comma and trims whitespace from each element,
+// omitting any elements that are empty after trimming.
+func splitTrimmed(s string) []string {
+	parts := strings.Split(s, ",")
+	result := make([]string, 0, len(parts))
+	for _, p := range parts {
+		if t := strings.TrimSpace(p); t != "" {
+			result = append(result, t)
+		}
+	}
+	return result
+}
+
 func parseFlowStreamFilter(c *gin.Context) (*FlowStreamFilter, error) {
 	filter := &FlowStreamFilter{}
 
 	if ns := c.Query("namespaces"); ns != "" {
-		filter.Namespaces = strings.Split(ns, ",")
+		filter.Namespaces = splitTrimmed(ns)
 	}
 	if pods := c.Query("pods"); pods != "" {
-		filter.PodNames = strings.Split(pods, ",")
+		filter.PodNames = splitTrimmed(pods)
 	}
 	if svcs := c.Query("services"); svcs != "" {
-		filter.ServiceNames = strings.Split(svcs, ",")
+		filter.ServiceNames = splitTrimmed(svcs)
 	}
 	if selector := c.Query("podLabelSelector"); selector != "" {
 		filter.PodLabelSelector = selector
@@ -89,7 +102,7 @@ func parseFlowStreamFilter(c *gin.Context) (*FlowStreamFilter, error) {
 		}
 	}
 	if ips := c.Query("ips"); ips != "" {
-		filter.IPs = strings.Split(ips, ",")
+		filter.IPs = splitTrimmed(ips)
 	}
 	if dir := c.Query("direction"); dir != "" {
 		switch strings.ToLower(dir) {
@@ -190,6 +203,9 @@ func (h *SSEHandler) StreamFlows(c *gin.Context) {
 			return true
 		case streamErr, ok := <-errCh:
 			if !ok {
+				// errCh closed means the gRPC goroutine exited cleanly (EOF or
+				// context cancelled). flowsCh is closed after errCh (LIFO defers),
+				// so any buffered flows were already delivered. End the SSE stream.
 				return false
 			}
 			errEvent := apisv1.FlowStreamErrorEvent{Message: streamErr.Error()}