Add stub webhook endpoint for Event Grid

In order to receive notifications that jobs have finished (portfolio processed, report created), we're going to use event grid. The async job will send a message on completion, and it'll be received by our main API server.

This PR just adds a stub `/events/processed_portfolio` endpoint that does some logging, and middleware for validation, as described in https://learn.microsoft.com/en-us/azure/event-grid/webhook-event-delivery#validation-details

azure/azevents/BUILD.bazel

@@ -0,0 +1,12 @@
+load("@io_bazel_rules_go//go:def.bzl", "go_library")
+
+go_library(
+    name = "azevents",
+    srcs = ["azevents.go"],
+    importpath = "github.com/RMI/pacta/azure/azevents",
+    visibility = ["//visibility:public"],
+    deps = [
+        "@com_github_go_chi_chi_v5//:chi",
+        "@org_uber_go_zap//:zap",
+    ],
+)

azure/azevents/azevents.go

@@ -0,0 +1,108 @@
+// Package azevents handles incoming webhook events from Azure Event Grid. For
+// more info on the verification/validation logic, see
+// https://learn.microsoft.com/en-us/azure/event-grid/webhook-event-delivery#validation-details
+package azevents
+
+import (
+	"encoding/json"
+	"errors"
+	"fmt"
+	"io"
+	"net/http"
+	"time"
+
+	"github.com/go-chi/chi/v5"
+	"go.uber.org/zap"
+)
+
+type Config struct {
+	Logger *zap.Logger
+}
+
+func (c *Config) validate() error {
+	if c.Logger == nil {
+		return errors.New("no logger was given")
+	}
+	return nil
+}
+
+// Server handles both validating the Event Grid subscription and handling incoming events.
+type Server struct {
+	logger *zap.Logger
+}
+
+func NewServer(cfg *Config) (*Server, error) {
+	if err := cfg.validate(); err != nil {
+		return nil, fmt.Errorf("invalid config given: %w", err)
+	}
+
+	return &Server{
+		logger: cfg.Logger,
+	}, nil
+}
+
+func (s *Server) verifyWebhook(next http.Handler) http.Handler {
+	return http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+		if r.Header.Get("aeg-event-type") != "SubscriptionValidation" {
+			next.ServeHTTP(w, r)
+			return
+		}
+
+		// If we're here, we're validating to Azure that we own this endpoint and want
+		// to accept webhook calls here. Only calls to valid webhook endpoints will
+		// trigger this middleware (the rest will just get 404s), so if we've made it
+		// this far, validate that yes, we'll take webhook invocations.
+		var reqs []struct {
+			Id              string    `json:"id"`
+			Topic           string    `json:"topic"`
+			Subject         string    `json:"subject"`
+			EventType       string    `json:"eventType"`
+			EventTime       time.Time `json:"eventTime"`
+			MetadataVersion string    `json:"metadataVersion"`
+			DataVersion     string    `json:"dataVersion"`
+			Data            *struct {
+				ValidationCode string `json:"validationCode"`
+				ValidationUrl  string `json:"validationUrl"`
+			} `json:"data"`
+		}
+		if err := json.NewDecoder(r.Body).Decode(&reqs); err != nil {
+			s.logger.Error("failed to decode subscription validation request", zap.Error(err))
+			http.Error(w, http.StatusText(http.StatusBadRequest), http.StatusBadRequest)
+			return
+		}
+
+		if len(reqs) != 1 {
+			s.logger.Error("unexpected number of validation requests", zap.Any("reqs", reqs))
+			http.Error(w, http.StatusText(http.StatusBadRequest), http.StatusBadRequest)
+			return
+		}
+		req := reqs[0]
+		if req.Data == nil {
+			s.logger.Error("no data provided in validation request", zap.Any("req", req))
+			http.Error(w, http.StatusText(http.StatusBadRequest), http.StatusBadRequest)
+			return
+		}
+
+		s.logger.Info("received SubscriptionValidation request", zap.Any("req", req))
+
+		resp := struct {
+			ValidationResponse string `json:"validationResponse"`
+		}{req.Data.ValidationCode}
+		if err := json.NewEncoder(w).Encode(resp); err != nil {
+			s.logger.Error("failed to encode JSON validation response", zap.Error(err))
+		}
+	})
+}
+
+func (s *Server) RegisterHandlers(r chi.Router) {
+	r.Use(s.verifyWebhook)
+	r.Post("/events/processed_portfolio", func(w http.ResponseWriter, r *http.Request) {
+		dat, err := io.ReadAll(r.Body)
+		if err != nil {
+			s.logger.Error("failed to read webhook request body", zap.Error(err))
+			http.Error(w, http.StatusText(http.StatusInternalServerError), http.StatusInternalServerError)
+			return
+		}
+		s.logger.Info("processed porfolio", zap.String("portfolio_data", string(dat)))
+	})
+}

cmd/server/BUILD.bazel

@@ -9,6 +9,7 @@ go_library(
     visibility = ["//visibility:private"],
     deps = [
         "//azure/azblob",
+        "//azure/azevents",
         "//azure/aztask",
         "//cmd/runner/taskrunner",
         "//cmd/server/pactasrv",

cmd/server/main.go

@@ -14,6 +14,7 @@ import (
 	"github.com/Azure/azure-sdk-for-go/sdk/azcore"
 	"github.com/Azure/azure-sdk-for-go/sdk/azidentity"
 	"github.com/RMI/pacta/azure/azblob"
+	"github.com/RMI/pacta/azure/azevents"
 	"github.com/RMI/pacta/azure/aztask"
 	"github.com/RMI/pacta/cmd/runner/taskrunner"
 	"github.com/RMI/pacta/cmd/server/pactasrv"
@@ -274,7 +275,15 @@ func run(args []string) error {
 		}),
 	})
 
+	eventSrv, err := azevents.NewServer(&azevents.Config{
+		Logger: logger,
+	})
+	if err != nil {
+		return fmt.Errorf("failed to init Azure Event Grid handler: %w", err)
+	}
+
 	r := chi.NewRouter()
+	r.Group(eventSrv.RegisterHandlers)
 
 	jwKey, err := jwk.FromRaw(sec.AuthVerificationKey.PublicKey)
 	if err != nil {