5 files changed, 445 insertions, 0 deletions
diff --git a/vendor/codeberg.org/gruf/go-sched/LICENSE b/vendor/codeberg.org/gruf/go-sched/LICENSE
new file mode 100644
index 000000000..e4163ae35
--- /dev/null
+++ b/vendor/codeberg.org/gruf/go-sched/LICENSE
@@ -0,0 +1,9 @@
+MIT License
+
+Copyright (c) 2022 gruf
+
+Permission is hereby granted, free of charge, to any person obtaining a copy of this software and associated documentation files (the "Software"), to deal in the Software without restriction, including without limitation the rights to use, copy, modify, merge, publish, distribute, sublicense, and/or sell copies of the Software, and to permit persons to whom the Software is furnished to do so, subject to the following conditions:
+
+The above copyright notice and this permission notice shall be included in all copies or substantial portions of the Software.
+
+THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY, FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM, OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE SOFTWARE.
diff --git a/vendor/codeberg.org/gruf/go-sched/README.md b/vendor/codeberg.org/gruf/go-sched/README.md
new file mode 100644
index 000000000..d32a961ae
--- /dev/null
+++ b/vendor/codeberg.org/gruf/go-sched/README.md
@@ -0,0 +1,5 @@
+# go-sched
+
+A simple job (both run-once and recurring) queueing library with down-to millisecond precision.
+
+Precision estimates based on test output (running on i7-11800h): 1ms precision with 80% tolerance.
+\ No newline at end of file
diff --git a/vendor/codeberg.org/gruf/go-sched/job.go b/vendor/codeberg.org/gruf/go-sched/job.go
new file mode 100644
index 000000000..66e24fe9a
--- /dev/null
+++ b/vendor/codeberg.org/gruf/go-sched/job.go
@@ -0,0 +1,99 @@
+package sched
+
+import (
+	"time"
+
+	"codeberg.org/gruf/go-atomics"
+)
+
+// Job encapsulates logic for a scheduled job to be run according
+// to a set Timing, executing the job with a set panic handler, and
+// holding onto a next execution time safely in a concurrent environment.
+type Job struct {
+	id     uint64
+	next   atomics.Time
+	timing Timing
+	call   func(time.Time)
+	panic  func(interface{})
+}
+
+// NewJob returns a new Job to run given function.
+func NewJob(fn func(now time.Time)) *Job {
+	if fn == nil {
+		// Ensure a function
+		panic("nil func")
+	}
+
+	j := &Job{ // set defaults
+		timing: emptytiming, // i.e. fire immediately
+		call:   fn,
+		panic:  func(i interface{}) { panic(i) },
+	}
+
+	// Init next time ptr
+	j.next.Store(zerotime)
+
+	return j
+}
+
+// At sets this Job to execute at time, by passing (*sched.Once)(&at) to .With(). See .With() for details.
+func (job *Job) At(at time.Time) *Job {
+	return job.With((*Once)(&at))
+}
+
+// Every sets this Job to execute every period, by passing sched.Period(period) to .With(). See .With() for details.
+func (job *Job) Every(period time.Duration) *Job {
+	return job.With(Periodic(period))
+}
+
+// EveryAt sets this Job to execute every period starting at time, by passing &PeriodicAt{once: Once(at), period: Periodic(period)} to .With(). See .With() for details.
+func (job *Job) EveryAt(at time.Time, period time.Duration) *Job {
+	return job.With(&PeriodicAt{Once: Once(at), Period: Periodic(period)})
+}
+
+// With sets this Job's timing to given implementation, or if already set will wrap existing using sched.TimingWrap{}.
+func (job *Job) With(t Timing) *Job {
+	if t == nil {
+		// Ensure a timing
+		panic("nil Timing")
+	}
+
+	if job.timing == emptytiming {
+		// Set new timing
+		job.timing = t
+	} else {
+		// Wrap old timing
+		old := job.timing
+		job.timing = &TimingWrap{
+			Outer: t,
+			Inner: old,
+		}
+	}
+
+	return job
+}
+
+// Panic specifics how this job handles panics, default is an actual panic.
+func (job *Job) Panic(fn func(interface{})) *Job {
+	if fn == nil {
+		// Ensure a function
+		panic("nil func")
+	}
+	job.panic = fn
+	return job
+}
+
+// Next returns the next time this Job is expected to run.
+func (job *Job) Next() time.Time {
+	return job.next.Load()
+}
+
+// Run will execute this Job and pass through given now time.
+func (job *Job) Run(now time.Time) {
+	defer func() {
+		if r := recover(); r != nil {
+			job.panic(r)
+		}
+	}()
+	job.call(now)
+}
diff --git a/vendor/codeberg.org/gruf/go-sched/scheduler.go b/vendor/codeberg.org/gruf/go-sched/scheduler.go
new file mode 100644
index 000000000..d017ddcf6
--- /dev/null
+++ b/vendor/codeberg.org/gruf/go-sched/scheduler.go
@@ -0,0 +1,240 @@
+package sched
+
+import (
+	"context"
+	"sort"
+	"time"
+
+	"codeberg.org/gruf/go-atomics"
+	"codeberg.org/gruf/go-runners"
+)
+
+var (
+	// neverticks is a timer channel that never ticks (it's starved).
+	neverticks = make(chan time.Time)
+
+	// alwaysticks is a timer channel that always ticks (it's closed).
+	alwaysticks = func() chan time.Time {
+		ch := make(chan time.Time)
+		close(ch)
+		return ch
+	}()
+)
+
+// Scheduler provides a means of running jobs at specific times and
+// regular intervals, all while sharing a single underlying timer.
+type Scheduler struct {
+	jobs []*Job           // jobs is a list of tracked Jobs to be executed
+	jch  chan interface{} // jch accepts either Jobs or job IDs to notify new/removed jobs
+	svc  runners.Service  // svc manages the main scheduler routine
+	jid  atomics.Uint64   // jid is used to iteratively generate unique IDs for jobs
+}
+
+// New returns a new Scheduler instance with given job change queue size.
+func NewScheduler(queue int) Scheduler {
+	if queue < 0 {
+		queue = 10
+	}
+	return Scheduler{jch: make(chan interface{}, queue)}
+}
+
+// Start will attempt to start the Scheduler. Immediately returns false if the Service is already running, and true after completed run.
+func (sch *Scheduler) Start() bool {
+	return sch.svc.Run(sch.run)
+}
+
+// Stop will attempt to stop the Scheduler. Immediately returns false if not running, and true only after Scheduler is fully stopped.
+func (sch *Scheduler) Stop() bool {
+	return sch.svc.Stop()
+}
+
+// Running will return whether Scheduler is running.
+func (sch *Scheduler) Running() bool {
+	return sch.svc.Running()
+}
+
+// Schedule will add provided Job to the Scheduler, returning a cancel function.
+func (sch *Scheduler) Schedule(job *Job) (cancel func()) {
+	if job == nil {
+		// Ensure there's a job!
+		panic("nil job")
+	}
+
+	// Get last known job ID
+	last := sch.jid.Load()
+
+	// Give this job an ID and check overflow
+	if job.id = sch.jid.Add(1); job.id < last {
+		panic("scheduler job id overflow")
+	}
+
+	// Pass job to scheduler
+	sch.jch <- job
+
+	// Return cancel function for job ID
+	return func() { sch.jch <- job.id }
+}
+
+// run is the main scheduler run routine, which runs for as long as ctx is valid.
+func (sch *Scheduler) run(ctx context.Context) {
+	var (
+		// timerset represents whether timer was running
+		// for a particular run of the loop. false means
+		// that tch == neverticks || tch == alwaysticks
+		timerset bool
+
+		// timer tick channel (or a never-tick channel)
+		tch <-chan time.Time
+
+		// timer notifies this main routine to wake when
+		// the job queued needs to be checked for executions
+		timer *time.Timer
+
+		// stopdrain will stop and drain the timer
+		// if it has been running (i.e. timerset == true)
+		stopdrain = func() {
+			if timerset && !timer.Stop() {
+				<-timer.C
+			}
+		}
+	)
+
+	for {
+		select {
+		// Handle received job/id
+		case v := <-sch.jch:
+			sch.handle(v)
+			continue
+
+		// No more
+		default:
+		}
+
+		// Done
+		break
+	}
+
+	// Create a stopped timer
+	timer = time.NewTimer(1)
+	<-timer.C
+
+	for {
+		// Reset timer state
+		timerset = false
+
+		if len(sch.jobs) > 0 {
+			// Sort jobs by next occurring
+			sort.Sort(byNext(sch.jobs))
+
+			// Get execution time
+			now := time.Now()
+
+			// Get next job time
+			next := sch.jobs[0].Next()
+
+			if until := next.Sub(now); until <= 0 {
+				// This job is behind schedule,
+				// set timer to always tick
+				tch = alwaysticks
+			} else {
+				// Reset timer to period
+				timer.Reset(until)
+				tch = timer.C
+				timerset = true
+			}
+		} else {
+			// Unset timer
+			tch = neverticks
+		}
+
+		select {
+		// Scheduler stopped
+		case <-ctx.Done():
+			stopdrain()
+			return
+
+		// Timer ticked, run scheduled
+		case now := <-tch:
+			sch.schedule(now)
+
+		// Received update, handle job/id
+		case v := <-sch.jch:
+			sch.handle(v)
+			stopdrain()
+		}
+	}
+}
+
+// handle takes an interfaces received from Scheduler.jch and handles either:
+// - Job --> new job to add.
+// - uint64 --> job ID to remove.
+func (sch *Scheduler) handle(v interface{}) {
+	switch v := v.(type) {
+	// New job added
+	case *Job:
+		// Get current time
+		now := time.Now()
+
+		// Update the next call time
+		next := v.timing.Next(now)
+		v.next.Store(next)
+
+		// Append this job to queued
+		sch.jobs = append(sch.jobs, v)
+
+	// Job removed
+	case uint64:
+		for i := 0; i < len(sch.jobs); i++ {
+			if sch.jobs[i].id == v {
+				// This is the job we're looking for! Drop this
+				sch.jobs = append(sch.jobs[:i], sch.jobs[i+1:]...)
+				return
+			}
+		}
+	}
+}
+
+// schedule will iterate through the scheduler jobs and execute those necessary, updating their next call time.
+func (sch *Scheduler) schedule(now time.Time) {
+	for i := 0; i < len(sch.jobs); {
+		// Scope our own var
+		job := sch.jobs[i]
+
+		// We know these jobs are ordered by .Next(), so as soon
+		// as we reach one with .Next() after now, we can return
+		if job.Next().After(now) {
+			return
+		}
+
+		// Update the next call time
+		next := job.timing.Next(now)
+		job.next.Store(next)
+
+		// Run this job async!
+		go job.Run(now)
+
+		if job.Next().IsZero() {
+			// Zero time, this job is done and can be dropped
+			sch.jobs = append(sch.jobs[:i], sch.jobs[i+1:]...)
+			continue
+		}
+
+		// Iter
+		i++
+	}
+}
+
+// byNext is an implementation of sort.Interface to sort Jobs by their .Next() time.
+type byNext []*Job
+
+func (by byNext) Len() int {
+	return len(by)
+}
+
+func (by byNext) Less(i int, j int) bool {
+	return by[i].Next().Before(by[j].Next())
+}
+
+func (by byNext) Swap(i int, j int) {
+	by[i], by[j] = by[j], by[i]
+}
diff --git a/vendor/codeberg.org/gruf/go-sched/timing.go b/vendor/codeberg.org/gruf/go-sched/timing.go
new file mode 100644
index 000000000..33c230fa5
--- /dev/null
+++ b/vendor/codeberg.org/gruf/go-sched/timing.go
@@ -0,0 +1,92 @@
+package sched
+
+import (
+	"time"
+)
+
+var (
+	// zerotime is zero time.Time (unix epoch).
+	zerotime = time.Time{}
+
+	// emptytiming is a global timingempty to check against.
+	emptytiming = timingempty{}
+)
+
+// Timing provides scheduling for a Job, determining the next time
+// for given current time that execution is required. Please note that
+// calls to .Next() may alter the results of the next call, and should
+// only be called by the Scheduler.
+type Timing interface {
+	Next(time.Time) time.Time
+}
+
+// timingempty is a 'zero' Timing implementation that always returns zero time.
+type timingempty struct{}
+
+func (timingempty) Next(time.Time) time.Time {
+	return zerotime
+}
+
+// Once implements Timing to provide a run-once Job execution.
+type Once time.Time
+
+func (o *Once) Next(time.Time) time.Time {
+	ret := *(*time.Time)(o)
+	*o = Once(zerotime) // reset
+	return ret
+}
+
+// Periodic implements Timing to provide a recurring Job execution.
+type Periodic time.Duration
+
+func (p Periodic) Next(now time.Time) time.Time {
+	return now.Add(time.Duration(p))
+}
+
+// PeriodicAt implements Timing to provide a recurring Job execution starting at 'Once' time.
+type PeriodicAt struct {
+	Once   Once
+	Period Periodic
+}
+
+func (p *PeriodicAt) Next(now time.Time) time.Time {
+	if next := p.Once.Next(now); !next.IsZero() {
+		return next
+	}
+	return p.Period.Next(now)
+}
+
+// TimingWrap allows combining two different Timing implementations.
+type TimingWrap struct {
+	Outer Timing
+	Inner Timing
+
+	// determined next times
+	outerNext time.Time
+	innerNext time.Time
+}
+
+func (t *TimingWrap) Next(now time.Time) time.Time {
+	if t.outerNext.IsZero() {
+		// Regenerate outermost next run time
+		t.outerNext = t.Outer.Next(now)
+	}
+
+	if t.innerNext.IsZero() {
+		// Regenerate innermost next run time
+		t.innerNext = t.Inner.Next(now)
+	}
+
+	// If outer comes before inner, return outer
+	if t.outerNext != zerotime &&
+		t.outerNext.Before(t.innerNext) {
+		next := t.outerNext
+		t.outerNext = zerotime
+		return next
+	}
+
+	// Else, return inner
+	next := t.innerNext
+	t.innerNext = zerotime
+	return next
+}