package sqlite3 import ( "context" "database/sql" "embed" "errors" "fmt" "net/url" "os" "path/filepath" "sync" "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 sqlite3DDLAssets embed.FS //go:embed dml/*.sql var sqlite3DMLAssets embed.FS const ( sqlite3DDLAssetPath = "ddl" sqlite3DMLAssetPath = "dml" deleteDeviceByIDSQLFile = sqlite3DMLAssetPath + "/deleteDeviceByID.sql" deleteDeviceByNameSQLFile = sqlite3DMLAssetPath + "/deleteDeviceByName.sql" deleteSensorByIDSQLFile = sqlite3DMLAssetPath + "/deleteSensorByID.sql" deleteSensorByNameSQLFile = sqlite3DMLAssetPath + "/deleteSensorByName.sql" insertDeviceSQLFile = sqlite3DMLAssetPath + "/insertDevice.sql" insertSensorSQLFile = sqlite3DMLAssetPath + "/insertSensor.sql" insertHumiditySQLFile = sqlite3DMLAssetPath + "/insertHumidity.sql" insertPressureSQLFile = sqlite3DMLAssetPath + "/insertPressure.sql" insertTemperatureSQLFile = sqlite3DMLAssetPath + "/insertTemperature.sql" insertOrUpdateDeviceSQLFile = sqlite3DMLAssetPath + "/insertOrUpdateDevice.sql" insertOrUpdateSensorSQLFile = sqlite3DMLAssetPath + "/insertOrUpdateSensor.sql" insertOrUpdateHumiditySQLFile = sqlite3DMLAssetPath + "/insertOrUpdateHumidity.sql" insertOrUpdatePressureSQLFile = sqlite3DMLAssetPath + "/insertOrUpdatePressure.sql" insertOrUpdateTemperatureSQLFile = sqlite3DMLAssetPath + "/insertOrUpdateTemperature.sql" selectDeviceByIDSQLFile = sqlite3DMLAssetPath + "/selectDeviceByID.sql" selectDeviceByNameSQLFile = sqlite3DMLAssetPath + "/selectDeviceByName.sql" selectDevicesSQLFile = sqlite3DMLAssetPath + "/selectDevices.sql" selectSensorByIDSQLFile = sqlite3DMLAssetPath + "/selectSensorByID.sql" selectSensorsSQLFile = sqlite3DMLAssetPath + "/selectSensors.sql" selectSensorsByDeviceIDSQLFile = sqlite3DMLAssetPath + "/selectSensorsByDeviceID.sql" selectSensorsByModelSQLFile = sqlite3DMLAssetPath + "/selectSensorsByModel.sql" selectSensorsByNameSQLFile = sqlite3DMLAssetPath + "/selectSensorsByName.sql" selectHumidityByIDSQLFile = sqlite3DMLAssetPath + "/selectHumidityByID.sql" selectHumiditiesSQLFile = sqlite3DMLAssetPath + "/selectHumidities.sql" selectPressureByIDSQLFile = sqlite3DMLAssetPath + "/selectPressureByID.sql" selectPressuresSQLFile = sqlite3DMLAssetPath + "/selectPressures.sql" selectTemperatureByIDSQLFile = sqlite3DMLAssetPath + "/selectTemperatureByID.sql" selectTemperaturesSQLFile = sqlite3DMLAssetPath + "/selectTemperatures.sql" updateDeviceSQLFile = sqlite3DMLAssetPath + "/updateDevice.sql" updateSensorSQLFile = sqlite3DMLAssetPath + "/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 != "sqlite3" { return fmt.Errorf("Unsupported database scheme %v for database backend sqlite3", o.DatabaseURL.Scheme) } return nil } func New(opts Opts) (*SQLite, error) { if err := opts.Validate(); err != nil { return nil, err } // Create directory if not exist if _, err := os.Stat(filepath.Dir(opts.DatabaseURL.Path)); os.IsNotExist(err) { err := os.MkdirAll(filepath.Dir(opts.DatabaseURL.Path), 0755) if err != nil { return nil, err } } // enable foreign keys values := opts.DatabaseURL.Query() values.Set("_foreign_keys", "on") customRawURL := fmt.Sprintf("file://%v?%v", opts.DatabaseURL.Path, values.Encode()) sqlDBO, err := sql.Open(opts.DatabaseURL.Scheme, customRawURL) if err != nil { return nil, err } return &SQLite{ databaseURL: opts.DatabaseURL, dbo: sqlDBO, flogger: opts.Logger, mutex: new(sync.Mutex), }, nil } // SQLite implementation type SQLite struct { databaseURL *url.URL dbo *sql.DB flogger logger.Logger mutex *sync.Mutex } // AddDevices into the database func (sqlite *SQLite) AddDevices(ctx context.Context, devices ...*types.Device) error { sqlite.mutex.Lock() defer sqlite.mutex.Unlock() f, err := sqlite3DMLAssets.ReadFile(insertDeviceSQLFile) if err != nil { return fmt.Errorf("Failed to read file %v: %w", insertDeviceSQLFile, err) } tx, err := sqlite.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 _, device := range devices { _, 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() } // AddOrUpdateDevices into the database func (sqlite *SQLite) AddOrUpdateDevices(ctx context.Context, devices ...*types.Device) error { sqlite.mutex.Lock() defer sqlite.mutex.Unlock() f, err := sqlite3DMLAssets.ReadFile(insertOrUpdateDeviceSQLFile) if err != nil { return fmt.Errorf("Failed to read file %v: %w", insertOrUpdateDeviceSQLFile, err) } tx, err := sqlite.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 _, device := range devices { _, 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 (sqlite *SQLite) AddMeasuredValues(ctx context.Context, measuredValues ...*types.MeasuredValue) error { sqlite.mutex.Lock() defer sqlite.mutex.Unlock() 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 := sqlite.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 := sqlite3DMLAssets.ReadFile(queryFile) if err != nil { return fmt.Errorf("Failed to read file %v: %w", deleteDeviceByIDSQLFile, 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 { _, 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 (sqlite *SQLite) AddOrUpdateMeasuredValues(ctx context.Context, measuredValues ...*types.MeasuredValue) error { sqlite.mutex.Lock() defer sqlite.mutex.Unlock() 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 := sqlite.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 := sqlite3DMLAssets.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 { _, 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 (sqlite *SQLite) AddSensors(ctx context.Context, sensors ...*types.Sensor) error { sqlite.mutex.Lock() defer sqlite.mutex.Unlock() f, err := sqlite3DMLAssets.ReadFile(insertSensorSQLFile) if err != nil { return fmt.Errorf("Failed to read file %v: %w", insertSensorSQLFile, err) } tx, err := sqlite.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 { _, 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 (sqlite *SQLite) AddOrUpdateSensors(ctx context.Context, sensors ...*types.Sensor) error { sqlite.mutex.Lock() defer sqlite.mutex.Unlock() f, err := sqlite3DMLAssets.ReadFile(insertOrUpdateSensorSQLFile) if err != nil { return fmt.Errorf("Failed to read file %v: %w", insertOrUpdateSensorSQLFile, err) } tx, err := sqlite.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 { _, 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 (sqlite *SQLite) Close() error { sqlite.mutex.Lock() defer sqlite.mutex.Unlock() return sqlite.dbo.Close() } // Migrate creates all required tables if not exist func (sqlite *SQLite) Migrate(ctx context.Context) error { sqlite.mutex.Lock() defer sqlite.mutex.Unlock() sourceDriver, err := iofs.New(sqlite3DDLAssets, sqlite3DDLAssetPath) if err != nil { return err } m, err := migrate.NewWithSourceInstance("iofs", sourceDriver, sqlite.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 (sqlite *SQLite) GetDeviceByID(ctx context.Context, id string) (*types.Device, error) { sqlite.mutex.Lock() defer sqlite.mutex.Unlock() f, err := sqlite3DMLAssets.ReadFile(selectDeviceByIDSQLFile) if err != nil { return nil, fmt.Errorf("Failed to read file %v: %w", selectDeviceByIDSQLFile, err) } tx, err := sqlite.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 := sqlite.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 (sqlite *SQLite) GetDeviceByName(ctx context.Context, name string) (*types.Device, error) { sqlite.mutex.Lock() defer sqlite.mutex.Unlock() f, err := sqlite3DMLAssets.ReadFile(selectDeviceByNameSQLFile) if err != nil { return nil, fmt.Errorf("Failed to read file %v: %w", selectDeviceByNameSQLFile, err) } tx, err := sqlite.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 := sqlite.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 (sqlite *SQLite) GetDevices(ctx context.Context) ([]*types.Device, error) { sqlite.mutex.Lock() defer sqlite.mutex.Unlock() f, err := sqlite3DMLAssets.ReadFile(selectDevicesSQLFile) if err != nil { return nil, fmt.Errorf("Failed to read file %v: %w", selectDevicesSQLFile, err) } tx, err := sqlite.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 := sqlite.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 (sqlite *SQLite) 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) } defer rows.Close() 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 (sqlite *SQLite) GetHumidityByID(ctx context.Context, id string) (*types.MeasuredValue, error) { sqlite.mutex.Lock() defer sqlite.mutex.Unlock() f, err := sqlite3DMLAssets.ReadFile(selectHumidityByIDSQLFile) if err != nil { return nil, fmt.Errorf("Failed to read file %v: %w", selectHumidityByIDSQLFile, err) } tx, err := sqlite.dbo.BeginTx(ctx, &sql.TxOptions{ReadOnly: true}) if err != nil { return nil, err } defer tx.Rollback() measuredValues, err := sqlite.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 (sqlite *SQLite) GetHumidities(ctx context.Context) ([]*types.MeasuredValue, error) { sqlite.mutex.Lock() defer sqlite.mutex.Unlock() f, err := sqlite3DMLAssets.ReadFile(selectHumiditiesSQLFile) if err != nil { return nil, fmt.Errorf("Failed to read file %v: %w", selectHumiditiesSQLFile, err) } tx, err := sqlite.dbo.BeginTx(ctx, &sql.TxOptions{ReadOnly: true}) if err != nil { return nil, err } defer tx.Rollback() measuredValues, err := sqlite.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 (sqlite *SQLite) 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 } defer rows.Close() 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 (sqlite *SQLite) GetPressureByID(ctx context.Context, id string) (*types.MeasuredValue, error) { sqlite.mutex.Lock() defer sqlite.mutex.Unlock() f, err := sqlite3DMLAssets.ReadFile(selectPressureByIDSQLFile) if err != nil { return nil, fmt.Errorf("Failed to read file %v: %w", selectPressureByIDSQLFile, err) } tx, err := sqlite.dbo.BeginTx(ctx, &sql.TxOptions{ReadOnly: true}) if err != nil { return nil, err } defer tx.Rollback() measuredValues, err := sqlite.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 (sqlite *SQLite) GetPressures(ctx context.Context) ([]*types.MeasuredValue, error) { sqlite.mutex.Lock() defer sqlite.mutex.Unlock() f, err := sqlite3DMLAssets.ReadFile(selectPressuresSQLFile) if err != nil { return nil, fmt.Errorf("Failed to read file %v: %w", deleteDeviceByIDSQLFile, err) } tx, err := sqlite.dbo.BeginTx(ctx, &sql.TxOptions{ReadOnly: true}) if err != nil { return nil, err } defer tx.Rollback() measuredValues, err := sqlite.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 (sqlite *SQLite) GetSensorByID(ctx context.Context, id string) (*types.Sensor, error) { sqlite.mutex.Lock() defer sqlite.mutex.Unlock() f, err := sqlite3DMLAssets.ReadFile(selectSensorByIDSQLFile) if err != nil { return nil, fmt.Errorf("Failed to read file %v: %w", selectSensorByIDSQLFile, err) } tx, err := sqlite.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 := sqlite.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 (sqlite *SQLite) GetSensors(ctx context.Context) ([]*types.Sensor, error) { sqlite.mutex.Lock() defer sqlite.mutex.Unlock() f, err := sqlite3DMLAssets.ReadFile(selectSensorsSQLFile) if err != nil { return nil, fmt.Errorf("Failed to read file %v: %w", selectSensorsSQLFile, err) } tx, err := sqlite.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 := sqlite.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 (sqlite *SQLite) GetSensorsByDeviceIDs(ctx context.Context, deviceIDs ...string) ([]*types.Sensor, error) { sqlite.mutex.Lock() defer sqlite.mutex.Unlock() f, err := sqlite3DMLAssets.ReadFile(selectSensorsByDeviceIDSQLFile) if err != nil { return nil, fmt.Errorf("Failed to read file %v: %w", selectSensorsByDeviceIDSQLFile, err) } tx, err := sqlite.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 := sqlite.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 } // GetSensorsByModels from the database func (sqlite *SQLite) GetSensorsByModels(ctx context.Context, sensorModels ...string) ([]*types.Sensor, error) { sqlite.mutex.Lock() defer sqlite.mutex.Unlock() f, err := sqlite3DMLAssets.ReadFile(selectSensorsByModelSQLFile) if err != nil { return nil, fmt.Errorf("Failed to read file %v: %w", selectSensorsByModelSQLFile, err) } tx, err := sqlite.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 sensorModels { sensors, err := sqlite.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 } // GetSensorsByModels from the database func (sqlite *SQLite) GetSensorsByNames(ctx context.Context, sensorNames ...string) ([]*types.Sensor, error) { sqlite.mutex.Lock() defer sqlite.mutex.Unlock() f, err := sqlite3DMLAssets.ReadFile(selectSensorsByNameSQLFile) if err != nil { return nil, fmt.Errorf("Failed to read file %v: %w", selectSensorsByNameSQLFile, err) } tx, err := sqlite.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 := sqlite.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 (sqlite *SQLite) 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) } defer rows.Close() 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 (sqlite *SQLite) GetTemperatureByID(ctx context.Context, id string) (*types.MeasuredValue, error) { sqlite.mutex.Lock() defer sqlite.mutex.Unlock() f, err := sqlite3DMLAssets.ReadFile(selectTemperatureByIDSQLFile) if err != nil { return nil, fmt.Errorf("Failed to read file %v: %w", selectTemperatureByIDSQLFile, err) } tx, err := sqlite.dbo.BeginTx(ctx, &sql.TxOptions{ReadOnly: true}) if err != nil { return nil, err } defer tx.Rollback() measuredValues, err := sqlite.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 (sqlite *SQLite) GetTemperatures(ctx context.Context) ([]*types.MeasuredValue, error) { sqlite.mutex.Lock() defer sqlite.mutex.Unlock() f, err := sqlite3DMLAssets.ReadFile(selectTemperaturesSQLFile) if err != nil { return nil, fmt.Errorf("Failed to read file %v: %w", selectTemperaturesSQLFile, err) } tx, err := sqlite.dbo.BeginTx(ctx, &sql.TxOptions{ReadOnly: true}) if err != nil { return nil, err } defer tx.Rollback() measuredValues, err := sqlite.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 (sqlite *SQLite) RemoveDevicesByIDs(ctx context.Context, deviceIDs ...string) error { sqlite.mutex.Lock() defer sqlite.mutex.Unlock() f, err := sqlite3DMLAssets.ReadFile(deleteDeviceByIDSQLFile) if err != nil { return fmt.Errorf("Failed to read file %v: %w", deleteDeviceByIDSQLFile, err) } tx, err := sqlite.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 (sqlite *SQLite) RemoveDevicesByNames(ctx context.Context, names ...string) error { sqlite.mutex.Lock() defer sqlite.mutex.Unlock() f, err := sqlite3DMLAssets.ReadFile(deleteDeviceByNameSQLFile) if err != nil { return fmt.Errorf("Failed to read file %v: %w", deleteDeviceByNameSQLFile, err) } tx, err := sqlite.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 (sqlite *SQLite) RemoveSensorsByIDs(ctx context.Context, sensorIDs ...string) error { sqlite.mutex.Lock() defer sqlite.mutex.Unlock() f, err := sqlite3DMLAssets.ReadFile(deleteSensorByIDSQLFile) if err != nil { return fmt.Errorf("Failed to read file %v: %w", deleteSensorByIDSQLFile, err) } tx, err := sqlite.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 (sqlite *SQLite) RemoveSensorsByNames(ctx context.Context, names ...string) error { sqlite.mutex.Lock() defer sqlite.mutex.Unlock() f, err := sqlite3DMLAssets.ReadFile(deleteSensorByNameSQLFile) if err != nil { return fmt.Errorf("Failed to read file %v: %w", deleteSensorByNameSQLFile, err) } tx, err := sqlite.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 names { _, err = stmt.Exec(sensorID) if err != nil { return err } } return tx.Commit() } // UpdateDevices updates a device in the database func (sqlite *SQLite) UpdateDevices(ctx context.Context, devices ...*types.Device) error { sqlite.mutex.Lock() defer sqlite.mutex.Unlock() f, err := sqlite3DMLAssets.ReadFile(updateDeviceSQLFile) if err != nil { return fmt.Errorf("Failed to read file %v: %w", updateDeviceSQLFile, err) } tx, err := sqlite.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 { _, 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 (sqlite *SQLite) UpdateSensors(ctx context.Context, sensors ...*types.Sensor) error { sqlite.mutex.Lock() defer sqlite.mutex.Unlock() f, err := sqlite3DMLAssets.ReadFile(updateSensorSQLFile) if err != nil { return fmt.Errorf("Failed to read file %v: %w", updateSensorSQLFile, err) } tx, err := sqlite.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 { _, 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() }