Feature/set watering plan status on tree crud

internal/service/domain/tree/handle_new_sensor_data.go

@@ -0,0 +1,39 @@
+package tree
+
+import (
+	"context"
+	"log/slog"
+
+	"github.com/green-ecolution/green-ecolution-backend/internal/entities"
+	"github.com/green-ecolution/green-ecolution-backend/internal/logger"
+	"github.com/green-ecolution/green-ecolution-backend/internal/service/domain/utils"
+	"github.com/green-ecolution/green-ecolution-backend/internal/storage/postgres/tree"
+)
+
+func (s *TreeService) HandleNewSensorData(ctx context.Context, event *entities.EventNewSensorData) error {
+	log := logger.GetLogger(ctx)
+	log.Debug("handle event", "event", event.Type(), "service", "TreeService")
+	t, err := s.treeRepo.GetBySensorID(ctx, event.New.SensorID)
+	if err != nil {
+		log.Error("failed to get tree by sensor id", "sensor_id", event.New.SensorID, "err", err)
+		return nil
+	}
+
+	status := utils.CalculateWateringStatus(ctx, t.PlantingYear, event.New.Data.Watermarks)
+
+	if status == t.WateringStatus {
+		log.Debug("sensor status has not changed", "sensor_status", status)
+		return nil
+	}
+
+	newTree, err := s.treeRepo.Update(ctx, t.ID, tree.WithWateringStatus(status))
+	if err != nil {
+		log.Error("failed to update tree with new watering status", "tree_id", t.ID, "watering_status", status, "err", err)
+		return err
+	}
+
+	slog.Info("updating tree watering status", "prev_status", t.WateringStatus, "new_status", status)
+
+	s.publishUpdateTreeEvent(ctx, t, newTree)
+	return nil
+}

internal/service/domain/tree/handle_new_sensor_data_test.go

