package cortex

import (
	"flag"
	"fmt"
	"os"

	"github.com/go-kit/kit/log/level"
	"github.com/pkg/errors"
	"github.com/weaveworks/common/middleware"
	"github.com/weaveworks/common/server"
	"google.golang.org/grpc"
	"gopkg.in/yaml.v2"

	"github.com/cortexproject/cortex/pkg/alertmanager"
	"github.com/cortexproject/cortex/pkg/chunk"
	"github.com/cortexproject/cortex/pkg/chunk/cache"
	"github.com/cortexproject/cortex/pkg/chunk/encoding"
	"github.com/cortexproject/cortex/pkg/chunk/storage"
	chunk_util "github.com/cortexproject/cortex/pkg/chunk/util"
	"github.com/cortexproject/cortex/pkg/compactor"
	"github.com/cortexproject/cortex/pkg/configs/api"
	config_client "github.com/cortexproject/cortex/pkg/configs/client"
	"github.com/cortexproject/cortex/pkg/configs/db"
	"github.com/cortexproject/cortex/pkg/distributor"
	"github.com/cortexproject/cortex/pkg/ingester"
	"github.com/cortexproject/cortex/pkg/ingester/client"
	"github.com/cortexproject/cortex/pkg/querier"
	"github.com/cortexproject/cortex/pkg/querier/frontend"
	"github.com/cortexproject/cortex/pkg/querier/queryrange"
	"github.com/cortexproject/cortex/pkg/ring"
	"github.com/cortexproject/cortex/pkg/ruler"
	"github.com/cortexproject/cortex/pkg/storage/tsdb"
	"github.com/cortexproject/cortex/pkg/util"
	"github.com/cortexproject/cortex/pkg/util/runtimeconfig"
	"github.com/cortexproject/cortex/pkg/util/validation"
)

// The design pattern for Cortex is a series of config objects, which are
// registered for command line flags, and then a series of components that
// are instantiated and composed.  Some rules of thumb:
// - Config types should only contain 'simple' types (ints, strings, urls etc).
// - Flag validation should be done by the flag; use a flag.Value where
//   appropriate.
// - Config types should map 1:1 with a component type.
// - Config types should define flags with a common prefix.
// - It's fine to nest configs within configs, but this should match the
//   nesting of components within components.
// - Limit as much is possible sharing of configuration between config types.
//   Where necessary, use a pointer for this - avoid repetition.
// - Where a nesting of components its not obvious, it's fine to pass
//   references to other components constructors to compose them.
// - First argument for a components constructor should be its matching config
//   object.

// Config is the root config for Cortex.
type Config struct {
	Target      moduleName `yaml:"target,omitempty"`
	AuthEnabled bool       `yaml:"auth_enabled,omitempty"`
	PrintConfig bool       `yaml:"-"`
	HTTPPrefix  string     `yaml:"http_prefix"`

	Server         server.Config            `yaml:"server,omitempty"`
	Distributor    distributor.Config       `yaml:"distributor,omitempty"`
	Querier        querier.Config           `yaml:"querier,omitempty"`
	IngesterClient client.Config            `yaml:"ingester_client,omitempty"`
	Ingester       ingester.Config          `yaml:"ingester,omitempty"`
	Storage        storage.Config           `yaml:"storage,omitempty"`
	ChunkStore     chunk.StoreConfig        `yaml:"chunk_store,omitempty"`
	Schema         chunk.SchemaConfig       `yaml:"schema,omitempty" doc:"hidden"` // Doc generation tool doesn't support it because part of the SchemaConfig doesn't support CLI flags (needs manual documentation)
	LimitsConfig   validation.Limits        `yaml:"limits,omitempty"`
	Prealloc       client.PreallocConfig    `yaml:"prealloc,omitempty" doc:"hidden"`
	Worker         frontend.WorkerConfig    `yaml:"frontend_worker,omitempty"`
	Frontend       frontend.Config          `yaml:"frontend,omitempty"`
	QueryRange     queryrange.Config        `yaml:"query_range,omitempty"`
	TableManager   chunk.TableManagerConfig `yaml:"table_manager,omitempty"`
	Encoding       encoding.Config          `yaml:"-"` // No yaml for this, it only works with flags.
	TSDB           tsdb.Config              `yaml:"tsdb" doc:"hidden"`
	Compactor      compactor.Config         `yaml:"compactor,omitempty" doc:"hidden"`

	Ruler         ruler.Config                               `yaml:"ruler,omitempty"`
	ConfigDB      db.Config                                  `yaml:"configdb,omitempty"`
	ConfigStore   config_client.Config                       `yaml:"config_store,omitempty"`
	Alertmanager  alertmanager.MultitenantAlertmanagerConfig `yaml:"alertmanager,omitempty"`
	RuntimeConfig runtimeconfig.ManagerConfig                `yaml:"runtime_config,omitempty"`
}

