broker_test.go

// Copyright (c) HashiCorp, Inc.
// SPDX-License-Identifier: MPL-2.0

package eventlogger

import (
	"context"
	"errors"
	"fmt"
	"os"
	"path/filepath"
	"sync"
	"testing"
	"time"

	"github.com/go-test/deep"
	"github.com/hashicorp/go-multierror"
	"github.com/hashicorp/go-uuid"
	"github.com/stretchr/testify/assert"
	"github.com/stretchr/testify/require"
)

// nodesToNodeIDs takes the supplied nodes and registers them with a corresponding,
// generated ID which follows the format 'node-{argument_index}' starting from 0.
func nodesToNodeIDs(t *testing.T, broker *Broker, nodes ...Node) []NodeID {
	t.Helper()
	nodeIDs := make([]NodeID, len(nodes))
	for i, node := range nodes {
		id := NodeID(fmt.Sprintf("node-%d", i))
		err := broker.RegisterNode(id, node)
		if err != nil {
			t.Fatal(err)
		}
		nodeIDs[i] = id
	}
	return nodeIDs
}

func TestBroker(t *testing.T) {
	// Filter out the purple nodes
	filter := &Filter{
		Predicate: func(e *Event) (bool, error) {
			color, ok := e.Payload.(map[string]interface{})["color"]
			return !ok || color != "purple", nil
		},
	}
	// Marshal to JSON
	formatter := &JSONFormatter{}

	formatterFilter := &JSONFormatterFilter{
		Predicate: func(e interface{}) (bool, error) {
			color, ok := e.(*Event).Payload.(map[string]interface{})["color"]
			return !ok || color != "purple", nil
		},
	}

	// Create a broker
	broker, err := NewBroker()
	require.NoError(t, err)
	now := time.Now()
	broker.clock = &clock{now}

	tests := map[string][]Node{
		"with-formatter":        {filter, formatter},
		"with-formatter-filter": {formatterFilter},
	}

	for name, nodes := range tests {
		t.Run(name, func(t *testing.T) {
			tmpDir := t.TempDir()

			// Send to FileSink
			sink := &FileSink{Path: tmpDir, FileName: "file.log"}
			nodes = append(nodes, sink)

			// Register the graph with the broker
			et := EventType("Foo")
			nodeIDs := nodesToNodeIDs(t, broker, nodes...)
			err := broker.RegisterPipeline(Pipeline{
				EventType:  et,
				PipelineID: "id",
				NodeIDs:    nodeIDs,
			})
			if err != nil {
				t.Fatal(err)
			}

			// Set success threshold to 1
			err = broker.SetSuccessThreshold(et, 1)
			if err != nil {
				t.Fatal(err)
			}

			// Process some Events
			payloads := []interface{}{
				map[string]interface{}{
					"color": "red",
					"width": 1,
				},
				map[string]interface{}{
					"color": "green",
					"width": 2,
				},
				map[string]interface{}{
					"color": "purple",
					"width": 3,
				},
				map[string]interface{}{
					"color": "blue",
					"width": 4,
				},
			}
			for _, p := range payloads {
				_, err = broker.Send(context.Background(), et, p)
				if err != nil {
					t.Fatal(err)
				}
			}

			// Check the contents of the log
			dirEntry, err := os.ReadDir(tmpDir)
			if err != nil {
				t.Fatal(err)
			}
			if len(dirEntry) > 1 {
				t.Errorf("Expected 1 log file, got %d", len(dirEntry))
			}

			dat, err := os.ReadFile(filepath.Join(tmpDir, dirEntry[0].Name()))
			if err != nil {
				t.Fatal(err)
			}

			prefix := fmt.Sprintf(`{"created_at":"%s","event_type":"Foo","payload":`, now.Format(time.RFC3339Nano))
			suffix := "}\n"
			var expect string
			for _, s := range []string{`{"color":"red","width":1}`, `{"color":"green","width":2}`, `{"color":"blue","width":4}`} {
				expect += fmt.Sprintf("%s%s%s", prefix, s, suffix)
			}
			if diff := deep.Equal(string(dat), expect); diff != nil {
				t.Fatal(diff)
			}
		})
	}
}

