fix (sinks): set sink status when deployment status change

sinks/api/http/logging.go

@@ -6,11 +6,12 @@ package http
 
 import (
 	"context"
+	"time"
+
 	"github.com/orb-community/orb/sinks"
 	"github.com/orb-community/orb/sinks/authentication_type"
 	"github.com/orb-community/orb/sinks/backend"
 	"go.uber.org/zap"
-	"time"
 )
 
 var _ sinks.SinkService = (*loggingMiddleware)(nil)

sinks/api/http/metrics.go

@@ -6,14 +6,15 @@ package http
 
 import (
 	"context"
+	"time"
+
 	"github.com/go-kit/kit/metrics"
 	"github.com/mainflux/mainflux"
 	"github.com/orb-community/orb/pkg/errors"
 	"github.com/orb-community/orb/sinks"
 	"github.com/orb-community/orb/sinks/authentication_type"
 	"github.com/orb-community/orb/sinks/backend"
 	"go.uber.org/zap"
-	"time"
 )
 
 var _ sinks.SinkService = (*metricsMiddleware)(nil)

sinks/redis/consumer/sink_status_listener.go

@@ -89,7 +89,7 @@ func (s *sinkStatusListener) ReceiveMessage(ctx context.Context, message redis.X
 			gotSink.Error = event.Msg
 		}
 		gotSink.State = newState
-		_, err = s.sinkService.UpdateSinkInternal(ctx, gotSink)
+		err = s.sinkService.ChangeSinkStateInternal(ctx, gotSink.ID, gotSink.Error, gotSink.MFOwnerID, gotSink.State)
 		if err != nil {
 			logger.Error("failed to update sink", zap.String("owner_id", event.OwnerID),
 				zap.String("sink_id", event.SinkID), zap.Error(err))

sinks/redis/producer/streams.go

@@ -10,6 +10,7 @@ package producer
 
 import (
 	"context"
+
 	"github.com/orb-community/orb/sinks/authentication_type"
 
 	"github.com/go-redis/redis/v8"

sinks/sinks.go

@@ -7,13 +7,14 @@ package sinks
 import (
 	"context"
 	"database/sql/driver"
+	"time"
+
 	"github.com/orb-community/orb/pkg/errors"
 	"github.com/orb-community/orb/pkg/types"
 	"github.com/orb-community/orb/sinks/authentication_type"
 	"github.com/orb-community/orb/sinks/authentication_type/basicauth"
 	"github.com/orb-community/orb/sinks/backend"
 	"go.uber.org/zap"
-	"time"
 )
 
 var (