// RegisterFlags registers flag.
func (c *Config) RegisterFlags(f *flag.FlagSet) {
	c.Server.MetricsNamespace = "cortex"
	c.Target = All
	c.Server.ExcludeRequestInLog = true
	f.Var(&c.Target, "target", "The Cortex service to run. Supported values are: all, distributor, ingester, querier, query-frontend, table-manager, ruler, alertmanager, configs.")
	f.BoolVar(&c.AuthEnabled, "auth.enabled", true, "Set to false to disable auth.")
	f.BoolVar(&c.PrintConfig, "print.config", false, "Print the config and exit.")
	f.StringVar(&c.HTTPPrefix, "http.prefix", "/api/prom", "HTTP path prefix for Cortex API.")

	c.Server.RegisterFlags(f)
	c.Distributor.RegisterFlags(f)
	c.Querier.RegisterFlags(f)
	c.IngesterClient.RegisterFlags(f)
	c.Ingester.RegisterFlags(f)
	c.Storage.RegisterFlags(f)
	c.ChunkStore.RegisterFlags(f)
	c.Schema.RegisterFlags(f)
	c.LimitsConfig.RegisterFlags(f)
	c.Prealloc.RegisterFlags(f)
	c.Worker.RegisterFlags(f)
	c.Frontend.RegisterFlags(f)
	c.QueryRange.RegisterFlags(f)
	c.TableManager.RegisterFlags(f)
	c.Encoding.RegisterFlags(f)
	c.TSDB.RegisterFlags(f)
	c.Compactor.RegisterFlags(f)

	c.Ruler.RegisterFlags(f)
	c.ConfigDB.RegisterFlags(f)
	c.ConfigStore.RegisterFlagsWithPrefix("alertmanager.", f)
	c.Alertmanager.RegisterFlags(f)
	c.RuntimeConfig.RegisterFlags(f)

	// These don't seem to have a home.
	flag.IntVar(&chunk_util.QueryParallelism, "querier.query-parallelism", 100, "Max subqueries run in parallel per higher-level query.")
}

// Validate the cortex config and returns an error if the validation
// doesn't pass
func (c *Config) Validate() error {
	if err := c.Schema.Validate(); err != nil {
		return errors.Wrap(err, "invalid schema config")
	}
	if err := c.Encoding.Validate(); err != nil {
		return errors.Wrap(err, "invalid encoding config")
	}
	if err := c.Storage.Validate(); err != nil {
		return errors.Wrap(err, "invalid storage config")
	}
	if err := c.TSDB.Validate(); err != nil {
		return errors.Wrap(err, "invalid TSDB config")
	}
	if err := c.LimitsConfig.Validate(c.Distributor.ShardByAllLabels); err != nil {
		return errors.Wrap(err, "invalid limits config")
	}
	if err := c.Distributor.Validate(); err != nil {
		return errors.Wrap(err, "invalid distributor config")
	}
	return nil
}

// Cortex is the root datastructure for Cortex.
type Cortex struct {
	target             moduleName
	httpAuthMiddleware middleware.Interface

	server        *server.Server
	ring          *ring.Ring
	overrides     *validation.Overrides
	distributor   *distributor.Distributor
	ingester      *ingester.Ingester
	store         chunk.Store
	worker        frontend.Worker
	frontend      *frontend.Frontend
	tableManager  *chunk.TableManager
	cache         cache.Cache
	runtimeConfig *runtimeconfig.Manager

	ruler        *ruler.Ruler
	configAPI    *api.API
	configDB     db.DB
	alertmanager *alertmanager.MultitenantAlertmanager
	compactor    *compactor.Compactor
}

