Big reorg #23

Merged
amigan merged 4 commits from rest into trunk 2024-11-03 13:46:12 -05:00
8 changed files with 124 additions and 40 deletions
Showing only changes of commit d917e13c4f - Show all commits

View file

@ -168,8 +168,8 @@ func (as *alerter) eval(ctx context.Context, now time.Time, testMode bool) ([]Al
var notifications []Alert var notifications []Alert
for _, s := range as.scores { for _, s := range as.scores {
origScore := s.Score origScore := s.Score
tgr, has := as.tgCache.TG(ctx, s.ID) tgr, err := as.tgCache.TG(ctx, s.ID)
if has { if err == nil {
if !tgr.Talkgroup.Alert { if !tgr.Talkgroup.Alert {
continue continue
} }
@ -327,9 +327,9 @@ func (as *alerter) makeAlert(ctx context.Context, score trending.Score[talkgroup
OrigScore: origScore, OrigScore: origScore,
} }
tgRecord, has := as.tgCache.TG(ctx, score.ID) tgRecord, err := as.tgCache.TG(ctx, score.ID)
switch has { switch err {
case true: case nil:
d.Weight = tgRecord.Talkgroup.Weight d.Weight = tgRecord.Talkgroup.Weight
if tgRecord.System.Name == "" { if tgRecord.System.Name == "" {
tgRecord.System.Name = strconv.Itoa(int(score.ID.System)) tgRecord.System.Name = strconv.Itoa(int(score.ID.System))
@ -340,7 +340,7 @@ func (as *alerter) makeAlert(ctx context.Context, score trending.Score[talkgroup
} else { } else {
d.TGName = fmt.Sprintf("%s:%d", tgRecord.System.Name, int(score.ID.Talkgroup)) d.TGName = fmt.Sprintf("%s:%d", tgRecord.System.Name, int(score.ID.Talkgroup))
} }
case false: default:
system, has := as.tgCache.SystemName(ctx, int(score.ID.System)) system, has := as.tgCache.SystemName(ctx, int(score.ID.System))
if has { if has {
d.TGName = fmt.Sprintf("%s:%d", system, int(score.ID.Talkgroup)) d.TGName = fmt.Sprintf("%s:%d", system, int(score.ID.Talkgroup))

View file

@ -112,10 +112,10 @@ JOIN systems sys ON tg.system_id = sys.id
WHERE tg.id = systg2id($1, $2) WHERE tg.id = systg2id($1, $2)
UNION UNION
SELECT SELECT
tgl.id::INT8, tgl.system_id::INT4, sys.name system_name, tgl.tgid::INT4, tgl.name, tgl.id::INT8, tgl.system_id::INT4, tgl.tgid::INT4, tgl.name,
tgl.alpha_tag, NULL::INTEGER, NULL::JSONB, tgl.alpha_tag, tgl.alpha_tag, NULL::INTEGER, NULL::JSONB,
CASE WHEN tgl.alpha_tag IS NULL THEN NULL ELSE ARRAY[tgl.alpha_tag] END, tgl.alpha_tag, CASE WHEN tgl.alpha_tag IS NULL THEN NULL ELSE ARRAY[tgl.alpha_tag] END,
TRUE, 1.0, NULL::JSONB, TRUE, NULL::JSONB, 1.0, sys.id, sys.name,
TRUE learned TRUE learned
FROM talkgroups_learned tgl FROM talkgroups_learned tgl
JOIN systems sys ON tgl.system_id = sys.id JOIN systems sys ON tgl.system_id = sys.id
@ -160,10 +160,10 @@ JOIN systems sys ON tg.system_id = sys.id
WHERE tg.id = ANY($1::INT8[]) WHERE tg.id = ANY($1::INT8[])
UNION UNION
SELECT SELECT
tgl.id::INT8, tgl.system_id::INT4, sys.name system_name, tgl.tgid::INT4, tgl.name, tgl.id::INT8, tgl.system_id::INT4, tgl.tgid::INT4, tgl.name,
tgl.alpha_tag, NULL::INTEGER, NULL::JSONB, tgl.alpha_tag, tgl.alpha_tag, NULL::INTEGER, NULL::JSONB,
CASE WHEN tgl.alpha_tag IS NULL THEN NULL ELSE ARRAY[tgl.alpha_tag] END, tgl.alpha_tag, CASE WHEN tgl.alpha_tag IS NULL THEN NULL ELSE ARRAY[tgl.alpha_tag] END,
TRUE, 1.0, NULL::JSONB, TRUE, NULL::JSONB, 1.0, sys.id, sys.name,
TRUE learned TRUE learned
FROM talkgroups_learned tgl FROM talkgroups_learned tgl
JOIN systems sys ON tgl.system_id = sys.id JOIN systems sys ON tgl.system_id = sys.id

View file

@ -0,0 +1,49 @@
package database
import (
"testing"
"github.com/stretchr/testify/require"
)
const getTalkgroupWithLearnedByPackedIDsTest = `-- name: GetTalkgroupWithLearnedByPackedIDs :many
SELECT
tg.id, tg.system_id, tg.tgid, tg.name, tg.alpha_tag, tg.tg_group, tg.frequency, tg.metadata, tg.tags, tg.alert, tg.alert_config, tg.weight, sys.id, sys.name,
FALSE learned
FROM talkgroups tg
JOIN systems sys ON tg.system_id = sys.id
WHERE tg.id = ANY($1::INT8[])
UNION
SELECT
tgl.id::INT8, tgl.system_id::INT4, tgl.tgid::INT4, tgl.name,
tgl.alpha_tag, tgl.alpha_tag, NULL::INTEGER, NULL::JSONB,
CASE WHEN tgl.alpha_tag IS NULL THEN NULL ELSE ARRAY[tgl.alpha_tag] END,
TRUE, NULL::JSONB, 1.0, sys.id, sys.name,
TRUE learned
FROM talkgroups_learned tgl
JOIN systems sys ON tgl.system_id = sys.id
WHERE systg2id(tgl.system_id, tgl.tgid) = ANY($1::INT8[]) AND ignored IS NOT TRUE
`
const getTalkgroupWithLearnedTest = `-- name: GetTalkgroupWithLearned :one
SELECT
tg.id, tg.system_id, tg.tgid, tg.name, tg.alpha_tag, tg.tg_group, tg.frequency, tg.metadata, tg.tags, tg.alert, tg.alert_config, tg.weight, sys.id, sys.name,
FALSE learned
FROM talkgroups tg
JOIN systems sys ON tg.system_id = sys.id
WHERE tg.id = systg2id($1, $2)
UNION
SELECT
tgl.id::INT8, tgl.system_id::INT4, tgl.tgid::INT4, tgl.name,
tgl.alpha_tag, tgl.alpha_tag, NULL::INTEGER, NULL::JSONB,
CASE WHEN tgl.alpha_tag IS NULL THEN NULL ELSE ARRAY[tgl.alpha_tag] END,
TRUE, NULL::JSONB, 1.0, sys.id, sys.name,
TRUE learned
FROM talkgroups_learned tgl
JOIN systems sys ON tgl.system_id = sys.id
WHERE tgl.system_id = $1 AND tgl.tgid = $2 AND ignored IS NOT TRUE
`
func TestQueryColumnsMatch(t *testing.T) {
require.Equal(t, getTalkgroupWithLearnedByPackedIDsTest, getTalkgroupWithLearnedByPackedIDs)
require.Equal(t, getTalkgroupWithLearnedTest, getTalkgroupWithLearned)
}

View file

@ -5,10 +5,9 @@ import (
"encoding/json" "encoding/json"
"dynatron.me/x/stillbox/pkg/calls" "dynatron.me/x/stillbox/pkg/calls"
"dynatron.me/x/stillbox/pkg/database"
"dynatron.me/x/stillbox/pkg/pb" "dynatron.me/x/stillbox/pkg/pb"
"dynatron.me/x/stillbox/pkg/talkgroups"
"github.com/jackc/pgx/v5"
"github.com/rs/zerolog/log" "github.com/rs/zerolog/log"
"google.golang.org/protobuf/types/known/structpb" "google.golang.org/protobuf/types/known/structpb"
) )
@ -61,10 +60,9 @@ func (c *client) SendError(cmd *pb.Command, err error) {
} }
func (c *client) Talkgroup(ctx context.Context, tg *pb.Talkgroup) error { func (c *client) Talkgroup(ctx context.Context, tg *pb.Talkgroup) error {
db := database.FromCtx(ctx) tgi, err := talkgroups.StoreFrom(ctx).TG(ctx, talkgroups.TG(tg.System, tg.Talkgroup))
tgi, err := db.GetTalkgroupWithLearned(ctx, int(tg.System), int(tg.Talkgroup))
if err != nil { if err != nil {
if err != pgx.ErrNoRows { if err != talkgroups.ErrNoTG {
log.Error().Err(err).Int32("sys", tg.System).Int32("tg", tg.Talkgroup).Msg("get talkgroup fail") log.Error().Err(err).Int32("sys", tg.System).Int32("tg", tg.Talkgroup).Msg("get talkgroup fail")
} }
return err return err

View file

@ -104,6 +104,7 @@ func (s *Server) Go(ctx context.Context) error {
s.installHupHandler() s.installHupHandler()
ctx = database.CtxWithDB(ctx, s.db) ctx = database.CtxWithDB(ctx, s.db)
ctx = talkgroups.CtxWithStore(ctx, s.tgs)
httpSrv := &http.Server{ httpSrv := &http.Server{
Addr: s.conf.Listen, Addr: s.conf.Listen,

View file

@ -2,6 +2,7 @@ package talkgroups
import ( import (
"context" "context"
"errors"
"sync" "sync"
"time" "time"
@ -14,11 +15,11 @@ import (
"github.com/rs/zerolog/log" "github.com/rs/zerolog/log"
) )
type tgMap map[ID]database.GetTalkgroupWithLearnedByPackedIDsRow type tgMap map[ID]Talkgroup
type Store interface { type Store interface {
// TG retrieves a Talkgroup from the Store. It returns the record and whether one was found. // TG retrieves a Talkgroup from the Store.
TG(ctx context.Context, tg ID) (database.GetTalkgroupWithLearnedByPackedIDsRow, bool) TG(ctx context.Context, tg ID) (Talkgroup, error)
// SystemName retrieves a system name from the store. It returns the record and whether one was found. // SystemName retrieves a system name from the store. It returns the record and whether one was found.
SystemName(ctx context.Context, id int) (string, bool) SystemName(ctx context.Context, id int) (string, bool)
@ -36,6 +37,23 @@ type Store interface {
Invalidate() Invalidate()
} }
type CtxStoreKeyT string
const CtxStoreKey CtxStoreKeyT = "store"
func CtxWithStore(ctx context.Context, s Store) context.Context {
return context.WithValue(ctx, CtxStoreKey, s)
}
func StoreFrom(ctx context.Context) Store {
s, ok := ctx.Value(CtxStoreKey).(Store)
if !ok {
return NewCache()
}
return s
}
func (t *cache) Invalidate() { func (t *cache) Invalidate() {
t.Lock() t.Lock()
defer t.Unlock() defer t.Unlock()
@ -89,7 +107,7 @@ func (t *cache) Hint(ctx context.Context, tgs []ID) error {
return nil return nil
} }
func (t *cache) add(rec database.GetTalkgroupWithLearnedByPackedIDsRow) error { func (t *cache) add(rec Talkgroup) error {
tg := TG(rec.System.ID, int(rec.Talkgroup.Tgid)) tg := TG(rec.System.ID, int(rec.Talkgroup.Tgid))
t.tgs[tg] = rec t.tgs[tg] = rec
t.systems[int32(rec.System.ID)] = rec.System.Name t.systems[int32(rec.System.ID)] = rec.System.Name
@ -97,6 +115,14 @@ func (t *cache) add(rec database.GetTalkgroupWithLearnedByPackedIDsRow) error {
return t.AlertConfig.AddAlertConfig(tg, rec.Talkgroup.AlertConfig) return t.AlertConfig.AddAlertConfig(tg, rec.Talkgroup.AlertConfig)
} }
func rowToTalkgroup(r database.GetTalkgroupWithLearnedByPackedIDsRow) Talkgroup {
return Talkgroup{
Talkgroup: r.Talkgroup,
System: r.System,
Learned: r.Learned,
}
}
func (t *cache) Load(ctx context.Context, tgs []int64) error { func (t *cache) Load(ctx context.Context, tgs []int64) error {
tgRecords, err := database.FromCtx(ctx).GetTalkgroupWithLearnedByPackedIDs(ctx, tgs) tgRecords, err := database.FromCtx(ctx).GetTalkgroupWithLearnedByPackedIDs(ctx, tgs)
if err != nil { if err != nil {
@ -107,7 +133,7 @@ func (t *cache) Load(ctx context.Context, tgs []int64) error {
defer t.Unlock() defer t.Unlock()
for _, rec := range tgRecords { for _, rec := range tgRecords {
err := t.add(rec) err := t.add(rowToTalkgroup(rec))
if err != nil { if err != nil {
log.Error().Err(err).Msg("add alert config fail") log.Error().Err(err).Msg("add alert config fail")
@ -117,38 +143,40 @@ func (t *cache) Load(ctx context.Context, tgs []int64) error {
return nil return nil
} }
func (t *cache) TG(ctx context.Context, tg ID) (database.GetTalkgroupWithLearnedByPackedIDsRow, bool) { var ErrNoTG = errors.New("talkgroup not found")
func (t *cache) TG(ctx context.Context, tg ID) (Talkgroup, error) {
t.RLock() t.RLock()
rec, has := t.tgs[tg] rec, has := t.tgs[tg]
t.RUnlock() t.RUnlock()
if has { if has {
return rec, has return rec, nil
} }
recs, err := database.FromCtx(ctx).GetTalkgroupWithLearnedByPackedIDs(ctx, []int64{tg.Pack()}) recs, err := database.FromCtx(ctx).GetTalkgroupWithLearnedByPackedIDs(ctx, []int64{tg.Pack()})
switch err { switch err {
case nil: case nil:
case pgx.ErrNoRows: case pgx.ErrNoRows:
return database.GetTalkgroupWithLearnedByPackedIDsRow{}, false return Talkgroup{}, ErrNoTG
default: default:
log.Error().Err(err).Msg("TG() cache add db get") log.Error().Err(err).Msg("TG() cache add db get")
return database.GetTalkgroupWithLearnedByPackedIDsRow{}, false return Talkgroup{}, errors.Join(ErrNoTG, err)
} }
if len(recs) < 1 { if len(recs) < 1 {
return database.GetTalkgroupWithLearnedByPackedIDsRow{}, false return Talkgroup{}, ErrNoTG
} }
t.Lock() t.Lock()
defer t.Unlock() defer t.Unlock()
err = t.add(recs[0]) err = t.add(rowToTalkgroup(recs[0]))
if err != nil { if err != nil {
log.Error().Err(err).Msg("TG() cache add") log.Error().Err(err).Msg("TG() cache add")
return recs[0], false return rowToTalkgroup(recs[0]), errors.Join(ErrNoTG, err)
} }
return recs[0], true return rowToTalkgroup(recs[0]), nil
} }
func (t *cache) SystemName(ctx context.Context, id int) (name string, has bool) { func (t *cache) SystemName(ctx context.Context, id int) (name string, has bool) {

View file

@ -2,8 +2,16 @@ package talkgroups
import ( import (
"fmt" "fmt"
"dynatron.me/x/stillbox/pkg/database"
) )
type Talkgroup struct {
database.Talkgroup
System database.System `json:"system"`
Learned bool `json:"learned"`
}
type ID struct { type ID struct {
System uint32 System uint32
Talkgroup uint32 Talkgroup uint32

View file

@ -42,10 +42,10 @@ JOIN systems sys ON tg.system_id = sys.id
WHERE tg.id = systg2id(sqlc.arg(system_id), sqlc.arg(tgid)) WHERE tg.id = systg2id(sqlc.arg(system_id), sqlc.arg(tgid))
UNION UNION
SELECT SELECT
tgl.id::INT8, tgl.system_id::INT4, sys.name system_name, tgl.tgid::INT4, tgl.name, tgl.id::INT8, tgl.system_id::INT4, tgl.tgid::INT4, tgl.name,
tgl.alpha_tag, NULL::INTEGER, NULL::JSONB, tgl.alpha_tag, tgl.alpha_tag, NULL::INTEGER, NULL::JSONB,
CASE WHEN tgl.alpha_tag IS NULL THEN NULL ELSE ARRAY[tgl.alpha_tag] END, tgl.alpha_tag, CASE WHEN tgl.alpha_tag IS NULL THEN NULL ELSE ARRAY[tgl.alpha_tag] END,
TRUE, 1.0, NULL::JSONB, TRUE, NULL::JSONB, 1.0, sys.id, sys.name,
TRUE learned TRUE learned
FROM talkgroups_learned tgl FROM talkgroups_learned tgl
JOIN systems sys ON tgl.system_id = sys.id JOIN systems sys ON tgl.system_id = sys.id
@ -60,10 +60,10 @@ JOIN systems sys ON tg.system_id = sys.id
WHERE tg.id = ANY($1::INT8[]) WHERE tg.id = ANY($1::INT8[])
UNION UNION
SELECT SELECT
tgl.id::INT8, tgl.system_id::INT4, sys.name system_name, tgl.tgid::INT4, tgl.name, tgl.id::INT8, tgl.system_id::INT4, tgl.tgid::INT4, tgl.name,
tgl.alpha_tag, NULL::INTEGER, NULL::JSONB, tgl.alpha_tag, tgl.alpha_tag, NULL::INTEGER, NULL::JSONB,
CASE WHEN tgl.alpha_tag IS NULL THEN NULL ELSE ARRAY[tgl.alpha_tag] END, tgl.alpha_tag, CASE WHEN tgl.alpha_tag IS NULL THEN NULL ELSE ARRAY[tgl.alpha_tag] END,
TRUE, 1.0, NULL::JSONB, TRUE, NULL::JSONB, 1.0, sys.id, sys.name,
TRUE learned TRUE learned
FROM talkgroups_learned tgl FROM talkgroups_learned tgl
JOIN systems sys ON tgl.system_id = sys.id JOIN systems sys ON tgl.system_id = sys.id