// periodically scan chunks and close any that have not received data in a while
// TODO instrument occurences and duration of GC
func (ms *AggMetrics) GC() {
	ticker := time.Tick(time.Duration(*gcInterval) * time.Second)
	for now := range ticker {
		log.Info("checking for stale chunks that need persisting.")
		now := uint32(now.Unix())
		chunkMinTs := now - (now % ms.chunkSpan) - uint32(ms.chunkMaxStale)
		metricMinTs := now - (now % ms.chunkSpan) - uint32(ms.metricMaxStale)

		// as this is the only goroutine that can delete from ms.Metrics
		// we only need to lock long enough to get the list of actives metrics.
		// it doesnt matter if new metrics are added while we iterate this list.
		ms.RLock()
		keys := make([]string, 0, len(ms.Metrics))
		for k := range ms.Metrics {
			keys = append(keys, k)
		}
		ms.RUnlock()
		for _, key := range keys {
			ms.RLock()
			a := ms.Metrics[key]
			ms.RUnlock()
			if stale := a.GC(chunkMinTs, metricMinTs); stale {
				log.Info("metric %s is stale. Purging data from memory.", key)
				delete(ms.Metrics, key)
			}
		}

	}
}

func InitApiPluginRoutes(r *macaron.Macaron) {
	for _, plugin := range plugins.ApiPlugins {
		log.Info("Plugin: Adding proxy routes for api plugin")
		for _, route := range plugin.Routes {
			url := util.JoinUrlFragments("/api/plugin-proxy/", route.Path)
			handlers := make([]macaron.Handler, 0)
			if route.ReqSignedIn {
				handlers = append(handlers, middleware.Auth(&middleware.AuthOptions{ReqSignedIn: true}))
			}
			if route.ReqGrafanaAdmin {
				handlers = append(handlers, middleware.Auth(&middleware.AuthOptions{ReqSignedIn: true, ReqGrafanaAdmin: true}))
			}
			if route.ReqSignedIn && route.ReqRole != "" {
				if route.ReqRole == m.ROLE_ADMIN {
					handlers = append(handlers, middleware.RoleAuth(m.ROLE_ADMIN))
				} else if route.ReqRole == m.ROLE_EDITOR {
					handlers = append(handlers, middleware.RoleAuth(m.ROLE_EDITOR, m.ROLE_ADMIN))
				}
			}
			handlers = append(handlers, ApiPlugin(route.Url))
			r.Route(url, route.Method, handlers...)
			log.Info("Plugin: Adding route %s", url)
		}
	}
}

func upgrade() {
	UpgradeFileName := getUpgradeFileName()
	VersionFileName := getVersionFileName()
	UpgradeMarkerFileName := getUpgradeMarkerFileName()

	if setting.PathExists(UpgradeFileName) {
		if loadUpgradeFile(UpgradeFileName) {
			if !setting.PathExists(VersionFileName) {
				netCrunchSettings, err := readNetCrunchServerSettings()
				if err == nil {
					if updateNetCrunchDatasources(netCrunchSettings) &&
						writeVersionFile(VersionFileName) && removeFile(UpgradeFileName) {
						log.Info("NetCrunch: Upgrade")

						if setting.PathExists(UpgradeMarkerFileName) {
							SetInitializationSuccess()
							removeFile(UpgradeMarkerFileName)
						}
					} else {
						log.Info("NetCrunch: Upgrade error")
					}
				} else {
					log.Info("NetCrunch: Upgrade error")
				}
			}
		} else {
			log.Info("NetCrunch: Upgrade error")
		}
	}
}

func (mg *Migrator) exec(m Migration) error {
	if mg.LogLevel <= log.INFO {
		log.Info("Migrator: exec migration id: %v", m.Id())
	}

	err := mg.inTransaction(func(sess *xorm.Session) error {

		condition := m.GetCondition()
		if condition != nil {
			sql, args := condition.Sql(mg.dialect)
			results, err := sess.Query(sql, args...)
			if err != nil || len(results) == 0 {
				log.Info("Migrator: skipping migration id: %v, condition not fulfilled", m.Id())
				return sess.Rollback()
			}
		}

		_, err := sess.Exec(m.Sql(mg.dialect))
		if err != nil {
			log.Error(3, "Migrator: exec FAILED migration id: %v, err: %v", m.Id(), err)
			return err
		}
		return nil
	})

	if err != nil {
		return err
	}

	return nil
}

