package streamflow

import (
	"fmt"
	"log"
	lunar_messages "lunar/engine/messages"
	stream_config "lunar/engine/streams/config"
	stream_filter "lunar/engine/streams/filter"
	internal_types "lunar/engine/streams/internal-types"
	lunar_context "lunar/engine/streams/lunar-context"
	"lunar/engine/streams/processors"
	"lunar/engine/streams/resources"
	"lunar/engine/utils/environment"
	"os"
	"path/filepath"
	"testing"

	publicTypes "lunar/engine/streams/public-types"

	streamTypes "lunar/engine/streams/types"

	testProcessors "lunar/engine/streams/flow/test-processors"

	"github.com/stretchr/testify/require"
)

var sharedState = lunar_context.NewMemoryState[[]byte]()

func TestMain(m *testing.M) {
	prevVal := environment.SetProcessorsDirectory("test-processors")

	// Run the tests
	code := m.Run()

	// Clean up if necessary
	environment.SetProcessorsDirectory(prevVal)

	// Exit with the code from the tests
	os.Exit(code)
}

func createTestProcessorManager(
	t *testing.T,
	processorNames []string,
) *processors.ProcessorManager {
	processorMng := processors.NewProcessorManager(nil)
	for _, procName := range processorNames {
		log.Default().Printf("Setting factory for processor %s", procName)
		processorMng.SetFactory(procName, testProcessors.NewMockProcessor)
	}

	err := processorMng.Init()
	require.NoError(t, err)

	return processorMng
}

func newTestFlow(t *testing.T, processorsCount int) *Flow {
	flowRep := &stream_config.FlowRepresentation{
		Name: "testFlow",
		Filter: &stream_config.Filter{
			Name: "testFilter",
			URLs: []string{"*"},
		},
		Processors: make(map[string]*stream_config.Processor),
	}

	var processorNames []string
	for i := 1; i <= processorsCount; i++ {
		name := fmt.Sprintf("processor%d", i)
		flowRep.Processors[name] = &stream_config.Processor{
			Processor: name,
			Key:       name,
		}
		processorNames = append(processorNames, name)
	}

	processorMng := createTestProcessorManager(t, processorNames)

	for processorKey, processorData := range flowRep.GetProcessors() {
		_, errCreation := processorMng.CreateProcessor(flowRep.GetName(), processorData)
		if errCreation != nil {
			require.NoError(t, errCreation, "Failed to create processor for key: %s", processorKey)
		}
	}

	nodeBuilder := newGraphNodeBuilder(map[string]internal_types.FlowRepI{
		flowRep.Name: flowRep,
	}, processorMng)

	return NewFlow(nodeBuilder, flowRep, nil)
}

func newTestAPIStream(url string) publicTypes.APIStreamI {
	apiStream := streamTypes.NewAPIStream("APIStreamName", publicTypes.StreamTypeRequest, sharedState)
	apiStream.SetRequest(streamTypes.NewRequest(lunar_messages.OnRequest{
		Method:  "GET",
		URL:     url,
		Headers: map[string]string{},
	}))

	apiStream.SetContext(lunar_context.NewLunarContext(lunar_context.NewContext()))
	apiStream.SetResponse(streamTypes.NewResponse(lunar_messages.OnResponse{
		Status: 200,
	}))

	return apiStream
}

func addProcessors(t *testing.T,
	flowDirection *FlowDirection,
	namePrefix string,
	count int,
) {
	addProcessorsWithDetails(t, flowDirection, namePrefix, count)
}

func addProcessorsWithDetails(
	t *testing.T,
	flowDirection *FlowDirection,
	namePrefix string,
	count int,
) {
	for i := 1; i <= count; i++ {
		name := fmt.Sprintf("%s%d", namePrefix, i)
		proc := &stream_config.ProcessorRef{
			Name:          name,
			ReferenceName: name,
		}

		_, err := flowDirection.getOrCreateNode(flowDirection.flowName, proc)
		require.NoError(t, err)
	}
}

func TestGetNode(t *testing.T) {
	flowGraph := newTestFlow(t, 1)

	addProcessors(t, flowGraph.request, "processor", 1)

	node, err := flowGraph.request.GetNode("processor1")
	require.NoError(t, err)
	require.NotNil(t, node)

	_, err = flowGraph.request.GetNode("nonExistentNode")
	require.Error(t, err)
}

