Initial alerting

wip

kind of working

wip

.gitignore

@@ -1,4 +1,5 @@
 config.yaml
+config.test.yaml
 mydb.sql
 client/calls/
 !client/calls/.gitkeep
@@ -6,3 +7,4 @@ client/calls/
 /calls
 Session.vim
 *.log
+*.dlv

internal/timeseries/timeseries.go

@@ -85,9 +85,11 @@ type Clock interface {
 	Now() time.Time
 }
 
-// defaultClock is used in case no clock is provided to the constructor.
+// DefaultClock is used in case no clock is provided to the constructor.
 type defaultClock struct{}
 
+var DefaultClock Clock = &defaultClock{}
+
 func (c *defaultClock) Now() time.Time {
 	return time.Now()
 }
@@ -137,7 +139,7 @@ func NewTimeSeries(os ...Option) (*TimeSeries, error) {
 		o(&opts)
 	}
 	if opts.clock == nil {
-		opts.clock = &defaultClock{}
+		opts.clock = DefaultClock
 	}
 	if opts.granularities == nil {
 		opts.granularities = defaultGranularities

internal/timeseries/timeseries_test.go

@@ -3,8 +3,6 @@ package timeseries
 import (
 	"testing"
 	"time"
-
-	"github.com/benbjohnson/clock"
 )
 
 // TODO: do table based testing

internal/trending/item.go

@@ -3,25 +3,27 @@ package trending
 import (
 	"math"
 	"time"
+
+	timeseries "dynatron.me/x/stillbox/internal/timeseries"
 )
 
-type item struct {
+type item[K comparable] struct {
 	eventSeries TimeSeries
 	maxSeries   SlidingWindow
 
 	max     float64
 	maxTime time.Time
-	options *options
+	options *options[K]
 
 	// TODO: move outside of item because it's the same for all items
 	defaultExpectation float64
 	defaultHourlyCount float64
 }
 
-func newItem(id string, options *options) *item {
+func newItem[K comparable](id K, options *options[K]) *item[K] {
 	defaultHourlyCount := float64(options.baseCount) * float64(options.storageDuration/time.Hour)
 	defaultExpectation := float64(options.baseCount) / float64(time.Hour/options.recentDuration)
-	return &item{
+	return &item[K]{
 		eventSeries: options.creator(id),
 		maxSeries:   options.slidingWindowCreator(id),
 		options:     options,
@@ -31,10 +33,10 @@ func newItem(id string, options *options) *item {
 	}
 }
 
-func (i *item) score() score {
+func (i *item[K]) score() Score[K] {
 	recentCount, count := i.computeCounts()
 	if recentCount < i.options.countThreshold {
-		return score{}
+		return Score[K]{}
 	}
 	if recentCount == count {
 		// we see this for the first time so there is no historical data
@@ -58,39 +60,41 @@ func (i *item) score() score {
 	}
 	i.decayMax()
 
-	mixedScore := 0.5 * (klScore + i.max)
+	mixedScore := 5 * (klScore + i.max)
 
-	return score{
+	return Score[K]{
 		Score:       mixedScore,
 		Probability: probability,
 		Expectation: expectation,
 		Maximum:     i.max,
 		KLScore:     klScore,
+		Count:       count,
+		RecentCount: recentCount,
 	}
 }
 
-func (i *item) computeCounts() (float64, float64) {
+func (i *item[K]) computeCounts() (float64, float64) {
-	now := time.Now()
+	now := timeseries.DefaultClock.Now()
 	totalCount, _ := i.eventSeries.Range(now.Add(-i.options.storageDuration), now)
 	count, _ := i.eventSeries.Range(now.Add(-i.options.recentDuration), now)
 	return count, totalCount
 }
 
-func (i *item) computeRecentMax() float64 {
+func (i *item[K]) computeRecentMax() float64 {
 	return i.maxSeries.Max()
 }
 
-func (i *item) decayMax() {
+func (i *item[K]) decayMax() {
 	i.updateMax(i.max * i.computeExponentialDecayMultiplier())
 }
 
-func (i *item) updateMax(score float64) {
+func (i *item[K]) updateMax(score float64) {
 	i.max = score
-	i.maxTime = time.Now()
+	i.maxTime = timeseries.DefaultClock.Now()
 }
 
-func (i *item) computeExponentialDecayMultiplier() float64 {
+func (i *item[K]) computeExponentialDecayMultiplier() float64 {
-	return math.Pow(0.5, float64(time.Now().Unix()-i.maxTime.Unix())/i.options.halfLife.Seconds())
+	return math.Pow(0.5, float64(timeseries.DefaultClock.Now().Unix()-i.maxTime.Unix())/i.options.halfLife.Seconds())
 }
 
 func computeKullbackLeibler(probability float64, expectation float64) float64 {

internal/trending/score.go

@@ -1,36 +1,38 @@
 package trending
 
-type score struct {
+type Score[K comparable] struct {
-	ID          string
+	ID          K
 	Score       float64
 	Probability float64
 	Expectation float64
 	Maximum     float64
 	KLScore     float64
+	Count       float64
+	RecentCount float64
 }
 
-type Scores []score
+type Scores[K comparable] []Score[K]
 
-func (s Scores) Len() int {
+func (s Scores[K]) Len() int {
 	return len(s)
 }
 
-func (s Scores) Swap(i, j int) {
+func (s Scores[K]) Swap(i, j int) {
 	s[i], s[j] = s[j], s[i]
 }
 
-func (s Scores) Less(i, j int) bool {
+func (s Scores[K]) Less(i, j int) bool {
 	return s[i].Score > s[j].Score
 }
 
-func (s Scores) take(count int) Scores {
+func (s Scores[K]) take(count int) Scores[K] {
 	if count >= len(s) {
 		return s
 	}
 	return s[0 : count-1]
 }
 
-func (s Scores) threshold(t float64) Scores {
+func (s Scores[K]) threshold(t float64) Scores[K] {
 	for i := range s {
 		if s[i].Score < t {
 			return s[0:i]

internal/trending/trending.go

@@ -25,13 +25,13 @@ var defaultHalfLife = 2 * time.Hour
 var defaultRecentDuration = 5 * time.Minute
 var defaultStorageDuration = 7 * 24 * time.Hour
 var defaultMaxResults = 100
-var defaultBaseCount = 3
+var defaultBaseCount = 1
 var defaultScoreThreshold = 0.01
 var defaultCountThreshold = 3.0
 
-type options struct {
+type options[K comparable] struct {
-	creator              TimeSeriesCreator
+	creator              TimeSeriesCreator[K]
-	slidingWindowCreator SlidingWindowCreator
+	slidingWindowCreator SlidingWindowCreator[K]
 
 	halfLife time.Duration
 
@@ -44,59 +44,59 @@ type options struct {
 	countThreshold float64
 }
 
-type Option func(*options)
+type Option[K comparable] func(*options[K])
 
-func WithTimeSeries(creator TimeSeriesCreator) Option {
+func WithTimeSeries[K comparable](creator TimeSeriesCreator[K]) Option[K] {
-	return func(o *options) {
+	return func(o *options[K]) {
 		o.creator = creator
 	}
 }
 
-func WithSlidingWindow(creator SlidingWindowCreator) Option {
+func WithSlidingWindow[K comparable](creator SlidingWindowCreator[K]) Option[K] {
-	return func(o *options) {
+	return func(o *options[K]) {
 		o.slidingWindowCreator = creator
 	}
 }
 
-func WithHalfLife(halfLife time.Duration) Option {
+func WithHalfLife[K comparable](halfLife time.Duration) Option[K] {
-	return func(o *options) {
+	return func(o *options[K]) {
 		o.halfLife = halfLife
 	}
 }
 
-func WithRecentDuration(recentDuration time.Duration) Option {
+func WithRecentDuration[K comparable](recentDuration time.Duration) Option[K] {
-	return func(o *options) {
+	return func(o *options[K]) {
 		o.recentDuration = recentDuration
 	}
 }
 
-func WithStorageDuration(storageDuration time.Duration) Option {
+func WithStorageDuration[K comparable](storageDuration time.Duration) Option[K] {
-	return func(o *options) {
+	return func(o *options[K]) {
 		o.storageDuration = storageDuration
 	}
 }
 
-func WithMaxResults(maxResults int) Option {
+func WithMaxResults[K comparable](maxResults int) Option[K] {
-	return func(o *options) {
+	return func(o *options[K]) {
 		o.maxResults = maxResults
 	}
 }
 
-func WithScoreThreshold(threshold float64) Option {
+func WithScoreThreshold[K comparable](threshold float64) Option[K] {
-	return func(o *options) {
+	return func(o *options[K]) {
 		o.scoreThreshold = threshold
 	}
 }
 
-func WithCountThreshold(threshold float64) Option {
+func WithCountThreshold[K comparable](threshold float64) Option[K] {
-	return func(o *options) {
+	return func(o *options[K]) {
 		o.countThreshold = threshold
 	}
 }
 
-type Scorer struct {
+type Scorer[K comparable] struct {
-	options options
+	options options[K]
-	items   map[string]*item
+	items   map[K]*item[K]
 }
 
 type SlidingWindow interface {
@@ -104,16 +104,16 @@ type SlidingWindow interface {
 	Max() float64
 }
 
-type SlidingWindowCreator func(string) SlidingWindow
+type SlidingWindowCreator[K comparable] func(K) SlidingWindow
 
 type TimeSeries interface {
 	IncreaseAtTime(amount int, time time.Time)
 	Range(start, end time.Time) (float64, error)
 }
 
-type TimeSeriesCreator func(string) TimeSeries
+type TimeSeriesCreator[K comparable] func(K) TimeSeries
 
-func NewMemoryTimeSeries(id string) TimeSeries {
+func NewMemoryTimeSeries[K comparable](id K) TimeSeries {
 	ts, _ := timeseries.NewTimeSeries(timeseries.WithGranularities(
 		[]timeseries.Granularity{
 			{Granularity: time.Second, Count: 60},
@@ -125,13 +125,13 @@ func NewMemoryTimeSeries(id string) TimeSeries {
 	return ts
 }
 
-func NewScorer(options ...Option) Scorer {
+func NewScorer[K comparable](options ...Option[K]) Scorer[K] {
-	scorer := Scorer{items: make(map[string]*item)}
+	scorer := Scorer[K]{items: make(map[K]*item[K])}
 	for _, o := range options {
 		o(&scorer.options)
 	}
 	if scorer.options.creator == nil {
-		scorer.options.creator = NewMemoryTimeSeries
+		scorer.options.creator = NewMemoryTimeSeries[K]
 	}
 	if scorer.options.halfLife == 0 {
 		scorer.options.halfLife = defaultHalfLife
@@ -155,7 +155,7 @@ func NewScorer(options ...Option) Scorer {
 		scorer.options.baseCount = defaultBaseCount
 	}
 	if scorer.options.slidingWindowCreator == nil {
-		scorer.options.slidingWindowCreator = func(id string) SlidingWindow {
+		scorer.options.slidingWindowCreator = func(id K) SlidingWindow {
 			return slidingwindow.NewSlidingWindow(
 				slidingwindow.WithStep(time.Hour*24),
 				slidingwindow.WithDuration(scorer.options.storageDuration),
@@ -165,7 +165,7 @@ func NewScorer(options ...Option) Scorer {
 	return scorer
 }
 
-func (s *Scorer) AddEvent(id string, time time.Time) {
+func (s *Scorer[K]) AddEvent(id K, time time.Time) {
 	item := s.items[id]
 	if item == nil {
 		item = newItem(id, &s.options)
@@ -174,12 +174,12 @@ func (s *Scorer) AddEvent(id string, time time.Time) {
 	s.addToItem(item, time)
 }
 
-func (s *Scorer) addToItem(item *item, time time.Time) {
+func (s *Scorer[K]) addToItem(item *item[K], tm time.Time) {
-	item.eventSeries.IncreaseAtTime(1, time)
+	item.eventSeries.IncreaseAtTime(1, tm)
 }
 
-func (s *Scorer) Score() Scores {
+func (s *Scorer[K]) Score() Scores[K] {
-	var scores Scores
+	var scores Scores[K]
 	for id, item := range s.items {
 		score := item.score()
 		score.ID = id

pkg/gordio/alerting/alerting.go

@@ -0,0 +1,148 @@
+package alerting
+
+import (
+	"context"
+	"sort"
+	"sync"
+	"time"
+
+	cl "dynatron.me/x/stillbox/pkg/calls"
+	"dynatron.me/x/stillbox/pkg/gordio/database"
+
+	"dynatron.me/x/stillbox/internal/timeseries"
+	"dynatron.me/x/stillbox/internal/trending"
+
+	"github.com/rs/zerolog/log"
+)
+
+const (
+	StorageLookbackDays = 7
+	HalfLife            = 30 * time.Minute
+	RecentDuration      = 2 * time.Hour
+	ScoreThreshold      = -1
+	CountThreshold      = 1.0
+	AlerterTickInterval = time.Minute
+)
+
+type Alerter struct {
+	sync.RWMutex
+	scorer    trending.Scorer[cl.Talkgroup]
+	scores    trending.Scores[cl.Talkgroup]
+	lastScore time.Time
+}
+
+type myClock struct {
+	offset time.Duration
+}
+
+func (c *myClock) Now() time.Time {
+	return time.Now().Add(c.offset)
+}
+
+func New() *Alerter {
+	as := &Alerter{
+		scorer: trending.NewScorer[cl.Talkgroup](
+			trending.WithTimeSeries(newTimeSeries),
+			trending.WithStorageDuration[cl.Talkgroup](StorageLookbackDays*24*time.Hour),
+			trending.WithRecentDuration[cl.Talkgroup](RecentDuration),
+			trending.WithHalfLife[cl.Talkgroup](HalfLife),
+			trending.WithScoreThreshold[cl.Talkgroup](ScoreThreshold),
+			trending.WithCountThreshold[cl.Talkgroup](CountThreshold),
+		),
+	}
+
+	return as
+}
+
+func (as *Alerter) Go(ctx context.Context) {
+	as.startBackfill(ctx)
+
+	as.score(ctx, time.Now())
+	ticker := time.NewTicker(AlerterTickInterval)
+
+	for {
+		select {
+		case now := <-ticker.C:
+			as.score(ctx, now)
+		case <-ctx.Done():
+			ticker.Stop()
+			return
+		}
+	}
+
+}
+
+func newTimeSeries(id cl.Talkgroup) trending.TimeSeries {
+	ts, _ := timeseries.NewTimeSeries(timeseries.WithGranularities(
+		[]timeseries.Granularity{
+			{Granularity: time.Second, Count: 60},
+			{Granularity: time.Minute, Count: 10},
+			{Granularity: time.Hour, Count: 24},
+			{Granularity: time.Hour * 24, Count: StorageLookbackDays},
+		},
+	))
+	return ts
+}
+
+func (as *Alerter) startBackfill(ctx context.Context) {
+	now := time.Now()
+	since := now.Add(StorageLookbackDays * -24 * time.Hour)
+	log.Debug().Time("since", since).Msg("starting stats backfill")
+	count, err := as.backfill(ctx, since)
+	if err != nil {
+		log.Error().Err(err).Msg("backfill failed")
+		return
+	}
+	log.Debug().Int("count", count).Str("in", time.Now().Sub(now).String()).Int("len", as.scorer.Score().Len()).Msg("backfill finished")
+}
+
+func (as *Alerter) score(ctx context.Context, now time.Time) {
+	as.Lock()
+	defer as.Unlock()
+
+	as.scores = as.scorer.Score()
+	as.lastScore = now
+	sort.Sort(as.scores)
+}
+
+func (as *Alerter) backfill(ctx context.Context, since time.Time) (count int, err error) {
+	db := database.FromCtx(ctx)
+	const backfillStatsQuery = `SELECT system, talkgroup, call_date FROM calls WHERE call_date > $1 AND call_date < $2`
+
+	rows, err := db.Query(ctx, backfillStatsQuery, since, timeseries.DefaultClock.Now())
+	if err != nil {
+		return count, err
+	}
+	defer rows.Close()
+
+	as.Lock()
+	defer as.Unlock()
+
+	for rows.Next() {
+		var tg cl.Talkgroup
+		var callDate time.Time
+		if err := rows.Scan(&tg.System, &tg.Talkgroup, &callDate); err != nil {
+			return count, err
+		}
+		as.scorer.AddEvent(tg, callDate)
+		count++
+	}
+
+	if err := rows.Err(); err != nil {
+		return count, err
+	}
+
+	return count, nil
+}
+
+func (as *Alerter) SinkType() string {
+	return "alerting"
+}
+
+func (as *Alerter) Call(ctx context.Context, call *cl.Call) error {
+	as.Lock()
+	defer as.Unlock()
+	as.scorer.AddEvent(call.TalkgroupTuple(), call.DateTime)
+
+	return nil
+}

pkg/gordio/alerting/stats.go

@@ -0,0 +1,71 @@
+package alerting
+
+import (
+	_ "embed"
+	"fmt"
+	"html/template"
+	"net/http"
+	"time"
+
+	"dynatron.me/x/stillbox/pkg/calls"
+	"dynatron.me/x/stillbox/pkg/gordio/database"
+
+	"dynatron.me/x/stillbox/internal/trending"
+
+	"github.com/go-chi/chi/v5"
+	"github.com/rs/zerolog/log"
+)
+
+//go:embed stats.html
+var statsTemplateFile string
+
+var (
+	funcMap = template.FuncMap{
+		"f": func(v float64) string {
+			return fmt.Sprintf("%.4f", v)
+		},
+	}
+	statTmpl = template.Must(template.New("stats").Funcs(funcMap).Parse(statsTemplateFile))
+)
+
+func (as *Alerter) PrivateRoutes(r chi.Router) {
+	r.Get("/tgstats", as.tgStats)
+}
+
+func (as *Alerter) tgStats(w http.ResponseWriter, r *http.Request) {
+	ctx := r.Context()
+	db := database.FromCtx(ctx)
+
+	packed := make([]int64, 0, len(as.scores))
+	for _, s := range as.scores {
+		packed = append(packed, s.ID.Pack())
+	}
+
+	tgs, err := db.GetTalkgroupsByPackedIDs(ctx, packed)
+	if err != nil {
+		log.Error().Err(err).Msg("stats TG get failed")
+		http.Error(w, err.Error(), http.StatusInternalServerError)
+		return
+	}
+
+	tgMap := make(map[calls.Talkgroup]database.GetTalkgroupsByPackedIDsRow, len(tgs))
+	for _, t := range tgs {
+		tgMap[calls.Talkgroup{System: uint32(t.SystemID), Talkgroup: uint32(t.ID)}] = t
+	}
+
+	renderData := struct {
+		TGs       map[calls.Talkgroup]database.GetTalkgroupsByPackedIDsRow
+		Scores    trending.Scores[calls.Talkgroup]
+		LastScore time.Time
+	}{
+		TGs:       tgMap,
+		Scores:    as.scores,
+		LastScore: as.lastScore,
+	}
+
+	w.WriteHeader(http.StatusOK)
+	err = statTmpl.Execute(w, renderData)
+	if err != nil {
+		log.Error().Err(err).Msg("stat template exec")
+	}
+}

pkg/gordio/alerting/stats.html

@@ -0,0 +1,86 @@
+<!DOCTYPE html>
+<html>
+  <head>
+    <title>Stats</title>
+    <style>
+      *, *:before, *:after {
+        margin: 0;
+        padding: 0;
+        box-sizing: border-box;
+      }
+      body {
+        background: #105469;
+        font-family: sans-serif;
+      }
+      table {
+        background: #012B39;
+        border-radius: 0.25em;
+        border-collapse: collapse;
+        margin-top: 1em;
+        margin-bottom: 1em;
+        margin-left: auto;
+        margin-right: auto;
+      }
+      th {
+        border-bottom: 1px solid #364043;
+        color: #E2B842;
+        font-size: 0.85em;
+        font-weight: 600;
+        padding: 0.5em 1em;
+        text-align: left;
+      }
+      td {
+        color: #fff;
+        font-weight: 400;
+        padding: 0.65em 1em;
+      }
+      .disabled td {
+        color: #4F5F64;
+      }
+      tbody tr {
+        transition: background 0.25s ease;
+      }
+      tbody tr:hover {
+        background: #014055;
+      }
+    </style>
+  </head>
+  <body>
+    <table>
+      <tr>
+        <th>System</th>
+        <th>TG</th>
+        <th>TG ID</th>
+	<th>Count</th>
+	<th>Recent</th>
+        <th>Score</th>
+        <th>Probab</th>
+        <th>Expect</th>
+        <th>Max</th>
+        <th>KL</th>
+      </tr>
+      {{ range .Scores }}
+      {{ $tg := (index $.TGs .ID) }}
+      <tr>
+	<td>{{ $tg.Name_2}}</td>
+	<td>{{ $tg.Name}}</td>
+	<td>{{ .ID.Talkgroup }}</td>
+	<td>{{ .Count }}</td>
+	<td>{{ f .RecentCount }}</td>
+	<td>{{ f .Score }}</td>
+	<td>{{ f .Probability }}</td>
+	<td>{{ f .Expectation }}</td>
+	<td>{{ f .Maximum }}</td>
+	<td>{{ f .KLScore }}</td>
+      </tr>
+      {{else}}
+      <tr>
+        <td colspan="10">No Data</td>
+      </tr>
+      {{ end }}
+      <tr>
+      <td colspan="10">Last updated at {{ .LastScore }}</td>
+      </tr>
+    </table>
+  </body>
+</html>

pkg/gordio/database/calls.sql.go

@@ -7,9 +7,9 @@ package database
 
 import (
 	"context"
-	"time"
 
 	"github.com/google/uuid"
+	"github.com/jackc/pgx/v5/pgtype"
 )
 
 const addCall = `-- name: AddCall :one
@@ -39,7 +39,7 @@ type AddCallParams struct {
 	Submitter   *int32             `json:"submitter"`
 	System      int                `json:"system"`
 	Talkgroup   int                `json:"talkgroup"`
-	CallDate    time.Time `json:"call_date"`
+	CallDate    pgtype.Timestamptz `json:"call_date"`
 	AudioName   *string            `json:"audio_name"`
 	AudioBlob   []byte             `json:"audio_blob"`
 	AudioType   *string            `json:"audio_type"`

pkg/gordio/database/database.go

@@ -13,8 +13,6 @@ import (
 	"github.com/jackc/pgx/v5/pgxpool"
 )
 
-// This file will eventually turn into a postgres driver.
-
 // DB is a database handle.
 type DB struct {
 	*pgxpool.Pool
@@ -40,7 +38,12 @@ func NewClient(ctx context.Context, conf config.DB) (*DB, error) {
 
 	m.Close()
 
-	pool, err := pgxpool.New(ctx, conf.Connect)
+	pgConf, err := pgxpool.ParseConfig(conf.Connect)
+	if err != nil {
+		return nil, err
+	}
+
+	pool, err := pgxpool.NewWithConfig(ctx, pgConf)
 	if err != nil {
 		return nil, err
 	}

pkg/gordio/database/models.go

@@ -25,7 +25,7 @@ type Call struct {
 	Submitter   *int32             `json:"submitter"`
 	System      int                `json:"system"`
 	Talkgroup   int                `json:"talkgroup"`
-	CallDate    time.Time `json:"call_date"`
+	CallDate    pgtype.Timestamptz `json:"call_date"`
 	AudioName   *string            `json:"audio_name"`
 	AudioBlob   []byte             `json:"audio_blob"`
 	Duration    *int32             `json:"duration"`

pkg/gordio/database/querier.go

@@ -24,6 +24,7 @@ type Querier interface {
 	GetTalkgroupIDsByTags(ctx context.Context, anytags []string, alltags []string, nottags []string) ([]GetTalkgroupIDsByTagsRow, error)
 	GetTalkgroupTags(ctx context.Context, sys int, tg int) ([]string, error)
 	GetTalkgroupWithLearned(ctx context.Context, systemID int, tgid int) (GetTalkgroupWithLearnedRow, error)
+	GetTalkgroupsByPackedIDs(ctx context.Context, dollar_1 []int64) ([]GetTalkgroupsByPackedIDsRow, error)
 	GetTalkgroupsWithAllTags(ctx context.Context, tags []string) ([]Talkgroup, error)
 	GetTalkgroupsWithAnyTags(ctx context.Context, tags []string) ([]Talkgroup, error)
 	GetUserByID(ctx context.Context, id int32) (User, error)

pkg/gordio/database/talkgroups.sql.go

@@ -137,6 +137,58 @@ func (q *Queries) GetTalkgroupWithLearned(ctx context.Context, systemID int, tgi
 	return i, err
 }
 
+const getTalkgroupsByPackedIDs = `-- name: GetTalkgroupsByPackedIDs :many
+SELECT tg.id, system_id, tgid, tg.name, alpha_tag, tg_group, frequency, metadata, tags, sys.id, sys.name FROM talkgroups tg
+JOIN systems sys ON tg.system_id = sys.id
+WHERE tg.id = ANY($1::INT8[])
+`
+
+type GetTalkgroupsByPackedIDsRow struct {
+	ID        int64    `json:"id"`
+	SystemID  int32    `json:"system_id"`
+	Tgid      int32    `json:"tgid"`
+	Name      *string  `json:"name"`
+	AlphaTag  *string  `json:"alpha_tag"`
+	TgGroup   *string  `json:"tg_group"`
+	Frequency *int32   `json:"frequency"`
+	Metadata  []byte   `json:"metadata"`
+	Tags      []string `json:"tags"`
+	ID_2      int      `json:"id_2"`
+	Name_2    string   `json:"name_2"`
+}
+
+func (q *Queries) GetTalkgroupsByPackedIDs(ctx context.Context, dollar_1 []int64) ([]GetTalkgroupsByPackedIDsRow, error) {
+	rows, err := q.db.Query(ctx, getTalkgroupsByPackedIDs, dollar_1)
+	if err != nil {
+		return nil, err
+	}
+	defer rows.Close()
+	var items []GetTalkgroupsByPackedIDsRow
+	for rows.Next() {
+		var i GetTalkgroupsByPackedIDsRow
+		if err := rows.Scan(
+			&i.ID,
+			&i.SystemID,
+			&i.Tgid,
+			&i.Name,
+			&i.AlphaTag,
+			&i.TgGroup,
+			&i.Frequency,
+			&i.Metadata,
+			&i.Tags,
+			&i.ID_2,
+			&i.Name_2,
+		); err != nil {
+			return nil, err
+		}
+		items = append(items, i)
+	}
+	if err := rows.Err(); err != nil {
+		return nil, err
+	}
+	return items, nil
+}
+
 const getTalkgroupsWithAllTags = `-- name: GetTalkgroupsWithAllTags :many
 SELECT id, system_id, tgid, name, alpha_tag, tg_group, frequency, metadata, tags FROM talkgroups
 WHERE tags && ARRAY[$1]

pkg/gordio/server/routes.go

@@ -33,6 +33,7 @@ func (s *Server) setupRoutes() {
 		r.Use(s.auth.VerifyMiddleware(), s.auth.AuthMiddleware())
 		s.nex.PrivateRoutes(r)
 		s.auth.PrivateRoutes(r)
+		s.alerter.PrivateRoutes(r)
 	})
 
 	r.Group(func(r chi.Router) {

pkg/gordio/server/server.go

@@ -6,6 +6,7 @@ import (
 	"os"
 	"time"
 
+	"dynatron.me/x/stillbox/pkg/gordio/alerting"
 	"dynatron.me/x/stillbox/pkg/gordio/auth"
 	"dynatron.me/x/stillbox/pkg/gordio/config"
 	"dynatron.me/x/stillbox/pkg/gordio/database"
@@ -29,6 +30,7 @@ type Server struct {
 	sinks   sinks.Sinks
 	nex     *nexus.Nexus
 	logger  *Logger
+	alerter *alerting.Alerter
 	hup     chan os.Signal
 }
 
@@ -43,6 +45,8 @@ func New(ctx context.Context, cfg *config.Config) (*Server, error) {
 		return nil, err
 	}
 
+	ctx = database.CtxWithDB(ctx, db)
+
 	r := chi.NewRouter()
 	authenticator := auth.NewAuthenticator(cfg.Auth)
 	srv := &Server{
@@ -52,10 +56,12 @@ func New(ctx context.Context, cfg *config.Config) (*Server, error) {
 		r:       r,
 		nex:     nexus.New(),
 		logger:  logger,
+		alerter: alerting.New(),
 	}
 
 	srv.sinks.Register("database", sinks.NewDatabaseSink(srv.db), true)
 	srv.sinks.Register("nexus", sinks.NewNexusSink(srv.nex), false)
+	srv.sinks.Register("alerting", sinks.NewAlerterSink(srv.alerter), false)
 	srv.sources.Register("rdio-http", sources.NewRdioHTTP(authenticator, srv))
 
 	r.Use(middleware.RequestID)
@@ -88,6 +94,7 @@ func (s *Server) Go(ctx context.Context) error {
 	}
 
 	go s.nex.Go(ctx)
+	go s.alerter.Go(ctx)
 
 	var err error
 	go func() {
@@ -95,6 +102,8 @@ func (s *Server) Go(ctx context.Context) error {
 	}()
 	<-ctx.Done()
 
+	s.sinks.Shutdown()
+
 	ctxShutdown, cancel := context.WithTimeout(context.Background(), shutdownTimeout)
 	defer cancel()
 

pkg/gordio/sinks/alerter.go

@@ -0,0 +1,9 @@
+package sinks
+
+import (
+	"dynatron.me/x/stillbox/pkg/gordio/alerting"
+)
+
+func NewAlerterSink(a *alerting.Alerter) *alerting.Alerter {
+	return a
+}

pkg/gordio/sinks/database.go

@@ -8,6 +8,7 @@ import (
 	"dynatron.me/x/stillbox/pkg/calls"
 	"dynatron.me/x/stillbox/pkg/gordio/database"
 
+	"github.com/jackc/pgx/v5/pgtype"
 	"github.com/rs/zerolog/log"
 )
 
@@ -24,6 +25,7 @@ func (s *DatabaseSink) Call(ctx context.Context, call *calls.Call) error {
 		log.Debug().Str("call", call.String()).Msg("received dontStore call")
 		return nil
 	}
+
 	dbCall, err := s.db.AddCall(ctx, s.toAddCallParams(call))
 	if err != nil {
 		return fmt.Errorf("add call: %w", err)
@@ -43,7 +45,7 @@ func (s *DatabaseSink) toAddCallParams(call *calls.Call) database.AddCallParams
 		Submitter:   call.Submitter.Int32Ptr(),
 		System:      call.System,
 		Talkgroup:   call.Talkgroup,
-		CallDate:    call.DateTime,
+		CallDate:    pgtype.Timestamptz{Time: call.DateTime, Valid: true},
 		AudioName:   common.PtrOrNull(call.AudioName),
 		AudioBlob:   call.Audio,
 		AudioType:   common.PtrOrNull(call.AudioType),

pkg/gordio/sinks/sinks.go

@@ -40,6 +40,13 @@ func (s *Sinks) Register(name string, toAdd Sink, required bool) {
 	})
 }
 
+func (s *Sinks) Shutdown() {
+	s.Lock()
+	defer s.Unlock()
+
+	s.sinks = nil
+}
+
 func (s *Sinks) EmitCall(ctx context.Context, call *calls.Call) error {
 	s.Lock()
 	defer s.Unlock()

sql/postgres/migrations/001_initial.up.sql

@@ -91,7 +91,7 @@ CREATE TABLE IF NOT EXISTS calls(
 	submitter INTEGER REFERENCES api_keys(id) ON DELETE SET NULL,
 	system INTEGER NOT NULL,
 	talkgroup INTEGER NOT NULL,
-	call_date TIMESTAMP NOT NULL,
+	call_date TIMESTAMPTZ NOT NULL,
 	audio_name TEXT,
 	audio_blob BYTEA,
 	duration INTEGER,
@@ -107,7 +107,6 @@ CREATE TABLE IF NOT EXISTS calls(
 	transcript TEXT
 );
 
-
 CREATE OR REPLACE TRIGGER learn_tg AFTER INSERT ON calls
 FOR EACH ROW EXECUTE FUNCTION learn_talkgroup();
 

sql/postgres/queries/talkgroups.sql

@@ -28,6 +28,11 @@ WHERE id = ANY($1);
 SELECT * FROM talkgroups
 WHERE id = systg2id(sqlc.arg(system_id), sqlc.arg(tgid));
 
+-- name: GetTalkgroupsByPackedIDs :many
+SELECT * FROM talkgroups tg
+JOIN systems sys ON tg.system_id = sys.id
+WHERE tg.id = ANY($1::INT8[]);
+
 -- name: GetTalkgroupWithLearned :one
 SELECT
 tg.id, tg.system_id, sys.name system_name, tg.tgid, tg.name,