func (h *hub) run() {
	for {
		select {
		case c := <-h.register:
			h.connections[c] = true
			log.Info("Live: New connection (Total count: %v)", len(h.connections))

		case c := <-h.unregister:
			if _, ok := h.connections[c]; ok {
				log.Info("Live: Closing Connection (Total count: %v)", len(h.connections))
				delete(h.connections, c)
				close(c.send)
			}
		// hand stream subscriptions
		case sub := <-h.subChannel:
			log.Info("Live: Subscribing to: %v, remove: %v", sub.name, sub.remove)
			subscribers, exists := h.streams[sub.name]

			// handle unsubscribe
			if exists && sub.remove {
				delete(subscribers, sub.conn)
				continue
			}

			if !exists {
				subscribers = make(map[*connection]bool)
				h.streams[sub.name] = subscribers
			}
			subscribers[sub.conn] = true

			// handle stream messages
		case message := <-h.streamChannel:
			subscribers, exists := h.streams[message.Stream]
			if !exists || len(subscribers) == 0 {
				log.Info("Live: Message to stream without subscribers: %v", message.Stream)
				continue
			}

			messageBytes, _ := simplejson.NewFromAny(message).Encode()
			for sub := range subscribers {
				// check if channel is open
				if _, ok := h.connections[sub]; !ok {
					delete(subscribers, sub)
					continue
				}

				select {
				case sub.send <- messageBytes:
				default:
					close(sub.send)
					delete(h.connections, sub)
					delete(subscribers, sub)
				}
			}
		}
	}
}

func initRuntime() {
	setting.NewConfigContext(&setting.CommandLineArgs{
		Config:   *configFile,
		HomePath: *homePath,
		Args:     flag.Args(),
	})

	log.Info("Starting Grafana")
	log.Info("Version: %v, Commit: %v, Build date: %v", setting.BuildVersion, setting.BuildCommit, time.Unix(setting.BuildStamp, 0))
	setting.LogConfigurationInfo()

	sqlstore.NewEngine()
	sqlstore.EnsureAdminUser()
}

func (c *CollectorContext) Save() error {
	cmd := &m.AddCollectorSessionCommand{
		CollectorId: c.Collector.Id,
		SocketId:    c.Socket.Id(),
		OrgId:       c.OrgId,
		InstanceId:  instanceId,
	}
	if err := bus.Dispatch(cmd); err != nil {
		log.Info("could not write collector_sesison to DB.", err)
		return err
	}
	log.Info("collector_session %s for collector_id: %d saved to DB.", cmd.SocketId, cmd.CollectorId)
	return nil
}

func NewAggMetrics(chunkSpan, numChunks, chunkMaxStale, metricMaxStale uint32, aggSettings []aggSetting) *AggMetrics {
	ms := AggMetrics{
		Metrics:        make(map[string]*AggMetric),
		chunkSpan:      chunkSpan,
		numChunks:      numChunks,
		aggSettings:    aggSettings,
		chunkMaxStale:  chunkMaxStale,
		metricMaxStale: metricMaxStale,
	}
	// open data file
	dataFile, err := os.Open(*dumpFile)

	if err == nil {
		log.Info("loading aggMetrics from file " + *dumpFile)
		dataDecoder := gob.NewDecoder(dataFile)
		err = dataDecoder.Decode(&ms)
		if err != nil {
			log.Error(3, "failed to load aggMetrics from file. %s", err)
		}
		dataFile.Close()
		log.Info("aggMetrics loaded from file.")
		if ms.numChunks != numChunks {
			if ms.numChunks > numChunks {
				log.Fatal(3, "numChunks can not be decreased.")
			}
			log.Info("numChunks has changed. Updating memory structures.")
			sem := make(chan bool, *concurrency)
			for _, m := range ms.Metrics {
				sem <- true
				go func() {
					m.GrowNumChunks(numChunks)
					<-sem
				}()
			}
			for i := 0; i < cap(sem); i++ {
				sem <- true
			}

			ms.numChunks = numChunks
			log.Info("memory structures updated.")
		}
	} else {
		log.Info("starting with fresh aggmetrics.")
	}

	go ms.stats()
	go ms.GC()
	return &ms
}

