// setDefaults sets default values for any Worker fields that are
// uninitialized.
func (w *Worker) setDefaults() {
	if w.WorkerID == "" {
		// May as well use a UUID here, "it's what we've always done"
		w.WorkerID = uuid.NewV4().String()
	}

	if w.Concurrency == 0 {
		w.Concurrency = runtime.NumCPU()
	}

	if w.PollInterval == time.Duration(0) {
		w.PollInterval = time.Duration(1) * time.Second
	}

	if w.HeartbeatInterval == time.Duration(0) {
		w.HeartbeatInterval = time.Duration(15) * time.Second
	}

	if w.MaxAttempts == 0 {
		w.MaxAttempts = 100
	}

	if w.Clock == nil {
		w.Clock = clock.New()
	}
}

// Ensure that the clock's time matches the standary library.
func TestClock_Now(t *testing.T) {
	a := time.Now().Round(time.Second)
	b := clock.New().Now().Round(time.Second)
	if !a.Equal(b) {
		t.Errorf("not equal: %s != %s", a, b)
	}
}

func newRateLimiter(perSec int, maxBurst time.Duration) *rateLimiter {
	maxPerBatch := int64(perSec / int(time.Second/maxBurst))
	return &rateLimiter{
		limitPerSec: perSec,
		resolution:  maxBurst,
		time:        clock.New(),
		maxPerBatch: maxPerBatch,
	}
}

func NewLWWSetWithBias(bias BiasType) (*LWWSet, error) {
	if bias != BiasAdd && bias != BiasRemove {
		return nil, ErrNoSuchBias
	}

	return &LWWSet{
		addMap: make(map[interface{}]time.Time),
		rmMap:  make(map[interface{}]time.Time),
		bias:   bias,
		clock:  clock.New(),
	}, nil
}

func NewEngine() *Engine {
	e := &Engine{
		ticker:        NewTicker(time.Now(), time.Second*0, clock.New()),
		execQueue:     make(chan *Job, 1000),
		scheduler:     NewScheduler(),
		evalHandler:   NewEvalHandler(),
		ruleReader:    NewRuleReader(),
		log:           log.New("alerting.engine"),
		resultHandler: NewResultHandler(),
	}

	return e
}

// Create a new zone database
func NewZoneDb(config ZoneConfig) (zone *ZoneDb, err error) {
	zone = &ZoneDb{
		domain:           config.Domain,
		idents:           make(identRecordSet),
		mdnsCli:          config.MDNSClient,
		mdnsSrv:          config.MDNSServer,
		iface:            config.Iface,
		clock:            config.Clock,
		relevantLimit:    time.Duration(DefaultRelevantTime) * time.Second,
		refreshCloseChan: make(chan bool),
	}

	// fix the default configuration parameters
	if zone.clock == nil {
		zone.clock = clock.New()
	}
	if len(zone.domain) == 0 {
		zone.domain = DefaultLocalDomain
	}
	if config.RefreshInterval > 0 {
		zone.refreshInterval = time.Duration(config.RefreshInterval) * time.Second
	}
	if config.RelevantTime > 0 {
		zone.relevantLimit = time.Duration(config.RelevantTime) * time.Second
	}

	if zone.refreshInterval > 0 {
		zone.refreshScheds = NewSchedQueue(zone.clock)
	}

	// Create the mDNS client and server
	if zone.mdnsCli == nil {
		if zone.mdnsCli, err = NewMDNSClient(); err != nil {
			return
		}
	}
	if zone.mdnsSrv == nil {
		mdnsTTL := DefaultLocalTTL
		if config.LocalTTL > 0 {
			mdnsTTL = config.LocalTTL
		}

		if zone.mdnsSrv, err = NewMDNSServer(zone, false, mdnsTTL); err != nil {
			return
		}
	}

	return
}

// Spawn initializes the limiter
func (l *Limiter) Spawn(id int) utils.Composer {
	l.keepSending = make(chan struct{}, l.Config.Burst)
	if l.clk == nil {
		l.clk = clock.New()
	}

	go func() {
		for {
			<-l.clk.Timer(1 * time.Second).C
			l.keepSending <- struct{}{}
		}
	}()

	return l
}

