Method Handle

internal/gmailwatch/processor.go:127–217

(ctx context.Context, notification Notification)

}

func (p *Processor) Handle(ctx context.Context, notification Notification) (*Payload, error) {
	if p.Repository == nil {
		return nil, ErrMissingRepository
	}

	if notification.MessageID != "" {
		state := p.Repository.Get()
		if state.LastPushMessageID == notification.MessageID {
			p.logf("watch: ignoring duplicate push %s", notification.MessageID)

			return nil, ErrNoNewMessages
		}
	}

	if err := p.checkRateLimitCircuit(p.currentTime()); err != nil {
		return nil, err
	}

	startID, err := p.Repository.StartHistoryID(notification.HistoryID)
	if err != nil {
		return nil, err
	}

	if startID == 0 {
		p.logStaleNotification(notification.HistoryID)

		return nil, ErrNoNewMessages
	}

	if sleepErr := p.sleepForFetch(ctx); sleepErr != nil {
		return nil, sleepErr
	}

	if p.NewSource == nil {
		return nil, ErrMissingSourceFactory
	}

	source, err := p.NewSource(ctx)
	if err != nil {
		return nil, err
	}

	if source == nil {
		return nil, ErrMissingSource
	}

	historyPage, err := source.ListHistory(ctx, startID, p.Config.HistoryMax, p.Config.HistoryTypes)
	if err != nil {
		if p.IsStaleHistoryError != nil && p.IsStaleHistoryError(err) {
			return p.resync(ctx, source, notification)
		}

		return nil, p.openRateLimitCircuitIfNeeded(err)
	}

	nextHistoryID := notification.HistoryID
	if historyPage.HistoryID != "" {
		nextHistoryID = historyPage.HistoryID