PKGBUILD/pkg/repository/postgres/postgres.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
}

// 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
}

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()
}

// 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()
}

// 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()
}

// 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()
}

// 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()
}

// 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()
}

// 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
}

// 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()
}

// 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()
}