fix: reconnecting workers should get new dispatcher id, ticker active…

… bug (#91)

internal/repository/prisma/dbsqlc/tickers.sql

@@ -1,12 +1,12 @@
--- name: ListStaleTickers :many
+-- name: ListNewlyStaleTickers :many
 SELECT
     sqlc.embed(tickers)
 FROM "Ticker" as tickers
 WHERE
     -- last heartbeat older than 15 seconds
     "lastHeartbeatAt" < NOW () - INTERVAL '15 seconds'
-    -- not active
-    AND "isActive" = false;
+    -- active
+    AND "isActive" = true;
 
 -- name: ListActiveTickers :many
 SELECT

internal/repository/prisma/ticker.go

@@ -174,7 +174,7 @@ func (t *tickerRepository) UpdateStaleTickers(onStale func(tickerId string, getV
 
 	defer tx.Rollback(context.Background())
 
-	staleTickers, err := t.queries.ListStaleTickers(context.Background(), tx)
+	staleTickers, err := t.queries.ListNewlyStaleTickers(context.Background(), tx)
 
 	if err != nil {
 		return err

internal/repository/prisma/worker.go

@@ -255,6 +255,12 @@ func (w *workerRepository) UpdateWorker(tenantId, workerId string, opts *reposit
 		optionals = append(optionals, db.Worker.LastHeartbeatAt.Set(*opts.LastHeartbeatAt))
 	}
 
+	if opts.DispatcherId != nil {
+		optionals = append(optionals, db.Worker.Dispatcher.Link(
+			db.Dispatcher.ID.Equals(*opts.DispatcherId),
+		))
+	}
+
 	if len(opts.Actions) > 0 {
 		for _, action := range opts.Actions {
 			txs = append(txs, w.client.Action.UpsertOne(

internal/repository/worker.go

@@ -21,6 +21,9 @@ type CreateWorkerOpts struct {
 }
 
 type UpdateWorkerOpts struct {
+	// The id of the dispatcher
+	DispatcherId *string `validate:"omitempty,uuid"`
+
 	// The status of the worker
 	Status *db.WorkerStatus
 

internal/services/dispatcher/server.go

@@ -118,6 +118,25 @@ func (s *DispatcherImpl) Register(ctx context.Context, request *contracts.Worker
 func (s *DispatcherImpl) Listen(request *contracts.WorkerListenRequest, stream contracts.Dispatcher_ListenServer) error {
 	s.l.Debug().Msgf("Received subscribe request from ID: %s", request.WorkerId)
 
+	worker, err := s.repo.Worker().GetWorkerById(request.WorkerId)
+
+	if err != nil {
+		s.l.Error().Err(err).Msgf("could not get worker %s", request.WorkerId)
+		return err
+	}
+
+	// check the worker's dispatcher against the current dispatcher. if they don't match, then update the worker
+	if worker.DispatcherID != s.dispatcherId {
+		_, err = s.repo.Worker().UpdateWorker(request.TenantId, request.WorkerId, &repository.UpdateWorkerOpts{
+			DispatcherId: &s.dispatcherId,
+		})
+
+		if err != nil {
+			s.l.Error().Err(err).Msgf("could not update worker %s dispatcher", request.WorkerId)
+			return err
+		}
+	}
+
 	fin := make(chan bool)
 
 	s.workers.Store(request.WorkerId, subscribedWorker{stream: stream, finished: fin})