diff --git a/pkg/accessreview/source_name_worker.go b/pkg/accessreview/source_name_worker.go new file mode 100644 index 000000000..c64051bab --- /dev/null +++ b/pkg/accessreview/source_name_worker.go @@ -0,0 +1,290 @@ +// Copyright (c) 2026 Probo Inc . +// +// 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 + } +} diff --git a/pkg/accessreview/worker.go b/pkg/accessreview/worker.go new file mode 100644 index 000000000..39b620e6f --- /dev/null +++ b/pkg/accessreview/worker.go @@ -0,0 +1,338 @@ +// Copyright (c) 2026 Probo Inc . +// +// 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) + }, + ) +}