func TestPipeline(t *testing.T) {
	broker, err := NewBroker()
	require.NoError(t, err)

	// invalid pipeline
	nodeIDs := nodesToNodeIDs(t, broker, &Filter{Predicate: nil})
	err = broker.RegisterPipeline(Pipeline{
		EventType:  "t",
		PipelineID: "id",
		NodeIDs:    nodeIDs,
	})
	require.Error(t, err)

	if diff := deep.Equal("non-sink node has no children", err.Error()); diff != nil {
		t.Fatal(diff)
	}

	// Construct a graph
	f1 := &JSONFormatter{}
	s1 := &testSink{}
	p1 := nodesToNodeIDs(t, broker, f1, s1)
	err = broker.RegisterPipeline(Pipeline{
		EventType:  "t",
		PipelineID: "s1",
		NodeIDs:    p1,
	})
	require.NoError(t, err)

	// register again
	err = broker.RegisterPipeline(Pipeline{
		EventType:  "t",
		PipelineID: "s1",
		NodeIDs:    p1,
	})
	require.NoError(t, err)

	// send a payload
	payload := map[string]interface{}{
		"color": "red",
		"width": 1,
	}
	_, err = broker.Send(context.Background(), "t", payload)
	require.NoError(t, err)
	require.Equal(t, 1, s1.count)

	// Construct another graph
	s2 := &testSink{}
	p2 := nodesToNodeIDs(t, broker, f1, s2)
	err = broker.RegisterPipeline(Pipeline{
		EventType:  "t",
		PipelineID: "s2",
		NodeIDs:    p2,
	})
	require.NoError(t, err)

	// send a payload
	_, err = broker.Send(context.Background(), "t", payload)
	require.NoError(t, err)
	require.Equal(t, 2, s1.count)
	require.Equal(t, 1, s2.count)

	// remove second graph
	err = broker.RemovePipeline("t", "s2")
	require.NoError(t, err)

	// send a payload
	_, err = broker.Send(context.Background(), "t", payload)
	require.NoError(t, err)
	require.Equal(t, 3, s1.count)
	require.Equal(t, 1, s2.count)

	// remove
	err = broker.RemovePipeline("t", "s1")
	require.NoError(t, err)

	// remove again
	err = broker.RemovePipeline("t", "s1")
	require.NoError(t, err)
}

// TestPipelineRaceCondition can't fail, but it can check if there is a race condition
// in iterating through, adding, or removing pipelines.
func TestPipelineRaceCondition(t *testing.T) {
	broker, err := NewBroker()
	require.NoError(t, err)

	eventType := EventType("t")
	var pipelines []PipelineID
	var sinks []*testSink

	wg := sync.WaitGroup{}
	wg.Add(2)

	// register a bunch of pipelines
	go func(t *testing.T) {
		for i := 0; i < 10; i++ {
			// Construct a graph
			f1 := &JSONFormatter{}
			s1 := &testSink{}
			p1 := nodesToNodeIDs(t, broker, f1, s1)

			id, err := uuid.GenerateUUID()
			if err != nil {
				panic(err)
			}

			err = broker.RegisterPipeline(Pipeline{
				EventType:  eventType,
				PipelineID: PipelineID(id),
				NodeIDs:    p1,
			})
			if err != nil {
				panic(err)
			}

			pipelines = append(pipelines, PipelineID(id))
			sinks = append(sinks, s1)
		}

		for _, id := range pipelines {
			err := broker.RemovePipeline(eventType, id)
			if err != nil {
				panic(err)
			}
		}
		wg.Done()
	}(t)

	go func() {
		for i := 0; i < 100; i++ {
			// send payloads
			payload := map[string]interface{}{
				"color": "red",
				"width": 1,
			}
			_, _ = broker.Send(context.Background(), eventType, payload)
		}
		wg.Done()
	}()

	wg.Wait()
}

type testSink struct {
	count int
}

var _ Node = &testSink{}

func (ts *testSink) Type() NodeType {
	return NodeTypeSink
}

func (ts *testSink) Process(_ context.Context, _ *Event) (*Event, error) {
	ts.count++
	return nil, nil
}

func (ts *testSink) Reopen() error {
	return nil
}

func (ts *testSink) Name() string {
	return "testSink"
}

// TestSetSuccessThreshold tests that we can set the required success threshold for
// a specific event type in the graph.
func TestSetSuccessThreshold(t *testing.T) {
	t.Parallel()

	threshold := 2
	b, err := NewBroker()
	require.NoError(t, err)

	err = b.SetSuccessThreshold("", threshold)
	require.Error(t, err)
	require.EqualError(t, err, "event type cannot be empty")

	err = b.SetSuccessThreshold("t", threshold)
	require.NoError(t, err)

	g, ok := b.graphs["t"]
	require.True(t, ok)
	require.Equal(t, threshold, g.successThreshold)

	err = b.SetSuccessThreshold("t", -1)
	require.Error(t, err)
	require.EqualError(t, err, "successThreshold must be 0 or greater")
}

// TestSetSuccessThresholdSinks tests that we can set the required sink success
// threshold for a specific event type in the graph.
func TestSetSuccessThresholdSinks(t *testing.T) {
	t.Parallel()

	threshold := 2
	b, err := NewBroker()
	require.NoError(t, err)

	err = b.SetSuccessThresholdSinks("", threshold)
	require.Error(t, err)
	require.EqualError(t, err, "event type cannot be empty")

	err = b.SetSuccessThresholdSinks("t", threshold)
	require.NoError(t, err)

	g, ok := b.graphs["t"]
	require.True(t, ok)
	require.Equal(t, threshold, g.successThresholdSinks)

	err = b.SetSuccessThresholdSinks("t", -1)
	require.Error(t, err)
	require.EqualError(t, err, "successThresholdSinks must be 0 or greater")
}