@@ -0,0 +1,162 @@
+package tree
+
+import (
+	"context"
+	"testing"
+	"time"
+
+	"github.com/green-ecolution/green-ecolution-backend/internal/entities"
+	"github.com/green-ecolution/green-ecolution-backend/internal/storage"
+	storageMock "github.com/green-ecolution/green-ecolution-backend/internal/storage/_mock"
+	"github.com/green-ecolution/green-ecolution-backend/internal/worker"
+	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/mock"
+)
+
+func TestTreeService_HandleNewSensorData(t *testing.T) {
+	t.Run("should update watering status on new sensor data event", func(t *testing.T) {
+		treeRepo := storageMock.NewMockTreeRepository(t)
+		sensorRepo := storageMock.NewMockSensorRepository(t)
+		imageRepo := storageMock.NewMockImageRepository(t)
+		clusterRepo := storageMock.NewMockTreeClusterRepository(t)
+		eventManager := worker.NewEventManager(entities.EventTypeUpdateTree)
+		svc := NewTreeService(treeRepo, sensorRepo, imageRepo, clusterRepo, eventManager)
+
+		_, ch, _ := eventManager.Subscribe(entities.EventTypeUpdateTree)
+		ctx, cancel := context.WithCancel(context.Background())
+		defer cancel()
+		go eventManager.Run(ctx)
+
+		sensorDataEvent := entities.SensorData{
+			SensorID: "sensor-1",
+			Data: &entities.MqttPayload{
+				Watermarks: []entities.Watermark{
+					{Centibar: 30, Depth: 30},
+					{Centibar: 40, Depth: 60},
+					{Centibar: 50, Depth: 90},
+				},
+			},
+		}
+
+		treeNew := entities.Tree{
+			ID:             1,
+			PlantingYear:   int32(time.Now().Year() - 2),
+			WateringStatus: entities.WateringStatusGood,
+		}
+
+		tree := entities.Tree{
+			ID:             1,
+			PlantingYear:   int32(time.Now().Year() - 2),
+			WateringStatus: entities.WateringStatusUnknown,
+		}
+
+		event := entities.NewEventSensorData(&sensorDataEvent)
+
+		treeRepo.EXPECT().GetBySensorID(mock.Anything, "sensor-1").Return(&tree, nil)
+		treeRepo.EXPECT().Update(mock.Anything, mock.Anything, mock.Anything).Return(&treeNew, nil)
+
+		err := svc.HandleNewSensorData(context.Background(), &event)
+
+		assert.NoError(t, err)
+		select {
+		case receivedEvent := <-ch:
+			e, ok := receivedEvent.(entities.EventUpdateTree)
+			assert.True(t, ok)
+			assert.Equal(t, *e.Prev, tree)
+			assert.Equal(t, *e.New, treeNew)
+		case <-time.After(100 * time.Millisecond):
+			t.Fatal("event was not received")
+		}
+	})
+
+	t.Run("should not update and not send event if the sensor has no linked tree", func(t *testing.T) {
+		treeRepo := storageMock.NewMockTreeRepository(t)
+		sensorRepo := storageMock.NewMockSensorRepository(t)
+		imageRepo := storageMock.NewMockImageRepository(t)
+		clusterRepo := storageMock.NewMockTreeClusterRepository(t)
+		eventManager := worker.NewEventManager(entities.EventTypeUpdateTree)
+		svc := NewTreeService(treeRepo, sensorRepo, imageRepo, clusterRepo, eventManager)
+
+		// event
+		_, ch, _ := eventManager.Subscribe(entities.EventTypeUpdateTree)
+		ctx, cancel := context.WithCancel(context.Background())
+		defer cancel()
+		go eventManager.Run(ctx)
+
+		sensorDataEvent := entities.SensorData{
+			SensorID: "sensor-1",
+			Data: &entities.MqttPayload{
+				Watermarks: []entities.Watermark{
+					{Centibar: 61, Depth: 30},
+					{Centibar: 24, Depth: 60},
+					{Centibar: 24, Depth: 90},
+				},
+			},
+		}
+
+		event := entities.NewEventSensorData(&sensorDataEvent)
+
+		treeRepo.EXPECT().GetBySensorID(mock.Anything, "sensor-1").Return(nil, storage.ErrTreeNotFound)
+
+		// when
+		err := svc.HandleNewSensorData(context.Background(), &event)
+
+		// then
+		assert.NoError(t, err)
+		select {
+		case <-ch:
+			t.Fatal("event was received. It should not have been sent")
+		case <-time.After(100 * time.Millisecond):
+			assert.True(t, true)
+		}
+	})
+
+	t.Run("should not update and not send event if tree could not be updated", func(t *testing.T) {
+		treeRepo := storageMock.NewMockTreeRepository(t)
+		sensorRepo := storageMock.NewMockSensorRepository(t)
+		imageRepo := storageMock.NewMockImageRepository(t)
+		clusterRepo := storageMock.NewMockTreeClusterRepository(t)
+		eventManager := worker.NewEventManager(entities.EventTypeUpdateTree)
+		svc := NewTreeService(treeRepo, sensorRepo, imageRepo, clusterRepo, eventManager)
+
+		// event
+		_, ch, _ := eventManager.Subscribe(entities.EventTypeUpdateTree)
+		ctx, cancel := context.WithCancel(context.Background())
+		defer cancel()
+		go eventManager.Run(ctx)
+
+		sensorDataEvent := entities.SensorData{
+			SensorID: "sensor-1",
+			Data: &entities.MqttPayload{
+				Watermarks: []entities.Watermark{
+					{Centibar: 30, Depth: 30},
+					{Centibar: 40, Depth: 60},
+					{Centibar: 50, Depth: 90},
+				},
+			},
+		}
+
+		tree := entities.Tree{
+			ID:             1,
+			PlantingYear:   int32(time.Now().Year() - 2),
+			WateringStatus: entities.WateringStatusUnknown,
+		}
+
+		event := entities.NewEventSensorData(&sensorDataEvent)
+
+		treeRepo.EXPECT().GetBySensorID(mock.Anything, "sensor-1").Return(&tree, nil)
+		treeRepo.EXPECT().Update(mock.Anything, mock.Anything, mock.Anything).Return(nil, storage.ErrTreeNotFound)
+
+		// when
+		err := svc.HandleNewSensorData(context.Background(), &event)
+
+		// then
+		assert.ErrorIs(t, err, storage.ErrTreeNotFound)
+		select {
+		case <-ch:
+			t.Fatal("event was received. It should not have been sent")
+		case <-time.After(100 * time.Millisecond):
+			assert.True(t, true)
+		}
+	})
+}

internal/service/domain/tree/import_tree_test.go

@@ -65,6 +65,7 @@ func TestTreeService_ImportTree(t *testing.T) {
 			mock.Anything,
 			mock.Anything,
 			mock.Anything,
+			mock.Anything,
 			mock.Anything).Return(updatedTree, nil)
 
 		// When

internal/service/domain/tree/tree.go

@@ -42,32 +42,6 @@ func NewTreeService(
 	}
 }
 
