| 83 | } |
| 84 | |
| 85 | func (p *Processor) Process(ctx context.Context, notification Notification) (*ProcessedPayload, error) { |
| 86 | if p.Repository == nil { |
| 87 | return nil, ErrMissingRepository |
| 88 | } |
| 89 | |
| 90 | progressBefore := p.Repository.Get() |
| 91 | |
| 92 | payload, err := p.Handle(ctx, notification) |
| 93 | if err != nil { |
| 94 | return nil, err |
| 95 | } |
| 96 | |
| 97 | if payload == nil { |
| 98 | return nil, ErrNoNewMessages |
| 99 | } |
| 100 | |
| 101 | processed := &ProcessedPayload{Payload: payload} |
| 102 | if p.Deliver == nil { |
| 103 | return processed, nil |
| 104 | } |
| 105 | |
| 106 | delivery := p.Deliver(ctx, payload) |
| 107 | if delivery.Record { |
| 108 | if recordErr := p.Repository.RecordDelivery(delivery.Status, delivery.Note, p.currentTime()); recordErr != nil { |
| 109 | p.warnf("watch: failed to update delivery state: %v", recordErr) |
| 110 | } |
| 111 | } |
| 112 | |
| 113 | if delivery.Err == nil { |
| 114 | return processed, nil |
| 115 | } |
| 116 | |
| 117 | p.warnf("watch: hook failed: %v", delivery.Err) |
| 118 | processed.HookFailed = true |
| 119 | |
| 120 | if _, restoreErr := p.Repository.RestoreProgress(progressBefore, payload.HistoryID, notification.MessageID); restoreErr != nil { |
| 121 | p.warnf("watch: failed to preserve retry state after hook failure: %v", restoreErr) |
| 122 | } |
| 123 | |
| 124 | return processed, &HookDeliveryError{Err: delivery.Err} |
| 125 | } |
| 126 | |
| 127 | func (p *Processor) Handle(ctx context.Context, notification Notification) (*Payload, error) { |
| 128 | if p.Repository == nil { |