Implement Gitlab event handling

internal/providers/gitlab/manager/manager.go

@@ -29,6 +29,7 @@ import (
 	"github.com/stacklok/minder/internal/config/server"
 	"github.com/stacklok/minder/internal/crypto"
 	"github.com/stacklok/minder/internal/db"
+	"github.com/stacklok/minder/internal/events"
 	"github.com/stacklok/minder/internal/providers/credentials"
 	"github.com/stacklok/minder/internal/providers/gitlab"
 	v1 "github.com/stacklok/minder/pkg/providers/v1"
@@ -41,6 +42,7 @@ type providerClassManager struct {
 	glpcfg        *server.GitLabConfig
 	webhookURL    string
 	parentContext context.Context
+	pub           events.Publisher
 
 	// secrets for the webhook. These are stored in the
 	// structure to allow efficient fetching. Rotation
@@ -51,7 +53,8 @@ type providerClassManager struct {
 
 // NewGitLabProviderClassManager creates a new provider class manager for the dockerhub provider
 func NewGitLabProviderClassManager(
-	ctx context.Context, crypteng crypto.Engine, store db.Store, cfg *server.GitLabConfig, wgCfg server.WebhookConfig,
+	ctx context.Context, crypteng crypto.Engine, store db.Store, pub events.Publisher,
+	cfg *server.GitLabConfig, wgCfg server.WebhookConfig,
 ) (*providerClassManager, error) {
 	webhookURLBase := wgCfg.ExternalWebhookURL
 	if webhookURLBase == "" {
@@ -80,6 +83,7 @@ func NewGitLabProviderClassManager(
 	return &providerClassManager{
 		store:                  store,
 		crypteng:               crypteng,
+		pub:                    pub,
 		glpcfg:                 cfg,
 		webhookURL:             webhookURL,
 		parentContext:          ctx,

internal/providers/gitlab/manager/webhook.go

@@ -22,6 +22,7 @@ import (
 
 	"github.com/google/uuid"
 	"github.com/rs/zerolog"
+	gitlablib "github.com/xanzy/go-gitlab"
 
 	"github.com/stacklok/minder/internal/providers/gitlab/webhooksecret"
 )
@@ -47,13 +48,30 @@ func (m *providerClassManager) GetWebhookHandler() http.Handler {
 			return
 		}
 
-		l.Debug().Msg("received webhook")
+		eventType := gitlablib.HookEventType(r)
+		if eventType == "" {
+			l.Error().Msg("missing X-Gitlab-Event header")
+			http.Error(w, "missing X-Gitlab-Event header", http.StatusBadRequest)
+			return
+		}
+
+		l = l.With().Str("event", string(eventType)).Logger()
+
+		disp := m.getWebhookEventDispatcher(eventType)
+
+		if err := disp(l, r); err != nil {
+			l.Error().Err(err).Msg("error handling webhook event")
+			http.Error(w, "error handling webhook event", http.StatusInternalServerError)
+			return
+		}
+
+		l.Debug().Msg("processed webhook event successfully")
 	})
 }
 
 func (m *providerClassManager) validateRequest(r *http.Request) error {
 	// Validate the webhook secret
-	gltok := r.Header.Get("X-Gitlab-Token")
+	gltok := gitlablib.HookEventToken(r)
 	if gltok == "" {
 		return errors.New("missing X-Gitlab-Token header")
 	}

internal/providers/gitlab/manager/webhook_handlers.go

@@ -0,0 +1,148 @@
+// Copyright 2024 Stacklok, Inc.
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+package manager
+
+import (
+	"encoding/json"
+	"fmt"
+	"io"
+	"net/http"
+
+	"github.com/ThreeDotsLabs/watermill/message"
+	"github.com/google/uuid"
+	"github.com/rs/zerolog"
+	gitlablib "github.com/xanzy/go-gitlab"
+
+	entmsg "github.com/stacklok/minder/internal/entities/handlers/message"
+	"github.com/stacklok/minder/internal/entities/properties"
+	"github.com/stacklok/minder/internal/events"
+	"github.com/stacklok/minder/internal/providers/gitlab"
+	minderv1 "github.com/stacklok/minder/pkg/api/protobuf/go/minder/v1"
+)
+
+const (
+	// MaxBytesLimit is the maximum number of bytes to read from the response body
+	// We limit to 1MB to prevent abuse
+	MaxBytesLimit int64 = 1 << 20
+)
+
+// getWebhookEventDispatcher returns the appropriate webhook event dispatcher for the given event type
+// It returns a function that is meant to do the actual handling of the event.
+// Note that we pass the request to the handler function, so we don't even try to
+// parse the request body here unless it's necessary.
+func (m *providerClassManager) getWebhookEventDispatcher(
+	eventType gitlablib.EventType,
+) func(l zerolog.Logger, r *http.Request) error {
+	//nolint:exhaustive // We only handle a subset of the possible events
+	switch eventType {
+	case gitlablib.EventTypePush:
+		return m.handleRepoPush
+	case gitlablib.EventTypeTagPush:
+		return m.handleTagPush
+	default:
+		return m.handleNoop
+	}
+}
+
+// handleNoop is a no-op handler for unhandled webhook events
+func (_ *providerClassManager) handleNoop(l zerolog.Logger, _ *http.Request) error {
+	l.Debug().Msg("unhandled webhook event")
+	return nil
+}
+
+func (m *providerClassManager) handleRepoPush(l zerolog.Logger, r *http.Request) error {
+	l.Debug().Msg("handling push event")
+
+	pushEvent := gitlablib.PushEvent{}
+	if err := decodeJSONSafe(r.Body, &pushEvent); err != nil {
+		l.Error().Err(err).Msg("error decoding push event")
+		return fmt.Errorf("error decoding push event: %w", err)
+	}
+
+	rawID := pushEvent.ProjectID
+	if rawID == 0 {
+		l.Error().Msg("push event missing project ID")
+		return fmt.Errorf("push event missing project ID")
+	}
+
+	return m.publishRefreshAndEvalForGitlabProject(l, rawID)
+}
+
+func (m *providerClassManager) handleTagPush(l zerolog.Logger, r *http.Request) error {
+	l.Debug().Msg("handling tag push event")
+
+	tagPushEvent := gitlablib.TagEvent{}
+	if err := decodeJSONSafe(r.Body, &tagPushEvent); err != nil {
+		l.Error().Err(err).Msg("error decoding tag push event")
+		return fmt.Errorf("error decoding tag push event: %w", err)
+	}
+
+	rawID := tagPushEvent.ProjectID
+	if rawID == 0 {
+		l.Error().Msg("tag push event missing project ID")
+		return fmt.Errorf("tag push event missing project ID")
+	}
+
+	return m.publishRefreshAndEvalForGitlabProject(l, rawID)
+}
+
+func (m *providerClassManager) publishRefreshAndEvalForGitlabProject(
+	l zerolog.Logger, rawProjectID int) error {
+	upstreamID := gitlab.FormatRepositoryUpstreamID(rawProjectID)
+
+	// Form identifying properties
+	identifyingProps, err := properties.NewProperties(map[string]any{
+		properties.PropertyUpstreamID: upstreamID,
+	})
+	if err != nil {
+		l.Error().Err(err).Msg("error creating identifying properties")
+		return fmt.Errorf("error creating identifying properties: %w", err)
+	}
+
+	// Form message to publish
+	outm := entmsg.NewEntityRefreshAndDoMessage()
+	outm.WithEntity(minderv1.Entity_ENTITY_REPOSITORIES, identifyingProps)
+	outm.WithProviderClassHint(gitlab.Class)
+
+	// Convert message for publishing
+	msgID := uuid.New().String()
+	msg := message.NewMessage(msgID, nil)
+	if err := outm.ToMessage(msg); err != nil {
+		l.Error().Err(err).Msg("error converting message to protobuf")
+		return fmt.Errorf("error converting message to protobuf: %w", err)
+	}
+
+	// Publish message
+	l.Debug().Str("msg_id", msgID).Msg("publishing refresh and eval message")
+	if err := m.pub.Publish(events.TopicQueueRefreshEntityAndEvaluate, msg); err != nil {
+		l.Error().Err(err).Msg("error publishing refresh and eval message")
+		return fmt.Errorf("error publishing refresh and eval message: %w", err)
+	}
+
+	return nil
+}
+
+func decodeJSONSafe[T any](r io.ReadCloser, v *T) error {
+	rs := wrapSafe(r)
+	defer r.Close()
+
+	dec := json.NewDecoder(rs)
+	return dec.Decode(v)
+}
+
+// wrapSafe wraps the io.Reader in a LimitReader to prevent abuse
+func wrapSafe(r io.Reader) io.Reader {
+	return io.LimitReader(r, MaxBytesLimit)
+}

internal/providers/gitlab/properties.go

@@ -95,6 +95,13 @@ func (c *gitlabClient) PropertiesToProtoMessage(
 	return repoV1FromProperties(props)
 }
 
+// FormatRepositoryUpstreamID returns the upstream ID for a gitlab project
+// This is done so we don't have to deal with conversions in the provider
+// when dealing with entities
+func FormatRepositoryUpstreamID(id int) string {
+	return fmt.Sprintf("%d", id)
+}
+
 func getStringProp(props *properties.Properties, key string) (string, error) {
 	value, err := props.GetProperty(key).AsString()
 	if err != nil {

internal/providers/gitlab/repository_properties.go

@@ -91,7 +91,7 @@ func gitlabProjectToProperties(proj *gitlab.Project) (*properties.Properties, er
 	}
 
 	outProps, err := properties.NewProperties(map[string]any{
-		properties.PropertyUpstreamID:     fmt.Sprintf("%d", proj.ID),
+		properties.PropertyUpstreamID:     FormatRepositoryUpstreamID(proj.ID),
 		properties.PropertyName:           formatRepoName(owner, proj.Name),
 		properties.RepoPropertyIsPrivate:  proj.Visibility == gitlab.PrivateVisibility,
 		properties.RepoPropertyIsArchived: proj.Archived,

internal/service/service.go

@@ -153,6 +153,7 @@ func AllInOneServerService(
 			ctx,
 			cryptoEngine,
 			store,
+			evt,
 			cfg.Provider.GitLab,
 			cfg.WebhookConfig,
 		)