// AmqpExecutor reads jobs from rabbitmq, executes them, and acknowledges them
// if they processed succesfully or encountered a fatal error
// (i.e. an error that we know won't recover on future retries, so no point in retrying)
func AmqpExecutor(fn GraphiteReturner, consumer rabbitmq.Consumer, cache *lru.Cache) {
	executorNum.Inc(1)
	defer executorNum.Dec(1)
	consumer.Consume(func(msg *amqp.Delivery) error {
		job := Job{}
		if err := json.Unmarshal(msg.Body, &job); err != nil {
			log.Error(0, "failed to unmarshal msg body.", err)
			return nil
		}
		job.StoreMetricFunc = api.StoreMetric
		var err error
		if setting.AlertingInspect {
			inspect(GraphiteAuthContextReturner, &job, cache)
		} else {
			err = execute(GraphiteAuthContextReturner, &job, cache)
		}
		if err != nil {
			if strings.HasPrefix(err.Error(), "fatal:") {
				log.Error(0, "%s: removing job from queue", err.Error())
				return nil
			}
			log.Error(0, "%s: not acking message. retry later", err.Error())
		}
		return err
	})
}

func (h *Handler) HandleMessage(m *nsq.Message) error {
	ms, err := msg.MetricDataFromMsg(m.Body)
	if err != nil {
		log.Error(3, "skipping message. %s", err)
		return nil
	}
	msgsAge.Value(time.Now().Sub(ms.Produced).Nanoseconds() / 1000)

	err = ms.DecodeMetricData()
	if err != nil {
		log.Error(3, "skipping message. %s", err)
		return nil
	}
	metricsPerMessage.Value(int64(len(ms.Metrics)))

	metricsReceived.Inc(int64(len(ms.Metrics)))
	for _, metric := range ms.Metrics {
		if metric.Time == 0 {
			log.Warn("invalid metric. metric.Time is 0. %s", metric.Id())
		} else {
			m := h.metrics.GetOrCreate(metric.Id())
			m.Add(uint32(metric.Time), metric.Value)
		}
	}

	return nil
}

func sendUsageStats() {
	log.Trace("Sending anonymous usage stats to stats.grafana.org")

	version := strings.Replace(setting.BuildVersion, ".", "_", -1)

	metrics := map[string]interface{}{}
	report := map[string]interface{}{
		"version": version,
		"metrics": metrics,
	}

	UsageStats.Each(func(name string, i interface{}) {
		switch metric := i.(type) {
		case Counter:
			if metric.Count() > 0 {
				metrics[name+".count"] = metric.Count()
				metric.Clear()
			}
		}
	})

	statsQuery := m.GetSystemStatsQuery{}
	if err := bus.Dispatch(&statsQuery); err != nil {
		log.Error(3, "Failed to get system stats", err)
		return
	}

	metrics["stats.dashboards.count"] = statsQuery.Result.DashboardCount
	metrics["stats.users.count"] = statsQuery.Result.UserCount
	metrics["stats.orgs.count"] = statsQuery.Result.OrgCount
	metrics["stats.playlist.count"] = statsQuery.Result.PlaylistCount
	metrics["stats.plugins.apps.count"] = len(plugins.Apps)
	metrics["stats.plugins.panels.count"] = len(plugins.Panels)
	metrics["stats.plugins.datasources.count"] = len(plugins.DataSources)

	dsStats := m.GetDataSourceStatsQuery{}
	if err := bus.Dispatch(&dsStats); err != nil {
		log.Error(3, "Failed to get datasource stats", err)
		return
	}

	// send counters for each data source
	// but ignore any custom data sources
	// as sending that name could be sensitive information
	dsOtherCount := 0
	for _, dsStat := range dsStats.Result {
		if m.IsKnownDataSourcePlugin(dsStat.Type) {
			metrics["stats.ds."+dsStat.Type+".count"] = dsStat.Count
		} else {
			dsOtherCount += dsStat.Count
		}
	}
	metrics["stats.ds.other.count"] = dsOtherCount

	out, _ := json.MarshalIndent(report, "", " ")
	data := bytes.NewBuffer(out)

	client := http.Client{Timeout: time.Duration(5 * time.Second)}
	go client.Post("https://stats.grafana.org/grafana-usage-report", "application/json", data)
}

