webhooker/internal/handlers/webhook.go

package handlers

import (
	"encoding/json"
	"fmt"
	"io"
	"net/http"

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

const (
	// maxWebhookBodySize is the maximum allowed webhook
	// request body (1 MB).
	maxWebhookBodySize = 1 << maxBodyShift
)

// HandleWebhook handles incoming webhook requests at entrypoint
// URLs.
func (h *Handlers) HandleWebhook() http.HandlerFunc {
	return func(w http.ResponseWriter, r *http.Request) {
		if r.Method != http.MethodPost {
			w.Header().Set("Allow", "POST")
			http.Error(
				w,
				"Method Not Allowed",
				http.StatusMethodNotAllowed,
			)

			return
		}

		entrypointUUID := chi.URLParam(r, "uuid")
		if entrypointUUID == "" {
			http.NotFound(w, r)

			return
		}

		entrypoint, ok := h.lookupEntrypoint(
			w, r, entrypointUUID,
		)
		if !ok {
			return
		}

		// Logged only once the UUID is known to name a real
		// entrypoint. The UUID comes straight out of the path on
		// the one unauthenticated endpoint, so logging it before
		// the lookup let a client write an INFO line per invented
		// path; the request itself is already in the access log
		// and a miss is already logged at DEBUG.
		h.log.Info("webhook request received",
			"entrypoint_uuid", entrypointUUID,
			"method", r.Method,
			"remote_addr", r.RemoteAddr,
		)

		if !entrypoint.Active {
			http.Error(w, "Gone", http.StatusGone)

			return
		}

		h.processWebhookRequest(w, r, entrypoint)
	}
}

// processWebhookRequest reads the body, serializes headers, loads
// targets, and delivers the event.
//
// Nothing about the request itself is authenticated: the entrypoint
// UUID in the path is the credential, and reaching here means it
// matched an active entrypoint.
func (h *Handlers) processWebhookRequest(
	w http.ResponseWriter,
	r *http.Request,
	entrypoint database.Entrypoint,
) {
	body, ok := h.readWebhookBody(w, r)
	if !ok {
		return
	}

	headersJSON, err := json.Marshal(r.Header)
	if err != nil {
		h.serverError(w, "failed to serialize headers", err)

		return
	}

	targets, err := h.loadActiveTargets(entrypoint.WebhookID)
	if err != nil {
		h.serverError(w, "failed to query targets", err)

		return
	}

	h.createAndDeliverEvent(
		w, r, entrypoint, body, headersJSON, targets,
	)
}

// loadActiveTargets returns all active targets for a webhook.
func (h *Handlers) loadActiveTargets(
	webhookID string,
) ([]database.Target, error) {
	var targets []database.Target

	err := h.db.DB().Where(
		"webhook_id = ? AND active = ?",
		webhookID, true,
	).Find(&targets).Error

	return targets, err
}

// lookupEntrypoint finds an entrypoint by UUID path.
func (h *Handlers) lookupEntrypoint(
	w http.ResponseWriter,
	r *http.Request,
	entrypointUUID string,
) (database.Entrypoint, bool) {
	var entrypoint database.Entrypoint

	result := h.db.DB().Where(
		"path = ?", entrypointUUID,
	).First(&entrypoint)
	if result.Error != nil {
		// The receiver is unauthenticated and /h/{uuid}
		// matches any single segment, so this value is entirely
		// client-chosen on exactly the branch where the lookup
		// failed. DEBUG is off by default; the cap is what keeps
		// turning it on from restoring an unbounded write.
		h.log.Debug(
			"entrypoint not found",
			"path", logfield.Truncate(
				entrypointUUID, logfield.MaxBytes,
			),
		)
		http.NotFound(w, r)

		return entrypoint, false
	}

	return entrypoint, true
}

// readWebhookBody reads and validates the request body size.
func (h *Handlers) readWebhookBody(
	w http.ResponseWriter,
	r *http.Request,
) ([]byte, bool) {
	body, err := io.ReadAll(
		io.LimitReader(r.Body, maxWebhookBodySize+1),
	)
	if err != nil {
		h.log.Error(
			"failed to read request body", "error", err,
		)
		http.Error(
			w, "Bad request", http.StatusBadRequest,
		)

		return nil, false
	}

	if len(body) > maxWebhookBodySize {
		http.Error(
			w,
			"Request body too large",
			http.StatusRequestEntityTooLarge,
		)

		return nil, false
	}

	return body, true
}

// createAndDeliverEvent stores the received event, fans it out to the
// webhook's targets, and answers the sender.
func (h *Handlers) createAndDeliverEvent(
	w http.ResponseWriter,
	r *http.Request,
	entrypoint database.Entrypoint,
	body, headersJSON []byte,
	targets []database.Target,
) {
	event, tasks, err := h.createAndFanOut(
		requestEventSource(r, entrypoint, headersJSON, body),
		targets,
	)
	if err != nil {
		h.serverError(w, "failed to store webhook event", err)

		return
	}

	h.finishWebhookResponse(w, event, entrypoint, tasks)
}

// eventSource carries the fields a new event is built from. The
// receiver fills it from the live request; the resubmit handler fills
// it from a stored event. Both then go through createAndFanOut, so an
// event is constructed and fanned out in one place however it entered
// the system.
type eventSource struct {
	WebhookID    string
	EntrypointID string
	Method       string
	HeadersJSON  string
	ContentType  string
	Body         []byte

	// ResubmittedFromID names the event this one copies. Only the
	// resubmit path sets it.
	ResubmittedFromID *string
}

// event builds the row this source stores.
func (s eventSource) event() *database.Event {
	return &database.Event{
		WebhookID:         s.WebhookID,
		EntrypointID:      s.EntrypointID,
		Method:            s.Method,
		Headers:           s.HeadersJSON,
		Body:              string(s.Body),
		ContentType:       s.ContentType,
		ResubmittedFromID: s.ResubmittedFromID,
	}
}

// requestEventSource describes the event a live receiver request
// stores.
func requestEventSource(
	r *http.Request,
	entrypoint database.Entrypoint,
	headersJSON, body []byte,
) eventSource {
	return eventSource{
		WebhookID:    entrypoint.WebhookID,
		EntrypointID: entrypoint.ID,
		Method:       r.Method,
		HeadersJSON:  string(headersJSON),
		ContentType:  r.Header.Get("Content-Type"),
		Body:         body,
	}
}

// createAndFanOut writes the event and one pending delivery per target
// in a single transaction, then hands the tasks to the delivery
// engine. It is the only path by which an event and its deliveries are
// created, so a resubmitted event is retried, SSRF-guarded and
// circuit-broken exactly as a received one is.
//
// The tasks are returned as well as queued, so a caller can report how
// many targets the event went to.
func (h *Handlers) createAndFanOut(
	src eventSource,
	targets []database.Target,
) (*database.Event, []delivery.Task, error) {
	webhookDB, err := h.dbMgr.GetDB(src.WebhookID)
	if err != nil {
		return nil, nil, fmt.Errorf(
			"getting webhook database: %w", err,
		)
	}

	tx := webhookDB.Begin()
	if tx.Error != nil {
		return nil, nil, fmt.Errorf(
			"beginning transaction: %w", tx.Error,
		)
	}

	event := src.event()

	err = tx.Create(event).Error
	if err != nil {
		tx.Rollback()

		return nil, nil, fmt.Errorf("creating event: %w", err)
	}

	tasks, err := buildDeliveryTasks(
		tx, event, targets, inlineBody(src.Body),
	)
	if err != nil {
		tx.Rollback()

		return nil, nil, err
	}

	err = tx.Commit().Error
	if err != nil {
		return nil, nil, fmt.Errorf(
			"committing transaction: %w", err,
		)
	}

	// Counted here, after the commit: an event exists once it is
	// durably stored, which is what the delivery counters are
	// compared against on a dashboard. A resubmitted event counts
	// too, because it produces deliveries that the delivery side
	// counts; the resubmit counter is what separates the two.
	h.mtr.EventReceived()

	if len(tasks) > 0 {
		h.notifier.Notify(tasks)
	}

	return event, tasks, nil
}

// inlineBody returns a pointer to body as a string if it fits
// within the inline size limit, or nil otherwise.
func inlineBody(body []byte) *string {
	if len(body) < delivery.MaxInlineBodySize {
		s := string(body)

		return &s
	}

	return nil
}

// finishWebhookResponse logs the event and writes the HTTP response.
func (h *Handlers) finishWebhookResponse(
	w http.ResponseWriter,
	event *database.Event,
	entrypoint database.Entrypoint,
	tasks []delivery.Task,
) {
	h.log.Info("webhook event created",
		"event_id", event.ID,
		"webhook_id", entrypoint.WebhookID,
		"entrypoint_id", entrypoint.ID,
		"target_count", len(tasks),
	)

	w.WriteHeader(http.StatusOK)

	_, err := w.Write([]byte(`{"status":"ok"}`))
	if err != nil {
		h.log.Error(
			"failed to write response", "error", err,
		)
	}
}

// buildDeliveryTasks creates one pending delivery per target in the
// transaction and returns the tasks for the delivery engine. The
// caller owns the transaction and rolls it back on error.
func buildDeliveryTasks(
	tx *gorm.DB,
	event *database.Event,
	targets []database.Target,
	bodyPtr *string,
) ([]delivery.Task, error) {
	tasks := make([]delivery.Task, 0, len(targets))

	for i := range targets {
		dlv := &database.Delivery{
			EventID:  event.ID,
			TargetID: targets[i].ID,
			Status:   database.DeliveryStatusPending,
		}

		err := tx.Create(dlv).Error
		if err != nil {
			return nil, fmt.Errorf(
				"creating delivery for target %s: %w",
				targets[i].ID, err,
			)
		}

		tasks = append(tasks, delivery.Task{
			DeliveryID:   dlv.ID,
			EventID:      event.ID,
			WebhookID:    event.WebhookID,
			EntrypointID: event.EntrypointID,
			TargetID:     targets[i].ID,
			TargetName:   targets[i].Name,
			TargetType:   targets[i].Type,
			TargetConfig: targets[i].Config,
			MaxRetries:   targets[i].MaxRetries,
			Method:       event.Method,
			Headers:      event.Headers,
			ContentType:  event.ContentType,
			Body:         bodyPtr,
			AttemptNum:   1,
		})
	}

	return tasks, nil
}