notify-gateway/internal/gateway/queue.go

package gateway

import (
	"context"
	"crypto/sha256"
	"encoding/hex"
	"encoding/json"
	"errors"
	"fmt"
	"log"
	"math/rand/v2"
	"strings"
	"time"

	"github.com/example/notify-gateway/internal/config"
	"github.com/example/notify-gateway/internal/divera"
	"github.com/example/notify-gateway/internal/model"
	"github.com/example/notify-gateway/internal/outbound"
	"github.com/example/notify-gateway/internal/outbox"
)

type Delivery struct {
	Kind     string                 `json:"kind"`
	Payload  map[string]any         `json:"payload"`
	Outbound *config.OutboundConfig `json:"outbound,omitempty"`
	Divera   *config.DiveraConfig   `json:"divera,omitempty"`
}
type Queue struct {
	Store      *outbox.Store
	Dispatcher *Dispatcher
	Logger     *log.Logger
}
type InputError struct{ Err error }

func (e *InputError) Error() string { return e.Err.Error() }
func (e *InputError) Unwrap() error { return e.Err }

func (q *Queue) Accept(ctx context.Context, msg model.InboundMessage, key string) (outbox.Receipt, error) {
	if len(key) > 256 {
		return outbox.Receipt{}, &InputError{errors.New("Idempotency-Key exceeds 256 bytes")}
	}
	original := msg
	original.ReceivedAt = time.Time{}
	b, err := json.Marshal(original)
	if err != nil {
		return outbox.Receipt{}, &InputError{err}
	}
	hash := sha256.Sum256(b)
	digest := hex.EncodeToString(hash[:])
	scopeBytes, _ := json.Marshal([]string{msg.Source, msg.Channel})
	scope := string(scopeBytes)
	if key == "" {
		key = outbox.ID()
	} else {
		r, err := q.Store.Lookup(ctx, scope, key, digest)
		if err != nil || r.ID != "" {
			return r, err
		}
	}
	cfg := q.Dispatcher.store.Get()
	var jobs []outbox.Job
	for _, m := range cfg.Mappings {
		if !m.Enabled || !matches(m, msg) {
			continue
		}
		payload, kind, err := buildPayload(m, msg)
		if err != nil {
			return outbox.Receipt{}, &InputError{fmt.Errorf("mapping %s: %w", m.ID, err)}
		}
		delivery := Delivery{Kind: kind, Payload: payload}
		if config.IsOutbound(kind) {
			for _, o := range cfg.Outbounds {
				if o.ID == m.OutboundID && o.Provider == kind {
					delivery.Outbound = &o
					break
				}
			}
			if delivery.Outbound == nil {
				return outbox.Receipt{}, &InputError{errors.New("outbound destination missing")}
			}
		} else {
			delivery.Divera = &cfg.Divera
		}
		data, err := json.Marshal(delivery)
		if err != nil {
			return outbox.Receipt{}, err
		}
		jobs = append(jobs, outbox.Job{MappingID: m.ID, Target: m.Target, Data: data})
	}
	if len(jobs) == 0 {
		return outbox.Receipt{}, &InputError{errors.New("no mapping matched")}
	}
	return q.Store.Enqueue(ctx, scope, key, digest, jobs)
}

func retryDelay(attempt int, after time.Duration) time.Duration {
	d := 5 * time.Second * time.Duration(1<<min(max(attempt-1, 0), 8))
	d += time.Duration(rand.Int64N(int64(d/5) + 1))
	if after > d {
		d = after
	}
	return min(d, 24*time.Hour)
}

// ProcessOne is also used by integration tests. A lease survives process crashes.
func (q *Queue) ProcessOne(ctx context.Context) (bool, error) {
	j, err := q.Store.Claim(ctx, time.Now())
	if err != nil || j == nil {
		return false, err
	}
	var d Delivery
	var resp outbound.Response
	err = json.Unmarshal(j.Data, &d)
	dry := false
	retry := false
	if err == nil {
		if d.Outbound != nil {
			dry = !d.Outbound.Live
			resp, err = outbound.SendWithID(ctx, *d.Outbound, d.Payload, j.ID)
			retry = resp.Retryable
		} else if d.Divera != nil {
			dry = d.Divera.DryRun
			c := divera.New(func() config.DiveraConfig { return *d.Divera })
			r, e := c.Create(ctx, d.Kind, d.Payload)
			resp.StatusCode = r.StatusCode
			resp.RetryAfter = r.RetryAfter
			err = e
			retry = outbound.RetryableHTTP(r.StatusCode)
		} else {
			err = errors.New("invalid persisted destination")
		}
	}
	state := "succeeded"
	message := ""
	if dry {
		state = "dry_run"
	}
	if err != nil {
		// Never persist provider error strings: URLs, SMTP responses and templates may contain secrets.
		message = "Zustellung fehlgeschlagen"
		if resp.StatusCode != 0 {
			message = fmt.Sprintf("Provider-Status %d", resp.StatusCode)
		}
		state = "dead"
		if retry && j.Attempts < 8 {
			state = "pending"
		}
	}
	next := time.Now().Add(retryDelay(j.Attempts, resp.RetryAfter))
	// Persist the outcome even if shutdown canceled the network request.
	finishCtx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
	defer cancel()
	if err := q.Store.Finish(finishCtx, *j, state, resp.StatusCode, message, next); err != nil {
		return true, err
	}
	if q.Logger != nil {
		q.Logger.Printf("delivery id=%s state=%s attempt=%d status=%d", j.ID, state, j.Attempts, resp.StatusCode)
	}
	return true, nil
}
func (q *Queue) Run(ctx context.Context) {
	ticker := time.NewTicker(time.Second)
	defer ticker.Stop()
	for {
		if ctx.Err() != nil {
			return
		}
		worked, err := q.ProcessOne(ctx)
		if err != nil && q.Logger != nil {
			q.Logger.Print("outbox worker: database operation failed")
		}
		if worked && err == nil {
			continue
		}
		select {
		case <-ctx.Done():
			return
		case <-ticker.C:
		}
	}
}

// Message keys should never contain transport credentials.
func ScopedKey(parts ...string) string {
	b, _ := json.Marshal(parts)
	sum := sha256.Sum256(b)
	return strings.ToLower(hex.EncodeToString(sum[:]))
}