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

wip

Browse source
This commit is contained in:
Daniel 2024-11-17 21:46:10 -05:00
parent 584ad46679
commit 705250ad47
23 changed files with 2218 additions and 1762 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

2
.mockery.yaml
View file

@ -7,4 +7,4 @@ packages:
dynatron.me/x/stillbox/pkg/database: dynatron.me/x/stillbox/pkg/database:
config: config:
interfaces: interfaces:
DB: Store:

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

@ -10,12 +10,11 @@ import (
"dynatron.me/x/stillbox/pkg/database" "dynatron.me/x/stillbox/pkg/database"
"dynatron.me/x/stillbox/pkg/talkgroups" "dynatron.me/x/stillbox/pkg/talkgroups"
"github.com/google/uuid"
"github.com/jackc/pgx/v5/pgtype" "github.com/jackc/pgx/v5/pgtype"
) )
type Alert struct { type Alert struct {
ID uuid.UUID ID int
Timestamp time.Time Timestamp time.Time
TGName string TGName string
Score trending.Score[talkgroups.ID] Score trending.Score[talkgroups.ID]
@ -34,7 +33,6 @@ func (a *Alert) ToAddAlertParams() database.AddAlertParams {
} }
return database.AddAlertParams{ return database.AddAlertParams{
ID: a.ID,
Time: pgtype.Timestamptz{Time: a.Timestamp, Valid: true}, Time: pgtype.Timestamptz{Time: a.Timestamp, Valid: true},
SystemID: int(a.Score.ID.System), SystemID: int(a.Score.ID.System),
TGID: int(a.Score.ID.Talkgroup), TGID: int(a.Score.ID.Talkgroup),
@ -48,7 +46,6 @@ func (a *Alert) ToAddAlertParams() database.AddAlertParams {
// Make creates an alert for later rendering or storage. // 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) { func Make(ctx context.Context, store talkgroups.Store, score trending.Score[talkgroups.ID], origScore float64) (Alert, error) {
d := Alert{ d := Alert{
ID: uuid.New(),
Score: score, Score: score,
Timestamp: time.Now(), Timestamp: time.Now(),
Weight: 1.0, Weight: 1.0,

16
pkg/calls/call.go
View file

@ -1,11 +1,13 @@
package calls package calls
import ( import (
"context"
"fmt" "fmt"
"time" "time"
"dynatron.me/x/stillbox/internal/audio" "dynatron.me/x/stillbox/internal/audio"
"dynatron.me/x/stillbox/pkg/auth" "dynatron.me/x/stillbox/pkg/auth"
"dynatron.me/x/stillbox/pkg/database"
"dynatron.me/x/stillbox/pkg/pb" "dynatron.me/x/stillbox/pkg/pb"
"dynatron.me/x/stillbox/pkg/talkgroups" "dynatron.me/x/stillbox/pkg/talkgroups"
@ -111,6 +113,20 @@ func (c *Call) ToPB() *pb.Call {
} }
} }
func (c *Call) LearnTG(ctx context.Context, db database.Store) (learnedId int, err error) {
err = db.AddTalkgroupWithLearnedFlag(ctx, int32(c.System), int32(c.Talkgroup))
if err != nil {
return 0, err
}
return db.AddLearnedTalkgroup(ctx, database.AddLearnedTalkgroupParams{
SystemID: c.System,
TGID: c.Talkgroup,
Name: c.TalkgroupLabel,
AlphaTag: c.TGAlphaTag,
})
}
func (c *Call) computeLength() (err error) { func (c *Call) computeLength() (err error) {
var td time.Duration var td time.Duration

19
pkg/database/calls.sql.go
View file

@ -13,7 +13,7 @@ import (
) )
const addAlert = `-- name: AddAlert :exec const addAlert = `-- name: AddAlert :exec
INSERT INTO alerts (id, time, tgid, system_id, weight, score, orig_score, notified, metadata) INSERT INTO alerts (time, tgid, system_id, weight, score, orig_score, notified, metadata)
VALUES VALUES
( (
$1, $1,
@ -23,13 +23,11 @@ VALUES
$5, $5,
$6, $6,
$7, $7,
$8, $8
$9
) )
` `
type AddAlertParams struct { type AddAlertParams struct {
ID uuid.UUID `json:"id"`
Time pgtype.Timestamptz `json:"time"` Time pgtype.Timestamptz `json:"time"`
TGID int `json:"tgid"` TGID int `json:"tgid"`
SystemID int `json:"system_id"` SystemID int `json:"system_id"`
@ -42,7 +40,6 @@ type AddAlertParams struct {
func (q *Queries) AddAlert(ctx context.Context, arg AddAlertParams) error { func (q *Queries) AddAlert(ctx context.Context, arg AddAlertParams) error {
_, err := q.db.Exec(ctx, addAlert, _, err := q.db.Exec(ctx, addAlert,
arg.ID,
arg.Time, arg.Time,
arg.TGID, arg.TGID,
arg.SystemID, arg.SystemID,
@ -109,9 +106,9 @@ type AddCallParams struct {
Frequency int `json:"frequency"` Frequency int `json:"frequency"`
Frequencies []int `json:"frequencies"` Frequencies []int `json:"frequencies"`
Patches []int `json:"patches"` Patches []int `json:"patches"`
TgLabel *string `json:"tg_label"` TGLabel *string `json:"tg_label"`
TgAlphaTag *string `json:"tg_alpha_tag"` TGAlphaTag *string `json:"tg_alpha_tag"`
TgGroup *string `json:"tg_group"` TGGroup *string `json:"tg_group"`
Source int `json:"source"` Source int `json:"source"`
} }
@ -130,9 +127,9 @@ func (q *Queries) AddCall(ctx context.Context, arg AddCallParams) error {
arg.Frequency, arg.Frequency,
arg.Frequencies, arg.Frequencies,
arg.Patches, arg.Patches,
arg.TgLabel, arg.TGLabel,
arg.TgAlphaTag, arg.TGAlphaTag,
arg.TgGroup, arg.TGGroup,
arg.Source, arg.Source,
) )
return err return err

39
pkg/database/database.go
View file

@ -3,6 +3,7 @@ package database
import ( import (
"context" "context"
"errors" "errors"
"fmt"
"strings" "strings"
"dynatron.me/x/stillbox/pkg/config" "dynatron.me/x/stillbox/pkg/config"
@ -19,11 +20,12 @@ import (
// DB is a database handle. // DB is a database handle.
//go:generate mockery //go:generate mockery
type DB interface { type Store interface {
Querier Querier
talkgroupQuerier talkgroupQuerier
DB() *Database DB() *Database
InTx(context.Context, func(Store) error, pgx.TxOptions) error
} }
type Database struct { type Database struct {
@ -35,14 +37,41 @@ func (db *Database) DB() *Database {
return db return db
} }
func (db *Database) InTx(ctx context.Context, f func(Store) error, opts pgx.TxOptions) error {
tx, err := db.DB().Pool.BeginTx(ctx, opts)
if err != nil {
return fmt.Errorf("Tx begin: %w", err)
}
defer tx.Rollback(ctx)
dbtx := &Database{Pool: db.Pool, Queries: db.Queries.WithTx(tx)}
err = f(dbtx)
if err != nil {
return fmt.Errorf("Tx: %w", err)
}
err = tx.Commit(ctx)
if err != nil {
return fmt.Errorf("Tx commit: %w", err)
}
return nil
}
type dbLogger struct{} type dbLogger struct{}
func (m dbLogger) Log(ctx context.Context, level tracelog.LogLevel, msg string, data map[string]any) { func (m dbLogger) Log(ctx context.Context, level tracelog.LogLevel, msg string, data map[string]any) {
log.Debug().Fields(data).Msg(msg) log.Debug().Fields(data).Msg(msg)
} }
func Close(c Store) {
c.(*Database).Pool.Close()
}
// NewClient creates a new DB using the provided config. // NewClient creates a new DB using the provided config.
func NewClient(ctx context.Context, conf config.DB) (DB, error) { func NewClient(ctx context.Context, conf config.DB) (Store, error) {
dir, err := iofs.New(sqlembed.Migrations, "postgres/migrations") dir, err := iofs.New(sqlembed.Migrations, "postgres/migrations")
if err != nil { if err != nil {
return nil, err return nil, err
@ -90,8 +119,8 @@ type dBCtxKey string
const DBCtxKey dBCtxKey = "dbctx" const DBCtxKey dBCtxKey = "dbctx"
// FromCtx returns the database handle from the provided Context. // FromCtx returns the database handle from the provided Context.
func FromCtx(ctx context.Context) DB { func FromCtx(ctx context.Context) Store {
c, ok := ctx.Value(DBCtxKey).(DB) c, ok := ctx.Value(DBCtxKey).(Store)
if !ok { if !ok {
panic("no DB in context") panic("no DB in context")
} }
@ -100,7 +129,7 @@ func FromCtx(ctx context.Context) DB {
} }
// CtxWithDB returns a Context with the provided database handle. // CtxWithDB returns a Context with the provided database handle.
func CtxWithDB(ctx context.Context, conn DB) context.Context { func CtxWithDB(ctx context.Context, conn Store) context.Context {
return context.WithValue(ctx, DBCtxKey, conn) return context.WithValue(ctx, DBCtxKey, conn)
} }

8
pkg/database/extend.go
View file

@ -2,16 +2,16 @@ package database
func (d GetTalkgroupsRow) GetTalkgroup() Talkgroup { return d.Talkgroup } func (d GetTalkgroupsRow) GetTalkgroup() Talkgroup { return d.Talkgroup }
func (d GetTalkgroupsRow) GetSystem() System { return d.System } func (d GetTalkgroupsRow) GetSystem() System { return d.System }
func (d GetTalkgroupsRow) GetLearned() bool { return d.Learned } func (d GetTalkgroupsRow) GetLearned() bool { return d.Talkgroup.Learned }
func (g GetTalkgroupWithLearnedRow) GetTalkgroup() Talkgroup { return g.Talkgroup } func (g GetTalkgroupWithLearnedRow) GetTalkgroup() Talkgroup { return g.Talkgroup }
func (g GetTalkgroupWithLearnedRow) GetSystem() System { return g.System } func (g GetTalkgroupWithLearnedRow) GetSystem() System { return g.System }
func (g GetTalkgroupWithLearnedRow) GetLearned() bool { return g.Learned } func (g GetTalkgroupWithLearnedRow) GetLearned() bool { return g.Talkgroup.Learned }
func (g GetTalkgroupsWithLearnedRow) GetTalkgroup() Talkgroup { return g.Talkgroup } func (g GetTalkgroupsWithLearnedRow) GetTalkgroup() Talkgroup { return g.Talkgroup }
func (g GetTalkgroupsWithLearnedRow) GetSystem() System { return g.System } func (g GetTalkgroupsWithLearnedRow) GetSystem() System { return g.System }
func (g GetTalkgroupsWithLearnedRow) GetLearned() bool { return g.Learned } func (g GetTalkgroupsWithLearnedRow) GetLearned() bool { return g.Talkgroup.Learned }
func (g GetTalkgroupsWithLearnedBySystemRow) GetTalkgroup() Talkgroup { return g.Talkgroup } func (g GetTalkgroupsWithLearnedBySystemRow) GetTalkgroup() Talkgroup { return g.Talkgroup }
func (g GetTalkgroupsWithLearnedBySystemRow) GetSystem() System { return g.System } func (g GetTalkgroupsWithLearnedBySystemRow) GetSystem() System { return g.System }
func (g GetTalkgroupsWithLearnedBySystemRow) GetLearned() bool { return g.Learned } func (g GetTalkgroupsWithLearnedBySystemRow) GetLearned() bool { return g.Talkgroup.Learned }
func (g Talkgroup) GetTalkgroup() Talkgroup { return g } func (g Talkgroup) GetTalkgroup() Talkgroup { return g }
func (g Talkgroup) GetSystem() System { return System{ID: int(g.SystemID)} } func (g Talkgroup) GetSystem() System { return System{ID: int(g.SystemID)} }
func (g Talkgroup) GetLearned() bool { return false } func (g Talkgroup) GetLearned() bool { return false }

1631
pkg/database/mocks/DB.go
View file

File diff suppressed because it is too large Load diff

1786
pkg/database/mocks/Store.go Normal file
View file

File diff suppressed because it is too large Load diff

26
pkg/database/models.go
View file

@ -14,7 +14,7 @@ import (
) )
type Alert struct { type Alert struct {
ID uuid.UUID `json:"id"` ID int `json:"id"`
Time pgtype.Timestamptz `json:"time"` Time pgtype.Timestamptz `json:"time"`
TGID int `json:"tgid"` TGID int `json:"tgid"`
SystemID int `json:"system_id"` SystemID int `json:"system_id"`
@ -48,9 +48,9 @@ type Call struct {
Frequency int `json:"frequency"` Frequency int `json:"frequency"`
Frequencies []int `json:"frequencies"` Frequencies []int `json:"frequencies"`
Patches []int `json:"patches"` Patches []int `json:"patches"`
TgLabel *string `json:"tg_label"` TGLabel *string `json:"tg_label"`
TgAlphaTag *string `json:"tg_alpha_tag"` TGAlphaTag *string `json:"tg_alpha_tag"`
TgGroup *string `json:"tg_group"` TGGroup *string `json:"tg_group"`
Source int `json:"source"` Source int `json:"source"`
Transcript *string `json:"transcript"` Transcript *string `json:"transcript"`
} }
@ -83,27 +83,29 @@ type System struct {
} }
type Talkgroup struct { type Talkgroup struct {
ID uuid.UUID `json:"id"` ID int `json:"id"`
SystemID int32 `json:"system_id"` SystemID int32 `json:"system_id"`
TGID int32 `json:"tgid"` TGID int32 `json:"tgid"`
Name *string `json:"name"` Name *string `json:"name"`
AlphaTag *string `json:"alpha_tag"` AlphaTag *string `json:"alpha_tag"`
TgGroup *string `json:"tg_group"` TGGroup *string `json:"tg_group"`
Frequency *int32 `json:"frequency"` Frequency *int32 `json:"frequency"`
Metadata jsontypes.Metadata `json:"metadata"` Metadata jsontypes.Metadata `json:"metadata"`
Tags []string `json:"tags"` Tags []string `json:"tags"`
Alert bool `json:"alert"` Alert bool `json:"alert"`
AlertConfig rules.AlertRules `json:"alert_config"` AlertConfig rules.AlertRules `json:"alert_config"`
Weight float32 `json:"weight"` Weight float32 `json:"weight"`
Learned bool `json:"learned"`
} }
type TalkgroupsLearned struct { type TalkgroupsLearned struct {
ID uuid.UUID `json:"id"` ID int `json:"id"`
SystemID int `json:"system_id"` SystemID int `json:"system_id"`
TGID int `json:"tgid"` TGID int `json:"tgid"`
Name string `json:"name"` Name string `json:"name"`
AlphaTag *string `json:"alpha_tag"` AlphaTag *string `json:"alpha_tag"`
Ignored *bool `json:"ignored"` TGGroup *string `json:"tg_group"`
Ignored *bool `json:"ignored"`
} }
type User struct { type User struct {

8
pkg/database/querier.go
View file

@ -14,6 +14,8 @@ import (
type Querier interface { type Querier interface {
AddAlert(ctx context.Context, arg AddAlertParams) error AddAlert(ctx context.Context, arg AddAlertParams) error
AddCall(ctx context.Context, arg AddCallParams) error AddCall(ctx context.Context, arg AddCallParams) error
AddLearnedTalkgroup(ctx context.Context, arg AddLearnedTalkgroupParams) (int, error)
AddTalkgroupWithLearnedFlag(ctx context.Context, systemID int32, tGID int32) error
CreateAPIKey(ctx context.Context, owner int, expires pgtype.Timestamp, disabled *bool) (ApiKey, error) CreateAPIKey(ctx context.Context, owner int, expires pgtype.Timestamp, disabled *bool) (ApiKey, error)
CreateUser(ctx context.Context, arg CreateUserParams) (User, error) CreateUser(ctx context.Context, arg CreateUserParams) (User, error)
DeleteAPIKey(ctx context.Context, apiKey string) error DeleteAPIKey(ctx context.Context, apiKey string) error
@ -21,9 +23,9 @@ type Querier interface {
GetAPIKey(ctx context.Context, apiKey string) (ApiKey, error) GetAPIKey(ctx context.Context, apiKey string) (ApiKey, error)
GetDatabaseSize(ctx context.Context) (string, error) GetDatabaseSize(ctx context.Context) (string, error)
GetSystemName(ctx context.Context, systemID int) (string, error) GetSystemName(ctx context.Context, systemID int) (string, error)
GetTalkgroup(ctx context.Context, systemID int32, tgID int32) (GetTalkgroupRow, error) GetTalkgroup(ctx context.Context, systemID int32, tGID int32) (GetTalkgroupRow, error)
GetTalkgroupIDsByTags(ctx context.Context, anyTags []string, allTags []string, notTags []string) ([]GetTalkgroupIDsByTagsRow, error) GetTalkgroupIDsByTags(ctx context.Context, anyTags []string, allTags []string, notTags []string) ([]GetTalkgroupIDsByTagsRow, error)
GetTalkgroupTags(ctx context.Context, systemID int32, tgID int32) ([]string, error) GetTalkgroupTags(ctx context.Context, systemID int32, tGID int32) ([]string, error)
GetTalkgroupWithLearned(ctx context.Context, systemID int32, tGID int32) (GetTalkgroupWithLearnedRow, error) GetTalkgroupWithLearned(ctx context.Context, systemID int32, tGID int32) (GetTalkgroupWithLearnedRow, error)
GetTalkgroupsWithAllTags(ctx context.Context, tags []string) ([]GetTalkgroupsWithAllTagsRow, error) GetTalkgroupsWithAllTags(ctx context.Context, tags []string) ([]GetTalkgroupsWithAllTagsRow, error)
GetTalkgroupsWithAnyTags(ctx context.Context, tags []string) ([]GetTalkgroupsWithAnyTagsRow, error) GetTalkgroupsWithAnyTags(ctx context.Context, tags []string) ([]GetTalkgroupsWithAnyTagsRow, error)
@ -34,7 +36,7 @@ type Querier interface {
GetUserByUsername(ctx context.Context, username string) (User, error) GetUserByUsername(ctx context.Context, username string) (User, error)
GetUsers(ctx context.Context) ([]User, error) GetUsers(ctx context.Context) ([]User, error)
SetCallTranscript(ctx context.Context, iD uuid.UUID, transcript *string) error SetCallTranscript(ctx context.Context, iD uuid.UUID, transcript *string) error
SetTalkgroupTags(ctx context.Context, tags []string, systemID int32, tgID int32) error SetTalkgroupTags(ctx context.Context, tags []string, systemID int32, tGID int32) error
UpdatePassword(ctx context.Context, username string, password string) error UpdatePassword(ctx context.Context, username string, password string) error
UpdateTalkgroup(ctx context.Context, arg UpdateTalkgroupParams) (Talkgroup, error) UpdateTalkgroup(ctx context.Context, arg UpdateTalkgroupParams) (Talkgroup, error)
} }

28
pkg/database/talkgroups.go
View file

@ -2,6 +2,9 @@ package database
import ( import (
"context" "context"
"errors"
"github.com/jackc/pgx/v5/pgconn"
) )
type talkgroupQuerier interface { type talkgroupQuerier interface {
@ -12,6 +15,17 @@ type talkgroupQuerier interface {
type TGTuples [2][]uint32 type TGTuples [2][]uint32
const TGConstraintName = ""
func IsTGConstraintViolation(e error) bool {
var err *pgconn.PgError
if errors.As(e, &err) && err.Code == "23503" && err.ConstraintName == TGConstraintName {
return true
}
return false
}
func MakeTGTuples(cap int) TGTuples { func MakeTGTuples(cap int) TGTuples {
return [2][]uint32{ return [2][]uint32{
make([]uint32, 0, cap), make([]uint32, 0, cap),
@ -27,11 +41,11 @@ func (t *TGTuples) Append(sys, tg uint32) {
// Below queries are here because sqlc refuses to parse unnest(x, y) // Below queries are here because sqlc refuses to parse unnest(x, y)
const getTalkgroupsWithLearnedBySysTGID = `SELECT const getTalkgroupsWithLearnedBySysTGID = `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, 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, tg.learned
FALSE learned
FROM talkgroups tg FROM talkgroups tg
JOIN systems sys ON tg.system_id = sys.id JOIN systems sys ON tg.system_id = sys.id
JOIN UNNEST($1::INT4[], $2::INT4[]) AS tgt(sys, tg) ON (tg.system_id = tgt.sys AND tg.tgid = tgt.tg) JOIN UNNEST($1::INT4[], $2::INT4[]) AS tgt(sys, tg) ON (tg.system_id = tgt.sys AND tg.tgid = tgt.tg)
WHERE tg.learned IS NOT TRUE
UNION UNION
SELECT SELECT
tgl.id, tgl.system_id::INT4, tgl.tgid::INT4, tgl.name, tgl.id, tgl.system_id::INT4, tgl.tgid::INT4, tgl.name,
@ -46,7 +60,6 @@ JOIN UNNEST($1::INT4[], $2::INT4[]) AS tgt(sys, tg) ON (tgl.system_id = tgt.sys
type GetTalkgroupsRow struct { type GetTalkgroupsRow struct {
Talkgroup Talkgroup `json:"talkgroup"` Talkgroup Talkgroup `json:"talkgroup"`
System System `json:"system"` System System `json:"system"`
Learned bool `json:"learned"`
} }
func (q *Queries) GetTalkgroupsWithLearnedBySysTGID(ctx context.Context, ids TGTuples) ([]GetTalkgroupsRow, error) { func (q *Queries) GetTalkgroupsWithLearnedBySysTGID(ctx context.Context, ids TGTuples) ([]GetTalkgroupsRow, error) {
@ -64,7 +77,7 @@ func (q *Queries) GetTalkgroupsWithLearnedBySysTGID(ctx context.Context, ids TGT
&i.Talkgroup.TGID, &i.Talkgroup.TGID,
&i.Talkgroup.Name, &i.Talkgroup.Name,
&i.Talkgroup.AlphaTag, &i.Talkgroup.AlphaTag,
&i.Talkgroup.TgGroup, &i.Talkgroup.TGGroup,
&i.Talkgroup.Frequency, &i.Talkgroup.Frequency,
&i.Talkgroup.Metadata, &i.Talkgroup.Metadata,
&i.Talkgroup.Tags, &i.Talkgroup.Tags,
@ -73,7 +86,7 @@ func (q *Queries) GetTalkgroupsWithLearnedBySysTGID(ctx context.Context, ids TGT
&i.Talkgroup.Weight, &i.Talkgroup.Weight,
&i.System.ID, &i.System.ID,
&i.System.Name, &i.System.Name,
&i.Learned, &i.Talkgroup.Learned,
); err != nil { ); err != nil {
return nil, err return nil, err
} }
@ -87,7 +100,8 @@ func (q *Queries) GetTalkgroupsWithLearnedBySysTGID(ctx context.Context, ids TGT
const getTalkgroupsBySysTGID = `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 FROM talkgroups tg const getTalkgroupsBySysTGID = `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 FROM talkgroups tg
JOIN systems sys ON tg.system_id = sys.id JOIN systems sys ON tg.system_id = sys.id
JOIN UNNEST($1::INT4[], $2::INT4[]) AS tgt(sys, tg) ON (tg.system_id = tgt.sys AND tg.tgid = tgt.tg);` JOIN UNNEST($1::INT4[], $2::INT4[]) AS tgt(sys, tg) ON (tg.system_id = tgt.sys AND tg.tgid = tgt.tg)
WHERE tg.learned IS NOT TRUE;`
func (q *Queries) GetTalkgroupsBySysTGID(ctx context.Context, ids TGTuples) ([]GetTalkgroupsRow, error) { func (q *Queries) GetTalkgroupsBySysTGID(ctx context.Context, ids TGTuples) ([]GetTalkgroupsRow, error) {
rows, err := q.db.Query(ctx, getTalkgroupsBySysTGID, ids[0], ids[1]) rows, err := q.db.Query(ctx, getTalkgroupsBySysTGID, ids[0], ids[1])
@ -104,7 +118,7 @@ func (q *Queries) GetTalkgroupsBySysTGID(ctx context.Context, ids TGTuples) ([]G
&i.Talkgroup.TGID, &i.Talkgroup.TGID,
&i.Talkgroup.Name, &i.Talkgroup.Name,
&i.Talkgroup.AlphaTag, &i.Talkgroup.AlphaTag,
&i.Talkgroup.TgGroup, &i.Talkgroup.TGGroup,
&i.Talkgroup.Frequency, &i.Talkgroup.Frequency,
&i.Talkgroup.Metadata, &i.Talkgroup.Metadata,
&i.Talkgroup.Tags, &i.Talkgroup.Tags,

122
pkg/database/talkgroups.sql.go
View file

@ -10,9 +10,62 @@ import (
"dynatron.me/x/stillbox/internal/jsontypes" "dynatron.me/x/stillbox/internal/jsontypes"
"dynatron.me/x/stillbox/pkg/alerting/rules" "dynatron.me/x/stillbox/pkg/alerting/rules"
"github.com/jackc/pgx/v5/pgtype"
) )
const addLearnedTalkgroup = `-- name: AddLearnedTalkgroup :one
INSERT INTO talkgroups_learned(
system_id,
tgid,
name,
alpha_tag,
tg_group
) VALUES (
$1,
$2,
$3,
$4,
$5
) RETURNING id
`
type AddLearnedTalkgroupParams struct {
SystemID int `json:"system_id"`
TGID int `json:"tgid"`
Name *string `json:"name"`
AlphaTag *string `json:"alpha_tag"`
TGGroup *string `json:"tg_group"`
}
func (q *Queries) AddLearnedTalkgroup(ctx context.Context, arg AddLearnedTalkgroupParams) (int, error) {
row := q.db.QueryRow(ctx, addLearnedTalkgroup,
arg.SystemID,
arg.TGID,
arg.Name,
arg.AlphaTag,
arg.TGGroup,
)
var id int
err := row.Scan(&id)
return id, err
}
const addTalkgroupWithLearnedFlag = `-- name: AddTalkgroupWithLearnedFlag :exec
INSERT INTO talkgroups (
system_id,
tgid,
learned
) VALUES(
$1,
$2,
't'
)
`
func (q *Queries) AddTalkgroupWithLearnedFlag(ctx context.Context, systemID int32, tGID int32) error {
_, err := q.db.Exec(ctx, addTalkgroupWithLearnedFlag, systemID, tGID)
return err
}
const getSystemName = `-- name: GetSystemName :one const getSystemName = `-- name: GetSystemName :one
SELECT name FROM systems WHERE id = $1 SELECT name FROM systems WHERE id = $1
` `
@ -25,7 +78,7 @@ func (q *Queries) GetSystemName(ctx context.Context, systemID int) (string, erro
} }
const getTalkgroup = `-- name: GetTalkgroup :one const getTalkgroup = `-- name: GetTalkgroup :one
SELECT talkgroups.id, talkgroups.system_id, talkgroups.tgid, talkgroups.name, talkgroups.alpha_tag, talkgroups.tg_group, talkgroups.frequency, talkgroups.metadata, talkgroups.tags, talkgroups.alert, talkgroups.alert_config, talkgroups.weight FROM talkgroups SELECT talkgroups.id, talkgroups.system_id, talkgroups.tgid, talkgroups.name, talkgroups.alpha_tag, talkgroups.tg_group, talkgroups.frequency, talkgroups.metadata, talkgroups.tags, talkgroups.alert, talkgroups.alert_config, talkgroups.weight, talkgroups.learned FROM talkgroups
WHERE (system_id, tgid) = ($1, $2) WHERE (system_id, tgid) = ($1, $2)
` `
@ -33,8 +86,8 @@ type GetTalkgroupRow struct {
Talkgroup Talkgroup `json:"talkgroup"` Talkgroup Talkgroup `json:"talkgroup"`
} }
func (q *Queries) GetTalkgroup(ctx context.Context, systemID int32, tgID int32) (GetTalkgroupRow, error) { func (q *Queries) GetTalkgroup(ctx context.Context, systemID int32, tGID int32) (GetTalkgroupRow, error) {
row := q.db.QueryRow(ctx, getTalkgroup, systemID, tgID) row := q.db.QueryRow(ctx, getTalkgroup, systemID, tGID)
var i GetTalkgroupRow var i GetTalkgroupRow
err := row.Scan( err := row.Scan(
&i.Talkgroup.ID, &i.Talkgroup.ID,
@ -42,13 +95,14 @@ func (q *Queries) GetTalkgroup(ctx context.Context, systemID int32, tgID int32)
&i.Talkgroup.TGID, &i.Talkgroup.TGID,
&i.Talkgroup.Name, &i.Talkgroup.Name,
&i.Talkgroup.AlphaTag, &i.Talkgroup.AlphaTag,
&i.Talkgroup.TgGroup, &i.Talkgroup.TGGroup,
&i.Talkgroup.Frequency, &i.Talkgroup.Frequency,
&i.Talkgroup.Metadata, &i.Talkgroup.Metadata,
&i.Talkgroup.Tags, &i.Talkgroup.Tags,
&i.Talkgroup.Alert, &i.Talkgroup.Alert,
&i.Talkgroup.AlertConfig, &i.Talkgroup.AlertConfig,
&i.Talkgroup.Weight, &i.Talkgroup.Weight,
&i.Talkgroup.Learned,
) )
return i, err return i, err
} }
@ -90,8 +144,8 @@ SELECT tags FROM talkgroups
WHERE system_id = $1 AND tgid = $2 WHERE system_id = $1 AND tgid = $2
` `
func (q *Queries) GetTalkgroupTags(ctx context.Context, systemID int32, tgID int32) ([]string, error) { func (q *Queries) GetTalkgroupTags(ctx context.Context, systemID int32, tGID int32) ([]string, error) {
row := q.db.QueryRow(ctx, getTalkgroupTags, systemID, tgID) row := q.db.QueryRow(ctx, getTalkgroupTags, systemID, tGID)
var tags []string var tags []string
err := row.Scan(&tags) err := row.Scan(&tags)
return tags, err return tags, err
@ -99,11 +153,10 @@ func (q *Queries) GetTalkgroupTags(ctx context.Context, systemID int32, tgID int
const getTalkgroupWithLearned = `-- name: GetTalkgroupWithLearned :one const getTalkgroupWithLearned = `-- name: GetTalkgroupWithLearned :one
SELECT 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, 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, tg.learned, sys.id, sys.name
FALSE learned
FROM talkgroups tg FROM talkgroups tg
JOIN systems sys ON tg.system_id = sys.id JOIN systems sys ON tg.system_id = sys.id
WHERE (tg.system_id, tg.tgid) = ($1, $2) WHERE (tg.system_id, tg.tgid) = ($1, $2) AND tg.learned IS NOT TRUE
UNION UNION
SELECT SELECT
tgl.id, tgl.system_id::INT4, tgl.tgid::INT4, tgl.name, tgl.id, tgl.system_id::INT4, tgl.tgid::INT4, tgl.name,
@ -119,7 +172,6 @@ WHERE tgl.system_id = $1 AND tgl.tgid = $2 AND ignored IS NOT TRUE
type GetTalkgroupWithLearnedRow struct { type GetTalkgroupWithLearnedRow struct {
Talkgroup Talkgroup `json:"talkgroup"` Talkgroup Talkgroup `json:"talkgroup"`
System System `json:"system"` System System `json:"system"`
Learned bool `json:"learned"`
} }
func (q *Queries) GetTalkgroupWithLearned(ctx context.Context, systemID int32, tGID int32) (GetTalkgroupWithLearnedRow, error) { func (q *Queries) GetTalkgroupWithLearned(ctx context.Context, systemID int32, tGID int32) (GetTalkgroupWithLearnedRow, error) {
@ -131,22 +183,22 @@ func (q *Queries) GetTalkgroupWithLearned(ctx context.Context, systemID int32, t
&i.Talkgroup.TGID, &i.Talkgroup.TGID,
&i.Talkgroup.Name, &i.Talkgroup.Name,
&i.Talkgroup.AlphaTag, &i.Talkgroup.AlphaTag,
&i.Talkgroup.TgGroup, &i.Talkgroup.TGGroup,
&i.Talkgroup.Frequency, &i.Talkgroup.Frequency,
&i.Talkgroup.Metadata, &i.Talkgroup.Metadata,
&i.Talkgroup.Tags, &i.Talkgroup.Tags,
&i.Talkgroup.Alert, &i.Talkgroup.Alert,
&i.Talkgroup.AlertConfig, &i.Talkgroup.AlertConfig,
&i.Talkgroup.Weight, &i.Talkgroup.Weight,
&i.Talkgroup.Learned,
&i.System.ID, &i.System.ID,
&i.System.Name, &i.System.Name,
&i.Learned,
) )
return i, err return i, err
} }
const getTalkgroupsWithAllTags = `-- name: GetTalkgroupsWithAllTags :many const getTalkgroupsWithAllTags = `-- name: GetTalkgroupsWithAllTags :many
SELECT talkgroups.id, talkgroups.system_id, talkgroups.tgid, talkgroups.name, talkgroups.alpha_tag, talkgroups.tg_group, talkgroups.frequency, talkgroups.metadata, talkgroups.tags, talkgroups.alert, talkgroups.alert_config, talkgroups.weight FROM talkgroups SELECT talkgroups.id, talkgroups.system_id, talkgroups.tgid, talkgroups.name, talkgroups.alpha_tag, talkgroups.tg_group, talkgroups.frequency, talkgroups.metadata, talkgroups.tags, talkgroups.alert, talkgroups.alert_config, talkgroups.weight, talkgroups.learned FROM talkgroups
WHERE tags && ARRAY[$1] WHERE tags && ARRAY[$1]
` `
@ -169,13 +221,14 @@ func (q *Queries) GetTalkgroupsWithAllTags(ctx context.Context, tags []string) (
&i.Talkgroup.TGID, &i.Talkgroup.TGID,
&i.Talkgroup.Name, &i.Talkgroup.Name,
&i.Talkgroup.AlphaTag, &i.Talkgroup.AlphaTag,
&i.Talkgroup.TgGroup, &i.Talkgroup.TGGroup,
&i.Talkgroup.Frequency, &i.Talkgroup.Frequency,
&i.Talkgroup.Metadata, &i.Talkgroup.Metadata,
&i.Talkgroup.Tags, &i.Talkgroup.Tags,
&i.Talkgroup.Alert, &i.Talkgroup.Alert,
&i.Talkgroup.AlertConfig, &i.Talkgroup.AlertConfig,
&i.Talkgroup.Weight, &i.Talkgroup.Weight,
&i.Talkgroup.Learned,
); err != nil { ); err != nil {
return nil, err return nil, err
} }
@ -188,7 +241,7 @@ func (q *Queries) GetTalkgroupsWithAllTags(ctx context.Context, tags []string) (
} }
const getTalkgroupsWithAnyTags = `-- name: GetTalkgroupsWithAnyTags :many const getTalkgroupsWithAnyTags = `-- name: GetTalkgroupsWithAnyTags :many
SELECT talkgroups.id, talkgroups.system_id, talkgroups.tgid, talkgroups.name, talkgroups.alpha_tag, talkgroups.tg_group, talkgroups.frequency, talkgroups.metadata, talkgroups.tags, talkgroups.alert, talkgroups.alert_config, talkgroups.weight FROM talkgroups SELECT talkgroups.id, talkgroups.system_id, talkgroups.tgid, talkgroups.name, talkgroups.alpha_tag, talkgroups.tg_group, talkgroups.frequency, talkgroups.metadata, talkgroups.tags, talkgroups.alert, talkgroups.alert_config, talkgroups.weight, talkgroups.learned FROM talkgroups
WHERE tags @> ARRAY[$1] WHERE tags @> ARRAY[$1]
` `
@ -211,13 +264,14 @@ func (q *Queries) GetTalkgroupsWithAnyTags(ctx context.Context, tags []string) (
&i.Talkgroup.TGID, &i.Talkgroup.TGID,
&i.Talkgroup.Name, &i.Talkgroup.Name,
&i.Talkgroup.AlphaTag, &i.Talkgroup.AlphaTag,
&i.Talkgroup.TgGroup, &i.Talkgroup.TGGroup,
&i.Talkgroup.Frequency, &i.Talkgroup.Frequency,
&i.Talkgroup.Metadata, &i.Talkgroup.Metadata,
&i.Talkgroup.Tags, &i.Talkgroup.Tags,
&i.Talkgroup.Alert, &i.Talkgroup.Alert,
&i.Talkgroup.AlertConfig, &i.Talkgroup.AlertConfig,
&i.Talkgroup.Weight, &i.Talkgroup.Weight,
&i.Talkgroup.Learned,
); err != nil { ); err != nil {
return nil, err return nil, err
} }
@ -231,10 +285,10 @@ func (q *Queries) GetTalkgroupsWithAnyTags(ctx context.Context, tags []string) (
const getTalkgroupsWithLearned = `-- name: GetTalkgroupsWithLearned :many const getTalkgroupsWithLearned = `-- name: GetTalkgroupsWithLearned :many
SELECT 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, 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, tg.learned, sys.id, sys.name
FALSE learned
FROM talkgroups tg FROM talkgroups tg
JOIN systems sys ON tg.system_id = sys.id JOIN systems sys ON tg.system_id = sys.id
WHERE tg.learned IS NOT TRUE
UNION UNION
SELECT SELECT
tgl.id, tgl.system_id::INT4, tgl.tgid::INT4, tgl.name, tgl.id, tgl.system_id::INT4, tgl.tgid::INT4, tgl.name,
@ -250,7 +304,6 @@ WHERE ignored IS NOT TRUE
type GetTalkgroupsWithLearnedRow struct { type GetTalkgroupsWithLearnedRow struct {
Talkgroup Talkgroup `json:"talkgroup"` Talkgroup Talkgroup `json:"talkgroup"`
System System `json:"system"` System System `json:"system"`
Learned bool `json:"learned"`
} }
func (q *Queries) GetTalkgroupsWithLearned(ctx context.Context) ([]GetTalkgroupsWithLearnedRow, error) { func (q *Queries) GetTalkgroupsWithLearned(ctx context.Context) ([]GetTalkgroupsWithLearnedRow, error) {
@ -268,16 +321,16 @@ func (q *Queries) GetTalkgroupsWithLearned(ctx context.Context) ([]GetTalkgroups
&i.Talkgroup.TGID, &i.Talkgroup.TGID,
&i.Talkgroup.Name, &i.Talkgroup.Name,
&i.Talkgroup.AlphaTag, &i.Talkgroup.AlphaTag,
&i.Talkgroup.TgGroup, &i.Talkgroup.TGGroup,
&i.Talkgroup.Frequency, &i.Talkgroup.Frequency,
&i.Talkgroup.Metadata, &i.Talkgroup.Metadata,
&i.Talkgroup.Tags, &i.Talkgroup.Tags,
&i.Talkgroup.Alert, &i.Talkgroup.Alert,
&i.Talkgroup.AlertConfig, &i.Talkgroup.AlertConfig,
&i.Talkgroup.Weight, &i.Talkgroup.Weight,
&i.Talkgroup.Learned,
&i.System.ID, &i.System.ID,
&i.System.Name, &i.System.Name,
&i.Learned,
); err != nil { ); err != nil {
return nil, err return nil, err
} }
@ -291,11 +344,10 @@ func (q *Queries) GetTalkgroupsWithLearned(ctx context.Context) ([]GetTalkgroups
const getTalkgroupsWithLearnedBySystem = `-- name: GetTalkgroupsWithLearnedBySystem :many const getTalkgroupsWithLearnedBySystem = `-- name: GetTalkgroupsWithLearnedBySystem :many
SELECT 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, 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, tg.learned, sys.id, sys.name
FALSE learned
FROM talkgroups tg FROM talkgroups tg
JOIN systems sys ON tg.system_id = sys.id JOIN systems sys ON tg.system_id = sys.id
WHERE tg.system_id = $1 WHERE tg.system_id = $1 AND tg.learned IS NOT TRUE
UNION UNION
SELECT SELECT
tgl.id, tgl.system_id::INT4, tgl.tgid::INT4, tgl.name, tgl.id, tgl.system_id::INT4, tgl.tgid::INT4, tgl.name,
@ -311,7 +363,6 @@ WHERE tgl.system_id = $1 AND ignored IS NOT TRUE
type GetTalkgroupsWithLearnedBySystemRow struct { type GetTalkgroupsWithLearnedBySystemRow struct {
Talkgroup Talkgroup `json:"talkgroup"` Talkgroup Talkgroup `json:"talkgroup"`
System System `json:"system"` System System `json:"system"`
Learned bool `json:"learned"`
} }
func (q *Queries) GetTalkgroupsWithLearnedBySystem(ctx context.Context, system int32) ([]GetTalkgroupsWithLearnedBySystemRow, error) { func (q *Queries) GetTalkgroupsWithLearnedBySystem(ctx context.Context, system int32) ([]GetTalkgroupsWithLearnedBySystemRow, error) {
@ -329,16 +380,16 @@ func (q *Queries) GetTalkgroupsWithLearnedBySystem(ctx context.Context, system i
&i.Talkgroup.TGID, &i.Talkgroup.TGID,
&i.Talkgroup.Name, &i.Talkgroup.Name,
&i.Talkgroup.AlphaTag, &i.Talkgroup.AlphaTag,
&i.Talkgroup.TgGroup, &i.Talkgroup.TGGroup,
&i.Talkgroup.Frequency, &i.Talkgroup.Frequency,
&i.Talkgroup.Metadata, &i.Talkgroup.Metadata,
&i.Talkgroup.Tags, &i.Talkgroup.Tags,
&i.Talkgroup.Alert, &i.Talkgroup.Alert,
&i.Talkgroup.AlertConfig, &i.Talkgroup.AlertConfig,
&i.Talkgroup.Weight, &i.Talkgroup.Weight,
&i.Talkgroup.Learned,
&i.System.ID, &i.System.ID,
&i.System.Name, &i.System.Name,
&i.Learned,
); err != nil { ); err != nil {
return nil, err return nil, err
} }
@ -355,8 +406,8 @@ UPDATE talkgroups SET tags = $1
WHERE system_id = $2 AND tgid = $3 WHERE system_id = $2 AND tgid = $3
` `
func (q *Queries) SetTalkgroupTags(ctx context.Context, tags []string, systemID int32, tgID int32) error { func (q *Queries) SetTalkgroupTags(ctx context.Context, tags []string, systemID int32, tGID int32) error {
_, err := q.db.Exec(ctx, setTalkgroupTags, tags, systemID, tgID) _, err := q.db.Exec(ctx, setTalkgroupTags, tags, systemID, tGID)
return err return err
} }
@ -373,20 +424,20 @@ SET
alert_config = COALESCE($8, alert_config), alert_config = COALESCE($8, alert_config),
weight = COALESCE($9, weight) weight = COALESCE($9, weight)
WHERE id = $10 OR (system_id = $11 AND tgid = $12) WHERE id = $10 OR (system_id = $11 AND tgid = $12)
RETURNING id, system_id, tgid, name, alpha_tag, tg_group, frequency, metadata, tags, alert, alert_config, weight RETURNING id, system_id, tgid, name, alpha_tag, tg_group, frequency, metadata, tags, alert, alert_config, weight, learned
` `
type UpdateTalkgroupParams struct { type UpdateTalkgroupParams struct {
Name *string `json:"name"` Name *string `json:"name"`
AlphaTag *string `json:"alpha_tag"` AlphaTag *string `json:"alpha_tag"`
TgGroup *string `json:"tg_group"` TGGroup *string `json:"tg_group"`
Frequency *int32 `json:"frequency"` Frequency *int32 `json:"frequency"`
Metadata jsontypes.Metadata `json:"metadata"` Metadata jsontypes.Metadata `json:"metadata"`
Tags []string `json:"tags"` Tags []string `json:"tags"`
Alert *bool `json:"alert"` Alert *bool `json:"alert"`
AlertConfig rules.AlertRules `json:"alert_config"` AlertConfig rules.AlertRules `json:"alert_config"`
Weight *float32 `json:"weight"` Weight *float32 `json:"weight"`
ID pgtype.UUID `json:"id"` ID *int32 `json:"id"`
SystemID *int32 `json:"system_id"` SystemID *int32 `json:"system_id"`
TGID *int32 `json:"tgid"` TGID *int32 `json:"tgid"`
} }
@ -395,7 +446,7 @@ func (q *Queries) UpdateTalkgroup(ctx context.Context, arg UpdateTalkgroupParams
row := q.db.QueryRow(ctx, updateTalkgroup, row := q.db.QueryRow(ctx, updateTalkgroup,
arg.Name, arg.Name,
arg.AlphaTag, arg.AlphaTag,
arg.TgGroup, arg.TGGroup,
arg.Frequency, arg.Frequency,
arg.Metadata, arg.Metadata,
arg.Tags, arg.Tags,
@ -413,13 +464,14 @@ func (q *Queries) UpdateTalkgroup(ctx context.Context, arg UpdateTalkgroupParams
&i.TGID, &i.TGID,
&i.Name, &i.Name,
&i.AlphaTag, &i.AlphaTag,
&i.TgGroup, &i.TGGroup,
&i.Frequency, &i.Frequency,
&i.Metadata, &i.Metadata,
&i.Tags, &i.Tags,
&i.Alert, &i.Alert,
&i.AlertConfig, &i.AlertConfig,
&i.Weight, &i.Weight,
&i.Learned,
) )
return i, err return i, err
} }

2
pkg/nexus/commands.go
View file

@ -78,7 +78,7 @@ func (c *client) Talkgroup(ctx context.Context, tg *pb.Talkgroup) error {
resp := &pb.TalkgroupInfo{ resp := &pb.TalkgroupInfo{
Tg: tg, Tg: tg,
Name: tgi.Talkgroup.Name, Name: tgi.Talkgroup.Name,
Group: tgi.Talkgroup.TgGroup, Group: tgi.Talkgroup.TGGroup,
Frequency: tgi.Talkgroup.Frequency, Frequency: tgi.Talkgroup.Frequency,
Metadata: md, Metadata: md,
Tags: tgi.Talkgroup.Tags, Tags: tgi.Talkgroup.Tags,

6
pkg/server/server.go
View file

@ -27,7 +27,7 @@ const shutdownTimeout = 5 * time.Second
type Server struct { type Server struct {
auth *auth.Auth auth *auth.Auth
conf *config.Config conf *config.Config
db database.DB db database.Store
r *chi.Mux r *chi.Mux
sources sources.Sources sources sources.Sources
sinks sinks.Sinks sinks sinks.Sinks
@ -78,7 +78,7 @@ func New(ctx context.Context, cfg *config.Config) (*Server, error) {
rest: api, rest: api,
} }
srv.sinks.Register("database", sinks.NewDatabaseSink(srv.db), true) srv.sinks.Register("database", sinks.NewDatabaseSink(), true)
srv.sinks.Register("nexus", sinks.NewNexusSink(srv.nex), false) srv.sinks.Register("nexus", sinks.NewNexusSink(srv.nex), false)
if srv.alerter.Enabled() { if srv.alerter.Enabled() {
@ -112,7 +112,7 @@ func New(ctx context.Context, cfg *config.Config) (*Server, error) {
} }
func (s *Server) Go(ctx context.Context) error { func (s *Server) Go(ctx context.Context) error {
defer s.db.DB().Close() defer database.Close(s.db)
s.installHupHandler() s.installHupHandler()

38
pkg/sinks/database.go
View file

@ -8,16 +8,16 @@ import (
"dynatron.me/x/stillbox/pkg/calls" "dynatron.me/x/stillbox/pkg/calls"
"dynatron.me/x/stillbox/pkg/database" "dynatron.me/x/stillbox/pkg/database"
"github.com/jackc/pgx/v5"
"github.com/jackc/pgx/v5/pgtype" "github.com/jackc/pgx/v5/pgtype"
"github.com/rs/zerolog/log" "github.com/rs/zerolog/log"
) )
type DatabaseSink struct { type DatabaseSink struct {
db database.DB
} }
func NewDatabaseSink(db database.DB) *DatabaseSink { func NewDatabaseSink() *DatabaseSink {
return &DatabaseSink{db: db} return &DatabaseSink{}
} }
func (s *DatabaseSink) Call(ctx context.Context, call *calls.Call) error { func (s *DatabaseSink) Call(ctx context.Context, call *calls.Call) error {
@ -26,14 +26,28 @@ func (s *DatabaseSink) Call(ctx context.Context, call *calls.Call) error {
return nil return nil
} }
err := s.db.AddCall(ctx, s.toAddCallParams(call)) return database.FromCtx(ctx).InTx(ctx, func(tx database.Store) error {
if err != nil { params := s.toAddCallParams(call)
return fmt.Errorf("add call: %w", err) err := tx.AddCall(ctx, params)
} if err != nil {
if database.IsTGConstraintViolation(err) {
_, err := call.LearnTG(ctx, tx)
if err != nil {
return fmt.Errorf("add call: learn tg: %w", err)
}
log.Debug().Str("id", call.ID.String()).Int("system", call.System).Int("tgid", call.Talkgroup).Msg("stored") err = tx.AddCall(ctx, params)
if err != nil {
return fmt.Errorf("add call: retry: %w", err)
}
}
return fmt.Errorf("add call: %w", err)
}
return nil log.Debug().Str("id", call.ID.String()).Int("system", call.System).Int("tgid", call.Talkgroup).Msg("stored")
return nil
}, pgx.TxOptions{})
} }
func (s *DatabaseSink) SinkType() string { func (s *DatabaseSink) SinkType() string {
@ -54,9 +68,9 @@ func (s *DatabaseSink) toAddCallParams(call *calls.Call) database.AddCallParams
Frequency: call.Frequency, Frequency: call.Frequency,
Frequencies: call.Frequencies, Frequencies: call.Frequencies,
Patches: call.Patches, Patches: call.Patches,
TgLabel: call.TalkgroupLabel, TGLabel: call.TalkgroupLabel,
TgAlphaTag: call.TGAlphaTag, TGAlphaTag: call.TGAlphaTag,
TgGroup: call.TalkgroupGroup, TGGroup: call.TalkgroupGroup,
Source: call.Source, Source: call.Source,
} }
} }

5
pkg/talkgroups/importer/import.go
View file

@ -10,8 +10,6 @@ import (
"strconv" "strconv"
"strings" "strings"
"github.com/google/uuid"
"dynatron.me/x/stillbox/internal/jsontypes" "dynatron.me/x/stillbox/internal/jsontypes"
"dynatron.me/x/stillbox/pkg/database" "dynatron.me/x/stillbox/pkg/database"
"dynatron.me/x/stillbox/pkg/talkgroups" "dynatron.me/x/stillbox/pkg/talkgroups"
@ -112,12 +110,11 @@ func (rr *radioReferenceImporter) importTalkgroups(ctx context.Context, sys int,
gn := groupName // must take a copy gn := groupName // must take a copy
tgs = append(tgs, talkgroups.Talkgroup{ tgs = append(tgs, talkgroups.Talkgroup{
Talkgroup: database.Talkgroup{ Talkgroup: database.Talkgroup{
ID: uuid.New(),
TGID: int32(tgt.Talkgroup), TGID: int32(tgt.Talkgroup),
SystemID: int32(tgt.System), SystemID: int32(tgt.System),
Name: &fields[4], Name: &fields[4],
AlphaTag: &fields[3], AlphaTag: &fields[3],
TgGroup: &gn, TGGroup: &gn,
Metadata: metadata, Metadata: metadata,
Tags: tags, Tags: tags,
Weight: 1.0, Weight: 1.0,

20
sql/postgres/migrations/001_initial.up.sql
View file

@ -49,6 +49,7 @@ CREATE TABLE IF NOT EXISTS talkgroups_learned(
tgid INTEGER NOT NULL, tgid INTEGER NOT NULL,
name TEXT NOT NULL, name TEXT NOT NULL,
alpha_tag TEXT, alpha_tag TEXT,
tg_group TEXT,
ignored BOOLEAN, ignored BOOLEAN,
UNIQUE (system_id, tgid, name) UNIQUE (system_id, tgid, name)
); );
@ -65,22 +66,6 @@ CREATE TABLE IF NOT EXISTS alerts(
metadata JSONB metadata JSONB
); );
CREATE OR REPLACE FUNCTION learn_talkgroup()
RETURNS TRIGGER AS $$
BEGIN
IF NOT EXISTS (
SELECT tg.system_id, tg.tgid, tg.name, tg.alpha_tag FROM talkgroups tg WHERE tg.system_id = NEW.system AND tg.tgid = NEW.talkgroup
UNION
SELECT tgl.system_id, tgl.tgid, tgl.name, tgl.alpha_tag FROM talkgroups_learned tgl WHERE tgl.system_id = NEW.system AND tgl.tgid = NEW.talkgroup
) THEN
INSERT INTO talkgroups_learned(system_id, tgid, name, alpha_tag) VALUES(
NEW.system, NEW.talkgroup, NEW.tg_label, NEW.tg_alpha_tag
) ON CONFLICT DO NOTHING;
END IF;
RETURN NEW;
END
$$ LANGUAGE plpgsql;
CREATE TABLE IF NOT EXISTS calls( CREATE TABLE IF NOT EXISTS calls(
id UUID PRIMARY KEY, id UUID PRIMARY KEY,
submitter INTEGER REFERENCES api_keys(id) ON DELETE SET NULL, submitter INTEGER REFERENCES api_keys(id) ON DELETE SET NULL,
@ -102,9 +87,6 @@ CREATE TABLE IF NOT EXISTS calls(
transcript TEXT transcript TEXT
); );
CREATE OR REPLACE TRIGGER learn_tg AFTER INSERT ON calls
FOR EACH ROW EXECUTE FUNCTION learn_talkgroup();
CREATE INDEX IF NOT EXISTS calls_transcript_idx ON calls USING GIN (to_tsvector('english', transcript)); CREATE INDEX IF NOT EXISTS calls_transcript_idx ON calls USING GIN (to_tsvector('english', transcript));
CREATE INDEX IF NOT EXISTS calls_call_date_tg_idx ON calls(system, talkgroup, call_date); CREATE INDEX IF NOT EXISTS calls_call_date_tg_idx ON calls(system, talkgroup, call_date);

34
sql/postgres/migrations/002_tglearned.down.sql Normal file
View file

@ -0,0 +1,34 @@
ALTER TABLE calls DROP CONSTRAINT IF EXISTS calls_talkgroup_id_fkey;
ALTER TABLE talkgroups ALTER COLUMN id DROP IDENTITY IF EXISTS;
ALTER TABLE talkgroups_learned ALTER COLUMN id SET DATA TYPE UUID USING (gen_random_uuid());
DROP SEQUENCE IF EXISTS talkgroups_id_seq;
ALTER TABLE talkgroups DROP COLUMN IF EXISTS learned;
CREATE OR REPLACE FUNCTION learn_talkgroup()
RETURNS TRIGGER AS $$
BEGIN
IF NOT EXISTS (
SELECT tg.system_id, tg.tgid, tg.name, tg.alpha_tag FROM talkgroups tg WHERE tg.system_id = NEW.system AND tg.tgid = NEW.talkgroup
UNION
SELECT tgl.system_id, tgl.tgid, tgl.name, tgl.alpha_tag FROM talkgroups_learned tgl WHERE tgl.system_id = NEW.system AND tgl.tgid = NEW.talkgroup
) THEN
INSERT INTO talkgroups_learned(system_id, tgid, name, alpha_tag) VALUES(
NEW.system, NEW.talkgroup, NEW.tg_label, NEW.tg_alpha_tag
) ON CONFLICT DO NOTHING;
END IF;
RETURN NEW;
END
$$ LANGUAGE plpgsql;
CREATE OR REPLACE TRIGGER learn_tg AFTER INSERT ON calls FOR EACH ROW EXECUTE FUNCTION learn_talkgroup();
ALTER TABLE talkgroups_learned ALTER COLUMN id DROP IDENTITY IF EXISTS;
ALTER TABLE talkgroups_learned ALTER COLUMN id SET DATA TYPE UUID USING (gen_random_uuid());
DROP SEQUENCE IF EXISTS talkgroups_learned_id_seq;
ALTER TABLE alerts ALTER COLUMN id DROP IDENTITY IF EXISTS;
ALTER TABLE alerts ALTER COLUMN id SET DATA TYPE UUID USING (gen_random_uuid());
DROP SEQUENCE IF EXISTS alerts_id_seq;

21
sql/postgres/migrations/002_tglearned.up.sql Normal file
View file

@ -0,0 +1,21 @@
CREATE SEQUENCE IF NOT EXISTS alerts_id_seq START WITH 1;
ALTER TABLE alerts ALTER COLUMN id SET DATA TYPE INTEGER USING (nextval('alerts_id_seq'));
ALTER TABLE alerts ALTER COLUMN id ADD GENERATED ALWAYS AS IDENTITY;
DROP SEQUENCE IF EXISTS alerts_id_seq;
CREATE SEQUENCE IF NOT EXISTS talkgroups_learned_id_seq START WITH 1;
ALTER TABLE talkgroups_learned ALTER COLUMN id SET DATA TYPE INTEGER USING (nextval('talkgroups_learned_id_seq'));
ALTER TABLE talkgroups_learned ALTER COLUMN id ADD GENERATED ALWAYS AS IDENTITY;
DROP SEQUENCE IF EXISTS talkgroup_learned_id_seq;
DROP TRIGGER IF EXISTS learn_tg ON calls;
DROP FUNCTION IF EXISTS learn_talkgroup();
ALTER TABLE talkgroups ADD COLUMN IF NOT EXISTS learned BOOLEAN NOT NULL DEFAULT FALSE;
CREATE SEQUENCE IF NOT EXISTS talkgroups_id_seq START WITH 1;
ALTER TABLE talkgroups ALTER COLUMN id SET DATA TYPE INTEGER USING (nextval('talkgroups_id_seq'));
ALTER TABLE talkgroups ALTER COLUMN id ADD GENERATED ALWAYS AS IDENTITY;
DROP SEQUENCE IF EXISTS talkgroups_id_seq;
ALTER TABLE calls ADD CONSTRAINT calls_talkgroup_id_fkey FOREIGN KEY (system, talkgroup_id) REFERENCES talkgroups(system_id, tgid);

120
sql/postgres/migrations/initial.sql Normal file
View file

@ -0,0 +1,120 @@
CREATE TABLE IF NOT EXISTS users(
id INTEGER PRIMARY KEY GENERATED ALWAYS AS IDENTITY,
username VARCHAR (255) UNIQUE NOT NULL,
password TEXT NOT NULL,
email TEXT NOT NULL,
is_admin BOOLEAN NOT NULL,
prefs JSONB
);
CREATE INDEX IF NOT EXISTS users_username_idx ON users(username);
CREATE TABLE IF NOT EXISTS api_keys(
id INTEGER PRIMARY KEY GENERATED ALWAYS AS IDENTITY,
owner INTEGER REFERENCES users(id) NOT NULL,
created_at TIMESTAMP NOT NULL,
expires TIMESTAMP,
disabled BOOLEAN,
api_key TEXT UNIQUE NOT NULL
);
CREATE TABLE IF NOT EXISTS systems(
id INTEGER PRIMARY KEY,
name TEXT NOT NULL
);
CREATE TABLE IF NOT EXISTS talkgroups(
id INTEGER PRIMARY KEY GENERATED ALWAYS AS IDENTITY,
system_id INT4 REFERENCES systems(id) NOT NULL,
tgid INT4 NOT NULL,
name TEXT,
alpha_tag TEXT,
tg_group TEXT,
frequency INTEGER,
metadata JSONB,
tags TEXT[] NOT NULL DEFAULT '{}',
alert BOOLEAN NOT NULL DEFAULT 'true',
alert_config JSONB,
weight REAL NOT NULL DEFAULT 1.0,
learned BOOLEAN,
UNIQUE (system_id, tgid)
);
CREATE INDEX talkgroups_system_tgid_idx ON talkgroups (system_id, tgid);
CREATE INDEX IF NOT EXISTS talkgroup_id_tags ON talkgroups USING GIN (tags);
CREATE TABLE IF NOT EXISTS talkgroups_learned(
id INTEGER PRIMARY KEY GENERATED ALWAYS AS IDENTITY,
system_id INTEGER REFERENCES systems(id) NOT NULL,
tgid INTEGER NOT NULL,
name TEXT NOT NULL,
alpha_tag TEXT,
tg_group TEXT,
ignored BOOLEAN,
UNIQUE (system_id, tgid, name)
);
CREATE TABLE IF NOT EXISTS alerts(
id INTEGER PRIMARY KEY GENERATED ALWAYS AS IDENTITY,
time TIMESTAMPTZ NOT NULL,
tgid INTEGER NOT NULL,
system_id INTEGER REFERENCES systems(id) NOT NULL,
weight REAL,
score REAL,
orig_score REAL,
notified BOOLEAN NOT NULL DEFAULT 'false',
metadata JSONB
);
CREATE TABLE IF NOT EXISTS calls(
id UUID PRIMARY KEY,
submitter INTEGER REFERENCES api_keys(id) ON DELETE SET NULL,
system INTEGER NOT NULL,
talkgroup INTEGER NOT NULL,
call_date TIMESTAMPTZ NOT NULL,
audio_name TEXT,
audio_blob BYTEA,
duration INTEGER,
audio_type TEXT,
audio_url TEXT,
frequency INTEGER NOT NULL,
frequencies INTEGER[],
patches INTEGER[],
tg_label TEXT,
tg_alpha_tag TEXT,
tg_group TEXT,
source INTEGER NOT NULL,
transcript TEXT
);
CREATE INDEX IF NOT EXISTS calls_transcript_idx ON calls USING GIN (to_tsvector('english', transcript));
CREATE INDEX IF NOT EXISTS calls_call_date_tg_idx ON calls(system, talkgroup, call_date);
CREATE TABLE IF NOT EXISTS settings(
name TEXT PRIMARY KEY,
updated_by INTEGER REFERENCES users(id),
value JSONB
);
CREATE TABLE IF NOT EXISTS incidents(
id UUID PRIMARY KEY,
name TEXT NOT NULL,
description TEXT,
start_time TIMESTAMP,
end_time TIMESTAMP,
location JSONB,
metadata JSONB
);
CREATE INDEX IF NOT EXISTS incidents_name_description_idx ON incidents USING GIN (
(to_tsvector('english', name) || to_tsvector('english', coalesce(description, ''))
)
);
CREATE TABLE IF NOT EXISTS incidents_calls(
incident_id UUID REFERENCES incidents(id) ON UPDATE CASCADE ON DELETE CASCADE,
call_id UUID REFERENCES calls(id) ON UPDATE CASCADE,
notes JSONB,
PRIMARY KEY (incident_id, call_id)
);

3
sql/postgres/queries/calls.sql
View file

@ -41,10 +41,9 @@ source
UPDATE calls SET transcript = $2 WHERE id = $1; UPDATE calls SET transcript = $2 WHERE id = $1;
-- name: AddAlert :exec -- name: AddAlert :exec
INSERT INTO alerts (id, time, tgid, system_id, weight, score, orig_score, notified, metadata) INSERT INTO alerts (time, tgid, system_id, weight, score, orig_score, notified, metadata)
VALUES VALUES
( (
sqlc.arg(id),
sqlc.arg(time), sqlc.arg(time),
sqlc.arg(tgid), sqlc.arg(tgid),
sqlc.arg(system_id), sqlc.arg(system_id),

40
sql/postgres/queries/talkgroups.sql
View file

@ -26,11 +26,10 @@ WHERE (system_id, tgid) = (@system_id, @tg_id);
-- name: GetTalkgroupWithLearned :one -- name: GetTalkgroupWithLearned :one
SELECT SELECT
sqlc.embed(tg), sqlc.embed(sys), sqlc.embed(tg), sqlc.embed(sys)
FALSE learned
FROM talkgroups tg FROM talkgroups tg
JOIN systems sys ON tg.system_id = sys.id JOIN systems sys ON tg.system_id = sys.id
WHERE (tg.system_id, tg.tgid) = (@system_id, @tgid) WHERE (tg.system_id, tg.tgid) = (@system_id, @tgid) AND tg.learned IS NOT TRUE
UNION UNION
SELECT SELECT
tgl.id, tgl.system_id::INT4, tgl.tgid::INT4, tgl.name, tgl.id, tgl.system_id::INT4, tgl.tgid::INT4, tgl.name,
@ -44,11 +43,10 @@ WHERE tgl.system_id = @system_id AND tgl.tgid = @tgid AND ignored IS NOT TRUE;
-- name: GetTalkgroupsWithLearnedBySystem :many -- name: GetTalkgroupsWithLearnedBySystem :many
SELECT SELECT
sqlc.embed(tg), sqlc.embed(sys), sqlc.embed(tg), sqlc.embed(sys)
FALSE learned
FROM talkgroups tg FROM talkgroups tg
JOIN systems sys ON tg.system_id = sys.id JOIN systems sys ON tg.system_id = sys.id
WHERE tg.system_id = @system WHERE tg.system_id = @system AND tg.learned IS NOT TRUE
UNION UNION
SELECT SELECT
tgl.id, tgl.system_id::INT4, tgl.tgid::INT4, tgl.name, tgl.id, tgl.system_id::INT4, tgl.tgid::INT4, tgl.name,
@ -62,10 +60,10 @@ WHERE tgl.system_id = @system AND ignored IS NOT TRUE;
-- name: GetTalkgroupsWithLearned :many -- name: GetTalkgroupsWithLearned :many
SELECT SELECT
sqlc.embed(tg), sqlc.embed(sys), sqlc.embed(tg), sqlc.embed(sys)
FALSE learned
FROM talkgroups tg FROM talkgroups tg
JOIN systems sys ON tg.system_id = sys.id JOIN systems sys ON tg.system_id = sys.id
WHERE tg.learned IS NOT TRUE
UNION UNION
SELECT SELECT
tgl.id, tgl.system_id::INT4, tgl.tgid::INT4, tgl.name, tgl.id, tgl.system_id::INT4, tgl.tgid::INT4, tgl.name,
@ -94,3 +92,29 @@ SET
weight = COALESCE(sqlc.narg('weight'), weight) weight = COALESCE(sqlc.narg('weight'), weight)
WHERE id = sqlc.narg('id') OR (system_id = sqlc.narg('system_id') AND tgid = sqlc.narg('tgid')) WHERE id = sqlc.narg('id') OR (system_id = sqlc.narg('system_id') AND tgid = sqlc.narg('tgid'))
RETURNING *; RETURNING *;
-- name: AddTalkgroupWithLearnedFlag :exec
INSERT INTO talkgroups (
system_id,
tgid,
learned
) VALUES(
@system_id,
@tgid,
't'
);
-- name: AddLearnedTalkgroup :one
INSERT INTO talkgroups_learned(
system_id,
tgid,
name,
alpha_tag,
tg_group
) VALUES (
@system_id,
@tgid,
sqlc.narg('name'),
sqlc.narg('alpha_tag'),
sqlc.narg('tg_group')
) RETURNING id;

1
sql/sqlc.yaml
View file

@ -14,6 +14,7 @@ sql:
initialisms: initialisms:
- id - id
- tgid - tgid
- tg
emit_pointers_for_null_types: true emit_pointers_for_null_types: true
overrides: overrides:
- db_type: "uuid" - db_type: "uuid"