// Licensed to Apache Software Foundation (ASF) under one or more contributor
// license agreements. See the NOTICE file distributed with
// this work for additional information regarding copyright
// ownership. Apache Software Foundation (ASF) licenses this file to you 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 cmdsetup

import (
	"context"

	grpcprom "github.com/grpc-ecosystem/go-grpc-middleware/providers/prometheus"
	"github.com/spf13/cobra"

	"github.com/apache/skywalking-banyandb/api/common"
	"github.com/apache/skywalking-banyandb/banyand/liaison/grpc"
	"github.com/apache/skywalking-banyandb/banyand/liaison/http"
	"github.com/apache/skywalking-banyandb/banyand/measure"
	"github.com/apache/skywalking-banyandb/banyand/metadata/service"
	"github.com/apache/skywalking-banyandb/banyand/observability"
	"github.com/apache/skywalking-banyandb/banyand/observability/services"
	"github.com/apache/skywalking-banyandb/banyand/property"
	"github.com/apache/skywalking-banyandb/banyand/protector"
	"github.com/apache/skywalking-banyandb/banyand/query"
	"github.com/apache/skywalking-banyandb/banyand/queue"
	"github.com/apache/skywalking-banyandb/banyand/stream"
	"github.com/apache/skywalking-banyandb/banyand/trace"
	"github.com/apache/skywalking-banyandb/pkg/logger"
	"github.com/apache/skywalking-banyandb/pkg/run"
	"github.com/apache/skywalking-banyandb/pkg/version"
)

func newStandaloneCmd(runners ...run.Unit) *cobra.Command {
	l := logger.GetLogger("bootstrap")
	ctx := context.Background()
	dataPipeline := queue.Local()
	metaSvc, err := service.NewService()
	if err != nil {
		l.Fatal().Err(err).Msg("failed to initiate metadata service")
	}
	metaSvc.SetSnapshotPipeline(dataPipeline)
	metricSvc := services.NewMetricService(metaSvc, dataPipeline, "standalone", nil)
	metaSvc.SetMetricsRegistry(metricSvc)
	pm := protector.NewMemory(metricSvc)
	propertySvc, err := property.NewService(metaSvc, dataPipeline, nil, metricSvc, pm)
	if err != nil {
		l.Fatal().Err(err).Msg("failed to initiate property service")
	}
	streamSvc, err := stream.NewService(metaSvc, dataPipeline, metricSvc, pm, nil)
	if err != nil {
		l.Fatal().Err(err).Msg("failed to initiate stream service")
	}
	traceSvc, err := trace.NewService(metaSvc, dataPipeline, metricSvc, pm)
	if err != nil {
		l.Fatal().Err(err).Msg("failed to initiate trace service")
	}
	var srvMetrics *grpcprom.ServerMetrics
	srvMetrics.UnaryServerInterceptor()
	srvMetrics.UnaryServerInterceptor()
	measureSvc, err := measure.NewStandalone(metaSvc, dataPipeline, nil, metricSvc, pm)
	if err != nil {
		l.Fatal().Err(err).Msg("failed to initiate measure service")
	}
	q, err := query.NewService(ctx, streamSvc, measureSvc, traceSvc, metaSvc, dataPipeline, metricSvc, false)
	if err != nil {
		l.Fatal().Err(err).Msg("failed to initiate query processor")
	}
	nr := grpc.NewLocalNodeRegistry()
	grpcServer := grpc.NewServer(ctx, dataPipeline, dataPipeline, dataPipeline, metaSvc, grpc.NodeRegistries{
		MeasureLiaisonNodeRegistry: nr,
		StreamLiaisonNodeRegistry:  nr,
		PropertyNodeRegistry:       nr,
		TraceLiaisonNodeRegistry:   nr,
	}, metricSvc, pm, nil)
	profSvc := observability.NewProfService()
	httpServer := http.NewServer(grpcServer.GetAuthReloader())

	var units []run.Unit
	units = append(units, runners...)
	units = append(units,
		dataPipeline,
		metricSvc,
		metaSvc,
		pm,
		propertySvc,
		measureSvc,
		streamSvc,
		traceSvc,
		q,
		grpcServer,
		httpServer,
		profSvc,
	)
	standaloneGroup := run.NewGroup("standalone")
	// Meta the run Group units.
	standaloneGroup.Register(units...)

	standaloneCmd := &cobra.Command{
		Use:     "standalone",
		Version: version.Build(),
		Short:   "Run as the standalone server",
		RunE: func(_ *cobra.Command, _ []string) (err error) {
			// Cobra short-circuits the post-run hooks on a non-nil RunE
			// return (command.go: `if err := c.RunE(...); err != nil { return err }`),
			// so PersistentPostRunE in root.go does not fire on the
			// error path. Drain the artifact sink here so queued panic
			// records make it to disk before main exits with non-zero.
			// ShutdownSupervisor is idempotent; the success-path call
			// in PersistentPostRunE is harmless.
			defer ShutdownSupervisor()
			nodeID, err := common.GenerateNode(grpcServer.GetPort(), httpServer.GetPort(), nil,
				metaSvc.GetSchemaServerPort(), metaSvc.GetSchemaGossipPort())
			if err != nil {
				return err
			}
			logger.GetLogger().Info().Msg("starting as a standalone server")
			// Spawn our go routines and wait for shutdown. The Run context is
			// derived from SupervisorContext so that a panic recovered anywhere
			// in the process, including goroutines spawned via run.Go, fires
			// cancellation here.
			if err := standaloneGroup.Run(context.WithValue(SupervisorContext(), common.ContextNodeKey, nodeID)); err != nil {
				logger.GetLogger().Error().Err(err).Stack().Str("name", standaloneGroup.Name()).Msg("Exit")
				return err
			}
			return nil
		},
	}
	standaloneCmd.Flags().AddFlagSet(standaloneGroup.RegisterFlags().FlagSet)
	return standaloneCmd
}
