diff --git a/pkg/accessreview/campaign_service.go b/pkg/accessreview/campaign_service.go index ce26c4376..be68e769b 100644 --- a/pkg/accessreview/campaign_service.go +++ b/pkg/accessreview/campaign_service.go @@ -422,6 +422,7 @@ func (s *Service) upsertCampaignSource( campaignSource := &coredata.AccessReviewCampaignSource{ ID: gid.New(scope.GetTenantID(), coredata.AccessReviewCampaignSourceEntityType), + OrganizationID: source.OrganizationID, AccessReviewCampaignID: campaignID, AccessReviewSourceID: &sourceID, Name: source.Name, @@ -448,6 +449,7 @@ func (s *Service) enqueueSourceFetches( for _, campaignSource := range campaignSources { attempt := &coredata.AccessReviewCampaignSourceFetchAttempt{ ID: gid.New(scope.GetTenantID(), coredata.AccessReviewCampaignSourceFetchAttemptEntityType), + OrganizationID: campaignSource.OrganizationID, AccessReviewCampaignSourceID: campaignSource.ID, Status: coredata.AccessReviewCampaignSourceFetchStatusQueued, CreatedAt: now, diff --git a/pkg/coredata/access_review_campaign_source.go b/pkg/coredata/access_review_campaign_source.go index 303c71738..7c52ff099 100644 --- a/pkg/coredata/access_review_campaign_source.go +++ b/pkg/coredata/access_review_campaign_source.go @@ -35,6 +35,7 @@ type ( // entries and fetch attempts reference this snapshot, not the live source. AccessReviewCampaignSource struct { ID gid.GID `db:"id"` + OrganizationID gid.GID `db:"organization_id"` TenantID gid.TenantID `db:"tenant_id"` AccessReviewCampaignID gid.GID `db:"access_review_campaign_id"` AccessReviewSourceID *gid.GID `db:"access_review_source_id"` @@ -52,17 +53,7 @@ func (s *AccessReviewCampaignSource) AuthorizationAttributes( conn pg.Querier, resourceIDs []gid.GID, ) (policy.AttributesByID, error) { - q := ` -SELECT - cs.id, - c.organization_id -FROM - access_review_campaign_sources cs -JOIN - access_review_campaigns c ON c.id = cs.access_review_campaign_id -WHERE - cs.id = ANY(@resource_ids::text[]) -` + q := `SELECT id, organization_id FROM access_review_campaign_sources WHERE id = ANY(@resource_ids::text[])` args := pgx.StrictNamedArgs{ "resource_ids": resourceIDs, @@ -108,6 +99,7 @@ func (s *AccessReviewCampaignSource) Upsert( q := ` INSERT INTO access_review_campaign_sources ( id, + organization_id, tenant_id, access_review_campaign_id, access_review_source_id, @@ -117,6 +109,7 @@ INSERT INTO access_review_campaign_sources ( updated_at ) VALUES ( @id, + @organization_id, @tenant_id, @access_review_campaign_id, @access_review_source_id, @@ -133,6 +126,7 @@ RETURNING id ` args := pgx.StrictNamedArgs{ "id": s.ID, + "organization_id": s.OrganizationID, "tenant_id": scope.GetTenantID(), "access_review_campaign_id": s.AccessReviewCampaignID, "access_review_source_id": s.AccessReviewSourceID, @@ -158,6 +152,7 @@ func (s *AccessReviewCampaignSource) LoadByID( q := ` SELECT id, + organization_id, tenant_id, access_review_campaign_id, access_review_source_id, @@ -233,6 +228,7 @@ func (sources *AccessReviewCampaignSources) LoadByCampaignID( q := ` SELECT id, + organization_id, tenant_id, access_review_campaign_id, access_review_source_id, diff --git a/pkg/coredata/access_review_campaign_source_fetch_attempt.go b/pkg/coredata/access_review_campaign_source_fetch_attempt.go index 2a368dc81..d9f551d16 100644 --- a/pkg/coredata/access_review_campaign_source_fetch_attempt.go +++ b/pkg/coredata/access_review_campaign_source_fetch_attempt.go @@ -24,6 +24,7 @@ import ( "github.com/jackc/pgx/v5" "go.gearno.de/kit/pg" "go.probo.inc/probo/pkg/gid" + "go.probo.inc/probo/pkg/iam/policy" "go.probo.inc/probo/pkg/page" ) @@ -39,6 +40,7 @@ type ( // tenant to construct a Scope for subsequent operations. AccessReviewCampaignSourceFetchAttempt struct { ID gid.GID `db:"id"` + OrganizationID gid.GID `db:"organization_id"` TenantID gid.TenantID `db:"tenant_id"` AccessReviewCampaignSourceID gid.GID `db:"access_review_campaign_source_id"` Status AccessReviewCampaignSourceFetchStatus `db:"status"` @@ -69,6 +71,45 @@ var ( ErrNoAccessReviewCampaignSourceFetchAttemptAvailable = errors.New("no access review source fetch attempt available") ) +func (a *AccessReviewCampaignSourceFetchAttempt) AuthorizationAttributes( + ctx context.Context, + conn pg.Querier, + resourceIDs []gid.GID, +) (policy.AttributesByID, error) { + q := `SELECT id, organization_id FROM access_review_campaign_source_fetch_attempts WHERE id = ANY(@resource_ids::text[])` + + args := pgx.StrictNamedArgs{ + "resource_ids": resourceIDs, + } + + rows, err := conn.Query(ctx, q, args) + if err != nil { + return nil, fmt.Errorf("cannot query authorization attributes: %w", err) + } + + defer rows.Close() + + attrsByID := make(policy.AttributesByID) + + for rows.Next() { + var id, organizationID gid.GID + + if err := rows.Scan(&id, &organizationID); err != nil { + return nil, fmt.Errorf("cannot scan authorization attributes: %w", err) + } + + attrsByID[id] = policy.Attributes{ + "organization_id": organizationID.String(), + } + } + + if err := rows.Err(); err != nil { + return nil, fmt.Errorf("cannot iterate authorization attributes: %w", err) + } + + return attrsByID, nil +} + // Insert appends a new attempt for the snapshot, assigning the next // attempt_number atomically. The receiver's AttemptNumber is synced from the // database. @@ -80,6 +121,7 @@ func (a *AccessReviewCampaignSourceFetchAttempt) Insert( q := ` INSERT INTO access_review_campaign_source_fetch_attempts ( id, + organization_id, tenant_id, access_review_campaign_source_id, attempt_number, @@ -92,6 +134,7 @@ INSERT INTO access_review_campaign_source_fetch_attempts ( updated_at ) VALUES ( @id, + @organization_id, @tenant_id, @access_review_campaign_source_id, COALESCE(( @@ -111,6 +154,7 @@ RETURNING attempt_number ` args := pgx.StrictNamedArgs{ "id": a.ID, + "organization_id": a.OrganizationID, "tenant_id": scope.GetTenantID(), "access_review_campaign_source_id": a.AccessReviewCampaignSourceID, "status": a.Status, @@ -185,6 +229,7 @@ func (a *AccessReviewCampaignSourceFetchAttempt) LoadNextQueuedForUpdateSkipLock q := ` SELECT id, + organization_id, tenant_id, access_review_campaign_source_id, attempt_number, @@ -235,6 +280,7 @@ func (attempts *AccessReviewCampaignSourceFetchAttempts) LoadLatestByCampaignID( q := ` SELECT DISTINCT ON (access_review_campaign_source_id) id, + organization_id, tenant_id, access_review_campaign_source_id, attempt_number, @@ -286,6 +332,7 @@ func (attempts *AccessReviewCampaignSourceFetchAttempts) LoadByCampaignSourceID( q := ` SELECT id, + organization_id, tenant_id, access_review_campaign_source_id, status, @@ -299,6 +346,7 @@ SELECT FROM ( SELECT id, + organization_id, tenant_id, access_review_campaign_source_id, status, @@ -349,6 +397,7 @@ func (attempts *AccessReviewCampaignSourceFetchAttempts) LoadAllByCampaignSource q := ` SELECT id, + organization_id, tenant_id, access_review_campaign_source_id, attempt_number, @@ -424,6 +473,7 @@ func (attempts *AccessReviewCampaignSourceFetchAttempts) RecoverStale( q := ` SELECT id, + organization_id, tenant_id, access_review_campaign_source_id, attempt_number, @@ -466,6 +516,7 @@ FOR UPDATE SKIP LOCKED retry := &AccessReviewCampaignSourceFetchAttempt{ ID: gid.New(attempt.TenantID, AccessReviewCampaignSourceFetchAttemptEntityType), + OrganizationID: attempt.OrganizationID, AccessReviewCampaignSourceID: attempt.AccessReviewCampaignSourceID, Status: AccessReviewCampaignSourceFetchStatusQueued, CreatedAt: now, diff --git a/pkg/coredata/access_review_campaign_source_test.go b/pkg/coredata/access_review_campaign_source_test.go index 2db48ed26..83b7deee6 100644 --- a/pkg/coredata/access_review_campaign_source_test.go +++ b/pkg/coredata/access_review_campaign_source_test.go @@ -111,6 +111,7 @@ func TestSourceFetchAttempts_AppendOnly(t *testing.T) { require.NoError(t, client.WithTx(ctx, func(ctx context.Context, tx pg.Tx) error { first := &coredata.AccessReviewCampaignSourceFetchAttempt{ ID: gid.New(tenantID, coredata.AccessReviewCampaignSourceFetchAttemptEntityType), + OrganizationID: fx.organizationID, AccessReviewCampaignSourceID: fx.campaignSourceID, Status: coredata.AccessReviewCampaignSourceFetchStatusFailed, Error: &failureMsg, @@ -124,6 +125,7 @@ func TestSourceFetchAttempts_AppendOnly(t *testing.T) { second := &coredata.AccessReviewCampaignSourceFetchAttempt{ ID: gid.New(tenantID, coredata.AccessReviewCampaignSourceFetchAttemptEntityType), + OrganizationID: fx.organizationID, AccessReviewCampaignSourceID: fx.campaignSourceID, Status: coredata.AccessReviewCampaignSourceFetchStatusSuccess, FetchedAccountsCount: 7, diff --git a/pkg/coredata/access_review_entry_upsert_test.go b/pkg/coredata/access_review_entry_upsert_test.go index 0be9d7d4f..46f95b3b9 100644 --- a/pkg/coredata/access_review_entry_upsert_test.go +++ b/pkg/coredata/access_review_entry_upsert_test.go @@ -87,6 +87,7 @@ func seedAccessReviewEntryFixture(t *testing.T, ctx context.Context, client *pg. campaignSource := &coredata.AccessReviewCampaignSource{ ID: campaignSourceID, + OrganizationID: organizationID, TenantID: tenantID, AccessReviewCampaignID: campaignID, AccessReviewSourceID: &sourceID, diff --git a/pkg/coredata/migrations/20260615T130000Z.sql b/pkg/coredata/migrations/20260615T130000Z.sql new file mode 100644 index 000000000..d4e0c042a --- /dev/null +++ b/pkg/coredata/migrations/20260615T130000Z.sql @@ -0,0 +1,40 @@ +-- 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. + +-- Add organization_id to campaign source snapshots and fetch attempts to +-- avoid JOINs in AuthorizationAttributes lookups. + +-- 1. access_review_campaign_sources +ALTER TABLE access_review_campaign_sources + ADD COLUMN organization_id TEXT REFERENCES organizations(id); + +UPDATE access_review_campaign_sources cs +SET organization_id = c.organization_id +FROM access_review_campaigns c +WHERE cs.access_review_campaign_id = c.id; + +ALTER TABLE access_review_campaign_sources + ALTER COLUMN organization_id SET NOT NULL; + +-- 2. access_review_campaign_source_fetch_attempts +ALTER TABLE access_review_campaign_source_fetch_attempts + ADD COLUMN organization_id TEXT REFERENCES organizations(id); + +UPDATE access_review_campaign_source_fetch_attempts fa +SET organization_id = cs.organization_id +FROM access_review_campaign_sources cs +WHERE fa.access_review_campaign_source_id = cs.id; + +ALTER TABLE access_review_campaign_source_fetch_attempts + ALTER COLUMN organization_id SET NOT NULL;