webhooker/internal/delivery/queue_depth.go

package delivery

import (
	"context"
	"fmt"
	"time"

	"sneak.berlin/go/webhooker/internal/database"
)

// queueDepthSampleInterval is how often the pending and retrying
// queue depths are counted and published as gauges.
const queueDepthSampleInterval = 30 * time.Second

// queueDepthSampler publishes the pending and retrying queue depths
// on a timer for as long as the engine runs.
//
// The depths are counted out of the databases rather than tracked as
// deltas alongside the status transitions. A delta counter would have
// to be seeded correctly at startup from rows written by a previous
// process, and would drift permanently on any transition that failed
// to persist. Counting is the measurement that cannot go wrong, and
// it is the same whole-database walk the retry sweep already makes.
func (e *Engine) queueDepthSampler(ctx context.Context) {
	defer e.wg.Done()

	ticker := time.NewTicker(queueDepthSampleInterval)
	defer ticker.Stop()

	e.sampleQueueDepths(ctx)

	for {
		select {
		case <-ctx.Done():
			return
		case <-ticker.C:
			e.sampleQueueDepths(ctx)
		}
	}
}

// sampleQueueDepths counts every queued delivery across all
// per-webhook databases and publishes the result.
func (e *Engine) sampleQueueDepths(ctx context.Context) {
	if e.database == nil || e.dbManager == nil {
		return
	}

	types, err := e.targetTypesByID()
	if err != nil {
		e.log.Error(
			"queue depth sample: failed to load target types",
			"error", err,
		)

		return
	}

	var webhookIDs []string

	err = e.database.DB().
		Model(&database.Webhook{}).
		Pluck("id", &webhookIDs).Error
	if err != nil {
		e.log.Error(
			"queue depth sample: failed to query webhook IDs",
			"error", err,
		)

		return
	}

	pending := make(map[database.TargetType]int)
	retrying := make(map[database.TargetType]int)

	for _, webhookID := range webhookIDs {
		select {
		case <-ctx.Done():
			return
		default:
		}

		if !e.dbManager.DBExists(webhookID) {
			continue
		}

		e.sampleWebhookQueueDepths(
			webhookID, types, pending, retrying,
		)
	}

	e.mx.SetQueueDepths(pending, retrying)
}

// targetTypesByID maps every configured target id to its type. The
// deliveries live in the per-webhook databases but carry only a
// target id, so the type label has to come from the main database.
func (e *Engine) targetTypesByID() (
	map[string]database.TargetType, error,
) {
	var rows []struct {
		ID   string
		Type database.TargetType
	}

	err := e.database.DB().
		Model(&database.Target{}).
		Select("id", "type").
		Scan(&rows).Error
	if err != nil {
		return nil, fmt.Errorf("loading targets: %w", err)
	}

	types := make(map[string]database.TargetType, len(rows))

	for _, row := range rows {
		types[row.ID] = row.Type
	}

	return types, nil
}

// sampleWebhookQueueDepths adds one webhook's queued deliveries into
// the running totals. A delivery whose target has since been deleted
// resolves to the empty type and lands in the unknown bucket rather
// than being dropped.
func (e *Engine) sampleWebhookQueueDepths(
	webhookID string,
	types map[string]database.TargetType,
	pending, retrying map[database.TargetType]int,
) {
	webhookDB, err := e.dbManager.GetDB(webhookID)
	if err != nil {
		e.log.Error(
			"queue depth sample: failed to get webhook database",
			"webhook_id", webhookID,
			"error", err,
		)

		return
	}

	var rows []struct {
		TargetID string
		Status   database.DeliveryStatus
		Depth    int
	}

	err = webhookDB.
		Model(&database.Delivery{}).
		Select("target_id", "status", "count(*) as depth").
		Where("status IN ?", []database.DeliveryStatus{
			database.DeliveryStatusPending,
			database.DeliveryStatusRetrying,
		}).
		Group("target_id, status").
		Scan(&rows).Error
	if err != nil {
		e.log.Error(
			"queue depth sample: "+
				"failed to count queued deliveries",
			"webhook_id", webhookID,
			"error", err,
		)

		return
	}

	for _, row := range rows {
		targetType := types[row.TargetID]

		switch row.Status {
		case database.DeliveryStatusPending:
			pending[targetType] += row.Depth
		case database.DeliveryStatusRetrying:
			retrying[targetType] += row.Depth
		case database.DeliveryStatusDelivered,
			database.DeliveryStatusFailed:
			// Excluded by the query above: a delivery that has
			// reached a terminal state is not queued.
		}
	}
}