Repinoid
3a73207abd
store.Create классифицирует 23505 по имени constraint: (namespace,device_id) → ErrDeviceIDExists, иначе ErrAlreadyExists. Handler отдаёт разные 409-сообщения. Помогает Terraform понять, какое поле конфликтует.
216 lines
7.3 KiB
Go
216 lines
7.3 KiB
Go
package store
|
||
|
||
import (
|
||
"context"
|
||
"database/sql"
|
||
"encoding/json"
|
||
"errors"
|
||
"fmt"
|
||
"strings"
|
||
|
||
"github.com/lib/pq"
|
||
)
|
||
|
||
// Create вставляет новое устройство. Конфликт (namespace, name) или
|
||
// (namespace, device_id) возвращает ErrAlreadyExists.
|
||
func (s *DeviceStore) Create(ctx context.Context, d *Device) error {
|
||
metadata, err := json.Marshal(d.Metadata)
|
||
if err != nil {
|
||
return fmt.Errorf("store: marshal metadata: %w", err)
|
||
}
|
||
// phase пишем ЯВНО через phaseOf(enabled): колонка имеет DEFAULT 'Active',
|
||
// поэтому без явной записи устройство с enabled=false лежало бы в БД как
|
||
// 'Active', а POST-ответ считал бы 'Disabled' → GET отдал бы другое значение
|
||
// (ревью 2026-10-10, 🟡 store). Согласовано с UpdateEnabled (тот же phaseOf).
|
||
_, err = s.db.ExecContext(ctx, `
|
||
INSERT INTO iot_devices (namespace, name, device_id, enabled, mqtt_password, metadata, phase)
|
||
VALUES ($1, $2, $3, $4, $5, $6, $7)`,
|
||
d.Namespace, d.Name, d.DeviceID, d.Enabled, d.MQTTPassword, metadata, phaseOf(d.Enabled),
|
||
)
|
||
if err != nil {
|
||
if isUniqueViolation(err) {
|
||
// Различаем конфликт name и device_id по имени constraint (ревью
|
||
// 2026-10-10): PK — iot_devices_pkey (namespace,name), уникальный
|
||
// индекс — idx_iot_devices_ns_device (namespace,device_id).
|
||
var pqErr *pq.Error
|
||
if errors.As(err, &pqErr) && strings.Contains(pqErr.Constraint, "ns_device") {
|
||
return ErrDeviceIDExists
|
||
}
|
||
return ErrAlreadyExists
|
||
}
|
||
return fmt.Errorf("store: insert device: %w", err)
|
||
}
|
||
return nil
|
||
}
|
||
|
||
// List возвращает устройства namespace (без паролей — API их не отдаёт).
|
||
// limit <= 0 — без ограничения; offset — смещение (пагинация API).
|
||
func (s *DeviceStore) List(ctx context.Context, namespace string, limit, offset int) ([]Device, error) {
|
||
query := `SELECT namespace, name, device_id, enabled, mqtt_password, metadata, phase,
|
||
last_connected, created_at
|
||
FROM iot_devices WHERE namespace = $1 ORDER BY name`
|
||
args := []any{namespace}
|
||
if limit > 0 {
|
||
query += fmt.Sprintf(" LIMIT $%d", len(args)+1)
|
||
args = append(args, limit)
|
||
}
|
||
if offset > 0 {
|
||
query += fmt.Sprintf(" OFFSET $%d", len(args)+1)
|
||
args = append(args, offset)
|
||
}
|
||
rows, err := s.db.QueryContext(ctx, query, args...)
|
||
if err != nil {
|
||
return nil, fmt.Errorf("store: list devices: %w", err)
|
||
}
|
||
defer rows.Close()
|
||
return scanDevices(rows)
|
||
}
|
||
|
||
// GetByName возвращает устройство по (namespace, name).
|
||
func (s *DeviceStore) GetByName(ctx context.Context, namespace, name string) (*Device, error) {
|
||
return s.get(ctx, `SELECT namespace, name, device_id, enabled, mqtt_password, metadata, phase,
|
||
last_connected, created_at
|
||
FROM iot_devices WHERE namespace = $1 AND name = $2`, namespace, name)
|
||
}
|
||
|
||
// GetByDeviceID возвращает устройство по (namespace, device_id) — для MQTT auth.
|
||
func (s *DeviceStore) GetByDeviceID(ctx context.Context, namespace, deviceID string) (*Device, error) {
|
||
return s.get(ctx, `SELECT namespace, name, device_id, enabled, mqtt_password, metadata, phase,
|
||
last_connected, created_at
|
||
FROM iot_devices WHERE namespace = $1 AND device_id = $2`, namespace, deviceID)
|
||
}
|
||
|
||
// get — общая выборка одного устройства.
|
||
func (s *DeviceStore) get(ctx context.Context, query string, args ...any) (*Device, error) {
|
||
d, err := scanDevice(s.db.QueryRowContext(ctx, query, args...))
|
||
if errors.Is(err, sql.ErrNoRows) {
|
||
return nil, ErrNotFound
|
||
}
|
||
if err != nil {
|
||
return nil, fmt.Errorf("store: get device: %w", err)
|
||
}
|
||
return d, nil
|
||
}
|
||
|
||
// UpdateEnabled включает/выключает устройство.
|
||
func (s *DeviceStore) UpdateEnabled(ctx context.Context, namespace, name string, enabled bool) error {
|
||
res, err := s.db.ExecContext(ctx,
|
||
`UPDATE iot_devices SET enabled = $3, phase = $4
|
||
WHERE namespace = $1 AND name = $2`,
|
||
namespace, name, enabled, phaseOf(enabled))
|
||
if err != nil {
|
||
return fmt.Errorf("store: update device: %w", err)
|
||
}
|
||
if n, _ := res.RowsAffected(); n == 0 {
|
||
return ErrNotFound
|
||
}
|
||
return nil
|
||
}
|
||
|
||
// UpdateMetadata обновляет metadata устройства.
|
||
func (s *DeviceStore) UpdateMetadata(ctx context.Context, namespace, name string, metadata map[string]string) error {
|
||
b, err := json.Marshal(metadata)
|
||
if err != nil {
|
||
return fmt.Errorf("store: marshal metadata: %w", err)
|
||
}
|
||
res, err := s.db.ExecContext(ctx,
|
||
`UPDATE iot_devices SET metadata = $3 WHERE namespace = $1 AND name = $2`,
|
||
namespace, name, b)
|
||
if err != nil {
|
||
return fmt.Errorf("store: update metadata: %w", err)
|
||
}
|
||
if n, _ := res.RowsAffected(); n == 0 {
|
||
return ErrNotFound
|
||
}
|
||
return nil
|
||
}
|
||
|
||
// Delete удаляет устройство.
|
||
func (s *DeviceStore) Delete(ctx context.Context, namespace, name string) error {
|
||
res, err := s.db.ExecContext(ctx,
|
||
`DELETE FROM iot_devices WHERE namespace = $1 AND name = $2`, namespace, name)
|
||
if err != nil {
|
||
return fmt.Errorf("store: delete device: %w", err)
|
||
}
|
||
if n, _ := res.RowsAffected(); n == 0 {
|
||
return ErrNotFound
|
||
}
|
||
return nil
|
||
}
|
||
|
||
// TouchLastConnected обновляет last_connected (best-effort при MQTT auth).
|
||
func (s *DeviceStore) TouchLastConnected(ctx context.Context, namespace, deviceID string) {
|
||
_, _ = s.db.ExecContext(ctx, `
|
||
UPDATE iot_devices SET last_connected = now()
|
||
WHERE namespace = $1 AND device_id = $2`, namespace, deviceID)
|
||
}
|
||
|
||
// Count возвращает количество устройств в namespace (для админ-статистики).
|
||
func (s *DeviceStore) Count(ctx context.Context, namespace string) (int64, error) {
|
||
var n int64
|
||
err := s.db.QueryRowContext(ctx,
|
||
`SELECT COUNT(*) FROM iot_devices WHERE namespace = $1`, namespace).Scan(&n)
|
||
if err != nil {
|
||
return 0, fmt.Errorf("store: count devices: %w", err)
|
||
}
|
||
return n, nil
|
||
}
|
||
|
||
// scanDevices читает все строки результата.
|
||
func scanDevices(rows *sql.Rows) ([]Device, error) {
|
||
var out []Device
|
||
for rows.Next() {
|
||
d, err := scanDevice(rows)
|
||
if err != nil {
|
||
return nil, err
|
||
}
|
||
out = append(out, *d)
|
||
}
|
||
return out, rows.Err()
|
||
}
|
||
|
||
// scanDevice читает одну строку устройства.
|
||
type rowScanner interface {
|
||
Scan(dest ...any) error
|
||
}
|
||
|
||
func scanDevice(r rowScanner) (*Device, error) {
|
||
var (
|
||
d Device
|
||
metadata []byte
|
||
lastConn sql.NullTime
|
||
)
|
||
err := r.Scan(&d.Namespace, &d.Name, &d.DeviceID, &d.Enabled, &d.MQTTPassword,
|
||
&metadata, &d.Phase, &lastConn, &d.CreatedAt)
|
||
if err != nil {
|
||
return nil, err
|
||
}
|
||
if len(metadata) > 0 {
|
||
_ = json.Unmarshal(metadata, &d.Metadata)
|
||
}
|
||
if lastConn.Valid {
|
||
t := lastConn.Time
|
||
d.LastConnected = &t
|
||
}
|
||
return &d, nil
|
||
}
|
||
|
||
// phaseOf — статус устройства в зависимости от enabled.
|
||
func phaseOf(enabled bool) string {
|
||
if enabled {
|
||
return "Active"
|
||
}
|
||
return "Disabled"
|
||
}
|
||
|
||
// isUniqueViolation определяет нарушение уникальности PostgreSQL.
|
||
func isUniqueViolation(err error) bool {
|
||
return err != nil && err.Error() != "" && hasSQLState(err, "23505")
|
||
}
|
||
|
||
func hasSQLState(err error, state string) bool {
|
||
type stater interface{ SQLState() string }
|
||
var e stater
|
||
return errors.As(err, &e) && e.SQLState() == state
|
||
}
|