func TestSuccessThreshold_NoEventType(t *testing.T) {
	t.Parallel()

	b, err := NewBroker()
	require.NoError(t, err)
	err = b.SetSuccessThreshold("t", 5)
	require.NoError(t, err)
	n, ok := b.SuccessThreshold("x")
	require.False(t, ok)
	require.Equal(t, 0, n)
}

func TestSuccessThreshold(t *testing.T) {
	t.Parallel()

	b, err := NewBroker()
	require.NoError(t, err)
	err = b.SetSuccessThreshold("t", 5)
	require.NoError(t, err)
	n, ok := b.SuccessThreshold("t")
	require.True(t, ok)
	require.Equal(t, 5, n)
}

func TestSuccessThresholdSinks_NoEventType(t *testing.T) {
	t.Parallel()

	b, err := NewBroker()
	require.NoError(t, err)
	err = b.SetSuccessThresholdSinks("t", 5)
	require.NoError(t, err)
	n, ok := b.SuccessThresholdSinks("x")
	require.False(t, ok)
	require.Equal(t, 0, n)
}

func TestSuccessThresholdSinks(t *testing.T) {
	t.Parallel()

	b, err := NewBroker()
	require.NoError(t, err)
	err = b.SetSuccessThresholdSinks("t", 5)
	require.NoError(t, err)
	n, ok := b.SuccessThresholdSinks("t")
	require.True(t, ok)
	require.Equal(t, 5, n)
}

// TestRemovePipelineAndNodes exercises the behavior that removes a pipeline and
// any nodes associated with that pipeline, if they are not referenced by other pipelines.
// The test is relatively long as it is focused on the state of the broker across
// multiple operations.
func TestRemovePipelineAndNodes(t *testing.T) {
	t.Parallel()

	ctx := context.Background()
	broker, err := NewBroker()
	require.NoError(t, err)

	// Construct a graph
	f1 := &JSONFormatter{}
	s1 := &testSink{}
	nodeIDs := nodesToNodeIDs(t, broker, f1, s1)

	// Register single pipeline
	err = broker.RegisterPipeline(Pipeline{
		EventType:  "t",
		PipelineID: "p1",
		NodeIDs:    nodeIDs,
	})
	require.NoError(t, err)

	// Deregister the only pipeline we have
	ok, err := broker.RemovePipelineAndNodes(ctx, EventType("t"), PipelineID("p1"))
	require.NoError(t, err)
	require.True(t, ok)
	require.Empty(t, broker.nodes)

	// Attempt to register 2nd pipeline which references now deleted nodes
	err = broker.RegisterPipeline(Pipeline{
		EventType:  "t",
		PipelineID: "p2",
		NodeIDs:    nodeIDs,
	})
	require.Error(t, err)
	require.EqualError(t, err, "node ID \"node-0\" not registered")

	// Re-register nodes and 2 pipelines which use the same nodes
	nodeIDs = nodesToNodeIDs(t, broker, f1, s1)
	err = broker.RegisterPipeline(Pipeline{
		EventType:  "t",
		PipelineID: "p1",
		NodeIDs:    nodeIDs,
	})
	require.NoError(t, err)

	err = broker.RegisterPipeline(Pipeline{
		EventType:  "t",
		PipelineID: "p2",
		NodeIDs:    nodeIDs,
	})
	require.NoError(t, err)

	// Deregister pipeline p1, leave pipeline p2 in place
	ok, err = broker.RemovePipelineAndNodes(ctx, EventType("t"), PipelineID("p1"))
	require.NoError(t, err)
	require.True(t, ok)
	require.NotEmpty(t, broker.nodes)
	require.Equal(t, 2, len(broker.nodes))

	// Whip the nodes out from underneath a pipeline and then try to deregister it
	broker.nodes = nil
	ok, err = broker.RemovePipelineAndNodes(ctx, EventType("t"), "p2")
	require.Error(t, err)
	require.True(t, ok)
	me, ok := err.(*multierror.Error)
	require.True(t, ok)
	require.Equal(t, 2, me.Len())
}

// TestRemovePipelineAndNodes_BadEventType tests attempting to remove a pipeline
// with an event type we haven't previously registered.
func TestRemovePipelineAndNodes_BadEventType(t *testing.T) {
	t.Parallel()

	ctx := context.Background()
	broker, err := NewBroker()
	require.NoError(t, err)
	ok, err := broker.RemovePipelineAndNodes(ctx, EventType("foo"), PipelineID("p2"))
	require.Error(t, err)
	require.False(t, ok)
	require.EqualError(t, err, "no graph for EventType foo")
}

