Add access review background workers

Add SourceFetchWorker for campaign source fetching with bounded concurrency and SourceNameWorker for resolving provider instance names via OAuth connectors. Signed-off-by: Aurélien Sibiril <81782+aureliensibiril@users.noreply.github.com>
2026-04-02 11:51:12 +02:00
parent 5efd5136e2
commit 761840929f
2 changed files with 628 additions and 0 deletions
--- a/pkg/accessreview/source_name_worker.go
+++ b/pkg/accessreview/source_name_worker.go
@@ -0,0 +1,290 @@
+// 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 accessreview
+
+import (
+	"context"
+	"errors"
+	"fmt"
+	"net/http"
+	"time"
+
+	"go.gearno.de/kit/log"
+	"go.gearno.de/kit/pg"
+	"go.probo.inc/probo/pkg/accessreview/drivers"
+	"go.probo.inc/probo/pkg/connector"
+	"go.probo.inc/probo/pkg/coredata"
+	"go.probo.inc/probo/pkg/crypto/cipher"
+)
+
+// SourceNameWorker polls for access sources that have a connector but no
+// synced name, resolves the provider instance name, and updates the source.
+type SourceNameWorker struct {
+	pg                *pg.Client
+	encryptionKey     cipher.EncryptionKey
+	connectorRegistry *connector.ConnectorRegistry
+	logger            *log.Logger
+	interval          time.Duration
+}
+
+func NewSourceNameWorker(
+	pgClient *pg.Client,
+	encryptionKey cipher.EncryptionKey,
+	connectorRegistry *connector.ConnectorRegistry,
+	logger *log.Logger,
+) *SourceNameWorker {
+	return &SourceNameWorker{
+		pg:                pgClient,
+		encryptionKey:     encryptionKey,
+		connectorRegistry: connectorRegistry,
+		logger:            logger,
+		interval:          10 * time.Second,
+	}
+}
+
+func (w *SourceNameWorker) Run(ctx context.Context) error {
+	w.logger.InfoCtx(ctx, "source name worker started",
+		log.String("interval", w.interval.String()),
+	)
+
+	for {
+		select {
+		case <-ctx.Done():
+			w.logger.InfoCtx(context.WithoutCancel(ctx), "source name worker stopping")
+			return ctx.Err()
+		case <-time.After(w.interval):
+			nonCancelableCtx := context.WithoutCancel(ctx)
+			for {
+				if err := w.processNext(nonCancelableCtx); err != nil {
+					if !errors.Is(err, coredata.ErrNoAccessSourceNameSyncAvailable) {
+						w.logger.ErrorCtx(nonCancelableCtx, "cannot sync source name", log.Error(err))
+					}
+					break
+				}
+			}
+		}
+	}
+}
+
+func (w *SourceNameWorker) processNext(ctx context.Context) error {
+	var source coredata.AccessSource
+
+	err := w.pg.WithTx(
+		ctx,
+		func(tx pg.Conn) error {
+			return source.LoadNextUnsyncedNameForUpdateSkipLocked(ctx, tx)
+		},
+	)
+	if err != nil {
+		return err
+	}
+
+	w.logger.InfoCtx(ctx, "syncing source name",
+		log.String("source_id", source.ID.String()),
+		log.String("current_name", source.Name),
+	)
+
+	var (
+		dbConnector coredata.Connector
+		resolver    drivers.NameResolver
+	)
+
+	err = w.pg.WithConn(
+		ctx,
+		func(conn pg.Conn) error {
+			scope := coredata.NewScopeFromObjectID(source.ID)
+			if source.ConnectorID == nil {
+				return fmt.Errorf("source %s has no connector", source.ID)
+			}
+
+			if err := dbConnector.LoadByID(ctx, conn, scope, *source.ConnectorID, w.encryptionKey); err != nil {
+				return fmt.Errorf("cannot load connector %s: %w", *source.ConnectorID, err)
+			}
+
+			var tokenBefore string
+			if oauth2Conn, ok := dbConnector.Connection.(*connector.OAuth2Connection); ok {
+				tokenBefore = oauth2Conn.AccessToken
+			}
+
+			httpClient, err := w.connectorHTTPClient(ctx, &dbConnector)
+			if err != nil {
+				return fmt.Errorf("cannot create HTTP client for connector: %w", err)
+			}
+
+			if oauth2Conn, ok := dbConnector.Connection.(*connector.OAuth2Connection); ok {
+				if oauth2Conn.AccessToken != tokenBefore {
+					dbConnector.UpdatedAt = time.Now()
+					if err := dbConnector.Update(ctx, conn, scope, w.encryptionKey); err != nil {
+						return fmt.Errorf("cannot persist refreshed token for connector %s: %w", *source.ConnectorID, err)
+					}
+				}
+			}
+
+			resolver = w.buildResolver(&dbConnector, httpClient)
+			return nil
+		},
+	)
+	if err != nil {
+		w.logger.ErrorCtx(ctx, "cannot load connector for source name sync",
+			log.String("source_id", source.ID.String()),
+			log.Error(err),
+		)
+		return nil
+	}
+
+	if resolver == nil {
+		w.logger.InfoCtx(ctx, "no name resolver for provider, keeping generic name",
+			log.String("source_id", source.ID.String()),
+			log.String("provider", dbConnector.Provider.String()),
+		)
+		return w.markNameSynced(ctx, &source)
+	}
+
+	resolveCtx, cancel := context.WithTimeout(ctx, 10*time.Second)
+	defer cancel()
+
+	instanceName, err := resolver.ResolveInstanceName(resolveCtx)
+	if err != nil {
+		w.logger.ErrorCtx(ctx, "cannot resolve instance name",
+			log.String("source_id", source.ID.String()),
+			log.String("provider", dbConnector.Provider.String()),
+			log.Error(err),
+		)
+		return fmt.Errorf("cannot resolve instance name for source %s: %w", source.ID, err)
+	}
+
+	if instanceName == "" {
+		w.logger.InfoCtx(ctx, "instance name is empty, keeping generic name",
+			log.String("source_id", source.ID.String()),
+			log.String("provider", dbConnector.Provider.String()),
+		)
+		return w.markNameSynced(ctx, &source)
+	}
+
+	displayName := drivers.ProviderDisplayName(dbConnector.Provider)
+	newName := displayName + " " + instanceName
+
+	w.logger.InfoCtx(ctx, "resolved source name",
+		log.String("source_id", source.ID.String()),
+		log.String("old_name", source.Name),
+		log.String("new_name", newName),
+	)
+
+	source.Name = newName
+	return w.markNameSynced(ctx, &source)
+}
+
+func (w *SourceNameWorker) markNameSynced(
+	ctx context.Context,
+	source *coredata.AccessSource,
+) error {
+	return w.pg.WithTx(
+		ctx,
+		func(tx pg.Conn) error {
+			scope := coredata.NewScopeFromObjectID(source.ID)
+			now := time.Now()
+
+			source.NameSyncedAt = new(now)
+			source.UpdatedAt = now
+
+			if err := source.Update(ctx, tx, scope); err != nil {
+				return fmt.Errorf("cannot update access source: %w", err)
+			}
+
+			return nil
+		},
+	)
+}
+
+// connectorHTTPClient returns an HTTP client for the given connector.
+// For OAuth2 connections it uses RefreshableClient when a refresh config
+// is registered for the provider, so that short-lived tokens are
+// transparently refreshed.
+func (w *SourceNameWorker) connectorHTTPClient(
+	ctx context.Context,
+	dbConnector *coredata.Connector,
+) (*http.Client, error) {
+	oauth2Conn, ok := dbConnector.Connection.(*connector.OAuth2Connection)
+	if !ok {
+		return dbConnector.Connection.Client(ctx)
+	}
+
+	if w.connectorRegistry != nil {
+		refreshCfg := w.connectorRegistry.GetOAuth2RefreshConfig(string(dbConnector.Provider))
+		if refreshCfg != nil {
+			return oauth2Conn.RefreshableClient(ctx, *refreshCfg)
+		}
+	}
+
+	return oauth2Conn.Client(ctx)
+}
+
+func (w *SourceNameWorker) buildResolver(
+	dbConnector *coredata.Connector,
+	httpClient *http.Client,
+) drivers.NameResolver {
+	switch dbConnector.Provider {
+	case coredata.ConnectorProviderSlack:
+		return drivers.NewSlackNameResolver(httpClient)
+	case coredata.ConnectorProviderGoogleWorkspace:
+		return drivers.NewGoogleWorkspaceNameResolver(httpClient)
+	case coredata.ConnectorProviderLinear:
+		return drivers.NewLinearNameResolver(httpClient)
+	case coredata.ConnectorProviderCloudflare:
+		return drivers.NewCloudflareNameResolver(httpClient)
+	case coredata.ConnectorProviderBrex:
+		return drivers.NewBrexNameResolver(httpClient)
+	case coredata.ConnectorProviderTally:
+		tallySettings, err := dbConnector.TallySettings()
+		if err != nil {
+			w.logger.Error("cannot read tally connector settings", log.Error(err))
+			return nil
+		}
+		return drivers.NewTallyNameResolver(httpClient, tallySettings.OrganizationID)
+	case coredata.ConnectorProviderHubSpot:
+		return drivers.NewHubSpotNameResolver(httpClient)
+	case coredata.ConnectorProviderDocuSign:
+		return drivers.NewDocuSignNameResolver(httpClient)
+	case coredata.ConnectorProviderOpenAI:
+		return drivers.NewOpenAINameResolver(httpClient)
+	case coredata.ConnectorProviderSentry:
+		sentrySettings, err := dbConnector.SentrySettings()
+		if err != nil {
+			w.logger.Error("cannot read sentry connector settings", log.Error(err))
+			return nil
+		}
+		return drivers.NewSentryNameResolver(httpClient, sentrySettings.OrganizationSlug)
+	case coredata.ConnectorProviderGitHub:
+		githubSettings, err := dbConnector.GitHubSettings()
+		if err != nil {
+			w.logger.Error("cannot read github connector settings", log.Error(err))
+			return nil
+		}
+		return drivers.NewGitHubNameResolver(httpClient, githubSettings.Organization)
+	case coredata.ConnectorProviderSupabase:
+		supabaseSettings, err := dbConnector.SupabaseSettings()
+		if err != nil {
+			w.logger.Error("cannot read supabase connector settings", log.Error(err))
+			return nil
+		}
+		return drivers.NewSupabaseNameResolver(supabaseSettings.OrganizationSlug)
+	case coredata.ConnectorProviderIntercom:
+		return drivers.NewIntercomNameResolver(httpClient)
+	case coredata.ConnectorProviderResend:
+		return drivers.NewResendNameResolver()
+	default:
+		return nil
+	}
+}
--- a/pkg/accessreview/worker.go
+++ b/pkg/accessreview/worker.go
@@ -0,0 +1,338 @@
+// 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 accessreview
+
+import (
+	"context"
+	"errors"
+	"fmt"
+	"sync"
+	"time"
+
+	"go.gearno.de/kit/log"
+	"go.gearno.de/kit/pg"
+	"go.probo.inc/probo/pkg/coredata"
+	"go.probo.inc/probo/pkg/gid"
+)
+
+type (
+	SourceFetchWorker struct {
+		svc            *Service
+		pg             *pg.Client
+		logger         *log.Logger
+		interval       time.Duration
+		staleAfter     time.Duration
+		maxConcurrency int
+	}
+
+	SourceFetchWorkerOption func(*SourceFetchWorker)
+)
+
+func WithSourceFetchWorkerIntervalDuration(interval time.Duration) SourceFetchWorkerOption {
+	return func(w *SourceFetchWorker) {
+		w.interval = interval
+	}
+}
+
+func WithSourceFetchWorkerStaleAfter(staleAfter time.Duration) SourceFetchWorkerOption {
+	return func(w *SourceFetchWorker) {
+		w.staleAfter = staleAfter
+	}
+}
+
+func WithSourceFetchWorkerMaxConcurrency(maxConcurrency int) SourceFetchWorkerOption {
+	return func(w *SourceFetchWorker) {
+		w.maxConcurrency = maxConcurrency
+	}
+}
+
+func NewSourceFetchWorker(
+	svc *Service,
+	pgClient *pg.Client,
+	logger *log.Logger,
+	opts ...SourceFetchWorkerOption,
+) *SourceFetchWorker {
+	w := &SourceFetchWorker{
+		svc:            svc,
+		pg:             pgClient,
+		logger:         logger,
+		interval:       30 * time.Second,
+		staleAfter:     5 * time.Minute,
+		maxConcurrency: 20,
+	}
+
+	for _, opt := range opts {
+		opt(w)
+	}
+
+	return w
+}
+
+func (w *SourceFetchWorker) Run(ctx context.Context) error {
+	var (
+		wg     sync.WaitGroup
+		sem    = make(chan struct{}, w.maxConcurrency)
+		ticker = time.NewTicker(w.interval)
+	)
+	defer ticker.Stop()
+	defer wg.Wait()
+
+	for {
+		select {
+		case <-ctx.Done():
+			return ctx.Err()
+		case <-ticker.C:
+			nonCancelableCtx := context.WithoutCancel(ctx)
+			w.recoverStaleRows(nonCancelableCtx)
+			for {
+				if err := w.processNext(ctx, sem, &wg); err != nil {
+					if !errors.Is(err, coredata.ErrNoAccessReviewCampaignSourceFetchAvailable) {
+						w.logger.ErrorCtx(nonCancelableCtx, "cannot claim item", log.Error(err))
+					}
+					break
+				}
+			}
+		}
+	}
+}
+
+func (w *SourceFetchWorker) processNext(
+	ctx context.Context,
+	sem chan struct{},
+	wg *sync.WaitGroup,
+) error {
+	select {
+	case sem <- struct{}{}:
+	case <-ctx.Done():
+		return ctx.Err()
+	}
+
+	var (
+		sourceFetch      coredata.AccessReviewCampaignSourceFetch
+		now              = time.Now()
+		nonCancelableCtx = context.WithoutCancel(ctx)
+	)
+
+	if err := w.pg.WithTx(
+		nonCancelableCtx,
+		func(tx pg.Conn) error {
+			if err := sourceFetch.LoadNextQueuedForUpdateSkipLocked(nonCancelableCtx, tx); err != nil {
+				return err // sentinel errors checked by caller
+			}
+
+			sourceFetch.Status = coredata.AccessReviewCampaignSourceFetchStatusFetching
+			sourceFetch.AttemptCount++
+			sourceFetch.LastError = nil
+			sourceFetch.StartedAt = new(now)
+			sourceFetch.CompletedAt = nil
+			sourceFetch.UpdatedAt = now
+
+			scope := coredata.NewScope(sourceFetch.TenantID)
+			if err := sourceFetch.Update(nonCancelableCtx, tx, scope); err != nil {
+				return fmt.Errorf("cannot update source fetch status: %w", err)
+			}
+			return nil
+		},
+	); err != nil {
+		<-sem
+		return fmt.Errorf("cannot claim source fetch: %w", err)
+	}
+
+	wg.Add(1)
+	go func(sourceFetch coredata.AccessReviewCampaignSourceFetch) {
+		defer wg.Done()
+		defer func() { <-sem }()
+
+		if err := w.handle(nonCancelableCtx, &sourceFetch); err != nil {
+			w.logger.ErrorCtx(nonCancelableCtx, "cannot process source fetch", log.Error(err))
+		}
+	}(sourceFetch)
+
+	return nil
+}
+
+func (w *SourceFetchWorker) handle(
+	ctx context.Context,
+	sourceFetch *coredata.AccessReviewCampaignSourceFetch,
+) error {
+	scope := coredata.NewScope(sourceFetch.TenantID)
+
+	campaign, err := w.svc.Campaigns(scope).Get(ctx, sourceFetch.AccessReviewCampaignID)
+	if err != nil {
+		commitErr := w.commitFailedSourceFetch(
+			ctx,
+			sourceFetch,
+			fmt.Errorf("cannot load campaign: %w", err),
+		)
+		if commitErr != nil {
+			return fmt.Errorf("cannot load campaign: %w, and cannot commit failed source fetch: %w", err, commitErr)
+		}
+		return fmt.Errorf("cannot load campaign: %w", err)
+	}
+
+	count, err := w.svc.Engine(scope).FetchSource(ctx, campaign, sourceFetch.AccessSourceID)
+	if err != nil {
+		commitErr := w.commitFailedSourceFetch(ctx, sourceFetch, err)
+		if commitErr != nil {
+			return fmt.Errorf("cannot fetch source: %w, and cannot commit failed source fetch: %w", err, commitErr)
+		}
+
+		if finalizeErr := w.finalizeCampaignFetchLifecycle(ctx, sourceFetch.TenantID, sourceFetch.AccessReviewCampaignID); finalizeErr != nil {
+			return fmt.Errorf("cannot finalize campaign after failed source fetch: %w", finalizeErr)
+		}
+		return fmt.Errorf("cannot fetch source: %w", err)
+	}
+
+	if err := w.commitSuccessfulSourceFetch(ctx, sourceFetch, count); err != nil {
+		return fmt.Errorf("cannot commit successful source fetch: %w", err)
+	}
+
+	if err := w.finalizeCampaignFetchLifecycle(ctx, sourceFetch.TenantID, sourceFetch.AccessReviewCampaignID); err != nil {
+		return fmt.Errorf("cannot finalize campaign fetch lifecycle: %w", err)
+	}
+
+	return nil
+}
+
+func (w *SourceFetchWorker) recoverStaleRows(ctx context.Context) {
+	now := time.Now()
+	staleThreshold := now.Add(-w.staleAfter)
+
+	err := w.pg.WithTx(
+		ctx,
+		func(tx pg.Conn) error {
+			var fetches coredata.AccessReviewCampaignSourceFetches
+			count, err := fetches.RecoverStale(ctx, tx, staleThreshold, now)
+			if err != nil {
+				return fmt.Errorf("cannot recover stale source fetches: %w", err)
+			}
+
+			if count > 0 {
+				w.logger.InfoCtx(
+					ctx,
+					"recovered stale source fetches",
+					log.Int64("count", count),
+				)
+			}
+
+			return nil
+		},
+	)
+	if err != nil {
+		w.logger.ErrorCtx(ctx, "cannot recover stale rows", log.Error(err))
+	}
+}
+
+func (w *SourceFetchWorker) commitFailedSourceFetch(
+	ctx context.Context,
+	sourceFetch *coredata.AccessReviewCampaignSourceFetch,
+	failureErr error,
+) error {
+	var (
+		now    = time.Now()
+		errMsg = failureErr.Error()
+		scope  = coredata.NewScopeFromObjectID(sourceFetch.AccessReviewCampaignID)
+	)
+
+	sourceFetch.Status = coredata.AccessReviewCampaignSourceFetchStatusFailed
+	sourceFetch.LastError = &errMsg
+	sourceFetch.CompletedAt = new(now)
+	sourceFetch.UpdatedAt = now
+
+	return w.pg.WithConn(
+		ctx,
+		func(conn pg.Conn) error {
+			return sourceFetch.Update(ctx, conn, scope)
+		},
+	)
+}
+
+func (w *SourceFetchWorker) commitSuccessfulSourceFetch(
+	ctx context.Context,
+	sourceFetch *coredata.AccessReviewCampaignSourceFetch,
+	fetchedAccountsCount int,
+) error {
+	var (
+		now   = time.Now()
+		scope = coredata.NewScopeFromObjectID(sourceFetch.AccessReviewCampaignID)
+	)
+
+	sourceFetch.Status = coredata.AccessReviewCampaignSourceFetchStatusSuccess
+	sourceFetch.FetchedAccountsCount = fetchedAccountsCount
+	sourceFetch.LastError = nil
+	sourceFetch.CompletedAt = new(now)
+	sourceFetch.UpdatedAt = now
+
+	return w.pg.WithConn(
+		ctx,
+		func(conn pg.Conn) error {
+			return sourceFetch.Update(ctx, conn, scope)
+		},
+	)
+}
+
+func (w *SourceFetchWorker) finalizeCampaignFetchLifecycle(
+	ctx context.Context,
+	tenantID gid.TenantID,
+	campaignID gid.GID,
+) error {
+	scope := coredata.NewScope(tenantID)
+
+	return w.pg.WithTx(
+		ctx,
+		func(tx pg.Conn) error {
+			if err := lockCampaignForUpdate(ctx, tx, scope, campaignID); err != nil {
+				return fmt.Errorf("cannot lock campaign: %w", err)
+			}
+
+			campaign := &coredata.AccessReviewCampaign{}
+			if err := campaign.LoadByID(ctx, tx, scope, campaignID); err != nil {
+				return fmt.Errorf("cannot load campaign: %w", err)
+			}
+
+			if campaign.Status != coredata.AccessReviewCampaignStatusInProgress {
+				return nil
+			}
+
+			fetches := coredata.AccessReviewCampaignSourceFetches{}
+			if err := fetches.LoadByCampaignID(ctx, tx, scope, campaignID); err != nil {
+				return fmt.Errorf("cannot load source fetches: %w", err)
+			}
+
+			if len(fetches) == 0 {
+				return nil
+			}
+
+			hasFailure := false
+			for _, fetch := range fetches {
+				if !fetch.Status.IsTerminal() {
+					return nil
+				}
+				if fetch.Status == coredata.AccessReviewCampaignSourceFetchStatusFailed {
+					hasFailure = true
+				}
+			}
+
+			if hasFailure {
+				campaign.Status = coredata.AccessReviewCampaignStatusFailed
+			} else {
+				campaign.Status = coredata.AccessReviewCampaignStatusPendingActions
+			}
+
+			campaign.UpdatedAt = time.Now()
+			return campaign.Update(ctx, tx, scope)
+		},
+	)
+}