package statscollection

import (
	"database/sql"
	"fmt"
	"log"
	"strconv"
	"time"

	"github.com/cactus/go-statsd-client/statsd"
	_ "github.com/lib/pq"
)

func vlogf(verbose bool, fmt string, args ...interface{}) {
	if verbose {
		log.Printf(fmt, args...)
	}
}

func RecordLoadRow(rows *sql.Rows, stats StatsTracker) (err error) {
	var relation, bytes string
	var seqTupRead, idxTupFetch, nTupIns, nTupUpd, nTupDel, nTupHotUpd string
	var nLiveTup, nDeadTup string
	var lastVacuum, lastAutoVacuum string
	var nVacuumCount, nAutoVacuumCount string
	if err := rows.Scan(&relation, &bytes, &seqTupRead, &idxTupFetch, &nTupIns, &nTupUpd, &nTupDel,
		&nTupHotUpd, &nLiveTup, &nDeadTup, &lastVacuum, &lastAutoVacuum, &nVacuumCount, &nAutoVacuumCount); err != nil {
		return err
	}

	//counter for RecordDelta and RecordAbsolute to be gauges
	stats.RecordDelta(relation, "bytes", bytes)
	stats.RecordDelta(relation, "seq_tup_red", seqTupRead)
	stats.RecordDelta(relation, "idx_tup_fetch", idxTupFetch)
	stats.RecordDelta(relation, "n_tup_ins", nTupIns)
	stats.RecordDelta(relation, "n_tup_upd", nTupUpd)
	stats.RecordDelta(relation, "n_tup_del", nTupDel)
	stats.RecordDelta(relation, "n_tup_hot_upd", nTupHotUpd)
	stats.RecordAbsolute(relation, "n_live_tup", nLiveTup)
	stats.RecordAbsolute(relation, "n_dead_tup", nDeadTup)
	stats.RecordDelta(relation, "last_vacuum", lastVacuum)
	stats.RecordDelta(relation, "last_autovacuum", lastAutoVacuum)
	stats.RecordAbsolute(relation, "vacuum_count", nVacuumCount)
	stats.RecordAbsolute(relation, "autovacuum_count", nAutoVacuumCount)
	return nil
}

// The usage pattern for a statsTracker is to start a 'frame' by
// calling ClearOldHistory(), call Record*() several times, and
// finally call Format() to finish out the frame.
type StatsTracker interface {
	Reset()
	ClearOldHistory(now int64, globalInterval int64)
	RecordAbsolute(relation string, column string, value string)
	RecordDelta(relation string, column string, value string)
	FormatStats(stats statsd.Statter, now int64, cluster string, nodeRole string, hostname string, schema string, verbose bool) (err error)
	tick(now int64)
}

// The statsCollection struct stores the absolute values found in
// last_stats and records what should be logged in the current poll.
// Therefore, deltas will never be logged on the first frame or on
// a frame after a discontinuity.
type columnValues map[string]int64
type relationStats map[string]columnValues
type statsCollection struct {
	currentStats relationStats
	lastStats    relationStats
	lastAt       int64
}

func NewStatsCollection() *statsCollection {
	stats := new(statsCollection)
	stats.Reset()
	return stats
}

func (stats statsCollection) String() string {
	return fmt.Sprintf("last=%v current=%v at=%v", stats.lastStats, stats.currentStats, stats.lastAt)
}

func (stats *statsCollection) Reset() {
	stats.currentStats = make(relationStats)
	stats.lastStats = make(relationStats)
	stats.lastAt = time.Now().Unix()
}

// If the stats are found to be too old, we don't want to draw
// confusing delta graphs. By zero-ing out history, this frame of
// deltas will not be logged.
func (stats *statsCollection) ClearOldHistory(now int64, globalInterval int64) {
	if stats.lastAt < now-2*globalInterval {
		log.Println("Previous results are from too long ago. Assuming discontinuity.")
		stats.lastStats = make(relationStats)
	}
}

// This just records the value found
func (stats *statsCollection) RecordAbsolute(relation string, column string, value string) {
	val, err := strconv.ParseInt(value, 10, 64)
	if err != nil {
		log.Printf("Unable to parse %v.%v value '%v': %v\n", relation, column, value, err)
		return // we want to go forward, so we don't bubble up the error here.
	}
	// We want to record an absolute value so we only need to look in current.
	current, ok := stats.currentStats[relation]
	if val == 0 {
		if ok {
			// Zero is not an interesting value for recording metrics. Omit it.
			delete(stats.currentStats[relation], column)
		}
		return
	}
	if !ok {
		current = make(columnValues)
		stats.currentStats[relation] = current
	}
	current[column] = val
}

func (stats *statsCollection) RecordDelta(relation string, column string, value string) {
	val, err := strconv.ParseInt(value, 10, 64)
	if err != nil {
		log.Printf("Unable to parse %v.%v value '%v': %v\n", relation, column, value, err)
		return // we want to go forward, so we don't bubble up the error here.
	}
	last, ok := stats.lastStats[relation]
	if !ok {
		last = make(columnValues)
		stats.lastStats[relation] = last
	}
	oldValue, ok := last[column]
	last[column] = val
	if ok {
		delta := val - oldValue
		if delta <= 0 {
			if delta < 0 {
				log.Printf("Backwards movement in %v.%v. Assuming discontinuity.", relation, column)
			}
			// We don't want to log it since 0 is uninteresting and negative leads to
			// confusing graphs.
			delete(stats.currentStats[relation], column)
			return
		}
		current, ok := stats.currentStats[relation]
		if !ok {
			current = make(columnValues)
			stats.currentStats[relation] = current
		}
		current[column] = delta
	}
}

// Advance to the next frame.
func (stats *statsCollection) tick(now int64) {
	stats.currentStats = make(relationStats)
	stats.lastAt = now
}

func (stats *statsCollection) FormatStats(statsd statsd.Statter, now int64, cluster string, nodeRole string, hostname string, schema string, verbose bool) (err error) {
	defer stats.tick(now)
	for relation, current := range stats.currentStats {
		for column, value := range current {
			if column == "n_live_tup" || column == "n_dead_tup" || column == "last_vacuum" || column == "last_autovacuum" || column == "vacuum_count" || column == "autovacuum_count" {
				line := fmt.Sprintf("postgres.%v.%v.%v.%v.relations.%v.%v", cluster, nodeRole, hostname, schema, relation, column)
				vlogf(verbose, "Logging: %v %v", line, value)
				if err := statsd.Gauge(line, value, 1.0); err != nil {
					return err
				}
			} else {
				line := fmt.Sprintf("postgres.%v.%v.%v.%v.relations.%v.%v", cluster, nodeRole, hostname, schema, relation, column)
				vlogf(verbose, "Logging: %v %v", line, value)
				if err := statsd.Inc(line, value, 1.0); err != nil {
					return err
				}
			}
		}
	}
	return nil
}