// TestRegisterPipeline_BadParameters ensures that we perform sanity checking
// on the parameters passed in when we attempt to register a pipeline.
func TestRegisterPipeline_BadParameters(t *testing.T) {
	t.Parallel()

	tests := map[string]struct {
		pipelineID string
		eventType  string
		nodes      []NodeID
		error      string
	}{
		"no-pipelineID": {
			pipelineID: "",
			eventType:  "t",
			nodes:      []NodeID{"1", "2", "3"},
			error:      "pipeline ID is required",
		},
		"no-eventType": {
			pipelineID: "1",
			eventType:  "",
			nodes:      []NodeID{"1", "2", "3"},
			error:      "event type is required",
		},
		"nil-nodes": {
			pipelineID: "1",
			eventType:  "t",
			nodes:      nil,
			error:      "node IDs are required",
		},
		"empty-nodes": {
			pipelineID: "1",
			eventType:  "t",
			nodes:      []NodeID{},
			error:      "node IDs are required",
		},
		"bad-nodes": {
			pipelineID: "1",
			eventType:  "t",
			nodes:      []NodeID{"", ""},
			error:      "node ID cannot be empty",
		},
	}

	for name, tc := range tests {
		name := name
		tc := tc
		t.Run(name, func(t *testing.T) {
			t.Parallel()
			broker, err := NewBroker()
			require.NoError(t, err)

			// Register some nodes so they exist (1, 2, 3 should appear in the test cases)
			node := &JSONFormatter{}
			err = broker.RegisterNode("1", node)
			require.NoError(t, err)
			err = broker.RegisterNode("2", node)
			require.NoError(t, err)
			err = broker.RegisterNode("3", node)
			require.NoError(t, err)

			err = broker.RegisterPipeline(Pipeline{
				PipelineID: PipelineID(tc.pipelineID),
				EventType:  EventType(tc.eventType),
				NodeIDs:    tc.nodes,
			})

			require.Error(t, err)
			me, ok := err.(*multierror.Error)
			require.True(t, ok)
			require.EqualError(t, me.Unwrap(), tc.error)
		})
	}
}

// TestRemovePipelineAndNodes_BadParameters ensures that we perform sanity checking
// on the parameters passed in when we attempt to remove both individual pipelines
// and also pipelines and nodes together.
func TestRemovePipelineAndNodes_BadParameters(t *testing.T) {
	t.Parallel()

	ctx := context.Background()
	tests := map[string]struct {
		pipelineID string
		eventType  string
		error      string
	}{
		"no-pipelineID": {
			pipelineID: "",
			eventType:  "t",
			error:      "pipeline ID cannot be empty",
		},
		"no-eventType": {
			pipelineID: "1",
			eventType:  "",
			error:      "event type cannot be empty",
		},
		"wrong-eventType": {
			pipelineID: "1",
			eventType:  "foo",
			error:      "no graph for EventType foo",
		},
	}

	for name, tc := range tests {
		name := name
		tc := tc
		t.Run(name, func(t *testing.T) {
			t.Parallel()

			broker, err := NewBroker()
			require.NoError(t, err)

			// Test removing the pipeline and nodes
			ok, err := broker.RemovePipelineAndNodes(ctx, EventType(tc.eventType), PipelineID(tc.pipelineID))
			require.Error(t, err)
			require.False(t, ok)
			require.EqualError(t, err, tc.error)
			// Test removing just the pipeline
			err = broker.RemovePipeline(EventType(tc.eventType), PipelineID(tc.pipelineID))
			require.Error(t, err)
			require.EqualError(t, err, tc.error)
		})
	}
}

// TestPipelineValidate tests that given a Pipeline in various states we can assert
// that the Pipeline is valid or invalid.
func TestPipelineValidate(t *testing.T) {
	t.Parallel()

	tests := map[string]struct {
		pipelineID       string
		eventType        string
		nodes            []NodeID
		expectValid      bool
		expectErrorCount int
	}{
		"valid": {
			pipelineID:  "1",
			eventType:   "t",
			nodes:       []NodeID{"a", "b"},
			expectValid: true,
		},
		"no-pipelineID": {
			pipelineID:       "",
			eventType:        "t",
			nodes:            []NodeID{"a", "b"},
			expectValid:      false,
			expectErrorCount: 1,
		},
		"no-event-type": {
			pipelineID:       "1",
			eventType:        "",
			nodes:            []NodeID{"a", "b"},
			expectValid:      false,
			expectErrorCount: 1,
		},
		"empty-nodes": {
			pipelineID:       "1",
			eventType:        "t",
			nodes:            []NodeID{},
			expectValid:      false,
			expectErrorCount: 1,
		},
		"fully-invalid": {
			pipelineID:       "",
			eventType:        "",
			nodes:            []NodeID{},
			expectValid:      false,
			expectErrorCount: 3,
		},
		"fully-invalid-no-nodeIDs": {
			pipelineID:       "",
			eventType:        "",
			nodes:            []NodeID{"", ""},
			expectValid:      false,
			expectErrorCount: 3,
		},
	}

	for name, tc := range tests {
		name := name
		tc := tc
		t.Run(name, func(t *testing.T) {
			t.Parallel()

			p := Pipeline{
				PipelineID: PipelineID(tc.pipelineID),
				EventType:  EventType(tc.eventType),
				NodeIDs:    tc.nodes,
			}

			err := p.validate()
			switch tc.expectValid {
			case true:
				require.NoError(t, err)
			default:
				require.Error(t, err)
				me, ok := err.(*multierror.Error)
				require.True(t, ok)
				require.Equal(t, tc.expectErrorCount, me.Len())
			}
		})
	}
}