// Ensure that the clock's timer can be stopped.
func TestClock_Timer_Stop(t *testing.T) {
	var ok bool
	go func() {
		time.Sleep(10 * time.Millisecond)
		ok = true
	}()

	timer := clock.New().Timer(20 * time.Millisecond)
	timer.Stop()
	select {
	case <-timer.C:
		t.Fatal("unexpected send")
	case <-time.After(30 * time.Millisecond):
	}
}

// Ensure that the clock sleeps for the appropriate amount of time.
func TestClock_Sleep(t *testing.T) {
	var ok bool
	go func() {
		time.Sleep(10 * time.Millisecond)
		ok = true
	}()
	go func() {
		time.Sleep(30 * time.Millisecond)
		t.Fatal("too late")
	}()
	gosched()

	clock.New().Sleep(20 * time.Millisecond)
	if !ok {
		t.Fatal("too early")
	}
}

// NewCache creates a cache of the given capacity
func NewCache(capacity int, clk clock.Clock) (*Cache, error) {
	if capacity <= 0 {
		return nil, errInvalidCapacity
	}
	c := &Cache{
		capacity: capacity,
		entries:  make(entries, capacity),
		clock:    clk,
	}

	if c.clock == nil {
		c.clock = clock.New()
	}

	heap.Init(&c.entriesH)
	return c, nil
}

// Ensure that the clock's ticker ticks correctly.
func TestClock_Ticker(t *testing.T) {
	var ok bool
	go func() {
		time.Sleep(100 * time.Millisecond)
		ok = true
	}()
	go func() {
		time.Sleep(200 * time.Millisecond)
		t.Fatal("too late")
	}()
	gosched()

	ticker := clock.New().Ticker(50 * time.Millisecond)
	<-ticker.C
	<-ticker.C
	if !ok {
		t.Fatal("too early")
	}
}

func defaultOptions() *Options {
	opts := &Options{
		SuspicionTimeout:  5000 * time.Millisecond,
		MinProtocolPeriod: 200 * time.Millisecond,

		JoinTimeout:        1000 * time.Millisecond,
		PingTimeout:        1500 * time.Millisecond,
		PingRequestTimeout: 5000 * time.Millisecond,

		PingRequestSize: 3,

		RollupFlushInterval: 5000 * time.Millisecond,
		RollupMaxUpdates:    250,

		Clock: clock.New(),
	}

	return opts
}

// Ensure that the clock ticks correctly.
func TestClock_Tick(t *testing.T) {
	var ok bool
	go func() {
		time.Sleep(10 * time.Millisecond)
		ok = true
	}()
	go func() {
		time.Sleep(50 * time.Millisecond)
		t.Fatal("too late")
	}()
	gosched()

	c := clock.New().Tick(20 * time.Millisecond)
	<-c
	<-c
	if !ok {
		t.Fatal("too early")
	}
}

// start statsdaemon instance with standard network daemon behaviors
func (s *StatsDaemon) Run(listen_addr, admin_addr, graphite_addr string) {
	s.listen_addr = listen_addr
	s.admin_addr = admin_addr
	s.graphite_addr = graphite_addr
	s.submitFunc = s.GraphiteQueue
	s.Clock = clock.New()
	s.graphiteQueue = make(chan []byte, 1000)
	log.Printf("statsdaemon instance '%s' starting\n", s.instance)
	output := &common.Output{
		Metrics:       s.Metrics,
		MetricAmounts: s.metricAmounts,
		Valid_lines:   s.valid_lines,
		Invalid_lines: s.Invalid_lines,
	}
	go udp.StatsListener(s.listen_addr, s.prefix, output) // set up udp listener that writes messages to output's channels (i.e. s's channels)
	go s.adminListener()                                  // tcp admin_addr to handle requests
	go s.metricStatsMonitor()                             // handles requests fired by telnet api
	go s.graphiteWriter()                                 // writes to graphite in the background
	s.metricsMonitor()                                    // takes data from s.Metrics and puts them in the guage/timers/etc objects. pointers guarded by select. also listens for signals.
}