// don't ever call with a ts of 0, cause we use 0 to mean not initialized!
func (a *AggMetric) Add(ts uint32, val float64) {
	a.Lock()
	defer a.Unlock()

	t0 := ts - (ts % a.ChunkSpan)

	currentChunk := a.getChunk(a.CurrentChunkPos)
	if currentChunk == nil {
		chunkCreate.Inc(1)
		// no data has been added to this metric at all.
		log.Debug("instantiating new circular buffer.")
		a.Chunks = append(a.Chunks, NewChunk(t0))

		if err := a.Chunks[0].Push(ts, val); err != nil {
			panic(fmt.Sprintf("FATAL ERROR: this should never happen. Pushing initial value <%d,%f> to new chunk at pos 0 failed: %q", ts, val, err))
		}

		log.Debug("created new chunk. %s:  %v", a.Key, a.Chunks[0])
	} else if t0 == currentChunk.T0 {
		if currentChunk.Saved {
			//TODO(awoods): allow the chunk to be re-opened.
			log.Error(3, "cant write to chunk that has already been saved. %s T0:%d", a.Key, currentChunk.T0)
			return
		}
		// last prior data was in same chunk as new point
		if err := a.Chunks[a.CurrentChunkPos].Push(ts, val); err != nil {
			log.Error(3, "failed to add metric to chunk for %s. %s", a.Key, err)
			return
		}
	} else if t0 < currentChunk.T0 {
		log.Error(3, "Point at %d has t0 %d, goes back into previous chunk. CurrentChunk t0: %d, LastTs: %d", ts, t0, currentChunk.T0, currentChunk.LastTs)
		return
	} else {
		currentChunk.Finish()
		go a.Persist(currentChunk)

		a.CurrentChunkPos++
		if a.CurrentChunkPos >= int(a.NumChunks) {
			a.CurrentChunkPos = 0
		}

		chunkCreate.Inc(1)
		if len(a.Chunks) < int(a.NumChunks) {
			log.Debug("adding new chunk to cirular Buffer. now %d chunks", a.CurrentChunkPos+1)
			a.Chunks = append(a.Chunks, NewChunk(t0))
		} else {
			chunkClear.Inc(1)
			log.Debug("numChunks: %d  currentPos: %d", len(a.Chunks), a.CurrentChunkPos)
			log.Debug("clearing chunk from circular buffer. %v", a.Chunks[a.CurrentChunkPos])
			a.Chunks[a.CurrentChunkPos] = NewChunk(t0)
		}
		log.Debug("created new chunk. %s: %v", a.Key, a.Chunks[a.CurrentChunkPos])

		if err := a.Chunks[a.CurrentChunkPos].Push(ts, val); err != nil {
			panic(fmt.Sprintf("FATAL ERROR: this should never happen. Pushing initial value <%d,%f> to new chunk at pos %d failed: %q", ts, val, a.CurrentChunkPos, err))
		}
	}
	a.addAggregators(ts, val)
}

