amigan/stillbox
1
0
Fork 0
Code Issues 16 Pull requests Projects Releases Packages Wiki Activity Actions

Compare commits

base: amigan:36c982473c3100eae170034546007afcb84e0d41
Branches Tags
amigan:trunk
amigan:flattenLearned
amigan:adminui
amigan:putTalkgroups
amigan:primebug
amigan:rrimport
amigan:archive/duration
amigan:archive/old_filters
amigan:archive/react
amigan:archive/rustimpl
..
compare: amigan:3f133e152a9a10f64a2a8990f094ffb4117e7dd6
Branches Tags
amigan:trunk
amigan:flattenLearned
amigan:adminui
amigan:putTalkgroups
amigan:primebug
amigan:rrimport
amigan:archive/duration
amigan:archive/old_filters
amigan:archive/react
amigan:archive/rustimpl

No commits in common. "36c982473c3100eae170034546007afcb84e0d41" and "3f133e152a9a10f64a2a8990f094ffb4117e7dd6" have entirely different histories.
36c982473c ... 3f133e152a

20 changed files with 341 additions and 641 deletions
Show stats Expand all files Collapse all files

6
config.sample.yaml
View file

@ -33,11 +33,5 @@ alerting:
renotify: 30m
notify:
- provider: slackwebhook
# subjectTemplate: "Stillbox Alert ({{ highest . }})"
# bodyTemplate: |
# {{ range . -}}
# {{ .TGName }} is active with a score of {{ f .Score.Score 4 }}! ({{ f .Score.RecentCount 0 }}/{{ .Score.Count }} recent calls)
#
# {{ end -}}
config:
webhookURL: "http://somewhere"

10
internal/common/common.go
View file

@ -1,6 +1,9 @@
package common
import (
"fmt"
"strconv"
"github.com/spf13/cobra"
)
@ -44,3 +47,10 @@ func PtrOrNull[T comparable](val T) *T {
return &val
}
func FmtFloat(v float64, places ...int) string {
if len(places) > 0 {
return fmt.Sprintf("%."+strconv.Itoa(places[0])+"f", v)
}
return fmt.Sprintf("%.4f", v)
}

48
internal/common/template.go
View file

@ -1,48 +0,0 @@
package common
import (
"errors"
"fmt"
"strconv"
"text/template"
"time"
"dynatron.me/x/stillbox/internal/jsontime"
)
var (
FuncMap = template.FuncMap{
"f": fmtFloat,
"dict": func(values ...interface{}) (map[string]interface{}, error) {
if len(values)%2 != 0 {
return nil, errors.New("invalid dict call")
}
dict := make(map[string]interface{}, len(values)/2)
for i := 0; i < len(values); i += 2 {
key, ok := values[i].(string)
if !ok {
return nil, errors.New("dict keys must be strings")
}
dict[key] = values[i+1]
}
return dict, nil
},
"formTime": func(t jsontime.Time) string {
return time.Time(t).Format("2006-01-02T15:04")
},
"ago": func(s string) (string, error) {
d, err := time.ParseDuration(s)
if err != nil {
return "", err
}
return time.Now().Add(-d).Format("2006-01-02T15:04"), nil
},
}
)
func fmtFloat(v float64, places ...int) string {
if len(places) > 0 {
return fmt.Sprintf("%."+strconv.Itoa(places[0])+"f", v)
}
return fmt.Sprintf("%.4f", v)
}

215
internal/forms/forms.go
View file

