diff --git a/.gitignore b/.gitignore index a37f180..0223701 100644 --- a/.gitignore +++ b/.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 diff --git a/internal/timeseries/timeseries.go b/internal/timeseries/timeseries.go index 889cec5..a0a8747 100644 --- a/internal/timeseries/timeseries.go +++ b/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 diff --git a/internal/timeseries/timeseries_test.go b/internal/timeseries/timeseries_test.go index 2d0f745..8d87c30 100644 --- a/internal/timeseries/timeseries_test.go +++ b/internal/timeseries/timeseries_test.go @@ -3,8 +3,6 @@ package timeseries import ( "testing" "time" - - "github.com/benbjohnson/clock" ) // TODO: do table based testing diff --git a/internal/trending/item.go b/internal/trending/item.go index 534ef41..625a943 100644 --- a/internal/trending/item.go +++ b/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) { - now := time.Now() +func (i *item[K]) computeCounts() (float64, float64) { + 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 { - return math.Pow(0.5, float64(time.Now().Unix()-i.maxTime.Unix())/i.options.halfLife.Seconds()) +func (i *item[K]) computeExponentialDecayMultiplier() float64 { + return math.Pow(0.5, float64(timeseries.DefaultClock.Now().Unix()-i.maxTime.Unix())/i.options.halfLife.Seconds()) } func computeKullbackLeibler(probability float64, expectation float64) float64 { diff --git a/internal/trending/score.go b/internal/trending/score.go index 22e5f3a..f59368b 100644 --- a/internal/trending/score.go +++ b/internal/trending/score.go @@ -1,36 +1,38 @@ package trending -type score struct { - ID string +type Score[K comparable] struct { + 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] diff --git a/internal/trending/trending.go b/internal/trending/trending.go index 6117872..20e1df2 100644 --- a/internal/trending/trending.go +++ b/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 { - creator TimeSeriesCreator - slidingWindowCreator SlidingWindowCreator +type options[K comparable] struct { + creator TimeSeriesCreator[K] + 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 { - return func(o *options) { +func WithTimeSeries[K comparable](creator TimeSeriesCreator[K]) Option[K] { + return func(o *options[K]) { o.creator = creator } } -func WithSlidingWindow(creator SlidingWindowCreator) Option { - return func(o *options) { +func WithSlidingWindow[K comparable](creator SlidingWindowCreator[K]) Option[K] { + return func(o *options[K]) { o.slidingWindowCreator = creator } } -func WithHalfLife(halfLife time.Duration) Option { - return func(o *options) { +func WithHalfLife[K comparable](halfLife time.Duration) Option[K] { + return func(o *options[K]) { o.halfLife = halfLife } } -func WithRecentDuration(recentDuration time.Duration) Option { - return func(o *options) { +func WithRecentDuration[K comparable](recentDuration time.Duration) Option[K] { + return func(o *options[K]) { o.recentDuration = recentDuration } } -func WithStorageDuration(storageDuration time.Duration) Option { - return func(o *options) { +func WithStorageDuration[K comparable](storageDuration time.Duration) Option[K] { + return func(o *options[K]) { o.storageDuration = storageDuration } } -func WithMaxResults(maxResults int) Option { - return func(o *options) { +func WithMaxResults[K comparable](maxResults int) Option[K] { + return func(o *options[K]) { o.maxResults = maxResults } } -func WithScoreThreshold(threshold float64) Option { - return func(o *options) { +func WithScoreThreshold[K comparable](threshold float64) Option[K] { + return func(o *options[K]) { o.scoreThreshold = threshold } } -func WithCountThreshold(threshold float64) Option { - return func(o *options) { +func WithCountThreshold[K comparable](threshold float64) Option[K] { + return func(o *options[K]) { o.countThreshold = threshold } } -type Scorer struct { - options options - items map[string]*item +type Scorer[K comparable] struct { + options options[K] + 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 { - scorer := Scorer{items: make(map[string]*item)} +func NewScorer[K comparable](options ...Option[K]) Scorer[K] { + 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) { - item.eventSeries.IncreaseAtTime(1, time) +func (s *Scorer[K]) addToItem(item *item[K], tm time.Time) { + item.eventSeries.IncreaseAtTime(1, tm) } -func (s *Scorer) Score() Scores { - var scores Scores +func (s *Scorer[K]) Score() Scores[K] { + var scores Scores[K] for id, item := range s.items { score := item.score() score.ID = id diff --git a/pkg/gordio/alerting/alerting.go b/pkg/gordio/alerting/alerting.go new file mode 100644 index 0000000..cb5192a --- /dev/null +++ b/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 +} diff --git a/pkg/gordio/alerting/stats.go b/pkg/gordio/alerting/stats.go new file mode 100644 index 0000000..9e1fafa --- /dev/null +++ b/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") + } +} diff --git a/pkg/gordio/alerting/stats.html b/pkg/gordio/alerting/stats.html new file mode 100644 index 0000000..61e4e56 --- /dev/null +++ b/pkg/gordio/alerting/stats.html @@ -0,0 +1,86 @@ + + + + Stats + + + + + + + + + + + + + + + + + {{ range .Scores }} + {{ $tg := (index $.TGs .ID) }} + + + + + + + + + + + + + {{else}} + + + + {{ end }} + + + +
SystemTGTG IDCountRecentScoreProbabExpectMaxKL
{{ $tg.Name_2}}{{ $tg.Name}}{{ .ID.Talkgroup }}{{ .Count }}{{ f .RecentCount }}{{ f .Score }}{{ f .Probability }}{{ f .Expectation }}{{ f .Maximum }}{{ f .KLScore }}
No Data
Last updated at {{ .LastScore }}
+ + diff --git a/pkg/gordio/database/calls.sql.go b/pkg/gordio/database/calls.sql.go index 8c0301a..9089bd9 100644 --- a/pkg/gordio/database/calls.sql.go +++ b/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 @@ -36,22 +36,22 @@ RETURNING id ` type AddCallParams struct { - Submitter *int32 `json:"submitter"` - System int `json:"system"` - Talkgroup int `json:"talkgroup"` - CallDate time.Time `json:"call_date"` - AudioName *string `json:"audio_name"` - AudioBlob []byte `json:"audio_blob"` - AudioType *string `json:"audio_type"` - AudioUrl *string `json:"audio_url"` - Duration *int32 `json:"duration"` - Frequency int `json:"frequency"` - Frequencies []int `json:"frequencies"` - Patches []int `json:"patches"` - TgLabel *string `json:"tg_label"` - TgAlphaTag *string `json:"tg_alpha_tag"` - TgGroup *string `json:"tg_group"` - Source int `json:"source"` + Submitter *int32 `json:"submitter"` + System int `json:"system"` + Talkgroup int `json:"talkgroup"` + CallDate pgtype.Timestamptz `json:"call_date"` + AudioName *string `json:"audio_name"` + AudioBlob []byte `json:"audio_blob"` + AudioType *string `json:"audio_type"` + AudioUrl *string `json:"audio_url"` + Duration *int32 `json:"duration"` + Frequency int `json:"frequency"` + Frequencies []int `json:"frequencies"` + Patches []int `json:"patches"` + TgLabel *string `json:"tg_label"` + TgAlphaTag *string `json:"tg_alpha_tag"` + TgGroup *string `json:"tg_group"` + Source int `json:"source"` } func (q *Queries) AddCall(ctx context.Context, arg AddCallParams) (uuid.UUID, error) { diff --git a/pkg/gordio/database/database.go b/pkg/gordio/database/database.go index 8aee6fd..591f02c 100644 --- a/pkg/gordio/database/database.go +++ b/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 } diff --git a/pkg/gordio/database/models.go b/pkg/gordio/database/models.go index 6683e3a..25df383 100644 --- a/pkg/gordio/database/models.go +++ b/pkg/gordio/database/models.go @@ -21,24 +21,24 @@ type ApiKey struct { } type Call struct { - ID uuid.UUID `json:"id"` - Submitter *int32 `json:"submitter"` - System int `json:"system"` - Talkgroup int `json:"talkgroup"` - CallDate time.Time `json:"call_date"` - AudioName *string `json:"audio_name"` - AudioBlob []byte `json:"audio_blob"` - Duration *int32 `json:"duration"` - AudioType *string `json:"audio_type"` - AudioUrl *string `json:"audio_url"` - Frequency int `json:"frequency"` - Frequencies []int `json:"frequencies"` - Patches []int `json:"patches"` - TgLabel *string `json:"tg_label"` - TgAlphaTag *string `json:"tg_alpha_tag"` - TgGroup *string `json:"tg_group"` - Source int `json:"source"` - Transcript *string `json:"transcript"` + ID uuid.UUID `json:"id"` + Submitter *int32 `json:"submitter"` + System int `json:"system"` + Talkgroup int `json:"talkgroup"` + CallDate pgtype.Timestamptz `json:"call_date"` + AudioName *string `json:"audio_name"` + AudioBlob []byte `json:"audio_blob"` + Duration *int32 `json:"duration"` + AudioType *string `json:"audio_type"` + AudioUrl *string `json:"audio_url"` + Frequency int `json:"frequency"` + Frequencies []int `json:"frequencies"` + Patches []int `json:"patches"` + TgLabel *string `json:"tg_label"` + TgAlphaTag *string `json:"tg_alpha_tag"` + TgGroup *string `json:"tg_group"` + Source int `json:"source"` + Transcript *string `json:"transcript"` } type Incident struct { diff --git a/pkg/gordio/database/querier.go b/pkg/gordio/database/querier.go index 0f9d2a4..50ffb6b 100644 --- a/pkg/gordio/database/querier.go +++ b/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) diff --git a/pkg/gordio/database/talkgroups.sql.go b/pkg/gordio/database/talkgroups.sql.go index 98d7788..6ff6b1e 100644 --- a/pkg/gordio/database/talkgroups.sql.go +++ b/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] diff --git a/pkg/gordio/server/routes.go b/pkg/gordio/server/routes.go index 1ba9d77..601c2db 100644 --- a/pkg/gordio/server/routes.go +++ b/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) { diff --git a/pkg/gordio/server/server.go b/pkg/gordio/server/server.go index 104f31f..bd64082 100644 --- a/pkg/gordio/server/server.go +++ b/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,19 +45,23 @@ 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{ - auth: authenticator, - conf: cfg, - db: db, - r: r, - nex: nexus.New(), - logger: logger, + auth: authenticator, + conf: cfg, + db: db, + 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() diff --git a/pkg/gordio/sinks/alerter.go b/pkg/gordio/sinks/alerter.go new file mode 100644 index 0000000..1b63bd8 --- /dev/null +++ b/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 +} diff --git a/pkg/gordio/sinks/database.go b/pkg/gordio/sinks/database.go index ca29298..a2ff46d 100644 --- a/pkg/gordio/sinks/database.go +++ b/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), diff --git a/pkg/gordio/sinks/sinks.go b/pkg/gordio/sinks/sinks.go index 7495370..3c8afd9 100644 --- a/pkg/gordio/sinks/sinks.go +++ b/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() diff --git a/sql/postgres/migrations/001_initial.up.sql b/sql/postgres/migrations/001_initial.up.sql index 1fb4ef2..6fb7874 100644 --- a/sql/postgres/migrations/001_initial.up.sql +++ b/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(); diff --git a/sql/postgres/queries/talkgroups.sql b/sql/postgres/queries/talkgroups.sql index e316dae..fe902e9 100644 --- a/sql/postgres/queries/talkgroups.sql +++ b/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,