mirror of
https://github.com/idanoo/autobrr
synced 2025-07-23 00:39:13 +00:00
feat(feeds): add force run (#1243)
* feat(feeds): add force run * fix: simplify ForceRun * add confirmation modal * handle errors by using the test func * require user input to run * make sure to reschedule next job after forcerun * refactor modal centering with grid * refactor: Simplify startJob and forceRun logic - Refactor `startJob` to accept a `runImmediately` flag. This flag controls whether the job should be run immediately or scheduled for later. This change simplifies the `ForceRun` function by allowing it to call `startJob` with `runImmediately` set to `true`. - Remove redundant checks in `ForceRun` related to feed type. These checks are handled in `startJob`. BREAKING CHANGE: The `startJob` function now requires a second argument, `runImmediately`. This change affects all calls to `startJob`. * fix(web) Invalidate queries after forceRun * refactor(feeds): init and test run --------- Co-authored-by: ze0s <43699394+zze0s@users.noreply.github.com>
This commit is contained in:
soup
GitHub
parent
ff70a341ad
commit
2bd1a68a94
10 changed files with 318 additions and 38 deletions
|
|
@ -36,7 +36,7 @@ type NewznabJob struct {
|
|||
JobID int
|
||||
}
|
||||
|
||||
func NewNewznabJob(feed *domain.Feed, name string, indexerIdentifier string, log zerolog.Logger, url string, client newznab.Client, repo domain.FeedRepo, cacheRepo domain.FeedCacheRepo, releaseSvc release.Service) *NewznabJob {
|
||||
func NewNewznabJob(feed *domain.Feed, name string, indexerIdentifier string, log zerolog.Logger, url string, client newznab.Client, repo domain.FeedRepo, cacheRepo domain.FeedCacheRepo, releaseSvc release.Service) FeedJob {
|
||||
return &NewznabJob{
|
||||
Feed: feed,
|
||||
Name: name,
|
||||
|
|
@ -53,7 +53,7 @@ func NewNewznabJob(feed *domain.Feed, name string, indexerIdentifier string, log
|
|||
func (j *NewznabJob) Run() {
|
||||
ctx := context.Background()
|
||||
|
||||
if err := j.process(ctx); err != nil {
|
||||
if err := j.RunE(ctx); err != nil {
|
||||
j.Log.Err(err).Int("attempts", j.attempts).Msg("newznab process error")
|
||||
|
||||
j.errors = append(j.errors, err)
|
||||
|
|
@ -63,6 +63,15 @@ func (j *NewznabJob) Run() {
|
|||
j.errors = j.errors[:0]
|
||||
}
|
||||
|
||||
func (j *NewznabJob) RunE(ctx context.Context) error {
|
||||
if err := j.process(ctx); err != nil {
|
||||
j.Log.Err(err).Msg("newznab process error")
|
||||
return err
|
||||
}
|
||||
|
||||
return nil
|
||||
}
|
||||
|
||||
func (j *NewznabJob) process(ctx context.Context) error {
|
||||
// get feed
|
||||
items, err := j.getFeed(ctx)
|
||||
|
|
|
|||
|
|
@ -41,7 +41,7 @@ type RSSJob struct {
|
|||
JobID int
|
||||
}
|
||||
|
||||
func NewRSSJob(feed *domain.Feed, name string, indexerIdentifier string, log zerolog.Logger, url string, repo domain.FeedRepo, cacheRepo domain.FeedCacheRepo, releaseSvc release.Service, timeout time.Duration) *RSSJob {
|
||||
func NewRSSJob(feed *domain.Feed, name string, indexerIdentifier string, log zerolog.Logger, url string, repo domain.FeedRepo, cacheRepo domain.FeedCacheRepo, releaseSvc release.Service, timeout time.Duration) FeedJob {
|
||||
return &RSSJob{
|
||||
Feed: feed,
|
||||
Name: name,
|
||||
|
|
@ -58,15 +58,23 @@ func NewRSSJob(feed *domain.Feed, name string, indexerIdentifier string, log zer
|
|||
func (j *RSSJob) Run() {
|
||||
ctx := context.Background()
|
||||
|
||||
if err := j.process(ctx); err != nil {
|
||||
j.Log.Error().Err(err).Int("attempts", j.attempts).Msg("rss feed process error")
|
||||
if err := j.RunE(ctx); err != nil {
|
||||
j.Log.Err(err).Int("attempts", j.attempts).Msg("rss feed process error")
|
||||
|
||||
j.errors = append(j.errors, err)
|
||||
return
|
||||
}
|
||||
|
||||
j.attempts = 0
|
||||
j.errors = []error{}
|
||||
j.errors = j.errors[:0]
|
||||
}
|
||||
|
||||
func (j *RSSJob) RunE(ctx context.Context) error {
|
||||
if err := j.process(ctx); err != nil {
|
||||
j.Log.Err(err).Msg("rss feed process error")
|
||||
return err
|
||||
}
|
||||
|
||||
return nil
|
||||
}
|
||||
|
||||
func (j *RSSJob) process(ctx context.Context) error {
|
||||
|
|
|
|||
|
|
@ -36,6 +36,7 @@ type Service interface {
|
|||
DeleteFeedCache(ctx context.Context, id int) error
|
||||
GetLastRunData(ctx context.Context, id int) (string, error)
|
||||
DeleteFeedCacheStale(ctx context.Context) error
|
||||
ForceRun(ctx context.Context, id int) error
|
||||
|
||||
Start() error
|
||||
}
|
||||
|
|
@ -310,6 +311,12 @@ func (s *service) start() error {
|
|||
|
||||
for _, feed := range feeds {
|
||||
feed := feed
|
||||
|
||||
if !feed.Enabled {
|
||||
s.log.Trace().Msgf("feed disabled, skipping... %s", feed.Name)
|
||||
continue
|
||||
}
|
||||
|
||||
if err := s.startJob(&feed); err != nil {
|
||||
s.log.Error().Err(err).Msgf("failed to initialize feed job: %s", feed.Name)
|
||||
continue
|
||||
|
|
@ -339,18 +346,7 @@ func (s *service) restartJob(f *domain.Feed) error {
|
|||
|
||||
return nil
|
||||
}
|
||||
|
||||
func (s *service) startJob(f *domain.Feed) error {
|
||||
// if it's not enabled we should not start it
|
||||
if !f.Enabled {
|
||||
return nil
|
||||
}
|
||||
|
||||
// get torznab_url from settings
|
||||
if f.URL == "" {
|
||||
return errors.New("no URL provided for feed: %s", f.Name)
|
||||
}
|
||||
|
||||
func newFeedInstance(f *domain.Feed) feedInstance {
|
||||
// cron schedule to run every X minutes
|
||||
fi := feedInstance{
|
||||
Feed: f,
|
||||
|
|
@ -363,8 +359,12 @@ func (s *service) startJob(f *domain.Feed) error {
|
|||
Timeout: time.Duration(f.Timeout) * time.Second,
|
||||
}
|
||||
|
||||
return fi
|
||||
}
|
||||
|
||||
func (s *service) initializeFeedJob(fi feedInstance) (FeedJob, error) {
|
||||
var err error
|
||||
var job cron.Job
|
||||
var job FeedJob
|
||||
|
||||
switch fi.Implementation {
|
||||
case string(domain.FeedTypeTorznab):
|
||||
|
|
@ -377,15 +377,46 @@ func (s *service) startJob(f *domain.Feed) error {
|
|||
job, err = s.createRSSJob(fi)
|
||||
|
||||
default:
|
||||
return errors.New("unsupported feed type: %s", fi.Implementation)
|
||||
return nil, errors.New("unsupported feed type: %s", fi.Implementation)
|
||||
}
|
||||
|
||||
if err != nil {
|
||||
s.log.Error().Err(err).Msgf("failed to initialize %s feed", fi.Implementation)
|
||||
return err
|
||||
return nil, err
|
||||
}
|
||||
|
||||
identifierKey := feedKey{f.ID}.ToString()
|
||||
return job, nil
|
||||
}
|
||||
|
||||
func (s *service) startJob(f *domain.Feed) error {
|
||||
// if it's not enabled we should not start it
|
||||
if !f.Enabled {
|
||||
return errors.New("feed %s not enabled", f.Name)
|
||||
}
|
||||
|
||||
// get url from settings
|
||||
if f.URL == "" {
|
||||
return errors.New("no URL provided for feed: %s", f.Name)
|
||||
}
|
||||
|
||||
fi := newFeedInstance(f)
|
||||
|
||||
job, err := s.initializeFeedJob(fi)
|
||||
if err != nil {
|
||||
return errors.Wrap(err, "initialize job %s failed", f.Indexer)
|
||||
}
|
||||
|
||||
if err := s.scheduleJob(fi, job); err != nil {
|
||||
return errors.Wrap(err, "schedule job %s failed", f.Indexer)
|
||||
}
|
||||
|
||||
s.log.Debug().Msgf("successfully started feed: %s", f.Name)
|
||||
|
||||
return nil
|
||||
}
|
||||
|
||||
func (s *service) scheduleJob(fi feedInstance, job cron.Job) error {
|
||||
identifierKey := feedKey{fi.Feed.ID}.ToString()
|
||||
|
||||
// schedule job
|
||||
id, err := s.scheduler.ScheduleJob(job, fi.CronSchedule, identifierKey)
|
||||
|
|
@ -396,12 +427,10 @@ func (s *service) startJob(f *domain.Feed) error {
|
|||
// add to job map
|
||||
s.jobs[identifierKey] = id
|
||||
|
||||
s.log.Debug().Msgf("successfully started feed: %s", f.Name)
|
||||
|
||||
return nil
|
||||
}
|
||||
|
||||
func (s *service) createTorznabJob(f feedInstance) (cron.Job, error) {
|
||||
func (s *service) createTorznabJob(f feedInstance) (FeedJob, error) {
|
||||
s.log.Debug().Msgf("create torznab job: %s", f.Name)
|
||||
|
||||
if f.URL == "" {
|
||||
|
|
@ -424,8 +453,8 @@ func (s *service) createTorznabJob(f feedInstance) (cron.Job, error) {
|
|||
return job, nil
|
||||
}
|
||||
|
||||
func (s *service) createNewznabJob(f feedInstance) (cron.Job, error) {
|
||||
s.log.Debug().Msgf("add newznab job: %s", f.Name)
|
||||
func (s *service) createNewznabJob(f feedInstance) (FeedJob, error) {
|
||||
s.log.Debug().Msgf("create newznab job: %s", f.Name)
|
||||
|
||||
if f.URL == "" {
|
||||
return nil, errors.New("newznab feed requires URL")
|
||||
|
|
@ -443,8 +472,8 @@ func (s *service) createNewznabJob(f feedInstance) (cron.Job, error) {
|
|||
return job, nil
|
||||
}
|
||||
|
||||
func (s *service) createRSSJob(f feedInstance) (cron.Job, error) {
|
||||
s.log.Debug().Msgf("add rss job: %s", f.Name)
|
||||
func (s *service) createRSSJob(f feedInstance) (FeedJob, error) {
|
||||
s.log.Debug().Msgf("create rss job: %s", f.Name)
|
||||
|
||||
if f.URL == "" {
|
||||
return nil, errors.New("rss feed requires URL")
|
||||
|
|
@ -507,3 +536,25 @@ func (s *service) GetLastRunData(ctx context.Context, id int) (string, error) {
|
|||
|
||||
return feed, nil
|
||||
}
|
||||
|
||||
func (s *service) ForceRun(ctx context.Context, id int) error {
|
||||
feed, err := s.FindByID(ctx, id)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
|
||||
fi := newFeedInstance(feed)
|
||||
|
||||
job, err := s.initializeFeedJob(fi)
|
||||
if err != nil {
|
||||
s.log.Error().Err(err).Msg("failed to initialize feed job")
|
||||
return err
|
||||
}
|
||||
|
||||
if err := job.RunE(ctx); err != nil {
|
||||
s.log.Error().Err(err).Msg("failed to refresh feed")
|
||||
return err
|
||||
}
|
||||
|
||||
return nil
|
||||
}
|
||||
|
|
|
|||
|
|
@ -37,7 +37,12 @@ type TorznabJob struct {
|
|||
JobID int
|
||||
}
|
||||
|
||||
func NewTorznabJob(feed *domain.Feed, name string, indexerIdentifier string, log zerolog.Logger, url string, client torznab.Client, repo domain.FeedRepo, cacheRepo domain.FeedCacheRepo, releaseSvc release.Service) *TorznabJob {
|
||||
type FeedJob interface {
|
||||
Run()
|
||||
RunE(ctx context.Context) error
|
||||
}
|
||||
|
||||
func NewTorznabJob(feed *domain.Feed, name string, indexerIdentifier string, log zerolog.Logger, url string, client torznab.Client, repo domain.FeedRepo, cacheRepo domain.FeedCacheRepo, releaseSvc release.Service) FeedJob {
|
||||
return &TorznabJob{
|
||||
Feed: feed,
|
||||
Name: name,
|
||||
|
|
@ -54,7 +59,7 @@ func NewTorznabJob(feed *domain.Feed, name string, indexerIdentifier string, log
|
|||
func (j *TorznabJob) Run() {
|
||||
ctx := context.Background()
|
||||
|
||||
if err := j.process(ctx); err != nil {
|
||||
if err := j.RunE(ctx); err != nil {
|
||||
j.Log.Err(err).Int("attempts", j.attempts).Msg("torznab process error")
|
||||
|
||||
j.errors = append(j.errors, err)
|
||||
|
|
@ -64,6 +69,15 @@ func (j *TorznabJob) Run() {
|
|||
j.errors = j.errors[:0]
|
||||
}
|
||||
|
||||
func (j *TorznabJob) RunE(ctx context.Context) error {
|
||||
if err := j.process(ctx); err != nil {
|
||||
j.Log.Err(err).Int("attempts", j.attempts).Msg("torznab process error")
|
||||
return err
|
||||
}
|
||||
|
||||
return nil
|
||||
}
|
||||
|
||||
func (j *TorznabJob) process(ctx context.Context) error {
|
||||
// get feed
|
||||
items, err := j.getFeed(ctx)
|
||||
|
|
|
|||
Loading…
Add table
Add a link
Reference in a new issue