Batch-load and validate campaign sources before merging

MergeByCampaignID joined access_review_sources directly from coredata
to resolve live sources, so unrecognized or out-of-scope IDs were
silently dropped instead of erroring, and in the worst case (every ID
invalid) the NOT MATCHED BY SOURCE clause deleted every existing
campaign source. The syncCampaignSources ErrResourceNotFound check
was therefore unreachable dead code.

Add AccessReviewSources.LoadByIDs, matching the existing scoped
LoadByIDs pattern (id = ANY(@ids) plus a resolved-count check), and
have CreateCampaign, AddCampaignSource, and syncCampaignSources
resolve and validate sources up front. MergeByCampaignID now takes
the already-loaded sources and builds its desired-state CTE from an
unnest() of their values instead of joining access_review_sources,
keeping the merge inside the campaign-source entity boundary.

Signed-off-by: Bryan Frimin <bryan@probo.com>

pkg/accessreview/campaign_service.go

@@ -59,16 +59,16 @@ func (s *Service) CreateCampaign(
 				return fmt.Errorf("cannot insert access review campaign: %w", err)
 			}
 
-			for _, sourceID := range req.AccessReviewSourceIDs {
+			var sources coredata.AccessReviewSources
-				source := &coredata.AccessReviewSource{}
+			if err := sources.LoadByIDs(ctx, conn, scope, req.AccessReviewSourceIDs); err != nil {
-				if err := source.LoadByID(ctx, conn, scope, sourceID); err != nil {
+				if errors.Is(err, coredata.ErrResourceNotFound) {
-					if errors.Is(err, coredata.ErrResourceNotFound) {
+					return coredata.ErrResourceNotFound
-						return coredata.ErrResourceNotFound
-					}
-
-					return fmt.Errorf("cannot load access source: %w", err)
 				}
 
+				return fmt.Errorf("cannot load access sources: %w", err)
+			}
+
+			for _, source := range sources {
 				if err := s.upsertCampaignSource(ctx, conn, scope, campaign.ID, source); err != nil {
 					return fmt.Errorf("cannot snapshot scope source: %w", err)
 				}
@@ -308,12 +308,17 @@ func (s *Service) syncCampaignSources(
 	campaign *coredata.AccessReviewCampaign,
 	sourceIDs []gid.GID,
 ) error {
-	var campaignSources coredata.AccessReviewCampaignSources
+	var sources coredata.AccessReviewSources
-	if err := campaignSources.MergeByCampaignID(ctx, conn, scope, campaign.ID, sourceIDs); err != nil {
+	if err := sources.LoadByIDs(ctx, conn, scope, sourceIDs); err != nil {
 		if errors.Is(err, coredata.ErrResourceNotFound) {
 			return coredata.ErrResourceNotFound
 		}
 
+		return fmt.Errorf("cannot load access sources: %w", err)
+	}
+
+	var campaignSources coredata.AccessReviewCampaignSources
+	if err := campaignSources.MergeByCampaignID(ctx, conn, scope, campaign.ID, sources); err != nil {
 		return fmt.Errorf("cannot merge campaign sources: %w", err)
 	}
 

pkg/coredata/access_review_campaign_source.go

@@ -149,21 +149,40 @@ RETURNING id
 	return nil
 }
 
-// MergeByCampaignID syncs scoped access-review source snapshots for a campaign:
+// MergeByCampaignID syncs scoped campaign source snapshots against the given,
-// upserts snapshots for the given live source IDs and deletes snapshots no
+// already-loaded and validated live access sources: upserts a snapshot for
-// longer in the set.
+// each source and deletes snapshots no longer in the set. Callers are
+// responsible for resolving accessReviewSources (existence, tenant, and
+// organization checks) before calling this method.
 func (sources *AccessReviewCampaignSources) MergeByCampaignID(
 	ctx context.Context,
 	conn pg.Tx,
 	scope Scoper,
 	campaignID gid.GID,
-	accessReviewSourceIDs []gid.GID,
+	accessReviewSources AccessReviewSources,
 ) error {
-	sourceIDs := gid.NewSet(accessReviewSourceIDs...)
+	ids := make([]string, 0, len(accessReviewSources))
+	organizationIDs := make([]string, 0, len(accessReviewSources))
+	names := make([]string, 0, len(accessReviewSources))
+	connectorIDs := make([]*string, 0, len(accessReviewSources))
 
-	sourceIDStrings := make([]string, 0, len(sourceIDs))
+	seen := make(map[gid.GID]struct{}, len(accessReviewSources))
-	for id := range sourceIDs {
+	for _, source := range accessReviewSources {
-		sourceIDStrings = append(sourceIDStrings, id.String())
+		if _, ok := seen[source.ID]; ok {
+			continue
+		}
+		seen[source.ID] = struct{}{}
+
+		var connectorID *string
+		if source.ConnectorID != nil {
+			s := source.ConnectorID.String()
+			connectorID = &s
+		}
+
+		ids = append(ids, source.ID.String())
+		organizationIDs = append(organizationIDs, source.OrganizationID.String())
+		names = append(names, source.Name)
+		connectorIDs = append(connectorIDs, connectorID)
 	}
 
 	now := time.Now()
@@ -175,10 +194,12 @@ WITH desired_sources AS (
 		organization_id,
 		name,
 		connector_id
-	FROM access_review_sources
+	FROM unnest(
-	WHERE
+		@access_review_source_ids::text[],
-		%s
+		@organization_ids::text[],
-		AND id = ANY(@access_review_source_ids::text[])
+		@names::text[],
+		@connector_ids::text[]
+	) AS t(id, organization_id, name, connector_id)
 )
 MERGE INTO access_review_campaign_sources AS target
 USING desired_sources AS source
@@ -220,11 +241,14 @@ WHEN NOT MATCHED BY SOURCE
 	DELETE
 `
 
-	q = fmt.Sprintf(q, scope.SQLFragment(), scope.SQLFragment(), scope.SQLFragment())
+	q = fmt.Sprintf(q, scope.SQLFragment(), scope.SQLFragment())
 
 	args := pgx.StrictNamedArgs{
-		"access_review_campaign_id":                 campaignID,
+		"access_review_campaign_id": campaignID,
-		"access_review_source_ids":                  sourceIDStrings,
+		"access_review_source_ids":  ids,
+		"organization_ids":          organizationIDs,
+		"names":                     names,
+		"connector_ids":             connectorIDs,
 		"access_review_campaign_source_entity_type": AccessReviewCampaignSourceEntityType,
 		"tenant_id": scope.GetTenantID(),
 		"now":       now,

pkg/coredata/access_review_source.go

@@ -144,6 +144,52 @@ LIMIT 1;
 	return nil
 }
 
+func (sources *AccessReviewSources) LoadByIDs(
+	ctx context.Context,
+	conn pg.Querier,
+	scope Scoper,
+	ids []gid.GID,
+) error {
+	q := `
+SELECT
+    id,
+    organization_id,
+    connector_id,
+    name,
+    csv_data,
+    name_synced_at,
+    created_at,
+    updated_at
+FROM
+    access_review_sources
+WHERE
+    %s
+    AND id = ANY(@ids)
+`
+	q = fmt.Sprintf(q, scope.SQLFragment())
+
+	args := pgx.StrictNamedArgs{"ids": ids}
+	maps.Copy(args, scope.SQLArguments())
+
+	rows, err := conn.Query(ctx, q, args)
+	if err != nil {
+		return fmt.Errorf("cannot query access_review_sources: %w", err)
+	}
+
+	result, err := pgx.CollectRows(rows, pgx.RowToAddrOfStructByName[AccessReviewSource])
+	if err != nil {
+		return fmt.Errorf("cannot collect access_review_sources: %w", err)
+	}
+
+	*sources = result
+
+	if len(result) != len(gid.NewSet(ids...)) {
+		return ErrResourceNotFound
+	}
+
+	return nil
+}
+
 func (as *AccessReviewSource) Insert(
 	ctx context.Context,
 	conn pg.Tx,