// Copyright The OpenTelemetry Authors
//
// Licensed under the Apache License, Version 2.0 (the "License");
// you may not use this file except in compliance with the License.
// You may obtain a copy of the License at
//
//       http://www.apache.org/licenses/LICENSE-2.0
//
// Unless required by applicable law or agreed to in writing, software
// distributed under the License is distributed on an "AS IS" BASIS,
// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
// See the License for the specific language governing permissions and
// limitations under the License.

package service // import "go.opentelemetry.io/collector/service"

import (
	"context"
	"hash/fnv"
	"strings"

	"go.opentelemetry.io/collector/component"
	"go.opentelemetry.io/collector/connector"
	"go.opentelemetry.io/collector/consumer"
	"go.opentelemetry.io/collector/service/internal/components"
	"go.opentelemetry.io/collector/service/internal/fanoutconsumer"
)

const (
	receiverSeed      = "receiver"
	processorSeed     = "processor"
	exporterSeed      = "exporter"
	connectorSeed     = "connector"
	capabilitiesSeed  = "capabilities"
	fanOutToExporters = "fanout_to_exporters"
)

type nodeID int64

func (n nodeID) ID() int64 {
	return int64(n)
}

func newNodeID(parts ...string) nodeID {
	h := fnv.New64a()
	h.Write([]byte(strings.Join(parts, "|")))
	return nodeID(h.Sum64())
}

type consumerNode interface {
	getConsumer() baseConsumer
}

// A receiver instance can be shared by multiple pipelines of the same type.
// Therefore, nodeID is derived from "pipeline type" and "component ID".
type receiverNode struct {
	nodeID
	componentID  component.ID
	pipelineType component.DataType
	component.Component
}

func newReceiverNode(pipelineID component.ID, recvID component.ID) *receiverNode {
	return &receiverNode{
		nodeID:       newNodeID(receiverSeed, string(pipelineID.Type()), recvID.String()),
		componentID:  recvID,
		pipelineType: pipelineID.Type(),
	}
}

var _ consumerNode = &processorNode{}

// Every processor instance is unique to one pipeline.
// Therefore, nodeID is derived from "pipeline ID" and "component ID".
type processorNode struct {
	nodeID
	componentID component.ID
	pipelineID  component.ID
	component.Component
}

func newProcessorNode(pipelineID, procID component.ID) *processorNode {
	return &processorNode{
		nodeID:      newNodeID(processorSeed, pipelineID.String(), procID.String()),
		componentID: procID,
		pipelineID:  pipelineID,
	}
}

func (n *processorNode) getConsumer() baseConsumer {
	return n.Component.(baseConsumer)
}

var _ consumerNode = &exporterNode{}

// An exporter instance can be shared by multiple pipelines of the same type.
// Therefore, nodeID is derived from "pipeline type" and "component ID".
type exporterNode struct {
	nodeID
	componentID  component.ID
	pipelineType component.DataType
	component.Component
}

func newExporterNode(pipelineID component.ID, exprID component.ID) *exporterNode {
	return &exporterNode{
		nodeID:       newNodeID(exporterSeed, string(pipelineID.Type()), exprID.String()),
		componentID:  exprID,
		pipelineType: pipelineID.Type(),
	}
}

func (n *exporterNode) getConsumer() baseConsumer {
	return n.Component.(baseConsumer)
}

var _ consumerNode = &connectorNode{}

// A connector instance connects one pipeline type to one other pipeline type.
// Therefore, nodeID is derived from "exporter pipeline type", "receiver pipeline type", and "component ID".
type connectorNode struct {
	nodeID
	componentID      component.ID
	exprPipelineType component.DataType
	rcvrPipelineType component.DataType
	component.Component
}

func newConnectorNode(exprPipelineType, rcvrPipelineType component.DataType, connID component.ID) *connectorNode {
	return &connectorNode{
		nodeID:           newNodeID(connectorSeed, connID.String(), string(exprPipelineType), string(rcvrPipelineType)),
		componentID:      connID,
		exprPipelineType: exprPipelineType,
		rcvrPipelineType: rcvrPipelineType,
	}
}

func (n *connectorNode) getConsumer() baseConsumer {
	return n.Component.(baseConsumer)
}

