From 68a64647bf5075ce14f90e54f8e519113772ca62 Mon Sep 17 00:00:00 2001 From: Bryan Frimin Date: Mon, 27 Jul 2026 18:30:55 +0200 Subject: [PATCH] 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 --- pkg/accessreview/campaign_service.go | 25 +++++---- pkg/coredata/access_review_campaign_source.go | 54 +++++++++++++------ pkg/coredata/access_review_source.go | 46 ++++++++++++++++ 3 files changed, 100 insertions(+), 25 deletions(-) diff --git a/pkg/accessreview/campaign_service.go b/pkg/accessreview/campaign_service.go index cbd3baa56..4ca5ecb8a 100644 --- a/pkg/accessreview/campaign_service.go +++ b/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 { - source := &coredata.AccessReviewSource{} - if err := source.LoadByID(ctx, conn, scope, sourceID); err != nil { - if errors.Is(err, coredata.ErrResourceNotFound) { - return coredata.ErrResourceNotFound - } - - return fmt.Errorf("cannot load access source: %w", err) + var sources coredata.AccessReviewSources + if err := sources.LoadByIDs(ctx, conn, scope, req.AccessReviewSourceIDs); err != nil { + if errors.Is(err, coredata.ErrResourceNotFound) { + return coredata.ErrResourceNotFound } + 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 - if err := campaignSources.MergeByCampaignID(ctx, conn, scope, campaign.ID, sourceIDs); err != nil { + var sources coredata.AccessReviewSources + 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) } diff --git a/pkg/coredata/access_review_campaign_source.go b/pkg/coredata/access_review_campaign_source.go index 621414050..ea55d56c0 100644 --- a/pkg/coredata/access_review_campaign_source.go +++ b/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: -// upserts snapshots for the given live source IDs and deletes snapshots no -// longer in the set. +// MergeByCampaignID syncs scoped campaign source snapshots against the given, +// already-loaded and validated live access sources: upserts a snapshot for +// 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)) - for id := range sourceIDs { - sourceIDStrings = append(sourceIDStrings, id.String()) + seen := make(map[gid.GID]struct{}, len(accessReviewSources)) + for _, source := range accessReviewSources { + 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 - WHERE - %s - AND id = ANY(@access_review_source_ids::text[]) + FROM unnest( + @access_review_source_ids::text[], + @organization_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_source_ids": sourceIDStrings, + "access_review_campaign_id": campaignID, + "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, diff --git a/pkg/coredata/access_review_source.go b/pkg/coredata/access_review_source.go index 0f4a50ea6..6721d2026 100644 --- a/pkg/coredata/access_review_source.go +++ b/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,