From 4ff22d7d6317aabb500f34b9042180398c93096c Mon Sep 17 00:00:00 2001 From: Gisle Aune Date: Sat, 31 Dec 2022 17:17:10 +0100 Subject: [PATCH] add dat base --- bus.go | 6 +- cmd/lucifer4-color/main.go | 24 ++ commands/assign.go | 8 +- commands/device.go | 8 + device/pointer.go | 33 +- effects/serializable.go | 2 +- events/alias.go | 21 ++ events/assignment.go | 34 ++ events/device.go | 21 +- events/log.go | 2 +- events/program.go | 7 + go.mod | 9 +- go.sum | 4 + services/effectenforcer.go | 47 ++- services/hue/bridge.go | 21 +- .../20221229182826_device_hwstate.sql | 16 + .../20221229213541_device_assignment.sql | 15 + .../migrations/20221230104518_device_auth.sql | 14 + .../20221231155702_device_alias.sql | 15 + services/mysqldb/mysqlgen/db.go | 238 +++++++++++++ services/mysqldb/mysqlgen/device.sql.go | 301 ++++++++++++++++ services/mysqldb/mysqlgen/models.go | 36 ++ services/mysqldb/queries/device.sql | 74 ++++ services/mysqldb/service.go | 321 ++++++++++++++++++ services/mysqldb/sqltypes/nullrawmessage.go | 36 ++ services/resolver.go | 30 +- sqlc.yaml | 28 ++ 27 files changed, 1336 insertions(+), 35 deletions(-) create mode 100644 cmd/lucifer4-color/main.go create mode 100644 events/alias.go create mode 100644 events/assignment.go create mode 100644 events/program.go create mode 100644 services/mysqldb/migrations/20221229182826_device_hwstate.sql create mode 100644 services/mysqldb/migrations/20221229213541_device_assignment.sql create mode 100644 services/mysqldb/migrations/20221230104518_device_auth.sql create mode 100644 services/mysqldb/migrations/20221231155702_device_alias.sql create mode 100644 services/mysqldb/mysqlgen/db.go create mode 100644 services/mysqldb/mysqlgen/device.sql.go create mode 100644 services/mysqldb/mysqlgen/models.go create mode 100644 services/mysqldb/queries/device.sql create mode 100644 services/mysqldb/service.go create mode 100644 services/mysqldb/sqltypes/nullrawmessage.go create mode 100644 sqlc.yaml diff --git a/bus.go b/bus.go index 0244a2d..fa78b8e 100644 --- a/bus.go +++ b/bus.go @@ -65,7 +65,7 @@ func (b *EventBus) JoinPrivileged(service ActiveService) { } func (b *EventBus) RunCommand(command Command) { - if cd := command.CommandDescription(); !strings.HasPrefix(cd, "SetStateBatch(") { + if cd := command.CommandDescription(); !strings.HasPrefix(cd, "SetState") { if setStates := atomic.LoadInt32(&b.setStates); setStates > 0 { fmt.Println("[INFO]", setStates, "SetStates commands hidden.") atomic.AddInt32(&b.setStates, -setStates) @@ -74,7 +74,7 @@ func (b *EventBus) RunCommand(command Command) { fmt.Println("[COMMAND]", cd) } else { if atomic.AddInt32(&b.setStates, 1) >= 1000 { - fmt.Println("[INFO] 100 SetStates commands hidden.") + fmt.Println("[INFO] 1000 SetStates commands hidden.") atomic.AddInt32(&b.setStates, -1000) } } @@ -99,7 +99,7 @@ func (b *EventBus) send(message serviceMessage) { for _, service := range b.privilegedList { if message.command != nil { - service.HandleCommand(nil, message.command) + service.HandleCommand(b, message.command) } if message.event != nil { service.HandleEvent(nil, message.event) diff --git a/cmd/lucifer4-color/main.go b/cmd/lucifer4-color/main.go new file mode 100644 index 0000000..e963f8e --- /dev/null +++ b/cmd/lucifer4-color/main.go @@ -0,0 +1,24 @@ +package main + +import ( + "fmt" + "git.aiterp.net/lucifer3/server/internal/color" + "os" +) + +func main() { + col := color.MustParse(os.Args[len(os.Args)-1]) + + if col, ok := col.ToXY(); ok { + fmt.Println(col.String()) + } + if col, ok := col.ToHS(); ok { + fmt.Println(col.String()) + } + if col, ok := col.ToHSK(); ok { + fmt.Println(col.String()) + } + if col, ok := col.ToRGB(); ok { + fmt.Println(col.String()) + } +} diff --git a/commands/assign.go b/commands/assign.go index 7bb3459..4c42da2 100644 --- a/commands/assign.go +++ b/commands/assign.go @@ -3,13 +3,19 @@ package commands import ( "fmt" lucifer3 "git.aiterp.net/lucifer3/server" + "github.com/google/uuid" ) type Assign struct { + ID *uuid.UUID `json:"id"` Match string `json:"match"` Effect lucifer3.Effect `json:"effect"` } func (c Assign) CommandDescription() string { - return fmt.Sprintf("Assign(%s, %s)", c.Match, c.Effect.EffectDescription()) + if c.ID != nil { + return fmt.Sprintf("Assign(%s, %s, id=%s)", c.Match, c.Effect.EffectDescription(), *c.ID) + } else { + return fmt.Sprintf("Assign(%s, %s)", c.Match, c.Effect.EffectDescription()) + } } diff --git a/commands/device.go b/commands/device.go index 718d083..8d2b3c6 100644 --- a/commands/device.go +++ b/commands/device.go @@ -58,3 +58,11 @@ func (c SearchDevices) Matches(driver string) (string, bool) { func (c SearchDevices) CommandDescription() string { return fmt.Sprintf("SearchDevices(%s, %#+v)", c.ID, c.Hint) } + +type ForgetDevice struct { + ID string `json:"id"` +} + +func (c ForgetDevice) CommandDescription() string { + return fmt.Sprintf("ForgetDevice(%s)", c.ID) +} diff --git a/device/pointer.go b/device/pointer.go index 73a5011..0c0f317 100644 --- a/device/pointer.go +++ b/device/pointer.go @@ -1,6 +1,7 @@ package device import ( + "git.aiterp.net/lucifer3/server/internal/gentools" "strings" ) @@ -13,32 +14,46 @@ type Pointer struct { Aliases []string `json:"aliases"` } -func (p *Pointer) AddAlias(alias string) { +func (p *Pointer) AddAlias(alias string) (added *string, removed *string) { + if !strings.HasPrefix(alias, "lucifer:") { + return + } + + for _, alias2 := range p.Aliases { + if alias2 == alias { + return + } + } + if strings.HasPrefix(alias, "lucifer:name:") { for i, alias2 := range p.Aliases { if strings.HasPrefix(alias2, "lucifer:name:") { p.Aliases = append(p.Aliases[:i], p.Aliases[i+1:]...) + removed = gentools.ShallowCopy(&alias2) break } } } - for _, alias2 := range p.Aliases { - if alias2 == alias { - return - } - } - + added = gentools.ShallowCopy(&alias) p.Aliases = append(p.Aliases, alias) + return } -func (p *Pointer) RemoveAlias(alias string) { +func (p *Pointer) RemoveAlias(alias string) bool { + if strings.HasPrefix(alias, "lucifer:name:") { + // Name cannot be removed like this. + return false + } + for i, alias2 := range p.Aliases { if alias2 == alias { p.Aliases = append(p.Aliases[:i], p.Aliases[i+1:]...) - break + return true } } + + return false } func (p *Pointer) Name() string { diff --git a/effects/serializable.go b/effects/serializable.go index 5a3b641..5524ad7 100644 --- a/effects/serializable.go +++ b/effects/serializable.go @@ -14,7 +14,7 @@ type serializedEffect struct { } type Serializable struct { - lucifer3.Effect + Effect lucifer3.Effect } func (s *Serializable) UnmarshalJSON(raw []byte) error { diff --git a/events/alias.go b/events/alias.go new file mode 100644 index 0000000..75ce92c --- /dev/null +++ b/events/alias.go @@ -0,0 +1,21 @@ +package events + +import "fmt" + +type AliasAdded struct { + ID string + Alias string +} + +func (e AliasAdded) EventDescription() string { + return fmt.Sprintf("AliasAdded(id=%s, alias=%s)", e.ID, e.Alias) +} + +type AliasRemoved struct { + ID string + Alias string +} + +func (e AliasRemoved) EventDescription() string { + return fmt.Sprintf("AliasRemoved(id=%s, alias=%s)", e.ID, e.Alias) +} diff --git a/events/assignment.go b/events/assignment.go new file mode 100644 index 0000000..37906d4 --- /dev/null +++ b/events/assignment.go @@ -0,0 +1,34 @@ +package events + +import ( + "fmt" + lucifer3 "git.aiterp.net/lucifer3/server" + "github.com/google/uuid" +) + +type AssignmentCreated struct { + ID uuid.UUID + Match string + Effect lucifer3.Effect +} + +func (e AssignmentCreated) EventDescription() string { + return fmt.Sprintf("AssignmentCreated(id=%s, match=%s, effect=%s)", e.ID, e.Match, e.Effect.EffectDescription()) +} + +type AssignmentRemoved struct { + ID uuid.UUID +} + +func (e AssignmentRemoved) EventDescription() string { + return fmt.Sprintf("AssignmentRemoved(id=%s)", e.ID) +} + +type DeviceAssigned struct { + DeviceID string `json:"deviceId"` + AssignmentID *uuid.UUID `json:"assignmentId"` +} + +func (e DeviceAssigned) EventDescription() string { + return fmt.Sprintf("DeviceAssigned(device=%s, %s)", e.DeviceID, e.AssignmentID) +} diff --git a/events/device.go b/events/device.go index d96bd05..620caf4 100644 --- a/events/device.go +++ b/events/device.go @@ -4,6 +4,8 @@ import ( "fmt" "git.aiterp.net/lucifer3/server/device" "git.aiterp.net/lucifer3/server/internal/color" + "git.aiterp.net/lucifer3/server/internal/formattools" + "os" ) type DeviceConnected struct { @@ -24,7 +26,7 @@ func (e DeviceDisconnected) EventDescription() string { } type HardwareState struct { - ID string `json:"internalId"` + ID string `json:"id"` InternalName string `json:"internalName"` SupportFlags device.SupportFlags `json:"supportFlags"` ColorFlags device.ColorFlags `json:"colorFlags"` @@ -34,6 +36,7 @@ type HardwareState struct { State device.State `json:"state"` BatteryPercentage *int `json:"batteryPercentage"` Unreachable bool `json:"unreachable"` + Stale bool `json:"stale,omitempty"` } func (e HardwareState) EventDescription() string { @@ -53,6 +56,7 @@ type HardwareMetadata struct { Icon string `json:"icon,omitempty"` SerialNumber string `json:"serialNumber,omitempty"` FirmwareVersion string `json:"firmwareVersion,omitempty"` + Stale bool `json:"stale,omitempty"` } func (e HardwareMetadata) EventDescription() string { @@ -85,6 +89,17 @@ type DeviceAccepted struct { } func (e DeviceAccepted) EventDescription() string { - // TODO: Use formattools.Asterisks - return fmt.Sprintf("DeviceAccepted(id:%s, apiKey:%s)", e.ID, e.APIKey) + if os.Getenv("DEV_SHOW_API_KEYS") == "true" { + return fmt.Sprintf("DeviceAccepted(id:%s, apiKey:%s)", e.ID, e.APIKey) + } else { + return fmt.Sprintf("DeviceAccepted(id:%s, apiKey:%s)", e.ID, formattools.Asterisks(e.APIKey)) + } +} + +type DeviceForgotten struct { + ID string `json:"id"` +} + +func (e DeviceForgotten) EventDescription() string { + return fmt.Sprintf("DeviceForgotten(id:%s)", e.ID) } diff --git a/events/log.go b/events/log.go index 081f34a..0c247cf 100644 --- a/events/log.go +++ b/events/log.go @@ -3,7 +3,7 @@ package events import "fmt" type Log struct { - ID string `json:"id"` + ID string `json:"id,omitempty"` Level string `json:"level"` Code string `json:"code"` Message string `json:"message"` diff --git a/events/program.go b/events/program.go new file mode 100644 index 0000000..4acede4 --- /dev/null +++ b/events/program.go @@ -0,0 +1,7 @@ +package events + +type Started struct{} + +func (Started) EventDescription() string { + return "Started()" +} diff --git a/go.mod b/go.mod index ca45253..d08a76e 100644 --- a/go.mod +++ b/go.mod @@ -4,6 +4,9 @@ go 1.19 require github.com/lucasb-eyer/go-colorful v1.2.0 -require github.com/gobwas/glob v0.2.3 - -require golang.org/x/sync v0.1.0 // indirect +require ( + github.com/go-sql-driver/mysql v1.7.0 + github.com/gobwas/glob v0.2.3 + github.com/google/uuid v1.3.0 + golang.org/x/sync v0.1.0 +) diff --git a/go.sum b/go.sum index fbf32fd..10e2d3f 100644 --- a/go.sum +++ b/go.sum @@ -1,5 +1,9 @@ +github.com/go-sql-driver/mysql v1.7.0 h1:ueSltNNllEqE3qcWBTD0iQd3IpL/6U+mJxLkazJ7YPc= +github.com/go-sql-driver/mysql v1.7.0/go.mod h1:OXbVy3sEdcQ2Doequ6Z5BW6fXNQTmx+9S1MCJN5yJMI= github.com/gobwas/glob v0.2.3 h1:A4xDbljILXROh+kObIiy5kIaPYD8e96x1tgBhUI5J+Y= github.com/gobwas/glob v0.2.3/go.mod h1:d3Ez4x06l9bZtSvzIay5+Yzi0fmZzPgnTbPcKjJAkT8= +github.com/google/uuid v1.3.0 h1:t6JiXgmwXMjEs8VusXIJk2BXHsn+wx8BZdTaoZ5fu7I= +github.com/google/uuid v1.3.0/go.mod h1:TIyPZe4MgqvfeYDBFedMoGGpEw/LqOeaOT+nhxU+yHo= github.com/lucasb-eyer/go-colorful v1.2.0 h1:1nnpGOrhyZZuNyfu1QjKiUICQ74+3FNCN69Aj6K7nkY= github.com/lucasb-eyer/go-colorful v1.2.0/go.mod h1:R4dSotOR9KMtayYi1e77YzuveK+i7ruzyGqttikkLy0= golang.org/x/sync v0.1.0 h1:wsuoTGHzEhffawBOhz5CYhcrV4IdKZbEyZjBMuTp12o= diff --git a/services/effectenforcer.go b/services/effectenforcer.go index 22f2bde..e505e20 100644 --- a/services/effectenforcer.go +++ b/services/effectenforcer.go @@ -5,6 +5,8 @@ import ( "git.aiterp.net/lucifer3/server/commands" "git.aiterp.net/lucifer3/server/device" "git.aiterp.net/lucifer3/server/events" + "git.aiterp.net/lucifer3/server/internal/gentools" + "github.com/google/uuid" "sync" "sync/atomic" "time" @@ -42,7 +44,7 @@ func (s *effectEnforcer) HandleEvent(_ *lucifer3.EventBus, event lucifer3.Event) // If the device is managed by the effect enforcer, cause the effect to be // re-ran. s.mu.Lock() - if run, ok := s.index[event.ID]; ok && run.due.IsZero() { + if run, ok := s.index[event.ID]; ok { run.due = time.Now() } s.mu.Unlock() @@ -59,10 +61,35 @@ func (s *effectEnforcer) HandleCommand(bus *lucifer3.EventBus, command lucifer3. allowedIDs = append(allowedIDs, ptr.ID) } } + if len(pointers) == 0 { + if command.ID != nil { + bus.RunEvent(events.AssignmentRemoved{ID: *command.ID}) + } else { + bus.RunEvent(events.Log{ + Level: "info", + Code: "assignment_matched_none", + Message: "Assignment to \"" + command.Match + "\" matched no devices.", + }) + } + + return + } + + id := uuid.New() + if command.ID != nil { + id = *command.ID + } + + bus.RunEvent(events.AssignmentCreated{ + ID: id, + Match: command.Match, + Effect: command.Effect, + }) s.mu.Lock() // Create a new run newRun := &effectEnforcerRun{ + id: id, due: time.Now(), ids: allowedIDs, effect: command.Effect, @@ -81,17 +108,28 @@ func (s *effectEnforcer) HandleCommand(bus *lucifer3.EventBus, command lucifer3. if atomic.CompareAndSwapUint32(&s.started, 0, 1) { go s.runLoop(bus) } + + for _, deviceID := range allowedIDs { + bus.RunEvent(events.DeviceAssigned{ + DeviceID: deviceID, + AssignmentID: gentools.Ptr(id), + }) + } } } func (s *effectEnforcer) runLoop(bus *lucifer3.EventBus) { deleteList := make([]int, 0, 8) batch := make(commands.SetStateBatch, 64) + deleteIDs := make([]uuid.UUID, 0) for now := range time.NewTicker(time.Millisecond * 50).C { + deleteIDs = deleteIDs[:0] + s.mu.Lock() for i, run := range s.list { if run.dead { + deleteIDs = append(deleteIDs, run.id) deleteList = append(deleteList, i-len(deleteList)) continue } @@ -123,13 +161,17 @@ func (s *effectEnforcer) runLoop(bus *lucifer3.EventBus) { if len(deleteList) > 0 { for _, i := range deleteList { - deleteList = append(deleteList[:i], deleteList[i+1:]...) + s.list = append(s.list[:i], s.list[i+1:]...) } deleteList = deleteList[:0] } s.mu.Unlock() + for _, id := range deleteIDs { + bus.RunEvent(events.AssignmentRemoved{ID: id}) + } + if len(batch) > 0 { bus.RunCommand(batch) batch = make(commands.SetStateBatch, 64) @@ -138,6 +180,7 @@ func (s *effectEnforcer) runLoop(bus *lucifer3.EventBus) { } type effectEnforcerRun struct { + id uuid.UUID due time.Time ids []string effect lucifer3.Effect diff --git a/services/hue/bridge.go b/services/hue/bridge.go index bfe0ca8..3006d36 100644 --- a/services/hue/bridge.go +++ b/services/hue/bridge.go @@ -196,10 +196,12 @@ func (b *Bridge) ApplyPatches(resources []ResourceData) (events []lucifer3.Event func (b *Bridge) SetStates(patch map[string]device.State) { b.mu.Lock() - newStates := gentools.CopyMap(b.desiredStates) + desiredStates := b.desiredStates resources := b.resources b.mu.Unlock() + desiredStates = gentools.CopyMap(desiredStates) + prefix := "hue:" + b.host + ":" for id, state := range patch { if !strings.HasPrefix(id, prefix) { @@ -213,14 +215,14 @@ func (b *Bridge) SetStates(patch map[string]device.State) { } if !state.Empty() { - newStates[id] = resource.FixState(state, resources) + desiredStates[id] = resource.FixState(state, resources) } else { - delete(newStates, id) + delete(desiredStates, id) } } b.mu.Lock() - b.desiredStates = newStates + b.desiredStates = desiredStates b.mu.Unlock() b.triggerCongruenceCheck() @@ -289,8 +291,8 @@ func (b *Bridge) makeCongruentLoop(ctx context.Context) { break } - // Make sure this loop takes half a second - rateLimit := time.After(time.Second / 2) + // Make sure this loop doesn't spam too hard + rateLimit := time.After(time.Second / 10) // Take states b.mu.Lock() @@ -305,6 +307,7 @@ func (b *Bridge) makeCongruentLoop(ctx context.Context) { updates := make(map[string]ResourceUpdate) for id, desired := range desiredStates { + active, activeOK := activeStates[id] lightID := resources[id].ServiceID("light") if !reachable[id] || !activeOK || lightID == nil { @@ -423,10 +426,10 @@ func (b *Bridge) makeCongruentLoop(ctx context.Context) { b.mu.Unlock() cancel() - } - // Wait the remaining time for the rate limit - <-rateLimit + // Wait the remaining time for the rate limit + <-rateLimit + } } } diff --git a/services/mysqldb/migrations/20221229182826_device_hwstate.sql b/services/mysqldb/migrations/20221229182826_device_hwstate.sql new file mode 100644 index 0000000..61c1a63 --- /dev/null +++ b/services/mysqldb/migrations/20221229182826_device_hwstate.sql @@ -0,0 +1,16 @@ +-- +goose Up +-- +goose StatementBegin +CREATE TABLE device_info +( + id VARCHAR(255) NOT NULL, + kind VARCHAR(255) NOT NULL, + data JSON NOT NULL, + + PRIMARY KEY (id, kind) +); +-- +goose StatementEnd + +-- +goose Down +-- +goose StatementBegin +DROP TABLE device_info; +-- +goose StatementEnd diff --git a/services/mysqldb/migrations/20221229213541_device_assignment.sql b/services/mysqldb/migrations/20221229213541_device_assignment.sql new file mode 100644 index 0000000..ecc9148 --- /dev/null +++ b/services/mysqldb/migrations/20221229213541_device_assignment.sql @@ -0,0 +1,15 @@ +-- +goose Up +-- +goose StatementBegin +CREATE TABLE device_assignment +( + id CHAR(36) NOT NULL PRIMARY KEY, + created_date TIMESTAMP NOT NULL, + `match` TEXT NOT NULL, + effect JSON NOT NULL +); +-- +goose StatementEnd + +-- +goose Down +-- +goose StatementBegin +DROP TABLE device_assignment; +-- +goose StatementEnd diff --git a/services/mysqldb/migrations/20221230104518_device_auth.sql b/services/mysqldb/migrations/20221230104518_device_auth.sql new file mode 100644 index 0000000..b3587f1 --- /dev/null +++ b/services/mysqldb/migrations/20221230104518_device_auth.sql @@ -0,0 +1,14 @@ +-- +goose Up +-- +goose StatementBegin +CREATE TABLE device_auth +( + id VARCHAR(255) NOT NULL PRIMARY KEY, + api_key TEXT NOT NULL, + extras JSON NOT NULL +); +-- +goose StatementEnd + +-- +goose Down +-- +goose StatementBegin +DROP TABLE device_auth; +-- +goose StatementEnd diff --git a/services/mysqldb/migrations/20221231155702_device_alias.sql b/services/mysqldb/migrations/20221231155702_device_alias.sql new file mode 100644 index 0000000..fb588c4 --- /dev/null +++ b/services/mysqldb/migrations/20221231155702_device_alias.sql @@ -0,0 +1,15 @@ +-- +goose Up +-- +goose StatementBegin +CREATE TABLE device_alias +( + id VARCHAR(255) NOT NULL, + alias VARCHAR(1023) NOT NULL, + + PRIMARY KEY (id, alias) +); +-- +goose StatementEnd + +-- +goose Down +-- +goose StatementBegin +DROP TABLE device_alias; +-- +goose StatementEnd diff --git a/services/mysqldb/mysqlgen/db.go b/services/mysqldb/mysqlgen/db.go new file mode 100644 index 0000000..284c7db --- /dev/null +++ b/services/mysqldb/mysqlgen/db.go @@ -0,0 +1,238 @@ +// Code generated by sqlc. DO NOT EDIT. +// versions: +// sqlc v1.13.0 + +package mysqlgen + +import ( + "context" + "database/sql" + "fmt" +) + +type DBTX interface { + ExecContext(context.Context, string, ...interface{}) (sql.Result, error) + PrepareContext(context.Context, string) (*sql.Stmt, error) + QueryContext(context.Context, string, ...interface{}) (*sql.Rows, error) + QueryRowContext(context.Context, string, ...interface{}) *sql.Row +} + +func New(db DBTX) *Queries { + return &Queries{db: db} +} + +func Prepare(ctx context.Context, db DBTX) (*Queries, error) { + q := Queries{db: db} + var err error + if q.deleteDeviceAliasStmt, err = db.PrepareContext(ctx, deleteDeviceAlias); err != nil { + return nil, fmt.Errorf("error preparing query DeleteDeviceAlias: %w", err) + } + if q.deleteDeviceAliasByIDStmt, err = db.PrepareContext(ctx, deleteDeviceAliasByID); err != nil { + return nil, fmt.Errorf("error preparing query DeleteDeviceAliasByID: %w", err) + } + if q.deleteDeviceAliasByIDLikeStmt, err = db.PrepareContext(ctx, deleteDeviceAliasByIDLike); err != nil { + return nil, fmt.Errorf("error preparing query DeleteDeviceAliasByIDLike: %w", err) + } + if q.deleteDeviceAssignmentStmt, err = db.PrepareContext(ctx, deleteDeviceAssignment); err != nil { + return nil, fmt.Errorf("error preparing query DeleteDeviceAssignment: %w", err) + } + if q.deleteDeviceAuthStmt, err = db.PrepareContext(ctx, deleteDeviceAuth); err != nil { + return nil, fmt.Errorf("error preparing query DeleteDeviceAuth: %w", err) + } + if q.deleteDeviceInfoStmt, err = db.PrepareContext(ctx, deleteDeviceInfo); err != nil { + return nil, fmt.Errorf("error preparing query DeleteDeviceInfo: %w", err) + } + if q.deleteDeviceInfoByIDStmt, err = db.PrepareContext(ctx, deleteDeviceInfoByID); err != nil { + return nil, fmt.Errorf("error preparing query DeleteDeviceInfoByID: %w", err) + } + if q.deleteDeviceInfoLikeStmt, err = db.PrepareContext(ctx, deleteDeviceInfoLike); err != nil { + return nil, fmt.Errorf("error preparing query DeleteDeviceInfoLike: %w", err) + } + if q.insertDeviceAliasStmt, err = db.PrepareContext(ctx, insertDeviceAlias); err != nil { + return nil, fmt.Errorf("error preparing query InsertDeviceAlias: %w", err) + } + if q.listDeviceAliasesStmt, err = db.PrepareContext(ctx, listDeviceAliases); err != nil { + return nil, fmt.Errorf("error preparing query ListDeviceAliases: %w", err) + } + if q.listDeviceAssignmentsStmt, err = db.PrepareContext(ctx, listDeviceAssignments); err != nil { + return nil, fmt.Errorf("error preparing query ListDeviceAssignments: %w", err) + } + if q.listDeviceAuthStmt, err = db.PrepareContext(ctx, listDeviceAuth); err != nil { + return nil, fmt.Errorf("error preparing query ListDeviceAuth: %w", err) + } + if q.listDeviceInfosStmt, err = db.PrepareContext(ctx, listDeviceInfos); err != nil { + return nil, fmt.Errorf("error preparing query ListDeviceInfos: %w", err) + } + if q.replaceDeviceAssignmentStmt, err = db.PrepareContext(ctx, replaceDeviceAssignment); err != nil { + return nil, fmt.Errorf("error preparing query ReplaceDeviceAssignment: %w", err) + } + if q.replaceDeviceAuthStmt, err = db.PrepareContext(ctx, replaceDeviceAuth); err != nil { + return nil, fmt.Errorf("error preparing query ReplaceDeviceAuth: %w", err) + } + if q.replaceDeviceInfoStmt, err = db.PrepareContext(ctx, replaceDeviceInfo); err != nil { + return nil, fmt.Errorf("error preparing query ReplaceDeviceInfo: %w", err) + } + return &q, nil +} + +func (q *Queries) Close() error { + var err error + if q.deleteDeviceAliasStmt != nil { + if cerr := q.deleteDeviceAliasStmt.Close(); cerr != nil { + err = fmt.Errorf("error closing deleteDeviceAliasStmt: %w", cerr) + } + } + if q.deleteDeviceAliasByIDStmt != nil { + if cerr := q.deleteDeviceAliasByIDStmt.Close(); cerr != nil { + err = fmt.Errorf("error closing deleteDeviceAliasByIDStmt: %w", cerr) + } + } + if q.deleteDeviceAliasByIDLikeStmt != nil { + if cerr := q.deleteDeviceAliasByIDLikeStmt.Close(); cerr != nil { + err = fmt.Errorf("error closing deleteDeviceAliasByIDLikeStmt: %w", cerr) + } + } + if q.deleteDeviceAssignmentStmt != nil { + if cerr := q.deleteDeviceAssignmentStmt.Close(); cerr != nil { + err = fmt.Errorf("error closing deleteDeviceAssignmentStmt: %w", cerr) + } + } + if q.deleteDeviceAuthStmt != nil { + if cerr := q.deleteDeviceAuthStmt.Close(); cerr != nil { + err = fmt.Errorf("error closing deleteDeviceAuthStmt: %w", cerr) + } + } + if q.deleteDeviceInfoStmt != nil { + if cerr := q.deleteDeviceInfoStmt.Close(); cerr != nil { + err = fmt.Errorf("error closing deleteDeviceInfoStmt: %w", cerr) + } + } + if q.deleteDeviceInfoByIDStmt != nil { + if cerr := q.deleteDeviceInfoByIDStmt.Close(); cerr != nil { + err = fmt.Errorf("error closing deleteDeviceInfoByIDStmt: %w", cerr) + } + } + if q.deleteDeviceInfoLikeStmt != nil { + if cerr := q.deleteDeviceInfoLikeStmt.Close(); cerr != nil { + err = fmt.Errorf("error closing deleteDeviceInfoLikeStmt: %w", cerr) + } + } + if q.insertDeviceAliasStmt != nil { + if cerr := q.insertDeviceAliasStmt.Close(); cerr != nil { + err = fmt.Errorf("error closing insertDeviceAliasStmt: %w", cerr) + } + } + if q.listDeviceAliasesStmt != nil { + if cerr := q.listDeviceAliasesStmt.Close(); cerr != nil { + err = fmt.Errorf("error closing listDeviceAliasesStmt: %w", cerr) + } + } + if q.listDeviceAssignmentsStmt != nil { + if cerr := q.listDeviceAssignmentsStmt.Close(); cerr != nil { + err = fmt.Errorf("error closing listDeviceAssignmentsStmt: %w", cerr) + } + } + if q.listDeviceAuthStmt != nil { + if cerr := q.listDeviceAuthStmt.Close(); cerr != nil { + err = fmt.Errorf("error closing listDeviceAuthStmt: %w", cerr) + } + } + if q.listDeviceInfosStmt != nil { + if cerr := q.listDeviceInfosStmt.Close(); cerr != nil { + err = fmt.Errorf("error closing listDeviceInfosStmt: %w", cerr) + } + } + if q.replaceDeviceAssignmentStmt != nil { + if cerr := q.replaceDeviceAssignmentStmt.Close(); cerr != nil { + err = fmt.Errorf("error closing replaceDeviceAssignmentStmt: %w", cerr) + } + } + if q.replaceDeviceAuthStmt != nil { + if cerr := q.replaceDeviceAuthStmt.Close(); cerr != nil { + err = fmt.Errorf("error closing replaceDeviceAuthStmt: %w", cerr) + } + } + if q.replaceDeviceInfoStmt != nil { + if cerr := q.replaceDeviceInfoStmt.Close(); cerr != nil { + err = fmt.Errorf("error closing replaceDeviceInfoStmt: %w", cerr) + } + } + return err +} + +func (q *Queries) exec(ctx context.Context, stmt *sql.Stmt, query string, args ...interface{}) (sql.Result, error) { + switch { + case stmt != nil && q.tx != nil: + return q.tx.StmtContext(ctx, stmt).ExecContext(ctx, args...) + case stmt != nil: + return stmt.ExecContext(ctx, args...) + default: + return q.db.ExecContext(ctx, query, args...) + } +} + +func (q *Queries) query(ctx context.Context, stmt *sql.Stmt, query string, args ...interface{}) (*sql.Rows, error) { + switch { + case stmt != nil && q.tx != nil: + return q.tx.StmtContext(ctx, stmt).QueryContext(ctx, args...) + case stmt != nil: + return stmt.QueryContext(ctx, args...) + default: + return q.db.QueryContext(ctx, query, args...) + } +} + +func (q *Queries) queryRow(ctx context.Context, stmt *sql.Stmt, query string, args ...interface{}) *sql.Row { + switch { + case stmt != nil && q.tx != nil: + return q.tx.StmtContext(ctx, stmt).QueryRowContext(ctx, args...) + case stmt != nil: + return stmt.QueryRowContext(ctx, args...) + default: + return q.db.QueryRowContext(ctx, query, args...) + } +} + +type Queries struct { + db DBTX + tx *sql.Tx + deleteDeviceAliasStmt *sql.Stmt + deleteDeviceAliasByIDStmt *sql.Stmt + deleteDeviceAliasByIDLikeStmt *sql.Stmt + deleteDeviceAssignmentStmt *sql.Stmt + deleteDeviceAuthStmt *sql.Stmt + deleteDeviceInfoStmt *sql.Stmt + deleteDeviceInfoByIDStmt *sql.Stmt + deleteDeviceInfoLikeStmt *sql.Stmt + insertDeviceAliasStmt *sql.Stmt + listDeviceAliasesStmt *sql.Stmt + listDeviceAssignmentsStmt *sql.Stmt + listDeviceAuthStmt *sql.Stmt + listDeviceInfosStmt *sql.Stmt + replaceDeviceAssignmentStmt *sql.Stmt + replaceDeviceAuthStmt *sql.Stmt + replaceDeviceInfoStmt *sql.Stmt +} + +func (q *Queries) WithTx(tx *sql.Tx) *Queries { + return &Queries{ + db: tx, + tx: tx, + deleteDeviceAliasStmt: q.deleteDeviceAliasStmt, + deleteDeviceAliasByIDStmt: q.deleteDeviceAliasByIDStmt, + deleteDeviceAliasByIDLikeStmt: q.deleteDeviceAliasByIDLikeStmt, + deleteDeviceAssignmentStmt: q.deleteDeviceAssignmentStmt, + deleteDeviceAuthStmt: q.deleteDeviceAuthStmt, + deleteDeviceInfoStmt: q.deleteDeviceInfoStmt, + deleteDeviceInfoByIDStmt: q.deleteDeviceInfoByIDStmt, + deleteDeviceInfoLikeStmt: q.deleteDeviceInfoLikeStmt, + insertDeviceAliasStmt: q.insertDeviceAliasStmt, + listDeviceAliasesStmt: q.listDeviceAliasesStmt, + listDeviceAssignmentsStmt: q.listDeviceAssignmentsStmt, + listDeviceAuthStmt: q.listDeviceAuthStmt, + listDeviceInfosStmt: q.listDeviceInfosStmt, + replaceDeviceAssignmentStmt: q.replaceDeviceAssignmentStmt, + replaceDeviceAuthStmt: q.replaceDeviceAuthStmt, + replaceDeviceInfoStmt: q.replaceDeviceInfoStmt, + } +} diff --git a/services/mysqldb/mysqlgen/device.sql.go b/services/mysqldb/mysqlgen/device.sql.go new file mode 100644 index 0000000..8e0fb6b --- /dev/null +++ b/services/mysqldb/mysqlgen/device.sql.go @@ -0,0 +1,301 @@ +// Code generated by sqlc. DO NOT EDIT. +// versions: +// sqlc v1.13.0 +// source: device.sql + +package mysqlgen + +import ( + "context" + "encoding/json" + "time" + + "github.com/google/uuid" +) + +const deleteDeviceAlias = `-- name: DeleteDeviceAlias :exec +DELETE +FROM device_alias +WHERE id = ? + AND alias = ? +` + +type DeleteDeviceAliasParams struct { + ID string + Alias string +} + +func (q *Queries) DeleteDeviceAlias(ctx context.Context, arg DeleteDeviceAliasParams) error { + _, err := q.exec(ctx, q.deleteDeviceAliasStmt, deleteDeviceAlias, arg.ID, arg.Alias) + return err +} + +const deleteDeviceAliasByID = `-- name: DeleteDeviceAliasByID :exec +DELETE +FROM device_alias +WHERE id = ? +` + +func (q *Queries) DeleteDeviceAliasByID(ctx context.Context, id string) error { + _, err := q.exec(ctx, q.deleteDeviceAliasByIDStmt, deleteDeviceAliasByID, id) + return err +} + +const deleteDeviceAliasByIDLike = `-- name: DeleteDeviceAliasByIDLike :exec +DELETE +FROM device_alias +WHERE id LIKE ? +` + +func (q *Queries) DeleteDeviceAliasByIDLike(ctx context.Context, id string) error { + _, err := q.exec(ctx, q.deleteDeviceAliasByIDLikeStmt, deleteDeviceAliasByIDLike, id) + return err +} + +const deleteDeviceAssignment = `-- name: DeleteDeviceAssignment :exec +DELETE +FROM device_assignment +WHERE id = ? +` + +func (q *Queries) DeleteDeviceAssignment(ctx context.Context, id uuid.UUID) error { + _, err := q.exec(ctx, q.deleteDeviceAssignmentStmt, deleteDeviceAssignment, id) + return err +} + +const deleteDeviceAuth = `-- name: DeleteDeviceAuth :exec +DELETE +FROM device_auth +WHERE id = ? +` + +func (q *Queries) DeleteDeviceAuth(ctx context.Context, id string) error { + _, err := q.exec(ctx, q.deleteDeviceAuthStmt, deleteDeviceAuth, id) + return err +} + +const deleteDeviceInfo = `-- name: DeleteDeviceInfo :exec +DELETE +FROM device_info +WHERE id = ? + AND kind = ? +` + +type DeleteDeviceInfoParams struct { + ID string + Kind string +} + +func (q *Queries) DeleteDeviceInfo(ctx context.Context, arg DeleteDeviceInfoParams) error { + _, err := q.exec(ctx, q.deleteDeviceInfoStmt, deleteDeviceInfo, arg.ID, arg.Kind) + return err +} + +const deleteDeviceInfoByID = `-- name: DeleteDeviceInfoByID :exec +DELETE +FROM device_info +WHERE id = ? +` + +func (q *Queries) DeleteDeviceInfoByID(ctx context.Context, id string) error { + _, err := q.exec(ctx, q.deleteDeviceInfoByIDStmt, deleteDeviceInfoByID, id) + return err +} + +const deleteDeviceInfoLike = `-- name: DeleteDeviceInfoLike :exec +DELETE +FROM device_info +WHERE id LIKE ? +` + +func (q *Queries) DeleteDeviceInfoLike(ctx context.Context, id string) error { + _, err := q.exec(ctx, q.deleteDeviceInfoLikeStmt, deleteDeviceInfoLike, id) + return err +} + +const insertDeviceAlias = `-- name: InsertDeviceAlias :exec +INSERT INTO device_alias (id, alias) +VALUES (?, ?) +` + +type InsertDeviceAliasParams struct { + ID string + Alias string +} + +func (q *Queries) InsertDeviceAlias(ctx context.Context, arg InsertDeviceAliasParams) error { + _, err := q.exec(ctx, q.insertDeviceAliasStmt, insertDeviceAlias, arg.ID, arg.Alias) + return err +} + +const listDeviceAliases = `-- name: ListDeviceAliases :many +SELECT id, alias +FROM device_alias +` + +func (q *Queries) ListDeviceAliases(ctx context.Context) ([]DeviceAlias, error) { + rows, err := q.query(ctx, q.listDeviceAliasesStmt, listDeviceAliases) + if err != nil { + return nil, err + } + defer rows.Close() + items := []DeviceAlias{} + for rows.Next() { + var i DeviceAlias + if err := rows.Scan(&i.ID, &i.Alias); err != nil { + return nil, err + } + items = append(items, i) + } + if err := rows.Close(); err != nil { + return nil, err + } + if err := rows.Err(); err != nil { + return nil, err + } + return items, nil +} + +const listDeviceAssignments = `-- name: ListDeviceAssignments :many +SELECT id, created_date, ` + "`" + `match` + "`" + `, effect +FROM device_assignment +ORDER BY created_date +` + +func (q *Queries) ListDeviceAssignments(ctx context.Context) ([]DeviceAssignment, error) { + rows, err := q.query(ctx, q.listDeviceAssignmentsStmt, listDeviceAssignments) + if err != nil { + return nil, err + } + defer rows.Close() + items := []DeviceAssignment{} + for rows.Next() { + var i DeviceAssignment + if err := rows.Scan( + &i.ID, + &i.CreatedDate, + &i.Match, + &i.Effect, + ); err != nil { + return nil, err + } + items = append(items, i) + } + if err := rows.Close(); err != nil { + return nil, err + } + if err := rows.Err(); err != nil { + return nil, err + } + return items, nil +} + +const listDeviceAuth = `-- name: ListDeviceAuth :many +SELECT id, api_key, extras +FROM device_auth +` + +func (q *Queries) ListDeviceAuth(ctx context.Context) ([]DeviceAuth, error) { + rows, err := q.query(ctx, q.listDeviceAuthStmt, listDeviceAuth) + if err != nil { + return nil, err + } + defer rows.Close() + items := []DeviceAuth{} + for rows.Next() { + var i DeviceAuth + if err := rows.Scan(&i.ID, &i.ApiKey, &i.Extras); err != nil { + return nil, err + } + items = append(items, i) + } + if err := rows.Close(); err != nil { + return nil, err + } + if err := rows.Err(); err != nil { + return nil, err + } + return items, nil +} + +const listDeviceInfos = `-- name: ListDeviceInfos :many +SELECT id, kind, data +FROM device_info +` + +func (q *Queries) ListDeviceInfos(ctx context.Context) ([]DeviceInfo, error) { + rows, err := q.query(ctx, q.listDeviceInfosStmt, listDeviceInfos) + if err != nil { + return nil, err + } + defer rows.Close() + items := []DeviceInfo{} + for rows.Next() { + var i DeviceInfo + if err := rows.Scan(&i.ID, &i.Kind, &i.Data); err != nil { + return nil, err + } + items = append(items, i) + } + if err := rows.Close(); err != nil { + return nil, err + } + if err := rows.Err(); err != nil { + return nil, err + } + return items, nil +} + +const replaceDeviceAssignment = `-- name: ReplaceDeviceAssignment :exec +REPLACE INTO device_assignment (id, created_date, ` + "`" + `match` + "`" + `, effect) +VALUES (?, ?, ?, ?) +` + +type ReplaceDeviceAssignmentParams struct { + ID uuid.UUID + CreatedDate time.Time + Match string + Effect json.RawMessage +} + +func (q *Queries) ReplaceDeviceAssignment(ctx context.Context, arg ReplaceDeviceAssignmentParams) error { + _, err := q.exec(ctx, q.replaceDeviceAssignmentStmt, replaceDeviceAssignment, + arg.ID, + arg.CreatedDate, + arg.Match, + arg.Effect, + ) + return err +} + +const replaceDeviceAuth = `-- name: ReplaceDeviceAuth :exec +REPLACE INTO device_auth (id, api_key, extras) +VALUES (?, ?, ?) +` + +type ReplaceDeviceAuthParams struct { + ID string + ApiKey string + Extras json.RawMessage +} + +func (q *Queries) ReplaceDeviceAuth(ctx context.Context, arg ReplaceDeviceAuthParams) error { + _, err := q.exec(ctx, q.replaceDeviceAuthStmt, replaceDeviceAuth, arg.ID, arg.ApiKey, arg.Extras) + return err +} + +const replaceDeviceInfo = `-- name: ReplaceDeviceInfo :exec +REPLACE INTO device_info (id, kind, data) +VALUES (?, ?, ?) +` + +type ReplaceDeviceInfoParams struct { + ID string + Kind string + Data json.RawMessage +} + +func (q *Queries) ReplaceDeviceInfo(ctx context.Context, arg ReplaceDeviceInfoParams) error { + _, err := q.exec(ctx, q.replaceDeviceInfoStmt, replaceDeviceInfo, arg.ID, arg.Kind, arg.Data) + return err +} diff --git a/services/mysqldb/mysqlgen/models.go b/services/mysqldb/mysqlgen/models.go new file mode 100644 index 0000000..2f19750 --- /dev/null +++ b/services/mysqldb/mysqlgen/models.go @@ -0,0 +1,36 @@ +// Code generated by sqlc. DO NOT EDIT. +// versions: +// sqlc v1.13.0 + +package mysqlgen + +import ( + "encoding/json" + "time" + + "github.com/google/uuid" +) + +type DeviceAlias struct { + ID string + Alias string +} + +type DeviceAssignment struct { + ID uuid.UUID + CreatedDate time.Time + Match string + Effect json.RawMessage +} + +type DeviceAuth struct { + ID string + ApiKey string + Extras json.RawMessage +} + +type DeviceInfo struct { + ID string + Kind string + Data json.RawMessage +} diff --git a/services/mysqldb/queries/device.sql b/services/mysqldb/queries/device.sql new file mode 100644 index 0000000..184db05 --- /dev/null +++ b/services/mysqldb/queries/device.sql @@ -0,0 +1,74 @@ +-- name: ListDeviceInfos :many +SELECT * +FROM device_info; + +-- name: ListDeviceAssignments :many +SELECT * +FROM device_assignment +ORDER BY created_date; + +-- name: ListDeviceAuth :many +SELECT * +FROM device_auth; + +-- name: ListDeviceAliases :many +SELECT * +FROM device_alias; + +-- name: InsertDeviceAlias :exec +INSERT INTO device_alias (id, alias) +VALUES (?, ?); + +-- name: ReplaceDeviceAuth :exec +REPLACE INTO device_auth (id, api_key, extras) +VALUES (?, ?, ?); + +-- name: ReplaceDeviceInfo :exec +REPLACE INTO device_info (id, kind, data) +VALUES (?, ?, ?); + +-- name: ReplaceDeviceAssignment :exec +REPLACE INTO device_assignment (id, created_date, `match`, effect) +VALUES (?, ?, ?, ?); + +-- name: DeleteDeviceInfo :exec +DELETE +FROM device_info +WHERE id = ? + AND kind = ?; + +-- name: DeleteDeviceInfoByID :exec +DELETE +FROM device_info +WHERE id = ?; + +-- name: DeleteDeviceInfoLike :exec +DELETE +FROM device_info +WHERE id LIKE ?; + +-- name: DeleteDeviceAssignment :exec +DELETE +FROM device_assignment +WHERE id = ?; + +-- name: DeleteDeviceAuth :exec +DELETE +FROM device_auth +WHERE id = ?; + +-- name: DeleteDeviceAlias :exec +DELETE +FROM device_alias +WHERE id = ? + AND alias = ?; + +-- name: DeleteDeviceAliasByID :exec +DELETE +FROM device_alias +WHERE id = ?; + +-- name: DeleteDeviceAliasByIDLike :exec +DELETE +FROM device_alias +WHERE id LIKE ?; diff --git a/services/mysqldb/service.go b/services/mysqldb/service.go new file mode 100644 index 0000000..8d64a2f --- /dev/null +++ b/services/mysqldb/service.go @@ -0,0 +1,321 @@ +package mysqldb + +import ( + "context" + "database/sql" + "encoding/json" + "fmt" + lucifer3 "git.aiterp.net/lucifer3/server" + "git.aiterp.net/lucifer3/server/commands" + "git.aiterp.net/lucifer3/server/effects" + "git.aiterp.net/lucifer3/server/events" + "git.aiterp.net/lucifer3/server/internal/gentools" + "git.aiterp.net/lucifer3/server/services/mysqldb/mysqlgen" + "time" + + _ "github.com/go-sql-driver/mysql" +) + +type database struct { + db *sql.DB +} + +func (d *database) Active() bool { + return true +} + +func (d *database) HandleEvent(bus *lucifer3.EventBus, event lucifer3.Event) { + q := mysqlgen.New(d.db) + timeout, cancel := context.WithTimeout(context.Background(), time.Second/2) + defer cancel() + + switch event := event.(type) { + case events.Started: + timeout, cancel := context.WithTimeout(context.Background(), time.Second*10) + defer cancel() + + // Fetch all aliases first + aliases, err := q.ListDeviceAliases(timeout) + if err != nil { + bus.RunEvent(events.Log{ + Level: "error", + Code: "database_could_not_list_aliases", + Message: "Database could not list aliases: " + err.Error(), + }) + } + + auths, err := q.ListDeviceAuth(timeout) + if err != nil { + bus.RunEvent(events.Log{ + Level: "error", + Code: "database_could_not_list_device_infos", + Message: "Hardware State serialization failed: " + err.Error(), + }) + } + for _, auth := range auths { + bus.RunCommand(commands.ConnectDevice{ + ID: auth.ID, + APIKey: auth.ApiKey, + }) + } + + infos, err := q.ListDeviceInfos(timeout) + if err != nil { + bus.RunEvent(events.Log{ + Level: "error", + Code: "database_could_not_list_device_infos", + Message: "Hardware State serialization failed: " + err.Error(), + }) + } + for _, info := range infos { + switch info.Kind { + case "hardware_state": + staleEvent := events.HardwareState{} + err := json.Unmarshal(info.Data, &staleEvent) + if err == nil { + bus.RunEvent(staleEvent) + } + case "hardware_metadata": + staleEvent := events.HardwareMetadata{} + err := json.Unmarshal(info.Data, &staleEvent) + if err == nil { + bus.RunEvent(staleEvent) + } + } + } + + // Run the aliases now + for _, alias := range aliases { + bus.RunCommand(commands.AddAlias{ + Match: alias.ID, + Alias: alias.Alias, + }) + } + + assignments, err := q.ListDeviceAssignments(timeout) + if err != nil { + bus.RunEvent(events.Log{ + Level: "error", + Code: "database_could_not_list_assignments", + Message: "Database could not list assignments: " + err.Error(), + }) + } + for _, ass := range assignments { + effect := effects.Serializable{} + err := json.Unmarshal(ass.Effect, &effect) + if err != nil { + bus.RunEvent(events.Log{ + Level: "error", + Code: "database_could_not_deserialize_effect", + Message: err.Error(), + }) + continue + } + + bus.RunCommand(commands.Assign{ + ID: gentools.ShallowCopy(&ass.ID), + Match: ass.Match, + Effect: effect.Effect, + }) + } + + case events.DeviceAccepted: + if event.Extras == nil { + event.Extras = map[string]string{} + } + + extras, err := json.Marshal(event.Extras) + if err != nil { + extras = []byte("{}") + } + + err = q.ReplaceDeviceAuth(timeout, mysqlgen.ReplaceDeviceAuthParams{ + ID: event.ID, + ApiKey: event.APIKey, + Extras: extras, + }) + if err != nil { + bus.RunEvent(events.Log{ + Level: "error", + Code: "database_could_not_save_device_auth", + Message: "Device auth save failed: " + err.Error(), + }) + } + + case events.HardwareState: + if event.Stale { + return + } + + event.Stale = true + serialized, err := json.Marshal(event) + if err != nil { + bus.RunEvent(events.Log{ + Level: "error", + Code: "database_could_not_serialize_hardware_state", + Message: "Hardware State serialization failed: " + err.Error(), + }) + return + } + + err = q.ReplaceDeviceInfo(timeout, mysqlgen.ReplaceDeviceInfoParams{ + ID: event.ID, + Kind: "hardware_state", + Data: serialized, + }) + if err != nil { + bus.RunEvent(events.Log{ + Level: "error", + Code: "database_could_not_save_hardware_state", + Message: err.Error(), + }) + + return + } + + case events.HardwareMetadata: + if event.Stale { + return + } + + event.Stale = true + serialized, err := json.Marshal(event) + if err != nil { + bus.RunEvent(events.Log{ + Level: "error", + Code: "database_could_not_serialize_hardware_state", + Message: "Hardware State serialization failed: " + err.Error(), + }) + return + } + + err = q.ReplaceDeviceInfo(timeout, mysqlgen.ReplaceDeviceInfoParams{ + ID: event.ID, + Kind: "hardware_metadata", + Data: serialized, + }) + if err != nil { + bus.RunEvent(events.Log{ + Level: "error", + Code: "database_could_not_save_hardware_state", + Message: err.Error(), + }) + + return + } + + case events.AssignmentCreated: + serialized, err := json.Marshal(&effects.Serializable{Effect: event.Effect}) + if err != nil { + bus.RunEvent(events.Log{ + Level: "error", + Code: "database_could_not_serialize_effect", + Message: err.Error(), + }) + return + } + + err = q.ReplaceDeviceAssignment(timeout, mysqlgen.ReplaceDeviceAssignmentParams{ + ID: event.ID, + CreatedDate: time.Now().UTC(), + Match: event.Match, + Effect: serialized, + }) + if err != nil { + bus.RunEvent(events.Log{ + Level: "error", + Code: "database_could_not_save_assignment", + Message: err.Error(), + }) + return + } + + case events.AssignmentRemoved: + err := q.DeleteDeviceAssignment(timeout, event.ID) + if err != nil { + bus.RunEvent(events.Log{ + Level: "error", + Code: "database_could_not_remove_assignment", + Message: err.Error(), + }) + return + } + + case events.AliasAdded: + err := q.InsertDeviceAlias(timeout, mysqlgen.InsertDeviceAliasParams{ + ID: event.ID, + Alias: event.Alias, + }) + if err != nil { + bus.RunEvent(events.Log{ + Level: "error", + Code: "database_could_not_remove_assignment", + Message: err.Error(), + }) + return + } + + case events.AliasRemoved: + err := q.DeleteDeviceAlias(timeout, mysqlgen.DeleteDeviceAliasParams{ + ID: event.ID, + Alias: event.Alias, + }) + if err != nil { + bus.RunEvent(events.Log{ + Level: "error", + Code: "database_could_not_remove_assignment", + Message: err.Error(), + }) + return + } + + case events.DeviceForgotten: + err := q.DeleteDeviceInfoByID(timeout, event.ID) + if err != nil { + bus.RunEvent(events.Log{ + Level: "error", + Code: "database_could_not_remove_info", + Message: "Failed to remove device info: " + err.Error(), + }) + } + _ = q.DeleteDeviceAuth(timeout, event.ID) + err = q.DeleteDeviceAliasByID(timeout, event.ID) + if err != nil { + bus.RunEvent(events.Log{ + Level: "error", + Code: "database_could_not_remove_aliases", + Message: "Failed to remove device aliases: " + err.Error(), + }) + } + err = q.DeleteDeviceAliasByIDLike(timeout, event.ID+":%") + if err != nil { + bus.RunEvent(events.Log{ + Level: "error", + Code: "database_could_not_remove_sub_aliases", + Message: "Failed to remove sub-device aliases: " + err.Error(), + }) + } + } +} + +func (d *database) HandleCommand(*lucifer3.EventBus, lucifer3.Command) {} + +func Connect(host string, port int, username, password, dbname string) (lucifer3.ActiveService, error) { + db, err := sql.Open("mysql", fmt.Sprintf( + "%s:%s@(%s:%d)/%s?parseTime=true", username, password, host, port, dbname, + )) + if err != nil { + return nil, err + } + + db.SetMaxOpenConns(10) + db.SetMaxIdleConns(10) + db.SetConnMaxIdleTime(time.Minute) + + err = db.Ping() + if err != nil { + return nil, err + } + + return &database{db: db}, nil +} diff --git a/services/mysqldb/sqltypes/nullrawmessage.go b/services/mysqldb/sqltypes/nullrawmessage.go new file mode 100644 index 0000000..6fb957f --- /dev/null +++ b/services/mysqldb/sqltypes/nullrawmessage.go @@ -0,0 +1,36 @@ +package sqltypes + +import ( + "database/sql/driver" + "encoding/json" + "errors" +) + +type NullRawMessage struct { + RawMessage json.RawMessage + Valid bool +} + +func (n *NullRawMessage) Scan(value interface{}) error { + if value == nil { + n.RawMessage, n.Valid = json.RawMessage{}, false + return nil + } + buf, ok := value.([]byte) + + if !ok { + return errors.New("cannot parse to bytes") + } + + n.RawMessage, n.Valid = buf, true + + return nil +} + +func (n NullRawMessage) Value() (driver.Value, error) { + if !n.Valid { + return nil, nil + } + + return []byte(n.RawMessage), nil +} diff --git a/services/resolver.go b/services/resolver.go index b5b35c9..b63bf00 100644 --- a/services/resolver.go +++ b/services/resolver.go @@ -79,12 +79,27 @@ func (r *Resolver) HandleEvent(_ *lucifer3.EventBus, event lucifer3.Event) { } } -func (r *Resolver) HandleCommand(_ *lucifer3.EventBus, command lucifer3.Command) { +func (r *Resolver) HandleCommand(bus *lucifer3.EventBus, command lucifer3.Command) { + var aliasEvents []lucifer3.Event + switch command := command.(type) { case commands.AddAlias: r.mu.Lock() for _, ptr := range r.resolve(command.Match) { - ptr.AddAlias(command.Alias) + added, removed := ptr.AddAlias(command.Alias) + + if added != nil { + aliasEvents = append(aliasEvents, events.AliasAdded{ + ID: ptr.ID, + Alias: *added, + }) + } + if removed != nil { + aliasEvents = append(aliasEvents, events.AliasRemoved{ + ID: ptr.ID, + Alias: *removed, + }) + } } if strings.HasPrefix(command.Alias, "lucifer:name:") { r.sortPointers() @@ -94,13 +109,22 @@ func (r *Resolver) HandleCommand(_ *lucifer3.EventBus, command lucifer3.Command) case commands.RemoveAlias: r.mu.Lock() for _, ptr := range r.resolve(command.Match) { - ptr.RemoveAlias(command.Alias) + if ptr.RemoveAlias(command.Alias) { + aliasEvents = append(aliasEvents, events.AliasRemoved{ + ID: ptr.ID, + Alias: command.Alias, + }) + } } if strings.HasPrefix(command.Alias, "lucifer:name:") { r.sortPointers() } r.mu.Unlock() } + + if len(aliasEvents) > 0 { + go bus.RunEvents(aliasEvents) + } } func (r *Resolver) ensure(id string) *device.Pointer { diff --git a/sqlc.yaml b/sqlc.yaml new file mode 100644 index 0000000..30c32e4 --- /dev/null +++ b/sqlc.yaml @@ -0,0 +1,28 @@ +version: "1" +packages: + - name: "mysqlgen" + path: "./services/mysqldb/mysqlgen" + queries: "./services/mysqldb/queries" + schema: "./services/mysqldb/migrations" + engine: "mysql" + emit_prepared_queries: true + emit_interface: false + emit_exact_table_names: false + emit_empty_slices: true + emit_json_tags: false +overrides: + - go_type: "git.aiterp.net/lucifer3/server/services/mysqldb/sqltypes.NullRawMessage" + db_type: "json" + nullable: true + - go_type: "float64" + db_type: "float" + - go_type: "float64" + db_type: "float" + nullable: true + - go_type: "int" + db_type: "int" + - go_type: "github.com/google/uuid.UUID" + db_type: "char" + - go_type: "github.com/google/uuid.NullUUID" + db_type: "char" + nullable: true