121 lines
3.9 KiB
Go
121 lines
3.9 KiB
Go
package area
|
|
|
|
import (
|
|
"context"
|
|
"database/sql"
|
|
"encoding/json"
|
|
"errors"
|
|
"sync"
|
|
)
|
|
|
|
var (
|
|
ErrNotFound = errors.New("area not found")
|
|
ErrConflict = errors.New("area version conflict")
|
|
)
|
|
|
|
type Store interface {
|
|
Save(context.Context, Area, int64) error
|
|
Get(context.Context, string) (Area, error)
|
|
List(context.Context, string) ([]Area, error)
|
|
}
|
|
type MemoryStore struct {
|
|
mu sync.RWMutex
|
|
versions map[string][]Area
|
|
}
|
|
|
|
func NewMemoryStore() *MemoryStore { return &MemoryStore{versions: map[string][]Area{}} }
|
|
func (s *MemoryStore) Save(_ context.Context, item Area, expected int64) error {
|
|
s.mu.Lock()
|
|
defer s.mu.Unlock()
|
|
values := s.versions[item.ID]
|
|
if len(values) == 0 {
|
|
if expected != 0 {
|
|
return ErrConflict
|
|
}
|
|
} else if values[len(values)-1].Version != expected {
|
|
return ErrConflict
|
|
}
|
|
s.versions[item.ID] = append(values, item)
|
|
return nil
|
|
}
|
|
func (s *MemoryStore) Get(_ context.Context, id string) (Area, error) {
|
|
s.mu.RLock()
|
|
defer s.mu.RUnlock()
|
|
values := s.versions[id]
|
|
if len(values) == 0 {
|
|
return Area{}, ErrNotFound
|
|
}
|
|
return values[len(values)-1], nil
|
|
}
|
|
func (s *MemoryStore) List(_ context.Context, deviceID string) ([]Area, error) {
|
|
s.mu.RLock()
|
|
defer s.mu.RUnlock()
|
|
items := []Area{}
|
|
for _, values := range s.versions {
|
|
item := values[len(values)-1]
|
|
if deviceID == "" || item.DeviceID == deviceID {
|
|
items = append(items, item)
|
|
}
|
|
}
|
|
return items, nil
|
|
}
|
|
|
|
type PostgresStore struct{ database *sql.DB }
|
|
|
|
func NewPostgresStore(db *sql.DB) *PostgresStore { return &PostgresStore{database: db} }
|
|
func (s *PostgresStore) Save(ctx context.Context, item Area, expected int64) error {
|
|
tx, err := s.database.BeginTx(ctx, &sql.TxOptions{Isolation: sql.LevelSerializable})
|
|
if err != nil {
|
|
return err
|
|
}
|
|
defer tx.Rollback()
|
|
var current int64
|
|
err = tx.QueryRowContext(ctx, `SELECT COALESCE(max(version),0) FROM sense_area_versions WHERE id=$1`, item.ID).Scan(¤t)
|
|
if err != nil {
|
|
return err
|
|
}
|
|
if current != expected {
|
|
return ErrConflict
|
|
}
|
|
points, _ := json.Marshal(item.Points)
|
|
_, err = tx.ExecContext(ctx, `INSERT INTO sense_area_versions(id,version,device_id,profile_token,width,height,name,kind,direction,points,enabled,recalibration_required,created_at,updated_at) VALUES($1,$2,$3,$4,$5,$6,$7,$8,$9,$10,$11,$12,$13,$14)`, item.ID, item.Version, item.DeviceID, item.ProfileToken, item.Width, item.Height, item.Name, item.Kind, item.Direction, points, item.Enabled, item.RecalibrationRequired, item.CreatedAt, item.UpdatedAt)
|
|
if err != nil {
|
|
return err
|
|
}
|
|
return tx.Commit()
|
|
}
|
|
func (s *PostgresStore) Get(ctx context.Context, id string) (Area, error) {
|
|
return scan(s.database.QueryRowContext(ctx, `SELECT id,version,device_id,profile_token,width,height,name,kind,direction,points,enabled,recalibration_required,created_at,updated_at FROM sense_area_versions WHERE id=$1 ORDER BY version DESC LIMIT 1`, id))
|
|
}
|
|
|
|
type scanner interface{ Scan(...any) error }
|
|
|
|
func scan(row scanner) (Area, error) {
|
|
var item Area
|
|
var points []byte
|
|
err := row.Scan(&item.ID, &item.Version, &item.DeviceID, &item.ProfileToken, &item.Width, &item.Height, &item.Name, &item.Kind, &item.Direction, &points, &item.Enabled, &item.RecalibrationRequired, &item.CreatedAt, &item.UpdatedAt)
|
|
if errors.Is(err, sql.ErrNoRows) {
|
|
return Area{}, ErrNotFound
|
|
}
|
|
if err == nil {
|
|
err = json.Unmarshal(points, &item.Points)
|
|
}
|
|
return item, err
|
|
}
|
|
func (s *PostgresStore) List(ctx context.Context, deviceID string) ([]Area, error) {
|
|
rows, err := s.database.QueryContext(ctx, `SELECT DISTINCT ON(id) id,version,device_id,profile_token,width,height,name,kind,direction,points,enabled,recalibration_required,created_at,updated_at FROM sense_area_versions WHERE ($1='' OR device_id=$1) ORDER BY id,version DESC`, deviceID)
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
defer rows.Close()
|
|
items := []Area{}
|
|
for rows.Next() {
|
|
item, err := scan(rows)
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
items = append(items, item)
|
|
}
|
|
return items, rows.Err()
|
|
}
|