1207 lines
30 KiB
Go
1207 lines
30 KiB
Go
package postgres
|
|
|
|
import (
|
|
"context"
|
|
"database/sql"
|
|
"embed"
|
|
"errors"
|
|
"fmt"
|
|
"net/url"
|
|
"time"
|
|
|
|
"git.cryptic.systems/volker.raschek/flucky/pkg/types"
|
|
"git.cryptic.systems/volker.raschek/go-logger"
|
|
"github.com/golang-migrate/migrate/v4"
|
|
"github.com/johejo/golang-migrate-extra/source/iofs"
|
|
)
|
|
|
|
//go:embed ddl/*.sql
|
|
var postgresDDLAssets embed.FS
|
|
|
|
//go:embed dml/*.sql
|
|
var postgresDMLAssets embed.FS
|
|
|
|
const (
|
|
postgresDDLAssetPath = "ddl"
|
|
postgresDMLAssetPath = "dml"
|
|
|
|
deleteDeviceByIDSQLFile = postgresDMLAssetPath + "/deleteDeviceByID.sql"
|
|
deleteDeviceByNameSQLFile = postgresDMLAssetPath + "/deleteDeviceByName.sql"
|
|
|
|
deleteSensorByIDSQLFile = postgresDMLAssetPath + "/deleteSensorByID.sql"
|
|
deleteSensorByNameSQLFile = postgresDMLAssetPath + "/deleteSensorByName.sql"
|
|
|
|
insertDeviceSQLFile = postgresDMLAssetPath + "/insertDevice.sql"
|
|
insertSensorSQLFile = postgresDMLAssetPath + "/insertSensor.sql"
|
|
insertHumiditySQLFile = postgresDMLAssetPath + "/insertHumidity.sql"
|
|
insertPressureSQLFile = postgresDMLAssetPath + "/insertPressure.sql"
|
|
insertTemperatureSQLFile = postgresDMLAssetPath + "/insertTemperature.sql"
|
|
|
|
insertOrUpdateDeviceSQLFile = postgresDMLAssetPath + "/insertOrUpdateDevice.sql"
|
|
insertOrUpdateSensorSQLFile = postgresDMLAssetPath + "/insertOrUpdateSensor.sql"
|
|
insertOrUpdateHumiditySQLFile = postgresDMLAssetPath + "/insertOrUpdateHumidity.sql"
|
|
insertOrUpdatePressureSQLFile = postgresDMLAssetPath + "/insertOrUpdatePressure.sql"
|
|
insertOrUpdateTemperatureSQLFile = postgresDMLAssetPath + "/insertOrUpdateTemperature.sql"
|
|
|
|
selectDeviceByIDSQLFile = postgresDMLAssetPath + "/selectDeviceByID.sql"
|
|
selectDeviceByNameSQLFile = postgresDMLAssetPath + "/selectDeviceByName.sql"
|
|
selectDevicesSQLFile = postgresDMLAssetPath + "/selectDevices.sql"
|
|
|
|
selectSensorByIDSQLFile = postgresDMLAssetPath + "/selectSensorByID.sql"
|
|
selectSensorsSQLFile = postgresDMLAssetPath + "/selectSensors.sql"
|
|
selectSensorsByDeviceIDSQLFile = postgresDMLAssetPath + "/selectSensorsByDeviceID.sql"
|
|
selectSensorsByModelSQLFile = postgresDMLAssetPath + "/selectSensorsByModel.sql"
|
|
selectSensorsByNameSQLFile = postgresDMLAssetPath + "/selectSensorsByName.sql"
|
|
|
|
selectHumidityByIDSQLFile = postgresDMLAssetPath + "/selectHumidityByID.sql"
|
|
selectHumiditiesSQLFile = postgresDMLAssetPath + "/selectHumidities.sql"
|
|
|
|
selectPressureByIDSQLFile = postgresDMLAssetPath + "/selectPressureByID.sql"
|
|
selectPressuresSQLFile = postgresDMLAssetPath + "/selectPressures.sql"
|
|
|
|
selectTemperatureByIDSQLFile = postgresDMLAssetPath + "/selectTemperatureByID.sql"
|
|
selectTemperaturesSQLFile = postgresDMLAssetPath + "/selectTemperatures.sql"
|
|
|
|
updateDeviceSQLFile = postgresDMLAssetPath + "/updateDevice.sql"
|
|
updateSensorSQLFile = postgresDMLAssetPath + "/updateSensor.sql"
|
|
)
|
|
|
|
type Opts struct {
|
|
DatabaseURL *url.URL
|
|
Logger logger.Logger
|
|
}
|
|
|
|
func (o *Opts) Validate() error {
|
|
for k, v := range map[string]interface{}{
|
|
"DatabaseURL": o.DatabaseURL,
|
|
"Logger": o.Logger,
|
|
} {
|
|
if v == nil {
|
|
return fmt.Errorf("%v does not be empty or nil", k)
|
|
}
|
|
}
|
|
|
|
if o.DatabaseURL.Scheme != "postgres" {
|
|
return fmt.Errorf("Unsupported database scheme %v for database backend postgres", o.DatabaseURL.Scheme)
|
|
}
|
|
|
|
return nil
|
|
}
|
|
|
|
func New(opts Opts) (*Postgres, error) {
|
|
if err := opts.Validate(); err != nil {
|
|
return nil, err
|
|
}
|
|
|
|
sqlDBO, err := sql.Open(opts.DatabaseURL.Scheme, opts.DatabaseURL.String())
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
|
|
return &Postgres{
|
|
databaseURL: opts.DatabaseURL,
|
|
dbo: sqlDBO,
|
|
flogger: opts.Logger,
|
|
}, nil
|
|
}
|
|
|
|
// Postgres implementation
|
|
type Postgres struct {
|
|
databaseURL *url.URL
|
|
dbo *sql.DB
|
|
flogger logger.Logger
|
|
}
|
|
|
|
// Close closes the database and prevents new queries from starting. Close then
|
|
// waits for all queries that have started processing on the server to finish.
|
|
func (postgres *Postgres) Close() error {
|
|
return postgres.dbo.Close()
|
|
}
|
|
|
|
// RemoveDevices from the database
|
|
func (postgres *Postgres) RemoveDevicesByIDs(ctx context.Context, deviceIDs ...string) error {
|
|
f, err := postgresDMLAssets.ReadFile(deleteDeviceByIDSQLFile)
|
|
if err != nil {
|
|
return fmt.Errorf("Failed to read file %v: %w", deleteDeviceByIDSQLFile, err)
|
|
}
|
|
|
|
tx, err := postgres.dbo.BeginTx(ctx, &sql.TxOptions{ReadOnly: false})
|
|
if err != nil {
|
|
return fmt.Errorf("Failed to begin new transaction: %v", err)
|
|
}
|
|
defer tx.Rollback()
|
|
|
|
stmt, err := tx.Prepare(string(f))
|
|
if err != nil {
|
|
return fmt.Errorf("Failed to prepare statement: %v", err)
|
|
}
|
|
defer stmt.Close()
|
|
|
|
for _, deviceID := range deviceIDs {
|
|
_, err = stmt.Exec(deviceID)
|
|
if err != nil {
|
|
return err
|
|
}
|
|
}
|
|
|
|
return tx.Commit()
|
|
}
|
|
|
|
// RemoveDevices from the database
|
|
func (postgres *Postgres) RemoveDevicesByNames(ctx context.Context, names ...string) error {
|
|
f, err := postgresDMLAssets.ReadFile(deleteDeviceByNameSQLFile)
|
|
if err != nil {
|
|
return fmt.Errorf("Failed to read file %v: %w", deleteDeviceByNameSQLFile, err)
|
|
}
|
|
|
|
tx, err := postgres.dbo.BeginTx(ctx, &sql.TxOptions{ReadOnly: false})
|
|
if err != nil {
|
|
return fmt.Errorf("Failed to begin new transaction: %v", err)
|
|
}
|
|
defer tx.Rollback()
|
|
|
|
stmt, err := tx.Prepare(string(f))
|
|
if err != nil {
|
|
return fmt.Errorf("Failed to prepare statement: %v", err)
|
|
}
|
|
defer stmt.Close()
|
|
|
|
for _, deviceID := range names {
|
|
_, err = stmt.Exec(deviceID)
|
|
if err != nil {
|
|
return err
|
|
}
|
|
}
|
|
|
|
return tx.Commit()
|
|
}
|
|
|
|
// RemoveSensors from the database
|
|
func (postgres *Postgres) RemoveSensorsByIDs(ctx context.Context, sensorIDs ...string) error {
|
|
f, err := postgresDMLAssets.ReadFile(deleteSensorByIDSQLFile)
|
|
if err != nil {
|
|
return fmt.Errorf("Failed to read file %v: %w", deleteSensorByIDSQLFile, err)
|
|
}
|
|
|
|
tx, err := postgres.dbo.BeginTx(ctx, &sql.TxOptions{ReadOnly: false})
|
|
if err != nil {
|
|
return fmt.Errorf("Failed to begin new transaction: %v", err)
|
|
}
|
|
defer tx.Rollback()
|
|
|
|
stmt, err := tx.Prepare(string(f))
|
|
if err != nil {
|
|
return fmt.Errorf("Failed to prepare statement: %v", err)
|
|
}
|
|
defer stmt.Close()
|
|
|
|
for _, sensorID := range sensorIDs {
|
|
_, err = stmt.Exec(sensorID)
|
|
if err != nil {
|
|
return err
|
|
}
|
|
}
|
|
|
|
return tx.Commit()
|
|
}
|
|
|
|
// RemoveSensors from the database
|
|
func (postgres *Postgres) RemoveSensorsByNames(ctx context.Context, sensorIDs ...string) error {
|
|
f, err := postgresDMLAssets.ReadFile(deleteSensorByNameSQLFile)
|
|
if err != nil {
|
|
return fmt.Errorf("Failed to read file %v: %w", deleteSensorByNameSQLFile, err)
|
|
}
|
|
|
|
tx, err := postgres.dbo.BeginTx(ctx, &sql.TxOptions{ReadOnly: false})
|
|
if err != nil {
|
|
return fmt.Errorf("Failed to begin new transaction: %v", err)
|
|
}
|
|
defer tx.Rollback()
|
|
|
|
stmt, err := tx.Prepare(string(f))
|
|
if err != nil {
|
|
return fmt.Errorf("Failed to prepare statement: %v", err)
|
|
}
|
|
defer stmt.Close()
|
|
|
|
for _, sensorID := range sensorIDs {
|
|
_, err = stmt.Exec(sensorID)
|
|
if err != nil {
|
|
return err
|
|
}
|
|
}
|
|
|
|
return tx.Commit()
|
|
}
|
|
|
|
// AddDevices into the database
|
|
func (postgres *Postgres) AddDevices(ctx context.Context, devices ...*types.Device) error {
|
|
tx, err := postgres.dbo.BeginTx(ctx, &sql.TxOptions{ReadOnly: false})
|
|
if err != nil {
|
|
return fmt.Errorf("Failed to begin new transaction: %v", err)
|
|
}
|
|
defer tx.Rollback()
|
|
|
|
err = postgres.insertDevices(tx, devices...)
|
|
if err != nil {
|
|
return err
|
|
}
|
|
|
|
return tx.Commit()
|
|
}
|
|
|
|
func (postgres *Postgres) insertDevices(tx *sql.Tx, devices ...*types.Device) error {
|
|
f, err := postgresDMLAssets.ReadFile(insertDeviceSQLFile)
|
|
if err != nil {
|
|
return fmt.Errorf("Failed to read file %v: %w", insertDeviceSQLFile, err)
|
|
}
|
|
|
|
stmt, err := tx.Prepare(string(f))
|
|
if err != nil {
|
|
return fmt.Errorf("Failed to prepare statement: %v", err)
|
|
}
|
|
defer stmt.Close()
|
|
|
|
for _, device := range devices {
|
|
|
|
if device.CreationDate.Equal(time.Time{}) {
|
|
device.CreationDate = time.Now()
|
|
}
|
|
|
|
_, err = stmt.Exec(
|
|
&device.ID,
|
|
&device.Name,
|
|
&device.Location,
|
|
&device.CreationDate,
|
|
&device.UpdateDate,
|
|
)
|
|
if err != nil {
|
|
return fmt.Errorf("Failed to execute statement: %v", err)
|
|
}
|
|
}
|
|
|
|
return nil
|
|
}
|
|
|
|
// AddMeasuredValues into the database
|
|
func (postgres *Postgres) AddMeasuredValues(ctx context.Context, measuredValues ...*types.MeasuredValue) error {
|
|
splittedMeasuredValues := make(map[types.MeasuredValueType][]*types.MeasuredValue, 0)
|
|
|
|
for _, measuredValue := range measuredValues {
|
|
if _, ok := splittedMeasuredValues[measuredValue.ValueType]; !ok {
|
|
splittedMeasuredValues[measuredValue.ValueType] = make([]*types.MeasuredValue, 0)
|
|
}
|
|
splittedMeasuredValues[measuredValue.ValueType] = append(splittedMeasuredValues[measuredValue.ValueType], measuredValue)
|
|
}
|
|
|
|
tx, err := postgres.dbo.BeginTx(ctx, &sql.TxOptions{ReadOnly: false})
|
|
if err != nil {
|
|
return fmt.Errorf("Failed to begin new transaction: %v", err)
|
|
}
|
|
defer tx.Rollback()
|
|
|
|
// General insert function
|
|
insert := func(tx *sql.Tx, queryFile string, measuredValues []*types.MeasuredValue) error {
|
|
f, err := postgresDMLAssets.ReadFile(queryFile)
|
|
if err != nil {
|
|
return fmt.Errorf("Failed to read file %v: %w", queryFile, err)
|
|
}
|
|
|
|
stmt, err := tx.Prepare(string(f))
|
|
if err != nil {
|
|
return fmt.Errorf("Failed to prepare statement: %v", err)
|
|
}
|
|
defer stmt.Close()
|
|
|
|
for _, measuredValue := range measuredValues {
|
|
|
|
if measuredValue.CreationDate.Equal(time.Time{}) {
|
|
measuredValue.CreationDate = time.Now()
|
|
}
|
|
|
|
_, err := stmt.Exec(
|
|
&measuredValue.ID,
|
|
&measuredValue.Value,
|
|
&measuredValue.Date,
|
|
&measuredValue.SensorID,
|
|
&measuredValue.CreationDate,
|
|
&measuredValue.UpdateDate,
|
|
)
|
|
|
|
if err != nil {
|
|
return fmt.Errorf("Failed to execute statement: %v", err)
|
|
}
|
|
}
|
|
|
|
return nil
|
|
}
|
|
|
|
for measuredValueType, measuredValues := range splittedMeasuredValues {
|
|
var queryFile string
|
|
|
|
switch measuredValueType {
|
|
case types.Humidity:
|
|
queryFile = insertHumiditySQLFile
|
|
case types.Pressure:
|
|
queryFile = insertPressureSQLFile
|
|
case types.Temperature:
|
|
queryFile = insertTemperatureSQLFile
|
|
default:
|
|
return fmt.Errorf("Measured value type %v not supported", measuredValueType)
|
|
}
|
|
|
|
err := insert(tx, queryFile, measuredValues)
|
|
if err != nil {
|
|
return err
|
|
}
|
|
}
|
|
|
|
return tx.Commit()
|
|
}
|
|
|
|
func (postgres *Postgres) AddOrUpdateDevices(ctx context.Context, devices ...*types.Device) error {
|
|
f, err := postgresDMLAssets.ReadFile(insertOrUpdateDeviceSQLFile)
|
|
if err != nil {
|
|
return fmt.Errorf("Failed to read file %v: %w", insertOrUpdateDeviceSQLFile, err)
|
|
}
|
|
|
|
tx, err := postgres.dbo.BeginTx(ctx, &sql.TxOptions{ReadOnly: false})
|
|
if err != nil {
|
|
return fmt.Errorf("Failed to begin new transaction: %v", err)
|
|
}
|
|
defer tx.Rollback()
|
|
|
|
stmt, err := tx.Prepare(string(f))
|
|
if err != nil {
|
|
return fmt.Errorf("Failed to prepare statement: %w", err)
|
|
}
|
|
defer stmt.Close()
|
|
|
|
for _, device := range devices {
|
|
|
|
if device.CreationDate.Equal(time.Time{}) {
|
|
device.CreationDate = time.Now()
|
|
}
|
|
|
|
_, err = stmt.Exec(
|
|
&device.ID,
|
|
&device.Name,
|
|
&device.Location,
|
|
&device.CreationDate,
|
|
&device.UpdateDate,
|
|
)
|
|
if err != nil {
|
|
return fmt.Errorf("Failed to execute statement: %v", err)
|
|
}
|
|
}
|
|
|
|
return tx.Commit()
|
|
}
|
|
|
|
// AddOrUpdateMeasuredValues into the database
|
|
func (postgres *Postgres) AddOrUpdateMeasuredValues(ctx context.Context, measuredValues ...*types.MeasuredValue) error {
|
|
splittedMeasuredValues := make(map[types.MeasuredValueType][]*types.MeasuredValue, 0)
|
|
|
|
for _, measuredValue := range measuredValues {
|
|
if _, ok := splittedMeasuredValues[measuredValue.ValueType]; !ok {
|
|
splittedMeasuredValues[measuredValue.ValueType] = make([]*types.MeasuredValue, 0)
|
|
}
|
|
splittedMeasuredValues[measuredValue.ValueType] = append(splittedMeasuredValues[measuredValue.ValueType], measuredValue)
|
|
}
|
|
|
|
tx, err := postgres.dbo.BeginTx(ctx, &sql.TxOptions{ReadOnly: false})
|
|
if err != nil {
|
|
return fmt.Errorf("Failed to begin new transaction: %v", err)
|
|
}
|
|
defer tx.Rollback()
|
|
|
|
// General insert function
|
|
insert := func(tx *sql.Tx, queryFile string, measuredValues []*types.MeasuredValue) error {
|
|
f, err := postgresDMLAssets.ReadFile(queryFile)
|
|
if err != nil {
|
|
return fmt.Errorf("Failed to read file %v: %w", queryFile, err)
|
|
}
|
|
|
|
stmt, err := tx.Prepare(string(f))
|
|
if err != nil {
|
|
return fmt.Errorf("Failed to prepare statement: %v", err)
|
|
}
|
|
defer stmt.Close()
|
|
|
|
for _, measuredValue := range measuredValues {
|
|
|
|
if measuredValue.CreationDate.Equal(time.Time{}) {
|
|
measuredValue.CreationDate = time.Now()
|
|
}
|
|
|
|
_, err := stmt.Exec(
|
|
&measuredValue.ID,
|
|
&measuredValue.Value,
|
|
&measuredValue.Date,
|
|
&measuredValue.SensorID,
|
|
&measuredValue.CreationDate,
|
|
&measuredValue.UpdateDate,
|
|
)
|
|
|
|
if err != nil {
|
|
return fmt.Errorf("Failed to execute statement: %v", err)
|
|
}
|
|
}
|
|
|
|
return nil
|
|
}
|
|
|
|
for measuredValueType, measuredValues := range splittedMeasuredValues {
|
|
var queryFile string
|
|
|
|
switch measuredValueType {
|
|
case types.Humidity:
|
|
queryFile = insertOrUpdateHumiditySQLFile
|
|
case types.Pressure:
|
|
queryFile = insertOrUpdatePressureSQLFile
|
|
case types.Temperature:
|
|
queryFile = insertOrUpdateTemperatureSQLFile
|
|
default:
|
|
return fmt.Errorf("Measured value type %v not supported", measuredValueType)
|
|
}
|
|
|
|
err := insert(tx, queryFile, measuredValues)
|
|
if err != nil {
|
|
return err
|
|
}
|
|
}
|
|
|
|
return tx.Commit()
|
|
}
|
|
|
|
// AddOrUpdateSensors into the database
|
|
func (postgres *Postgres) AddOrUpdateSensors(ctx context.Context, sensors ...*types.Sensor) error {
|
|
f, err := postgresDMLAssets.ReadFile(insertOrUpdateSensorSQLFile)
|
|
if err != nil {
|
|
return fmt.Errorf("Failed to read file %v: %w", insertOrUpdateSensorSQLFile, err)
|
|
}
|
|
|
|
tx, err := postgres.dbo.BeginTx(ctx, &sql.TxOptions{ReadOnly: false})
|
|
if err != nil {
|
|
return fmt.Errorf("Failed to begin new transaction: %v", err)
|
|
}
|
|
defer tx.Rollback()
|
|
|
|
stmt, err := tx.Prepare(string(f))
|
|
if err != nil {
|
|
return fmt.Errorf("Failed to prepare statement: %v", err)
|
|
}
|
|
defer stmt.Close()
|
|
|
|
for _, sensor := range sensors {
|
|
|
|
if sensor.CreationDate.Equal(time.Time{}) {
|
|
sensor.CreationDate = time.Now()
|
|
}
|
|
|
|
_, err = stmt.Exec(
|
|
&sensor.ID,
|
|
&sensor.Name,
|
|
&sensor.Location,
|
|
&sensor.WireID,
|
|
&sensor.I2CBus,
|
|
&sensor.I2CAddress,
|
|
&sensor.GPIONumber,
|
|
&sensor.Model,
|
|
&sensor.Enabled,
|
|
&sensor.TickDuration,
|
|
&sensor.DeviceID,
|
|
&sensor.CreationDate,
|
|
&sensor.UpdateDate,
|
|
)
|
|
if err != nil {
|
|
return fmt.Errorf("Failed to execute statement: %v", err)
|
|
}
|
|
}
|
|
|
|
return tx.Commit()
|
|
}
|
|
|
|
// AddSensors into the database
|
|
func (postgres *Postgres) AddSensors(ctx context.Context, sensors ...*types.Sensor) error {
|
|
f, err := postgresDMLAssets.ReadFile(insertSensorSQLFile)
|
|
if err != nil {
|
|
return fmt.Errorf("Failed to read file %v: %w", insertSensorSQLFile, err)
|
|
}
|
|
|
|
tx, err := postgres.dbo.BeginTx(ctx, &sql.TxOptions{ReadOnly: false})
|
|
if err != nil {
|
|
return fmt.Errorf("Failed to begin new transaction: %v", err)
|
|
}
|
|
defer tx.Rollback()
|
|
|
|
stmt, err := tx.Prepare(string(f))
|
|
if err != nil {
|
|
return fmt.Errorf("Failed to prepare statement: %v", err)
|
|
}
|
|
defer stmt.Close()
|
|
|
|
for _, sensor := range sensors {
|
|
|
|
if sensor.CreationDate.Equal(time.Time{}) {
|
|
sensor.CreationDate = time.Now()
|
|
}
|
|
|
|
_, err = stmt.Exec(
|
|
&sensor.ID,
|
|
&sensor.Name,
|
|
&sensor.Location,
|
|
&sensor.WireID,
|
|
&sensor.I2CBus,
|
|
&sensor.I2CAddress,
|
|
&sensor.GPIONumber,
|
|
&sensor.Model,
|
|
&sensor.Enabled,
|
|
&sensor.TickDuration,
|
|
&sensor.DeviceID,
|
|
&sensor.CreationDate,
|
|
&sensor.UpdateDate,
|
|
)
|
|
if err != nil {
|
|
return fmt.Errorf("Failed to execute statement: %v", err)
|
|
}
|
|
}
|
|
|
|
return tx.Commit()
|
|
}
|
|
|
|
// Migrate creates all required tables if not exist
|
|
func (postgres *Postgres) Migrate(ctx context.Context) error {
|
|
sourceDriver, err := iofs.New(postgresDDLAssets, postgresDDLAssetPath)
|
|
if err != nil {
|
|
return err
|
|
}
|
|
|
|
m, err := migrate.NewWithSourceInstance("iofs", sourceDriver, postgres.databaseURL.String())
|
|
if err != nil {
|
|
return err
|
|
}
|
|
|
|
err = m.Up()
|
|
switch {
|
|
case errors.Is(err, migrate.ErrNoChange):
|
|
return nil
|
|
default:
|
|
return err
|
|
}
|
|
}
|
|
|
|
// GetDevice from database
|
|
func (postgres *Postgres) GetDeviceByID(ctx context.Context, id string) (*types.Device, error) {
|
|
f, err := postgresDMLAssets.ReadFile(selectDeviceByIDSQLFile)
|
|
if err != nil {
|
|
return nil, fmt.Errorf("Failed to read file %v: %w", selectDeviceByIDSQLFile, err)
|
|
}
|
|
|
|
tx, err := postgres.dbo.BeginTx(ctx, &sql.TxOptions{ReadOnly: true})
|
|
if err != nil {
|
|
return nil, fmt.Errorf("Failed to begin new transaction: %v", err)
|
|
}
|
|
defer tx.Rollback()
|
|
|
|
devices, err := postgres.selectDevices(tx, string(f), id)
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
|
|
err = tx.Commit()
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
|
|
if len(devices) == 0 {
|
|
return nil, nil
|
|
}
|
|
|
|
return devices[0], nil
|
|
}
|
|
|
|
// GetDevice from database
|
|
func (postgres *Postgres) GetDeviceByName(ctx context.Context, name string) (*types.Device, error) {
|
|
f, err := postgresDMLAssets.ReadFile(selectDeviceByNameSQLFile)
|
|
if err != nil {
|
|
return nil, fmt.Errorf("Failed to read file %v: %w", selectDeviceByNameSQLFile, err)
|
|
}
|
|
|
|
tx, err := postgres.dbo.BeginTx(ctx, &sql.TxOptions{ReadOnly: true})
|
|
if err != nil {
|
|
return nil, fmt.Errorf("Failed to begin new transaction: %v", err)
|
|
}
|
|
defer tx.Rollback()
|
|
|
|
devices, err := postgres.selectDevices(tx, string(f), name)
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
|
|
err = tx.Commit()
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
|
|
if len(devices) == 0 {
|
|
return nil, nil
|
|
}
|
|
|
|
return devices[0], nil
|
|
}
|
|
|
|
// GetDevices from the database
|
|
func (postgres *Postgres) GetDevices(ctx context.Context) ([]*types.Device, error) {
|
|
f, err := postgresDMLAssets.ReadFile(selectDevicesSQLFile)
|
|
if err != nil {
|
|
return nil, fmt.Errorf("Failed to read file %v: %w", selectDevicesSQLFile, err)
|
|
}
|
|
|
|
tx, err := postgres.dbo.BeginTx(ctx, &sql.TxOptions{ReadOnly: true})
|
|
if err != nil {
|
|
return nil, fmt.Errorf("Failed to begin new transaction: %v", err)
|
|
}
|
|
defer tx.Rollback()
|
|
|
|
devices, err := postgres.selectDevices(tx, string(f))
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
|
|
err = tx.Commit()
|
|
if err != nil {
|
|
return nil, fmt.Errorf("Failed to commit transaction: %v", err)
|
|
}
|
|
|
|
return devices, nil
|
|
}
|
|
|
|
func (postgres *Postgres) selectDevices(tx *sql.Tx, query string, args ...interface{}) ([]*types.Device, error) {
|
|
stmt, err := tx.Prepare(query)
|
|
if err != nil {
|
|
return nil, fmt.Errorf("Failed to prepare statement: %v", err)
|
|
}
|
|
defer stmt.Close()
|
|
|
|
rows, err := stmt.Query(args...)
|
|
if err != nil {
|
|
return nil, fmt.Errorf("Failed to query statement: %v", err)
|
|
}
|
|
|
|
devices := make([]*types.Device, 0)
|
|
for rows.Next() {
|
|
device := new(types.Device)
|
|
err = rows.Scan(
|
|
&device.ID,
|
|
&device.Name,
|
|
&device.Location,
|
|
&device.CreationDate,
|
|
&device.UpdateDate,
|
|
)
|
|
if err != nil {
|
|
return nil, fmt.Errorf("Failed to scan row: %v", err)
|
|
}
|
|
devices = append(devices, device)
|
|
}
|
|
|
|
return devices, nil
|
|
}
|
|
|
|
// GetHumidity returns humidity from the database
|
|
func (postgres *Postgres) GetHumidityByID(ctx context.Context, id string) (*types.MeasuredValue, error) {
|
|
f, err := postgresDMLAssets.ReadFile(selectHumidityByIDSQLFile)
|
|
if err != nil {
|
|
return nil, fmt.Errorf("Failed to read file %v: %w", selectHumidityByIDSQLFile, err)
|
|
}
|
|
|
|
tx, err := postgres.dbo.BeginTx(ctx, &sql.TxOptions{ReadOnly: true})
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
defer tx.Rollback()
|
|
|
|
measuredValues, err := postgres.selectMeasuredValue(tx, string(f), id)
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
|
|
err = tx.Commit()
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
|
|
if measuredValues == nil {
|
|
return nil, nil
|
|
}
|
|
|
|
for _, measuredValue := range measuredValues {
|
|
measuredValue.ValueType = types.Humidity
|
|
}
|
|
|
|
return measuredValues[0], nil
|
|
}
|
|
|
|
// GetHumidities returns humidities from the database
|
|
func (postgres *Postgres) GetHumidities(ctx context.Context) ([]*types.MeasuredValue, error) {
|
|
f, err := postgresDMLAssets.ReadFile(selectHumiditiesSQLFile)
|
|
if err != nil {
|
|
return nil, fmt.Errorf("Failed to read file %v: %w", selectHumiditiesSQLFile, err)
|
|
}
|
|
|
|
tx, err := postgres.dbo.BeginTx(ctx, &sql.TxOptions{ReadOnly: true})
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
defer tx.Rollback()
|
|
|
|
measuredValues, err := postgres.selectMeasuredValue(tx, string(f))
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
|
|
err = tx.Commit()
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
|
|
for _, measuredValue := range measuredValues {
|
|
measuredValue.ValueType = types.Humidity
|
|
}
|
|
|
|
return measuredValues, nil
|
|
}
|
|
|
|
func (postgres *Postgres) selectMeasuredValue(tx *sql.Tx, query string, args ...interface{}) ([]*types.MeasuredValue, error) {
|
|
stmt, err := tx.Prepare(query)
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
defer stmt.Close()
|
|
|
|
rows, err := stmt.Query(args...)
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
|
|
measuredValues := make([]*types.MeasuredValue, 0)
|
|
for rows.Next() {
|
|
measuredValue := new(types.MeasuredValue)
|
|
err := rows.Scan(
|
|
&measuredValue.ID,
|
|
&measuredValue.Value,
|
|
&measuredValue.Date,
|
|
&measuredValue.SensorID,
|
|
&measuredValue.CreationDate,
|
|
&measuredValue.UpdateDate,
|
|
)
|
|
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
|
|
measuredValues = append(measuredValues, measuredValue)
|
|
}
|
|
|
|
return measuredValues, nil
|
|
}
|
|
|
|
// GetPressure returns pressure from the database
|
|
func (postgres *Postgres) GetPressureByID(ctx context.Context, id string) (*types.MeasuredValue, error) {
|
|
f, err := postgresDMLAssets.ReadFile(selectPressureByIDSQLFile)
|
|
if err != nil {
|
|
return nil, fmt.Errorf("Failed to read file %v: %w", selectPressureByIDSQLFile, err)
|
|
}
|
|
|
|
tx, err := postgres.dbo.BeginTx(ctx, &sql.TxOptions{ReadOnly: true})
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
defer tx.Rollback()
|
|
|
|
measuredValues, err := postgres.selectMeasuredValue(tx, string(f), id)
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
|
|
err = tx.Commit()
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
|
|
if measuredValues == nil {
|
|
return nil, nil
|
|
}
|
|
|
|
for _, measuredValue := range measuredValues {
|
|
measuredValue.ValueType = types.Pressure
|
|
}
|
|
|
|
return measuredValues[0], nil
|
|
}
|
|
|
|
// GetPressures returns pressure from the database
|
|
func (postgres *Postgres) GetPressures(ctx context.Context) ([]*types.MeasuredValue, error) {
|
|
f, err := postgresDMLAssets.ReadFile(selectPressuresSQLFile)
|
|
if err != nil {
|
|
return nil, fmt.Errorf("Failed to read file %v: %w", selectPressuresSQLFile, err)
|
|
}
|
|
|
|
tx, err := postgres.dbo.BeginTx(ctx, &sql.TxOptions{ReadOnly: true})
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
defer tx.Rollback()
|
|
|
|
measuredValues, err := postgres.selectMeasuredValue(tx, string(f))
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
|
|
err = tx.Commit()
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
|
|
for _, measuredValue := range measuredValues {
|
|
measuredValue.ValueType = types.Pressure
|
|
}
|
|
|
|
return measuredValues, nil
|
|
}
|
|
|
|
// GetSensor from database
|
|
func (postgres *Postgres) GetSensorByID(ctx context.Context, id string) (*types.Sensor, error) {
|
|
f, err := postgresDMLAssets.ReadFile(selectSensorByIDSQLFile)
|
|
if err != nil {
|
|
return nil, fmt.Errorf("Failed to read file %v: %w", selectSensorByIDSQLFile, err)
|
|
}
|
|
|
|
tx, err := postgres.dbo.BeginTx(ctx, &sql.TxOptions{ReadOnly: true})
|
|
if err != nil {
|
|
return nil, fmt.Errorf("Failed to begin new transaction: %v", err)
|
|
}
|
|
defer tx.Rollback()
|
|
|
|
sensors, err := postgres.selectSensors(tx, string(f), id)
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
|
|
err = tx.Commit()
|
|
if err != nil {
|
|
return nil, fmt.Errorf("Failed to commit transaction: %v", err)
|
|
}
|
|
|
|
if len(sensors) == 0 {
|
|
return nil, nil
|
|
}
|
|
|
|
return sensors[0], nil
|
|
}
|
|
|
|
// GetSensors from the database
|
|
func (postgres *Postgres) GetSensors(ctx context.Context) ([]*types.Sensor, error) {
|
|
f, err := postgresDMLAssets.ReadFile(selectSensorsSQLFile)
|
|
if err != nil {
|
|
return nil, fmt.Errorf("Failed to read file %v: %w", selectSensorsSQLFile, err)
|
|
}
|
|
|
|
tx, err := postgres.dbo.BeginTx(ctx, &sql.TxOptions{ReadOnly: true})
|
|
if err != nil {
|
|
return nil, fmt.Errorf("Failed to begin new transaction: %v", err)
|
|
}
|
|
defer tx.Rollback()
|
|
|
|
sensors, err := postgres.selectSensors(tx, string(f))
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
|
|
err = tx.Commit()
|
|
if err != nil {
|
|
return nil, fmt.Errorf("Failed to commit transaction: %v", err)
|
|
}
|
|
|
|
return sensors, nil
|
|
}
|
|
|
|
// GetSensorsByModels from the database
|
|
func (postgres *Postgres) GetSensorsByDeviceIDs(ctx context.Context, deviceIDs ...string) ([]*types.Sensor, error) {
|
|
f, err := postgresDMLAssets.ReadFile(selectSensorsByDeviceIDSQLFile)
|
|
if err != nil {
|
|
return nil, fmt.Errorf("Failed to read file %v: %w", selectSensorsByDeviceIDSQLFile, err)
|
|
}
|
|
|
|
tx, err := postgres.dbo.BeginTx(ctx, &sql.TxOptions{ReadOnly: true})
|
|
if err != nil {
|
|
return nil, fmt.Errorf("Failed to begin new transaction: %v", err)
|
|
}
|
|
defer tx.Rollback()
|
|
|
|
cachedSensors := make([]*types.Sensor, 0)
|
|
for i := range deviceIDs {
|
|
sensors, err := postgres.selectSensors(tx, string(f), deviceIDs[i])
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
|
|
cachedSensors = append(cachedSensors, sensors...)
|
|
}
|
|
|
|
err = tx.Commit()
|
|
if err != nil {
|
|
return nil, fmt.Errorf("Failed to commit transaction: %v", err)
|
|
}
|
|
|
|
return cachedSensors, nil
|
|
}
|
|
|
|
// GetSensorsByModel from the database
|
|
func (postgres *Postgres) GetSensorsByModels(ctx context.Context, sensorModels ...string) ([]*types.Sensor, error) {
|
|
f, err := postgresDMLAssets.ReadFile(selectSensorsByModelSQLFile)
|
|
if err != nil {
|
|
return nil, fmt.Errorf("Failed to read file %v: %w", selectSensorsByModelSQLFile, err)
|
|
}
|
|
|
|
tx, err := postgres.dbo.BeginTx(ctx, &sql.TxOptions{ReadOnly: true})
|
|
if err != nil {
|
|
return nil, fmt.Errorf("Failed to begin new transaction: %v", err)
|
|
}
|
|
|
|
cachedSensors := make([]*types.Sensor, 0)
|
|
for i := range sensorModels {
|
|
sensors, err := postgres.selectSensors(tx, string(f), sensorModels[i])
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
|
|
cachedSensors = append(cachedSensors, sensors...)
|
|
}
|
|
|
|
err = tx.Commit()
|
|
if err != nil {
|
|
return nil, fmt.Errorf("Failed to commit transaction: %v", err)
|
|
}
|
|
|
|
return cachedSensors, nil
|
|
}
|
|
|
|
// GetSensorsByModel from the database
|
|
func (postgres *Postgres) GetSensorsByNames(ctx context.Context, sensorNames ...string) ([]*types.Sensor, error) {
|
|
f, err := postgresDMLAssets.ReadFile(selectSensorsByNameSQLFile)
|
|
if err != nil {
|
|
return nil, fmt.Errorf("Failed to read file %v: %w", selectSensorsByNameSQLFile, err)
|
|
}
|
|
|
|
tx, err := postgres.dbo.BeginTx(ctx, &sql.TxOptions{ReadOnly: true})
|
|
if err != nil {
|
|
return nil, fmt.Errorf("Failed to begin new transaction: %v", err)
|
|
}
|
|
defer tx.Rollback()
|
|
|
|
cachedSensors := make([]*types.Sensor, 0)
|
|
for i := range sensorNames {
|
|
sensors, err := postgres.selectSensors(tx, string(f), sensorNames[i])
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
|
|
cachedSensors = append(cachedSensors, sensors...)
|
|
}
|
|
|
|
err = tx.Commit()
|
|
if err != nil {
|
|
return nil, fmt.Errorf("Failed to commit transaction: %v", err)
|
|
}
|
|
|
|
return cachedSensors, nil
|
|
}
|
|
|
|
func (postgres *Postgres) selectSensors(tx *sql.Tx, query string, args ...interface{}) ([]*types.Sensor, error) {
|
|
stmt, err := tx.Prepare(query)
|
|
if err != nil {
|
|
return nil, fmt.Errorf("Failed to prepare statement: %v", err)
|
|
}
|
|
defer stmt.Close()
|
|
|
|
rows, err := stmt.Query(args...)
|
|
if err != nil {
|
|
return nil, fmt.Errorf("Failed to query statement: %v", err)
|
|
}
|
|
|
|
sensors := make([]*types.Sensor, 0)
|
|
for rows.Next() {
|
|
sensor := new(types.Sensor)
|
|
err = rows.Scan(
|
|
&sensor.ID,
|
|
&sensor.Name,
|
|
&sensor.Location,
|
|
&sensor.WireID,
|
|
&sensor.I2CBus,
|
|
&sensor.I2CAddress,
|
|
&sensor.GPIONumber,
|
|
&sensor.Model,
|
|
&sensor.Enabled,
|
|
&sensor.TickDuration,
|
|
&sensor.DeviceID,
|
|
&sensor.CreationDate,
|
|
&sensor.UpdateDate,
|
|
)
|
|
|
|
if err != nil {
|
|
return nil, fmt.Errorf("Failed to scan row: %v", err)
|
|
}
|
|
sensors = append(sensors, sensor)
|
|
}
|
|
|
|
return sensors, nil
|
|
}
|
|
|
|
// GetTemperature returns temperatures from the database
|
|
func (postgres *Postgres) GetTemperatureByID(ctx context.Context, id string) (*types.MeasuredValue, error) {
|
|
f, err := postgresDMLAssets.ReadFile(selectTemperatureByIDSQLFile)
|
|
if err != nil {
|
|
return nil, fmt.Errorf("Failed to read file %v: %w", selectTemperatureByIDSQLFile, err)
|
|
}
|
|
|
|
tx, err := postgres.dbo.BeginTx(ctx, &sql.TxOptions{ReadOnly: true})
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
defer tx.Rollback()
|
|
|
|
measuredValues, err := postgres.selectMeasuredValue(tx, string(f), id)
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
|
|
err = tx.Commit()
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
|
|
if measuredValues == nil {
|
|
return nil, nil
|
|
}
|
|
|
|
for _, measuredValue := range measuredValues {
|
|
measuredValue.ValueType = types.Temperature
|
|
}
|
|
|
|
return measuredValues[0], nil
|
|
}
|
|
|
|
// GetTemperatures returns temperatures from the database
|
|
func (postgres *Postgres) GetTemperatures(ctx context.Context) ([]*types.MeasuredValue, error) {
|
|
f, err := postgresDMLAssets.ReadFile(selectTemperaturesSQLFile)
|
|
if err != nil {
|
|
return nil, fmt.Errorf("Failed to read file %v: %w", selectTemperaturesSQLFile, err)
|
|
}
|
|
|
|
tx, err := postgres.dbo.BeginTx(ctx, &sql.TxOptions{ReadOnly: true})
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
defer tx.Rollback()
|
|
|
|
measuredValues, err := postgres.selectMeasuredValue(tx, string(f))
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
|
|
err = tx.Commit()
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
|
|
for _, measuredValue := range measuredValues {
|
|
measuredValue.ValueType = types.Temperature
|
|
}
|
|
|
|
return measuredValues, nil
|
|
}
|
|
|
|
// UpdateDevices updates a device in the database
|
|
func (postgres *Postgres) UpdateDevices(ctx context.Context, devices ...*types.Device) error {
|
|
f, err := postgresDMLAssets.ReadFile(updateDeviceSQLFile)
|
|
if err != nil {
|
|
return fmt.Errorf("Failed to read file %v: %w", updateDeviceSQLFile, err)
|
|
}
|
|
|
|
tx, err := postgres.dbo.BeginTx(ctx, &sql.TxOptions{ReadOnly: false})
|
|
if err != nil {
|
|
return err
|
|
}
|
|
defer tx.Rollback()
|
|
|
|
stmt, err := tx.Prepare(string(f))
|
|
if err != nil {
|
|
return err
|
|
}
|
|
defer stmt.Close()
|
|
|
|
for _, device := range devices {
|
|
now := time.Now()
|
|
device.UpdateDate = &now
|
|
|
|
_, err := stmt.Exec(
|
|
&device.Name,
|
|
&device.Location,
|
|
&device.CreationDate,
|
|
&device.UpdateDate,
|
|
&device.ID,
|
|
)
|
|
if err != nil {
|
|
return err
|
|
}
|
|
}
|
|
|
|
return tx.Commit()
|
|
}
|
|
|
|
// UpdateSensors updates a sensor in the database
|
|
func (postgres *Postgres) UpdateSensors(ctx context.Context, sensors ...*types.Sensor) error {
|
|
f, err := postgresDMLAssets.ReadFile(updateSensorSQLFile)
|
|
if err != nil {
|
|
return fmt.Errorf("Failed to read file %v: %w", updateSensorSQLFile, err)
|
|
}
|
|
|
|
tx, err := postgres.dbo.BeginTx(ctx, &sql.TxOptions{ReadOnly: false})
|
|
if err != nil {
|
|
return err
|
|
}
|
|
defer tx.Rollback()
|
|
|
|
stmt, err := tx.Prepare(string(f))
|
|
if err != nil {
|
|
return err
|
|
}
|
|
defer stmt.Close()
|
|
|
|
for _, sensor := range sensors {
|
|
now := time.Now()
|
|
sensor.UpdateDate = &now
|
|
_, err := stmt.Exec(
|
|
&sensor.Name,
|
|
&sensor.Location,
|
|
&sensor.WireID,
|
|
&sensor.I2CBus,
|
|
&sensor.I2CAddress,
|
|
&sensor.GPIONumber,
|
|
&sensor.Model,
|
|
&sensor.Enabled,
|
|
&sensor.TickDuration,
|
|
&sensor.DeviceID,
|
|
&sensor.CreationDate,
|
|
&sensor.UpdateDate,
|
|
&sensor.ID,
|
|
)
|
|
if err != nil {
|
|
return err
|
|
}
|
|
}
|
|
|
|
return tx.Commit()
|
|
}
|