Gallery

// StartProcessing marks a token as processing. It returns a callback that must be called when work on the token is finished in order to mark
// it as finished. If withRetry is true, the callback will attempt to reenqueue the token if an error is passed. attemps is ignored when MaxRetries
// is set to the default value of 0.
func (m Manager) StartProcessing(ctx context.Context, td db.TokenDefinition, attempts int, cause persist.ProcessingCause) (func(db.TokenMedia, error) error, error) {
	if m.Paused(ctx, td) {
		recordPipelinePaused(ctx, m.metricReporter, td.Chain, td.ContractAddress, cause)
		err := ErrContractPaused{Chain: td.Chain, Contract: td.ContractAddress}
		sentryutil.ReportError(ctx, err)
		return nil, err
	}

	err := m.throttle.Lock(ctx, "lock:"+td.ID.String())
	if err != nil {
		return nil, err
	}

	stop := make(chan bool)
	done := make(chan bool)
	tick := time.NewTicker(10 * time.Second)

	go func() {
		defer tick.Stop()
		m.Registry.keepAlive(ctx, td.ID)
		for {
			select {
			case <-tick.C:
				m.Registry.keepAlive(ctx, td.ID)
			case <-stop:
				close(done)
				return
			}
		}
	}()

	start := time.Now()

	callback := func(tm db.TokenMedia, err error) error {
		close(stop)
		<-done
		m.tickTokenF(td) // mark that the token ran so if an error occured tryRetry delays the next run appropriately
		m.recordError(ctx, td, err)
		m.tryRetry(ctx, td, err, attempts)
		m.throttle.Unlock(ctx, "lock:"+td.ID.String())
		recordMetrics(ctx, m.metricReporter, td.Chain, tm.Media.MediaType, err, time.Since(start), cause)
		return nil
	}

	return callback, err
}