feat(eventsources/bitbucketserver): add OneEventPerChange config opti…

…on for webhook event handling

Add OneEventPerChange config option to control whether to process each change in a repo:refs_changed webhook event as a separate event. This allows independent processing of each tag or reference update in a single webhook event useful for triggering distinct workflows in Argo Workflows.

Signed-off-by: Ryan Currah <ryan@currah.ca>

api/jsonschema/schema.json

@@ -1055,7 +1055,7 @@
       "type": "object"
     },
     "io.argoproj.eventsource.v1alpha1.BitbucketBasicAuth": {
-      "description": "BasicAuth holds the information required to authenticate user via basic auth mechanism",
+      "description": "BitbucketBasicAuth holds the information required to authenticate user via basic auth mechanism",
       "properties": {
         "password": {
           "$ref": "#/definitions/io.k8s.api.core.v1.SecretKeySelector",
@@ -1186,6 +1186,10 @@
           "description": "Metadata holds the user defined metadata which will passed along the event payload.",
           "type": "object"
         },
+        "oneEventPerChange": {
+          "description": "OneEventPerChange controls whether to process each change in a repo:refs_changed webhook event as a separate event. This setting is useful when multiple tags are pushed simultaneously for the same commit, and each tag needs to independently trigger an action, such as a distinct workflow in Argo Workflows. When enabled, the BitbucketServerEventSource publishes an individual BitbucketServerEventData for each change, ensuring independent processing of each tag or reference update in a single webhook event.",
+          "type": "boolean"
+        },
         "projectKey": {
           "description": "DeprecatedProjectKey is the key of project for which integration needs to set up. Deprecated: use Repositories instead. Will be unsupported in v1.8.",
           "type": "string"

eventsources/sources/bitbucketserver/bitbucketserverclients.go

@@ -0,0 +1,148 @@
+package bitbucketserver
+
+import (
+	"context"
+	"fmt"
+	"net/http"
+
+	"github.com/argoproj/argo-events/common"
+	"github.com/argoproj/argo-events/pkg/apis/eventsource/v1alpha1"
+	bitbucketv1 "github.com/gfleury/go-bitbucket-v1"
+	"github.com/mitchellh/mapstructure"
+)
+
+func newBitbucketServerClientCfg(bitbucketserverEventSource *v1alpha1.BitbucketServerEventSource) (*bitbucketv1.Configuration, error) {
+	bitbucketCfg := bitbucketv1.NewConfiguration(bitbucketserverEventSource.BitbucketServerBaseURL)
+	bitbucketCfg.AddDefaultHeader("x-atlassian-token", "no-check")
+	bitbucketCfg.AddDefaultHeader("x-requested-with", "XMLHttpRequest")
+	bitbucketCfg.HTTPClient = &http.Client{}
+
+	if bitbucketserverEventSource.TLS != nil {
+		tlsConfig, err := common.GetTLSConfig(bitbucketserverEventSource.TLS)
+		if err != nil {
+			return nil, fmt.Errorf("failed to get the tls configuration: %w", err)
+		}
+
+		bitbucketCfg.HTTPClient.Transport = &http.Transport{
+			TLSClientConfig: tlsConfig,
+		}
+	}
+
+	return bitbucketCfg, nil
+}
+
+func newBitbucketServerClient(ctx context.Context, bitbucketConfig *bitbucketv1.Configuration, bitbucketToken string) *bitbucketv1.APIClient {
+	ctx = context.WithValue(ctx, bitbucketv1.ContextAccessToken, bitbucketToken)
+	return bitbucketv1.NewAPIClient(ctx, bitbucketConfig)
+}
+
+type bitbucketServerReposPager struct {
+	Size          int                      `json:"size"`
+	Limit         int                      `json:"limit"`
+	Start         int                      `json:"start"`
+	NextPageStart int                      `json:"nextPageStart"`
+	IsLastPage    bool                     `json:"isLastPage"`
+	Values        []bitbucketv1.Repository `json:"values"`
+}
+
+// getProjectRepositories returns all the Bitbucket Server repositories in the provided projects.
+func getProjectRepositories(client *bitbucketv1.APIClient, projects []string) ([]v1alpha1.BitbucketServerRepository, error) {
+	var bitbucketRepos []bitbucketv1.Repository
+	for _, project := range projects {
+		paginationOptions := map[string]interface{}{"start": 0, "limit": 500}
+		for {
+			response, err := client.DefaultApi.GetRepositoriesWithOptions(project, paginationOptions)
+			if err != nil {
+				return nil, fmt.Errorf("unable to list repositories for project %s: %w", project, err)
+			}
+
+			var reposPager bitbucketServerReposPager
+			err = mapstructure.Decode(response.Values, &reposPager)
+			if err != nil {
+				return nil, fmt.Errorf("unable to decode repositories for project %s: %w", project, err)
+			}
+
+			bitbucketRepos = append(bitbucketRepos, reposPager.Values...)
+
+			if reposPager.IsLastPage {
+				break
+			}
+
+			paginationOptions["start"] = reposPager.NextPageStart
+		}
+	}
+
+	var repositories []v1alpha1.BitbucketServerRepository
+	for n := range bitbucketRepos {
+		repositories = append(repositories, v1alpha1.BitbucketServerRepository{
+			ProjectKey:     bitbucketRepos[n].Project.Key,
+			RepositorySlug: bitbucketRepos[n].Slug,
+		})
+	}
+
+	return repositories, nil
+}
+
+type refsChangedWebhookEvent struct {
+	EventKey string `json:"eventKey"`
+	Date     string `json:"date"`
+	Actor    struct {
+		Name         string `json:"name"`
+		EmailAddress string `json:"emailAddress"`
+		ID           int    `json:"id"`
+		DisplayName  string `json:"displayName"`
+		Active       bool   `json:"active"`
+		Slug         string `json:"slug"`
+		Type         string `json:"type"`
+		Links        struct {
+			Self []struct {
+				Href string `json:"href"`
+			} `json:"self"`
+		} `json:"links"`
+	} `json:"actor"`
+	Repository struct {
+		Slug          string `json:"slug"`
+		ID            int    `json:"id"`
+		Name          string `json:"name"`
+		HierarchyID   string `json:"hierarchyId"`
+		ScmID         string `json:"scmId"`
+		State         string `json:"state"`
+		StatusMessage string `json:"statusMessage"`
+		Forkable      bool   `json:"forkable"`
+		Project       struct {
+			Key    string `json:"key"`
+			ID     int    `json:"id"`
+			Name   string `json:"name"`
+			Public bool   `json:"public"`
+			Type   string `json:"type"`
+			Links  struct {
+				Self []struct {
+					Href string `json:"href"`
+				} `json:"self"`
+			} `json:"links"`
+		} `json:"project"`
+		Public bool `json:"public"`
+		Links  struct {
+			Clone []struct {
+				Href string `json:"href"`
+				Name string `json:"name"`
+			} `json:"clone"`
+			Self []struct {
+				Href string `json:"href"`
+			} `json:"self"`
+		} `json:"links"`
+	} `json:"repository"`
+	Changes []refsChangedWebHookEventChange `json:"changes"`
+}
+
+type refsChangedWebHookEventChange struct {
+	Ref struct {
+		ID        string `json:"id"`
+		DisplayID string `json:"displayId"`
+		Type      string `json:"type"`
+	} `json:"ref"`
+	RefID    string `json:"refId"`
+	FromHash string `json:"fromHash"`
+	ToHash   string `json:"toHash"`
+	Type     string `json:"type"`
+}

eventsources/sources/bitbucketserver/custombitbucketserverclient.go

@@ -0,0 +1,117 @@
+package bitbucketserver
+
+import (
+	"context"
+	"encoding/json"
+	"fmt"
+	"io"
+	"net/http"
+	"net/url"
+	"strconv"
+)
+
+// customBitbucketServerClient returns a Bitbucket Server HTTP client that implements methods that gfleury/go-bitbucket-v1 does not.
+// Specifically getting Pull Requests associated to a commit is not supported by gfleury/go-bitbucket-v1.
+type customBitbucketServerClient struct {
+	client *http.Client
+	ctx    context.Context
+	token  string
+	url    *url.URL
+}
+
+// pullRequestRes is a struct containing information about the Pull Request.
+type pullRequestRes struct {
+	ID    int    `json:"id"`
+	State string `json:"state"`
+}
+
+// pagedPullRequestsRes is a paged response with values of pullRequestRes.
+type pagedPullRequestsRes struct {
+	Size          int              `json:"size"`
+	Limit         int              `json:"limit"`
+	IsLastPage    bool             `json:"isLastPage"`
+	Values        []pullRequestRes `json:"values"`
+	Start         int              `json:"start"`
+	NextPageStart int              `json:"nextPageStart"`
+}
+
+type pagination struct {
+	Start int
+	Limit int
+}
+
+func (p *pagination) StartStr() string {
+	return strconv.Itoa(p.Start)
+}
+
+func (p *pagination) LimitStr() string {
+	return strconv.Itoa(p.Limit)
+}
+
+func (c *customBitbucketServerClient) authHeader(req *http.Request) {
+	req.Header.Set("Authorization", fmt.Sprintf("Bearer %s", c.token))
+}
+
+func (c *customBitbucketServerClient) get(u string) ([]byte, error) {
+	req, err := http.NewRequestWithContext(c.ctx, http.MethodGet, u, nil)
+	if err != nil {
+		return nil, err
+	}
+
+	c.authHeader(req)
+
+	res, err := c.client.Do(req)
+	if err != nil {
+		return nil, err
+	}
+	defer func() {
+		_ = res.Body.Close()
+	}()
+
+	if res.StatusCode > 299 {
+		resBody, readErr := io.ReadAll(res.Body)
+		if readErr != nil {
+			return nil, readErr
+		}
+
+		return nil, fmt.Errorf("calling endpoint '%s' failed: status %d: response body: %s", u, res.StatusCode, resBody)
+	}
+
+	return io.ReadAll(res.Body)
+}
+
+// GetCommitPullRequests returns all the Pull Requests associated to the commit id.
+func (c *customBitbucketServerClient) GetCommitPullRequests(project, repository, commit string) ([]pullRequestRes, error) {
+	p := pagination{Start: 0, Limit: 500}
+
+	commitsURL := c.url.JoinPath(fmt.Sprintf("api/1.0/projects/%s/repos/%s/commits/%s/pull-requests", project, repository, commit))
+	query := commitsURL.Query()
+	query.Set("limit", p.LimitStr())
+
+	var pullRequests []pullRequestRes
+	for {
+		query.Set("start", p.StartStr())
+		commitsURL.RawQuery = query.Encode()
+
+		body, err := c.get(commitsURL.String())
+		if err != nil {
+			return nil, err
+		}
+
+		var pagedPullRequests pagedPullRequestsRes
+		err = json.Unmarshal(body, &pagedPullRequests)
+		if err != nil {
+			return nil, err
+		}
+
+		pullRequests = append(pullRequests, pagedPullRequests.Values...)
+
+		if pagedPullRequests.IsLastPage {
+			break
+		}
+
+		p.Start = pagedPullRequests.NextPageStart
+	}
+
+	return pullRequests, nil
+}