func main() {
	buildstampInt64, _ := strconv.ParseInt(buildstamp, 10, 64)

	setting.BuildVersion = version
	setting.BuildCommit = commit
	setting.BuildStamp = buildstampInt64

	go listenToSystemSignels()

	flag.Parse()
	writePIDFile()
	initRuntime()

	if setting.ProfileHeapMB > 0 {
		errors := make(chan error)
		go func() {
			for e := range errors {
				log.Error(0, e.Error())
			}
		}()
		heap, _ := heap.New(setting.ProfileHeapDir, setting.ProfileHeapMB*1000000, setting.ProfileHeapWait, time.Duration(1)*time.Second, errors)
		go heap.Run()
	}

	search.Init()
	login.Init()
	social.NewOAuthService()
	eventpublisher.Init()
	plugins.Init()
	elasticstore.Init()

	metricsBackend, err := helper.New(setting.StatsdEnabled, setting.StatsdAddr, setting.StatsdType, "grafana", setting.InstanceId)
	if err != nil {
		log.Error(3, "Statsd client:", err)
	}
	metricpublisher.Init(metricsBackend)
	collectoreventpublisher.Init(metricsBackend)
	api.InitCollectorController(metricsBackend)
	if setting.AlertingEnabled {
		alerting.Init(metricsBackend)
		alerting.Construct()
	}

	if err := notifications.Init(); err != nil {
		log.Fatal(3, "Notification service failed to initialize", err)
	}

	if setting.ReportingEnabled {
		go metrics.StartUsageReportLoop()
	}

	cmd.StartServer()
	exitChan <- 0
}

func init() {
	contextCache = NewContextCache()
	var err error
	server, err = socketio.NewServer([]string{"polling", "websocket"})
	if err != nil {
		log.Fatal(4, "failed to initialize socketio.", err)
		return
	}
	server.On("connection", func(so socketio.Socket) {
		c, err := register(so)
		if err != nil {
			if err == m.ErrInvalidApiKey {
				log.Info("collector failed to authenticate.")
			} else if err.Error() == "invalid collector version. Please upgrade." {
				log.Info("collector is wrong version")
			} else {
				log.Error(0, "Failed to initialize collector.", err)
			}
			so.Emit("error", err.Error())
			return
		}
		log.Info("connection registered without error")
		//get list of monitorTypes
		cmd := &m.GetMonitorTypesQuery{}
		if err := bus.Dispatch(cmd); err != nil {
			log.Error(0, "Failed to initialize collector.", err)
			so.Emit("error", err)
			return
		}
		log.Info("sending ready event to collector %s", c.Collector.Name)
		readyPayload := map[string]interface{}{
			"collector":     c.Collector,
			"monitor_types": cmd.Result,
			"socket_id":     c.SocketId,
		}
		c.Socket.Emit("ready", readyPayload)
		log.Info("binding event handlers for collector %s owned by OrgId: %d", c.Collector.Name, c.OrgId)
		c.Socket.On("event", c.OnEvent)
		c.Socket.On("results", c.OnResults)
		c.Socket.On("disconnection", c.OnDisconnection)
		log.Info("calling refresh for collector %s owned by OrgId: %d", c.Collector.Name, c.OrgId)
	})

	server.On("error", func(so socketio.Socket, err error) {
		log.Error(0, "socket emitted error", err)
	})

}

func ProcessBuffer(c <-chan m.MetricDefinition) {
	buf := make(map[uint32][]m.MetricDefinition)

	// flush buffer 10 times every second
	t := time.NewTicker(time.Millisecond * 100)
	for {
		select {
		case b := <-c:
			if b.OrgId != 0 {
				//get hash.
				h := crc32.NewIEEE()
				h.Write([]byte(b.Name))
				hash := h.Sum32() % uint32(1024)
				if _, ok := buf[hash]; !ok {
					buf[hash] = make([]m.MetricDefinition, 0)
				}
				buf[hash] = append(buf[hash], b)
			}
		case <-t.C:
			//copy contents of buffer
			for hash, metrics := range buf {
				currentBuf := make([]m.MetricDefinition, len(metrics))
				copy(currentBuf, metrics)
				delete(buf, hash)
				//log.Info(fmt.Sprintf("flushing %d items in buffer now", len(currentBuf)))
				msgString, err := json.Marshal(currentBuf)
				if err != nil {
					log.Error(0, "Failed to marshal metrics payload.", err)
				} else {
					go Publish(fmt.Sprintf("%d", hash), msgString)
				}
			}
		}
	}
}

