feat: return action rejections from arrs (#103)

* refactor: push status

* feat: return push status for arr actions

internal/action/lidarr.go

@@ -10,7 +10,7 @@ import (
 	"github.com/rs/zerolog/log"
 )
 
-func (s *service) lidarr(release domain.Release, action domain.Action) error {
+func (s *service) lidarr(release domain.Release, action domain.Action) ([]string, error) {
 	log.Trace().Msg("action LIDARR")
 
 	// TODO validate data
@@ -18,13 +18,13 @@ func (s *service) lidarr(release domain.Release, action domain.Action) error {
 	// get client for action
 	client, err := s.clientSvc.FindByID(context.TODO(), action.ClientID)
 	if err != nil {
-		log.Error().Err(err).Msgf("error finding client: %v", action.ClientID)
-		return err
+		log.Error().Err(err).Msgf("lidarr: error finding client: %v", action.ClientID)
+		return nil, err
 	}
 
 	// return early if no client found
 	if client == nil {
-		return err
+		return nil, err
 	}
 
 	// initial config
@@ -52,23 +52,19 @@ func (s *service) lidarr(release domain.Release, action domain.Action) error {
 		PublishDate:      time.Now().Format(time.RFC3339),
 	}
 
-	success, rejections, err := arr.Push(r)
+	rejections, err := arr.Push(r)
 	if err != nil {
 		log.Error().Stack().Err(err).Msgf("lidarr: failed to push release: %v", r)
-		return err
+		return nil, err
 	}
 
-	if !success {
+	if rejections != nil {
 		log.Debug().Msgf("lidarr: release push rejected: %v, indexer %v to %v reasons: '%v'", r.Title, r.Indexer, client.Host, rejections)
 
-		// save pushed release
-		s.bus.Publish("release:update-push-status-rejected", release.ID, rejections)
-		return nil
+		return rejections, nil
 	}
 
 	log.Debug().Msgf("lidarr: successfully pushed release: %v, indexer %v to %v", r.Title, r.Indexer, client.Host)
 
-	s.bus.Publish("release:update-push-status", release.ID, domain.ReleasePushStatusApproved)
-
-	return nil
+	return nil, nil
 }

internal/action/qbittorrent.go

@@ -10,8 +10,8 @@ import (
 	"github.com/rs/zerolog/log"
 )
 
-const REANNOUNCE_MAX_ATTEMPTS = 30
-const REANNOUNCE_INTERVAL = 7000
+const ReannounceMaxAttempts = 30
+const ReannounceInterval = 7000
 
 func (s *service) qbittorrent(qbt *qbittorrent.Client, action domain.Action, hash string, torrentFile string) error {
 	log.Debug().Msgf("action qBittorrent: %v", action.Name)
@@ -134,7 +134,7 @@ func checkTrackerStatus(qb qbittorrent.Client, hash string) error {
 	// initial sleep to give tracker a head start
 	time.Sleep(2 * time.Second)
 
-	for attempts < REANNOUNCE_MAX_ATTEMPTS {
+	for attempts < ReannounceMaxAttempts {
 		log.Debug().Msgf("qBittorrent - run re-announce %v attempt: %v", hash, attempts)
 
 		trackers, err := qb.GetTorrentTrackers(hash)
@@ -157,7 +157,7 @@ func checkTrackerStatus(qb qbittorrent.Client, hash string) error {
 			attempts++
 
 			// add delay for next run
-			time.Sleep(REANNOUNCE_INTERVAL * time.Millisecond)
+			time.Sleep(ReannounceInterval * time.Millisecond)
 
 			continue
 		} else {

internal/action/radarr.go

@@ -10,7 +10,7 @@ import (
 	"github.com/rs/zerolog/log"
 )
 
-func (s *service) radarr(release domain.Release, action domain.Action) error {
+func (s *service) radarr(release domain.Release, action domain.Action) ([]string, error) {
 	log.Trace().Msg("action RADARR")
 
 	// TODO validate data
@@ -18,13 +18,13 @@ func (s *service) radarr(release domain.Release, action domain.Action) error {
 	// get client for action
 	client, err := s.clientSvc.FindByID(context.TODO(), action.ClientID)
 	if err != nil {
-		log.Error().Err(err).Msgf("error finding client: %v", action.ClientID)
-		return err
+		log.Error().Err(err).Msgf("radarr: error finding client: %v", action.ClientID)
+		return nil, err
 	}
 
 	// return early if no client found
 	if client == nil {
-		return err
+		return nil, err
 	}
 
 	// initial config
@@ -52,23 +52,19 @@ func (s *service) radarr(release domain.Release, action domain.Action) error {
 		PublishDate:      time.Now().Format(time.RFC3339),
 	}
 
-	success, rejections, err := arr.Push(r)
+	rejections, err := arr.Push(r)
 	if err != nil {
 		log.Error().Stack().Err(err).Msgf("radarr: failed to push release: %v", r)
-		return err
+		return nil, err
 	}
 
-	if !success {
+	if rejections != nil {
 		log.Debug().Msgf("radarr: release push rejected: %v, indexer %v to %v reasons: '%v'", r.Title, r.Indexer, client.Host, rejections)
 
-		// save pushed release
-		s.bus.Publish("release:update-push-status-rejected", release.ID, rejections)
-		return nil
+		return rejections, nil
 	}
 
 	log.Debug().Msgf("radarr: successfully pushed release: %v, indexer %v to %v", r.Title, r.Indexer, client.Host)
 
-	s.bus.Publish("release:update-push-status", release.ID, domain.ReleasePushStatusApproved)
-
-	return nil
+	return nil, nil
 }

internal/action/run.go

@@ -13,169 +13,30 @@ import (
 
 func (s *service) RunActions(actions []domain.Action, release domain.Release) error {
 
-	var err error
-	var tmpFile string
-	var hash string
-
 	for _, action := range actions {
+		// only run active actions
 		if !action.Enabled {
-			// only run active actions
 			continue
 		}
 
 		log.Debug().Msgf("process action: %v for '%v'", action.Name, release.TorrentName)
 
-		actionStatus := domain.ReleaseActionStatus{
-			ReleaseID:  release.ID,
-			Status:     domain.ReleasePushStatusPending,
-			Action:     action.Name,
-			Type:       action.Type,
-			Rejections: []string{},
-			Timestamp:  time.Now(),
-		}
-
-		s.bus.Publish("release:store-action-status", &actionStatus)
-
-		switch action.Type {
-		case domain.ActionTypeTest:
-			s.test(action.Name)
-
-		case domain.ActionTypeExec:
-			if release.TorrentTmpFile == "" {
-				t, err := release.DownloadTorrentFile(nil)
-				if err != nil {
-					log.Error().Stack().Err(err)
-					return err
-				}
-
-				tmpFile = t.TmpFileName
-			}
-
-			go func(release domain.Release, action domain.Action, tmpFile string) {
-				s.execCmd(release, action, tmpFile)
-			}(release, action, tmpFile)
-
-		case domain.ActionTypeWatchFolder:
-			if release.TorrentTmpFile == "" {
-				t, err := release.DownloadTorrentFile(nil)
-				if err != nil {
-					log.Error().Stack().Err(err)
-					return err
-				}
-
-				tmpFile = t.TmpFileName
-			}
-			s.watchFolder(action.WatchFolder, tmpFile)
-
-		case domain.ActionTypeDelugeV1, domain.ActionTypeDelugeV2:
-			canDownload, err := s.delugeCheckRulesCanDownload(action)
+		go func(release domain.Release, action domain.Action) {
+			err := s.runAction(action, release)
 			if err != nil {
-				log.Error().Stack().Err(err).Msgf("error checking client rules: %v", action.Name)
-				continue
-			}
-			if !canDownload {
+				log.Err(err).Stack().Msgf("process action failed: %v for '%v'", action.Name, release.TorrentName)
+
 				s.bus.Publish("release:store-action-status", &domain.ReleaseActionStatus{
-					ID:         actionStatus.ID,
 					ReleaseID:  release.ID,
-					Status:     domain.ReleasePushStatusRejected,
+					Status:     domain.ReleasePushStatusErr,
 					Action:     action.Name,
 					Type:       action.Type,
-					Rejections: []string{"deluge busy"},
+					Rejections: []string{err.Error()},
+					Timestamp:  time.Now(),
 				})
-				continue
+				return
 			}
-			if release.TorrentTmpFile == "" {
-				t, err := release.DownloadTorrentFile(nil)
-				if err != nil {
-					log.Error().Stack().Err(err)
-					return err
-				}
-
-				tmpFile = t.TmpFileName
-			}
-
-			go func(action domain.Action, tmpFile string) {
-				err = s.deluge(action, tmpFile)
-				if err != nil {
-					log.Error().Stack().Err(err).Msg("error sending torrent to Deluge")
-				}
-			}(action, tmpFile)
-
-		case domain.ActionTypeQbittorrent:
-			canDownload, client, err := s.qbittorrentCheckRulesCanDownload(action)
-			if err != nil {
-				log.Error().Stack().Err(err).Msgf("error checking client rules: %v", action.Name)
-				continue
-			}
-			if !canDownload {
-				s.bus.Publish("release:store-action-status", &domain.ReleaseActionStatus{
-					ID:         actionStatus.ID,
-					ReleaseID:  release.ID,
-					Status:     domain.ReleasePushStatusRejected,
-					Action:     action.Name,
-					Type:       action.Type,
-					Rejections: []string{"qbittorrent busy"},
-				})
-				continue
-			}
-
-			if release.TorrentTmpFile == "" {
-				t, err := release.DownloadTorrentFile(nil)
-				if err != nil {
-					log.Error().Stack().Err(err)
-					return err
-				}
-
-				tmpFile = t.TmpFileName
-				hash = t.MetaInfo.HashInfoBytes().String()
-			}
-
-			go func(action domain.Action, hash string, tmpFile string) {
-				err = s.qbittorrent(client, action, hash, tmpFile)
-				if err != nil {
-					log.Error().Stack().Err(err).Msg("error sending torrent to qBittorrent")
-				}
-			}(action, hash, tmpFile)
-
-		case domain.ActionTypeRadarr:
-			go func(release domain.Release, action domain.Action) {
-				err = s.radarr(release, action)
-				if err != nil {
-					log.Error().Stack().Err(err).Msg("error sending torrent to radarr")
-					//continue
-				}
-			}(release, action)
-
-		case domain.ActionTypeSonarr:
-			go func(release domain.Release, action domain.Action) {
-				err = s.sonarr(release, action)
-				if err != nil {
-					log.Error().Stack().Err(err).Msg("error sending torrent to sonarr")
-					//continue
-				}
-			}(release, action)
-
-		case domain.ActionTypeLidarr:
-			go func(release domain.Release, action domain.Action) {
-				err = s.lidarr(release, action)
-				if err != nil {
-					log.Error().Stack().Err(err).Msg("error sending torrent to lidarr")
-					//continue
-				}
-			}(release, action)
-
-		default:
-			log.Warn().Msgf("unsupported action: %v type: %v", action.Name, action.Type)
-		}
-
-		s.bus.Publish("release:store-action-status", &domain.ReleaseActionStatus{
-			ID:         actionStatus.ID,
-			ReleaseID:  release.ID,
-			Status:     domain.ReleasePushStatusApproved,
-			Action:     action.Name,
-			Type:       action.Type,
-			Rejections: []string{},
-		})
+		}(release, action)
 	}
 
 	// safe to delete tmp file
@@ -183,6 +44,144 @@ func (s *service) RunActions(actions []domain.Action, release domain.Release) er
 	return nil
 }
 
+func (s *service) runAction(action domain.Action, release domain.Release) error {
+
+	var err error
+	var tmpFile string
+	var hash string
+	var rejections []string
+
+	switch action.Type {
+	case domain.ActionTypeTest:
+		s.test(action.Name)
+
+	case domain.ActionTypeExec:
+		if release.TorrentTmpFile == "" {
+			t, err := release.DownloadTorrentFile(nil)
+			if err != nil {
+				log.Error().Stack().Err(err)
+				return err
+			}
+
+			tmpFile = t.TmpFileName
+		}
+
+		s.execCmd(release, action, tmpFile)
+
+	case domain.ActionTypeWatchFolder:
+		if release.TorrentTmpFile == "" {
+			t, err := release.DownloadTorrentFile(nil)
+			if err != nil {
+				log.Error().Stack().Err(err)
+				return err
+			}
+
+			tmpFile = t.TmpFileName
+		}
+		s.watchFolder(action.WatchFolder, tmpFile)
+
+	case domain.ActionTypeDelugeV1, domain.ActionTypeDelugeV2:
+		canDownload, err := s.delugeCheckRulesCanDownload(action)
+		if err != nil {
+			log.Error().Stack().Err(err).Msgf("error checking client rules: %v", action.Name)
+			return err
+		}
+		if !canDownload {
+			rejections = []string{"deluge busy"}
+		}
+
+		if release.TorrentTmpFile == "" {
+			t, err := release.DownloadTorrentFile(nil)
+			if err != nil {
+				log.Error().Stack().Err(err)
+				return err
+			}
+
+			tmpFile = t.TmpFileName
+		}
+		err = s.deluge(action, tmpFile)
+		if err != nil {
+			log.Error().Stack().Err(err).Msg("error sending torrent to Deluge")
+			return err
+		}
+
+	case domain.ActionTypeQbittorrent:
+		canDownload, client, err := s.qbittorrentCheckRulesCanDownload(action)
+		if err != nil {
+			log.Error().Stack().Err(err).Msgf("error checking client rules: %v", action.Name)
+			return err
+		}
+		if !canDownload {
+			rejections = []string{"qBittorrent busy"}
+		}
+
+		if release.TorrentTmpFile == "" {
+			t, err := release.DownloadTorrentFile(nil)
+			if err != nil {
+				log.Error().Stack().Err(err)
+				return err
+			}
+
+			tmpFile = t.TmpFileName
+			hash = t.MetaInfo.HashInfoBytes().String()
+		}
+		err = s.qbittorrent(client, action, hash, tmpFile)
+		if err != nil {
+			log.Error().Stack().Err(err).Msg("error sending torrent to qBittorrent")
+			return err
+		}
+
+	case domain.ActionTypeRadarr:
+		rejections, err = s.radarr(release, action)
+		if err != nil {
+			log.Error().Stack().Err(err).Msg("error sending torrent to radarr")
+			return err
+		}
+
+	case domain.ActionTypeSonarr:
+		rejections, err = s.sonarr(release, action)
+		if err != nil {
+			log.Error().Stack().Err(err).Msg("error sending torrent to sonarr")
+			return err
+		}
+
+	case domain.ActionTypeLidarr:
+		rejections, err = s.lidarr(release, action)
+		if err != nil {
+			log.Error().Stack().Err(err).Msg("error sending torrent to lidarr")
+			return err
+		}
+
+	default:
+		log.Warn().Msgf("unsupported action: %v type: %v", action.Name, action.Type)
+		return nil
+	}
+
+	if rejections != nil {
+		s.bus.Publish("release:push-rejected", &domain.ReleaseActionStatus{
+			ReleaseID:  release.ID,
+			Status:     domain.ReleasePushStatusRejected,
+			Action:     action.Name,
+			Type:       action.Type,
+			Rejections: rejections,
+			Timestamp:  time.Now(),
+		})
+
+		return nil
+	}
+
+	s.bus.Publish("release:push-approved", &domain.ReleaseActionStatus{
+		ReleaseID:  release.ID,
+		Status:     domain.ReleasePushStatusApproved,
+		Action:     action.Name,
+		Type:       action.Type,
+		Rejections: []string{},
+		Timestamp:  time.Now(),
+	})
+
+	return nil
+}
+
 func (s *service) CheckCanDownload(actions []domain.Action) bool {
 	for _, action := range actions {
 		if !action.Enabled {

internal/action/sonarr.go

@@ -10,7 +10,7 @@ import (
 	"github.com/rs/zerolog/log"
 )
 
-func (s *service) sonarr(release domain.Release, action domain.Action) error {
+func (s *service) sonarr(release domain.Release, action domain.Action) ([]string, error) {
 	log.Trace().Msg("action SONARR")
 
 	// TODO validate data
@@ -18,13 +18,13 @@ func (s *service) sonarr(release domain.Release, action domain.Action) error {
 	// get client for action
 	client, err := s.clientSvc.FindByID(context.TODO(), action.ClientID)
 	if err != nil {
-		log.Error().Err(err).Msgf("error finding client: %v", action.ClientID)
-		return err
+		log.Error().Err(err).Msgf("sonarr: error finding client: %v", action.ClientID)
+		return nil, err
 	}
 
 	// return early if no client found
 	if client == nil {
-		return err
+		return nil, err
 	}
 
 	// initial config
@@ -52,23 +52,19 @@ func (s *service) sonarr(release domain.Release, action domain.Action) error {
 		PublishDate:      time.Now().Format(time.RFC3339),
 	}
 
-	success, rejections, err := arr.Push(r)
+	rejections, err := arr.Push(r)
 	if err != nil {
 		log.Error().Stack().Err(err).Msgf("sonarr: failed to push release: %v", r)
-		return err
+		return nil, err
 	}
 
-	if !success {
+	if rejections != nil {
 		log.Debug().Msgf("sonarr: release push rejected: %v, indexer %v to %v reasons: '%v'", r.Title, r.Indexer, client.Host, rejections)
 
-		// save pushed release
-		s.bus.Publish("release:update-push-status-rejected", release.ID, rejections)
-		return nil
+		return rejections, nil
 	}
 
 	log.Debug().Msgf("sonarr: successfully pushed release: %v, indexer %v to %v", r.Title, r.Indexer, client.Host)
 
-	s.bus.Publish("release:update-push-status", release.ID, domain.ReleasePushStatusApproved)
-
-	return nil
+	return nil, nil
 }

internal/domain/release.go

@@ -1457,7 +1457,7 @@ type ReleasePushStatus string
 const (
 	ReleasePushStatusApproved ReleasePushStatus = "PUSH_APPROVED"
 	ReleasePushStatusRejected ReleasePushStatus = "PUSH_REJECTED"
-	ReleasePushStatusMixed    ReleasePushStatus = "MIXED"   // For multiple actions, one might go and the other not
+	ReleasePushStatusErr      ReleasePushStatus = "PUSH_ERROR"
 	ReleasePushStatusPending  ReleasePushStatus = "PENDING" // Initial status
 )
 

internal/events/subscribers.go

@@ -25,6 +25,8 @@ func NewSubscribers(eventbus EventBus.Bus, releaseSvc release.Service) Subscribe
 
 func (s Subscriber) Register() {
 	s.eventbus.Subscribe("release:store-action-status", s.releaseActionStatus)
+	s.eventbus.Subscribe("release:push-rejected", s.releasePushRejected)
+	s.eventbus.Subscribe("release:push-approved", s.releasePushApproved)
 }
 
 func (s Subscriber) releaseActionStatus(actionStatus *domain.ReleaseActionStatus) {
@@ -35,3 +37,21 @@ func (s Subscriber) releaseActionStatus(actionStatus *domain.ReleaseActionStatus
 		log.Error().Err(err).Msgf("events: 'release:store-action-status' error")
 	}
 }
+
+func (s Subscriber) releasePushRejected(actionStatus *domain.ReleaseActionStatus) {
+	log.Trace().Msgf("events: 'release:push-rejected' '%+v'", actionStatus)
+
+	err := s.releaseSvc.StoreReleaseActionStatus(context.Background(), actionStatus)
+	if err != nil {
+		log.Error().Err(err).Msgf("events: 'release:push-rejected' error")
+	}
+}
+
+func (s Subscriber) releasePushApproved(actionStatus *domain.ReleaseActionStatus) {
+	log.Trace().Msgf("events: 'release:push-approved' '%+v'", actionStatus)
+
+	err := s.releaseSvc.StoreReleaseActionStatus(context.Background(), actionStatus)
+	if err != nil {
+		log.Error().Err(err).Msgf("events: 'release:push-approved' error")
+	}
+}