// TestRegisterNode_NoID ensures we cannot register a Node with an empty ID.
func TestRegisterNode_NoID(t *testing.T) {
	t.Parallel()

	b, err := NewBroker()
	require.NoError(t, err)
	err = b.RegisterNode("", &JSONFormatter{})
	require.Error(t, err)
	require.EqualError(t, err, "unable to register node, node ID cannot be empty: invalid parameter")
}

// TestBroker_RegisterNode_AllowOverwrite_Implicit is used to prove that nodes can be
// overwritten when a Broker has been implicitly configured with the AllowOverwrite policy.
// This is the default in order to maintain pre-existing behavior.
func TestBroker_RegisterNode_AllowOverwrite_Implicit(t *testing.T) {
	t.Parallel()

	b, err := NewBroker()
	require.NoError(t, err)
	err = b.RegisterNode("n1", &JSONFormatter{})
	require.NoError(t, err)
	err = b.RegisterNode("n1", &FileSink{})
	require.NoError(t, err)
}

// TestBroker_RegisterNode_AllowOverwrite_Explicit is used to prove that nodes can be
// overwritten when a Broker has been explicitly configured with the AllowOverwrite policy.
func TestBroker_RegisterNode_AllowOverwrite_Explicit(t *testing.T) {
	t.Parallel()

	b, err := NewBroker()
	require.NoError(t, err)
	err = b.RegisterNode("n1", &JSONFormatter{}, WithNodeRegistrationPolicy(AllowOverwrite))
	require.NoError(t, err)
	err = b.RegisterNode("n1", &FileSink{})
	require.NoError(t, err)
}

// TestBroker_RegisterNode_DenyOverwrite is used to prove that nodes can't be
// overwritten when a Broker has been configured with the DenyOverwrite policy.
func TestBroker_RegisterNode_DenyOverwrite(t *testing.T) {
	t.Parallel()

	b, err := NewBroker()
	require.NoError(t, err)
	err = b.RegisterNode("n1", &JSONFormatter{}, WithNodeRegistrationPolicy(DenyOverwrite))
	require.NoError(t, err)
	err = b.RegisterNode("n1", &FileSink{})
	require.Error(t, err)
	require.EqualError(t, err, "node ID \"n1\" is already registered, configured policy prevents overwriting")
}

// TestBroker_RegisterNode_AllowThenDenyOverwrite is used to prove that nodes can be
// overwritten and then updated to prevent overwriting.
func TestBroker_RegisterNode_AllowThenDenyOverwrite(t *testing.T) {
	t.Parallel()

	b, err := NewBroker()
	require.NoError(t, err)
	err = b.RegisterNode("n1", &JSONFormatter{}, WithNodeRegistrationPolicy(AllowOverwrite))
	require.NoError(t, err)
	err = b.RegisterNode("n1", &FileSink{}, WithNodeRegistrationPolicy(DenyOverwrite))
	require.NoError(t, err)
	err = b.RegisterNode("n1", &FileSink{})
	require.Error(t, err)
}

// TestRemoveNode ensures we cannot remove a Node with an empty ID.
func TestRemoveNode(t *testing.T) {
	t.Parallel()

	b, err := NewBroker()
	require.NoError(t, err)
	err = b.RegisterNode("n1", &JSONFormatter{})
	require.NoError(t, err)
	err = b.RemoveNode(context.Background(), "n1")
	require.NoError(t, err)
}

// TestRemoveNode_NoID ensures we cannot remove a Node with an empty ID.
func TestRemoveNode_NoID(t *testing.T) {
	t.Parallel()

	b, err := NewBroker()
	require.NoError(t, err)
	err = b.RemoveNode(context.Background(), "")
	require.Error(t, err)
	require.EqualError(t, err, "unable to remove node, node ID cannot be empty: invalid parameter")
}

// TestRemoveNode_NotFound ensures we cannot remove a Node that has not been registered
func TestRemoveNode_NotFound(t *testing.T) {
	t.Parallel()

	b, err := NewBroker()
	require.NoError(t, err)
	err = b.RemoveNode(context.Background(), "n1")
	require.Error(t, err)
	require.EqualError(t, err, "node not found: \"n1\"")
}

// TestRemoveNode_StillReferenced ensures we cannot remote a Node that is still referenced by a pipeline
func TestRemoveNode_StillReferenced(t *testing.T) {
	t.Parallel()

	b, err := NewBroker()
	require.NoError(t, err)
	err = b.RegisterNode("n1", &JSONFormatter{})
	require.NoError(t, err)
	b.nodes["n1"].referenceCount = 2
	err = b.RemoveNode(context.Background(), "n1")
	require.Error(t, err)
	require.EqualError(t, err, "cannot remove node, as it is still in use by 1 or more pipelines: \"n1\"")
}

