/*
 * Copyright 2018-present Open Networking Foundation

 * 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 core

import (
	"context"
	"time"

	"github.com/opencord/voltha-go/db/model"
	"github.com/opencord/voltha-go/rw_core/config"
	"github.com/opencord/voltha-go/rw_core/core/adapter"
	"github.com/opencord/voltha-go/rw_core/core/api"
	"github.com/opencord/voltha-go/rw_core/core/device"
	conf "github.com/opencord/voltha-lib-go/v4/pkg/config"
	grpcserver "github.com/opencord/voltha-lib-go/v4/pkg/grpc"
	"github.com/opencord/voltha-lib-go/v4/pkg/kafka"
	"github.com/opencord/voltha-lib-go/v4/pkg/log"
	"github.com/opencord/voltha-lib-go/v4/pkg/probe"
	"github.com/opencord/voltha-protos/v4/go/extension"
	"github.com/opencord/voltha-protos/v4/go/voltha"
	"google.golang.org/grpc"
)

// Core represent read,write core attributes
type Core struct {
	shutdown context.CancelFunc
	stopped  chan struct{}
}

const (
	adapterMessageBus = "adapter-message-bus"
	clusterMessageBus = "cluster-message-bus"
)

// NewCore creates instance of rw core
func NewCore(ctx context.Context, id string, cf *config.RWCoreFlags) *Core {
	// If the context has a probe then fetch it and register our services
	if p := probe.GetProbeFromContext(ctx); p != nil {
		p.RegisterService(
			ctx,
			adapterMessageBus,
			"kv-store",
			"adapter-manager",
			"grpc-service",
			"adapter-request-handler",
		)

		if cf.KafkaAdapterAddress != cf.KafkaClusterAddress {
			p.RegisterService(
				ctx,
				clusterMessageBus,
			)
		}
	}

	// new threads will be given a new cancelable context, so that they can be aborted later when Stop() is called
	shutdownCtx, cancelCtx := context.WithCancel(ctx)

	core := &Core{shutdown: cancelCtx, stopped: make(chan struct{})}
	go core.start(shutdownCtx, id, cf)
	return core
}

func (core *Core) start(ctx context.Context, id string, cf *config.RWCoreFlags) {
	logger.Info(ctx, "starting-core-services", log.Fields{"coreId": id})

	// deferred functions are used to run cleanup
	// failing partway will stop anything that's been started
	defer close(core.stopped)
	defer core.shutdown()

	logger.Info(ctx, "starting-rw-core-components")

	// setup kv client
	logger.Debugw(ctx, "create-kv-client", log.Fields{"kvstore": cf.KVStoreType})
	kvClient, err := newKVClient(ctx, cf.KVStoreType, cf.KVStoreAddress, cf.KVStoreTimeout)
	if err != nil {
		logger.Fatal(ctx, err)
	}
	defer stopKVClient(log.WithSpanFromContext(context.Background(), ctx), kvClient)

	// sync logging config with kv store
	cm := conf.NewConfigManager(ctx, kvClient, cf.KVStoreType, cf.KVStoreAddress, cf.KVStoreTimeout)
	go conf.StartLogLevelConfigProcessing(cm, ctx)
	go conf.StartLogFeaturesConfigProcessing(cm, ctx)

	backend := cm.Backend
	backend.LivenessChannelInterval = cf.LiveProbeInterval / 2

	// wait until connection to KV Store is up
	if err := waitUntilKVStoreReachableOrMaxTries(ctx, kvClient, cf.MaxConnectionRetries, cf.ConnectionRetryInterval); err != nil {
		logger.Fatal(ctx, "unable-to-connect-to-kv-store")
	}
	go monitorKVStoreLiveness(ctx, backend, cf.LiveProbeInterval, cf.NotLiveProbeInterval)

	// create kafka client
	kafkaClient := kafka.NewSaramaClient(
		kafka.Address(cf.KafkaAdapterAddress),
		kafka.ConsumerType(kafka.GroupCustomer),
		kafka.ProducerReturnOnErrors(true),
		kafka.ProducerReturnOnSuccess(true),
		kafka.ProducerMaxRetries(6),
		kafka.NumPartitions(3),
		kafka.ConsumerGroupName(id),
		kafka.ConsumerGroupPrefix(id),
		kafka.AutoCreateTopic(true),
		kafka.ProducerFlushFrequency(5),
		kafka.ProducerRetryBackoff(time.Millisecond*30),
		kafka.LivenessChannelInterval(cf.LiveProbeInterval/2),
	)

	// create kafka client for events
	kafkaClientEvent := kafka.NewSaramaClient(
		kafka.Address(cf.KafkaClusterAddress),
		kafka.ProducerReturnOnErrors(true),
		kafka.ProducerReturnOnSuccess(true),
		kafka.ProducerMaxRetries(6),
		kafka.ProducerRetryBackoff(time.Millisecond*30),
		kafka.AutoCreateTopic(true),
		kafka.MetadatMaxRetries(15),
	)

	// create event proxy
	updateProbeClusterService := cf.KafkaAdapterAddress != cf.KafkaClusterAddress
	eventProxy, err := startEventProxy(ctx, kafkaClientEvent, cf.EventTopic, cf.ConnectionRetryInterval, updateProbeClusterService)
	if err != nil {
		logger.Warn(ctx, "failed-to-setup-kafka-event-proxy-connection")
		return
	}
	if cf.KafkaAdapterAddress != cf.KafkaClusterAddress {
		// if we're using a single kafka cluster we don't need two liveliness probes on the same cluster
		go monitorKafkaLiveness(ctx, eventProxy, cf.LiveProbeInterval, cf.NotLiveProbeInterval, clusterMessageBus)
	}

	defer stopEventProxy(ctx, kafkaClientEvent, eventProxy)

	// create kv path
	dbPath := model.NewDBPath(backend)

	// load adapters & device types while other things are starting
	adapterMgr := adapter.NewAdapterManager(ctx, dbPath, id, kafkaClient)
	go adapterMgr.Start(ctx)

	// connect to kafka, then wait until reachable and publisher/consumer created
	// core.kmp must be created before deviceMgr and adapterMgr
	kmp, err := startKafkInterContainerProxy(ctx, kafkaClient, cf.KafkaAdapterAddress, cf.CoreTopic, cf.ConnectionRetryInterval)
	if err != nil {
		logger.Warn(ctx, "failed-to-setup-kafka-adapter-proxy-connection")
		return
	}
	defer kmp.Stop(ctx)
	go monitorKafkaLiveness(ctx, kmp, cf.LiveProbeInterval, cf.NotLiveProbeInterval, adapterMessageBus)

	// create the core of the system, the device managers
	endpointMgr := kafka.NewEndpointManager(backend)
	deviceMgr, logicalDeviceMgr := device.NewManagers(dbPath, adapterMgr, kmp, endpointMgr, cf.CoreTopic, id, cf.DefaultCoreTimeout, eventProxy, cf.VolthaStackID)

	// register kafka RPC handler
	registerAdapterRequestHandlers(ctx, kmp, deviceMgr, adapterMgr, cf, "adapter-request-handler")

	// start gRPC handler
	grpcServer := grpcserver.NewGrpcServer(cf.GrpcAddress, nil, false, probe.GetProbeFromContext(ctx))

	//Register the 'Extension' service on this gRPC server
	addGRPCExtensionService(ctx, grpcServer, device.GetNewExtensionManager(deviceMgr))

	go startGRPCService(ctx, grpcServer, api.NewNBIHandler(deviceMgr, logicalDeviceMgr, adapterMgr))
	defer grpcServer.Stop()

	// wait for core to be stopped, via Stop() or context cancellation, before running deferred functions
	<-ctx.Done()
}

// Stop brings down core services
func (core *Core) Stop() {
	core.shutdown()
	<-core.stopped
}

// startGRPCService creates the grpc service handlers, registers it to the grpc server and starts the server
func startGRPCService(ctx context.Context, server *grpcserver.GrpcServer, handler voltha.VolthaServiceServer) {
	logger.Info(ctx, "grpc-server-created")

	server.AddService(func(gs *grpc.Server) { voltha.RegisterVolthaServiceServer(gs, handler) })
	logger.Info(ctx, "grpc-service-added")

	probe.UpdateStatusFromContext(ctx, "grpc-service", probe.ServiceStatusRunning)
	logger.Info(ctx, "grpc-server-started")
	// Note that there is a small window here in which the core could return its status as ready,
	// when it really isn't.  This is unlikely to cause issues, as the delay is incredibly short.
	server.Start(ctx)
	probe.UpdateStatusFromContext(ctx, "grpc-service", probe.ServiceStatusStopped)
}

func addGRPCExtensionService(ctx context.Context, server *grpcserver.GrpcServer, handler extension.ExtensionServer) {
	logger.Info(ctx, "extension-grpc-server-created")

	server.AddService(func(server *grpc.Server) {
		extension.RegisterExtensionServer(server, handler)
	})

}