webhooker/internal/handlers/event_log.go

package handlers

import (
	"net/http"
	"slices"
	"time"

	"github.com/dustin/go-humanize"
	"gorm.io/gorm"
	"sneak.berlin/go/webhooker/internal/database"
	"sneak.berlin/go/webhooker/internal/delivery"
)

// DeliveryView is the display-safe projection of a delivery
// for the event log page. Its target is a TargetView, so the
// stored configuration blob — which holds the target's
// credential — has no path to the template.
type DeliveryView struct {
	ID     string
	Status database.DeliveryStatus
	Target delivery.TargetView

	// Replay is set on a delivery the Replay action created.
	Replay bool

	// Created is how long ago the delivery was created, and
	// CreatedUTC the full timestamp the page shows on hover.
	Created    string
	CreatedUTC string

	// Results is this delivery's attempts in attempt order,
	// bounded by maxRenderedAttempts. Without them a failure
	// renders as the status word alone and says nothing about
	// why.
	Results []DeliveryResultView

	// AttemptCount is how many attempts were recorded, which
	// is more than len(Results) once the middle was dropped.
	AttemptCount int

	// AttemptsOmitted is how many attempts were dropped from
	// the middle of Results. The page must show it, or the
	// bound would hide history rather than fold it.
	AttemptsOmitted int

	// Paused is set while the delivery is retrying and its
	// target's circuit breaker is open, and nil otherwise.
	Paused *PausedView
}

// eventLogTarget is what the event log needs to know about
// one target: the display-safe view its template renders, and
// the redactor that keeps that target's own credential out of
// the text its remote peer chose. The two are kept together
// so a caller cannot pick up one without the other, and apart
// from TargetView so the secrets never reach a template.
type eventLogTarget struct {
	View     delivery.TargetView
	Redactor delivery.Redactor
}

// The event log's show query parameter and its two values: the events
// with a failed delivery, and those with a delivery still pending or
// retrying.
const (
	showParam   = "show"
	showFailed  = "failed"
	showPending = "pending"
)

// HandleSourceLogs shows the request/response logs for a
// webhook.
func (h *Handlers) HandleSourceLogs() http.HandlerFunc {
	return func(w http.ResponseWriter, r *http.Request) {
		webhook, ok := h.ownedWebhook(w, r)
		if !ok {
			return
		}

		targets, err := h.loadTargetMap(webhook.ID)
		if err != nil {
			// Without the map every delivery renders through a
			// zero redactor, so failing the page is the only
			// safe answer.
			h.serverError(w, r, "failed to load targets", err)

			return
		}

		// Any other value of show lists every event, as no value
		// does.
		show := r.URL.Query().Get(showParam)

		statuses := eventLogStatuses(show)
		if statuses == nil {
			show = ""
		}

		evts, total, ok := h.loadEventsWithDeliveries(
			w, r, webhook, targets, statuses,
		)
		if !ok {
			return
		}

		failed, pending, err := h.countFailedAndPendingEvents(webhook.ID)
		if err != nil {
			h.serverError(w, r, "failed to count events", err)

			return
		}

		data := map[string]any{
			tmplKeyWebhook:  &webhook,
			"Events":        evts,
			"TotalEvents":   total,
			"Show":          show,
			"FailedEvents":  failed,
			"PendingEvents": pending,
		}

		h.renderTemplate(w, r, "source_logs.html", data)
	}
}

// loadTargetMap loads targets into a map of display-safe
// views keyed by target ID, each paired with its redactor.
// The projection happens here so that no caller can hand a
// raw target, configuration blob and all, to a template: the
// raw rows do not leave this function.
//
// The load is Unscoped because deleting a target only soft
// deletes the row while its deliveries survive in the
// per-webhook database. Both halves of the map need those rows:
// a scoped load leaves an old delivery with a zero redactor,
// which renders its response bodies unredacted, and with a zero
// view, which renders its target as a blank name.
//
// This map is historical display only. It is built for the event
// log and an event's own page, and reaches nothing but
// DeliveryView.Target: the target list on the source detail page,
// the edit form and the replay path each resolve targets
// themselves, and a deleted row is refused there as before.
func (h *Handlers) loadTargetMap(
	webhookID string,
) (map[string]eventLogTarget, error) {
	var targets []database.Target

	err := h.db.DB().Unscoped().Where(
		"webhook_id = ?", webhookID,
	).Find(&targets).Error
	if err != nil {
		return nil, err
	}

	targetMap := make(
		map[string]eventLogTarget, len(targets),
	)

	for i := range targets {
		targetMap[targets[i].ID] = eventLogTarget{
			Redactor: delivery.NewRedactor(&targets[i]),
		}
	}

	// The views come from NewTargetViews rather than being
	// rebuilt here, so the masking rules stay in one place and a
	// deleted target's configuration is masked by the same code
	// that masks a live one's.
	for _, v := range delivery.NewTargetViews(targets) {
		entry := targetMap[v.ID]
		entry.View = v
		targetMap[v.ID] = entry
	}

	return targetMap, nil
}