-func (s *TreeService) HandleNewSensorData(ctx context.Context, event *entities.EventNewSensorData) error {
-	log := logger.GetLogger(ctx)
-	log.Debug("handle event", "event", event.Type(), "service", "TreeService")
-	t, err := s.treeRepo.GetBySensorID(ctx, event.New.SensorID)
-	if err != nil {
-		log.Error("failed to get tree by sensor id", "sensor_id", event.New.SensorID, "err", err)
-		return nil
-	}
-
-	status := utils.CalculateWateringStatus(ctx, t.PlantingYear, event.New.Data.Watermarks)
-
-	if status == t.WateringStatus {
-		return nil
-	}
-
-	newTree, err := s.treeRepo.Update(ctx, t.ID, tree.WithWateringStatus(status))
-	if err != nil {
-		log.Error("failed to update tree with new watering status", "tree_id", t.ID, "watering_status", status, "err", err)
-	}
-
-	slog.Info("updating tree watering status", "prev_status", t.WateringStatus, "new_status", status)
-
-	s.publishUpdateTreeEvent(ctx, t, newTree)
-	return nil
-}
-
 func (s *TreeService) GetAll(ctx context.Context) ([]*entities.Tree, error) {
 	log := logger.GetLogger(ctx)
 	trees, err := s.treeRepo.GetAll(ctx)
@@ -146,12 +120,17 @@ func (s *TreeService) Create(ctx context.Context, treeCreate *entities.TreeCreat
 	}
 
 	if treeCreate.SensorID != nil {
-		sensorID, err := s.sensorRepo.GetByID(ctx, *treeCreate.SensorID)
+		sensor, err := s.sensorRepo.GetByID(ctx, *treeCreate.SensorID)
 		if err != nil {
 			log.Debug("failed to fetch sensor by id specified in the tree create request", "sensor_id", treeCreate.SensorID)
 			return nil, service.MapError(ctx, err, service.ErrorLogEntityNotFound)
 		}
-		fn = append(fn, tree.WithSensor(sensorID))
+		fn = append(fn, tree.WithSensor(sensor))
+
+		if sensor.LatestData != nil && sensor.LatestData.Data != nil && len(sensor.LatestData.Data.Watermarks) > 0 {
+			status := utils.CalculateWateringStatus(ctx, treeCreate.PlantingYear, sensor.LatestData.Data.Watermarks)
+			fn = append(fn, tree.WithWateringStatus(status))
+		}
 	}
 
 	fn = append(fn,
@@ -202,6 +181,7 @@ func (s *TreeService) Update(ctx context.Context, id int32, tu *entities.TreeUpd
 		return nil, service.MapError(ctx, err, service.ErrorLogEntityNotFound)
 	}
 
+	// TODO: Why is this still commented out?
 	// Check if the tree is readonly (imported from csv)
 	// if currentTree.Readonly {
 	// 	return nil, handleError(fmt.Errorf("tree with ID %d is readonly and cannot be updated", id))
@@ -228,8 +208,15 @@ func (s *TreeService) Update(ctx context.Context, id int32, tu *entities.TreeUpd
 			return nil, service.MapError(ctx, fmt.Errorf("failed to find Sensor with ID %v: %w", *tu.SensorID, err), service.ErrorLogEntityNotFound)
 		}
 		fn = append(fn, tree.WithSensor(sensor))
+
+		if sensor.LatestData != nil && sensor.LatestData.Data != nil && len(sensor.LatestData.Data.Watermarks) > 0 {
+			status := utils.CalculateWateringStatus(ctx, tu.PlantingYear, sensor.LatestData.Data.Watermarks)
+			fn = append(fn, tree.WithWateringStatus(status))
+		}
 	} else {
-		fn = append(fn, tree.WithSensor(nil))
+		fn = append(fn,
+			tree.WithSensor(nil),
+			tree.WithWateringStatus(entities.WateringStatusUnknown))
 	}
 
 	fn = append(fn, tree.WithPlantingYear(tu.PlantingYear),

internal/service/domain/tree/tree_test.go

@@ -251,6 +251,7 @@ func TestTreeService_Create(t *testing.T) {
 			mock.Anything,
 			mock.Anything,
 			mock.Anything,
+			mock.Anything,
 			mock.Anything).Return(expectedTree, nil)
 
 		// when
@@ -359,6 +360,7 @@ func TestTreeService_Create(t *testing.T) {
 			mock.Anything,
 			mock.Anything,
 			mock.Anything,
+			mock.Anything,
 			mock.Anything).Return(nil, expectedError)
 
 		// when
@@ -509,6 +511,8 @@ func TestTreeService_Update(t *testing.T) {
 			mock.Anything,
 			mock.Anything,
 			mock.Anything,
+			mock.Anything,
+			mock.Anything,
 			mock.Anything).Return(updatedTree, nil)
 
 		// when
@@ -655,6 +659,8 @@ func TestTreeService_Update(t *testing.T) {
 			mock.Anything,
 			mock.Anything,
 			mock.Anything,
+			mock.Anything,
+			mock.Anything,
 			mock.Anything).Return(nil, expectedError)
 
 		// when
@@ -743,6 +749,7 @@ func TestTreeService_EventSystem(t *testing.T) {
 			mock.Anything,
 			mock.Anything,
 			mock.Anything,
+			mock.Anything,
 			mock.Anything).Return(&expectedTree, nil)
 
 		svc := tree.NewTreeService(treeRepo, sensorRepo, imageRepo, treeClusterRepo, eventManager)