// TestDeregisterNode_Force ensures we can decrement the reference to a Node that is still referenced
// by a pipeline by using the force option
func TestRemoveNode_StillReferencedDecrement(t *testing.T) {
	t.Parallel()

	b, err := NewBroker()
	require.NoError(t, err)
	err = b.RegisterNode("n1", &JSONFormatter{})
	require.NoError(t, err)
	b.nodes["n1"].referenceCount = 2
	err = b.removeNode(context.Background(), "n1", true)
	require.NoError(t, err)
	require.Equal(t, 1, b.nodes["n1"].referenceCount)
}

// TestBroker_RegisterPipeline_AllowOverwrite_Implicit is used to prove that pipelines can be
// overwritten when a Broker has been implicitly configured with the AllowOverwrite policy.
// This is the default in order to maintain pre-existing behavior.
func TestBroker_RegisterPipeline_AllowOverwrite_Implicit(t *testing.T) {
	t.Parallel()

	b, err := NewBroker()
	require.NoError(t, err)

	err = b.RegisterNode("f1", &JSONFormatter{})
	require.NoError(t, err)

	err = b.RegisterNode("f2", &JSONFormatter{})
	require.NoError(t, err)

	err = b.RegisterNode("s1", &FileSink{})
	require.NoError(t, err)

	err = b.RegisterPipeline(Pipeline{
		PipelineID: "p1",
		EventType:  "t",
		NodeIDs:    []NodeID{"f1", "s1"},
	})
	require.NoError(t, err)

	err = b.RegisterPipeline(Pipeline{
		PipelineID: "p1",
		EventType:  "t",
		NodeIDs:    []NodeID{"f2", "s1"},
	})
	require.NoError(t, err)
}

// TestBroker_RegisterPipeline_AllowOverwrite_Explicit is used to prove that pipelines can be
// overwritten when a Broker has been explicitly configured with the AllowOverwrite policy.
func TestBroker_RegisterPipeline_AllowOverwrite_Explicit(t *testing.T) {
	t.Parallel()

	b, err := NewBroker()
	require.NoError(t, err)

	err = b.RegisterNode("f1", &JSONFormatter{})
	require.NoError(t, err)

	err = b.RegisterNode("f2", &JSONFormatter{})
	require.NoError(t, err)

	err = b.RegisterNode("s1", &FileSink{})
	require.NoError(t, err)

	err = b.RegisterPipeline(Pipeline{
		PipelineID: "p1",
		EventType:  "t",
		NodeIDs:    []NodeID{"f1", "s1"},
	}, WithPipelineRegistrationPolicy(AllowOverwrite))
	require.NoError(t, err)

	err = b.RegisterPipeline(Pipeline{
		PipelineID: "p1",
		EventType:  "t",
		NodeIDs:    []NodeID{"f2", "s1"},
	})
	require.NoError(t, err)
}

// TestBroker_RegisterPipeline_DenyOverwrite is used to prove that pipelines can't
// be overwritten when a Broker has been configured with the DenyOverwrite policy.
func TestBroker_RegisterPipeline_DenyOverwrite(t *testing.T) {
	t.Parallel()

	b, err := NewBroker()
	require.NoError(t, err)
	require.NotNil(t, b)

	// Ensure no side effects from setting thresholds before we have registered a pipeline
	err = b.SetSuccessThreshold("t", 1)
	require.NoError(t, err)
	err = b.SetSuccessThresholdSinks("t", 1)
	require.NoError(t, err)

	err = b.RegisterNode("f1", &JSONFormatter{})
	require.NoError(t, err)

	err = b.RegisterNode("f2", &JSONFormatter{})
	require.NoError(t, err)

	err = b.RegisterNode("s1", &FileSink{})
	require.NoError(t, err)

	err = b.RegisterPipeline(Pipeline{
		PipelineID: "p1",
		EventType:  "t",
		NodeIDs:    []NodeID{"f1", "s1"},
	}, WithPipelineRegistrationPolicy(DenyOverwrite))
	require.NoError(t, err)

	err = b.RegisterPipeline(Pipeline{
		PipelineID: "p1",
		EventType:  "t",
		NodeIDs:    []NodeID{"f2", "s1"},
	})
	require.Error(t, err)
	require.EqualError(t, err, "pipeline ID \"p1\" is already registered, configured policy prevents overwriting")
}