// Ensure that the clock's AfterFunc executes at the correct time.
func TestClock_AfterFunc(t *testing.T) {
	var ok bool
	go func() {
		time.Sleep(10 * time.Millisecond)
		ok = true
	}()
	go func() {
		time.Sleep(30 * time.Millisecond)
		t.Fatal("too late")
	}()
	gosched()

	var wg sync.WaitGroup
	wg.Add(1)
	clock.New().AfterFunc(20*time.Millisecond, func() {
		wg.Done()
	})
	wg.Wait()
	if !ok {
		t.Fatal("too early")
	}
}

// ServeCBORRPC runs a CBOR-RPC server on the specified local address.
// This serves connections forever, and probably wants to be run in a
// goroutine.  Panics on any error in the initial setup or in accepting
// connections.
func ServeCBORRPC(
	coord coordinate.Coordinate,
	gConfig map[string]interface{},
	network, laddr string,
	reqLogger *logrus.Logger,
) {
	var (
		cbor      *codec.CborHandle
		err       error
		namespace coordinate.Namespace
		ln        net.Listener
		conn      net.Conn
		jobd      *jobserver.JobServer
	)

	cbor = new(codec.CborHandle)
	if err == nil {
		err = cborrpc.SetExts(cbor)
	}
	if err == nil {
		namespace, err = coord.Namespace("")
	}
	if err == nil {
		jobd = &jobserver.JobServer{
			Namespace:    namespace,
			GlobalConfig: gConfig,
			Clock:        clock.New(),
		}
		ln, err = net.Listen(network, laddr)
	}
	for err == nil {
		conn, err = ln.Accept()
		if err == nil {
			go handleConnection(conn, jobd, cbor, reqLogger)
		}
	}
	panic(err)
}

// Dispatcher dispatches, every second, all jobs that should run for that second
// every job has an id so that you can run multiple dispatchers (for HA) while still only processing each job once.
// (provided jobs get consistently routed to executors)
func Dispatcher(jobQueue JobQueue) {
	go dispatchJobs(jobQueue)
	offset := time.Duration(LoadOrSetOffset()) * time.Second
	// no need to try resuming where we left off in the past.
	// see https://github.com/raintank/grafana/issues/266
	lastProcessed := time.Now().Truncate(time.Second).Add(-offset)
	cl := clock.New()
	ticker := NewTicker(lastProcessed, offset, cl)
	go func() {
		offsetReadTicker := cl.Ticker(time.Duration(1) * time.Second)
		for range offsetReadTicker.C {
			offset := time.Duration(LoadOrSetOffset()) * time.Second
			ticker.updateOffset(offset)
		}
	}()
	for {
		select {
		case tick := <-ticker.C:
			tickQueueItems.Value(int64(len(tickQueue)))
			tickQueueSize.Value(int64(setting.TickQueueSize))

			// let's say jobs with freq 60 and offset 7 trigger at 7, 67, 127, ...
			// and offset was 30 seconds, so we query for data with last point at 37, 97, 157, ...
			// so we should find the checks where ts-30 % frequency == offset
			// and then ts-30 was a ts of the last point we should query for

			select {
			case tickQueue <- tick:
			default:
				dispatcherTicksSkippedDueToSlowTickQueue.Inc(1)
			}
			tickQueueItems.Value(int64(len(tickQueue)))
			tickQueueSize.Value(int64(setting.TickQueueSize))
		}
	}
}

func newCounter() *rateCounter {
	return &rateCounter{
		time: clock.New(),
	}
}

// Profile will wrap a writer and reader pair and profile where
// time is spent: writing or reading. The result is returned when
// the `done` func is called. The `done` func can be called multiple
// times.
//
// There is a small performance overhead of ~µs per Read/Write call.
// This is negligible in most I/O workloads. If the overhead is too
// much for your needs, use the `ProfileSample` call.
func Profile(w io.Writer, r io.Reader) (pw io.Writer, pr io.Reader, done func() TimeProfile) {
	return profile(clock.New(), w, r)
}

// New creates a new Coordinate interface that operates purely in
// memory.
func New() coordinate.Coordinate {
	clk := clock.New()
	return NewWithClock(clk)
}