// loadEventsWithDeliveries loads the recentEventLimit newest events
// and their deliveries from the per-webhook database, and the total
// number of events stored. Given delivery statuses, both cover only
// the events with a delivery in one of them. Events come back as
// capped projections rather than database.Event rows: see
// eventLogColumns for why the cut happens in SQL.
//
// The bool reports whether the load succeeded. It is false
// once this has answered the request with an error, and the
// caller must then render nothing further.
func (h *Handlers) loadEventsWithDeliveries(
	w http.ResponseWriter,
	r *http.Request,
	webhook database.Webhook,
	targetMap map[string]eventLogTarget,
	statuses []database.DeliveryStatus,
) ([]EventLogView, int64, bool) {
	if !h.dbMgr.DBExists(webhook.ID) {
		return nil, 0, true
	}

	webhookDB, err := h.dbMgr.GetDB(webhook.ID)
	if err != nil {
		h.serverError(
			w, r, "failed to get webhook database", err,
		)

		return nil, 0, false
	}

	rows, totalEvents, err := loadEventLogRows(
		webhookDB, webhook.ID, statuses,
	)
	if err != nil {
		h.serverError(w, r, "failed to load events", err)

		return nil, 0, false
	}

	result, ok := h.eventLogViews(
		w, r, webhookDB, webhook.ID, rows, targetMap,
		maxRenderedBodyBytes,
	)

	return result, totalEvents, ok
}

// eventLogViews projects loaded events for rendering, each with
// its deliveries, how many times it has been resubmitted and the
// entrypoint it arrived at (for a resubmitted copy, the one the
// request it copies arrived at), and with its request headers only
// when their text holds at most maxHeaderBytes. Like
// loadEventsWithDeliveries, it reports false once it has answered
// the request with an error.
func (h *Handlers) eventLogViews(
	w http.ResponseWriter,
	r *http.Request,
	webhookDB *gorm.DB,
	webhookID string,
	rows []eventLogRow,
	targetMap map[string]eventLogTarget,
	maxHeaderBytes int,
) ([]EventLogView, bool) {
	result := make([]EventLogView, len(rows))
	eventDeliveries := make([][]database.Delivery, len(rows))

	var deliveryIDs []string

	eventIDs := make([]string, len(rows))

	for i := range rows {
		result[i] = rows[i].view(webhookID, maxHeaderBytes)
		eventIDs[i] = rows[i].ID

		webhookDB.Where(
			"event_id = ?", rows[i].ID,
		).Find(&eventDeliveries[i])

		for j := range eventDeliveries[i] {
			deliveryIDs = append(
				deliveryIDs, eventDeliveries[i][j].ID,
			)
		}
	}

	attempts, err := h.loadDeliveryResults(
		webhookDB, deliveryIDs,
	)
	if err != nil {
		h.serverError(
			w, r, "failed to load delivery attempts", err,
		)

		return nil, false
	}

	resubmits, err := resubmitCounts(webhookDB, eventIDs)
	if err != nil {
		h.serverError(
			w, r, "failed to count event resubmissions", err,
		)

		return nil, false
	}

	entrypoints, err := h.entrypointNames(webhookID)
	if err != nil {
		h.serverError(w, r, "failed to load entrypoints", err)

		return nil, false
	}

	for i := range rows {
		result[i].Deliveries = h.newDeliveryViews(
			eventDeliveries[i], targetMap, attempts,
		)
		result[i].ResubmitCount = resubmits[rows[i].ID]

		name, ok := entrypoints[rows[i].EntrypointID]
		if !ok {
			name = "deleted entrypoint"
		}

		result[i].Entrypoint = name
	}

	return result, true
}

// loadEventLogRows reads the event log projection of the
// recentEventLimit newest events, newest first, and the total number
// of events stored, both narrowed by statuses as eventsWithStatus
// narrows them.
func loadEventLogRows(
	webhookDB *gorm.DB,
	webhookID string,
	statuses []database.DeliveryStatus,
) ([]eventLogRow, int64, error) {
	totalEvents, err := countEventsWithStatus(
		webhookDB, webhookID, statuses,
	)
	if err != nil {
		return nil, 0, err
	}

	var rows []eventLogRow

	err = eventsWithStatus(webhookDB, webhookID, statuses).Select(
		eventLogColumns, maxRenderedBodyBytes, maxRenderedBodyBytes,
	).Order("created_at DESC").Limit(recentEventLimit).Find(&rows).Error

	return rows, totalEvents, err
}