// TestBroker_RegisterPipeline_DenyOverwrite is used to prove that pipelines can't
// be overwritten when a Broker has been configured with the DenyOverwrite policy.
func TestBroker_RegisterPipeline_AllowThenDenyOverwrite(t *testing.T) {
	t.Parallel()

	b, err := NewBroker()
	require.NoError(t, err)
	require.NotNil(t, b)

	// Ensure no side effects from setting thresholds before we have registered a pipeline
	err = b.SetSuccessThreshold("t", 1)
	require.NoError(t, err)
	err = b.SetSuccessThresholdSinks("t", 1)
	require.NoError(t, err)

	err = b.RegisterNode("f1", &JSONFormatter{})
	require.NoError(t, err)

	err = b.RegisterNode("f2", &JSONFormatter{})
	require.NoError(t, err)

	err = b.RegisterNode("s1", &FileSink{})
	require.NoError(t, err)

	err = b.RegisterPipeline(Pipeline{
		PipelineID: "p1",
		EventType:  "t",
		NodeIDs:    []NodeID{"f1", "s1"},
	}, WithPipelineRegistrationPolicy(AllowOverwrite))
	require.NoError(t, err)

	err = b.RegisterPipeline(Pipeline{
		PipelineID: "p1",
		EventType:  "t",
		NodeIDs:    []NodeID{"f2", "s1"},
	}, WithPipelineRegistrationPolicy(DenyOverwrite))
	require.NoError(t, err)

	err = b.RegisterPipeline(Pipeline{
		PipelineID: "p1",
		EventType:  "t",
		NodeIDs:    []NodeID{"f1", "s1"},
	})
	require.Error(t, err)
	require.EqualError(t, err, "pipeline ID \"p1\" is already registered, configured policy prevents overwriting")
}

func TestBroker_RegisterPipeline_WithCloser(t *testing.T) {
	t.Parallel()

	ctx := context.Background()
	b, err := NewBroker()
	require.NoError(t, err)

	mc := &mockCloserWithWrapper{n: &mockCloser{}}
	err = b.RegisterNode("mc1", mc)
	require.NoError(t, err)

	err = b.RegisterNode("f1", &JSONFormatter{})
	require.NoError(t, err)

	err = b.RegisterPipeline(Pipeline{
		PipelineID: "p1",
		EventType:  "t",
		NodeIDs:    []NodeID{"f1", "mc1"},
	})
	require.NoError(t, err)

	ok, err := b.RemovePipelineAndNodes(ctx, "t", "p1")
	require.NoError(t, err)
	require.True(t, ok)

	assert.True(t, mc.n.closed)
}

func TestBroker_RegisterPipeline_WithCloserError(t *testing.T) {
	t.Parallel()

	ctx := context.Background()
	b, err := NewBroker()
	require.NoError(t, err)

	mc := &mockCloserWithWrapper{n: &mockCloser{closeErr: errors.New("close error")}}
	err = b.RegisterNode("mc1", mc)
	require.NoError(t, err)

	err = b.RegisterNode("f1", &JSONFormatter{})
	require.NoError(t, err)

	err = b.RegisterPipeline(Pipeline{
		PipelineID: "p1",
		EventType:  "t",
		NodeIDs:    []NodeID{"f1", "mc1"},
	})
	require.NoError(t, err)

	ok, err := b.RemovePipelineAndNodes(ctx, "t", "p1")
	require.Error(t, err)
	require.True(t, ok)

	assert.Contains(t, err.Error(), "close error")
	assert.False(t, mc.n.closed)
}

func TestBroker_IsAnyPipelineRegistered(t *testing.T) {
	t.Parallel()

	b, err := NewBroker()
	require.NoError(t, err)

	err = b.RegisterNode("f1", &JSONFormatter{})
	require.NoError(t, err)

	err = b.RegisterNode("s1", &FileSink{})
	require.NoError(t, err)

	err = b.RegisterPipeline(Pipeline{
		PipelineID: "p1",
		EventType:  "t",
		NodeIDs:    []NodeID{"f1", "s1"},
	})
	require.NoError(t, err)

	require.True(t, b.IsAnyPipelineRegistered("t"))
	require.False(t, b.IsAnyPipelineRegistered("i-do-not-exist"))
}

func TestBroker_IsAnyPipelineRegistered_WithFailedRegistration(t *testing.T) {
	t.Parallel()

	b, err := NewBroker()
	require.NoError(t, err)

	err = b.RegisterPipeline(Pipeline{
		PipelineID: "p1",
		EventType:  "t",
		NodeIDs:    []NodeID{},
	})
	require.Error(t, err)

	require.False(t, b.IsAnyPipelineRegistered("t"))
}

func TestBroker_IsAnyPipelineRegisteredRaceCondition(t *testing.T) {
	t.Parallel()

	b, err := NewBroker()
	require.NoError(t, err)

	err = b.RegisterNode("f1", &JSONFormatter{})
	require.NoError(t, err)

	err = b.RegisterNode("s1", &FileSink{})
	require.NoError(t, err)

	n := 100
	var wg sync.WaitGroup
	for i := 0; i < n; i++ {
		pipelineID, err := uuid.GenerateUUID()
		require.NoError(t, err)
		eventType, err := uuid.GenerateUUID()
		require.NoError(t, err)

		wg.Add(1)
		go func() {
			defer wg.Done()
			err := b.RegisterPipeline(Pipeline{
				PipelineID: PipelineID(pipelineID),
				EventType:  EventType(eventType),
				NodeIDs:    []NodeID{"f1", "s1"},
			})
			require.NoError(t, err)
		}()

		wg.Add(1)
		go func() {
			defer wg.Done()
			_ = b.IsAnyPipelineRegistered("t")
		}()
	}
	wg.Wait()
}