func (mg *Migrator) exec(m Migration) error {
	log.Info("Executing migration", "id", m.Id())

	err := mg.inTransaction(func(sess *xorm.Session) error {

		condition := m.GetCondition()
		if condition != nil {
			sql, args := condition.Sql(mg.dialect)
			results, err := sess.Query(sql, args...)
			if err != nil || len(results) == 0 {
				mg.Logger.Info("Skipping migration condition not fulfilled", "id", m.Id())
				return sess.Rollback()
			}
		}

		_, err := sess.Exec(m.Sql(mg.dialect))
		if err != nil {
			mg.Logger.Error("Executing migration failed", "id", m.Id(), "error", err)
			return err
		}
		return nil
	})

	if err != nil {
		return err
	}

	return nil
}

func (c *CollectorContext) OnDisconnection() {
	log.Info(fmt.Sprintf("%s disconnected", c.Collector.Name))
	if err := c.Remove(); err != nil {
		log.Error(4, fmt.Sprintf("Failed to remove collectorSession. %s", c.Collector.Name), err)
	}
	contextCache.Remove(c.SocketId)
}

func loadLdapConfig() {
	if !setting.LdapEnabled {
		return
	}

	log.Info("Login: Ldap enabled, reading config file: %s", setting.LdapConfigFile)

	_, err := toml.DecodeFile(setting.LdapConfigFile, &ldapCfg)
	if err != nil {
		log.Fatal(3, "Failed to load ldap config file: %s", err)
	}

	if len(ldapCfg.Servers) == 0 {
		log.Fatal(3, "ldap enabled but no ldap servers defined in config file: %s", setting.LdapConfigFile)
	}

	// set default org id
	for _, server := range ldapCfg.Servers {
		assertNotEmptyCfg(server.SearchFilter, "search_filter")
		assertNotEmptyCfg(server.SearchBaseDNs, "search_base_dns")

		for _, groupMap := range server.LdapGroups {
			if groupMap.OrgId == 0 {
				groupMap.OrgId = 1
			}
		}
	}
}

func (pb *PluginBase) registerPlugin(pluginDir string) error {
	if _, exists := Plugins[pb.Id]; exists {
		return errors.New("Plugin with same id already exists")
	}

	if !strings.HasPrefix(pluginDir, setting.StaticRootPath) {
		log.Info("Plugins: Registering plugin %v", pb.Name)
	}

	if len(pb.Dependencies.Plugins) == 0 {
		pb.Dependencies.Plugins = []PluginDependencyItem{}
	}

	if pb.Dependencies.GrafanaVersion == "" {
		pb.Dependencies.GrafanaVersion = "*"
	}

	for _, include := range pb.Includes {
		if include.Role == "" {
			include.Role = m.RoleType(m.ROLE_VIEWER)
		}
	}

	pb.PluginDir = pluginDir
	Plugins[pb.Id] = pb
	return nil
}

// Logger returns a middleware handler that logs the request as it goes in and the response as it goes out.
func Logger() macaron.Handler {
	return func(res http.ResponseWriter, req *http.Request, c *macaron.Context) {
		start := time.Now()

		rw := res.(macaron.ResponseWriter)
		c.Next()

		content := fmt.Sprintf("Completed %s %v %s in %v", req.URL.Path, rw.Status(), http.StatusText(rw.Status()), time.Since(start))
		if !isWindows {
			switch rw.Status() {
			case 200:
				content = fmt.Sprintf("\033[1;32m%s\033[0m", content)
				return
			case 304:
				//content = fmt.Sprintf("\033[1;33m%s\033[0m", content)
				return
			case 404:
				content = fmt.Sprintf("\033[1;31m%s\033[0m", content)
			case 500:
				content = fmt.Sprintf("\033[1;36m%s\033[0m", content)
			}
		}
		log.Info(content)
	}
}