@ -1,7 +1,6 @@
package forms
import (
"encoding/json"
"errors"
"fmt"
"io"
@ -17,22 +16,15 @@ import (
)
var (
ErrNotStruct = errors.New("destination is not a struct")
ErrNotPointer = errors.New("destination is not a pointer")
ErrContentType = errors.New("bad content type")
)
const (
MaxMultipartMemory int64 = 1024 * 1024 // 1MB
ErrNotStruct = errors.New("destination is not a struct")
ErrNotPointer = errors.New("destination is not a pointer")
)
type options struct {
tagOverride *string
parseTimeIn *time.Location
parseLocal bool
acceptBlank bool
maxMultipartMemory int64
defaultOmitEmpty bool
tagOverride *string
parseTimeIn *time.Location
parseLocal bool
acceptBlank bool
}
type Option func(*options)
@ -61,18 +53,6 @@ func WithTag(t string) Option {
}
}
func WithMaxMultipartSize(s int64) Option {
return func(o *options) {
o.maxMultipartMemory = s
}
}
func WithOmitEmpty() Option {
return func(o *options) {
o.defaultOmitEmpty = true
}
}
func (o *options) Tag() string {
if o.tagOverride != nil {
return *o.tagOverride
@ -167,19 +147,17 @@ func (o *options) parseDuration(s string) (v time.Duration, set bool, err error)
return
}
var typeOfByteSlice = reflect.TypeOf([]byte(nil))
func (o *options) iterFields(r *http.Request, destStruct reflect.Value) error {
structType := destStruct.Type()
for i := 0; i < destStruct.NumField(); i++ {
destFieldVal := destStruct.Field(i)
fieldType := structType.Field(i)
if !fieldType.IsExported() && !fieldType.Anonymous {
func (o *options) iterFields(r *http.Request, rv reflect.Value) error {
rt := rv.Type()
for i := 0; i < rv.NumField(); i++ {
f := rv.Field(i)
tf := rt.Field(i)
if !tf.IsExported() && !tf.Anonymous {
continue
}
if destFieldVal.Kind() == reflect.Struct && fieldType.Anonymous {
err := o.iterFields(r, destFieldVal)
if f.Kind() == reflect.Struct && tf.Anonymous {
err := o.iterFields(r, f)
if err != nil {
return err
}
@ -187,38 +165,51 @@ func (o *options) iterFields(r *http.Request, destStruct reflect.Value) error {
var tAr []string
var formField string
var omitEmpty bool
if o.defaultOmitEmpty {
omitEmpty = true
}
formTag, has := structType.Field(i).Tag.Lookup(o.Tag())
formTag, has := rt.Field(i).Tag.Lookup(o.Tag())
if has {
tAr = strings.Split(formTag, ",")
formField = tAr[0]
for _, v := range tAr[1:] {
if v == "omitempty" {
omitEmpty = true
break
}
}
}
if !has || formField == "-" {
continue
}
destFieldIntf := destFieldVal.Interface()
fi := f.Interface()
if destFieldVal.Kind() == reflect.Slice && destFieldVal.Type() == typeOfByteSlice {
switch v := fi.(type) {
case string, *string:
s := r.Form.Get(formField)
setVal(f, s != "" || o.acceptBlank, v, s)
case int, uint, *int, *uint:
ff := r.Form.Get(formField)
val, set, err := o.parseInt(ff)
if err != nil {
return err
}
setVal(f, set, v, val)
case float64:
ff := r.Form.Get(formField)
val, set, err := o.parseFloat64(ff)
if err != nil {
return err
}
setVal(f, set, v, val)
case bool, *bool:
ff := r.Form.Get(formField)
val, set, err := o.parseBool(ff)
if err != nil {
return err
}
setVal(f, set, v, val)
case []byte:
file, hdr, err := r.FormFile(formField)
if err != nil {
return fmt.Errorf("get form file: %w", err)
}
nameField, hasFilename := structType.Field(i).Tag.Lookup("filenameField")
nameField, hasFilename := rt.Field(i).Tag.Lookup("filenameField")
if hasFilename {
fnf := destStruct.FieldByName(nameField)
fnf := rv.FieldByName(nameField)
if fnf == (reflect.Value{}) {
panic(fmt.Errorf("filenameField '%s' does not exist", nameField))
}
@ -230,52 +221,23 @@ func (o *options) iterFields(r *http.Request, destStruct reflect.Value) error {
return fmt.Errorf("file read: %w", err)
}
destFieldVal.SetBytes(audioBytes)
continue
}
if !r.Form.Has(formField) && omitEmpty {
continue
}
ff := r.Form.Get(formField)
switch v := destFieldIntf.(type) {
case string, *string:
setVal(destFieldVal, ff != "" || o.acceptBlank, ff)
case int, uint, *int, *uint:
val, set, err := o.parseInt(ff)
if err != nil {
return err
}
setVal(destFieldVal, set, val)
case float64:
val, set, err := o.parseFloat64(ff)
if err != nil {
return err
}
setVal(destFieldVal, set, val)
case bool, *bool:
val, set, err := o.parseBool(ff)
if err != nil {
return err
}
setVal(destFieldVal, set, val)
f.SetBytes(audioBytes)
case time.Time, *time.Time, jsontime.Time, *jsontime.Time:
t, set, err := o.parseTime(ff)
tval := r.Form.Get(formField)
t, set, err := o.parseTime(tval)
if err != nil {
return err
}
setVal(destFieldVal, set, t)
setVal(f, set, v, t)
case time.Duration, *time.Duration, jsontime.Duration, *jsontime.Duration:
d, set, err := o.parseDuration(ff)
dval := r.Form.Get(formField)
d, set, err := o.parseDuration(dval)
if err != nil {
return err
}
setVal(destFieldVal, set, d)
setVal(f, set, v, d)
case []int:
val := strings.Trim(ff, "[]")
val := strings.Trim(r.Form.Get(formField), "[]")
if val == "" && o.acceptBlank {
continue
}
@ -287,7 +249,7 @@ func (o *options) iterFields(r *http.Request, destStruct reflect.Value) error {
ar = append(ar, i)
}
}
destFieldVal.Set(reflect.ValueOf(ar))
f.Set(reflect.ValueOf(ar))
default:
panic(fmt.Errorf("unsupported type %T", v))
}
@ -296,77 +258,48 @@ func (o *options) iterFields(r *http.Request, destStruct reflect.Value) error {
return nil
}
func setVal(destFieldVal reflect.Value, set bool, src any) {
func setVal(setField reflect.Value, set bool, fv any, sv any) {
if !set {
return
}
destType := destFieldVal.Type()
srcVal := reflect.ValueOf(src)
rv := reflect.TypeOf(fv)
svo := reflect.ValueOf(sv)
if srcVal.Kind() == reflect.Ptr {
srcVal = srcVal.Elem()
if svo.CanConvert(rv) {
svo = svo.Convert(rv)
}
if destType.Kind() == reflect.Ptr {
if !srcVal.CanAddr() {
if srcVal.CanConvert(destType.Elem()) {
srcVal = srcVal.Convert(destType.Elem())
}
copy := reflect.New(srcVal.Type())
copy.Elem().Set(srcVal)
srcVal = copy
}
} else if srcVal.CanConvert(destFieldVal.Type()) {
srcVal = srcVal.Convert(destFieldVal.Type())
if rv.Kind() == reflect.Ptr {
svo = svo.Addr()
}
destFieldVal.Set(srcVal)
setField.Set(svo)
}
func Unmarshal(r *http.Request, dest any, opt ...Option) error {
o := options{
maxMultipartMemory: MaxMultipartMemory,
}
o := options{}
for _, opt := range opt {
opt(&o)
}
contentType := strings.Split(r.Header.Get("Content-Type"), ";")[0]
switch contentType {
case "multipart/form-data":
err := r.ParseMultipartForm(o.maxMultipartMemory)
if err != nil {
return fmt.Errorf("ParseForm: %w", err)
}
return o.unmarshalForm(r, dest)
case "application/x-www-form-urlencoded":
err := r.ParseForm()
if err != nil {
return fmt.Errorf("ParseForm: %w", err)
}
return o.unmarshalForm(r, dest)
case "application/json":
return json.NewDecoder(r.Body).Decode(dest)
}
return ErrContentType
}
func (o *options) unmarshalForm(r *http.Request, dest any) error {
destVal := reflect.ValueOf(dest)
if k := destVal.Kind(); k == reflect.Ptr {
destVal = destVal.Elem()
rv := reflect.ValueOf(dest)
if k := rv.Kind(); k == reflect.Ptr {
rv = rv.Elem()
} else {
return ErrNotPointer
}
if destVal.Kind() != reflect.Struct {
if rv.Kind() != reflect.Struct {
return ErrNotStruct
}
return o.iterFields(r, destVal)
if strings.HasPrefix(r.Header.Get("Content-Type"), "application/x-www-form-urlencoded") {
err := r.ParseForm()
if err != nil {
return fmt.Errorf("ParseForm: %w", err)
}
}
return o.iterFields(r, rv)
}

87
internal/forms/forms_test.go
View file

@ -8,7 +8,6 @@ import (
"testing"
"time"
"dynatron.me/x/stillbox/internal/common"
"dynatron.me/x/stillbox/internal/forms"
"dynatron.me/x/stillbox/internal/jsontime"
@ -55,14 +54,6 @@ type urlEncTestJT struct {
ScoreEnd jsontime.Time `json:"scoreEnd"`
}
type ptrTestJT struct {
LookbackDays uint `form:"lookbackDays"`
HalfLife *jsontime.Duration `form:"halfLife"`
Recent *string `form:"recent"`
ScoreStart *jsontime.Time `form:"scoreStart"`
ScoreEnd jsontime.Time `form:"scoreEnd"`
}
var (
UrlEncTest = urlEncTest{
LookbackDays: 7,
@ -78,13 +69,6 @@ var (
ScoreStart: jsontime.Time(time.Date(2024, time.October, 28, 9, 25, 0, 0, time.UTC)),
}
PtrTestJT = ptrTestJT{
LookbackDays: 7,
HalfLife: common.PtrTo(jsontime.Duration(30 * time.Minute)),
Recent: common.PtrTo("2h0m0s"),
ScoreStart: common.PtrTo(jsontime.Time(time.Date(2024, time.October, 28, 9, 25, 0, 0, time.UTC))),
}
UrlEncTestJTLocal = urlEncTestJT{
LookbackDays: 7,
HalfLife: jsontime.Duration(30 * time.Minute),
@ -138,29 +122,29 @@ func TestUnmarshal(t *testing.T) {
name string
r *http.Request
dest any
expect any
compare any
expectErr error
opts []forms.Option
}{
{
name: "base case",
r: makeRequest("call1.http"),
dest: &callUploadRequest{},
expect: &Call1,
opts: []forms.Option{forms.WithAcceptBlank()},
name: "base case",
r: makeRequest("call1.http"),
dest: &callUploadRequest{},
compare: &Call1,
opts: []forms.Option{forms.WithAcceptBlank()},
},
{
name: "base case no accept blank",
r: makeRequest("call1.http"),
dest: &callUploadRequest{},
expect: &Call1,
compare: &Call1,
expectErr: errors.New(`parsebool(''): strconv.ParseBool: parsing "": invalid syntax`),
},
{
name: "not a pointer",
r: makeRequest("call1.http"),
dest: callUploadRequest{},
expect: callUploadRequest{},
compare: callUploadRequest{},
expectErr: forms.ErrNotPointer,
opts: []forms.Option{forms.WithAcceptBlank()},
},
@ -168,7 +152,7 @@ func TestUnmarshal(t *testing.T) {
name: "not a struct",
r: makeRequest("call1.http"),
dest: &str,
expect: callUploadRequest{},
compare: callUploadRequest{},
expectErr: forms.ErrNotStruct,
opts: []forms.Option{forms.WithAcceptBlank()},
},
@ -176,51 +160,44 @@ func TestUnmarshal(t *testing.T) {
name: "url encoded",
r: makeRequest("urlenc.http"),
dest: &urlEncTest{},
expect: &UrlEncTest,
compare: &UrlEncTest,
expectErr: errors.New(`Could not find format for ""`),
},
{
name: "url encoded accept blank",
r: makeRequest("urlenc.http"),
dest: &urlEncTest{},
expect: &UrlEncTest,
opts: []forms.Option{forms.WithAcceptBlank()},
},
{
name: "url encoded accept blank pointer",
r: makeRequest("urlenc.http"),
dest: &ptrTestJT{},
expect: &PtrTestJT,
opts: []forms.Option{forms.WithAcceptBlank()},
name: "url encoded accept blank",
r: makeRequest("urlenc.http"),
dest: &urlEncTest{},
compare: &UrlEncTest,
opts: []forms.Option{forms.WithAcceptBlank()},
},
{
name: "url encoded jsontime",
r: makeRequest("urlenc.http"),
dest: &urlEncTestJT{},
expect: &UrlEncTestJT,
compare: &UrlEncTestJT,
expectErr: errors.New(`Could not find format for ""`),
opts: []forms.Option{forms.WithTag("json")},
},
{
name: "url encoded jsontime with tz",
r: makeRequest("urlenc.http"),
dest: &urlEncTestJT{},
expect: &UrlEncTestJT,
opts: []forms.Option{forms.WithAcceptBlank(), forms.WithParseTimeInTZ(time.UTC), forms.WithTag("json")},
name: "url encoded jsontime with tz",
r: makeRequest("urlenc.http"),
dest: &urlEncTestJT{},
compare: &UrlEncTestJT,
opts: []forms.Option{forms.WithAcceptBlank(), forms.WithParseTimeInTZ(time.UTC), forms.WithTag("json")},
},
{
name: "url encoded jsontime with local",
r: makeRequest("urlenc.http"),
dest: &urlEncTestJT{},
expect: &UrlEncTestJTLocal,
opts: []forms.Option{forms.WithAcceptBlank(), forms.WithParseLocalTime(), forms.WithTag("json")},
name: "url encoded jsontime with local",
r: makeRequest("urlenc.http"),
dest: &urlEncTestJT{},
compare: &UrlEncTestJTLocal,
opts: []forms.Option{forms.WithAcceptBlank(), forms.WithParseLocalTime(), forms.WithTag("json")},
},
{
name: "sim real data",
r: makeRequest("urlenc2.http"),
dest: &alerting.Simulation{},
expect: realSim,
opts: []forms.Option{forms.WithAcceptBlank(), forms.WithParseLocalTime()},
name: "sim real data",
r: makeRequest("urlenc2.http"),
dest: &alerting.Simulation{},
compare: realSim,
opts: []forms.Option{forms.WithAcceptBlank(), forms.WithParseLocalTime()},
},
}
@ -232,7 +209,7 @@ func TestUnmarshal(t *testing.T) {
assert.Contains(t, tc.expectErr.Error(), err.Error())
} else {
require.NoError(t, err)
assert.Equal(t, tc.expect, tc.dest)
assert.Equal(t, tc.compare, tc.dest)
}
})
}

80
pkg/alerting/alert/alert.go
View file

@ -1,80 +0,0 @@
package alert
import (
"context"
"fmt"
"strconv"
"time"
"dynatron.me/x/stillbox/internal/trending"
"dynatron.me/x/stillbox/pkg/database"
"dynatron.me/x/stillbox/pkg/talkgroups"
"github.com/google/uuid"
"github.com/jackc/pgx/v5/pgtype"
)
type Alert struct {
ID uuid.UUID
Timestamp time.Time
TGName string
Score trending.Score[talkgroups.ID]
OrigScore float64
Weight float32
Suppressed bool
}
func (a *Alert) ToAddAlertParams() database.AddAlertParams {
f32score := float32(a.Score.Score)
f32origscore := float32(a.OrigScore)
var origScore *float32
if a.Score.Score != a.OrigScore {
origScore = &f32origscore
}
return database.AddAlertParams{
ID: a.ID,
Time: pgtype.Timestamptz{Time: a.Timestamp, Valid: true},
PackedTg: a.Score.ID.Pack(),
Weight: &a.Weight,
Score: &f32score,
OrigScore: origScore,
Notified: !a.Suppressed,
}
}
// Make creates an alert for later rendering or storage.
func Make(ctx context.Context, store talkgroups.Store, score trending.Score[talkgroups.ID], origScore float64) (Alert, error) {
d := Alert{
ID: uuid.New(),
Score: score,
Timestamp: time.Now(),
Weight: 1.0,
OrigScore: origScore,
}
tgRecord, err := store.TG(ctx, score.ID)
switch err {
case nil:
d.Weight = tgRecord.Talkgroup.Weight
if tgRecord.System.Name == "" {
tgRecord.System.Name = strconv.Itoa(int(score.ID.System))
}
if tgRecord.Talkgroup.Name != nil {
d.TGName = fmt.Sprintf("%s %s [%d]", tgRecord.System.Name, *tgRecord.Talkgroup.Name, score.ID.Talkgroup)
} else {
d.TGName = fmt.Sprintf("%s:%d", tgRecord.System.Name, int(score.ID.Talkgroup))
}
default:
system, has := store.SystemName(ctx, int(score.ID.System))
if has {
d.TGName = fmt.Sprintf("%s:%d", system, int(score.ID.Talkgroup))
} else {
d.TGName = fmt.Sprintf("%d:%d", int(score.ID.System), int(score.ID.Talkgroup))
}
}
return d, nil
}

112
pkg/alerting/alerting.go
View file

@ -1,14 +1,16 @@
package alerting
import (
"bytes"
"context"
"fmt"
"net/http"
"sort"
"strconv"
"sync"
"text/template"
"time"
"dynatron.me/x/stillbox/pkg/alerting/alert"
"dynatron.me/x/stillbox/pkg/calls"
"dynatron.me/x/stillbox/pkg/config"
"dynatron.me/x/stillbox/pkg/database"
@ -19,6 +21,8 @@ import (
"dynatron.me/x/stillbox/internal/timeseries"
"dynatron.me/x/stillbox/internal/trending"
"github.com/google/uuid"
"github.com/jackc/pgx/v5/pgtype"
"github.com/rs/zerolog/log"
)
@ -47,7 +51,7 @@ type alerter struct {
scores trending.Scores[talkgroups.ID]
lastScore time.Time
sim *Simulation
alertCache map[talkgroups.ID]alert.Alert
alertCache map[talkgroups.ID]Alert
renotify time.Duration
notifier notify.Notifier
tgCache talkgroups.Store
@ -92,7 +96,7 @@ func New(cfg config.Alerting, tgCache talkgroups.Store, opts ...AlertOption) Ale
as := &alerter{
cfg: cfg,
alertCache: make(map[talkgroups.ID]alert.Alert),
alertCache: make(map[talkgroups.ID]Alert),
clock: timeseries.DefaultClock,
renotify: DefaultRenotify,
tgCache: tgCache,
@ -146,18 +150,22 @@ func (as *alerter) Go(ctx context.Context) {
}
func (as *alerter) eval(ctx context.Context, now time.Time, testMode bool) ([]alert.Alert, error) {
const notificationTemplStr = `{{ range . -}}
{{ .TGName }} is active with a score of {{ f .Score.Score 4 }}! ({{ f .Score.RecentCount 0 }}/{{ .Score.Count }} recent calls)
{{ end -}}`
var notificationTemplate = template.Must(template.New("notification").Funcs(funcMap).Parse(notificationTemplStr))
func (as *alerter) eval(ctx context.Context, now time.Time, testMode bool) ([]Alert, error) {
err := as.tgCache.Hint(ctx, as.scoredTGs())
if err != nil {
return nil, fmt.Errorf("prime TG cache: %w", err)
}
as.Lock()
defer as.Unlock()
db := database.FromCtx(ctx)
var notifications []alert.Alert
var notifications []Alert
for _, s := range as.scores {
origScore := s.Score
tgr, err := as.tgCache.TG(ctx, s.ID)
@ -168,7 +176,7 @@ func (as *alerter) eval(ctx context.Context, now time.Time, testMode bool) ([]al
if s.Score > as.cfg.AlertThreshold || testMode {
if old, inCache := as.alertCache[s.ID]; !inCache || now.Sub(old.Timestamp) > as.renotify {
s.Score *= as.tgCache.Weight(ctx, s.ID, now)
a, err := alert.Make(ctx, as.tgCache, s, origScore)
a, err := as.makeAlert(ctx, s, origScore)
if err != nil {
return nil, fmt.Errorf("makeAlert: %w", err)
}
@ -198,7 +206,7 @@ func (as *alerter) eval(ctx context.Context, now time.Time, testMode bool) ([]al
}
func (as *alerter) testNotifyHandler(w http.ResponseWriter, r *http.Request) {
alerts := make([]alert.Alert, 0, len(as.scores))
alerts := make([]Alert, 0, len(as.scores))
ctx := r.Context()
alerts, err := as.eval(ctx, time.Now(), true)
@ -208,7 +216,7 @@ func (as *alerter) testNotifyHandler(w http.ResponseWriter, r *http.Request) {
return
}
err = as.notifier.Send(ctx, alerts)
err = as.sendNotification(ctx, alerts)
if err != nil {
log.Error().Err(err).Msg("test notification send")
http.Error(w, err.Error(), http.StatusInternalServerError)
@ -250,12 +258,92 @@ func (as *alerter) notify(ctx context.Context) error {
}
if len(notifications) > 0 {
return as.notifier.Send(ctx, notifications)
return as.sendNotification(ctx, notifications)
}
return nil
}
type Alert struct {
ID uuid.UUID
Timestamp time.Time
TGName string
Score trending.Score[talkgroups.ID]
OrigScore float64
Weight float32
Suppressed bool
}
func (a *Alert) ToAddAlertParams() database.AddAlertParams {
f32score := float32(a.Score.Score)
f32origscore := float32(a.OrigScore)
var origScore *float32
if a.Score.Score != a.OrigScore {
origScore = &f32origscore
}
return database.AddAlertParams{
ID: a.ID,
Time: pgtype.Timestamptz{Time: a.Timestamp, Valid: true},
PackedTg: a.Score.ID.Pack(),
Weight: &a.Weight,
Score: &f32score,
OrigScore: origScore,
Notified: !a.Suppressed,
}
}
// sendNotification renders and sends the notification.
func (as *alerter) sendNotification(ctx context.Context, n []Alert) error {
msgBuffer := new(bytes.Buffer)
err := notificationTemplate.Execute(msgBuffer, n)
if err != nil {
return fmt.Errorf("notification template render: %w", err)
}
log.Debug().Str("msg", msgBuffer.String()).Msg("notifying")
return as.notifier.Send(ctx, NotificationSubject, msgBuffer.String())
}
// makeAlert creates a notification for later rendering by the template.
// It takes a talkgroup Score as input.
func (as *alerter) makeAlert(ctx context.Context, score trending.Score[talkgroups.ID], origScore float64) (Alert, error) {
d := Alert{
ID: uuid.New(),
Score: score,
Timestamp: time.Now(),
Weight: 1.0,
OrigScore: origScore,
}
tgRecord, err := as.tgCache.TG(ctx, score.ID)
switch err {
case nil:
d.Weight = tgRecord.Talkgroup.Weight
if tgRecord.System.Name == "" {
tgRecord.System.Name = strconv.Itoa(int(score.ID.System))
}
if tgRecord.Talkgroup.Name != nil {
d.TGName = fmt.Sprintf("%s %s (%d)", tgRecord.System.Name, *tgRecord.Talkgroup.Name, score.ID.Talkgroup)
} else {
d.TGName = fmt.Sprintf("%s:%d", tgRecord.System.Name, int(score.ID.Talkgroup))
}
default:
system, has := as.tgCache.SystemName(ctx, int(score.ID.System))
if has {
d.TGName = fmt.Sprintf("%s:%d", system, int(score.ID.Talkgroup))
} else {
d.TGName = fmt.Sprintf("%d:%d", int(score.ID.System), int(score.ID.Talkgroup))
}
}
return d, nil
}
// cleanCache clears the cache of aged-out entries
func (as *alerter) cleanCache() {
if as.notifier == nil {

24
pkg/alerting/simulate.go
View file

@ -2,6 +2,7 @@ package alerting
import (
"context"
"encoding/json"
"errors"
"fmt"
"net/http"
@ -113,15 +114,24 @@ func (s *Simulation) Simulate(ctx context.Context) (trending.Scores[talkgroups.I
func (as *alerter) simulateHandler(w http.ResponseWriter, r *http.Request) {
ctx := r.Context()
s := new(Simulation)
err := forms.Unmarshal(r, s, forms.WithAcceptBlank(), forms.WithParseLocalTime())
if err != nil {
err = fmt.Errorf("simulate unmarshal: %w", err)
http.Error(w, err.Error(), http.StatusBadRequest)
return
switch r.Header.Get("Content-Type") {
case "application/json":
err := json.NewDecoder(r.Body).Decode(s)
if err != nil {
err = fmt.Errorf("simulate decode: %w", err)
http.Error(w, err.Error(), http.StatusBadRequest)
return
}
default:
err := forms.Unmarshal(r, s, forms.WithAcceptBlank(), forms.WithParseLocalTime())
if err != nil {
err = fmt.Errorf("simulate unmarshal: %w", err)
http.Error(w, err.Error(), http.StatusBadRequest)
return
}
}
err = s.verify()
err := s.verify()
if err != nil {
err = fmt.Errorf("simulation profile verify: %w", err)
http.Error(w, err.Error(), http.StatusBadRequest)

37
pkg/alerting/stats.go
View file

@ -2,6 +2,7 @@ package alerting
import (
_ "embed"
"errors"
"html/template"
"net/http"
"time"
@ -11,6 +12,7 @@ import (
"dynatron.me/x/stillbox/pkg/talkgroups"
"dynatron.me/x/stillbox/internal/common"
"dynatron.me/x/stillbox/internal/jsontime"
"dynatron.me/x/stillbox/internal/trending"
"github.com/go-chi/chi/v5"
@ -20,14 +22,41 @@ import (
//go:embed stats.html
var statsTemplateFile string
var (
statTmpl = template.Must(template.New("stats").Funcs(common.FuncMap).Parse(statsTemplateFile))
)
type stats interface {
PrivateRoutes(chi.Router)
}
var (
funcMap = template.FuncMap{
"f": common.FmtFloat,
"dict": func(values ...interface{}) (map[string]interface{}, error) {
if len(values)%2 != 0 {
return nil, errors.New("invalid dict call")
}
dict := make(map[string]interface{}, len(values)/2)
for i := 0; i < len(values); i += 2 {
key, ok := values[i].(string)
if !ok {
return nil, errors.New("dict keys must be strings")
}
dict[key] = values[i+1]
}
return dict, nil
},
"formTime": func(t jsontime.Time) string {
return time.Time(t).Format("2006-01-02T15:04")
},
"ago": func(s string) (string, error) {
d, err := time.ParseDuration(s)
if err != nil {
return "", err
}
return time.Now().Add(-d).Format("2006-01-02T15:04"), nil
},
}
statTmpl = template.Must(template.New("stats").Funcs(funcMap).Parse(statsTemplateFile))
)
func (as *alerter) PrivateRoutes(r chi.Router) {
r.Get("/tgstats", as.tgStatsHandler)
r.Post("/tgstats", as.simulateHandler)

48
pkg/api/api.go
View file

@ -18,10 +18,13 @@ type API interface {
}
type api struct {
tgs talkgroups.Store
}
func New() API {
s := new(api)
func New(tgs talkgroups.Store) API {
s := &api{
tgs: tgs,
}
return s
}
@ -29,8 +32,9 @@ func New() API {
func (a *api) Subrouter() http.Handler {
r := chi.NewMux()
r.Mount("/talkgroup", new(talkgroupAPI).routes())
r.Get("/talkgroup/{system:\\d+}/{id:\\d+}", a.talkgroup)
r.Get("/talkgroup/{system:\\d+}/", a.talkgroup)
r.Get("/talkgroup/", a.talkgroup)
return r
}
@ -54,7 +58,7 @@ func httpCode(err error) int {
return http.StatusInternalServerError
}
func writeResponse(w http.ResponseWriter, r *http.Request, data interface{}, err error) {
func (a *api) writeResponse(w http.ResponseWriter, r *http.Request, data interface{}, err error) {
if err != nil {
log.Error().Str("path", r.URL.Path).Err(err).Msg("request failed")
http.Error(w, err.Error(), httpCode(err))
@ -71,10 +75,6 @@ func writeResponse(w http.ResponseWriter, r *http.Request, data interface{}, err
}
}
func reqErr(w http.ResponseWriter, err error, code int) {
http.Error(w, err.Error(), code)
}
func decodeParams(d interface{}, r *http.Request) error {
params := chi.RouteContext(r.Context()).URLParams
m := make(map[string]string, len(params.Keys))
@ -96,6 +96,32 @@ func decodeParams(d interface{}, r *http.Request) error {
return dec.Decode(m)
}
func badReq(w http.ResponseWriter, err error) {
reqErr(w, err, http.StatusBadRequest)
func (a *api) badReq(w http.ResponseWriter, err error) {
http.Error(w, err.Error(), http.StatusBadRequest)
}
func (a *api) talkgroup(w http.ResponseWriter, r *http.Request) {
ctx := r.Context()
p := struct {
System *int `param:"system"`
ID *int `param:"id"`
}{}
err := decodeParams(&p, r)
if err != nil {
a.badReq(w, err)
return
}
var res interface{}
switch {
case p.System != nil && p.ID != nil:
res, err = a.tgs.TG(ctx, talkgroups.TG(*p.System, *p.ID))
case p.System != nil:
res, err = a.tgs.SystemTGs(ctx, int32(*p.System))
default:
res, err = a.tgs.TGs(ctx, nil)
}
a.writeResponse(w, r, res, err)
}

116
pkg/api/talkgroups.go
View file

@ -1,116 +0,0 @@
package api
import (
"fmt"
"net/http"
"dynatron.me/x/stillbox/internal/forms"
"dynatron.me/x/stillbox/pkg/talkgroups"
"github.com/go-chi/chi/v5"
)
type talkgroupAPI struct {
}
func (tga *talkgroupAPI) routes() http.Handler {
r := chi.NewMux()
r.Get("/{system:\\d+}/{id:\\d+}", tga.talkgroup)
r.Put("/{system:\\d+}/{id:\\d+}", tga.putTalkgroup)
r.Get("/{system:\\d+}/", tga.talkgroup)
r.Get("/", tga.talkgroup)
return r
}
type tgParams struct {
System *int `param:"system"`
ID *int `param:"id"`
}
func (t tgParams) haveBoth() bool {
return t.System != nil && t.ID != nil
}
func (t tgParams) ToID() talkgroups.ID {
nilOr := func(i *int) uint32 {
if i == nil {
return 0
}
return uint32(*i)
}
return talkgroups.ID{
System: nilOr(t.System),
Talkgroup: nilOr(t.ID),
}
}
func (tga *talkgroupAPI) talkgroup(w http.ResponseWriter, r *http.Request) {
ctx := r.Context()
tgs := talkgroups.StoreFrom(ctx)
var p tgParams
err := decodeParams(&p, r)
if err != nil {
badReq(w, err)
return
}
var res interface{}
switch {
case p.System != nil && p.ID != nil:
res, err = tgs.TG(ctx, talkgroups.TG(*p.System, *p.ID))
case p.System != nil:
res, err = tgs.SystemTGs(ctx, int32(*p.System))
default:
res, err = tgs.TGs(ctx, nil)
}
writeResponse(w, r, res, err)
}
func (tga *talkgroupAPI) putTalkgroup(w http.ResponseWriter, r *http.Request) {
var id tgParams
err := decodeParams(&id, r)
if err != nil {
badReq(w, err)
return
}
/*
ctx := r.Context()
tgs := talkgroups.StoreFrom(ctx)
tg, err := tgs.TG(ctx, id.ToID())
switch err {
case nil:
case talkgroups.ErrNotFound:
reqErr(w, err, http.StatusNotFound)
return
default:
reqErr(w, err, http.StatusInternalServerError)
}
*/
input := struct {
Name *string `form:"name"`
AlphaTag *string `form:"alpha_tag"`
TgGroup *string `form:"tg_group"`
Frequency *int32 `form:"frequency"`
Metadata []byte `form:"metadata"`
Tags []string `form:"tags"`
Alert *bool `form:"alert"`
AlertConfig []byte `form:"alert_config"`
Weight *float32 `form:"weight"`
}{}
err = forms.Unmarshal(r, &input, forms.WithAcceptBlank(), forms.WithOmitEmpty())
if err != nil {
reqErr(w, err, http.StatusBadRequest)
return
}
fmt.Fprintf(w, "%+v\n", input)
}

4
pkg/calls/call.go
View file

@ -19,7 +19,7 @@ func (d CallDuration) Duration() time.Duration {
return time.Duration(d)
}
func (d CallDuration) MsInt32Ptr() *int32 {
func (d CallDuration) Int32Ptr() *int32 {
if time.Duration(d) == 0 {
return nil
}
@ -106,7 +106,7 @@ func (c *Call) ToPB() *pb.Call {
Frequencies: toInt64Slice(c.Frequencies),
Patches: toInt32Slice(c.Patches),
Sources: toInt32Slice(c.Sources),
Duration: c.Duration.MsInt32Ptr(),
Duration: c.Duration.Int32Ptr(),
Audio: c.Audio,
}
}

6
pkg/config/config.go
View file

@ -65,10 +65,8 @@ type Alerting struct {
type Notify []NotifyService
type NotifyService struct {
Provider string `yaml:"provider" json:"provider"`
SubjectTemplate *string `yaml:"subjectTemplate" json:"subjectTemplate"`
BodyTemplate *string `yaml:"bodyTemplate" json:"bodyTemplate"`
Config map[string]interface{} `yaml:"config" json:"config"`
Provider string `json:"provider"`
Config map[string]interface{} `json:"config"`
}
func (n *NotifyService) GetS(k, defaultVal string) string {

6
pkg/database/database.go
View file

@ -58,11 +58,11 @@ func NewClient(ctx context.Context, conf config.DB) (*DB, error) {
type DBCtxKey string
const DBCtxKeyValue DBCtxKey = "dbctx"
const DBCTXKeyValue DBCtxKey = "dbctx"
// FromCtx returns the database handle from the provided Context.
func FromCtx(ctx context.Context) *DB {
c, ok := ctx.Value(DBCtxKeyValue).(*DB)
c, ok := ctx.Value(DBCTXKeyValue).(*DB)
if !ok {
panic("no DB in context")
}
@ -72,7 +72,7 @@ func FromCtx(ctx context.Context) *DB {
// CtxWithDB returns a Context with the provided database handle.
func CtxWithDB(ctx context.Context, conn *DB) context.Context {
return context.WithValue(ctx, DBCtxKeyValue, conn)
return context.WithValue(ctx, DBCTXKeyValue, conn)
}
// IsNoRows is a convenience function that returns whether a returned error is a database

143
pkg/notify/notify.go
View file

@ -1,15 +1,10 @@
package notify
import (
"bytes"
"context"
"fmt"
stdhttp "net/http"
"text/template"
"time"
"dynatron.me/x/stillbox/internal/common"
"dynatron.me/x/stillbox/pkg/alerting/alert"
"dynatron.me/x/stillbox/pkg/config"
"github.com/go-viper/mapstructure/v2"
@ -18,86 +13,15 @@ import (
)
type Notifier interface {
Send(ctx context.Context, alerts []alert.Alert) error
}
type backend struct {
*notify.Notify
subject *template.Template
body *template.Template
notify.Notifier
}
type notifier struct {
backends []backend
*notify.Notify
}
func highest(a []alert.Alert) string {
if len(a) < 1 {
return "none"
}
func (n *notifier) buildSlackWebhookPayload(cfg *slackWebhookConfig) func(string, string) any {
top := a[0]
for _, a := range a {
if a.Score.Score > top.Score.Score {
top = a
}
}
return top.TGName
}
var alertFm = template.FuncMap{
"highest": highest,
}
const (
defaultBodyTemplStr = `{{ range . -}}
{{ .TGName }} is active with a score of {{ f .Score.Score 4 }}! ({{ f .Score.RecentCount 0 }}/{{ .Score.Count }} recent calls)
{{ end -}}`
defaultSubjectTemplStr = `Stillbox Alert ({{ highest . }})`
)
var (
defaultTemplate *template.Template
)
func init() {
defaultTemplate = template.New("notification")
defaultTemplate.Funcs(common.FuncMap).Funcs(alertFm)
_, err := defaultTemplate.New("body").Parse(defaultBodyTemplStr)
if err != nil {
panic(err)
}
_, err = defaultTemplate.New("subject").Parse(defaultSubjectTemplStr)
if err != nil {
panic(err)
}
}
// Send renders and sends the Alerts.
func (b *backend) Send(ctx context.Context, alerts []alert.Alert) (err error) {
var subject, body bytes.Buffer
err = b.subject.ExecuteTemplate(&subject, "subject", alerts)
if err != nil {
return err
}
err = b.body.ExecuteTemplate(&body, "body", alerts)
if err != nil {
return err
}
err = b.Notify.Send(ctx, subject.String(), body.String())
if err != nil {
return err
}
return nil
}
func buildSlackWebhookPayload(cfg *slackWebhookConfig) func(string, string) any {
type Attachment struct {
Title string `json:"title"`
Text string `json:"text"`
@ -129,44 +53,12 @@ func buildSlackWebhookPayload(cfg *slackWebhookConfig) func(string, string) any
}
type slackWebhookConfig struct {
WebhookURL string `mapstructure:"webhookURL"`
Icon string `mapstructure:"icon"`
MessageURL string `mapstructure:"messageURL"`
SubjectTemplate string `mapstructure:"subjectTemplate"`
BodyTemplate string `mapstructure:"bodyTemplate"`
WebhookURL string `mapstructure:"webhookURL"`
Icon string `mapstructure:"icon"`
MessageURL string `mapstructure:"messageURL"`
}
func (n *notifier) addService(cfg config.NotifyService) (err error) {
be := backend{}
switch cfg.SubjectTemplate {
case nil:
be.subject = defaultTemplate.Lookup("subject")
if be.subject == nil {
panic("subject template nil")
}
default:
be.subject, err = template.New("subject").Funcs(common.FuncMap).Funcs(alertFm).Parse(*cfg.SubjectTemplate)
if err != nil {
return err
}
}
switch cfg.BodyTemplate {
case nil:
be.body = defaultTemplate.Lookup("body")
if be.body == nil {
panic("body template nil")
}
default:
be.body, err = template.New("body").Funcs(common.FuncMap).Funcs(alertFm).Parse(*cfg.BodyTemplate)
if err != nil {
return err
}
}
be.Notify = notify.New()
func (n *notifier) addService(cfg config.NotifyService) error {
switch cfg.Provider {
case "slackwebhook":
swc := &slackWebhookConfig{
@ -182,31 +74,20 @@ func (n *notifier) addService(cfg config.NotifyService) (err error) {
Header: make(stdhttp.Header),
Method: stdhttp.MethodPost,
URL: swc.WebhookURL,
BuildPayload: buildSlackWebhookPayload(swc),
BuildPayload: n.buildSlackWebhookPayload(swc),
})
be.UseServices(hs)
n.UseServices(hs)
default:
return fmt.Errorf("unknown provider '%s'", cfg.Provider)
}
n.backends = append(n.backends, be)
return nil
}
func (n *notifier) Send(ctx context.Context, alerts []alert.Alert) error {
for _, be := range n.backends {
err := be.Send(ctx, alerts)
if err != nil {
return err
}
}
return nil
}
func New(cfg config.Notify) (Notifier, error) {
n := new(notifier)
n := &notifier{
Notify: notify.NewWithServices(),
}
for _, s := range cfg {
err := n.addService(s)

4
pkg/server/routes.go
View file

@ -9,7 +9,6 @@ import (
"dynatron.me/x/stillbox/internal/version"
"dynatron.me/x/stillbox/pkg/config"
"dynatron.me/x/stillbox/pkg/database"
"dynatron.me/x/stillbox/pkg/talkgroups"
"github.com/go-chi/chi/v5"
"github.com/go-chi/chi/v5/middleware"
"github.com/go-chi/httprate"
@ -27,8 +26,7 @@ func (s *Server) setupRoutes() {
}
r := s.r
r.Use(middleware.WithValue(database.DBCtxKeyValue, s.db))
r.Use(middleware.WithValue(talkgroups.StoreCtxKeyValue, s.tgs))
r.Use(middleware.WithValue(database.DBCTXKeyValue, s.db))
s.installPprof()

2
pkg/server/server.go
View file

@ -61,7 +61,7 @@ func New(ctx context.Context, cfg *config.Config) (*Server, error) {
}
tgCache := talkgroups.NewCache()
api := api.New()
api := api.New(tgCache)
srv := &Server{
auth: authenticator,

2
pkg/sinks/database.go
View file

@ -50,7 +50,7 @@ func (s *DatabaseSink) toAddCallParams(call *calls.Call) database.AddCallParams
AudioName: common.PtrOrNull(call.AudioName),
AudioBlob: call.Audio,
AudioType: common.PtrOrNull(call.AudioType),
Duration: call.Duration.MsInt32Ptr(),
Duration: call.Duration.Int32Ptr(),
Frequency: call.Frequency,
Frequencies: call.Frequencies,
Patches: call.Patches,

10
pkg/talkgroups/cache.go
View file

@ -49,16 +49,16 @@ type Store interface {
HUP(*config.Config)
}
type CtxStoreKey string
type CtxStoreKeyT string
const StoreCtxKeyValue CtxStoreKey = "store"
const CtxStoreKey CtxStoreKeyT = "store"
func CtxWithStore(ctx context.Context, s Store) context.Context {
return context.WithValue(ctx, StoreCtxKeyValue, s)
return context.WithValue(ctx, CtxStoreKey, s)
}
func StoreFrom(ctx context.Context) Store {
s, ok := ctx.Value(StoreCtxKeyValue).(Store)
s, ok := ctx.Value(CtxStoreKey).(Store)
if !ok {
return NewCache()
}
@ -127,7 +127,7 @@ func (t *cache) add(rec *Talkgroup) error {
t.Lock()
defer t.Unlock()
tg := TG(rec.System.ID, rec.Talkgroup.Tgid)
tg := TG(rec.System.ID, int(rec.Talkgroup.Tgid))
t.tgs[tg] = rec
t.systems[int32(rec.System.ID)] = rec.System.Name

22
pkg/talkgroups/talkgroup.go
View file

@ -12,11 +12,6 @@ type Talkgroup struct {
Learned bool `json:"learned"`
}
type Names struct {
System string
Talkgroup string
}
type ID struct {
System uint32 `json:"sys"`
Talkgroup uint32 `json:"tg"`
@ -33,11 +28,7 @@ func (ids *IDs) Packed() []int64 {
return r
}
type intId interface {
int | uint | int64 | uint64 | int32 | uint32
}
func TG[T intId, U intId](sys T, tgid U) ID {
func TG[T int | uint | int64 | uint64 | int32 | uint32](sys, tgid T) ID {
return ID{
System: uint32(sys),
Talkgroup: uint32(tgid),
@ -51,5 +42,14 @@ func (t ID) Pack() int64 {
func (t ID) String() string {
return fmt.Sprintf("%d:%d", t.System, t.Talkgroup)
}
func PackedTGs(tg []ID) []int64 {
s := make([]int64, len(tg))
for i, v := range tg {
s[i] = v.Pack()
}
return s
}