func TestBuildFlows(t *testing.T) {
	globalStreamRefStart := &stream_config.StreamRef{Name: publicTypes.GlobalStream, At: "start"}
	globalStreamRefEnd := &stream_config.StreamRef{Name: publicTypes.GlobalStream, At: "end"}
	processorRef1Condition := &stream_config.ProcessorRef{
		Name:          "processor1",
		Condition:     "condition",
		ReferenceName: "processor1",
	}
	processorRef2Condition2 := &stream_config.ProcessorRef{
		Name:          "processor2",
		Condition:     "condition2",
		ReferenceName: "processor2",
	}
	processorRef3 := &stream_config.ProcessorRef{Name: "processor3", ReferenceName: "processor3"}
	processorRef4 := &stream_config.ProcessorRef{Name: "processor4", ReferenceName: "processor4"}
	processorRef5 := &stream_config.ProcessorRef{Name: "processor5", ReferenceName: "processor5"}
	processorRef6 := &stream_config.ProcessorRef{Name: "processor6", ReferenceName: "processor6"}
	filter := stream_config.Filter{Name: "filter1", URLs: []string{"example.com"}}
	processorsList := []string{
		"processor1",
		"processor2",
		"processor3",
		"processor4",
		"processor5",
		"processor6",
	}

	testCases := []struct {
		name         string
		processorMng *processors.ProcessorManager
		flowReps     map[string]internal_types.FlowRepI
		validateFn   func(t *testing.T,
			graphs map[string]*Flow,
			requestEntryPoint, responseEntryPoint internal_types.EntryPointI,
		)
		expectErr      bool
		expectedErrMsg string
	}{
		{
			name:         "Valid single graph",
			processorMng: createTestProcessorManager(t, processorsList),
			flowReps: map[string]internal_types.FlowRepI{
				"GraphWithEntryPoints": &stream_config.FlowRepresentation{
					Filter: &filter,
					Name:   "GraphWithEntryPoints",
					Processors: map[string]*stream_config.Processor{
						"processor1": {Processor: "processor1", Key: "processor1"},
					},
					Flow: stream_config.Flow{
						Request: []*stream_config.FlowConnection{
							{
								From: &stream_config.Connection{Stream: globalStreamRefStart},
								To:   &stream_config.Connection{Processor: processorRef1Condition},
							},
							{
								From: &stream_config.Connection{Processor: processorRef1Condition},
								To:   &stream_config.Connection{Stream: globalStreamRefEnd},
							},
						},
						Response: []*stream_config.FlowConnection{
							{
								From: &stream_config.Connection{Stream: globalStreamRefStart},
								To:   &stream_config.Connection{Processor: processorRef1Condition},
							},
							{
								From: &stream_config.Connection{Processor: processorRef1Condition},
								To:   &stream_config.Connection{Stream: globalStreamRefEnd},
							},
						},
					},
				},
			},
			expectErr: false,
		},
		{
			name:         "Graph with no direction defined",
			processorMng: createTestProcessorManager(t, processorsList),
			flowReps: map[string]internal_types.FlowRepI{
				"Graph1": &stream_config.FlowRepresentation{
					Filter: &filter,
					Name:   "Graph1",
					Processors: map[string]*stream_config.Processor{
						"processor1": {Processor: "processor1", Key: "processor1", Parameters: nil},
					},
					Flow: stream_config.Flow{},
				},
			},
			expectErr: true,
		},
		{
			name:         "Graph with single direction defined",
			processorMng: createTestProcessorManager(t, processorsList),
			flowReps: map[string]internal_types.FlowRepI{
				"Graph1": &stream_config.FlowRepresentation{
					Filter: &filter,
					Name:   "Graph1",
					Processors: map[string]*stream_config.Processor{
						"processor1": {Processor: "processor1", Key: "processor1", Parameters: nil},
					},
					Flow: stream_config.Flow{
						Request: []*stream_config.FlowConnection{
							{
								From: &stream_config.Connection{Stream: globalStreamRefStart},
								To:   &stream_config.Connection{Processor: processorRef1Condition},
							},
							{
								From: &stream_config.Connection{Processor: processorRef1Condition},
								To:   &stream_config.Connection{Stream: globalStreamRefEnd},
							},
						},
					},
				},
			},
			expectErr: false,
		},
		{
			// The following test case ensures that a flow such as below is valid - although we get to 5 twice.
			// +---+     	+---+     condition      +---+      +---+
			// | 1 | ---->| 2 | ------------------>| 3 | ---->| 5 |
			// +---+		 	+---+                    +---+      +---+
			//					 		|                                   ^
			//					 		|                                   |
			//					 condition2                             |
			//					 		|                                   |
			//					 		v                                   |
			//					 	+---+                                 |
			//					 	| 4 | ---------------------------------
			//					 	+---+
			name:         "Valid - revisiting nodes without circular connections",
			processorMng: createTestProcessorManager(t, processorsList),
			expectErr:    false,
			flowReps: map[string]internal_types.FlowRepI{
				"revisitingNodes": &stream_config.FlowRepresentation{
					Name:   "revisitingNodes",
					Filter: &filter,
					Processors: map[string]*stream_config.Processor{
						"processor1": {Processor: "processor1", Key: "processor1"},
						"processor2": {Processor: "processor2", Key: "processor2"},
						"processor3": {Processor: "processor3", Key: "processor3"},
						"processor4": {Processor: "processor4", Key: "processor4"},
						"processor5": {Processor: "processor5", Key: "processor5"},
					},
					Flow: stream_config.Flow{
						Request: []*stream_config.FlowConnection{
							{
								From: &stream_config.Connection{
									Stream: globalStreamRefStart,
								},
								To: &stream_config.Connection{
									Processor: processorRef1Condition,
								},
							},
							{
								From: &stream_config.Connection{
									Processor: processorRef1Condition,
								},
								To: &stream_config.Connection{
									Processor: processorRef2Condition2,
								},
							},
							{
								From: &stream_config.Connection{
									Processor: processorRef2Condition2,
								},
								To: &stream_config.Connection{
									Processor: processorRef3,
								},
							},
							{
								From: &stream_config.Connection{
									Processor: processorRef2Condition2,
								},
								To: &stream_config.Connection{
									Processor: processorRef4,
								},
							},
							{
								From: &stream_config.Connection{
									Processor: processorRef4,
								},
								To: &stream_config.Connection{
									Processor: processorRef5,
								},
							},
							{
								From: &stream_config.Connection{
									Processor: processorRef3,
								},
								To: &stream_config.Connection{
									Processor: processorRef5,
								},
							},

							{
								From: &stream_config.Connection{Processor: processorRef5},
								To:   &stream_config.Connection{Stream: globalStreamRefEnd},
							},
						},
					},
				},
			},
		},
		{
			// The following test case ensures that a flow such as below is invalid - because of cyclicism.
			// +---+				+---+     condition      +---+      +---+
			// | 1 | ------>| 2 | ------------------>| 3 | ---->| 5 |
			// +---+				+---+ <----------        +---+      +---+
			//						 		|             |
			//						 		|             |
			//						 condition2       |
			//						 		|             |
			//						 		v             |
			//						 	+---+        +---+
			//						 	| 4 | -------| 6 |
			//						 	+---+				 +---+
			// It is a little more elaborate then other tests in this file since it is not an A<=>B circular connection.
			name:           "Invalid - revisiting nodes without circular connections",
			processorMng:   createTestProcessorManager(t, processorsList),
			expectErr:      true,
			expectedErrMsg: "circular connection detected",
			flowReps: map[string]internal_types.FlowRepI{
				"invalidCyclicNodes": &stream_config.FlowRepresentation{
					Name:   "invalidCyclicNodes",
					Filter: &filter,
					Processors: map[string]*stream_config.Processor{
						"processor1": {Processor: "processor1", Key: "processor1"},
						"processor2": {Processor: "processor2", Key: "processor2"},
						"processor3": {Processor: "processor3", Key: "processor3"},
						"processor4": {Processor: "processor4", Key: "processor4"},
						"processor5": {Processor: "processor5", Key: "processor5"},
						"processor6": {Processor: "processor6", Key: "processor6"},
					},
					Flow: stream_config.Flow{
						Request: []*stream_config.FlowConnection{
							{
								From: &stream_config.Connection{
									Stream: globalStreamRefStart,
								},
								To: &stream_config.Connection{
									Processor: processorRef1Condition,
								},
							},
							{
								From: &stream_config.Connection{
									Processor: processorRef1Condition,
								},
								To: &stream_config.Connection{
									Processor: processorRef2Condition2,
								},
							},
							{
								From: &stream_config.Connection{
									Processor: processorRef2Condition2,
								},
								To: &stream_config.Connection{
									Processor: processorRef3,
								},
							},
							{
								From: &stream_config.Connection{
									Processor: processorRef2Condition2,
								},
								To: &stream_config.Connection{
									Processor: processorRef4,
								},
							},
							{
								From: &stream_config.Connection{
									Processor: processorRef3,
								},
								To: &stream_config.Connection{
									Processor: processorRef5,
								},
							},
							{
								From: &stream_config.Connection{
									Processor: processorRef4,
								},
								To: &stream_config.Connection{
									Processor: processorRef6,
								},
							},
							{
								From: &stream_config.Connection{
									Processor: processorRef6,
								},
								To: &stream_config.Connection{
									Processor: processorRef2Condition2,
								},
							},
						},
					},
				},
			},
		},
		{
			name:         "Valid Multiple Graphs Merging",
			processorMng: createTestProcessorManager(t, processorsList),
			flowReps: map[string]internal_types.FlowRepI{
				// Graph 1 with a request entry point
				"Graph1": &stream_config.FlowRepresentation{
					Filter: &filter,
					Name:   "Graph1",
					Processors: map[string]*stream_config.Processor{
						"processor1": {Processor: "processor1", Key: "processor1"},
						"processor2": {Processor: "processor2", Key: "processor2"},
					},
					Flow: stream_config.Flow{
						Request: []*stream_config.FlowConnection{
							{
								From: &stream_config.Connection{Stream: globalStreamRefStart},
								To: &stream_config.Connection{
									Processor: &stream_config.ProcessorRef{
										Name:          "processor1",
										ReferenceName: "processor1",
									},
								},
							},
							{
								From: &stream_config.Connection{
									Processor: &stream_config.ProcessorRef{
										Name:          "processor1",
										ReferenceName: "processor1",
										Condition:     "condition",
									},
								},
								To: &stream_config.Connection{Stream: globalStreamRefEnd},
							},
						},
						Response: []*stream_config.FlowConnection{
							{
								From: &stream_config.Connection{Stream: globalStreamRefStart},
								To:   &stream_config.Connection{Processor: processorRef2Condition2},
							},
							{
								From: &stream_config.Connection{Processor: processorRef2Condition2},
								To:   &stream_config.Connection{Stream: globalStreamRefEnd},
							},
						},
					},
				},
				// Graph 2 intended to be connected at the start of Graph 1
				"Graph2": &stream_config.FlowRepresentation{
					Filter: &filter,
					Name:   "Graph2",
					Processors: map[string]*stream_config.Processor{
						"processor3": {Processor: "processor3", Key: "processor3"},
						"processor4": {Processor: "processor4", Key: "processor4"},
					},
					Flow: stream_config.Flow{
						Request: []*stream_config.FlowConnection{
							{
								From: &stream_config.Connection{
									Flow: &stream_config.FlowRef{Name: "Graph1", At: "end"},
								},
								To: &stream_config.Connection{
									Processor: &stream_config.ProcessorRef{
										Name:          "processor3",
										ReferenceName: "processor3",
									},
								},
							},
							{
								From: &stream_config.Connection{
									Processor: &stream_config.ProcessorRef{
										Name:          "processor3",
										ReferenceName: "processor3",
									},
								},
								To: &stream_config.Connection{
									Processor: &stream_config.ProcessorRef{
										Name:          "processor4",
										ReferenceName: "processor4",
									},
								},
							},
							{
								From: &stream_config.Connection{
									Processor: &stream_config.ProcessorRef{
										Name:          "processor4",
										ReferenceName: "processor4",
									},
								},
								To: &stream_config.Connection{Stream: globalStreamRefEnd},
							},
						},
						Response: []*stream_config.FlowConnection{
							{
								From: &stream_config.Connection{Stream: globalStreamRefStart},
								To: &stream_config.Connection{
									Processor: &stream_config.ProcessorRef{
										Name:          "processor3",
										ReferenceName: "processor3",
									},
								},
							},
							{ // Connection indicating Graph2 flows into Graph1
								From: &stream_config.Connection{
									Processor: &stream_config.ProcessorRef{
										Name:          "processor3",
										ReferenceName: "processor3",
									},
								},
								To: &stream_config.Connection{
									Flow: &stream_config.FlowRef{Name: "Graph1", At: "start"},
								},
							},
						},
					},
				},
			},
			expectErr: false,
			validateFn: func(t *testing.T, graphs map[string]*Flow, requestEntryPoint, _ internal_types.EntryPointI) {
				// Validate that the TotalFlowGraph contains nodes from both Graph1 and Graph2
				require.NotNil(t, graphs["Graph1"], "Graph1 should be part of the total flow graph")
				require.NotNil(t, graphs["Graph2"], "Graph2 should be part of the total flow graph")

				// Validate correct configuration of entry points for the total flow.
				require.NotNil(
					t,
					requestEntryPoint,
					"The total flow graph should have a valid request entry point",
				)
				require.Equal(t, publicTypes.GlobalStream, requestEntryPoint.GetStream().GetName())

				// Validate integrity of connections between Graph1 and Graph2.
				graph2RequestRoot := graphs["Graph2"].request.root
				require.NotNil(t, graph2RequestRoot, "Graph2 should have response entry point")
				require.Equal(t, "processor3", graph2RequestRoot.GetNode().GetProcessor().GetName())
				require.Equal(t, graphs["Graph1"].GetName(), graph2RequestRoot.GetFlow().GetName())
				require.Equal(t, "end", graph2RequestRoot.GetFlow().GetAt())
				require.Equal(t, "Graph2", graph2RequestRoot.GetNode().GetFlowGraphName())

				graph2ResponseRoot := graphs["Graph2"].response.root
				require.NotNil(t, graph2ResponseRoot, "Graph2 should have response entry point")

				require.Equal(
					t,
					"processor3",
					graph2ResponseRoot.GetNode().GetProcessor().GetName(),
				)
				require.Equal(
					t,
					globalStreamRefStart.Name,
					graph2ResponseRoot.GetStream().GetName(),
				)
				require.Equal(t, "start", graph2ResponseRoot.GetStream().GetAt())
			},
		},
		{
			name:           "Invalid - Circular Processor Connections",
			processorMng:   createTestProcessorManager(t, processorsList),
			expectErr:      true,
			expectedErrMsg: "circular connection detected",
			flowReps: map[string]internal_types.FlowRepI{
				"circularProcessorConnections": &stream_config.FlowRepresentation{
					Name: "circularProcessorConnections",
					Processors: map[string]*stream_config.Processor{
						"processor1": {Processor: "processor1", Key: "processor1"},
						"processor2": {Processor: "processor2", Key: "processor2"},
					},
					Flow: stream_config.Flow{
						Request: []*stream_config.FlowConnection{
							{
								From: &stream_config.Connection{Stream: globalStreamRefStart},
								To:   &stream_config.Connection{Processor: processorRef1Condition},
							},
							{
								From: &stream_config.Connection{Processor: processorRef1Condition},
								To:   &stream_config.Connection{Processor: processorRef2Condition2},
							},
							{
								From: &stream_config.Connection{Processor: processorRef2Condition2},
								To:   &stream_config.Connection{Processor: processorRef1Condition},
							},
						},
					},
				},
			},
		},
		{
			name:           "Invalid - Circular Processor Connections with condition",
			processorMng:   createTestProcessorManager(t, processorsList),
			expectErr:      true,
			expectedErrMsg: "circular connection detected",
			flowReps: map[string]internal_types.FlowRepI{
				"circularProcessorConnections": &stream_config.FlowRepresentation{
					Name: "circularProcessorConnections",
					Processors: map[string]*stream_config.Processor{
						"processor1": {Processor: "processor1", Key: "processor1"},
						"processor2": {Processor: "processor2", Key: "processor2"},
					},
					Flow: stream_config.Flow{
						Request: []*stream_config.FlowConnection{
							{
								From: &stream_config.Connection{Stream: globalStreamRefStart},
								To:   &stream_config.Connection{Processor: processorRef1Condition},
							},
							{
								From: &stream_config.Connection{Processor: processorRef1Condition},
								To:   &stream_config.Connection{Processor: processorRef2Condition2},
							},
							{
								From: &stream_config.Connection{Processor: processorRef2Condition2},
								To:   &stream_config.Connection{Processor: processorRef1Condition},
							},
						},
					},
				},
			},
		},
		{
			name:           "Invalid - circular processor connections, with different conditions",
			processorMng:   createTestProcessorManager(t, processorsList),
			expectErr:      true,
			expectedErrMsg: "circular connection detected",
			flowReps: map[string]internal_types.FlowRepI{
				"circularProcessorConnections": &stream_config.FlowRepresentation{
					Name: "circularProcessorConnections",
					Processors: map[string]*stream_config.Processor{
						"processor1": {Processor: "processor1", Key: "processor1"},
						"processor2": {Processor: "processor2", Key: "processor2"},
					},
					Flow: stream_config.Flow{
						Request: []*stream_config.FlowConnection{
							{
								From: &stream_config.Connection{Stream: globalStreamRefStart},
								To:   &stream_config.Connection{Processor: processorRef1Condition},
							},
							{
								From: &stream_config.Connection{Processor: processorRef1Condition},
								To:   &stream_config.Connection{Processor: processorRef2Condition2},
							},
							{
								From: &stream_config.Connection{Processor: processorRef2Condition2},
								To:   &stream_config.Connection{Processor: processorRef1Condition},
							},
						},
					},
				},
			},
		},
		{
			name:         "Invalid - Processor Refers to Nonexistent Target",
			processorMng: createTestProcessorManager(t, processorsList),
			expectErr:    true,
			flowReps: map[string]internal_types.FlowRepI{
				"invalidProcessorRef": &stream_config.FlowRepresentation{
					Name: "invalidProcessorRef",
					Processors: map[string]*stream_config.Processor{
						"processor1": {Processor: "processor1", Key: "processor1"},
					},
					Flow: stream_config.Flow{
						Request: []*stream_config.FlowConnection{
							{
								From: &stream_config.Connection{Processor: processorRef1Condition},
								To: &stream_config.Connection{
									Processor: &stream_config.ProcessorRef{
										Name:          "nonexistent_processor",
										ReferenceName: "nonexistent_processor",
									},
								},
							},
						},
					},
				},
			},
		},
		{
			name:         "Invalid - Reference to nonexistent flow",
			processorMng: createTestProcessorManager(t, processorsList),
			expectErr:    true,
			flowReps: map[string]internal_types.FlowRepI{
				"invalidFlowRef": &stream_config.FlowRepresentation{
					Name: "invalidFlowRef",
					Flow: stream_config.Flow{
						Request: []*stream_config.FlowConnection{
							{
								From: &stream_config.Connection{
									Flow: &stream_config.FlowRef{
										Name: "nonexistent_flow",
										At:   "end",
									},
								},
								To: &stream_config.Connection{Processor: processorRef1Condition},
							},
						},
					},
				},
			},
		},
		{
			name:         "Invalid - Stream Refers to Nonexistent Processor",
			processorMng: createTestProcessorManager(t, processorsList),
			expectErr:    true,
			flowReps: map[string]internal_types.FlowRepI{
				"invalidStreamRef": &stream_config.FlowRepresentation{
					Name: "invalidStreamRef",
					Flow: stream_config.Flow{
						Request: []*stream_config.FlowConnection{
							{
								From: &stream_config.Connection{
									Stream: &stream_config.StreamRef{Name: publicTypes.GlobalStream},
								},
								To: &stream_config.Connection{
									Processor: &stream_config.ProcessorRef{
										Name:          "nonexistent_processor",
										ReferenceName: "nonexistent_processor",
									},
								},
							},
						},
					},
				},
			},
		},
		{
			name:         "Invalid - No Valid Root",
			processorMng: createTestProcessorManager(t, processorsList),
			expectErr:    true,
			flowReps: map[string]internal_types.FlowRepI{
				"No valid root flow": &stream_config.FlowRepresentation{
					Filter: &filter,
					Name:   "No valid root flow",
					Processors: map[string]*stream_config.Processor{
						"processor1": {Processor: "processor1", Key: "processor1"},
					},
					Flow: stream_config.Flow{
						Request: []*stream_config.FlowConnection{
							{
								From: &stream_config.Connection{Processor: processorRef1Condition},
								To:   &stream_config.Connection{Stream: globalStreamRefEnd},
							},
						},
						Response: []*stream_config.FlowConnection{
							{
								From: &stream_config.Connection{Stream: globalStreamRefStart},
								To:   &stream_config.Connection{Processor: processorRef1Condition},
							},
						},
					},
				},
			},
		},
		{
			name:         "Response flow with no root",
			processorMng: createTestProcessorManager(t, processorsList),
			expectErr:    false,
			flowReps: map[string]internal_types.FlowRepI{
				"No root for response flow flow": &stream_config.FlowRepresentation{
					Filter: &filter,
					Name:   "No root for response flow flow",
					Processors: map[string]*stream_config.Processor{
						"processor1": {Processor: "processor1", Key: "processor1"},
					},
					Flow: stream_config.Flow{
						Request: []*stream_config.FlowConnection{
							{
								From: &stream_config.Connection{Stream: globalStreamRefStart},
								To:   &stream_config.Connection{Processor: processorRef1Condition},
							},
							{
								From: &stream_config.Connection{Processor: processorRef1Condition},
								To:   &stream_config.Connection{Stream: globalStreamRefEnd},
							},
						},
						Response: []*stream_config.FlowConnection{
							{
								From: &stream_config.Connection{Processor: processorRef1Condition},
								To:   &stream_config.Connection{Stream: globalStreamRefEnd},
							},
						},
					},
				},
			},
		},
	}

	for _, testCase := range testCases {
		t.Run(testCase.name, func(t *testing.T) {
			filterTree := stream_filter.NewFilterTree()
			resourceM, _ := resources.NewResourceManagement()
			for _, flow := range testCase.flowReps {
				for key, processorData := range flow.GetProcessors() {
					_, errCreation := testCase.processorMng.CreateProcessor(flow.GetName(), processorData)
					if errCreation != nil {
						require.NoError(t, errCreation, "key: %s", key)
					}
				}
			}
			err := BuildFlows(filterTree, testCase.flowReps, testCase.processorMng, resourceM)
			if testCase.expectErr {
				require.Error(t, err)
				if testCase.expectedErrMsg != "" {
					require.Contains(t, err.Error(), testCase.expectedErrMsg)
				}
			} else {
				require.NoError(t, err)
			}
		})
	}
}