// New makes a new Cortex.
func New(cfg Config) (*Cortex, error) {
	if cfg.PrintConfig {
		if err := yaml.NewEncoder(os.Stdout).Encode(&cfg); err != nil {
			fmt.Println("Error encoding config:", err)
		}
		os.Exit(0)
	}

	cortex := &Cortex{
		target: cfg.Target,
	}

	cortex.setupAuthMiddleware(&cfg)

	if err := cortex.init(&cfg, cfg.Target); err != nil {
		return nil, err
	}

	return cortex, nil
}

func (t *Cortex) setupAuthMiddleware(cfg *Config) {
	if cfg.AuthEnabled {
		cfg.Server.GRPCMiddleware = []grpc.UnaryServerInterceptor{
			middleware.ServerUserHeaderInterceptor,
		}
		cfg.Server.GRPCStreamMiddleware = []grpc.StreamServerInterceptor{
			func(srv interface{}, ss grpc.ServerStream, info *grpc.StreamServerInfo, handler grpc.StreamHandler) error {
				switch info.FullMethod {
				// Don't check auth header on TransferChunks, as we weren't originally
				// sending it and this could cause transfers to fail on update.
				//
				// Also don't check auth /frontend.Frontend/Process, as this handles
				// queries for multiple users.
				case "/cortex.Ingester/TransferChunks", "/frontend.Frontend/Process":
					return handler(srv, ss)
				default:
					return middleware.StreamServerUserHeaderInterceptor(srv, ss, info, handler)
				}
			},
		}
		t.httpAuthMiddleware = middleware.AuthenticateUser
	} else {
		cfg.Server.GRPCMiddleware = []grpc.UnaryServerInterceptor{
			fakeGRPCAuthUniaryMiddleware,
		}
		cfg.Server.GRPCStreamMiddleware = []grpc.StreamServerInterceptor{
			fakeGRPCAuthStreamMiddleware,
		}
		t.httpAuthMiddleware = fakeHTTPAuthMiddleware
	}
}

func (t *Cortex) init(cfg *Config, m moduleName) error {
	// initialize all of our dependencies first
	for _, dep := range orderedDeps(m) {
		if err := t.initModule(cfg, dep); err != nil {
			return err
		}
	}
	// lastly, initialize the requested module
	return t.initModule(cfg, m)
}

func (t *Cortex) initModule(cfg *Config, m moduleName) error {
	level.Info(util.Logger).Log("msg", "initialising", "module", m)
	if modules[m].init != nil {
		if err := modules[m].init(t, cfg); err != nil {
			return errors.Wrap(err, fmt.Sprintf("error initialising module: %s", m))
		}
	}
	return nil
}

// Run starts Cortex running, and blocks until a signal is received.
func (t *Cortex) Run() error {
	return t.server.Run()
}

// Stop gracefully stops a Cortex.
func (t *Cortex) Stop() error {
	t.stopModule(t.target)
	deps := orderedDeps(t.target)
	// iterate over our deps in reverse order and call stopModule
	for i := len(deps) - 1; i >= 0; i-- {
		t.stopModule(deps[i])
	}
	return nil
}

func (t *Cortex) stopModule(m moduleName) {
	level.Info(util.Logger).Log("msg", "stopping", "module", m)
	if modules[m].stop != nil {
		if err := modules[m].stop(t); err != nil {
			level.Error(util.Logger).Log("msg", "error stopping", "module", m, "err", err)
		}
	}
}

// listDeps recursively gets a list of dependencies for a passed moduleName
func listDeps(m moduleName) []moduleName {
	deps := modules[m].deps
	for _, d := range modules[m].deps {
		deps = append(deps, listDeps(d)...)
	}
	return deps
}

// orderedDeps gets a list of all dependencies ordered so that items are always after any of their dependencies.
func orderedDeps(m moduleName) []moduleName {
	deps := listDeps(m)

	// get a unique list of moduleNames, with a flag for whether they have been added to our result
	uniq := map[moduleName]bool{}
	for _, dep := range deps {
		uniq[dep] = false
	}

	result := make([]moduleName, 0, len(uniq))

	// keep looping through all modules until they have all been added to the result.

	for len(result) < len(uniq) {
	OUTER:
		for name, added := range uniq {
			if added {
				continue
			}
			for _, dep := range modules[name].deps {
				// stop processing this module if one of its dependencies has
				// not been added to the result yet.
				if !uniq[dep] {
					continue OUTER
				}
			}

			// if all of the module's dependencies have been added to the result slice,
			// then we can safely add this module to the result slice as well.
			uniq[name] = true
			result = append(result, name)
		}
	}
	return result
}