func EncryptPassword(password string) string {
	key := []byte(setting.KeystoneCredentialAesKey)
	block, err := aes.NewCipher(key)
	if err != nil {
		log.Error(3, "Error: NewCipher(%d bytes) = %s", len(setting.KeystoneCredentialAesKey), err)
	}
	ciphertext := make([]byte, aes.BlockSize+len(password))
	iv := ciphertext[:aes.BlockSize]
	if _, err := io.ReadFull(rand.Reader, iv); err != nil {
		log.Error(3, "Error: %s", err)
	}
	stream := cipher.NewOFB(block, iv)
	stream.XORKeyStream(ciphertext[aes.BlockSize:], []byte(password))

	return base64.StdEncoding.EncodeToString(ciphertext)
}

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 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 inTransaction2(callback dbTransactionFunc2) error {
	var err error

	sess := session{Session: x.NewSession()}

	defer sess.Close()
	if err = sess.Begin(); err != nil {
		return err
	}

	err = callback(&sess)

	if err != nil {
		sess.Rollback()
		return err
	} else if err = sess.Commit(); err != nil {
		return err
	}

	if len(sess.events) > 0 {
		for _, e := range sess.events {
			if err = bus.Publish(e); err != nil {
				log.Error(3, "Failed to publish event after commit", err)
			}
		}
	}

	return nil
}

func (ctx *Context) JsonApiErr(status int, message string, err error) {
	resp := make(map[string]interface{})

	if err != nil {
		log.Error(4, "%s: %v", message, err)
		if setting.Env != setting.PROD {
			resp["error"] = err.Error()
		}
	}

	switch status {
	case 404:
		metrics.M_Api_Status_404.Inc(1)
		resp["message"] = "Not Found"
	case 500:
		metrics.M_Api_Status_500.Inc(1)
		resp["message"] = "Internal Server Error"
	}

	if message != "" {
		resp["message"] = message
	}

	ctx.JSON(status, resp)
}

func ApiError(status int, message string, err error) *NormalResponse {
	resp := make(map[string]interface{})

	if err != nil {
		log.Error(4, "%s: %v", message, err)
		if setting.Env != setting.PROD {
			resp["error"] = err.Error()
		}
	}

	switch status {
	case 404:
		resp["message"] = "Not Found"
		metrics.M_Api_Status_500.Inc(1)
	case 500:
		metrics.M_Api_Status_404.Inc(1)
		resp["message"] = "Internal Server Error"
	}

	if message != "" {
		resp["message"] = message
	}

	return Json(status, resp)
}

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 dispatchJobs(jobQueue JobQueue) {
	for lastPointAt := range tickQueue {
		tickQueueItems.Value(int64(len(tickQueue)))
		tickQueueSize.Value(int64(setting.TickQueueSize))

		pre := time.Now()
		jobs, err := getJobs(lastPointAt.Unix())
		dispatcherNumGetSchedules.Inc(1)
		dispatcherGetSchedules.Value(time.Since(pre))

		if err != nil {
			log.Error(0, "getJobs() failed: %q", err)
			continue
		}

		dispatcherJobSchedulesSeen.Inc(int64(len(jobs)))
		for _, job := range jobs {
			job.GeneratedAt = time.Now()
			job.LastPointTs = lastPointAt
			job.AssertStart = lastPointAt.Add(-time.Duration(job.Freq) * time.Second)

			jobQueue.Put(job)

			dispatcherJobsScheduled.Inc(1)
		}
	}
}

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 (c *CollectorContext) Refresh() {
	log.Info("Collector %d refreshing", c.Collector.Id)
	//step 1. get list of collectorSessions for this collector.
	q := m.GetCollectorSessionsQuery{CollectorId: c.Collector.Id}
	if err := bus.Dispatch(&q); err != nil {
		log.Error(0, "failed to get list of collectorSessions.", err)
		return
	}
	org := c.Collector.OrgId
	if c.Collector.Public {
		org = 0
	}
	totalSessions := len(q.Result)
	//step 2. for each session
	for pos, sess := range q.Result {
		//we only need to refresh the 1 socket.
		if sess.SocketId != c.SocketId {
			continue
		}
		//step 3. get list of monitors configured for this colletor.
		monQuery := m.GetMonitorsQuery{
			OrgId:          org,
			IsGrafanaAdmin: true,
			Modulo:         int64(totalSessions),
			ModuloOffset:   int64(pos),
			Enabled:        "true",
		}
		if err := bus.Dispatch(&monQuery); err != nil {
			log.Error(0, "failed to get list of monitors.", err)
			return
		}
		log.Info("sending refresh to " + sess.SocketId)
		//step 5. send to socket.
		monitors := make([]*m.MonitorDTO, 0)
		for _, mon := range monQuery.Result {
			for _, col := range mon.Collectors {
				if col == c.Collector.Id {
					monitors = append(monitors, mon)
					break
				}
			}
		}
		c.Socket.Emit("refresh", monitors)
	}
}

