From abbb1ee87580a3e300f5b77547fa3bca00e0b52b Mon Sep 17 00:00:00 2001 From: Cursor Agent Date: Mon, 27 Jul 2026 14:19:08 +0000 Subject: [PATCH] Merge campaign scope sources in one SQL statement Add AccessReviewCampaignSources.MergeByCampaignID and use it from syncCampaignSources instead of per-row load, upsert, and delete. Signed-off-by: Cursor Agent Co-authored-by: Bryan FRIMIN --- pkg/accessreview/campaign_service.go | 57 +------- pkg/coredata/access_review_campaign_source.go | 137 ++++++++++++++++++ 2 files changed, 141 insertions(+), 53 deletions(-) diff --git a/pkg/accessreview/campaign_service.go b/pkg/accessreview/campaign_service.go index 6658af7aa..760177e1c 100644 --- a/pkg/accessreview/campaign_service.go +++ b/pkg/accessreview/campaign_service.go @@ -307,66 +307,17 @@ func (s *Service) syncCampaignSources( sourceIDs []gid.GID, ) error { var campaignSources coredata.AccessReviewCampaignSources - if err := campaignSources.LoadByCampaignID(ctx, conn, scope, campaign.ID); err != nil { - return fmt.Errorf("cannot load campaign sources: %w", err) - } - - existingSourceIDs := make([]gid.GID, 0, len(campaignSources)) - for _, campaignSource := range campaignSources { - if campaignSource.AccessReviewSourceID != nil { - existingSourceIDs = append(existingSourceIDs, *campaignSource.AccessReviewSourceID) - } - } - - for _, sourceID := range sourceIDs { - if containsGID(existingSourceIDs, sourceID) { - continue + if err := campaignSources.MergeByCampaignID(ctx, conn, scope, campaign.ID, sourceIDs); err != nil { + if errors.Is(err, coredata.ErrResourceNotFound) { + return coredata.ErrResourceNotFound } - 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) - } - - if err := s.upsertCampaignSource(ctx, conn, scope, campaign.ID, source); err != nil { - return fmt.Errorf("cannot snapshot scope source: %w", err) - } - } - - for _, existingSourceID := range existingSourceIDs { - if containsGID(sourceIDs, existingSourceID) { - continue - } - - campaignSource := &coredata.AccessReviewCampaignSource{} - if err := campaignSource.DeleteByCampaignIDAndAccessReviewSourceID( - ctx, - conn, - scope, - campaign.ID, - existingSourceID, - ); err != nil { - return fmt.Errorf("cannot delete campaign source: %w", err) - } + return fmt.Errorf("cannot merge campaign sources: %w", err) } return nil } -func containsGID(ids []gid.GID, id gid.GID) bool { - for _, candidate := range ids { - if candidate == id { - return true - } - } - - return false -} - func (s *Service) StartCampaign( ctx context.Context, scope coredata.Scoper, diff --git a/pkg/coredata/access_review_campaign_source.go b/pkg/coredata/access_review_campaign_source.go index 48f342451..8c64f5d9a 100644 --- a/pkg/coredata/access_review_campaign_source.go +++ b/pkg/coredata/access_review_campaign_source.go @@ -149,6 +149,143 @@ 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. +func (sources *AccessReviewCampaignSources) MergeByCampaignID( + ctx context.Context, + conn pg.Tx, + scope Scoper, + campaignID gid.GID, + accessReviewSourceIDs []gid.GID, +) error { + uniqueSourceIDs := uniqueGIDs(accessReviewSourceIDs) + + if len(uniqueSourceIDs) > 0 { + countQ := ` +SELECT COUNT(DISTINCT id) +FROM access_review_sources +WHERE + %s + AND id = ANY(@access_review_source_ids::text[]) +` + countQ = fmt.Sprintf(countQ, scope.SQLFragment()) + + sourceIDStrings := make([]string, len(uniqueSourceIDs)) + for i, id := range uniqueSourceIDs { + sourceIDStrings[i] = id.String() + } + + countArgs := pgx.StrictNamedArgs{"access_review_source_ids": sourceIDStrings} + maps.Copy(countArgs, scope.SQLArguments()) + + var found int + if err := conn.QueryRow(ctx, countQ, countArgs).Scan(&found); err != nil { + return fmt.Errorf("cannot count access review sources: %w", err) + } + + if found != len(uniqueSourceIDs) { + return ErrResourceNotFound + } + } + + sourceIDStrings := make([]string, len(uniqueSourceIDs)) + for i, id := range uniqueSourceIDs { + sourceIDStrings[i] = id.String() + } + + now := time.Now() + + q := ` +WITH desired_sources AS ( + SELECT + id AS access_review_source_id, + organization_id, + name, + connector_id + FROM access_review_sources + WHERE + %s + AND id = ANY(@access_review_source_ids::text[]) +) +MERGE INTO access_review_campaign_sources AS target +USING desired_sources AS source +ON + %s + AND target.access_review_campaign_id = @access_review_campaign_id + AND target.access_review_source_id = source.access_review_source_id +WHEN MATCHED THEN + UPDATE SET + name = source.name, + connector_id = source.connector_id, + updated_at = @now +WHEN NOT MATCHED THEN + INSERT ( + id, + organization_id, + tenant_id, + access_review_campaign_id, + access_review_source_id, + name, + connector_id, + created_at, + updated_at + ) + VALUES ( + generate_gid(decode_base64_unpadded(@tenant_id), @access_review_campaign_source_entity_type), + source.organization_id, + @tenant_id, + @access_review_campaign_id, + source.access_review_source_id, + source.name, + source.connector_id, + @now, + @now + ) +WHEN NOT MATCHED BY SOURCE + AND %s + AND target.access_review_campaign_id = @access_review_campaign_id THEN + DELETE +` + + q = fmt.Sprintf(q, scope.SQLFragment(), scope.SQLFragment(), scope.SQLFragment()) + + args := pgx.StrictNamedArgs{ + "access_review_campaign_id": campaignID, + "access_review_source_ids": sourceIDStrings, + "access_review_campaign_source_entity_type": AccessReviewCampaignSourceEntityType, + "tenant_id": scope.GetTenantID(), + "now": now, + } + maps.Copy(args, scope.SQLArguments()) + + if _, err := conn.Exec(ctx, q, args); err != nil { + return fmt.Errorf("cannot merge campaign sources: %w", err) + } + + return nil +} + +func uniqueGIDs(ids []gid.GID) []gid.GID { + if len(ids) == 0 { + return nil + } + + seen := make(map[gid.GID]struct{}, len(ids)) + unique := make([]gid.GID, 0, len(ids)) + + for _, id := range ids { + if _, ok := seen[id]; ok { + continue + } + + seen[id] = struct{}{} + unique = append(unique, id) + } + + return unique +} + func (s *AccessReviewCampaignSource) LoadByID( ctx context.Context, conn pg.Querier,