func TestGetEdges(t *testing.T) {
	flowGraph := newTestFlow(t, 1)
	addProcessors(t, flowGraph.request, "processor", 1)

	node := flowGraph.request.nodes["processor1"]
	node.edges = append(node.edges, &ConnectionEdge{condition: "condition1"})

	requestEdges := node.GetEdges()
	require.Len(t, requestEdges, 1)
	require.Equal(t, "condition1", requestEdges[0].GetCondition())
}

func testEdges(
	t *testing.T,
	edges []internal_types.ConnectionEdgeI,
	expectedEdgeNodes, expectedConditions []string,
) {
	require.Len(t, edges, len(expectedEdgeNodes))
	for i, edge := range edges {
		require.True(t, edge.IsValid(), "Edge %d is not valid", i)
		require.Equal(
			t,
			expectedEdgeNodes[i],
			edge.GetTargetNode().GetProcessorKey(),
			"Edge %d processor key does not match",
			i,
		)
		require.Equal(
			t,
			expectedConditions[i],
			edge.GetCondition(),
			"Edge %d condition does not match",
			i,
		)
	}
}

func loadTestCase(t *testing.T, testCase string) map[string]internal_types.FlowRepI {
	pattern := filepath.Join("test-cases", testCase, "*.yaml")
	files, fileErr := filepath.Glob(pattern)
	require.NoError(t, fileErr, "Failed to find YAML files")

	flowReps := make(map[string]internal_types.FlowRepI)
	for _, file := range files {
		t.Run(filepath.Base(file), func(t *testing.T) {
			flowRep, err := stream_config.ReadStreamFlowConfig(file)
			require.NoError(t, err, "Failed to read YAML file")

			flowReps[flowRep.Name] = flowRep

			require.NotEmpty(t, flowRep.Name, "Flow representation name is empty")
			t.Log("Running test case:", flowRep.Name)
		})
	}
	return flowReps
}

