Fix missing organization id

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

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,

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 := `
+	q := `SELECT id, organization_id FROM access_review_campaign_sources WHERE id = ANY(@resource_ids::text[])`
-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[])
-`
 
 	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,

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,

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,

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,

pkg/coredata/migrations/20260615T130000Z.sql

@@ -0,0 +1,40 @@
+-- Copyright (c) 2026 Probo Inc <hello@probo.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.
+
+-- 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;