func newMacaron() *macaron.Macaron {
	macaron.Env = setting.Env
	m := macaron.New()

	m.Use(middleware.Logger())
	m.Use(macaron.Recovery())

	if setting.EnableGzip {
		m.Use(middleware.Gziper())
	}

	for _, route := range plugins.StaticRoutes {
		pluginRoute := path.Join("/public/plugins/", route.PluginId)
		log.Info("Plugins: Adding route %s -> %s", pluginRoute, route.Directory)
		mapStatic(m, route.Directory, "", pluginRoute)
	}

	mapStatic(m, setting.StaticRootPath, "", "public")
	mapStatic(m, setting.StaticRootPath, "robots.txt", "robots.txt")

	m.Use(macaron.Renderer(macaron.RenderOptions{
		Directory:  path.Join(setting.StaticRootPath, "views"),
		IndentJSON: macaron.Env != macaron.PROD,
		Delims:     macaron.Delims{Left: "[[", Right: "]]"},
	}))

	if setting.EnforceDomain {
		m.Use(middleware.ValidateHostHeader(setting.Domain))
	}

	m.Use(middleware.GetContextHandler())
	m.Use(middleware.Sessioner(&setting.SessionOptions))

	return m
}

func Publish(event *schema.ProbeEvent) error {
	if !enabled {
		return nil
	}
	version := uint8(msgFormatJson)

	buf := new(bytes.Buffer)
	err := binary.Write(buf, binary.LittleEndian, version)
	if err != nil {
		log.Fatal(0, "binary.Write failed: %s", err.Error())
	}
	id := time.Now().UnixNano()
	binary.Write(buf, binary.BigEndian, id)
	if err != nil {
		log.Fatal(0, "binary.Write failed: %s", err.Error())
	}
	msg, err := json.Marshal(event)
	if err != nil {
		return fmt.Errorf("Failed to marshal event payload: %s", err)
	}
	_, err = buf.Write(msg)
	if err != nil {
		log.Fatal(0, "buf.Write failed: %s", err.Error())
	}
	collectorEventPublisherMsgs.Inc(1)
	err = globalProducer.Publish(topic, buf.Bytes())
	if err != nil {
		panic(fmt.Errorf("can't publish to nsqd: %s", err))
	}
	log.Info("event published to NSQ %d", id)

	//globalProducer.Stop()
	return nil
}

// error is what is used to determine to ACK or NACK
func (kg *KairosGateway) process(job Job) error {
	msg := job.msg
	messagesSize.Value(int64(len(job.Msg.Msg)))
	log.Debug("processing metrics %s %d. timestamp: %s. format: %s. attempts: %d\n", job.qualifier, job.Msg.Id, time.Unix(0, msg.Timestamp), job.Msg.Format, msg.Attempts)

	err := job.Msg.DecodeMetricData()
	if err != nil {
		log.Info("%s: skipping message", err.Error())
		return nil
	}

	metricsPerMessage.Value(int64(len(job.Msg.Metrics)))
	if !kg.dryRun {
		pre := time.Now()
		err = kg.kairos.SendMetricPointers(job.Msg.Metrics)
		if err != nil {
			metricsToKairosFail.Inc(int64(len(job.Msg.Metrics)))
			log.Warn("can't send to kairosdb: %s. retrying later", err)
		} else {
			metricsToKairosOK.Inc(int64(len(job.Msg.Metrics)))
			kairosPutDuration.Value(time.Now().Sub(pre))
		}
	}
	log.Debug("finished metrics %s %d - %d metrics sent\n", job.qualifier, job.Msg.Id, len(job.Msg.Metrics))
	return err
}

