feat(graph): model Redis Pub/Sub event topology

[kokorolx]

Summary

• add explicit event roles for JavaScript/TypeScript publish, subscribe, and emit integration edges
• extend flow stitching to connect matching publisher → consumer → emit paths
• add Redis/Socket.IO event-role regression coverage

Why

Redis Pub/Sub and Socket.IO events are event topology, not direct calls. The graph needs explicit roles so trade-state flows can be traced without misrepresenting cross-process delivery as a function call.

Validation

• CGO_ENABLED=0 go build ./...
• go test -race ./internal/graph ./internal/flow

Closes #612

[gemini-code-assist]

Code Review

This pull request introduces tracking for event roles (such as publish, subscribe, and emit) within the JS integration extractor and updates the flow stitching logic to construct event_emit flow edges. It also adds a unit test to verify the extraction of these event roles. The review feedback highlights a critical bug in the stitching logic where the emitter's source node is incorrectly compared to the consumer's entry node instead of its target node, which prevents the flow edges from being generated. Additionally, a performance optimization is suggested to pre-filter metadata before unmarshaling JSON to reduce overhead in workspaces with many edges.

Important

The consumer version of Gemini Code Assist on GitHub is being sunset. Starting June 18, 2026, new organization installations will be blocked, and all code review activity will officially cease on July 17, 2026.
For more details on the timeline and next steps, please review the Help Documentation.

[gemini-code-assist]

The stitching logic for event_emit edges compares emitter.SourceNode with t.sourceNode:

if emitter.SourceNode != t.sourceNode {
    continue
}

However, t.sourceNode is the consumer's entry node (e.g., "CONSUME tradestate"), whereas emitter.SourceNode is the function/module where the emit call occurred (e.g., "socket.js::module"). Because these two values will never be equal, this condition is always true, and the event_emit flow edges are never appended to the result.

To fix this, we need to match the emitter's source node with the consumer's target node (the handler function/module). Since consumerEntry currently does not store the target node, we should update it to do so.

Please apply the following changes:

1. Update consumerEntry struct to include targetNode:

type consumerEntry struct {
	workspaceHash string
	sourceNode    string
	targetNode    string
}

2. Populate targetNode when building the consumers map (around line 54):

consumers[topic] = append(consumers[topic], consumerEntry{
	workspaceHash: wsID,
	sourceNode:    e.SourceNode,
	targetNode:    e.TargetNode,
})

3. Apply the code suggestion below to update the comparison in the stitching loop.

Additionally, consider adding a unit test in internal/flow that validates the end-to-end stitching of publisher -> consumer -> emit paths to prevent future regressions.

Suggested change

	for _, emitter := range emitters[topic] {
	if emitter.SourceNode != t.sourceNode {
	continue
	}
	result = append(result, FlowEdge{From: t.sourceNode, To: emitter.TargetNode, Kind: "event_emit", CrossServiceWorkspace: t.workspaceHash})
	}
	for _, emitter := range emitters[topic] {
	if emitter.SourceNode != t.targetNode {
	continue
	}
	result = append(result, FlowEdge{From: t.sourceNode, To: emitter.TargetNode, Kind: "event_emit", CrossServiceWorkspace: t.workspaceHash})
	}

[gemini-code-assist]

In workspaces with a large number of edges, calling json.Unmarshal on every single edge returned by ListAllEdgesByWorkspace can introduce significant CPU and memory allocation overhead.

Since we are only interested in edges with "event_role": "emit", we can perform a fast pre-filtering check on the raw e.Metadata bytes using strings.Contains before attempting to unmarshal the JSON.

				for _, e := range all {
					if len(e.Metadata) == 0 || !strings.Contains(string(e.Metadata), "\"emit\"") {
						continue
					}
					var metadata map[string]any
					_ = json.Unmarshal(e.Metadata, &metadata)
					if topic, ok := metadata["topic"].(string); ok && topic != "" && metadata["event_role"] == "emit" {
						emitters[topic] = append(emitters[topic], graph.Edge{SourceNode: e.SourceNode, TargetNode: e.TargetNode, Kind: graph.EdgeKind(e.EdgeType), SourceFile: e.SourceFile})
					}
				}