webhooker/internal/handlers/webhook.go

package handlers

import (
	"encoding/json"
	"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.
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 /webhook/{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 creates the event and delivery records
// then notifies the delivery engine.
func (h *Handlers) createAndDeliverEvent(
	w http.ResponseWriter,
	r *http.Request,
	entrypoint database.Entrypoint,
	body, headersJSON []byte,
	targets []database.Target,
) {
	tx, err := h.beginWebhookTx(w, entrypoint.WebhookID)
	if err != nil {
		return
	}

	event := h.buildEvent(r, entrypoint, headersJSON, body)

	err = tx.Create(event).Error
	if err != nil {
		tx.Rollback()
		h.serverError(w, "failed to create event", err)

		return
	}

	bodyPtr := inlineBody(body)

	tasks := h.buildDeliveryTasks(
		w, tx, event, entrypoint, targets, bodyPtr,
	)
	if tasks == nil {
		return
	}

	err = tx.Commit().Error
	if err != nil {
		h.serverError(w, "failed to commit transaction", err)

		return
	}

	// Counted here, after the commit: an event is received once it
	// is durably stored, which is what the delivery counters are
	// compared against on a dashboard.
	h.mtr.EventReceived()

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

// beginWebhookTx opens a transaction on the per-webhook DB.
func (h *Handlers) beginWebhookTx(
	w http.ResponseWriter,
	webhookID string,
) (*gorm.DB, error) {
	webhookDB, err := h.dbMgr.GetDB(webhookID)
	if err != nil {
		h.serverError(
			w, "failed to get webhook database", err,
		)

		return nil, err
	}

	tx := webhookDB.Begin()
	if tx.Error != nil {
		h.serverError(
			w, "failed to begin transaction", tx.Error,
		)

		return nil, tx.Error
	}

	return tx, 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 notifies the delivery engine, logs the
// event, and writes the HTTP response.
func (h *Handlers) finishWebhookResponse(
	w http.ResponseWriter,
	event *database.Event,
	entrypoint database.Entrypoint,
	tasks []delivery.Task,
) {
	if len(tasks) > 0 {
		h.notifier.Notify(tasks)
	}

	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,
		)
	}
}

// buildEvent creates a new Event struct from request data.
func (h *Handlers) buildEvent(
	r *http.Request,
	entrypoint database.Entrypoint,
	headersJSON, body []byte,
) *database.Event {
	return &database.Event{
		WebhookID:    entrypoint.WebhookID,
		EntrypointID: entrypoint.ID,
		Method:       r.Method,
		Headers:      string(headersJSON),
		Body:         string(body),
		ContentType:  r.Header.Get("Content-Type"),
	}
}

// buildDeliveryTasks creates delivery records in the
// transaction and returns tasks for the delivery engine.
// Returns nil if an error occurred.
func (h *Handlers) buildDeliveryTasks(
	w http.ResponseWriter,
	tx *gorm.DB,
	event *database.Event,
	entrypoint database.Entrypoint,
	targets []database.Target,
	bodyPtr *string,
) []delivery.Task {
	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 {
			tx.Rollback()
			h.log.Error(
				"failed to create delivery",
				"target_id", targets[i].ID,
				"error", err,
			)
			http.Error(
				w, "Internal server error",
				http.StatusInternalServerError,
			)

			return nil
		}

		tasks = append(tasks, delivery.Task{
			DeliveryID:   dlv.ID,
			EventID:      event.ID,
			WebhookID:    entrypoint.WebhookID,
			EntrypointID: entrypoint.ID,
			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
}