func EnsureAdminUser() {
	statsQuery := m.GetSystemStatsQuery{}

	if err := bus.Dispatch(&statsQuery); err != nil {
		log.Fatal(3, "Could not determine if admin user exists: %v", err)
		return
	}

	if statsQuery.Result.UserCount > 0 {
		return
	}

	cmd := m.CreateUserCommand{}
	cmd.Login = setting.AdminUser
	cmd.Email = setting.AdminUser + "@localhost"
	cmd.Password = setting.AdminPassword
	cmd.IsAdmin = true

	if err := bus.Dispatch(&cmd); err != nil {
		log.Error(3, "Failed to create default admin user", err)
		return
	}

	log.Info("Created default admin user: %v", setting.AdminUser)
}

func getEngine() (*xorm.Engine, error) {
	LoadConfig()

	cnnstr := ""
	switch DbCfg.Type {
	case "mysql":
		cnnstr = fmt.Sprintf("%s:%[email protected](%s)/%s?charset=utf8",
			DbCfg.User, DbCfg.Pwd, DbCfg.Host, DbCfg.Name)
	case "postgres":
		var host, port = "127.0.0.1", "5432"
		fields := strings.Split(DbCfg.Host, ":")
		if len(fields) > 0 && len(strings.TrimSpace(fields[0])) > 0 {
			host = fields[0]
		}
		if len(fields) > 1 && len(strings.TrimSpace(fields[1])) > 0 {
			port = fields[1]
		}
		cnnstr = fmt.Sprintf("user=%s password=%s host=%s port=%s dbname=%s sslmode=%s",
			DbCfg.User, DbCfg.Pwd, host, port, DbCfg.Name, DbCfg.SslMode)
	case "sqlite3":
		if !filepath.IsAbs(DbCfg.Path) {
			DbCfg.Path = filepath.Join(setting.DataPath, DbCfg.Path)
		}
		os.MkdirAll(path.Dir(DbCfg.Path), os.ModePerm)
		cnnstr = "file:" + DbCfg.Path + "?cache=shared&mode=rwc&_loc=Local"
	default:
		return nil, fmt.Errorf("Unknown database type: %s", DbCfg.Type)
	}

	log.Info("Database: %v", DbCfg.Type)

	return xorm.NewEngine(DbCfg.Type, cnnstr)
}

func GetHttpClient() *http.Client {
	if client != nil {
		return client
	} else {
		var certPool *x509.CertPool
		if pemfile := setting.KeystoneRootCAPEMFile; pemfile != "" {
			certPool = x509.NewCertPool()
			pemFileContent, err := ioutil.ReadFile(pemfile)
			if err != nil {
				panic(err)
			}
			if !certPool.AppendCertsFromPEM(pemFileContent) {
				log.Error(3, "Failed to load any certificates from Root CA PEM file %s", pemfile)
			} else {
				log.Info("Successfully loaded certificate(s) from %s", pemfile)
			}
		}
		tr := &http.Transport{
			TLSClientConfig: &tls.Config{RootCAs: certPool,
				InsecureSkipVerify: !setting.KeystoneVerifySSLCert},
		}
		tr.Proxy = http.ProxyFromEnvironment

		client = &http.Client{Transport: tr}
		return client
	}
}

func LogConfigurationInfo() {
	var text bytes.Buffer
	text.WriteString("Configuration Info\n")

	text.WriteString("Config files:\n")
	for i, file := range configFiles {
		text.WriteString(fmt.Sprintf("  [%d]: %s\n", i, file))
	}

	if len(appliedCommandLineProperties) > 0 {
		text.WriteString("Command lines overrides:\n")
		for i, prop := range appliedCommandLineProperties {
			text.WriteString(fmt.Sprintf("  [%d]: %s\n", i, prop))
		}
	}

	if len(appliedEnvOverrides) > 0 {
		text.WriteString("\tEnvironment variables used:\n")
		for i, prop := range appliedCommandLineProperties {
			text.WriteString(fmt.Sprintf("  [%d]: %s\n", i, prop))
		}
	}

	text.WriteString("Paths:\n")
	text.WriteString(fmt.Sprintf("  home: %s\n", HomePath))
	text.WriteString(fmt.Sprintf("  data: %s\n", DataPath))
	text.WriteString(fmt.Sprintf("  logs: %s\n", LogsPath))

	log.Info(text.String())
}