Add async third-party vetting

Queue vetting on third_parties with PENDING, PROCESSING, COMPLETED, and FAILED states. Expose enqueue and status through GraphQL, MCP, CLI, and n8n, validate vet requests, tune the worker via config, and poll the detail page while vetting runs. Signed-off-by: Sacha Al Himdani <sacha@getprobo.com>
2026-05-27 13:34:49 +02:00
parent 1a71d15bc5
commit 6e7c96732f
59 changed files with 3251 additions and 520 deletions
--- a/pkg/thirdparty/service.go
+++ b/pkg/thirdparty/service.go
@@ -26,14 +26,20 @@ import (
 )

 type Service struct {
-	pg   *pg.Client
-	file *file.Service
+	pg             *pg.Client
+	file           *file.Service
+	vetter         Vetter
+	vettingEnabled bool
 }

-func NewService(pgClient *pg.Client, fileSvc *file.Service) *Service {
+func NewService(pgClient *pg.Client, fileSvc *file.Service, vetter Vetter) *Service {
+	_, disabled := vetter.(DisabledVetter)
+
 	return &Service{
-		pg:   pgClient,
-		file: fileSvc,
+		pg:             pgClient,
+		file:           fileSvc,
+		vetter:         vetter,
+		vettingEnabled: !disabled,
 	}
 }

--- a/pkg/thirdparty/vetting.go
+++ b/pkg/thirdparty/vetting.go
@@ -0,0 +1,171 @@
+// Copyright (c) 2026 Probo Inc <hello@getprobo.com>.
+//
+// Permission to use, copy, modify, and/or distribute this software for any
+// purpose with or without fee is hereby granted, provided that the above
+// copyright notice and this permission notice appear in all copies.
+//
+// THE SOFTWARE IS PROVIDED "AS IS" AND THE AUTHOR DISCLAIMS ALL WARRANTIES WITH
+// REGARD TO THIS SOFTWARE INCLUDING ALL IMPLIED WARRANTIES OF MERCHANTABILITY
+// AND FITNESS. IN NO EVENT SHALL THE AUTHOR BE LIABLE FOR ANY SPECIAL, DIRECT,
+// INDIRECT, OR CONSEQUENTIAL DAMAGES OR ANY DAMAGES WHATSOEVER RESULTING FROM
+// LOSS OF USE, DATA OR PROFITS, WHETHER IN AN ACTION OF CONTRACT, NEGLIGENCE OR
+// OTHER TORTIOUS ACTION, ARISING OUT OF OR IN CONNECTION WITH THE USE OR
+// PERFORMANCE OF THIS SOFTWARE.
+
+package thirdparty
+
+import (
+	"context"
+	"errors"
+	"fmt"
+	"time"
+	"unicode/utf8"
+
+	"go.gearno.de/kit/pg"
+	"go.probo.inc/probo/pkg/agent"
+	"go.probo.inc/probo/pkg/coredata"
+	"go.probo.inc/probo/pkg/gid"
+	"go.probo.inc/probo/pkg/validator"
+	"go.probo.inc/probo/pkg/vetting"
+)
+
+const (
+	vettingErrorMessageMaxLen  = 512
+	vettingWebsiteURLMaxLength = 2048
+	vettingProcedureMaxLength  = 5000
+)
+
+var (
+	ErrVettingDisabled   = errors.New("thirdParty vetting is not configured on this deployment")
+	ErrVettingInProgress = errors.New("a vetting job is already in progress for this third party")
+)
+
+type (
+	Vetter interface {
+		Assess(
+			ctx context.Context,
+			websiteURL string,
+			procedure string,
+			reporter agent.ProgressReporter,
+			extraTools []agent.Tool,
+		) (*vetting.Result, error)
+	}
+
+	DisabledVetter struct{}
+
+	VetRequest struct {
+		ID         gid.GID
+		WebsiteURL string
+		Procedure  *string
+	}
+)
+
+var _ Vetter = DisabledVetter{}
+
+func (DisabledVetter) Assess(
+	_ context.Context,
+	_ string,
+	_ string,
+	_ agent.ProgressReporter,
+	_ []agent.Tool,
+) (*vetting.Result, error) {
+	return nil, ErrVettingDisabled
+}
+
+func (req VetRequest) Validate() error {
+	v := validator.New()
+
+	v.Check(req.ID, "id", validator.Required(), validator.GID(coredata.ThirdPartyEntityType))
+	v.Check(req.WebsiteURL, "website_url", validator.Required(), validator.SafeText(vettingWebsiteURLMaxLength))
+	v.Check(req.Procedure, "procedure", validator.SafeText(vettingProcedureMaxLength))
+
+	return v.Error()
+}
+
+func sanitizeVettingError(err error) string {
+	msg := err.Error()
+	if len(msg) <= vettingErrorMessageMaxLen {
+		return msg
+	}
+
+	cut := vettingErrorMessageMaxLen
+	for cut > 0 && !utf8.RuneStart(msg[cut]) {
+		cut--
+	}
+
+	return msg[:cut] + "…"
+}
+
+func (s *Service) Vet(
+	ctx context.Context,
+	scope coredata.Scoper,
+	req VetRequest,
+) (*coredata.ThirdParty, error) {
+	if err := req.Validate(); err != nil {
+		return nil, err
+	}
+
+	if !s.vettingEnabled {
+		return nil, ErrVettingDisabled
+	}
+
+	thirdParty := &coredata.ThirdParty{}
+
+	err := s.pg.WithTx(
+		ctx,
+		func(ctx context.Context, conn pg.Tx) error {
+			if err := thirdParty.LoadByIDForUpdate(ctx, conn, scope, req.ID); err != nil {
+				return fmt.Errorf("cannot load thirdParty %q: %w", req.ID, err)
+			}
+
+			if thirdParty.VettingStatus != nil && thirdParty.VettingStatus.IsActive() {
+				return ErrVettingInProgress
+			}
+
+			pending := coredata.ThirdPartyVettingStatusPending
+			websiteURL := req.WebsiteURL
+
+			thirdParty.VettingStatus = &pending
+			thirdParty.VettingWebsiteURL = &websiteURL
+			thirdParty.VettingProcedure = req.Procedure
+			thirdParty.VettingProcessingStartedAt = nil
+			thirdParty.VettingErrorMessage = nil
+			thirdParty.UpdatedAt = time.Now()
+
+			if err := thirdParty.Update(ctx, conn, scope); err != nil {
+				return fmt.Errorf("cannot enqueue vetting: %w", err)
+			}
+
+			return nil
+		},
+	)
+	if err != nil {
+		return nil, err
+	}
+
+	return thirdParty, nil
+}
+
+func (s *Service) VettingStatus(
+	ctx context.Context,
+	scope coredata.Scoper,
+	thirdPartyID gid.GID,
+) (*coredata.ThirdPartyVettingStatus, error) {
+	thirdParty := &coredata.ThirdParty{}
+
+	err := s.pg.WithConn(
+		ctx,
+		func(ctx context.Context, conn pg.Querier) error {
+			return thirdParty.LoadByID(ctx, conn, scope, thirdPartyID)
+		},
+	)
+	if err != nil {
+		return nil, err
+	}
+
+	if thirdParty.VettingStatus == nil {
+		return nil, nil
+	}
+
+	return thirdParty.VettingStatus, nil
+}
--- a/pkg/thirdparty/vetting_test.go
+++ b/pkg/thirdparty/vetting_test.go
@@ -0,0 +1,119 @@
+// Copyright (c) 2026 Probo Inc <hello@getprobo.com>.
+//
+// Permission to use, copy, modify, and/or distribute this software for any
+// purpose with or without fee is hereby granted, provided that the above
+// copyright notice and this permission notice appear in all copies.
+//
+// THE SOFTWARE IS PROVIDED "AS IS" AND THE AUTHOR DISCLAIMS ALL WARRANTIES WITH
+// REGARD TO THIS SOFTWARE INCLUDING ALL IMPLIED WARRANTIES OF MERCHANTABILITY
+// AND FITNESS. IN NO EVENT SHALL THE AUTHOR BE LIABLE FOR ANY SPECIAL, DIRECT,
+// INDIRECT, OR CONSEQUENTIAL DAMAGES OR ANY DAMAGES WHATSOEVER RESULTING FROM
+// LOSS OF USE, DATA OR PROFITS, WHETHER IN AN ACTION OF CONTRACT, NEGLIGENCE OR
+// OTHER TORTIOUS ACTION, ARISING OUT OF OR IN CONNECTION WITH THE USE OR
+// PERFORMANCE OF THIS SOFTWARE.
+
+package thirdparty
+
+import (
+	"context"
+	"errors"
+	"strings"
+	"testing"
+
+	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/require"
+	"go.probo.inc/probo/pkg/coredata"
+	"go.probo.inc/probo/pkg/gid"
+	"go.probo.inc/probo/pkg/validator"
+)
+
+func TestVetRequest_Validate(t *testing.T) {
+	t.Parallel()
+
+	validID := gid.New(gid.NewTenantID(), coredata.ThirdPartyEntityType)
+
+	t.Run("accepts a valid request", func(t *testing.T) {
+		t.Parallel()
+
+		procedure := "Focus on SOC 2"
+
+		err := VetRequest{
+			ID:         validID,
+			WebsiteURL: "https://example.com",
+			Procedure:  &procedure,
+		}.Validate()
+		require.NoError(t, err)
+	})
+
+	t.Run("requires id", func(t *testing.T) {
+		t.Parallel()
+
+		err := VetRequest{
+			WebsiteURL: "https://example.com",
+		}.Validate()
+		require.Error(t, err)
+
+		validationErrors, ok := errors.AsType[validator.ValidationErrors](err)
+		require.True(t, ok)
+		assert.NotEmpty(t, validationErrors.ByField("id"))
+	})
+
+	t.Run("requires website url", func(t *testing.T) {
+		t.Parallel()
+
+		err := VetRequest{ID: validID}.Validate()
+		require.Error(t, err)
+
+		validationErrors, ok := errors.AsType[validator.ValidationErrors](err)
+		require.True(t, ok)
+		assert.NotEmpty(t, validationErrors.ByField("website_url"))
+	})
+
+	t.Run("rejects an invalid third party id", func(t *testing.T) {
+		t.Parallel()
+
+		err := VetRequest{
+			ID:         gid.New(gid.NewTenantID(), coredata.OrganizationEntityType),
+			WebsiteURL: "https://example.com",
+		}.Validate()
+		require.Error(t, err)
+
+		validationErrors, ok := errors.AsType[validator.ValidationErrors](err)
+		require.True(t, ok)
+		assert.NotEmpty(t, validationErrors.ByField("id"))
+	})
+}
+
+func TestSanitizeVettingError(t *testing.T) {
+	t.Parallel()
+
+	t.Run("returns short messages unchanged", func(t *testing.T) {
+		t.Parallel()
+
+		assert.Equal(t, "cannot vet third party", sanitizeVettingError(errors.New("cannot vet third party")))
+	})
+
+	t.Run("truncates long messages on a rune boundary", func(t *testing.T) {
+		t.Parallel()
+
+		msg := strings.Repeat("x", vettingErrorMessageMaxLen+10)
+
+		sanitized := sanitizeVettingError(errors.New(msg))
+
+		assert.LessOrEqual(t, len(sanitized), vettingErrorMessageMaxLen+len("…"))
+		assert.True(t, strings.HasSuffix(sanitized, "…"))
+	})
+}
+
+func TestDisabledVetter_Assess(t *testing.T) {
+	t.Parallel()
+
+	_, err := DisabledVetter{}.Assess(context.Background(), "https://example.com", "", nil, nil)
+	require.ErrorIs(t, err, ErrVettingDisabled)
+}
+
+func TestDisabledVetter_ImplementsVetter(t *testing.T) {
+	t.Parallel()
+
+	var _ Vetter = DisabledVetter{}
+}
--- a/pkg/thirdparty/vetting_worker.go
+++ b/pkg/thirdparty/vetting_worker.go
@@ -0,0 +1,238 @@
+// Copyright (c) 2026 Probo Inc <hello@getprobo.com>.
+//
+// Permission to use, copy, modify, and/or distribute this software for any
+// purpose with or without fee is hereby granted, provided that the above
+// copyright notice and this permission notice appear in all copies.
+//
+// THE SOFTWARE IS PROVIDED "AS IS" AND THE AUTHOR DISCLAIMS ALL WARRANTIES WITH
+// REGARD TO THIS SOFTWARE INCLUDING ALL IMPLIED WARRANTIES OF MERCHANTABILITY
+// AND FITNESS. IN NO EVENT SHALL THE AUTHOR BE LIABLE FOR ANY SPECIAL, DIRECT,
+// INDIRECT, OR CONSEQUENTIAL DAMAGES OR ANY DAMAGES WHATSOEVER RESULTING FROM
+// LOSS OF USE, DATA OR PROFITS, WHETHER IN AN ACTION OF CONTRACT, NEGLIGENCE OR
+// OTHER TORTIOUS ACTION, ARISING OUT OF OR IN CONNECTION WITH THE USE OR
+// PERFORMANCE OF THIS SOFTWARE.
+
+package thirdparty
+
+import (
+	"context"
+	"errors"
+	"fmt"
+	"time"
+
+	"go.gearno.de/kit/log"
+	"go.gearno.de/kit/pg"
+	"go.gearno.de/kit/worker"
+	"go.probo.inc/probo/pkg/coredata"
+	"go.probo.inc/probo/pkg/gid"
+	"go.probo.inc/probo/pkg/vetting"
+)
+
+type (
+	vettingHandler struct {
+		pg         *pg.Client
+		vetter     Vetter
+		logger     *log.Logger
+		staleAfter time.Duration
+	}
+
+	VettingWorkerConfig struct {
+		StaleAfter time.Duration
+	}
+)
+
+var (
+	_ worker.Handler[coredata.ThirdParty] = (*vettingHandler)(nil)
+	_ worker.StaleRecoverer               = (*vettingHandler)(nil)
+)
+
+func NewVettingWorker(
+	pgClient *pg.Client,
+	vetter Vetter,
+	logger *log.Logger,
+	cfg VettingWorkerConfig,
+	opts ...worker.Option,
+) *worker.Worker[coredata.ThirdParty] {
+	staleAfter := cfg.StaleAfter
+	if staleAfter <= 0 {
+		staleAfter = 25 * time.Minute
+	}
+
+	h := &vettingHandler{
+		pg:         pgClient,
+		vetter:     vetter,
+		logger:     logger,
+		staleAfter: staleAfter,
+	}
+
+	return worker.New(
+		"vetting-worker",
+		h,
+		logger,
+		opts...,
+	)
+}
+
+func (h *vettingHandler) Claim(ctx context.Context) (coredata.ThirdParty, error) {
+	var thirdParty coredata.ThirdParty
+
+	if err := h.pg.WithTx(
+		ctx,
+		func(ctx context.Context, tx pg.Tx) error {
+			if err := thirdParty.LoadNextPendingVettingForUpdateSkipLocked(ctx, tx); err != nil {
+				return err
+			}
+
+			now := time.Now()
+			processing := coredata.ThirdPartyVettingStatusProcessing
+
+			thirdParty.VettingStatus = &processing
+			thirdParty.VettingProcessingStartedAt = &now
+			thirdParty.VettingErrorMessage = nil
+			thirdParty.UpdatedAt = now
+
+			if err := thirdParty.Update(ctx, tx, coredata.NewNoScope()); err != nil {
+				return fmt.Errorf("cannot update third party: %w", err)
+			}
+
+			return nil
+		},
+	); err != nil {
+		if errors.Is(err, coredata.ErrResourceNotFound) {
+			return coredata.ThirdParty{}, worker.ErrNoTask
+		}
+
+		return coredata.ThirdParty{}, err
+	}
+
+	return thirdParty, nil
+}
+
+func (h *vettingHandler) Process(ctx context.Context, thirdParty coredata.ThirdParty) error {
+	if err := h.processThirdParty(ctx, &thirdParty); err != nil {
+		h.logger.ErrorCtx(
+			ctx,
+			"vetting worker failure",
+			log.Error(err),
+			log.String("third_party_id", thirdParty.ID.String()),
+		)
+
+		if failErr := h.failThirdParty(ctx, &thirdParty, err); failErr != nil {
+			h.logger.ErrorCtx(ctx, "cannot mark third party vetting as failed", log.Error(failErr))
+		}
+
+		return err
+	}
+
+	return nil
+}
+
+func (h *vettingHandler) RecoverStale(ctx context.Context) error {
+	return h.pg.WithConn(
+		ctx,
+		func(ctx context.Context, conn pg.Querier) error {
+			if err := coredata.ResetStaleVettingProcessing(ctx, conn, h.staleAfter); err != nil {
+				return fmt.Errorf("cannot reset stale vetting processing: %w", err)
+			}
+
+			return nil
+		},
+	)
+}
+
+func (h *vettingHandler) processThirdParty(
+	ctx context.Context,
+	thirdParty *coredata.ThirdParty,
+) error {
+	if thirdParty.VettingWebsiteURL == nil {
+		return fmt.Errorf("third party %s has no vetting website URL", thirdParty.ID)
+	}
+
+	procedure := ""
+	if thirdParty.VettingProcedure != nil {
+		procedure = *thirdParty.VettingProcedure
+	}
+
+	pc := &vetting.PersistenceContext{
+		PG:             h.pg,
+		ThirdPartyID:   thirdParty.ID,
+		OrganizationID: thirdParty.OrganizationID,
+		WebsiteURL:     *thirdParty.VettingWebsiteURL,
+	}
+
+	// Assessment runs outside any database transaction. Persistence tools
+	// are not passed in so the agent cannot open DB transactions during
+	// the long LLM/browser phase; results are written afterward.
+	result, err := h.vetter.Assess(
+		ctx,
+		*thirdParty.VettingWebsiteURL,
+		procedure,
+		nil,
+		nil,
+	)
+	if err != nil {
+		return fmt.Errorf("cannot vet third party: %w", err)
+	}
+
+	if err := vetting.PersistAssessmentResult(ctx, pc, *result); err != nil {
+		return fmt.Errorf("cannot persist vetting results: %w", err)
+	}
+
+	return h.commitVettingOutcome(
+		ctx,
+		thirdParty.ID,
+		func(fresh *coredata.ThirdParty) {
+			completed := coredata.ThirdPartyVettingStatusCompleted
+
+			fresh.VettingStatus = &completed
+			fresh.VettingProcessingStartedAt = nil
+			fresh.VettingErrorMessage = nil
+		},
+	)
+}
+
+func (h *vettingHandler) failThirdParty(
+	ctx context.Context,
+	thirdParty *coredata.ThirdParty,
+	reason error,
+) error {
+	errMsg := sanitizeVettingError(reason)
+
+	return h.commitVettingOutcome(
+		ctx,
+		thirdParty.ID,
+		func(fresh *coredata.ThirdParty) {
+			failed := coredata.ThirdPartyVettingStatusFailed
+
+			fresh.VettingStatus = &failed
+			fresh.VettingProcessingStartedAt = nil
+			fresh.VettingErrorMessage = &errMsg
+		},
+	)
+}
+
+func (h *vettingHandler) commitVettingOutcome(
+	ctx context.Context,
+	thirdPartyID gid.GID,
+	apply func(*coredata.ThirdParty),
+) error {
+	return h.pg.WithTx(
+		ctx,
+		func(ctx context.Context, tx pg.Tx) error {
+			thirdParty := &coredata.ThirdParty{}
+
+			if err := thirdParty.LoadByID(ctx, tx, coredata.NewNoScope(), thirdPartyID); err != nil {
+				return fmt.Errorf("cannot reload third party: %w", err)
+			}
+
+			apply(thirdParty)
+			thirdParty.UpdatedAt = time.Now()
+
+			if err := thirdParty.Update(ctx, tx, coredata.NewNoScope()); err != nil {
+				return fmt.Errorf("cannot update third party: %w", err)
+			}
+
+			return nil
+		},
+	)
+}