func TestTwoFlowsTestCaseYAML(t *testing.T) {
	t.Skip("This test should be modify to only work when the filters are in the same level")
	flowReps := loadTestCase(t, "2-flows*")

	procMng := createTestProcessorManager(
		t,
		[]string{
			"removePII",
			"readCache",
			"checkLimit",
			"generateResponse",
			"globalStream",
			"writeCache",
			"LogAPM",
		},
	)

	// Test building flow
	filterTree := stream_filter.NewFilterTree()
	resourceM, _ := resources.NewResourceManagement()
	err := BuildFlows(filterTree, flowReps, procMng, resourceM)
	require.NoError(t, err, "Failed to build flow")

	// Test first URL
	flowRaw, _ := filterTree.GetFlow(newTestAPIStream("maps.googleapis.com/maps/api/geocode/json"))
	// flowRaw := flow.(*Flow)

	require.NotNil(t, flowRaw)
	rawUserFlow, found := flowRaw.GetUserFlow()
	if !found {
		t.Error("User flow not found")
	}
	require.Equal(t, "GoogleMapsGeocodingCache", rawUserFlow[0].GetName())
	require.Equal(t, "maps.googleapis.com/maps/api/geocode/json", rawUserFlow[0].GetFilter().GetURLs())
	root, err := rawUserFlow[0].GetRequestDirection().GetRoot()
	require.NoError(t, err)
	require.NotNil(t, root)
	// ----------------------------------- Request Flow -----------------------------------
	// GoogleMapsGeocodingCache starts from InfraTeam1. The whole flow graph should be like this
	// removePII (InfraTeam1) -> readCache -> (checkLimit/generateResponse) -> globalStream/generateResponse

	require.True(t, root.IsValid())
	require.Equal(t, "removePII", root.GetNode().GetProcessorKey())
	require.Equal(t, "InfraTeam1", root.GetNode().GetFlowGraphName())
	reqEdges := root.GetNode().GetEdges()
	testEdges(t, reqEdges, []string{"readCache"}, []string{""})

	readCacheNode := root.GetNode().GetEdges()[0].GetTargetNode()
	testEdges(
		t,
		readCacheNode.GetEdges(),
		[]string{"checkLimit", "generateResponse"},
		[]string{"cache_miss", "cache_hit"},
	)

	checkLimitNode := readCacheNode.GetEdges()[0].GetTargetNode()
	require.Len(t, checkLimitNode.GetEdges(), 2)
	require.Equal(t, "below_limit", checkLimitNode.GetEdges()[0].GetCondition())
	require.Equal(t, "globalStream", checkLimitNode.GetEdges()[0].GetTargetStream().GetName())

	require.Equal(t, "above_limit", checkLimitNode.GetEdges()[1].GetCondition())
	require.Equal(
		t,
		"generateResponse",
		checkLimitNode.GetEdges()[1].GetTargetNode().GetProcessorKey(),
	)

	// ----------------------------------- Response Flow -----------------------------------
	// GoogleMapsGeocodingCache starts from globalStream and at the end passes to InfraTeam1.
	// The whole flow graph should be like this:
	// globalStream -> writeCache -> LogAPM (InfraTeam1) -> globalStream
	rawUserFlow, found = flowRaw.GetUserFlow()
	if !found {
		t.Error("User flow not found")
	}
	root, err = rawUserFlow[0].GetResponseDirection().GetRoot()
	require.NoError(t, err)
	require.True(t, root.IsValid())
	require.Equal(t, "writeCache", root.GetNode().GetProcessorKey())
	require.Equal(t, "GoogleMapsGeocodingCache", root.GetNode().GetFlowGraphName())
	testEdges(t, root.GetNode().GetEdges(), []string{"LogAPM"}, []string{""})

	logAPMNode := root.GetNode().GetEdges()[0].GetTargetNode()
	require.Equal(t, "InfraTeam1", logAPMNode.GetFlowGraphName())
	require.Len(t, logAPMNode.GetEdges(), 1)
	require.Equal(t, "globalStream", logAPMNode.GetEdges()[0].GetTargetStream().GetName())

	// Test second URL
	flowTreeResult, found := filterTree.GetFlow(newTestAPIStream("maps.googleapis.com"))
	require.True(t, found, "Flow not found")

	// flowRaw = flow.(*Flow)
	userFlow, found := flowTreeResult.GetUserFlow()
	if !found {
		t.Error("User flow not found")
	}
	require.NotNil(t, flowRaw)
	require.Equal(t, "InfraTeam1", userFlow[0].GetName())
	require.Equal(t, "*", userFlow[0].GetFilter().GetURLs())
	root, err = userFlow[0].GetRequestDirection().GetRoot()
	require.NoError(t, err)
	require.NotNil(t, root)

	// ----------------------------------- Request Flow -----------------------------------
	// The InfraTeam1 flow graph should be like this: globalStream -> removePII -> globalStream
	require.True(t, root.IsValid())

	removePIINode := root.GetNode()
	require.Equal(t, "InfraTeam1", removePIINode.GetFlowGraphName())
	require.Len(t, removePIINode.GetEdges(), 1)
	require.Equal(t, "globalStream", removePIINode.GetEdges()[0].GetTargetStream().GetName())

	// ----------------------------------- Response Flow -----------------------------------
	// The InfraTeam1 flow graph should be like this: globalStream -> LogAPM -> globalStream
	root, err = userFlow[0].GetResponseDirection().GetRoot()
	require.NoError(t, err)
	require.True(t, root.IsValid())

	logAPMNode = root.GetNode()
	require.Equal(t, "InfraTeam1", logAPMNode.GetFlowGraphName())
	require.Len(t, logAPMNode.GetEdges(), 1)
	require.Equal(t, "globalStream", logAPMNode.GetEdges()[0].GetTargetStream().GetName())
}