// eventLogStatuses returns the delivery statuses the event log's show
// value lists events by, or nil for one that lists every event.
func eventLogStatuses(show string) []database.DeliveryStatus {
	switch show {
	case showFailed:
		return []database.DeliveryStatus{database.DeliveryStatusFailed}
	case showPending:
		return []database.DeliveryStatus{
			database.DeliveryStatusPending,
			database.DeliveryStatusRetrying,
		}
	default:
		return nil
	}
}

// eventsWithStatus selects the webhook's events, or, given statuses,
// the recentEventLimit newest of those with at least one delivery in
// one of them.
//
// Given statuses, its cost follows the matching deliveries. SQLite
// never reorders a CROSS JOIN, so it reads each join's left side
// first: the distinct event IDs of the matching deliveries, through
// idx_deliveries_status; then each of those events by ID, sorted to
// keep the newest; then the rows of only the events kept, so no other
// event's body is read. With a plain "id IN (matching deliveries)"
// condition instead, SQLite, which keeps no statistics on these
// tables, walks every event newest first.
func eventsWithStatus(
	webhookDB *gorm.DB,
	webhookID string,
	statuses []database.DeliveryStatus,
) *gorm.DB {
	if statuses == nil {
		return webhookDB.Model(&database.Event{}).Where(
			"webhook_id = ?", webhookID,
		)
	}

	matching := webhookDB.Model(&database.Delivery{}).
		Distinct("event_id").Where("status IN ?", statuses)

	newest := webhookDB.Table("(?) AS matching", matching).
		Joins("CROSS JOIN events ON events.id = matching.event_id").
		Where(
			"events.webhook_id = ? AND events.deleted_at IS NULL",
			webhookID,
		).
		Order("events.created_at DESC").Limit(recentEventLimit).
		Select("events.id AS event_id")

	return webhookDB.Table("(?) AS newest", newest).
		Joins("CROSS JOIN events ON events.id = newest.event_id")
}

// countEventsWithStatus counts the webhook's events with at least one
// delivery in one of the statuses, or every event when statuses is
// nil. Given statuses, it counts the distinct events of the matching
// deliveries and reads nothing but those deliveries, through
// idx_deliveries_status, where counting the events would read every
// event row. That is the same number, because retention deletes an
// event's deliveries with it.
func countEventsWithStatus(
	webhookDB *gorm.DB,
	webhookID string,
	statuses []database.DeliveryStatus,
) (int64, error) {
	var count int64

	if statuses == nil {
		err := webhookDB.Model(&database.Event{}).Where(
			"webhook_id = ?", webhookID,
		).Count(&count).Error

		return count, err
	}

	err := webhookDB.Model(&database.Delivery{}).Distinct("event_id").
		Where("status IN ?", statuses).Count(&count).Error

	return count, err
}

// countFailedAndPendingEvents returns how many of the webhook's events
// the event log lists when it shows only those with a failed delivery,
// and when it shows only those with a delivery pending or retrying.
func (h *Handlers) countFailedAndPendingEvents(
	webhookID string,
) (int64, int64, error) {
	if !h.dbMgr.DBExists(webhookID) {
		return 0, 0, nil
	}

	webhookDB, err := h.dbMgr.GetDB(webhookID)
	if err != nil {
		return 0, 0, err
	}

	failed, err := countEventsWithStatus(
		webhookDB, webhookID, eventLogStatuses(showFailed),
	)
	if err != nil {
		return 0, 0, err
	}

	pending, err := countEventsWithStatus(
		webhookDB, webhookID, eventLogStatuses(showPending),
	)
	if err != nil {
		return 0, 0, err
	}

	return failed, pending, nil
}

// resubmitCounts reports, for each of the page's events, how many
// events have been resubmitted from it.
//
// One grouped query covers the page rather than one query per event.
// The page shows at most recentEventLimit events, far below SQLite's
// bound parameter ceiling, so it needs no chunking as the delivery
// result load does.
func resubmitCounts(
	webhookDB *gorm.DB, eventIDs []string,
) (map[string]int, error) {
	counts := make(map[string]int, len(eventIDs))

	if len(eventIDs) == 0 {
		return counts, nil
	}

	var rows []struct {
		ResubmittedFromID string
		Total             int
	}

	err := webhookDB.Model(&database.Event{}).
		Select("resubmitted_from_id, count(*) AS total").
		Where("resubmitted_from_id IN ?", eventIDs).
		Group("resubmitted_from_id").
		Find(&rows).Error
	if err != nil {
		return nil, err
	}

	for _, row := range rows {
		counts[row.ResubmittedFromID] = row.Total
	}

	return counts, nil
}

