From 5efd5136e2c06a25dee52f8245bc197e2f7c2e0b Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Aur=C3=A9lien=20Sibiril?= <81782+aureliensibiril@users.noreply.github.com> Date: Thu, 2 Apr 2026 11:51:06 +0200 Subject: [PATCH] Add access review domain services MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Add AccessSourceService, AccessEntryService, CampaignService, and ReviewEngine in the accessreview package. Service exposes tenant-scoped sub-service accessors and an unscoped ResolveEntryOrganizationID. Register access review actions and policies. Signed-off-by: Aurélien Sibiril <81782+aureliensibiril@users.noreply.github.com> --- pkg/accessreview/access_entry_service.go | 471 +++++++++++++++++++ pkg/accessreview/access_source_service.go | 409 +++++++++++++++++ pkg/accessreview/campaign_service.go | 528 ++++++++++++++++++++++ pkg/accessreview/campaign_types.go | 68 +++ pkg/accessreview/review_engine.go | 378 ++++++++++++++++ pkg/accessreview/review_engine_test.go | 52 +++ pkg/accessreview/service.go | 147 ++++++ pkg/probo/actions.go | 28 ++ pkg/probo/policies.go | 3 + pkg/probo/service.go | 5 + 10 files changed, 2089 insertions(+) create mode 100644 pkg/accessreview/access_entry_service.go create mode 100644 pkg/accessreview/access_source_service.go create mode 100644 pkg/accessreview/campaign_service.go create mode 100644 pkg/accessreview/campaign_types.go create mode 100644 pkg/accessreview/review_engine.go create mode 100644 pkg/accessreview/review_engine_test.go create mode 100644 pkg/accessreview/service.go diff --git a/pkg/accessreview/access_entry_service.go b/pkg/accessreview/access_entry_service.go new file mode 100644 index 000000000..cc45186d4 --- /dev/null +++ b/pkg/accessreview/access_entry_service.go @@ -0,0 +1,471 @@ +// 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" + "fmt" + "strings" + "time" + + "go.gearno.de/kit/pg" + "go.probo.inc/probo/pkg/coredata" + "go.probo.inc/probo/pkg/gid" + "go.probo.inc/probo/pkg/page" +) + +type ( + AccessEntryService struct { + pg *pg.Client + scope coredata.Scoper + } + + RecordAccessEntryDecisionRequest struct { + EntryID gid.GID + Decision coredata.AccessEntryDecision + DecisionNote *string + DecidedByID *gid.GID + } + + FlagAccessEntryRequest struct { + EntryID gid.GID + Flags []coredata.AccessEntryFlag + FlagReasons []string + } +) + +func (s AccessEntryService) Get( + ctx context.Context, + entryID gid.GID, +) (*coredata.AccessEntry, error) { + entry := &coredata.AccessEntry{} + + err := s.pg.WithConn( + ctx, + func(conn pg.Conn) error { + return entry.LoadByID(ctx, conn, s.scope, entryID) + }, + ) + if err != nil { + return nil, fmt.Errorf("cannot get access entry: %w", err) + } + + return entry, nil +} + +func (s AccessEntryService) RecordDecision( + ctx context.Context, + req RecordAccessEntryDecisionRequest, +) (*coredata.AccessEntry, error) { + if req.Decision == coredata.AccessEntryDecisionPending { + return nil, fmt.Errorf("cannot decide access entry: invalid decision %q", req.Decision) + } + + if req.Decision != coredata.AccessEntryDecisionApproved { + if req.DecisionNote == nil || strings.TrimSpace(*req.DecisionNote) == "" { + return nil, fmt.Errorf("cannot decide access entry: note is required for non-approved decisions") + } + } + + entry := &coredata.AccessEntry{} + + err := s.pg.WithTx( + ctx, + func(conn pg.Conn) error { + if err := entry.LoadByID(ctx, conn, s.scope, req.EntryID); err != nil { + return fmt.Errorf("cannot load access entry: %w", err) + } + + campaign := &coredata.AccessReviewCampaign{} + if err := campaign.LoadByID(ctx, conn, s.scope, entry.AccessReviewCampaignID); err != nil { + return fmt.Errorf("cannot load campaign: %w", err) + } + + if campaign.Status != coredata.AccessReviewCampaignStatusPendingActions { + return fmt.Errorf("cannot decide access entry: campaign status is %s, expected PENDING_ACTIONS", campaign.Status) + } + + now := time.Now() + entry.Decision = req.Decision + entry.DecisionNote = req.DecisionNote + entry.DecidedBy = req.DecidedByID + entry.DecidedAt = &now + entry.UpdatedAt = now + if entry.Flags == nil { + entry.Flags = []coredata.AccessEntryFlag{} + } + if entry.FlagReasons == nil { + entry.FlagReasons = []string{} + } + if req.Decision == coredata.AccessEntryDecisionRevoke || req.Decision == coredata.AccessEntryDecisionEscalate { + if len(entry.Flags) == 0 { + entry.Flags = []coredata.AccessEntryFlag{coredata.AccessEntryFlagExcessive} + } + } + + if err := entry.Update(ctx, conn, s.scope); err != nil { + return fmt.Errorf("cannot record access entry decision: %w", err) + } + + history := &coredata.AccessEntryDecisionHistory{ + ID: gid.New(s.scope.GetTenantID(), coredata.AccessEntryDecisionHistoryEntityType), + OrganizationID: entry.OrganizationID, + AccessEntry: entry.ID, + Decision: entry.Decision, + DecisionNote: entry.DecisionNote, + DecidedBy: entry.DecidedBy, + DecidedAt: *entry.DecidedAt, + CreatedAt: now, + } + if err := history.Insert(ctx, conn, s.scope); err != nil { + return fmt.Errorf("cannot insert decision history: %w", err) + } + + return nil + }, + ) + if err != nil { + return nil, fmt.Errorf("cannot record access entry decision: %w", err) + } + + updatedEntry, err := s.Get(ctx, req.EntryID) + if err != nil { + return nil, fmt.Errorf("cannot reload access entry after decision: %w", err) + } + + return updatedEntry, nil +} + +func (s AccessEntryService) RecordDecisions( + ctx context.Context, + decisions []RecordAccessEntryDecisionRequest, +) ([]*coredata.AccessEntry, error) { + for _, d := range decisions { + if d.Decision == coredata.AccessEntryDecisionPending { + return nil, fmt.Errorf("cannot bulk decide access entries: invalid decision %q", d.Decision) + } + if d.Decision != coredata.AccessEntryDecisionApproved { + if d.DecisionNote == nil || strings.TrimSpace(*d.DecisionNote) == "" { + return nil, fmt.Errorf( + "cannot bulk decide access entries: note is required for non-approved decisions on entry %s", + d.EntryID, + ) + } + } + } + + entryIDs := make([]gid.GID, len(decisions)) + for i, d := range decisions { + entryIDs[i] = d.EntryID + } + + err := s.pg.WithTx( + ctx, + func(conn pg.Conn) error { + // Track verified campaigns to avoid repeated loads within the + // same transaction. + verifiedCampaigns := make(map[gid.GID]bool) + + for _, d := range decisions { + entry := &coredata.AccessEntry{} + if err := entry.LoadByID(ctx, conn, s.scope, d.EntryID); err != nil { + return fmt.Errorf("cannot load access entry %s: %w", d.EntryID, err) + } + + if !verifiedCampaigns[entry.AccessReviewCampaignID] { + campaign := &coredata.AccessReviewCampaign{} + if err := campaign.LoadByID(ctx, conn, s.scope, entry.AccessReviewCampaignID); err != nil { + return fmt.Errorf("cannot load campaign: %w", err) + } + if campaign.Status != coredata.AccessReviewCampaignStatusPendingActions { + return fmt.Errorf("cannot decide access entry: campaign status is %s, expected PENDING_ACTIONS", campaign.Status) + } + verifiedCampaigns[entry.AccessReviewCampaignID] = true + } + + now := time.Now() + entry.Decision = d.Decision + entry.DecisionNote = d.DecisionNote + entry.DecidedBy = d.DecidedByID + entry.DecidedAt = &now + entry.UpdatedAt = now + if entry.Flags == nil { + entry.Flags = []coredata.AccessEntryFlag{} + } + if entry.FlagReasons == nil { + entry.FlagReasons = []string{} + } + if d.Decision == coredata.AccessEntryDecisionRevoke || d.Decision == coredata.AccessEntryDecisionEscalate { + if len(entry.Flags) == 0 { + entry.Flags = []coredata.AccessEntryFlag{coredata.AccessEntryFlagExcessive} + } + } + + if err := entry.Update(ctx, conn, s.scope); err != nil { + return fmt.Errorf("cannot record decision for entry %s: %w", d.EntryID, err) + } + + history := &coredata.AccessEntryDecisionHistory{ + ID: gid.New(s.scope.GetTenantID(), coredata.AccessEntryDecisionHistoryEntityType), + OrganizationID: entry.OrganizationID, + AccessEntry: entry.ID, + Decision: entry.Decision, + DecisionNote: entry.DecisionNote, + DecidedBy: entry.DecidedBy, + DecidedAt: *entry.DecidedAt, + CreatedAt: now, + } + if err := history.Insert(ctx, conn, s.scope); err != nil { + return fmt.Errorf("cannot insert decision history for entry %s: %w", d.EntryID, err) + } + } + + return nil + }, + ) + if err != nil { + return nil, fmt.Errorf("cannot record access entry decisions: %w", err) + } + + entries := make([]*coredata.AccessEntry, len(entryIDs)) + for i, id := range entryIDs { + entry, err := s.Get(ctx, id) + if err != nil { + return nil, fmt.Errorf("cannot reload access entry %s: %w", id, err) + } + entries[i] = entry + } + + return entries, nil +} + +func (s AccessEntryService) FlagEntry( + ctx context.Context, + req FlagAccessEntryRequest, +) (*coredata.AccessEntry, error) { + entry := &coredata.AccessEntry{} + + err := s.pg.WithTx( + ctx, + func(conn pg.Conn) error { + if err := entry.LoadByID(ctx, conn, s.scope, req.EntryID); err != nil { + return fmt.Errorf("cannot load access entry: %w", err) + } + + campaign := &coredata.AccessReviewCampaign{} + if err := campaign.LoadByID(ctx, conn, s.scope, entry.AccessReviewCampaignID); err != nil { + return fmt.Errorf("cannot load campaign: %w", err) + } + + if campaign.Status != coredata.AccessReviewCampaignStatusPendingActions { + return fmt.Errorf("cannot flag access entry: campaign status is %s, expected PENDING_ACTIONS", campaign.Status) + } + + now := time.Now() + entry.Flags = req.Flags + if entry.Flags == nil { + entry.Flags = []coredata.AccessEntryFlag{} + } + entry.FlagReasons = req.FlagReasons + if entry.FlagReasons == nil { + entry.FlagReasons = []string{} + } + entry.UpdatedAt = now + + return entry.UpdateFlags(ctx, conn, s.scope) + }, + ) + if err != nil { + return nil, fmt.Errorf("cannot flag access entry: %w", err) + } + + return s.Get(ctx, req.EntryID) +} + +func (s AccessEntryService) ListForCampaignID( + ctx context.Context, + campaignID gid.GID, + cursor *page.Cursor[coredata.AccessEntryOrderField], + filter *coredata.AccessEntryFilter, +) (*page.Page[*coredata.AccessEntry, coredata.AccessEntryOrderField], error) { + var entries coredata.AccessEntries + + err := s.pg.WithConn( + ctx, + func(conn pg.Conn) error { + return entries.LoadByCampaignID(ctx, conn, s.scope, campaignID, cursor, filter) + }, + ) + if err != nil { + return nil, fmt.Errorf("cannot list access entries: %w", err) + } + + return page.NewPage(entries, cursor), nil +} + +func (s AccessEntryService) ListForCampaignIDAndSourceID( + ctx context.Context, + campaignID gid.GID, + sourceID gid.GID, + cursor *page.Cursor[coredata.AccessEntryOrderField], + filter *coredata.AccessEntryFilter, +) (*page.Page[*coredata.AccessEntry, coredata.AccessEntryOrderField], error) { + var entries coredata.AccessEntries + + err := s.pg.WithConn( + ctx, + func(conn pg.Conn) error { + return entries.LoadByCampaignIDAndSourceID(ctx, conn, s.scope, campaignID, sourceID, cursor, filter) + }, + ) + if err != nil { + return nil, fmt.Errorf("cannot list access entries: %w", err) + } + + return page.NewPage(entries, cursor), nil +} + +func (s AccessEntryService) CountForCampaignID( + ctx context.Context, + campaignID gid.GID, + filter *coredata.AccessEntryFilter, +) (int, error) { + var count int + + err := s.pg.WithConn( + ctx, + func(conn pg.Conn) (err error) { + entries := coredata.AccessEntries{} + count, err = entries.CountByCampaignID(ctx, conn, s.scope, campaignID, filter) + if err != nil { + return fmt.Errorf("cannot count access entries by campaign: %w", err) + } + return nil + }, + ) + if err != nil { + return 0, fmt.Errorf("cannot count access entries: %w", err) + } + + return count, nil +} + +func (s AccessEntryService) CountForCampaignIDAndSourceID( + ctx context.Context, + campaignID gid.GID, + sourceID gid.GID, + filter *coredata.AccessEntryFilter, +) (int, error) { + var count int + + err := s.pg.WithConn( + ctx, + func(conn pg.Conn) (err error) { + entries := coredata.AccessEntries{} + count, err = entries.CountByCampaignIDAndSourceID(ctx, conn, s.scope, campaignID, sourceID, filter) + if err != nil { + return fmt.Errorf("cannot count access entries by campaign and source: %w", err) + } + return nil + }, + ) + if err != nil { + return 0, fmt.Errorf("cannot count access entries: %w", err) + } + + return count, nil +} + +func (s AccessEntryService) CountPendingForCampaignID( + ctx context.Context, + campaignID gid.GID, +) (int, error) { + var count int + + err := s.pg.WithConn( + ctx, + func(conn pg.Conn) (err error) { + entries := coredata.AccessEntries{} + count, err = entries.CountPendingByCampaignID(ctx, conn, s.scope, campaignID) + if err != nil { + return fmt.Errorf("cannot count pending access entries: %w", err) + } + return nil + }, + ) + if err != nil { + return 0, fmt.Errorf("cannot count pending access entries: %w", err) + } + + return count, nil +} + +func (s AccessEntryService) DecisionHistory( + ctx context.Context, + entryID gid.GID, +) (coredata.AccessEntryDecisionHistories, error) { + var histories coredata.AccessEntryDecisionHistories + + err := s.pg.WithConn( + ctx, + func(conn pg.Conn) error { + return histories.LoadByEntryID(ctx, conn, s.scope, entryID) + }, + ) + if err != nil { + return nil, fmt.Errorf("cannot load decision history: %w", err) + } + + return histories, nil +} + +func (s AccessEntryService) Statistics( + ctx context.Context, + campaignID gid.GID, +) (*coredata.AccessEntryStatistics, error) { + stats := &coredata.AccessEntryStatistics{} + + err := s.pg.WithConn( + ctx, + func(conn pg.Conn) error { + return stats.LoadByCampaignID(ctx, conn, s.scope, campaignID) + }, + ) + if err != nil { + return nil, fmt.Errorf("cannot load campaign statistics: %w", err) + } + + return stats, nil +} + +func (s AccessEntryService) StatisticsForSource( + ctx context.Context, + campaignID gid.GID, + sourceID gid.GID, +) (*coredata.AccessEntryStatistics, error) { + stats := &coredata.AccessEntryStatistics{} + + err := s.pg.WithConn( + ctx, + func(conn pg.Conn) error { + return stats.LoadByCampaignIDAndSourceID(ctx, conn, s.scope, campaignID, sourceID) + }, + ) + if err != nil { + return nil, fmt.Errorf("cannot load source statistics: %w", err) + } + + return stats, nil +} diff --git a/pkg/accessreview/access_source_service.go b/pkg/accessreview/access_source_service.go new file mode 100644 index 000000000..f4a6e2913 --- /dev/null +++ b/pkg/accessreview/access_source_service.go @@ -0,0 +1,409 @@ +// 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" + "fmt" + "net/http" + "time" + + "go.gearno.de/kit/pg" + "go.probo.inc/probo/pkg/connector" + "go.probo.inc/probo/pkg/coredata" + "go.probo.inc/probo/pkg/crypto/cipher" + "go.probo.inc/probo/pkg/gid" + "go.probo.inc/probo/pkg/page" + "go.probo.inc/probo/pkg/validator" +) + +const ( + NameMaxLength = 1000 +) + +type ( + AccessSourceService struct { + pg *pg.Client + scope coredata.Scoper + encryptionKey cipher.EncryptionKey + connectorRegistry *connector.ConnectorRegistry + } + + CreateAccessSourceRequest struct { + OrganizationID gid.GID + ConnectorID *gid.GID + Name string + Category coredata.AccessSourceCategory + CsvData *string + } + + UpdateAccessSourceRequest struct { + AccessSourceID gid.GID + Name *string + Category *coredata.AccessSourceCategory + ConnectorID **gid.GID + CsvData **string + } + + ConfigureAccessSourceRequest struct { + AccessSourceID gid.GID + OrganizationSlug string + } +) + +func (r *CreateAccessSourceRequest) Validate() error { + v := validator.New() + + v.Check(r.OrganizationID, "organization_id", validator.Required(), validator.GID(coredata.OrganizationEntityType)) + v.Check(r.Name, "name", validator.SafeTextNoNewLine(NameMaxLength)) + v.Check(r.Category, "category", validator.OneOfSlice(coredata.AccessSourceCategories())) + + return v.Error() +} + +func (r *ConfigureAccessSourceRequest) Validate() error { + v := validator.New() + + v.Check(r.AccessSourceID, "access_source_id", validator.Required(), validator.GID(coredata.AccessSourceEntityType)) + v.Check(r.OrganizationSlug, "organization_slug", validator.Required()) + + return v.Error() +} + +func (r *UpdateAccessSourceRequest) Validate() error { + v := validator.New() + + v.Check(r.AccessSourceID, "access_source_id", validator.Required(), validator.GID(coredata.AccessSourceEntityType)) + v.Check(r.Name, "name", validator.SafeTextNoNewLine(NameMaxLength)) + v.Check(r.Category, "category", validator.OneOfSlice(coredata.AccessSourceCategories())) + + return v.Error() +} + +func (s AccessSourceService) Create( + ctx context.Context, + req CreateAccessSourceRequest, +) (*coredata.AccessSource, error) { + if err := req.Validate(); err != nil { + return nil, err + } + + now := time.Now() + source := &coredata.AccessSource{ + ID: gid.New(s.scope.GetTenantID(), coredata.AccessSourceEntityType), + OrganizationID: req.OrganizationID, + ConnectorID: req.ConnectorID, + Name: req.Name, + Category: req.Category, + CsvData: req.CsvData, + CreatedAt: now, + UpdatedAt: now, + } + + err := s.pg.WithTx( + ctx, + func(conn pg.Conn) error { + // Validate connector exists if provided + if req.ConnectorID != nil { + connector := &coredata.Connector{} + if err := connector.LoadMetadataByID(ctx, conn, s.scope, *req.ConnectorID); err != nil { + return fmt.Errorf("cannot load connector: %w", err) + } + } + + if err := source.Insert(ctx, conn, s.scope); err != nil { + return fmt.Errorf("cannot insert access source: %w", err) + } + + return nil + }, + ) + if err != nil { + return nil, fmt.Errorf("cannot create access source: %w", err) + } + + return source, nil +} + +func (s AccessSourceService) Get( + ctx context.Context, + accessSourceID gid.GID, +) (*coredata.AccessSource, error) { + source := &coredata.AccessSource{} + + err := s.pg.WithConn( + ctx, + func(conn pg.Conn) error { + return source.LoadByID(ctx, conn, s.scope, accessSourceID) + }, + ) + if err != nil { + return nil, fmt.Errorf("cannot get access source: %w", err) + } + + return source, nil +} + +func (s AccessSourceService) Update( + ctx context.Context, + req UpdateAccessSourceRequest, +) (*coredata.AccessSource, error) { + if err := req.Validate(); err != nil { + return nil, err + } + + source := &coredata.AccessSource{} + + err := s.pg.WithTx( + ctx, + func(conn pg.Conn) error { + if err := source.LoadByID(ctx, conn, s.scope, req.AccessSourceID); err != nil { + return fmt.Errorf("cannot load access source: %w", err) + } + + if req.Name != nil { + source.Name = *req.Name + } + + if req.Category != nil { + source.Category = *req.Category + } + + if req.ConnectorID != nil { + if *req.ConnectorID != nil { + connector := &coredata.Connector{} + if err := connector.LoadMetadataByID(ctx, conn, s.scope, **req.ConnectorID); err != nil { + return fmt.Errorf("cannot load connector: %w", err) + } + } + source.ConnectorID = *req.ConnectorID + } + + if req.CsvData != nil { + source.CsvData = *req.CsvData + } + + source.UpdatedAt = time.Now() + + if err := source.Update(ctx, conn, s.scope); err != nil { + return fmt.Errorf("cannot update access source: %w", err) + } + + return nil + }, + ) + if err != nil { + return nil, fmt.Errorf("cannot update access source: %w", err) + } + + return source, nil +} + +func (s AccessSourceService) Delete( + ctx context.Context, + accessSourceID gid.GID, +) error { + source := &coredata.AccessSource{ID: accessSourceID} + + return s.pg.WithTx( + ctx, + func(conn pg.Conn) error { + return source.Delete(ctx, conn, s.scope) + }, + ) +} + +func (s AccessSourceService) ListForOrganizationID( + ctx context.Context, + organizationID gid.GID, + cursor *page.Cursor[coredata.AccessSourceOrderField], +) (*page.Page[*coredata.AccessSource, coredata.AccessSourceOrderField], error) { + var sources coredata.AccessSources + + err := s.pg.WithConn( + ctx, + func(conn pg.Conn) error { + return sources.LoadByOrganizationID(ctx, conn, s.scope, organizationID, cursor) + }, + ) + if err != nil { + return nil, fmt.Errorf("cannot list access sources: %w", err) + } + + return page.NewPage(sources, cursor), nil +} + +func (s AccessSourceService) CountForOrganizationID( + ctx context.Context, + organizationID gid.GID, +) (int, error) { + var count int + + err := s.pg.WithConn( + ctx, + func(conn pg.Conn) (err error) { + sources := coredata.AccessSources{} + count, err = sources.CountByOrganizationID(ctx, conn, s.scope, organizationID) + return err + }, + ) + if err != nil { + return 0, fmt.Errorf("cannot count access sources: %w", err) + } + + return count, nil +} + +func (s AccessSourceService) ListScopeSourcesForCampaignID( + ctx context.Context, + campaignID gid.GID, +) ([]*coredata.AccessSource, error) { + var sources coredata.AccessSources + + err := s.pg.WithConn( + ctx, + func(conn pg.Conn) error { + return sources.LoadScopeSourcesByCampaignID(ctx, conn, s.scope, campaignID) + }, + ) + if err != nil { + return nil, fmt.Errorf("cannot list scope sources: %w", err) + } + + return sources, nil +} + +// ConnectorHTTPClient loads a connector by ID with decrypted credentials +// and returns an HTTP client with token refresh support. If the token was +// refreshed during client creation, the updated credentials are persisted. +func (s AccessSourceService) ConnectorHTTPClient( + ctx context.Context, + connectorID gid.GID, +) (*http.Client, *coredata.Connector, error) { + var dbConnector coredata.Connector + + err := s.pg.WithConn( + ctx, + func(conn pg.Conn) error { + if err := dbConnector.LoadByID(ctx, conn, s.scope, connectorID, s.encryptionKey); err != nil { + return fmt.Errorf("cannot load connector: %w", err) + } + return nil + }, + ) + if err != nil { + return nil, nil, err + } + + var tokenBefore string + oauth2Conn, isOAuth2 := dbConnector.Connection.(*connector.OAuth2Connection) + if isOAuth2 { + tokenBefore = oauth2Conn.AccessToken + } + + var httpClient *http.Client + if isOAuth2 && s.connectorRegistry != nil { + refreshCfg := s.connectorRegistry.GetOAuth2RefreshConfig(string(dbConnector.Provider)) + if refreshCfg != nil { + var err error + httpClient, err = oauth2Conn.RefreshableClient(ctx, *refreshCfg) + if err != nil { + return nil, nil, fmt.Errorf("cannot create refreshable HTTP client: %w", err) + } + } + } + + if httpClient == nil { + var err error + httpClient, err = dbConnector.Connection.Client(ctx) + if err != nil { + return nil, nil, fmt.Errorf("cannot create HTTP client: %w", err) + } + } + + // Persist refreshed token if it changed. + if isOAuth2 && oauth2Conn.AccessToken != tokenBefore { + dbConnector.UpdatedAt = time.Now() + if err := s.pg.WithConn( + ctx, + func(conn pg.Conn) error { + return dbConnector.Update(ctx, conn, s.scope, s.encryptionKey) + }, + ); err != nil { + return nil, nil, fmt.Errorf("cannot persist refreshed token: %w", err) + } + } + + return httpClient, &dbConnector, nil +} + +func (s AccessSourceService) ConfigureAccessSource( + ctx context.Context, + req ConfigureAccessSourceRequest, +) (*coredata.AccessSource, error) { + if err := req.Validate(); err != nil { + return nil, err + } + + source := &coredata.AccessSource{} + + err := s.pg.WithTx( + ctx, + func(conn pg.Conn) error { + if err := source.LoadByID(ctx, conn, s.scope, req.AccessSourceID); err != nil { + return fmt.Errorf("cannot load access source: %w", err) + } + + if source.ConnectorID == nil { + return fmt.Errorf("cannot configure access source: no connector attached") + } + + dbConnector := &coredata.Connector{} + if err := dbConnector.LoadByID(ctx, conn, s.scope, *source.ConnectorID, s.encryptionKey); err != nil { + return fmt.Errorf("cannot load connector: %w", err) + } + + switch dbConnector.Provider { + case coredata.ConnectorProviderGitHub: + if err := dbConnector.SetSettings(&coredata.GitHubConnectorSettings{ + Organization: req.OrganizationSlug, + }); err != nil { + return fmt.Errorf("cannot set github settings: %w", err) + } + case coredata.ConnectorProviderSentry: + if err := dbConnector.SetSettings(&coredata.SentryConnectorSettings{ + OrganizationSlug: req.OrganizationSlug, + }); err != nil { + return fmt.Errorf("cannot set sentry settings: %w", err) + } + default: + return fmt.Errorf("cannot configure access source: provider %s does not support organization configuration", dbConnector.Provider) + } + + dbConnector.UpdatedAt = time.Now() + + if err := dbConnector.Update(ctx, conn, s.scope, s.encryptionKey); err != nil { + return fmt.Errorf("cannot update connector: %w", err) + } + + return nil + }, + ) + if err != nil { + return nil, err + } + + return source, nil +} diff --git a/pkg/accessreview/campaign_service.go b/pkg/accessreview/campaign_service.go new file mode 100644 index 000000000..dd3b7b47b --- /dev/null +++ b/pkg/accessreview/campaign_service.go @@ -0,0 +1,528 @@ +// 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" + "fmt" + "time" + + "go.gearno.de/kit/pg" + "go.probo.inc/probo/pkg/coredata" + "go.probo.inc/probo/pkg/gid" + "go.probo.inc/probo/pkg/page" +) + +type CampaignService struct { + pg *pg.Client + scope coredata.Scoper +} + +func NewCampaignService(pgClient *pg.Client, scope coredata.Scoper) *CampaignService { + return &CampaignService{ + pg: pgClient, + scope: scope, + } +} + +func (s *CampaignService) Create( + ctx context.Context, + req CreateAccessReviewCampaignRequest, +) (*coredata.AccessReviewCampaign, error) { + if err := req.Validate(); err != nil { + return nil, err + } + + now := time.Now() + campaign := &coredata.AccessReviewCampaign{ + ID: gid.New(s.scope.GetTenantID(), coredata.AccessReviewCampaignEntityType), + OrganizationID: req.OrganizationID, + Name: req.Name, + Description: req.Description, + Status: coredata.AccessReviewCampaignStatusDraft, + FrameworkControls: req.FrameworkControls, + CreatedAt: now, + UpdatedAt: now, + } + + err := s.pg.WithTx( + ctx, + func(conn pg.Conn) error { + if err := campaign.Insert(ctx, conn, s.scope); err != nil { + return fmt.Errorf("cannot insert access review campaign: %w", err) + } + + for _, sourceID := range req.AccessSourceIDs { + source := &coredata.AccessSource{} + if err := source.LoadByID(ctx, conn, s.scope, sourceID); err != nil { + return fmt.Errorf("cannot load access source %s: %w", sourceID, err) + } + + if source.OrganizationID != campaign.OrganizationID { + return fmt.Errorf("cannot create campaign: access source %s does not belong to the same organization", sourceID) + } + + scopeSystem := coredata.AccessReviewCampaignScopeSystem{ + AccessReviewCampaignID: campaign.ID, + AccessSourceID: sourceID, + } + if err := scopeSystem.Insert(ctx, conn, s.scope); err != nil { + return fmt.Errorf("cannot insert scope system: %w", err) + } + } + + return nil + }, + ) + if err != nil { + return nil, err + } + + return campaign, nil +} + +func (s *CampaignService) Get( + ctx context.Context, + campaignID gid.GID, +) (*coredata.AccessReviewCampaign, error) { + campaign := &coredata.AccessReviewCampaign{} + + err := s.pg.WithConn( + ctx, + func(conn pg.Conn) error { + if err := campaign.LoadByID(ctx, conn, s.scope, campaignID); err != nil { + return fmt.Errorf("cannot load campaign: %w", err) + } + return nil + }, + ) + if err != nil { + return nil, err + } + + return campaign, nil +} + +func (s *CampaignService) Update( + ctx context.Context, + req UpdateAccessReviewCampaignRequest, +) (*coredata.AccessReviewCampaign, error) { + if err := req.Validate(); err != nil { + return nil, fmt.Errorf("cannot validate update campaign request: %w", err) + } + + campaign := &coredata.AccessReviewCampaign{} + + err := s.pg.WithTx( + ctx, + func(conn pg.Conn) error { + if err := lockCampaignForUpdate(ctx, conn, s.scope, req.CampaignID); err != nil { + return fmt.Errorf("cannot lock campaign: %w", err) + } + + if err := campaign.LoadByID(ctx, conn, s.scope, req.CampaignID); err != nil { + return fmt.Errorf("cannot load campaign: %w", err) + } + + if campaign.Status != coredata.AccessReviewCampaignStatusDraft { + return fmt.Errorf("cannot update campaign: status is %s, expected DRAFT", campaign.Status) + } + + if req.Name != nil { + campaign.Name = *req.Name + } + + if req.Description != nil { + campaign.Description = *req.Description + } + + if req.FrameworkControls != nil { + campaign.FrameworkControls = *req.FrameworkControls + } + + campaign.UpdatedAt = time.Now() + + if err := campaign.Update(ctx, conn, s.scope); err != nil { + return fmt.Errorf("cannot update campaign: %w", err) + } + + return nil + }, + ) + if err != nil { + return nil, err + } + + return campaign, nil +} + +func (s *CampaignService) Delete( + ctx context.Context, + campaignID gid.GID, +) error { + return s.pg.WithTx( + ctx, + func(conn pg.Conn) error { + if err := lockCampaignForUpdate(ctx, conn, s.scope, campaignID); err != nil { + return fmt.Errorf("cannot lock campaign: %w", err) + } + + campaign := &coredata.AccessReviewCampaign{} + if err := campaign.LoadByID(ctx, conn, s.scope, campaignID); err != nil { + return fmt.Errorf("cannot load campaign: %w", err) + } + + if campaign.Status != coredata.AccessReviewCampaignStatusDraft && + campaign.Status != coredata.AccessReviewCampaignStatusCancelled { + return fmt.Errorf("cannot delete campaign: status is %s, expected %s or %s", campaign.Status, coredata.AccessReviewCampaignStatusDraft, coredata.AccessReviewCampaignStatusCancelled) + } + + if err := campaign.Delete(ctx, conn, s.scope); err != nil { + return fmt.Errorf("cannot delete campaign: %w", err) + } + return nil + }, + ) +} + +func (s *CampaignService) AddScopeSource( + ctx context.Context, + req AddCampaignScopeSourceRequest, +) (*coredata.AccessReviewCampaign, error) { + campaign := &coredata.AccessReviewCampaign{} + + err := s.pg.WithTx( + ctx, + func(conn pg.Conn) error { + if err := lockCampaignForUpdate(ctx, conn, s.scope, req.CampaignID); err != nil { + return fmt.Errorf("cannot lock campaign: %w", err) + } + + if err := campaign.LoadByID(ctx, conn, s.scope, req.CampaignID); err != nil { + return fmt.Errorf("cannot load campaign: %w", err) + } + + if campaign.Status != coredata.AccessReviewCampaignStatusDraft { + return fmt.Errorf("cannot add scope source: campaign status is %s, expected %s", campaign.Status, coredata.AccessReviewCampaignStatusDraft) + } + + source := &coredata.AccessSource{} + if err := source.LoadByID(ctx, conn, s.scope, req.AccessSourceID); err != nil { + return fmt.Errorf("cannot load access source %s: %w", req.AccessSourceID, err) + } + + if source.OrganizationID != campaign.OrganizationID { + return fmt.Errorf("cannot add scope source: access source %q does not belong to the same organization", req.AccessSourceID) + } + + scopeSystem := coredata.AccessReviewCampaignScopeSystem{ + AccessReviewCampaignID: campaign.ID, + AccessSourceID: req.AccessSourceID, + } + if err := scopeSystem.Upsert(ctx, conn, s.scope); err != nil { + return fmt.Errorf("cannot upsert scope system: %w", err) + } + + return nil + }, + ) + if err != nil { + return nil, err + } + + return campaign, nil +} + +func (s *CampaignService) RemoveScopeSource( + ctx context.Context, + req RemoveCampaignScopeSourceRequest, +) (*coredata.AccessReviewCampaign, error) { + campaign := &coredata.AccessReviewCampaign{} + + err := s.pg.WithTx( + ctx, + func(conn pg.Conn) error { + if err := lockCampaignForUpdate(ctx, conn, s.scope, req.CampaignID); err != nil { + return fmt.Errorf("cannot lock campaign: %w", err) + } + + if err := campaign.LoadByID(ctx, conn, s.scope, req.CampaignID); err != nil { + return fmt.Errorf("cannot load campaign: %w", err) + } + + if campaign.Status != coredata.AccessReviewCampaignStatusDraft { + return fmt.Errorf("cannot remove scope source: campaign status is %s, expected DRAFT", campaign.Status) + } + + scopeSystem := coredata.AccessReviewCampaignScopeSystem{ + AccessReviewCampaignID: campaign.ID, + AccessSourceID: req.AccessSourceID, + } + if err := scopeSystem.Delete(ctx, conn, s.scope); err != nil { + return fmt.Errorf("cannot delete scope system: %w", err) + } + + return nil + }, + ) + if err != nil { + return nil, err + } + + return campaign, nil +} + +func (s *CampaignService) Start( + ctx context.Context, + campaignID gid.GID, +) (*coredata.AccessReviewCampaign, error) { + campaign := &coredata.AccessReviewCampaign{} + + err := s.pg.WithTx( + ctx, + func(conn pg.Conn) error { + if err := lockCampaignForUpdate(ctx, conn, s.scope, campaignID); err != nil { + return fmt.Errorf("cannot lock campaign: %w", err) + } + + if err := campaign.LoadByID(ctx, conn, s.scope, campaignID); err != nil { + return fmt.Errorf("cannot load campaign: %w", err) + } + + if campaign.Status != coredata.AccessReviewCampaignStatusDraft && + campaign.Status != coredata.AccessReviewCampaignStatusFailed { + return fmt.Errorf("cannot start campaign: status is %s, expected %s or %s", campaign.Status, coredata.AccessReviewCampaignStatusDraft, coredata.AccessReviewCampaignStatusFailed) + } + + var sources coredata.AccessSources + if err := sources.LoadScopeSourcesByCampaignID(ctx, conn, s.scope, campaign.ID); err != nil { + return fmt.Errorf("cannot load scope sources: %w", err) + } + + if len(sources) == 0 { + return fmt.Errorf("cannot start campaign: no scope sources configured") + } + + now := time.Now() + campaign.Status = coredata.AccessReviewCampaignStatusInProgress + campaign.StartedAt = &now + campaign.UpdatedAt = now + + if err := campaign.Update(ctx, conn, s.scope); err != nil { + return fmt.Errorf("cannot update campaign: %w", err) + } + + if err := s.enqueueSourceFetches(ctx, conn, campaign.ID, sources); err != nil { + return fmt.Errorf("cannot queue source fetches: %w", err) + } + + return nil + }, + ) + if err != nil { + return nil, err + } + + return campaign, nil +} + +func (s *CampaignService) Close( + ctx context.Context, + campaignID gid.GID, +) (*coredata.AccessReviewCampaign, error) { + campaign := &coredata.AccessReviewCampaign{} + + err := s.pg.WithTx( + ctx, + func(conn pg.Conn) error { + if err := lockCampaignForUpdate(ctx, conn, s.scope, campaignID); err != nil { + return fmt.Errorf("cannot lock campaign: %w", err) + } + + if err := campaign.LoadByID(ctx, conn, s.scope, campaignID); err != nil { + return fmt.Errorf("cannot load campaign: %w", err) + } + + if campaign.Status != coredata.AccessReviewCampaignStatusPendingActions { + return fmt.Errorf("cannot close campaign: status is %s, expected %s", campaign.Status, coredata.AccessReviewCampaignStatusPendingActions) + } + + entries := coredata.AccessEntries{} + pendingCount, err := entries.CountPendingByCampaignID(ctx, conn, s.scope, campaignID) + if err != nil { + return fmt.Errorf("cannot count pending entries: %w", err) + } + + if pendingCount > 0 { + return fmt.Errorf("cannot close campaign: %d entries still pending", pendingCount) + } + + now := time.Now() + campaign.Status = coredata.AccessReviewCampaignStatusCompleted + campaign.CompletedAt = &now + campaign.UpdatedAt = now + + if err := campaign.Update(ctx, conn, s.scope); err != nil { + return fmt.Errorf("cannot update campaign: %w", err) + } + + return nil + }, + ) + if err != nil { + return nil, err + } + + return campaign, nil +} + +func lockCampaignForUpdate(ctx context.Context, conn pg.Conn, scope coredata.Scoper, campaignID gid.GID) error { + c := &coredata.AccessReviewCampaign{ID: campaignID} + if err := c.LockForUpdate(ctx, conn, scope); err != nil { + return fmt.Errorf("cannot lock campaign for update: %w", err) + } + return nil +} + +func (s *CampaignService) enqueueSourceFetches( + ctx context.Context, + conn pg.Conn, + campaignID gid.GID, + sources coredata.AccessSources, +) error { + now := time.Now() + for _, source := range sources { + fetch := &coredata.AccessReviewCampaignSourceFetch{ + AccessReviewCampaignID: campaignID, + AccessSourceID: source.ID, + } + if err := fetch.UpsertQueued(ctx, conn, s.scope, now); err != nil { + return fmt.Errorf("cannot queue source fetch %s: %w", source.ID, err) + } + } + + return nil +} + +func (s *CampaignService) Cancel( + ctx context.Context, + campaignID gid.GID, +) (*coredata.AccessReviewCampaign, error) { + campaign := &coredata.AccessReviewCampaign{} + + err := s.pg.WithTx( + ctx, + func(conn pg.Conn) error { + if err := lockCampaignForUpdate(ctx, conn, s.scope, campaignID); err != nil { + return fmt.Errorf("cannot lock campaign: %w", err) + } + + if err := campaign.LoadByID(ctx, conn, s.scope, campaignID); err != nil { + return fmt.Errorf("cannot load campaign: %w", err) + } + + if campaign.Status == coredata.AccessReviewCampaignStatusCompleted || + campaign.Status == coredata.AccessReviewCampaignStatusCancelled { + return fmt.Errorf("cannot update campaign: already %s", campaign.Status) + } + + now := time.Now() + campaign.Status = coredata.AccessReviewCampaignStatusCancelled + campaign.CompletedAt = &now + campaign.UpdatedAt = now + + if err := campaign.Update(ctx, conn, s.scope); err != nil { + return fmt.Errorf("cannot update campaign: %w", err) + } + + return nil + }, + ) + if err != nil { + return nil, err + } + + return campaign, nil +} + +func (s *CampaignService) ListForOrganizationID( + ctx context.Context, + organizationID gid.GID, + cursor *page.Cursor[coredata.AccessReviewCampaignOrderField], +) (*page.Page[*coredata.AccessReviewCampaign, coredata.AccessReviewCampaignOrderField], error) { + var campaigns coredata.AccessReviewCampaigns + + err := s.pg.WithConn( + ctx, + func(conn pg.Conn) error { + if err := campaigns.LoadByOrganizationID(ctx, conn, s.scope, organizationID, cursor); err != nil { + return fmt.Errorf("cannot load campaigns by organization: %w", err) + } + return nil + }, + ) + if err != nil { + return nil, err + } + + return page.NewPage(campaigns, cursor), nil +} + +func (s *CampaignService) ListSourceFetches( + ctx context.Context, + campaignID gid.GID, +) (coredata.AccessReviewCampaignSourceFetches, error) { + var fetches coredata.AccessReviewCampaignSourceFetches + + err := s.pg.WithConn( + ctx, + func(conn pg.Conn) error { + if err := fetches.LoadByCampaignID(ctx, conn, s.scope, campaignID); err != nil { + return fmt.Errorf("cannot load source fetches by campaign: %w", err) + } + return nil + }, + ) + if err != nil { + return nil, err + } + + return fetches, nil +} + +func (s *CampaignService) CountForOrganizationID( + ctx context.Context, + organizationID gid.GID, +) (int, error) { + var count int + + err := s.pg.WithConn( + ctx, + func(conn pg.Conn) (err error) { + campaigns := coredata.AccessReviewCampaigns{} + count, err = campaigns.CountByOrganizationID(ctx, conn, s.scope, organizationID) + if err != nil { + return fmt.Errorf("cannot count campaigns by organization: %w", err) + } + + return nil + }, + ) + if err != nil { + return 0, err + } + + return count, nil +} diff --git a/pkg/accessreview/campaign_types.go b/pkg/accessreview/campaign_types.go new file mode 100644 index 000000000..e7c7cf7b2 --- /dev/null +++ b/pkg/accessreview/campaign_types.go @@ -0,0 +1,68 @@ +// 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 ( + "go.probo.inc/probo/pkg/coredata" + "go.probo.inc/probo/pkg/gid" + "go.probo.inc/probo/pkg/validator" +) + +const campaignNameMaxLength = 255 + +type ( + CreateAccessReviewCampaignRequest struct { + OrganizationID gid.GID + Name string + Description string + FrameworkControls []string + AccessSourceIDs []gid.GID + } + + UpdateAccessReviewCampaignRequest struct { + CampaignID gid.GID + Name *string + Description *string + FrameworkControls *[]string + } + + AddCampaignScopeSourceRequest struct { + CampaignID gid.GID + AccessSourceID gid.GID + } + + RemoveCampaignScopeSourceRequest struct { + CampaignID gid.GID + AccessSourceID gid.GID + } +) + +func (r *CreateAccessReviewCampaignRequest) Validate() error { + v := validator.New() + + v.Check(r.OrganizationID, "organization_id", validator.Required(), validator.GID(coredata.OrganizationEntityType)) + v.Check(r.Name, "name", validator.SafeTextNoNewLine(campaignNameMaxLength)) + + return v.Error() +} + +func (r *UpdateAccessReviewCampaignRequest) Validate() error { + v := validator.New() + + v.Check(r.CampaignID, "campaign_id", validator.Required(), validator.GID(coredata.AccessReviewCampaignEntityType)) + v.Check(r.Name, "name", validator.SafeTextNoNewLine(campaignNameMaxLength)) + + return v.Error() +} diff --git a/pkg/accessreview/review_engine.go b/pkg/accessreview/review_engine.go new file mode 100644 index 000000000..0ba86bee1 --- /dev/null +++ b/pkg/accessreview/review_engine.go @@ -0,0 +1,378 @@ +// 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" + "strings" + "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" + "go.probo.inc/probo/pkg/gid" +) + +// ReviewEngine contains the stateless core logic for access review campaigns: +// snapshot and source data collection. +type ReviewEngine struct { + pg *pg.Client + scope coredata.Scoper + encryptionKey cipher.EncryptionKey + connectorRegistry *connector.ConnectorRegistry + logger *log.Logger +} + +func NewReviewEngine( + pgClient *pg.Client, + scope coredata.Scoper, + encryptionKey cipher.EncryptionKey, + connectorRegistry *connector.ConnectorRegistry, + logger *log.Logger, +) *ReviewEngine { + return &ReviewEngine{ + pg: pgClient, + scope: scope, + encryptionKey: encryptionKey, + connectorRegistry: connectorRegistry, + logger: logger, + } +} + +// FetchSource pulls accounts from a single source and upserts access entries. +func (e *ReviewEngine) FetchSource( + ctx context.Context, + campaign *coredata.AccessReviewCampaign, + sourceID gid.GID, +) (int, error) { + fetchedCount := 0 + + // Resolve the driver and load baseline data outside the write transaction + // so that external HTTP calls do not hold a database connection. + var ( + source *coredata.AccessSource + driver drivers.Driver + baseline []coredata.BaselineAccountEntry + ) + + err := e.pg.WithConn( + ctx, + func(conn pg.Conn) error { + source = &coredata.AccessSource{} + if err := source.LoadByID(ctx, conn, e.scope, sourceID); err != nil { + return fmt.Errorf("cannot load access source %s: %w", sourceID, err) + } + if source.OrganizationID != campaign.OrganizationID { + return fmt.Errorf("cannot process access source: %s does not belong to campaign organization", sourceID) + } + + var err error + driver, err = e.resolveDriver(ctx, conn, source) + if err != nil { + return fmt.Errorf("cannot resolve driver for source %s: %w", source.Name, err) + } + + lastCompletedCampaign := &coredata.AccessReviewCampaign{} + if err := lastCompletedCampaign.LoadLastCompletedByOrganizationID(ctx, conn, e.scope, campaign.OrganizationID); err != nil { + if !errors.Is(err, coredata.ErrResourceNotFound) { + return fmt.Errorf("cannot load last completed campaign: %w", err) + } + } else { + entries := &coredata.AccessEntries{} + baseline, err = entries.LoadBaselineBySourceID(ctx, conn, e.scope, lastCompletedCampaign.ID, sourceID) + if err != nil { + return fmt.Errorf("cannot load baseline entries by source: %w", err) + } + } + + return nil + }, + ) + if err != nil { + return 0, err + } + + previousByAccountKey := make(map[string]coredata.BaselineAccountEntry, len(baseline)) + for _, entry := range baseline { + previousByAccountKey[entry.AccountKey] = entry + } + + sourceCtx, cancel := context.WithTimeout(ctx, 30*time.Second) + accounts, err := driver.ListAccounts(sourceCtx) + cancel() + if err != nil { + return 0, fmt.Errorf("cannot list accounts from source %s: %w", source.Name, err) + } + fetchedCount = len(accounts) + + err = e.pg.WithTx( + ctx, + func(conn pg.Conn) error { + now := time.Now() + seenAccountKeys := make(map[string]struct{}, len(accounts)) + + for _, account := range accounts { + accountKey := normalizeAccountKey(account.Email, account.ExternalID) + seenAccountKeys[accountKey] = struct{}{} + incrementalTag := coredata.AccessEntryIncrementalTagNew + if _, ok := previousByAccountKey[accountKey]; ok { + incrementalTag = coredata.AccessEntryIncrementalTagUnchanged + } + + entry := &coredata.AccessEntry{ + ID: gid.New(e.scope.GetTenantID(), coredata.AccessEntryEntityType), + OrganizationID: campaign.OrganizationID, + AccessReviewCampaignID: campaign.ID, + AccessSourceID: sourceID, + Email: account.Email, + FullName: account.FullName, + Role: account.Role, + JobTitle: account.JobTitle, + IsAdmin: account.IsAdmin, + MFAStatus: account.MFAStatus, + AuthMethod: account.AuthMethod, + AccountType: account.AccountType, + LastLogin: account.LastLogin, + AccountCreatedAt: account.CreatedAt, + ExternalID: account.ExternalID, + AccountKey: accountKey, + IncrementalTag: incrementalTag, + Flags: []coredata.AccessEntryFlag{}, + FlagReasons: []string{}, + Decision: coredata.AccessEntryDecisionPending, + CreatedAt: now, + UpdatedAt: now, + } + + if err := entry.Upsert(ctx, conn, e.scope); err != nil { + return fmt.Errorf("cannot upsert access entry: %w", err) + } + } + + // Create REMOVED entries for accounts that existed in the previous + // campaign but are no longer present in the current fetch. + for accountKey, prev := range previousByAccountKey { + if _, seen := seenAccountKeys[accountKey]; seen { + continue + } + + entry := &coredata.AccessEntry{ + ID: gid.New(e.scope.GetTenantID(), coredata.AccessEntryEntityType), + OrganizationID: campaign.OrganizationID, + AccessReviewCampaignID: campaign.ID, + AccessSourceID: sourceID, + Email: prev.Email, + FullName: prev.FullName, + AccountKey: accountKey, + IncrementalTag: coredata.AccessEntryIncrementalTagRemoved, + Flags: []coredata.AccessEntryFlag{}, + FlagReasons: []string{}, + Decision: coredata.AccessEntryDecisionPending, + MFAStatus: coredata.MFAStatusUnknown, + AuthMethod: coredata.AccessEntryAuthMethodUnknown, + AccountType: coredata.AccessEntryAccountTypeUser, + CreatedAt: now, + UpdatedAt: now, + } + + if err := entry.Upsert(ctx, conn, e.scope); err != nil { + return fmt.Errorf("cannot upsert removed access entry: %w", err) + } + } + + return nil + }, + ) + if err != nil { + return 0, err + } + + return fetchedCount, nil +} + +func normalizeAccountKey(email, externalID string) string { + emailKey := strings.ToLower(strings.TrimSpace(email)) + externalID = strings.TrimSpace(externalID) + if externalID != "" { + return emailKey + "|" + externalID + } + + return emailKey +} + +// oauthClient returns an HTTP client for an OAuth2 connection, using +// RefreshableClient when a refresh config is available for the provider. +func (e *ReviewEngine) oauthClient( + ctx context.Context, + conn *connector.OAuth2Connection, + provider coredata.ConnectorProvider, +) (*http.Client, error) { + if e.connectorRegistry != nil { + refreshCfg := e.connectorRegistry.GetOAuth2RefreshConfig(string(provider)) + if refreshCfg != nil { + return conn.RefreshableClient(ctx, *refreshCfg) + } + } + return conn.Client(ctx) +} + +// connectorHTTPClient returns an HTTP client for the given connector. +// For OAuth2 connections it delegates to oauthClient so that token refresh +// is handled transparently. For other connection types it falls back to +// the standard Client method. +func (e *ReviewEngine) connectorHTTPClient( + ctx context.Context, + dbConnector *coredata.Connector, +) (*http.Client, error) { + if oauth2Conn, ok := dbConnector.Connection.(*connector.OAuth2Connection); ok { + return e.oauthClient(ctx, oauth2Conn, dbConnector.Provider) + } + return dbConnector.Connection.Client(ctx) +} + +// resolveDriver creates a Driver for the given AccessSource based on +// connector_id (null = built-in, set = connector-backed). +func (e *ReviewEngine) resolveDriver( + ctx context.Context, + conn pg.Conn, + source *coredata.AccessSource, +) (drivers.Driver, error) { + if source.ConnectorID == nil { + // CSV-backed source: use CSVDriver when csv_data is present + if source.CsvData != nil && *source.CsvData != "" { + return drivers.NewCSVDriver(strings.NewReader(*source.CsvData)), nil + } + + // Built-in driver: default to ProboMemberships + return drivers.NewProboMembershipsDriver(e.pg, e.scope, source.OrganizationID), nil + } + + // Connector-backed: look up the connector and resolve driver by provider + dbConnector := &coredata.Connector{} + if err := dbConnector.LoadByID(ctx, conn, e.scope, *source.ConnectorID, e.encryptionKey); err != nil { + return nil, fmt.Errorf("cannot load connector %s: %w", *source.ConnectorID, err) + } + + // Capture token before refresh to detect changes. + var tokenBefore string + if oauth2Conn, ok := dbConnector.Connection.(*connector.OAuth2Connection); ok { + tokenBefore = oauth2Conn.AccessToken + } + + // Build an HTTP client. For OAuth2 connections, use RefreshableClient + // so that short-lived tokens are transparently refreshed. + httpClient, err := e.connectorHTTPClient(ctx, dbConnector) + if err != nil { + return nil, fmt.Errorf("cannot create HTTP client for %s connector: %w", dbConnector.Provider, err) + } + + // Persist the refreshed token back to the database so subsequent + // calls (and other workers) use the updated credentials. Providers + // that rotate refresh tokens (HubSpot, DocuSign) will fail on the + // next poll if the old refresh token is reused. + if oauth2Conn, ok := dbConnector.Connection.(*connector.OAuth2Connection); ok { + if oauth2Conn.AccessToken != tokenBefore { + dbConnector.UpdatedAt = time.Now() + if err := dbConnector.Update(ctx, conn, e.scope, e.encryptionKey); err != nil { + return nil, fmt.Errorf("cannot persist refreshed token for connector %s: %w", *source.ConnectorID, err) + } + } + } + + switch dbConnector.Provider { + case coredata.ConnectorProviderGoogleWorkspace: + return drivers.NewGoogleWorkspaceDriver(httpClient), nil + case coredata.ConnectorProviderLinear: + return drivers.NewLinearDriver(httpClient), nil + case coredata.ConnectorProviderSlack: + return drivers.NewSlackDriver(httpClient), nil + case coredata.ConnectorProviderOnePassword: + // Client credentials grant -> Users API driver (to be created in Phase 5). + // Authorization code / SCIM grant -> existing SCIM-based driver. + if oauth2Conn, ok := dbConnector.Connection.(*connector.OAuth2Connection); ok && oauth2Conn.GrantType == connector.OAuth2GrantTypeClientCredentials { + settings, err := dbConnector.OnePasswordUsersAPISettings() + if err != nil { + return nil, fmt.Errorf("cannot read 1password users api settings: %w", err) + } + return drivers.NewOnePasswordUsersAPIDriver(httpClient, settings.AccountID, settings.Region), nil + } + onePasswordSettings, err := dbConnector.OnePasswordSettings() + if err != nil { + return nil, fmt.Errorf("cannot read 1password connector settings: %w", err) + } + if onePasswordSettings.SCIMBridgeURL == "" { + return nil, fmt.Errorf("1password connector requires scim_bridge_url in settings") + } + return drivers.NewOnePasswordDriver(httpClient, onePasswordSettings.SCIMBridgeURL), nil + case coredata.ConnectorProviderHubSpot: + return drivers.NewHubSpotDriver(httpClient), nil + case coredata.ConnectorProviderDocuSign: + return drivers.NewDocuSignDriver(httpClient), nil + case coredata.ConnectorProviderNotion: + return drivers.NewNotionDriver(httpClient), nil + case coredata.ConnectorProviderBrex: + return drivers.NewBrexDriver(httpClient), nil + case coredata.ConnectorProviderTally: + tallySettings, err := dbConnector.TallySettings() + if err != nil { + return nil, fmt.Errorf("cannot read tally connector settings: %w", err) + } + if tallySettings.OrganizationID == "" { + return nil, fmt.Errorf("tally connector requires organization_id in settings") + } + return drivers.NewTallyDriver(httpClient, tallySettings.OrganizationID), nil + case coredata.ConnectorProviderCloudflare: + return drivers.NewCloudflareDriver(httpClient), nil + case coredata.ConnectorProviderOpenAI: + return drivers.NewOpenAIDriver(httpClient), nil + case coredata.ConnectorProviderSentry: + sentrySettings, err := dbConnector.SentrySettings() + if err != nil { + return nil, fmt.Errorf("cannot read sentry connector settings: %w", err) + } + // OrganizationSlug may be empty for OAuth connections; the driver auto-discovers it. + return drivers.NewSentryDriver(httpClient, sentrySettings.OrganizationSlug), nil + case coredata.ConnectorProviderSupabase: + supabaseSettings, err := dbConnector.SupabaseSettings() + if err != nil { + return nil, fmt.Errorf("cannot read supabase connector settings: %w", err) + } + if supabaseSettings.OrganizationSlug == "" { + return nil, fmt.Errorf("supabase connector requires organization_slug in settings") + } + return drivers.NewSupabaseDriver(httpClient, supabaseSettings.OrganizationSlug), nil + case coredata.ConnectorProviderGitHub: + githubSettings, err := dbConnector.GitHubSettings() + if err != nil { + return nil, fmt.Errorf("cannot read github connector settings: %w", err) + } + if githubSettings.Organization == "" { + return nil, fmt.Errorf("github connector requires organization in settings") + } + return drivers.NewGitHubDriver(httpClient, githubSettings.Organization, e.logger.Named("github")), nil + case coredata.ConnectorProviderIntercom: + return drivers.NewIntercomDriver(httpClient), nil + case coredata.ConnectorProviderResend: + return drivers.NewResendDriver(httpClient), nil + default: + return nil, fmt.Errorf("unsupported connector provider %q for access source driver", dbConnector.Provider) + } +} diff --git a/pkg/accessreview/review_engine_test.go b/pkg/accessreview/review_engine_test.go new file mode 100644 index 000000000..49e8008f2 --- /dev/null +++ b/pkg/accessreview/review_engine_test.go @@ -0,0 +1,52 @@ +// 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 "testing" + +func TestNormalizeAccountKey(t *testing.T) { + t.Parallel() + + tests := []struct { + name string + email string + externalID string + want string + }{ + { + name: "email only", + email: " Jane@Example.com ", + externalID: "", + want: "jane@example.com", + }, + { + name: "email and external id", + email: "Jane@Example.com", + externalID: " 123 ", + want: "jane@example.com|123", + }, + } + + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + t.Parallel() + + got := normalizeAccountKey(tt.email, tt.externalID) + if got != tt.want { + t.Fatalf("normalizeAccountKey(%q, %q) = %q, want %q", tt.email, tt.externalID, got, tt.want) + } + }) + } +} diff --git a/pkg/accessreview/service.go b/pkg/accessreview/service.go new file mode 100644 index 000000000..d58370d0e --- /dev/null +++ b/pkg/accessreview/service.go @@ -0,0 +1,147 @@ +// Copyright (c) 2025-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" + "fmt" + "time" + + "go.gearno.de/kit/log" + "go.gearno.de/kit/pg" + "go.probo.inc/probo/pkg/connector" + "go.probo.inc/probo/pkg/coredata" + "go.probo.inc/probo/pkg/crypto/cipher" + "go.probo.inc/probo/pkg/gid" + "golang.org/x/sync/errgroup" +) + +type ( + Service struct { + pg *pg.Client + encryptionKey cipher.EncryptionKey + connectorRegistry *connector.ConnectorRegistry + logger *log.Logger + + worker *SourceFetchWorker + sourceNameWorker *SourceNameWorker + } + + Option func(*Service) +) + +func WithFetchInterval(interval time.Duration) Option { + return func(s *Service) { + s.worker.interval = interval + } +} + +func NewService( + pgClient *pg.Client, + encryptionKey cipher.EncryptionKey, + connectorRegistry *connector.ConnectorRegistry, + logger *log.Logger, + opts ...Option, +) *Service { + s := &Service{ + pg: pgClient, + encryptionKey: encryptionKey, + connectorRegistry: connectorRegistry, + logger: logger, + } + + s.worker = NewSourceFetchWorker(s, pgClient, logger) + s.sourceNameWorker = NewSourceNameWorker( + pgClient, + encryptionKey, + connectorRegistry, + logger.Named("source-name"), + ) + + for _, opt := range opts { + opt(s) + } + + return s +} + +// Sources returns a tenant-scoped AccessSourceService. +func (s *Service) Sources(scope coredata.Scoper) *AccessSourceService { + return &AccessSourceService{ + pg: s.pg, + scope: scope, + encryptionKey: s.encryptionKey, + connectorRegistry: s.connectorRegistry, + } +} + +// Campaigns returns a tenant-scoped CampaignService. +func (s *Service) Campaigns(scope coredata.Scoper) *CampaignService { + return NewCampaignService(s.pg, scope) +} + +// Entries returns a tenant-scoped AccessEntryService. +func (s *Service) Entries(scope coredata.Scoper) *AccessEntryService { + return &AccessEntryService{pg: s.pg, scope: scope} +} + +// Engine returns a tenant-scoped ReviewEngine. +func (s *Service) Engine(scope coredata.Scoper) *ReviewEngine { + return NewReviewEngine( + s.pg, + scope, + s.encryptionKey, + s.connectorRegistry, + s.logger.Named("review_engine"), + ) +} + +// ResolveEntryOrganizationID resolves the organization ID for an access entry. +// This is unscoped because it is used by resolvers before authorization to +// find the organization from an entry ID. +func (s *Service) ResolveEntryOrganizationID(ctx context.Context, entryID gid.GID) (gid.GID, error) { + var organizationID gid.GID + + err := s.pg.WithConn( + ctx, + func(conn pg.Conn) error { + var err error + entry := &coredata.AccessEntry{} + organizationID, err = entry.LoadOrganizationID(ctx, conn, entryID) + if err != nil { + return fmt.Errorf("cannot load organization id: %w", err) + } + return nil + }, + ) + if err != nil { + return gid.GID{}, fmt.Errorf("cannot resolve organization id: %w", err) + } + + return organizationID, nil +} + +func (s *Service) Run(ctx context.Context) error { + gCtx, cancel := context.WithCancel(context.WithoutCancel(ctx)) + g, gCtx := errgroup.WithContext(gCtx) + + g.Go(func() error { return s.worker.Run(gCtx) }) + g.Go(func() error { return s.sourceNameWorker.Run(gCtx) }) + + <-ctx.Done() + cancel() + + return g.Wait() +} diff --git a/pkg/probo/actions.go b/pkg/probo/actions.go index b06cedaf6..e3c7b5385 100644 --- a/pkg/probo/actions.go +++ b/pkg/probo/actions.go @@ -19,6 +19,7 @@ package probo const ( // Organization actions ActionOrganizationGet = "core:organization:get" + ActionOrganizationUpdate = "core:organization:update" ActionOrganizationGetLogoUrl = "core:organization:get-logo-url" ActionOrganizationGetHorizontalLogoUrl = "core:organization:get-horizontal-logo-url" @@ -307,6 +308,7 @@ const ( ActionSlackConnectionList = "core:slack-connection:list" // Connector actions (generic) + ActionConnectorCreate = "core:connector:create" ActionConnectorList = "core:connector:list" ActionConnectorDelete = "core:connector:delete" @@ -356,4 +358,30 @@ const ( ActionWebhookSubscriptionCreate = "core:webhook-subscription:create" ActionWebhookSubscriptionUpdate = "core:webhook-subscription:update" ActionWebhookSubscriptionDelete = "core:webhook-subscription:delete" + + // AccessReviewCampaign actions + ActionAccessReviewCampaignGet = "core:access-review-campaign:get" + ActionAccessReviewCampaignList = "core:access-review-campaign:list" + ActionAccessReviewCampaignCreate = "core:access-review-campaign:create" + ActionAccessReviewCampaignUpdate = "core:access-review-campaign:update" + ActionAccessReviewCampaignDelete = "core:access-review-campaign:delete" + ActionAccessReviewCampaignStart = "core:access-review-campaign:start" + ActionAccessReviewCampaignClose = "core:access-review-campaign:close" + ActionAccessReviewCampaignCancel = "core:access-review-campaign:cancel" + ActionAccessReviewCampaignAddScopeSource = "core:access-review-campaign:add-scope-source" + ActionAccessReviewCampaignRemoveScopeSource = "core:access-review-campaign:remove-scope-source" + + // AccessEntry actions + ActionAccessEntryGet = "core:access-entry:get" + ActionAccessEntryList = "core:access-entry:list" + ActionAccessEntryDecide = "core:access-entry:decide" + ActionAccessEntryFlag = "core:access-entry:flag" + + // AccessSource actions + ActionAccessSourceGet = "core:access-source:get" + ActionAccessSourceList = "core:access-source:list" + ActionAccessSourceCreate = "core:access-source:create" + ActionAccessSourceUpdate = "core:access-source:update" + ActionAccessSourceDelete = "core:access-source:delete" + ActionAccessSourceSync = "core:access-source:sync" ) diff --git a/pkg/probo/policies.go b/pkg/probo/policies.go index a978a0c38..cf2f84853 100644 --- a/pkg/probo/policies.go +++ b/pkg/probo/policies.go @@ -148,6 +148,9 @@ var ViewerPolicy = policy.NewPolicy( ActionStateOfApplicabilityGet, ActionStateOfApplicabilityList, ActionApplicabilityStatementGet, ActionApplicabilityStatementList, ActionWebhookSubscriptionGet, ActionWebhookSubscriptionList, + ActionAccessReviewCampaignGet, ActionAccessReviewCampaignList, + ActionAccessEntryGet, ActionAccessEntryList, + ActionAccessSourceGet, ActionAccessSourceList, ).WithSID("entity-read-access").When(organizationCondition), policy.Allow( diff --git a/pkg/probo/service.go b/pkg/probo/service.go index 8925a5b79..93bd5ed8b 100644 --- a/pkg/probo/service.go +++ b/pkg/probo/service.go @@ -24,6 +24,7 @@ import ( "go.gearno.de/kit/pg" "go.probo.inc/probo/pkg/agents" "go.probo.inc/probo/pkg/certmanager" + "go.probo.inc/probo/pkg/connector" "go.probo.inc/probo/pkg/coredata" "go.probo.inc/probo/pkg/crypto/cipher" "go.probo.inc/probo/pkg/esign" @@ -66,6 +67,7 @@ type ( logger *log.Logger slack *slack.Service esign *esign.Service + connectorRegistry *connector.ConnectorRegistry invitationTokenValidity time.Duration } @@ -141,6 +143,7 @@ func NewService( slackService *slack.Service, iamService *iam.Service, esignService *esign.Service, + connectorRegistry *connector.ConnectorRegistry, invitationTokenValidity time.Duration, ) (*Service, error) { if bucket == "" { @@ -166,6 +169,7 @@ func NewService( logger: logger, slack: slackService, esign: esignService, + connectorRegistry: connectorRegistry, invitationTokenValidity: invitationTokenValidity, } @@ -291,6 +295,7 @@ func (s *Service) WithTenant(tenantID gid.TenantID) *TenantService { logger: s.logger.Named("custom_domains"), } tenantService.SlackMessages = s.slack.WithTenant(tenantID).SlackMessages + return tenantService }