func TestBroker_Status_Complete(t *testing.T) {
	t.Parallel()

	tests := map[string]struct {
		statusSinks   []NodeID
		expectedSinks []NodeID
	}{
		"none": {
			statusSinks:   nil,
			expectedSinks: nil,
		},
		"one": {
			statusSinks:   []NodeID{NodeID("foo")},
			expectedSinks: []NodeID{NodeID("foo")},
		},
		"many": {
			statusSinks:   []NodeID{NodeID("foo"), NodeID("bar")},
			expectedSinks: []NodeID{NodeID("foo"), NodeID("bar")},
		},
	}
	for name, tc := range tests {
		t.Run(name, func(t *testing.T) {
			t.Parallel()

			s := &Status{
				complete: tc.statusSinks,
			}

			require.Len(t, s.Complete(), len(tc.expectedSinks))
			require.Equal(t, tc.expectedSinks, s.Complete())
		})
	}
}

func TestBroker_Status_CompleteSinks(t *testing.T) {
	t.Parallel()

	tests := map[string]struct {
		statusSinks   []NodeID
		expectedSinks []NodeID
	}{
		"none": {
			statusSinks:   nil,
			expectedSinks: nil,
		},
		"one": {
			statusSinks:   []NodeID{NodeID("foo")},
			expectedSinks: []NodeID{NodeID("foo")},
		},
		"many": {
			statusSinks:   []NodeID{NodeID("foo"), NodeID("bar")},
			expectedSinks: []NodeID{NodeID("foo"), NodeID("bar")},
		},
	}
	for name, tc := range tests {
		name := name
		tc := tc
		t.Run(name, func(t *testing.T) {
			t.Parallel()

			s := &Status{
				completeSinks: tc.statusSinks,
			}

			require.Len(t, s.CompleteSinks(), len(tc.expectedSinks))
			require.Equal(t, tc.expectedSinks, s.CompleteSinks())
		})
	}
}

func TestStatus_getError(t *testing.T) {
	t.Parallel()

	tests := map[string]struct {
		complete             []NodeID
		completeSinks        []NodeID
		isErrorExpected      bool
		expectedErrorMessage string
		shouldCancelContext  bool
	}{
		"no-problems": {
			complete:        []NodeID{"1", "2"},
			completeSinks:   []NodeID{"2"},
			isErrorExpected: false,
		},
		"no-problems-context-cancelled": {
			complete:            []NodeID{"1", "2"},
			completeSinks:       []NodeID{"2"},
			shouldCancelContext: true,
			isErrorExpected:     false,
		},
		"not-enough-nodes": {
			complete:             []NodeID{},
			completeSinks:        []NodeID{},
			isErrorExpected:      true,
			expectedErrorMessage: "event not processed by enough 'filter' and 'sink' nodes",
		},
		"not-enough-nodes-context-cancelled": {
			complete:             []NodeID{},
			completeSinks:        []NodeID{},
			shouldCancelContext:  true,
			isErrorExpected:      true,
			expectedErrorMessage: "event not processed by enough 'filter' and 'sink' nodes\ncontext canceled",
		},
		"not-enough-sink-nodes": {
			complete:             []NodeID{"1"},
			completeSinks:        []NodeID{},
			isErrorExpected:      true,
			expectedErrorMessage: "event not processed by enough 'sink' nodes",
		},
		"not-enough-sink-nodes-context-cancelled": {
			complete:             []NodeID{"1"},
			completeSinks:        []NodeID{},
			shouldCancelContext:  true,
			isErrorExpected:      true,
			expectedErrorMessage: "event not processed by enough 'sink' nodes\ncontext canceled",
		},
	}

	for name, tc := range tests {
		name := name
		tc := tc
		t.Run(name, func(t *testing.T) {
			t.Parallel()

			ctx, cancel := context.WithCancel(context.Background())
			t.Cleanup(func() { cancel() })

			s := Status{
				complete:      tc.complete,
				completeSinks: tc.completeSinks,
			}

			if tc.shouldCancelContext {
				// Cancel the context NOW if we want to test the output from getError.
				cancel()
			}

			// NOTE: threshold and thresholdSinks can be added to tests struct
			// if needed for additional tests.
			err := s.getError(ctx.Err(), 1, 1)

			switch {
			case tc.isErrorExpected:
				require.Error(t, err)
				require.EqualError(t, err, tc.expectedErrorMessage)
				if tc.shouldCancelContext {
					errors.Is(err, context.Canceled)
				}
			default:
				require.NoError(t, err)
			}
		})
	}

}