func (k *StdoutHandler) HandleMessage(m *nsq.Message) error {
	ms, err := msg.MetricDataFromMsg(m.Body)
	if err != nil {
		log.Error(3, "%s: skipping message", err.Error())
		return nil
	}

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

	for _, m := range ms.Metrics {
		fmt.Println(m.Name, m.Time, m.Value, m.Tags)
	}
	return nil
}

func RenderToPng(params *RenderOpts) (string, error) {
	log.Info("PhantomRenderer::renderToPng url %v", params.Url)

	var executable = "phantomjs"
	if runtime.GOOS == "windows" {
		executable = executable + ".exe"
	}

	binPath, _ := filepath.Abs(filepath.Join(setting.PhantomDir, executable))
	scriptPath, _ := filepath.Abs(filepath.Join(setting.PhantomDir, "render.js"))
	pngPath, _ := filepath.Abs(filepath.Join(setting.ImagesDir, util.GetRandomString(20)))
	pngPath = pngPath + ".png"

	cmd := exec.Command(binPath, "--ignore-ssl-errors=true", scriptPath, "url="+params.Url, "width="+params.Width,
		"height="+params.Height, "png="+pngPath, "cookiename="+setting.SessionOptions.CookieName,
		"domain="+setting.Domain, "sessionid="+params.SessionId)
	stdout, err := cmd.StdoutPipe()

	if err != nil {
		return "", err
	}
	stderr, err := cmd.StderrPipe()
	if err != nil {
		return "", err
	}

	err = cmd.Start()
	if err != nil {
		return "", err
	}

	go io.Copy(os.Stdout, stdout)
	go io.Copy(os.Stdout, stderr)

	done := make(chan error)
	go func() {
		cmd.Wait()
		close(done)
	}()

	timeout, err := strconv.Atoi(params.Timeout)
	if err != nil {
		timeout = 15
	}

	select {
	case <-time.After(time.Duration(timeout) * time.Second):
		if err := cmd.Process.Kill(); err != nil {
			log.Error(4, "failed to kill: %v", err)
		}
		return "", fmt.Errorf("PhantomRenderer::renderToPng timeout (>%vs)", timeout)
	case <-done:
	}

	return pngPath, nil
}

func indexMetric(m *schema.MetricDefinition) error {
	log.Debug("indexing %s in redis", m.Id)
	metricStr, err := json.Marshal(m)
	if err != nil {
		return err
	}
	if rerr := rs.SetEx(m.Id, time.Duration(300)*time.Second, string(metricStr)).Err(); err != nil {
		log.Error(3, "redis err. %s", rerr)
	}

	log.Debug("indexing %s in elasticsearch", m.Id)
	err = Indexer.Index("metric", "metric_index", m.Id, "", "", nil, m)
	if err != nil {
		log.Error(3, "failed to send payload to BulkApi indexer. %s", err)
		return err
	}

	return nil
}