// deliveryIDChunkSize bounds how many delivery IDs go into one
// IN clause. SQLite refuses a statement carrying more than
// SQLITE_MAX_VARIABLE_NUMBER (32766) bound parameters, and a
// page holds one delivery per target per event, so a webhook
// with enough targets would turn the whole query into an error
// and the page into zero attempts.
const deliveryIDChunkSize = 500

// loadDeliveryResults loads the recorded attempts for the
// page's deliveries, keyed by delivery ID.
//
// Each response body is cut by SQLite rather than in Go, for
// the reason deliveryResultColumns gives. How many attempts a
// delivery has is the target's MaxRetries, which the
// authenticated operator sets; how many of them reach the page
// is bounded again by maxRenderedAttempts.
func (h *Handlers) loadDeliveryResults(
	webhookDB *gorm.DB,
	deliveryIDs []string,
) (map[string][]deliveryResultRow, error) {
	byDelivery := make(map[string][]deliveryResultRow)

	for chunk := range slices.Chunk(
		deliveryIDs, deliveryIDChunkSize,
	) {
		var rows []deliveryResultRow

		err := webhookDB.Model(
			&database.DeliveryResult{},
		).Select(
			deliveryResultColumns, maxRenderedResponseBytes,
		).Where(
			"delivery_id IN ?", chunk,
		).Order("attempt_num ASC").Find(&rows).Error
		if err != nil {
			// Returning what was loaded so far renders the
			// deliveries in the failed chunk as never having run,
			// which is indistinguishable from ones that really
			// never ran. The page fails instead.
			return nil, err
		}

		for i := range rows {
			byDelivery[rows[i].DeliveryID] = append(
				byDelivery[rows[i].DeliveryID], rows[i],
			)
		}
	}

	return byDelivery, nil
}

// newDeliveryViews projects deliveries for rendering,
// resolving each one's target to its display-safe view and
// each one's attempts through that target's redactor. A
// retrying delivery also reads its target's circuit breaker.
func (h *Handlers) newDeliveryViews(
	deliveries []database.Delivery,
	targetMap map[string]eventLogTarget,
	attempts map[string][]deliveryResultRow,
) []DeliveryView {
	views := make([]DeliveryView, len(deliveries))

	for i := range deliveries {
		target := targetMap[deliveries[i].TargetID]
		rows := attempts[deliveries[i].ID]
		created := deliveries[i].CreatedAt

		results, omitted := renderedAttempts(
			rows, target.Redactor,
		)

		views[i] = DeliveryView{
			ID:              deliveries[i].ID,
			Status:          deliveries[i].Status,
			Target:          target.View,
			Replay:          deliveries[i].Replay,
			Created:         humanize.Time(created),
			CreatedUTC:      created.UTC().Format(time.DateTime) + " UTC",
			Results:         results,
			AttemptCount:    len(rows),
			AttemptsOmitted: omitted,
		}

		if deliveries[i].Status == database.DeliveryStatusRetrying {
			views[i].Paused = h.deliveryPausedView(
				deliveries[i].TargetID, rows,
			)
		}
	}

	return views
}

// maxRenderedAttempts bounds how many of one delivery's
// attempts the page renders. Past it the middle is dropped and
// counted, keeping the first attempts and the last ones: how
// the delivery started failing and how it ended are what a
// reader needs, and the count says plainly that the rest was
// dropped rather than never recorded.
const (
	renderedAttemptsHead = 10
	renderedAttemptsTail = 10
	maxRenderedAttempts  = renderedAttemptsHead +
		renderedAttemptsTail
)

// renderedAttempts projects a delivery's attempts through the
// target's redactor, at most maxRenderedAttempts of them, and
// reports how many it dropped.
func renderedAttempts(
	rows []deliveryResultRow,
	redactor delivery.Redactor,
) ([]DeliveryResultView, int) {
	omitted := 0

	if len(rows) > maxRenderedAttempts {
		omitted = len(rows) - maxRenderedAttempts

		kept := make(
			[]deliveryResultRow, 0, maxRenderedAttempts,
		)
		kept = append(kept, rows[:renderedAttemptsHead]...)
		kept = append(
			kept, rows[len(rows)-renderedAttemptsTail:]...,
		)
		rows = kept
	}

	views := make([]DeliveryResultView, len(rows))
	for i := range rows {
		views[i] = rows[i].view(redactor)
	}

	return views, omitted
}