func buildConnector(
	ctx context.Context,
	componentID component.ID,
	tel component.TelemetrySettings,
	info component.BuildInfo,
	builder *connector.Builder,
	exprPipelineType component.Type,
	rcvrPipelineType component.Type,
	nexts []baseConsumer,
) (conn component.Component, err error) {
	set := connector.CreateSettings{ID: componentID, TelemetrySettings: tel, BuildInfo: info}
	set.TelemetrySettings.Logger = components.ConnectorLogger(set.TelemetrySettings.Logger, componentID, exprPipelineType, rcvrPipelineType)

	switch rcvrPipelineType {
	case component.DataTypeTraces:
		next := nexts[0].(consumer.Traces)
		if len(nexts) > 1 {
			consumers := make(map[component.ID]consumer.Traces, len(nexts))
			for _, next := range nexts {
				consumers[next.(*capabilitiesNode).pipelineID] = next.(consumer.Traces)
			}
			next = fanoutconsumer.NewTracesRouter(consumers)
		}
		switch exprPipelineType {
		case component.DataTypeTraces:
			conn, err = builder.CreateTracesToTraces(ctx, set, next)
		case component.DataTypeMetrics:
			conn, err = builder.CreateMetricsToTraces(ctx, set, next)
		case component.DataTypeLogs:
			conn, err = builder.CreateLogsToTraces(ctx, set, next)
		}
	case component.DataTypeMetrics:
		next := nexts[0].(consumer.Metrics)
		if len(nexts) > 1 {
			consumers := make(map[component.ID]consumer.Metrics, len(nexts))
			for _, next := range nexts {
				consumers[next.(*capabilitiesNode).pipelineID] = next.(consumer.Metrics)
			}
			next = fanoutconsumer.NewMetricsRouter(consumers)
		}
		switch exprPipelineType {
		case component.DataTypeTraces:
			conn, err = builder.CreateTracesToMetrics(ctx, set, next)
		case component.DataTypeMetrics:
			conn, err = builder.CreateMetricsToMetrics(ctx, set, next)
		case component.DataTypeLogs:
			conn, err = builder.CreateLogsToMetrics(ctx, set, next)
		}
	case component.DataTypeLogs:
		next := nexts[0].(consumer.Logs)
		if len(nexts) > 1 {
			consumers := make(map[component.ID]consumer.Logs, len(nexts))
			for _, next := range nexts {
				consumers[next.(*capabilitiesNode).pipelineID] = next.(consumer.Logs)
			}
			next = fanoutconsumer.NewLogsRouter(consumers)
		}
		switch exprPipelineType {
		case component.DataTypeTraces:
			conn, err = builder.CreateTracesToLogs(ctx, set, next)
		case component.DataTypeMetrics:
			conn, err = builder.CreateMetricsToLogs(ctx, set, next)
		case component.DataTypeLogs:
			conn, err = builder.CreateLogsToLogs(ctx, set, next)
		}
	}
	return
}

var _ consumerNode = &capabilitiesNode{}

// Every pipeline has a "virtual" capabilities node immediately after the receiver(s).
// There are two purposes for this node:
// 1. Present aggregated capabilities to receivers, such as whether the pipeline mutates data.
// 2. Present a consistent "first consumer" for each pipeline.
// The nodeID is derived from "pipeline ID".
type capabilitiesNode struct {
	nodeID
	pipelineID component.ID
	baseConsumer
	consumer.ConsumeTracesFunc
	consumer.ConsumeMetricsFunc
	consumer.ConsumeLogsFunc
}

func newCapabilitiesNode(pipelineID component.ID) *capabilitiesNode {
	return &capabilitiesNode{
		nodeID:     newNodeID(capabilitiesSeed, pipelineID.String()),
		pipelineID: pipelineID,
	}
}

func (n *capabilitiesNode) getConsumer() baseConsumer {
	return n
}

var _ consumerNode = &fanOutNode{}

// Each pipeline has one fan-out node before exporters.
// Therefore, nodeID is derived from "pipeline ID".
type fanOutNode struct {
	nodeID
	pipelineID component.ID
	baseConsumer
}

func newFanOutNode(pipelineID component.ID) *fanOutNode {
	return &fanOutNode{
		nodeID:     newNodeID(fanOutToExporters, pipelineID.String()),
		pipelineID: pipelineID,
	}
}

func (n *fanOutNode) getConsumer() baseConsumer {
	return n.baseConsumer
}
