Add access review domain services

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

@@ -0,0 +1,471 @@
+// 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"
+	"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
+}

pkg/accessreview/access_source_service.go

@@ -0,0 +1,409 @@
+// 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"
+	"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
+}

pkg/accessreview/campaign_service.go

@@ -0,0 +1,528 @@
+// 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"
+	"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
+}

pkg/accessreview/campaign_types.go

@@ -0,0 +1,68 @@
+// 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 (
+	"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()
+}

pkg/accessreview/review_engine.go

@@ -0,0 +1,378 @@
+// 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"
+	"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)
+	}
+}

pkg/accessreview/review_engine_test.go

@@ -0,0 +1,52 @@
+// 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 "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)
+			}
+		})
+	}
+}

pkg/accessreview/service.go

@@ -0,0 +1,147 @@
+// Copyright (c) 2025-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"
+	"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()
+}

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"
 )

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(

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
 }