Add access review data layer and migrations

Add coredata entities for access review campaigns,
access sources, access entries with decision history,
campaign source fetches, and scope systems. Include
migrations, entity type registrations, enum types for
flags, decisions, MFA status, and auth methods.

Signed-off-by: Aurélien Sibiril <81782+aureliensibiril@users.noreply.github.com>

pkg/coredata/access_entry.go

@@ -0,0 +1,803 @@
+// Copyright (c) 2026 Probo Inc <hello@getprobo.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.
+
+package coredata
+
+import (
+	"context"
+	"errors"
+	"fmt"
+	"maps"
+	"time"
+
+	"github.com/jackc/pgx/v5"
+	"go.gearno.de/kit/pg"
+	"go.probo.inc/probo/pkg/gid"
+	"go.probo.inc/probo/pkg/page"
+)
+
+type (
+	AccessEntry struct {
+		ID                     gid.GID                   `db:"id"`
+		OrganizationID         gid.GID                   `db:"organization_id"`
+		AccessReviewCampaignID gid.GID                   `db:"access_review_campaign_id"`
+		AccessSourceID         gid.GID                   `db:"access_source_id"`
+		IdentityID             *gid.GID                  `db:"identity_id"`
+		Email                  string                    `db:"email"`
+		FullName               string                    `db:"full_name"`
+		Role                   string                    `db:"role"`
+		JobTitle               string                    `db:"job_title"`
+		IsAdmin                bool                      `db:"is_admin"`
+		MFAStatus              MFAStatus                 `db:"mfa_status"`
+		AuthMethod             AccessEntryAuthMethod     `db:"auth_method"`
+		AccountType            AccessEntryAccountType    `db:"account_type"`
+		LastLogin              *time.Time                `db:"last_login"`
+		AccountCreatedAt       *time.Time                `db:"account_created_at"`
+		ExternalID             string                    `db:"external_id"`
+		AccountKey             string                    `db:"account_key"`
+		IncrementalTag         AccessEntryIncrementalTag `db:"incremental_tag"`
+		Flags                  []AccessEntryFlag         `db:"flags"`
+		FlagReasons            []string                  `db:"flag_reasons"`
+		Decision               AccessEntryDecision       `db:"decision"`
+		DecisionNote           *string                   `db:"decision_note"`
+		DecidedBy              *gid.GID                  `db:"decided_by"`
+		DecidedAt              *time.Time                `db:"decided_at"`
+		CreatedAt              time.Time                 `db:"created_at"`
+		UpdatedAt              time.Time                 `db:"updated_at"`
+	}
+
+	AccessEntries []*AccessEntry
+)
+
+func (e AccessEntry) CursorKey(orderBy AccessEntryOrderField) page.CursorKey {
+	switch orderBy {
+	case AccessEntryOrderFieldCreatedAt:
+		return page.NewCursorKey(e.ID, e.CreatedAt)
+	}
+
+	panic(fmt.Sprintf("unsupported order by: %s", orderBy))
+}
+
+func (e *AccessEntry) AuthorizationAttributes(ctx context.Context, conn pg.Conn) (map[string]string, error) {
+	q := `SELECT organization_id FROM access_entries WHERE id = $1 LIMIT 1;`
+
+	var organizationID gid.GID
+	if err := conn.QueryRow(ctx, q, e.ID).Scan(&organizationID); err != nil {
+		if errors.Is(err, pgx.ErrNoRows) {
+			return nil, ErrResourceNotFound
+		}
+		return nil, fmt.Errorf("cannot query access entry authorization attributes: %w", err)
+	}
+
+	return map[string]string{"organization_id": organizationID.String()}, nil
+}
+
+func (e *AccessEntry) LoadByID(
+	ctx context.Context,
+	conn pg.Conn,
+	scope Scoper,
+	id gid.GID,
+) error {
+	q := `
+SELECT
+    id,
+    organization_id,
+    access_review_campaign_id,
+    access_source_id,
+    identity_id,
+    email,
+    full_name,
+    role,
+    job_title,
+    is_admin,
+    mfa_status,
+    auth_method,
+    account_type,
+    last_login,
+    account_created_at,
+    external_id,
+    account_key,
+    incremental_tag,
+    flags,
+    flag_reasons,
+    decision,
+    decision_note,
+    decided_by,
+    decided_at,
+    created_at,
+    updated_at
+FROM
+    access_entries
+WHERE
+    %s
+    AND id = @id
+LIMIT 1;
+`
+	q = fmt.Sprintf(q, scope.SQLFragment())
+
+	args := pgx.StrictNamedArgs{"id": id}
+	maps.Copy(args, scope.SQLArguments())
+
+	rows, err := conn.Query(ctx, q, args)
+	if err != nil {
+		return fmt.Errorf("cannot query access_entries: %w", err)
+	}
+
+	entry, err := pgx.CollectExactlyOneRow(rows, pgx.RowToStructByName[AccessEntry])
+	if err != nil {
+		if errors.Is(err, pgx.ErrNoRows) {
+			return ErrResourceNotFound
+		}
+		return fmt.Errorf("cannot collect access entry: %w", err)
+	}
+
+	*e = entry
+
+	return nil
+}
+
+func (e *AccessEntry) Insert(
+	ctx context.Context,
+	conn pg.Conn,
+	scope Scoper,
+) error {
+	q := `
+INSERT INTO
+    access_entries (
+        id,
+        tenant_id,
+        organization_id,
+        access_review_campaign_id,
+        access_source_id,
+        identity_id,
+        email,
+        full_name,
+        role,
+        job_title,
+        is_admin,
+        mfa_status,
+        auth_method,
+        account_type,
+        last_login,
+        account_created_at,
+        external_id,
+        account_key,
+        incremental_tag,
+        flags,
+        flag_reasons,
+        decision,
+        decision_note,
+        decided_by,
+        decided_at,
+        created_at,
+        updated_at
+    )
+VALUES (
+    @id,
+    @tenant_id,
+    @organization_id,
+    @access_review_campaign_id,
+    @access_source_id,
+    @identity_id,
+    @email,
+    @full_name,
+    @role,
+    @job_title,
+    @is_admin,
+    @mfa_status,
+    @auth_method,
+    @account_type,
+    @last_login,
+    @account_created_at,
+    @external_id,
+    @account_key,
+    @incremental_tag,
+    @flags,
+    @flag_reasons,
+    @decision,
+    @decision_note,
+    @decided_by,
+    @decided_at,
+    @created_at,
+    @updated_at
+);
+`
+
+	args := pgx.StrictNamedArgs{
+		"id":                        e.ID,
+		"tenant_id":                 scope.GetTenantID(),
+		"organization_id":           e.OrganizationID,
+		"access_review_campaign_id": e.AccessReviewCampaignID,
+		"access_source_id":          e.AccessSourceID,
+		"identity_id":               e.IdentityID,
+		"email":                     e.Email,
+		"full_name":                 e.FullName,
+		"role":                      e.Role,
+		"job_title":                 e.JobTitle,
+		"is_admin":                  e.IsAdmin,
+		"mfa_status":                e.MFAStatus,
+		"auth_method":               e.AuthMethod,
+		"account_type":              e.AccountType,
+		"last_login":                e.LastLogin,
+		"account_created_at":        e.AccountCreatedAt,
+		"external_id":               e.ExternalID,
+		"account_key":               e.AccountKey,
+		"incremental_tag":           e.IncrementalTag,
+		"flags":                     e.Flags,
+		"flag_reasons":              e.FlagReasons,
+		"decision":                  e.Decision,
+		"decision_note":             e.DecisionNote,
+		"decided_by":                e.DecidedBy,
+		"decided_at":                e.DecidedAt,
+		"created_at":                e.CreatedAt,
+		"updated_at":                e.UpdatedAt,
+	}
+	_, err := conn.Exec(ctx, q, args)
+	if err != nil {
+		return fmt.Errorf("cannot insert access_entry: %w", err)
+	}
+
+	return nil
+}
+
+func (e *AccessEntry) Update(
+	ctx context.Context,
+	conn pg.Conn,
+	scope Scoper,
+) error {
+	q := `
+UPDATE access_entries
+SET
+    flags = @flags,
+    flag_reasons = @flag_reasons,
+    decision = @decision,
+    decision_note = @decision_note,
+    decided_by = @decided_by,
+    decided_at = @decided_at,
+    updated_at = @updated_at
+WHERE
+    %s
+    AND id = @id
+`
+	q = fmt.Sprintf(q, scope.SQLFragment())
+
+	args := pgx.StrictNamedArgs{
+		"id":            e.ID,
+		"flags":         e.Flags,
+		"flag_reasons":  e.FlagReasons,
+		"decision":      e.Decision,
+		"decision_note": e.DecisionNote,
+		"decided_by":    e.DecidedBy,
+		"decided_at":    e.DecidedAt,
+		"updated_at":    e.UpdatedAt,
+	}
+	maps.Copy(args, scope.SQLArguments())
+
+	result, err := conn.Exec(ctx, q, args)
+	if err != nil {
+		return fmt.Errorf("cannot update access_entry: %w", err)
+	}
+
+	if result.RowsAffected() == 0 {
+		return ErrResourceNotFound
+	}
+
+	return nil
+}
+
+func (entries *AccessEntries) LoadByCampaignID(
+	ctx context.Context,
+	conn pg.Conn,
+	scope Scoper,
+	campaignID gid.GID,
+	cursor *page.Cursor[AccessEntryOrderField],
+	filter *AccessEntryFilter,
+) error {
+	q := `
+SELECT
+    id,
+    organization_id,
+    access_review_campaign_id,
+    access_source_id,
+    identity_id,
+    email,
+    full_name,
+    role,
+    job_title,
+    is_admin,
+    mfa_status,
+    auth_method,
+    account_type,
+    last_login,
+    account_created_at,
+    external_id,
+    account_key,
+    incremental_tag,
+    flags,
+    flag_reasons,
+    decision,
+    decision_note,
+    decided_by,
+    decided_at,
+    created_at,
+    updated_at
+FROM
+    access_entries
+WHERE
+    %s
+    AND access_review_campaign_id = @campaign_id
+    AND %s
+    AND %s
+`
+	q = fmt.Sprintf(q, scope.SQLFragment(), filter.SQLFragment(), cursor.SQLFragment())
+
+	args := pgx.StrictNamedArgs{"campaign_id": campaignID}
+	maps.Copy(args, scope.SQLArguments())
+	maps.Copy(args, filter.SQLArguments())
+	maps.Copy(args, cursor.SQLArguments())
+
+	rows, err := conn.Query(ctx, q, args)
+	if err != nil {
+		return fmt.Errorf("cannot query access_entries: %w", err)
+	}
+
+	result, err := pgx.CollectRows(rows, pgx.RowToAddrOfStructByName[AccessEntry])
+	if err != nil {
+		return fmt.Errorf("cannot collect access_entries: %w", err)
+	}
+
+	*entries = result
+
+	return nil
+}
+
+func (entries *AccessEntries) LoadByCampaignIDAndSourceID(
+	ctx context.Context,
+	conn pg.Conn,
+	scope Scoper,
+	campaignID gid.GID,
+	sourceID gid.GID,
+	cursor *page.Cursor[AccessEntryOrderField],
+	filter *AccessEntryFilter,
+) error {
+	q := `
+SELECT
+    id,
+    organization_id,
+    access_review_campaign_id,
+    access_source_id,
+    identity_id,
+    email,
+    full_name,
+    role,
+    job_title,
+    is_admin,
+    mfa_status,
+    auth_method,
+    account_type,
+    last_login,
+    account_created_at,
+    external_id,
+    account_key,
+    incremental_tag,
+    flags,
+    flag_reasons,
+    decision,
+    decision_note,
+    decided_by,
+    decided_at,
+    created_at,
+    updated_at
+FROM
+    access_entries
+WHERE
+    %s
+    AND access_review_campaign_id = @campaign_id
+    AND access_source_id = @source_id
+    AND %s
+    AND %s
+`
+	q = fmt.Sprintf(q, scope.SQLFragment(), filter.SQLFragment(), cursor.SQLFragment())
+
+	args := pgx.StrictNamedArgs{"campaign_id": campaignID, "source_id": sourceID}
+	maps.Copy(args, scope.SQLArguments())
+	maps.Copy(args, filter.SQLArguments())
+	maps.Copy(args, cursor.SQLArguments())
+
+	rows, err := conn.Query(ctx, q, args)
+	if err != nil {
+		return fmt.Errorf("cannot query access_entries: %w", err)
+	}
+
+	result, err := pgx.CollectRows(rows, pgx.RowToAddrOfStructByName[AccessEntry])
+	if err != nil {
+		return fmt.Errorf("cannot collect access_entries: %w", err)
+	}
+
+	*entries = result
+
+	return nil
+}
+
+func (entries *AccessEntries) CountByCampaignID(
+	ctx context.Context,
+	conn pg.Conn,
+	scope Scoper,
+	campaignID gid.GID,
+	filter *AccessEntryFilter,
+) (int, error) {
+	q := `
+SELECT COUNT(id)
+FROM access_entries
+WHERE
+    %s
+    AND access_review_campaign_id = @campaign_id
+    AND %s;
+`
+	q = fmt.Sprintf(q, scope.SQLFragment(), filter.SQLFragment())
+
+	args := pgx.StrictNamedArgs{"campaign_id": campaignID}
+	maps.Copy(args, scope.SQLArguments())
+	maps.Copy(args, filter.SQLArguments())
+
+	var count int
+	if err := conn.QueryRow(ctx, q, args).Scan(&count); err != nil {
+		return 0, fmt.Errorf("cannot count access_entries: %w", err)
+	}
+
+	return count, nil
+}
+
+func (entries *AccessEntries) CountByCampaignIDAndSourceID(
+	ctx context.Context,
+	conn pg.Conn,
+	scope Scoper,
+	campaignID gid.GID,
+	sourceID gid.GID,
+	filter *AccessEntryFilter,
+) (int, error) {
+	q := `
+SELECT COUNT(id)
+FROM access_entries
+WHERE
+    %s
+    AND access_review_campaign_id = @campaign_id
+    AND access_source_id = @source_id
+    AND %s;
+`
+	q = fmt.Sprintf(q, scope.SQLFragment(), filter.SQLFragment())
+
+	args := pgx.StrictNamedArgs{"campaign_id": campaignID, "source_id": sourceID}
+	maps.Copy(args, scope.SQLArguments())
+	maps.Copy(args, filter.SQLArguments())
+
+	var count int
+	if err := conn.QueryRow(ctx, q, args).Scan(&count); err != nil {
+		return 0, fmt.Errorf("cannot count access_entries: %w", err)
+	}
+
+	return count, nil
+}
+
+func (entries *AccessEntries) CountPendingByCampaignID(
+	ctx context.Context,
+	conn pg.Conn,
+	scope Scoper,
+	campaignID gid.GID,
+) (int, error) {
+	q := `
+SELECT COUNT(id)
+FROM access_entries
+WHERE
+    %s
+    AND access_review_campaign_id = @campaign_id
+    AND decision = 'PENDING';
+`
+	q = fmt.Sprintf(q, scope.SQLFragment())
+
+	args := pgx.StrictNamedArgs{"campaign_id": campaignID}
+	maps.Copy(args, scope.SQLArguments())
+
+	var count int
+	if err := conn.QueryRow(ctx, q, args).Scan(&count); err != nil {
+		return 0, fmt.Errorf("cannot count pending access_entries: %w", err)
+	}
+
+	return count, nil
+}
+
+func (e *AccessEntry) LoadOrganizationID(
+	ctx context.Context,
+	conn pg.Conn,
+	entryID gid.GID,
+) (gid.GID, error) {
+	q := `SELECT organization_id FROM access_entries WHERE id = $1 LIMIT 1;`
+
+	var organizationID gid.GID
+	if err := conn.QueryRow(ctx, q, entryID).Scan(&organizationID); err != nil {
+		if errors.Is(err, pgx.ErrNoRows) {
+			return gid.GID{}, ErrResourceNotFound
+		}
+		return gid.GID{}, fmt.Errorf("cannot load organization id for access entry: %w", err)
+	}
+
+	return organizationID, nil
+}
+
+func (e *AccessEntry) UpdateFlags(
+	ctx context.Context,
+	conn pg.Conn,
+	scope Scoper,
+) error {
+	q := `
+UPDATE access_entries
+SET
+    flags = @flags,
+    flag_reasons = @flag_reasons,
+    updated_at = @updated_at
+WHERE
+    %s
+    AND id = @id
+`
+	q = fmt.Sprintf(q, scope.SQLFragment())
+
+	args := pgx.StrictNamedArgs{
+		"id":           e.ID,
+		"flags":        e.Flags,
+		"flag_reasons": e.FlagReasons,
+		"updated_at":   e.UpdatedAt,
+	}
+	maps.Copy(args, scope.SQLArguments())
+
+	result, err := conn.Exec(ctx, q, args)
+	if err != nil {
+		return fmt.Errorf("cannot update access entry flags: %w", err)
+	}
+
+	if result.RowsAffected() == 0 {
+		return ErrResourceNotFound
+	}
+
+	return nil
+}
+
+func (e *AccessEntry) Upsert(
+	ctx context.Context,
+	conn pg.Conn,
+	scope Scoper,
+) error {
+	q := `
+INSERT INTO access_entries (
+    id,
+    tenant_id,
+    organization_id,
+    access_review_campaign_id,
+    access_source_id,
+    identity_id,
+    email,
+    full_name,
+    role,
+    job_title,
+    is_admin,
+    mfa_status,
+    auth_method,
+    account_type,
+    last_login,
+    account_created_at,
+    external_id,
+    account_key,
+    incremental_tag,
+    flags,
+    flag_reasons,
+    decision,
+    decision_note,
+    decided_by,
+    decided_at,
+    created_at,
+    updated_at
+) VALUES (
+    @id,
+    @tenant_id,
+    @organization_id,
+    @access_review_campaign_id,
+    @access_source_id,
+    @identity_id,
+    @email,
+    @full_name,
+    @role,
+    @job_title,
+    @is_admin,
+    @mfa_status,
+    @auth_method,
+    @account_type,
+    @last_login,
+    @account_created_at,
+    @external_id,
+    @account_key,
+    @incremental_tag,
+    @flags,
+    @flag_reasons,
+    @decision,
+    @decision_note,
+    @decided_by,
+    @decided_at,
+    @created_at,
+    @updated_at
+)
+ON CONFLICT (access_review_campaign_id, access_source_id, account_key) DO UPDATE SET
+    email = EXCLUDED.email,
+    full_name = EXCLUDED.full_name,
+    role = EXCLUDED.role,
+    job_title = EXCLUDED.job_title,
+    is_admin = EXCLUDED.is_admin,
+    mfa_status = EXCLUDED.mfa_status,
+    auth_method = EXCLUDED.auth_method,
+    account_type = EXCLUDED.account_type,
+    last_login = EXCLUDED.last_login,
+    account_created_at = EXCLUDED.account_created_at,
+    external_id = EXCLUDED.external_id,
+    incremental_tag = EXCLUDED.incremental_tag,
+    updated_at = EXCLUDED.updated_at
+`
+	args := pgx.StrictNamedArgs{
+		"id":                        e.ID,
+		"tenant_id":                 scope.GetTenantID(),
+		"organization_id":           e.OrganizationID,
+		"access_review_campaign_id": e.AccessReviewCampaignID,
+		"access_source_id":          e.AccessSourceID,
+		"identity_id":               e.IdentityID,
+		"email":                     e.Email,
+		"full_name":                 e.FullName,
+		"role":                      e.Role,
+		"job_title":                 e.JobTitle,
+		"is_admin":                  e.IsAdmin,
+		"mfa_status":                e.MFAStatus,
+		"auth_method":               e.AuthMethod,
+		"account_type":              e.AccountType,
+		"last_login":                e.LastLogin,
+		"account_created_at":        e.AccountCreatedAt,
+		"external_id":               e.ExternalID,
+		"account_key":               e.AccountKey,
+		"incremental_tag":           e.IncrementalTag,
+		"flags":                     e.Flags,
+		"flag_reasons":              e.FlagReasons,
+		"decision":                  e.Decision,
+		"decision_note":             e.DecisionNote,
+		"decided_by":                e.DecidedBy,
+		"decided_at":                e.DecidedAt,
+		"created_at":                e.CreatedAt,
+		"updated_at":                e.UpdatedAt,
+	}
+
+	_, err := conn.Exec(ctx, q, args)
+	if err != nil {
+		return fmt.Errorf("cannot upsert access entry: %w", err)
+	}
+
+	return nil
+}
+
+// BaselineAccountEntry holds minimal data from a previous campaign's entries
+// for incremental diffing.
+type BaselineAccountEntry struct {
+	AccountKey string
+	Email      string
+	FullName   string
+}
+
+func (entries *AccessEntries) LoadBaselineBySourceID(
+	ctx context.Context,
+	conn pg.Conn,
+	scope Scoper,
+	campaignID gid.GID,
+	sourceID gid.GID,
+) ([]BaselineAccountEntry, error) {
+	q := fmt.Sprintf(`
+SELECT account_key, email, full_name
+FROM access_entries
+WHERE %s
+  AND access_review_campaign_id = @campaign_id
+  AND access_source_id = @source_id
+`, scope.SQLFragment())
+
+	args := pgx.StrictNamedArgs{
+		"campaign_id": campaignID,
+		"source_id":   sourceID,
+	}
+	maps.Copy(args, scope.SQLArguments())
+
+	rows, err := conn.Query(ctx, q, args)
+	if err != nil {
+		return nil, fmt.Errorf("cannot load baseline entries: %w", err)
+	}
+	defer rows.Close()
+
+	var result []BaselineAccountEntry
+	for rows.Next() {
+		var entry BaselineAccountEntry
+		if err := rows.Scan(&entry.AccountKey, &entry.Email, &entry.FullName); err != nil {
+			return nil, fmt.Errorf("cannot scan baseline entry: %w", err)
+		}
+		result = append(result, entry)
+	}
+	if err := rows.Err(); err != nil {
+		return nil, fmt.Errorf("cannot iterate baseline entries: %w", err)
+	}
+
+	return result, nil
+}
+
+// LoadMembershipAccountsByOrganizationID loads IAM membership accounts for the
+// given organization.
+type MembershipAccount struct {
+	ID        gid.GID
+	Email     string
+	FullName  string
+	State     string
+	Role      string
+	CreatedAt time.Time
+}
+
+func LoadMembershipAccountsByOrganizationID(
+	ctx context.Context,
+	conn pg.Conn,
+	scope Scoper,
+	organizationID gid.GID,
+) ([]MembershipAccount, error) {
+	q := `
+SELECT
+    m.id,
+    i.email_address,
+    i.full_name,
+    m.state,
+    m.role,
+    m.created_at
+FROM
+    iam_memberships m
+JOIN
+    identities i ON i.id = m.identity_id
+WHERE
+    m.%s
+    AND m.organization_id = @organization_id
+    AND m.state = 'ACTIVE'
+ORDER BY
+    i.email_address ASC
+`
+	q = fmt.Sprintf(q, scope.SQLFragment())
+
+	args := pgx.StrictNamedArgs{
+		"organization_id": organizationID,
+	}
+	maps.Copy(args, scope.SQLArguments())
+
+	rows, err := conn.Query(ctx, q, args)
+	if err != nil {
+		return nil, fmt.Errorf("cannot query membership accounts: %w", err)
+	}
+	defer rows.Close()
+
+	var result []MembershipAccount
+	for rows.Next() {
+		var account MembershipAccount
+		if err := rows.Scan(&account.ID, &account.Email, &account.FullName, &account.State, &account.Role, &account.CreatedAt); err != nil {
+			return nil, fmt.Errorf("cannot scan membership account: %w", err)
+		}
+		result = append(result, account)
+	}
+	if err := rows.Err(); err != nil {
+		return nil, fmt.Errorf("cannot iterate membership accounts: %w", err)
+	}
+
+	return result, nil
+}

pkg/coredata/access_entry_account_type.go

@@ -0,0 +1,64 @@
+// Copyright (c) 2026 Probo Inc <hello@getprobo.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.
+
+package coredata
+
+import (
+	"database/sql/driver"
+	"fmt"
+)
+
+type AccessEntryAccountType string
+
+const (
+	AccessEntryAccountTypeUser           AccessEntryAccountType = "USER"
+	AccessEntryAccountTypeServiceAccount AccessEntryAccountType = "SERVICE_ACCOUNT"
+)
+
+func AccessEntryAccountTypes() []AccessEntryAccountType {
+	return []AccessEntryAccountType{
+		AccessEntryAccountTypeUser,
+		AccessEntryAccountTypeServiceAccount,
+	}
+}
+
+func (a AccessEntryAccountType) String() string {
+	return string(a)
+}
+
+func (a *AccessEntryAccountType) Scan(value any) error {
+	var str string
+	switch v := value.(type) {
+	case string:
+		str = v
+	case []byte:
+		str = string(v)
+	default:
+		return fmt.Errorf("cannot scan AccessEntryAccountType: unsupported type %T", value)
+	}
+
+	switch str {
+	case "USER":
+		*a = AccessEntryAccountTypeUser
+	case "SERVICE_ACCOUNT":
+		*a = AccessEntryAccountTypeServiceAccount
+	default:
+		return fmt.Errorf("cannot parse AccessEntryAccountType: invalid value %q", str)
+	}
+	return nil
+}
+
+func (a AccessEntryAccountType) Value() (driver.Value, error) {
+	return a.String(), nil
+}

pkg/coredata/access_entry_decision.go

@@ -0,0 +1,66 @@
+// Copyright (c) 2025-2026 Probo Inc <hello@getprobo.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.
+
+package coredata
+
+import (
+	"database/sql/driver"
+	"fmt"
+)
+
+type AccessEntryDecision string
+
+const (
+	AccessEntryDecisionPending  AccessEntryDecision = "PENDING"
+	AccessEntryDecisionApproved AccessEntryDecision = "APPROVED"
+	AccessEntryDecisionRevoke   AccessEntryDecision = "REVOKE"
+	AccessEntryDecisionDefer    AccessEntryDecision = "DEFER"
+	AccessEntryDecisionEscalate AccessEntryDecision = "ESCALATE"
+)
+
+func (d AccessEntryDecision) String() string {
+	return string(d)
+}
+
+func (d *AccessEntryDecision) Scan(value any) error {
+	var str string
+	switch v := value.(type) {
+	case string:
+		str = v
+	case []byte:
+		str = string(v)
+	default:
+		return fmt.Errorf("cannot scan AccessEntryDecision: unsupported type %T", value)
+	}
+
+	switch str {
+	case "PENDING":
+		*d = AccessEntryDecisionPending
+	case "APPROVED":
+		*d = AccessEntryDecisionApproved
+	case "REVOKE":
+		*d = AccessEntryDecisionRevoke
+	case "DEFER":
+		*d = AccessEntryDecisionDefer
+	case "ESCALATE":
+		*d = AccessEntryDecisionEscalate
+	default:
+		return fmt.Errorf("cannot parse AccessEntryDecision: invalid value %q", str)
+	}
+	return nil
+}
+
+func (d AccessEntryDecision) Value() (driver.Value, error) {
+	return d.String(), nil
+}

pkg/coredata/access_entry_decision_history.go

@@ -0,0 +1,150 @@
+// Copyright (c) 2026 Probo Inc <hello@getprobo.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.
+
+package coredata
+
+import (
+	"context"
+	"errors"
+	"fmt"
+	"maps"
+	"time"
+
+	"github.com/jackc/pgx/v5"
+	"go.gearno.de/kit/pg"
+	"go.probo.inc/probo/pkg/gid"
+)
+
+type (
+	AccessEntryDecisionHistory struct {
+		ID             gid.GID             `db:"id"`
+		OrganizationID gid.GID             `db:"organization_id"`
+		AccessEntry    gid.GID             `db:"access_entry_id"`
+		Decision       AccessEntryDecision `db:"decision"`
+		DecisionNote   *string             `db:"decision_note"`
+		DecidedBy      *gid.GID            `db:"decided_by"`
+		DecidedAt      time.Time           `db:"decided_at"`
+		CreatedAt      time.Time           `db:"created_at"`
+	}
+
+	AccessEntryDecisionHistories []*AccessEntryDecisionHistory
+)
+
+func (h *AccessEntryDecisionHistory) Insert(
+	ctx context.Context,
+	conn pg.Conn,
+	scope Scoper,
+) error {
+	q := `
+INSERT INTO access_entry_decision_history (
+    id,
+    tenant_id,
+    organization_id,
+    access_entry_id,
+    decision,
+    decision_note,
+    decided_by,
+    decided_at,
+    created_at
+) VALUES (
+    @id,
+    @tenant_id,
+    @organization_id,
+    @access_entry_id,
+    @decision,
+    @decision_note,
+    @decided_by,
+    @decided_at,
+    @created_at
+);
+`
+	args := pgx.StrictNamedArgs{
+		"id":              h.ID,
+		"tenant_id":       scope.GetTenantID(),
+		"organization_id": h.OrganizationID,
+		"access_entry_id": h.AccessEntry,
+		"decision":        h.Decision,
+		"decision_note":   h.DecisionNote,
+		"decided_by":      h.DecidedBy,
+		"decided_at":      h.DecidedAt,
+		"created_at":      h.CreatedAt,
+	}
+
+	_, err := conn.Exec(ctx, q, args)
+	if err != nil {
+		return fmt.Errorf("cannot insert access entry decision history: %w", err)
+	}
+
+	return nil
+}
+
+func (h *AccessEntryDecisionHistory) AuthorizationAttributes(
+	ctx context.Context,
+	conn pg.Conn,
+) (map[string]string, error) {
+	q := `SELECT organization_id FROM access_entry_decision_history WHERE id = $1 LIMIT 1;`
+
+	var organizationID gid.GID
+	if err := conn.QueryRow(ctx, q, h.ID).Scan(&organizationID); err != nil {
+		if errors.Is(err, pgx.ErrNoRows) {
+			return nil, ErrResourceNotFound
+		}
+		return nil, fmt.Errorf("cannot load authorization attributes: %w", err)
+	}
+
+	return map[string]string{"organization_id": organizationID.String()}, nil
+}
+
+func (hs *AccessEntryDecisionHistories) LoadByEntryID(
+	ctx context.Context,
+	conn pg.Conn,
+	scope Scoper,
+	entryID gid.GID,
+) error {
+	q := `
+SELECT
+    id,
+    organization_id,
+    access_entry_id,
+    decision,
+    decision_note,
+    decided_by,
+    decided_at,
+    created_at
+FROM
+    access_entry_decision_history
+WHERE
+    %s
+    AND access_entry_id = @access_entry_id
+ORDER BY decided_at ASC;
+`
+	q = fmt.Sprintf(q, scope.SQLFragment())
+
+	args := pgx.StrictNamedArgs{"access_entry_id": entryID}
+	maps.Copy(args, scope.SQLArguments())
+
+	rows, err := conn.Query(ctx, q, args)
+	if err != nil {
+		return fmt.Errorf("cannot query access entry decision history: %w", err)
+	}
+
+	result, err := pgx.CollectRows(rows, pgx.RowToAddrOfStructByName[AccessEntryDecisionHistory])
+	if err != nil {
+		return fmt.Errorf("cannot collect access entry decision history: %w", err)
+	}
+
+	*hs = result
+
+	return nil
+}

pkg/coredata/access_entry_filter.go

@@ -0,0 +1,109 @@
+// Copyright (c) 2026 Probo Inc <hello@getprobo.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.
+
+package coredata
+
+import (
+	"github.com/jackc/pgx/v5"
+)
+
+type AccessEntryFilter struct {
+	Decision       *AccessEntryDecision
+	Flag           *AccessEntryFlag
+	IncrementalTag *AccessEntryIncrementalTag
+	IsAdmin        *bool
+	AuthMethod     *AccessEntryAuthMethod
+	AccountType    *AccessEntryAccountType
+}
+
+func (f *AccessEntryFilter) SQLFragment() string {
+	if f == nil {
+		return "TRUE"
+	}
+
+	return `
+(
+	CASE
+		WHEN @filter_decision::text IS NOT NULL THEN
+			decision = @filter_decision::text
+		ELSE TRUE
+	END
+	AND
+	CASE
+		WHEN @filter_flag::text IS NOT NULL THEN
+			@filter_flag::text = ANY(flags)
+		ELSE TRUE
+	END
+	AND
+	CASE
+		WHEN @filter_incremental_tag::text IS NOT NULL THEN
+			incremental_tag = @filter_incremental_tag::text
+		ELSE TRUE
+	END
+	AND
+	CASE
+		WHEN @filter_is_admin::boolean IS NOT NULL THEN
+			is_admin = @filter_is_admin::boolean
+		ELSE TRUE
+	END
+	AND
+	CASE
+		WHEN @filter_auth_method::text IS NOT NULL THEN
+			auth_method = @filter_auth_method::text
+		ELSE TRUE
+	END
+	AND
+	CASE
+		WHEN @filter_account_type::text IS NOT NULL THEN
+			account_type = @filter_account_type::text
+		ELSE TRUE
+	END
+)`
+}
+
+func (f *AccessEntryFilter) SQLArguments() pgx.StrictNamedArgs {
+	if f == nil {
+		return pgx.StrictNamedArgs{}
+	}
+
+	args := pgx.StrictNamedArgs{
+		"filter_decision":        nil,
+		"filter_flag":            nil,
+		"filter_incremental_tag": nil,
+		"filter_is_admin":        nil,
+		"filter_auth_method":     nil,
+		"filter_account_type":    nil,
+	}
+
+	if f.Decision != nil {
+		args["filter_decision"] = string(*f.Decision)
+	}
+	if f.Flag != nil {
+		args["filter_flag"] = string(*f.Flag)
+	}
+	if f.IncrementalTag != nil {
+		args["filter_incremental_tag"] = string(*f.IncrementalTag)
+	}
+	if f.IsAdmin != nil {
+		args["filter_is_admin"] = *f.IsAdmin
+	}
+	if f.AuthMethod != nil {
+		args["filter_auth_method"] = string(*f.AuthMethod)
+	}
+	if f.AccountType != nil {
+		args["filter_account_type"] = string(*f.AccountType)
+	}
+
+	return args
+}

pkg/coredata/access_entry_flag.go

@@ -0,0 +1,96 @@
+// Copyright (c) 2026 Probo Inc <hello@getprobo.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.
+
+package coredata
+
+import (
+	"database/sql/driver"
+	"fmt"
+)
+
+type AccessEntryFlag string
+
+const (
+	AccessEntryFlagNone                    AccessEntryFlag = "NONE"
+	AccessEntryFlagOrphaned                AccessEntryFlag = "ORPHANED"
+	AccessEntryFlagInactive                AccessEntryFlag = "INACTIVE"
+	AccessEntryFlagExcessive               AccessEntryFlag = "EXCESSIVE"
+	AccessEntryFlagRoleMismatch            AccessEntryFlag = "ROLE_MISMATCH"
+	AccessEntryFlagNew                     AccessEntryFlag = "NEW"
+	AccessEntryFlagDormant                 AccessEntryFlag = "DORMANT"
+	AccessEntryFlagTerminatedUser          AccessEntryFlag = "TERMINATED_USER"
+	AccessEntryFlagContractorExpired       AccessEntryFlag = "CONTRACTOR_EXPIRED"
+	AccessEntryFlagSoDConflict             AccessEntryFlag = "SOD_CONFLICT"
+	AccessEntryFlagPrivilegedAccess        AccessEntryFlag = "PRIVILEGED_ACCESS"
+	AccessEntryFlagRoleCreep               AccessEntryFlag = "ROLE_CREEP"
+	AccessEntryFlagNoBusinessJustification AccessEntryFlag = "NO_BUSINESS_JUSTIFICATION"
+	AccessEntryFlagOutOfDepartment         AccessEntryFlag = "OUT_OF_DEPARTMENT"
+	AccessEntryFlagSharedAccount           AccessEntryFlag = "SHARED_ACCOUNT"
+)
+
+func (f AccessEntryFlag) String() string {
+	return string(f)
+}
+
+func (f *AccessEntryFlag) Scan(value any) error {
+	var str string
+	switch v := value.(type) {
+	case string:
+		str = v
+	case []byte:
+		str = string(v)
+	default:
+		return fmt.Errorf("cannot scan AccessEntryFlag: unsupported type %T", value)
+	}
+
+	switch str {
+	case "NONE":
+		*f = AccessEntryFlagNone
+	case "ORPHANED":
+		*f = AccessEntryFlagOrphaned
+	case "INACTIVE":
+		*f = AccessEntryFlagInactive
+	case "EXCESSIVE":
+		*f = AccessEntryFlagExcessive
+	case "ROLE_MISMATCH":
+		*f = AccessEntryFlagRoleMismatch
+	case "NEW":
+		*f = AccessEntryFlagNew
+	case "DORMANT":
+		*f = AccessEntryFlagDormant
+	case "TERMINATED_USER":
+		*f = AccessEntryFlagTerminatedUser
+	case "CONTRACTOR_EXPIRED":
+		*f = AccessEntryFlagContractorExpired
+	case "SOD_CONFLICT":
+		*f = AccessEntryFlagSoDConflict
+	case "PRIVILEGED_ACCESS":
+		*f = AccessEntryFlagPrivilegedAccess
+	case "ROLE_CREEP":
+		*f = AccessEntryFlagRoleCreep
+	case "NO_BUSINESS_JUSTIFICATION":
+		*f = AccessEntryFlagNoBusinessJustification
+	case "OUT_OF_DEPARTMENT":
+		*f = AccessEntryFlagOutOfDepartment
+	case "SHARED_ACCOUNT":
+		*f = AccessEntryFlagSharedAccount
+	default:
+		return fmt.Errorf("cannot parse AccessEntryFlag: invalid value %q", str)
+	}
+	return nil
+}
+
+func (f AccessEntryFlag) Value() (driver.Value, error) {
+	return f.String(), nil
+}

pkg/coredata/access_entry_incremental_tag.go

@@ -0,0 +1,61 @@
+// Copyright (c) 2026 Probo Inc <hello@getprobo.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.
+
+package coredata
+
+import (
+	"database/sql/driver"
+	"fmt"
+)
+
+type AccessEntryIncrementalTag string
+
+const (
+	AccessEntryIncrementalTagNew       AccessEntryIncrementalTag = "NEW"
+	AccessEntryIncrementalTagRemoved   AccessEntryIncrementalTag = "REMOVED"
+	AccessEntryIncrementalTagUnchanged AccessEntryIncrementalTag = "UNCHANGED"
+)
+
+func (t AccessEntryIncrementalTag) String() string {
+	return string(t)
+}
+
+func (t *AccessEntryIncrementalTag) Scan(value any) error {
+	var str string
+	switch v := value.(type) {
+	case string:
+		str = v
+	case []byte:
+		str = string(v)
+	default:
+		return fmt.Errorf("cannot scan AccessEntryIncrementalTag: unsupported type %T", value)
+	}
+
+	switch str {
+	case "NEW":
+		*t = AccessEntryIncrementalTagNew
+	case "REMOVED":
+		*t = AccessEntryIncrementalTagRemoved
+	case "UNCHANGED":
+		*t = AccessEntryIncrementalTagUnchanged
+	default:
+		return fmt.Errorf("cannot parse AccessEntryIncrementalTag: invalid value %q", str)
+	}
+
+	return nil
+}
+
+func (t AccessEntryIncrementalTag) Value() (driver.Value, error) {
+	return t.String(), nil
+}

pkg/coredata/access_entry_order_field.go

@@ -0,0 +1,57 @@
+// Copyright (c) 2026 Probo Inc <hello@getprobo.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.
+
+package coredata
+
+import "fmt"
+
+type (
+	AccessEntryOrderField string
+)
+
+const (
+	AccessEntryOrderFieldCreatedAt AccessEntryOrderField = "CREATED_AT"
+)
+
+func (p AccessEntryOrderField) Column() string {
+	switch p {
+	case AccessEntryOrderFieldCreatedAt:
+		return "created_at"
+	}
+	panic(fmt.Sprintf("unsupported order by: %s", p))
+}
+
+func (p AccessEntryOrderField) IsValid() bool {
+	switch p {
+	case AccessEntryOrderFieldCreatedAt:
+		return true
+	}
+	return false
+}
+
+func (p AccessEntryOrderField) String() string {
+	return string(p)
+}
+
+func (p AccessEntryOrderField) MarshalText() ([]byte, error) {
+	return []byte(p.String()), nil
+}
+
+func (p *AccessEntryOrderField) UnmarshalText(text []byte) error {
+	*p = AccessEntryOrderField(text)
+	if !p.IsValid() {
+		return fmt.Errorf("%s is not a valid AccessEntryOrderField", string(text))
+	}
+	return nil
+}

pkg/coredata/access_entry_statistics.go

@@ -0,0 +1,243 @@
+// Copyright (c) 2026 Probo Inc <hello@getprobo.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.
+
+package coredata
+
+import (
+	"context"
+	"fmt"
+	"maps"
+
+	"github.com/jackc/pgx/v5"
+	"go.gearno.de/kit/pg"
+	"go.probo.inc/probo/pkg/gid"
+)
+
+type AccessEntryStatistics struct {
+	TotalCount           int
+	DecisionCounts       map[AccessEntryDecision]int
+	FlagCounts           map[AccessEntryFlag]int
+	IncrementalTagCounts map[AccessEntryIncrementalTag]int
+}
+
+func (s *AccessEntryStatistics) LoadByCampaignID(
+	ctx context.Context,
+	conn pg.Conn,
+	scope Scoper,
+	campaignID gid.GID,
+) error {
+	args := pgx.StrictNamedArgs{"campaign_id": campaignID}
+	maps.Copy(args, scope.SQLArguments())
+
+	s.DecisionCounts = make(map[AccessEntryDecision]int)
+	s.FlagCounts = make(map[AccessEntryFlag]int)
+	s.IncrementalTagCounts = make(map[AccessEntryIncrementalTag]int)
+	s.TotalCount = 0
+
+	q := `
+SELECT decision, COUNT(*) as count
+FROM access_entries
+WHERE
+    %s
+    AND access_review_campaign_id = @campaign_id
+GROUP BY decision;
+`
+	q = fmt.Sprintf(q, scope.SQLFragment())
+
+	rows, err := conn.Query(ctx, q, args)
+	if err != nil {
+		return fmt.Errorf("cannot query access entry decision counts: %w", err)
+	}
+	defer rows.Close()
+
+	for rows.Next() {
+		var decision AccessEntryDecision
+		var count int
+		if err := rows.Scan(&decision, &count); err != nil {
+			return fmt.Errorf("cannot scan decision count: %w", err)
+		}
+		s.DecisionCounts[decision] = count
+		s.TotalCount += count
+	}
+	if err := rows.Err(); err != nil {
+		return fmt.Errorf("cannot iterate decision counts: %w", err)
+	}
+
+	q = `
+SELECT f, COUNT(*) as count
+FROM access_entries, unnest(flags) AS f
+WHERE
+    %s
+    AND access_review_campaign_id = @campaign_id
+GROUP BY f;
+`
+	q = fmt.Sprintf(q, scope.SQLFragment())
+
+	rows, err = conn.Query(ctx, q, args)
+	if err != nil {
+		return fmt.Errorf("cannot query access entry flag counts: %w", err)
+	}
+	defer rows.Close()
+
+	for rows.Next() {
+		var flag AccessEntryFlag
+		var count int
+		if err := rows.Scan(&flag, &count); err != nil {
+			return fmt.Errorf("cannot scan flag count: %w", err)
+		}
+		s.FlagCounts[flag] = count
+	}
+	if err := rows.Err(); err != nil {
+		return fmt.Errorf("cannot iterate flag counts: %w", err)
+	}
+
+	q = `
+SELECT incremental_tag, COUNT(*) as count
+FROM access_entries
+WHERE
+    %s
+    AND access_review_campaign_id = @campaign_id
+GROUP BY incremental_tag;
+`
+	q = fmt.Sprintf(q, scope.SQLFragment())
+
+	rows, err = conn.Query(ctx, q, args)
+	if err != nil {
+		return fmt.Errorf("cannot query access entry incremental tag counts: %w", err)
+	}
+	defer rows.Close()
+
+	for rows.Next() {
+		var tag AccessEntryIncrementalTag
+		var count int
+		if err := rows.Scan(&tag, &count); err != nil {
+			return fmt.Errorf("cannot scan incremental tag count: %w", err)
+		}
+		s.IncrementalTagCounts[tag] = count
+	}
+	if err := rows.Err(); err != nil {
+		return fmt.Errorf("cannot iterate incremental tag counts: %w", err)
+	}
+
+	return nil
+}
+
+func (s *AccessEntryStatistics) LoadByCampaignIDAndSourceID(
+	ctx context.Context,
+	conn pg.Conn,
+	scope Scoper,
+	campaignID gid.GID,
+	sourceID gid.GID,
+) error {
+	args := pgx.StrictNamedArgs{
+		"campaign_id": campaignID,
+		"source_id":   sourceID,
+	}
+	maps.Copy(args, scope.SQLArguments())
+
+	s.DecisionCounts = make(map[AccessEntryDecision]int)
+	s.FlagCounts = make(map[AccessEntryFlag]int)
+	s.IncrementalTagCounts = make(map[AccessEntryIncrementalTag]int)
+	s.TotalCount = 0
+
+	q := `
+SELECT decision, COUNT(*) as count
+FROM access_entries
+WHERE
+    %s
+    AND access_review_campaign_id = @campaign_id
+    AND access_source_id = @source_id
+GROUP BY decision;
+`
+	q = fmt.Sprintf(q, scope.SQLFragment())
+
+	rows, err := conn.Query(ctx, q, args)
+	if err != nil {
+		return fmt.Errorf("cannot query access entry decision counts: %w", err)
+	}
+	defer rows.Close()
+
+	for rows.Next() {
+		var decision AccessEntryDecision
+		var count int
+		if err := rows.Scan(&decision, &count); err != nil {
+			return fmt.Errorf("cannot scan decision count: %w", err)
+		}
+		s.DecisionCounts[decision] = count
+		s.TotalCount += count
+	}
+	if err := rows.Err(); err != nil {
+		return fmt.Errorf("cannot iterate decision counts: %w", err)
+	}
+
+	q = `
+SELECT f, COUNT(*) as count
+FROM access_entries, unnest(flags) AS f
+WHERE
+    %s
+    AND access_review_campaign_id = @campaign_id
+    AND access_source_id = @source_id
+GROUP BY f;
+`
+	q = fmt.Sprintf(q, scope.SQLFragment())
+
+	rows, err = conn.Query(ctx, q, args)
+	if err != nil {
+		return fmt.Errorf("cannot query access entry flag counts: %w", err)
+	}
+	defer rows.Close()
+
+	for rows.Next() {
+		var flag AccessEntryFlag
+		var count int
+		if err := rows.Scan(&flag, &count); err != nil {
+			return fmt.Errorf("cannot scan flag count: %w", err)
+		}
+		s.FlagCounts[flag] = count
+	}
+	if err := rows.Err(); err != nil {
+		return fmt.Errorf("cannot iterate flag counts: %w", err)
+	}
+
+	q = `
+SELECT incremental_tag, COUNT(*) as count
+FROM access_entries
+WHERE
+    %s
+    AND access_review_campaign_id = @campaign_id
+    AND access_source_id = @source_id
+GROUP BY incremental_tag;
+`
+	q = fmt.Sprintf(q, scope.SQLFragment())
+
+	rows, err = conn.Query(ctx, q, args)
+	if err != nil {
+		return fmt.Errorf("cannot query access entry incremental tag counts: %w", err)
+	}
+	defer rows.Close()
+
+	for rows.Next() {
+		var tag AccessEntryIncrementalTag
+		var count int
+		if err := rows.Scan(&tag, &count); err != nil {
+			return fmt.Errorf("cannot scan incremental tag count: %w", err)
+		}
+		s.IncrementalTagCounts[tag] = count
+	}
+	if err := rows.Err(); err != nil {
+		return fmt.Errorf("cannot iterate incremental tag counts: %w", err)
+	}
+
+	return nil
+}

pkg/coredata/access_review_campaign.go

@@ -0,0 +1,366 @@
+// Copyright (c) 2026 Probo Inc <hello@getprobo.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.
+
+package coredata
+
+import (
+	"context"
+	"errors"
+	"fmt"
+	"maps"
+	"time"
+
+	"github.com/jackc/pgx/v5"
+	"go.gearno.de/kit/pg"
+	"go.probo.inc/probo/pkg/gid"
+	"go.probo.inc/probo/pkg/page"
+)
+
+type (
+	AccessReviewCampaign struct {
+		ID                gid.GID                    `db:"id"`
+		OrganizationID    gid.GID                    `db:"organization_id"`
+		Name              string                     `db:"name"`
+		Description       string                     `db:"description"`
+		Status            AccessReviewCampaignStatus `db:"status"`
+		StartedAt         *time.Time                 `db:"started_at"`
+		CompletedAt       *time.Time                 `db:"completed_at"`
+		FrameworkControls []string                   `db:"framework_controls"`
+		CreatedAt         time.Time                  `db:"created_at"`
+		UpdatedAt         time.Time                  `db:"updated_at"`
+	}
+
+	AccessReviewCampaigns []*AccessReviewCampaign
+)
+
+func (c AccessReviewCampaign) CursorKey(orderBy AccessReviewCampaignOrderField) page.CursorKey {
+	switch orderBy {
+	case AccessReviewCampaignOrderFieldCreatedAt:
+		return page.NewCursorKey(c.ID, c.CreatedAt)
+	}
+
+	panic(fmt.Sprintf("unsupported order by: %s", orderBy))
+}
+
+func (c *AccessReviewCampaign) AuthorizationAttributes(ctx context.Context, conn pg.Conn) (map[string]string, error) {
+	q := `SELECT organization_id FROM access_review_campaigns WHERE id = $1 LIMIT 1;`
+
+	var organizationID gid.GID
+	if err := conn.QueryRow(ctx, q, c.ID).Scan(&organizationID); err != nil {
+		if errors.Is(err, pgx.ErrNoRows) {
+			return nil, ErrResourceNotFound
+		}
+		return nil, fmt.Errorf("cannot query access review campaign authorization attributes: %w", err)
+	}
+
+	return map[string]string{"organization_id": organizationID.String()}, nil
+}
+
+func (c *AccessReviewCampaign) LoadByID(
+	ctx context.Context,
+	conn pg.Conn,
+	scope Scoper,
+	id gid.GID,
+) error {
+	q := `
+SELECT
+    id,
+    organization_id,
+    name,
+    description,
+    status,
+    started_at,
+    completed_at,
+    framework_controls,
+    created_at,
+    updated_at
+FROM
+    access_review_campaigns
+WHERE
+    %s
+    AND id = @id
+LIMIT 1;
+`
+	q = fmt.Sprintf(q, scope.SQLFragment())
+
+	args := pgx.StrictNamedArgs{"id": id}
+	maps.Copy(args, scope.SQLArguments())
+
+	rows, err := conn.Query(ctx, q, args)
+	if err != nil {
+		return fmt.Errorf("cannot query access_review_campaigns: %w", err)
+	}
+
+	campaign, err := pgx.CollectExactlyOneRow(rows, pgx.RowToStructByName[AccessReviewCampaign])
+	if err != nil {
+		if errors.Is(err, pgx.ErrNoRows) {
+			return ErrResourceNotFound
+		}
+		return fmt.Errorf("cannot collect access review campaign: %w", err)
+	}
+
+	*c = campaign
+
+	return nil
+}
+
+func (c *AccessReviewCampaign) Insert(
+	ctx context.Context,
+	conn pg.Conn,
+	scope Scoper,
+) error {
+	q := `
+INSERT INTO
+    access_review_campaigns (
+        id,
+        tenant_id,
+        organization_id,
+        name,
+        description,
+        status,
+        started_at,
+        completed_at,
+        framework_controls,
+        created_at,
+        updated_at
+    )
+VALUES (
+    @id,
+    @tenant_id,
+    @organization_id,
+    @name,
+    @description,
+    @status,
+    @started_at,
+    @completed_at,
+    @framework_controls,
+    @created_at,
+    @updated_at
+);
+`
+
+	args := pgx.StrictNamedArgs{
+		"id":                 c.ID,
+		"tenant_id":          scope.GetTenantID(),
+		"organization_id":    c.OrganizationID,
+		"name":               c.Name,
+		"description":        c.Description,
+		"status":             c.Status,
+		"started_at":         c.StartedAt,
+		"completed_at":       c.CompletedAt,
+		"framework_controls": c.FrameworkControls,
+		"created_at":         c.CreatedAt,
+		"updated_at":         c.UpdatedAt,
+	}
+	_, err := conn.Exec(ctx, q, args)
+	if err != nil {
+		return fmt.Errorf("cannot insert access_review_campaign: %w", err)
+	}
+
+	return nil
+}
+
+func (c *AccessReviewCampaign) Update(
+	ctx context.Context,
+	conn pg.Conn,
+	scope Scoper,
+) error {
+	q := `
+UPDATE access_review_campaigns
+SET
+    name = @name,
+    description = @description,
+    status = @status,
+    started_at = @started_at,
+    completed_at = @completed_at,
+    framework_controls = @framework_controls,
+    updated_at = @updated_at
+WHERE
+    %s
+    AND id = @id
+`
+	q = fmt.Sprintf(q, scope.SQLFragment())
+
+	args := pgx.StrictNamedArgs{
+		"id":                 c.ID,
+		"name":               c.Name,
+		"description":        c.Description,
+		"status":             c.Status,
+		"started_at":         c.StartedAt,
+		"completed_at":       c.CompletedAt,
+		"framework_controls": c.FrameworkControls,
+		"updated_at":         c.UpdatedAt,
+	}
+	maps.Copy(args, scope.SQLArguments())
+
+	result, err := conn.Exec(ctx, q, args)
+	if err != nil {
+		return fmt.Errorf("cannot update access_review_campaign: %w", err)
+	}
+
+	if result.RowsAffected() == 0 {
+		return ErrResourceNotFound
+	}
+
+	return nil
+}
+
+func (c *AccessReviewCampaign) Delete(
+	ctx context.Context,
+	conn pg.Conn,
+	scope Scoper,
+) error {
+	q := `
+DELETE FROM access_review_campaigns
+WHERE %s AND id = @id
+`
+	q = fmt.Sprintf(q, scope.SQLFragment())
+
+	args := pgx.StrictNamedArgs{"id": c.ID}
+	maps.Copy(args, scope.SQLArguments())
+
+	result, err := conn.Exec(ctx, q, args)
+	if err != nil {
+		return fmt.Errorf("cannot delete access_review_campaign: %w", err)
+	}
+
+	if result.RowsAffected() == 0 {
+		return ErrResourceNotFound
+	}
+
+	return nil
+}
+
+func (campaigns *AccessReviewCampaigns) LoadByOrganizationID(
+	ctx context.Context,
+	conn pg.Conn,
+	scope Scoper,
+	organizationID gid.GID,
+	cursor *page.Cursor[AccessReviewCampaignOrderField],
+) error {
+	q := `
+SELECT
+    id,
+    organization_id,
+    name,
+    description,
+    status,
+    started_at,
+    completed_at,
+    framework_controls,
+    created_at,
+    updated_at
+FROM
+    access_review_campaigns
+WHERE
+    %s
+    AND organization_id = @organization_id
+    AND %s
+`
+	q = fmt.Sprintf(q, scope.SQLFragment(), cursor.SQLFragment())
+
+	args := pgx.StrictNamedArgs{"organization_id": organizationID}
+	maps.Copy(args, scope.SQLArguments())
+	maps.Copy(args, cursor.SQLArguments())
+
+	rows, err := conn.Query(ctx, q, args)
+	if err != nil {
+		return fmt.Errorf("cannot query access_review_campaigns: %w", err)
+	}
+
+	result, err := pgx.CollectRows(rows, pgx.RowToAddrOfStructByName[AccessReviewCampaign])
+	if err != nil {
+		return fmt.Errorf("cannot collect access_review_campaigns: %w", err)
+	}
+
+	*campaigns = result
+
+	return nil
+}
+
+func (campaigns *AccessReviewCampaigns) CountByOrganizationID(
+	ctx context.Context,
+	conn pg.Conn,
+	scope Scoper,
+	organizationID gid.GID,
+) (int, error) {
+	q := `
+SELECT COUNT(id)
+FROM access_review_campaigns
+WHERE
+    %s
+    AND organization_id = @organization_id;
+`
+	q = fmt.Sprintf(q, scope.SQLFragment())
+
+	args := pgx.StrictNamedArgs{"organization_id": organizationID}
+	maps.Copy(args, scope.SQLArguments())
+
+	var count int
+	if err := conn.QueryRow(ctx, q, args).Scan(&count); err != nil {
+		return 0, fmt.Errorf("cannot count access_review_campaigns: %w", err)
+	}
+
+	return count, nil
+}
+
+func (c *AccessReviewCampaign) LoadLastCompletedByOrganizationID(
+	ctx context.Context,
+	conn pg.Conn,
+	scope Scoper,
+	organizationID gid.GID,
+) error {
+	q := `
+SELECT
+    id,
+    organization_id,
+    name,
+    description,
+    status,
+    started_at,
+    completed_at,
+    framework_controls,
+    created_at,
+    updated_at
+FROM
+    access_review_campaigns
+WHERE
+    %s
+    AND organization_id = @organization_id
+    AND status = 'COMPLETED'
+ORDER BY completed_at DESC
+LIMIT 1;
+`
+	q = fmt.Sprintf(q, scope.SQLFragment())
+
+	args := pgx.StrictNamedArgs{"organization_id": organizationID}
+	maps.Copy(args, scope.SQLArguments())
+
+	rows, err := conn.Query(ctx, q, args)
+	if err != nil {
+		return fmt.Errorf("cannot query access_review_campaigns: %w", err)
+	}
+
+	campaign, err := pgx.CollectExactlyOneRow(rows, pgx.RowToStructByName[AccessReviewCampaign])
+	if err != nil {
+		if errors.Is(err, pgx.ErrNoRows) {
+			return ErrResourceNotFound
+		}
+		return fmt.Errorf("cannot collect access review campaign: %w", err)
+	}
+
+	*c = campaign
+
+	return nil
+}

pkg/coredata/access_review_campaign_order_field.go

@@ -0,0 +1,57 @@
+// Copyright (c) 2026 Probo Inc <hello@getprobo.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.
+
+package coredata
+
+import "fmt"
+
+type (
+	AccessReviewCampaignOrderField string
+)
+
+const (
+	AccessReviewCampaignOrderFieldCreatedAt AccessReviewCampaignOrderField = "CREATED_AT"
+)
+
+func (p AccessReviewCampaignOrderField) Column() string {
+	switch p {
+	case AccessReviewCampaignOrderFieldCreatedAt:
+		return "created_at"
+	}
+	panic(fmt.Sprintf("unsupported order by: %s", p))
+}
+
+func (p AccessReviewCampaignOrderField) IsValid() bool {
+	switch p {
+	case AccessReviewCampaignOrderFieldCreatedAt:
+		return true
+	}
+	return false
+}
+
+func (p AccessReviewCampaignOrderField) String() string {
+	return string(p)
+}
+
+func (p AccessReviewCampaignOrderField) MarshalText() ([]byte, error) {
+	return []byte(p.String()), nil
+}
+
+func (p *AccessReviewCampaignOrderField) UnmarshalText(text []byte) error {
+	*p = AccessReviewCampaignOrderField(text)
+	if !p.IsValid() {
+		return fmt.Errorf("%s is not a valid AccessReviewCampaignOrderField", string(text))
+	}
+	return nil
+}

pkg/coredata/access_review_campaign_scope_system.go

@@ -0,0 +1,214 @@
+// Copyright (c) 2026 Probo Inc <hello@getprobo.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.
+
+package coredata
+
+import (
+	"context"
+	"errors"
+	"fmt"
+	"maps"
+	"time"
+
+	"github.com/jackc/pgx/v5"
+	"go.gearno.de/kit/pg"
+	"go.probo.inc/probo/pkg/gid"
+)
+
+type AccessReviewCampaignScopeSystem struct {
+	AccessReviewCampaignID gid.GID `db:"access_review_campaign_id"`
+	AccessSourceID         gid.GID `db:"access_source_id"`
+}
+
+func (ss AccessReviewCampaignScopeSystem) Insert(
+	ctx context.Context,
+	conn pg.Conn,
+	scope Scoper,
+) error {
+	q := `
+INSERT INTO access_review_campaign_scope_systems (access_review_campaign_id, access_source_id, tenant_id)
+VALUES (@access_review_campaign_id, @access_source_id, @tenant_id)
+`
+	args := pgx.StrictNamedArgs{
+		"access_review_campaign_id": ss.AccessReviewCampaignID,
+		"access_source_id":          ss.AccessSourceID,
+		"tenant_id":                 scope.GetTenantID(),
+	}
+
+	_, err := conn.Exec(ctx, q, args)
+	if err != nil {
+		return fmt.Errorf("cannot insert campaign scope system: %w", err)
+	}
+
+	return nil
+}
+
+func (ss AccessReviewCampaignScopeSystem) Upsert(
+	ctx context.Context,
+	conn pg.Conn,
+	scope Scoper,
+) error {
+	q := `
+INSERT INTO access_review_campaign_scope_systems (access_review_campaign_id, access_source_id, tenant_id)
+VALUES (@access_review_campaign_id, @access_source_id, @tenant_id)
+ON CONFLICT (access_review_campaign_id, access_source_id) DO NOTHING
+`
+	args := pgx.StrictNamedArgs{
+		"access_review_campaign_id": ss.AccessReviewCampaignID,
+		"access_source_id":          ss.AccessSourceID,
+		"tenant_id":                 scope.GetTenantID(),
+	}
+
+	_, err := conn.Exec(ctx, q, args)
+	if err != nil {
+		return fmt.Errorf("cannot upsert campaign scope system: %w", err)
+	}
+
+	return nil
+}
+
+func (ss AccessReviewCampaignScopeSystem) Delete(
+	ctx context.Context,
+	conn pg.Conn,
+	scope Scoper,
+) error {
+	q := `
+DELETE FROM access_review_campaign_scope_systems
+WHERE
+    %s
+    AND access_review_campaign_id = @access_review_campaign_id
+    AND access_source_id = @access_source_id
+`
+	q = fmt.Sprintf(q, scope.SQLFragment())
+
+	args := pgx.StrictNamedArgs{
+		"access_review_campaign_id": ss.AccessReviewCampaignID,
+		"access_source_id":          ss.AccessSourceID,
+	}
+	maps.Copy(args, scope.SQLArguments())
+
+	_, err := conn.Exec(ctx, q, args)
+	if err != nil {
+		return fmt.Errorf("cannot delete campaign scope system: %w", err)
+	}
+
+	return nil
+}
+
+func (c *AccessReviewCampaign) LockForUpdate(
+	ctx context.Context,
+	conn pg.Conn,
+	scope Scoper,
+) error {
+	q := `
+SELECT id
+FROM access_review_campaigns
+WHERE %s
+  AND id = @id
+FOR UPDATE
+`
+	q = fmt.Sprintf(q, scope.SQLFragment())
+	args := pgx.StrictNamedArgs{"id": c.ID}
+	maps.Copy(args, scope.SQLArguments())
+
+	var id gid.GID
+	if err := conn.QueryRow(ctx, q, args).Scan(&id); err != nil {
+		if errors.Is(err, pgx.ErrNoRows) {
+			return ErrResourceNotFound
+		}
+		return fmt.Errorf("cannot lock campaign: %w", err)
+	}
+
+	return nil
+}
+
+func (f *AccessReviewCampaignSourceFetch) UpsertQueued(
+	ctx context.Context,
+	conn pg.Conn,
+	scope Scoper,
+	now time.Time,
+) error {
+	q := `
+INSERT INTO access_review_campaign_source_fetches (
+	tenant_id,
+	access_review_campaign_id,
+	access_source_id,
+	status,
+	fetched_accounts_count,
+	attempt_count,
+	last_error,
+	started_at,
+	completed_at,
+	created_at,
+	updated_at
+) VALUES (
+	@tenant_id, @access_review_campaign_id, @access_source_id,
+	'QUEUED', 0, 0, NULL, NULL, NULL, @now, @now
+)
+ON CONFLICT (access_review_campaign_id, access_source_id) DO UPDATE SET
+	status = 'QUEUED',
+	fetched_accounts_count = 0,
+	attempt_count = 0,
+	last_error = NULL,
+	started_at = NULL,
+	completed_at = NULL,
+	updated_at = EXCLUDED.updated_at
+`
+	args := pgx.StrictNamedArgs{
+		"tenant_id":                 scope.GetTenantID(),
+		"access_review_campaign_id": f.AccessReviewCampaignID,
+		"access_source_id":          f.AccessSourceID,
+		"now":                       now,
+	}
+
+	_, err := conn.Exec(ctx, q, args)
+	if err != nil {
+		return fmt.Errorf("cannot upsert queued source fetch: %w", err)
+	}
+
+	return nil
+}
+
+// RecoverStale is intentionally cross-tenant: the background worker recovers
+// all stale fetches regardless of tenant.
+func (fs *AccessReviewCampaignSourceFetches) RecoverStale(
+	ctx context.Context,
+	conn pg.Conn,
+	staleThreshold time.Time,
+	now time.Time,
+) (int64, error) {
+	q := `
+UPDATE access_review_campaign_source_fetches
+SET
+	status = 'QUEUED',
+	last_error = 'recovered from stale FETCHING state',
+	started_at = NULL,
+	completed_at = NULL,
+	updated_at = @now
+WHERE
+	status = 'FETCHING'
+	AND updated_at < @stale_threshold
+`
+	args := pgx.StrictNamedArgs{
+		"now":             now,
+		"stale_threshold": staleThreshold,
+	}
+
+	result, err := conn.Exec(ctx, q, args)
+	if err != nil {
+		return 0, fmt.Errorf("cannot recover stale source fetches: %w", err)
+	}
+
+	return result.RowsAffected(), nil
+}

pkg/coredata/access_review_campaign_source_fetch.go

@@ -0,0 +1,302 @@
+// Copyright (c) 2026 Probo Inc <hello@getprobo.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.
+
+package coredata
+
+import (
+	"context"
+	"errors"
+	"fmt"
+	"maps"
+	"time"
+
+	"github.com/jackc/pgx/v5"
+	"go.gearno.de/kit/pg"
+	"go.probo.inc/probo/pkg/gid"
+)
+
+type (
+	// AccessReviewCampaignSourceFetch tracks per-source fetch lifecycle.
+	// TenantID is retained on the struct because the background worker claims
+	// rows cross-tenant via LoadNextQueuedForUpdateSkipLocked and needs the
+	// tenant to construct a Scope for subsequent operations.
+	AccessReviewCampaignSourceFetch struct {
+		TenantID               gid.TenantID                          `db:"tenant_id"`
+		AccessReviewCampaignID gid.GID                               `db:"access_review_campaign_id"`
+		AccessSourceID         gid.GID                               `db:"access_source_id"`
+		Status                 AccessReviewCampaignSourceFetchStatus `db:"status"`
+		FetchedAccountsCount   int                                   `db:"fetched_accounts_count"`
+		AttemptCount           int                                   `db:"attempt_count"`
+		LastError              *string                               `db:"last_error"`
+		StartedAt              *time.Time                            `db:"started_at"`
+		CompletedAt            *time.Time                            `db:"completed_at"`
+		CreatedAt              time.Time                             `db:"created_at"`
+		UpdatedAt              time.Time                             `db:"updated_at"`
+	}
+
+	AccessReviewCampaignSourceFetches []*AccessReviewCampaignSourceFetch
+)
+
+var (
+	ErrNoAccessReviewCampaignSourceFetchAvailable = errors.New("no access review campaign source fetch available")
+)
+
+func (f *AccessReviewCampaignSourceFetch) Insert(
+	ctx context.Context,
+	conn pg.Conn,
+	scope Scoper,
+) error {
+	q := `
+INSERT INTO access_review_campaign_source_fetches (
+	tenant_id,
+	access_review_campaign_id,
+	access_source_id,
+	status,
+	fetched_accounts_count,
+	attempt_count,
+	last_error,
+	started_at,
+	completed_at,
+	created_at,
+	updated_at
+) VALUES (
+	@tenant_id,
+	@access_review_campaign_id,
+	@access_source_id,
+	@status,
+	@fetched_accounts_count,
+	@attempt_count,
+	@last_error,
+	@started_at,
+	@completed_at,
+	@created_at,
+	@updated_at
+)
+`
+	args := pgx.StrictNamedArgs{
+		"tenant_id":                 scope.GetTenantID(),
+		"access_review_campaign_id": f.AccessReviewCampaignID,
+		"access_source_id":          f.AccessSourceID,
+		"status":                    f.Status,
+		"fetched_accounts_count":    f.FetchedAccountsCount,
+		"attempt_count":             f.AttemptCount,
+		"last_error":                f.LastError,
+		"started_at":                f.StartedAt,
+		"completed_at":              f.CompletedAt,
+		"created_at":                f.CreatedAt,
+		"updated_at":                f.UpdatedAt,
+	}
+	_, err := conn.Exec(ctx, q, args)
+	if err != nil {
+		return fmt.Errorf("cannot insert campaign source fetch: %w", err)
+	}
+
+	return nil
+}
+
+func (f *AccessReviewCampaignSourceFetch) Update(
+	ctx context.Context,
+	conn pg.Conn,
+	scope Scoper,
+) error {
+	q := `
+UPDATE access_review_campaign_source_fetches
+SET
+	status = @status,
+	fetched_accounts_count = @fetched_accounts_count,
+	attempt_count = @attempt_count,
+	last_error = @last_error,
+	started_at = @started_at,
+	completed_at = @completed_at,
+	updated_at = @updated_at
+WHERE
+	%s
+	AND access_review_campaign_id = @access_review_campaign_id
+	AND access_source_id = @access_source_id
+`
+	q = fmt.Sprintf(q, scope.SQLFragment())
+
+	args := pgx.StrictNamedArgs{
+		"status":                    f.Status,
+		"fetched_accounts_count":    f.FetchedAccountsCount,
+		"attempt_count":             f.AttemptCount,
+		"last_error":                f.LastError,
+		"started_at":                f.StartedAt,
+		"completed_at":              f.CompletedAt,
+		"updated_at":                f.UpdatedAt,
+		"access_review_campaign_id": f.AccessReviewCampaignID,
+		"access_source_id":          f.AccessSourceID,
+	}
+	maps.Copy(args, scope.SQLArguments())
+
+	result, err := conn.Exec(ctx, q, args)
+	if err != nil {
+		return fmt.Errorf("cannot update campaign source fetch: %w", err)
+	}
+
+	if result.RowsAffected() == 0 {
+		return ErrResourceNotFound
+	}
+
+	return nil
+}
+
+func (f *AccessReviewCampaignSourceFetch) LoadByID(
+	ctx context.Context,
+	conn pg.Conn,
+	scope Scoper,
+	campaignID gid.GID,
+	sourceID gid.GID,
+) error {
+	q := `
+SELECT
+	tenant_id,
+	access_review_campaign_id,
+	access_source_id,
+	status,
+	fetched_accounts_count,
+	attempt_count,
+	last_error,
+	started_at,
+	completed_at,
+	created_at,
+	updated_at
+FROM access_review_campaign_source_fetches
+WHERE
+	%s
+	AND access_review_campaign_id = @access_review_campaign_id
+	AND access_source_id = @access_source_id
+LIMIT 1
+`
+	q = fmt.Sprintf(q, scope.SQLFragment())
+
+	args := pgx.StrictNamedArgs{
+		"access_review_campaign_id": campaignID,
+		"access_source_id":          sourceID,
+	}
+	maps.Copy(args, scope.SQLArguments())
+
+	rows, err := conn.Query(ctx, q, args)
+	if err != nil {
+		return fmt.Errorf("cannot query campaign source fetch: %w", err)
+	}
+
+	result, err := pgx.CollectExactlyOneRow(rows, pgx.RowToStructByName[AccessReviewCampaignSourceFetch])
+	if err != nil {
+		if errors.Is(err, pgx.ErrNoRows) {
+			return ErrResourceNotFound
+		}
+		return fmt.Errorf("cannot collect campaign source fetch: %w", err)
+	}
+
+	*f = result
+
+	return nil
+}
+
+func (fs *AccessReviewCampaignSourceFetches) LoadByCampaignID(
+	ctx context.Context,
+	conn pg.Conn,
+	scope Scoper,
+	campaignID gid.GID,
+) error {
+	q := `
+SELECT
+	tenant_id,
+	access_review_campaign_id,
+	access_source_id,
+	status,
+	fetched_accounts_count,
+	attempt_count,
+	last_error,
+	started_at,
+	completed_at,
+	created_at,
+	updated_at
+FROM access_review_campaign_source_fetches
+WHERE
+	%s
+	AND access_review_campaign_id = @access_review_campaign_id
+ORDER BY created_at ASC
+`
+	q = fmt.Sprintf(q, scope.SQLFragment())
+
+	args := pgx.StrictNamedArgs{
+		"access_review_campaign_id": campaignID,
+	}
+	maps.Copy(args, scope.SQLArguments())
+
+	rows, err := conn.Query(ctx, q, args)
+	if err != nil {
+		return fmt.Errorf("cannot query campaign source fetches: %w", err)
+	}
+
+	result, err := pgx.CollectRows(rows, pgx.RowToAddrOfStructByName[AccessReviewCampaignSourceFetch])
+	if err != nil {
+		return fmt.Errorf("cannot collect campaign source fetches: %w", err)
+	}
+
+	*fs = result
+
+	return nil
+}
+
+// LoadNextQueuedForUpdateSkipLocked is intentionally cross-tenant: the
+// background worker claims the next available fetch regardless of tenant.
+// The caller extracts TenantID from the returned struct to construct a
+// Scope for subsequent operations.
+func (f *AccessReviewCampaignSourceFetch) LoadNextQueuedForUpdateSkipLocked(
+	ctx context.Context,
+	conn pg.Conn,
+) error {
+	q := `
+SELECT
+	tenant_id,
+	access_review_campaign_id,
+	access_source_id,
+	status,
+	fetched_accounts_count,
+	attempt_count,
+	last_error,
+	started_at,
+	completed_at,
+	created_at,
+	updated_at
+FROM access_review_campaign_source_fetches
+WHERE status = @status
+ORDER BY created_at ASC
+LIMIT 1
+FOR UPDATE SKIP LOCKED
+`
+	args := pgx.StrictNamedArgs{
+		"status": AccessReviewCampaignSourceFetchStatusQueued,
+	}
+
+	rows, err := conn.Query(ctx, q, args)
+	if err != nil {
+		return fmt.Errorf("cannot query next queued campaign source fetch: %w", err)
+	}
+
+	result, err := pgx.CollectExactlyOneRow(rows, pgx.RowToStructByName[AccessReviewCampaignSourceFetch])
+	if err != nil {
+		if errors.Is(err, pgx.ErrNoRows) {
+			return ErrNoAccessReviewCampaignSourceFetchAvailable
+		}
+		return fmt.Errorf("cannot collect campaign source fetch: %w", err)
+	}
+
+	*f = result
+
+	return nil
+}

pkg/coredata/access_review_campaign_source_fetch_status.go

@@ -0,0 +1,68 @@
+// Copyright (c) 2026 Probo Inc <hello@getprobo.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.
+
+package coredata
+
+import (
+	"database/sql/driver"
+	"fmt"
+)
+
+type AccessReviewCampaignSourceFetchStatus string
+
+const (
+	AccessReviewCampaignSourceFetchStatusQueued   AccessReviewCampaignSourceFetchStatus = "QUEUED"
+	AccessReviewCampaignSourceFetchStatusFetching AccessReviewCampaignSourceFetchStatus = "FETCHING"
+	AccessReviewCampaignSourceFetchStatusSuccess  AccessReviewCampaignSourceFetchStatus = "SUCCESS"
+	AccessReviewCampaignSourceFetchStatusFailed   AccessReviewCampaignSourceFetchStatus = "FAILED"
+)
+
+func (s AccessReviewCampaignSourceFetchStatus) IsTerminal() bool {
+	return s == AccessReviewCampaignSourceFetchStatusSuccess || s == AccessReviewCampaignSourceFetchStatusFailed
+}
+
+func (s AccessReviewCampaignSourceFetchStatus) String() string {
+	return string(s)
+}
+
+func (s *AccessReviewCampaignSourceFetchStatus) Scan(value any) error {
+	var str string
+	switch v := value.(type) {
+	case string:
+		str = v
+	case []byte:
+		str = string(v)
+	default:
+		return fmt.Errorf("cannot scan AccessReviewCampaignSourceFetchStatus: unsupported type %T", value)
+	}
+
+	switch str {
+	case "QUEUED":
+		*s = AccessReviewCampaignSourceFetchStatusQueued
+	case "FETCHING":
+		*s = AccessReviewCampaignSourceFetchStatusFetching
+	case "SUCCESS":
+		*s = AccessReviewCampaignSourceFetchStatusSuccess
+	case "FAILED":
+		*s = AccessReviewCampaignSourceFetchStatusFailed
+	default:
+		return fmt.Errorf("cannot parse AccessReviewCampaignSourceFetchStatus: invalid value %q", str)
+	}
+
+	return nil
+}
+
+func (s AccessReviewCampaignSourceFetchStatus) Value() (driver.Value, error) {
+	return s.String(), nil
+}

pkg/coredata/access_review_campaign_source_fetch_status_test.go

@@ -0,0 +1,83 @@
+// Copyright (c) 2026 Probo Inc <hello@getprobo.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.
+
+package coredata
+
+import "testing"
+
+func TestAccessReviewCampaignSourceFetchStatusIsTerminal(t *testing.T) {
+	t.Parallel()
+
+	if AccessReviewCampaignSourceFetchStatusQueued.IsTerminal() {
+		t.Fatalf("QUEUED should not be terminal")
+	}
+	if AccessReviewCampaignSourceFetchStatusFetching.IsTerminal() {
+		t.Fatalf("FETCHING should not be terminal")
+	}
+	if !AccessReviewCampaignSourceFetchStatusSuccess.IsTerminal() {
+		t.Fatalf("SUCCESS should be terminal")
+	}
+	if !AccessReviewCampaignSourceFetchStatusFailed.IsTerminal() {
+		t.Fatalf("FAILED should be terminal")
+	}
+}
+
+func TestAccessReviewCampaignSourceFetchStatusScan(t *testing.T) {
+	t.Parallel()
+
+	tests := []struct {
+		name    string
+		input   any
+		want    AccessReviewCampaignSourceFetchStatus
+		wantErr bool
+	}{
+		{
+			name:  "queued string",
+			input: "QUEUED",
+			want:  AccessReviewCampaignSourceFetchStatusQueued,
+		},
+		{
+			name:  "fetching bytes",
+			input: []byte("FETCHING"),
+			want:  AccessReviewCampaignSourceFetchStatusFetching,
+		},
+		{
+			name:    "invalid value",
+			input:   "BOGUS",
+			wantErr: true,
+		},
+	}
+
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			t.Parallel()
+
+			var got AccessReviewCampaignSourceFetchStatus
+			err := got.Scan(tt.input)
+			if tt.wantErr {
+				if err == nil {
+					t.Fatalf("Scan(%v) expected error", tt.input)
+				}
+				return
+			}
+
+			if err != nil {
+				t.Fatalf("Scan(%v) returned error: %v", tt.input, err)
+			}
+			if got != tt.want {
+				t.Fatalf("Scan(%v) = %q, want %q", tt.input, got, tt.want)
+			}
+		})
+	}
+}

pkg/coredata/access_review_campaign_status.go

@@ -0,0 +1,69 @@
+// Copyright (c) 2026 Probo Inc <hello@getprobo.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.
+
+package coredata
+
+import (
+	"database/sql/driver"
+	"fmt"
+)
+
+type AccessReviewCampaignStatus string
+
+const (
+	AccessReviewCampaignStatusDraft          AccessReviewCampaignStatus = "DRAFT"
+	AccessReviewCampaignStatusInProgress     AccessReviewCampaignStatus = "IN_PROGRESS"
+	AccessReviewCampaignStatusPendingActions AccessReviewCampaignStatus = "PENDING_ACTIONS"
+	AccessReviewCampaignStatusFailed         AccessReviewCampaignStatus = "FAILED"
+	AccessReviewCampaignStatusCompleted      AccessReviewCampaignStatus = "COMPLETED"
+	AccessReviewCampaignStatusCancelled      AccessReviewCampaignStatus = "CANCELLED"
+)
+
+func (s AccessReviewCampaignStatus) String() string {
+	return string(s)
+}
+
+func (s *AccessReviewCampaignStatus) Scan(value any) error {
+	var str string
+	switch v := value.(type) {
+	case string:
+		str = v
+	case []byte:
+		str = string(v)
+	default:
+		return fmt.Errorf("cannot scan AccessReviewCampaignStatus: unsupported type %T", value)
+	}
+
+	switch str {
+	case "DRAFT":
+		*s = AccessReviewCampaignStatusDraft
+	case "IN_PROGRESS":
+		*s = AccessReviewCampaignStatusInProgress
+	case "PENDING_ACTIONS":
+		*s = AccessReviewCampaignStatusPendingActions
+	case "FAILED":
+		*s = AccessReviewCampaignStatusFailed
+	case "COMPLETED":
+		*s = AccessReviewCampaignStatusCompleted
+	case "CANCELLED":
+		*s = AccessReviewCampaignStatusCancelled
+	default:
+		return fmt.Errorf("cannot parse AccessReviewCampaignStatus: invalid value %q", str)
+	}
+	return nil
+}
+
+func (s AccessReviewCampaignStatus) Value() (driver.Value, error) {
+	return s.String(), nil
+}

pkg/coredata/access_source.go

@@ -0,0 +1,408 @@
+// Copyright (c) 2025-2026 Probo Inc <hello@getprobo.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.
+
+package coredata
+
+import (
+	"context"
+	"errors"
+	"fmt"
+	"maps"
+	"time"
+
+	"github.com/jackc/pgx/v5"
+	"go.gearno.de/kit/pg"
+	"go.probo.inc/probo/pkg/gid"
+	"go.probo.inc/probo/pkg/page"
+)
+
+type (
+	AccessSource struct {
+		ID             gid.GID              `db:"id"`
+		OrganizationID gid.GID              `db:"organization_id"`
+		ConnectorID    *gid.GID             `db:"connector_id"`
+		Name           string               `db:"name"`
+		Category       AccessSourceCategory `db:"category"`
+		CsvData        *string              `db:"csv_data"`
+		NameSyncedAt   *time.Time           `db:"name_synced_at"`
+		CreatedAt      time.Time            `db:"created_at"`
+		UpdatedAt      time.Time            `db:"updated_at"`
+	}
+
+	AccessSources []*AccessSource
+)
+
+func (as AccessSource) CursorKey(orderBy AccessSourceOrderField) page.CursorKey {
+	switch orderBy {
+	case AccessSourceOrderFieldCreatedAt:
+		return page.NewCursorKey(as.ID, as.CreatedAt)
+	}
+
+	panic(fmt.Sprintf("unsupported order by: %s", orderBy))
+}
+
+func (as *AccessSource) AuthorizationAttributes(ctx context.Context, conn pg.Conn) (map[string]string, error) {
+	q := `SELECT organization_id FROM access_sources WHERE id = $1 LIMIT 1;`
+
+	var organizationID gid.GID
+	if err := conn.QueryRow(ctx, q, as.ID).Scan(&organizationID); err != nil {
+		if errors.Is(err, pgx.ErrNoRows) {
+			return nil, ErrResourceNotFound
+		}
+		return nil, fmt.Errorf("cannot query access source authorization attributes: %w", err)
+	}
+
+	return map[string]string{"organization_id": organizationID.String()}, nil
+}
+
+func (as *AccessSource) LoadByID(
+	ctx context.Context,
+	conn pg.Conn,
+	scope Scoper,
+	id gid.GID,
+) error {
+	q := `
+SELECT
+    id,
+    organization_id,
+    connector_id,
+    name,
+    category,
+    csv_data,
+    name_synced_at,
+    created_at,
+    updated_at
+FROM
+    access_sources
+WHERE
+    %s
+    AND id = @id
+LIMIT 1;
+`
+	q = fmt.Sprintf(q, scope.SQLFragment())
+
+	args := pgx.StrictNamedArgs{"id": id}
+	maps.Copy(args, scope.SQLArguments())
+
+	rows, err := conn.Query(ctx, q, args)
+	if err != nil {
+		return fmt.Errorf("cannot query access_sources: %w", err)
+	}
+
+	source, err := pgx.CollectExactlyOneRow(rows, pgx.RowToStructByName[AccessSource])
+	if err != nil {
+		if errors.Is(err, pgx.ErrNoRows) {
+			return ErrResourceNotFound
+		}
+		return fmt.Errorf("cannot collect access source: %w", err)
+	}
+
+	*as = source
+
+	return nil
+}
+
+func (as *AccessSource) Insert(
+	ctx context.Context,
+	conn pg.Conn,
+	scope Scoper,
+) error {
+	q := `
+INSERT INTO
+    access_sources (
+        id,
+        tenant_id,
+        organization_id,
+        connector_id,
+        name,
+        category,
+        csv_data,
+        name_synced_at,
+        created_at,
+        updated_at
+    )
+VALUES (
+    @id,
+    @tenant_id,
+    @organization_id,
+    @connector_id,
+    @name,
+    @category,
+    @csv_data,
+    @name_synced_at,
+    @created_at,
+    @updated_at
+);
+`
+
+	args := pgx.StrictNamedArgs{
+		"id":              as.ID,
+		"tenant_id":       scope.GetTenantID(),
+		"organization_id": as.OrganizationID,
+		"connector_id":    as.ConnectorID,
+		"name":            as.Name,
+		"category":        as.Category,
+		"csv_data":        as.CsvData,
+		"name_synced_at":  as.NameSyncedAt,
+		"created_at":      as.CreatedAt,
+		"updated_at":      as.UpdatedAt,
+	}
+	_, err := conn.Exec(ctx, q, args)
+	if err != nil {
+		return fmt.Errorf("cannot insert access_source: %w", err)
+	}
+
+	return nil
+}
+
+func (as *AccessSource) Update(
+	ctx context.Context,
+	conn pg.Conn,
+	scope Scoper,
+) error {
+	q := `
+UPDATE access_sources
+SET
+    name = @name,
+    category = @category,
+    connector_id = @connector_id,
+    csv_data = @csv_data,
+    name_synced_at = @name_synced_at,
+    updated_at = @updated_at
+WHERE
+    %s
+    AND id = @id
+`
+	q = fmt.Sprintf(q, scope.SQLFragment())
+
+	args := pgx.StrictNamedArgs{
+		"id":             as.ID,
+		"name":           as.Name,
+		"category":       as.Category,
+		"connector_id":   as.ConnectorID,
+		"csv_data":       as.CsvData,
+		"name_synced_at": as.NameSyncedAt,
+		"updated_at":     as.UpdatedAt,
+	}
+	maps.Copy(args, scope.SQLArguments())
+
+	result, err := conn.Exec(ctx, q, args)
+	if err != nil {
+		return fmt.Errorf("cannot update access_source: %w", err)
+	}
+
+	if result.RowsAffected() == 0 {
+		return ErrResourceNotFound
+	}
+
+	return nil
+}
+
+func (as *AccessSource) Delete(
+	ctx context.Context,
+	conn pg.Conn,
+	scope Scoper,
+) error {
+	q := `
+DELETE FROM access_sources
+WHERE %s AND id = @id
+`
+	q = fmt.Sprintf(q, scope.SQLFragment())
+
+	args := pgx.StrictNamedArgs{"id": as.ID}
+	maps.Copy(args, scope.SQLArguments())
+
+	result, err := conn.Exec(ctx, q, args)
+	if err != nil {
+		return fmt.Errorf("cannot delete access_source: %w", err)
+	}
+
+	if result.RowsAffected() == 0 {
+		return ErrResourceNotFound
+	}
+
+	return nil
+}
+
+func (sources *AccessSources) LoadByOrganizationID(
+	ctx context.Context,
+	conn pg.Conn,
+	scope Scoper,
+	organizationID gid.GID,
+	cursor *page.Cursor[AccessSourceOrderField],
+) error {
+	q := `
+SELECT
+    id,
+    organization_id,
+    connector_id,
+    name,
+    category,
+    csv_data,
+    name_synced_at,
+    created_at,
+    updated_at
+FROM
+    access_sources
+WHERE
+    %s
+    AND organization_id = @organization_id
+    AND %s
+`
+	q = fmt.Sprintf(q, scope.SQLFragment(), cursor.SQLFragment())
+
+	args := pgx.StrictNamedArgs{"organization_id": organizationID}
+	maps.Copy(args, scope.SQLArguments())
+	maps.Copy(args, cursor.SQLArguments())
+
+	rows, err := conn.Query(ctx, q, args)
+	if err != nil {
+		return fmt.Errorf("cannot query access_sources: %w", err)
+	}
+
+	result, err := pgx.CollectRows(rows, pgx.RowToAddrOfStructByName[AccessSource])
+	if err != nil {
+		return fmt.Errorf("cannot collect access_sources: %w", err)
+	}
+
+	*sources = result
+
+	return nil
+}
+
+func (sources *AccessSources) CountByOrganizationID(
+	ctx context.Context,
+	conn pg.Conn,
+	scope Scoper,
+	organizationID gid.GID,
+) (int, error) {
+	q := `
+SELECT COUNT(id)
+FROM access_sources
+WHERE
+    %s
+    AND organization_id = @organization_id;
+`
+	q = fmt.Sprintf(q, scope.SQLFragment())
+
+	args := pgx.StrictNamedArgs{"organization_id": organizationID}
+	maps.Copy(args, scope.SQLArguments())
+
+	var count int
+	if err := conn.QueryRow(ctx, q, args).Scan(&count); err != nil {
+		return 0, fmt.Errorf("cannot count access_sources: %w", err)
+	}
+
+	return count, nil
+}
+
+// LoadScopeSourcesByCampaignID loads the campaign scope sources in deterministic
+// name order. Only explicitly scoped sources are returned.
+func (sources *AccessSources) LoadScopeSourcesByCampaignID(
+	ctx context.Context,
+	conn pg.Conn,
+	scope Scoper,
+	campaignID gid.GID,
+) error {
+	q := `
+SELECT
+    id,
+    organization_id,
+    connector_id,
+    name,
+    category,
+    csv_data,
+    name_synced_at,
+    created_at,
+    updated_at
+FROM
+    access_sources
+WHERE
+    %s
+    AND id IN (
+        SELECT arcss.access_source_id
+        FROM access_review_campaign_scope_systems arcss
+        WHERE arcss.access_review_campaign_id = @campaign_id
+    )
+ORDER BY name ASC
+`
+	q = fmt.Sprintf(q, scope.SQLFragment())
+
+	args := pgx.StrictNamedArgs{"campaign_id": campaignID}
+	maps.Copy(args, scope.SQLArguments())
+
+	rows, err := conn.Query(ctx, q, args)
+	if err != nil {
+		return fmt.Errorf("cannot query scope access_sources: %w", err)
+	}
+
+	result, err := pgx.CollectRows(rows, pgx.RowToAddrOfStructByName[AccessSource])
+	if err != nil {
+		return fmt.Errorf("cannot collect scope access_sources: %w", err)
+	}
+
+	*sources = result
+
+	return nil
+}
+
+// ErrNoAccessSourceNameSyncAvailable is returned when no access source
+// needs its name synced from its connector.
+var ErrNoAccessSourceNameSyncAvailable = fmt.Errorf("no access source name sync available")
+
+// LoadNextUnsyncedNameForUpdateSkipLocked claims the next access source that
+// has a connector but has not yet had its name synced. The row is locked with
+// FOR UPDATE SKIP LOCKED so concurrent workers do not pick the same row.
+func (as *AccessSource) LoadNextUnsyncedNameForUpdateSkipLocked(
+	ctx context.Context,
+	conn pg.Conn,
+) error {
+	q := `
+SELECT
+    id,
+    organization_id,
+    connector_id,
+    name,
+    category,
+    csv_data,
+    name_synced_at,
+    created_at,
+    updated_at
+FROM
+    access_sources
+WHERE
+    connector_id IS NOT NULL
+    AND name_synced_at IS NULL
+ORDER BY
+    created_at ASC
+LIMIT 1
+FOR UPDATE SKIP LOCKED;
+`
+
+	rows, err := conn.Query(ctx, q)
+	if err != nil {
+		return fmt.Errorf("cannot query unsynced access_sources: %w", err)
+	}
+
+	row, err := pgx.CollectExactlyOneRow(rows, pgx.RowToStructByName[AccessSource])
+	if err != nil {
+		if errors.Is(err, pgx.ErrNoRows) {
+			return ErrNoAccessSourceNameSyncAvailable
+		}
+		return fmt.Errorf("cannot collect unsynced access source: %w", err)
+	}
+
+	*as = row
+	return nil
+}

pkg/coredata/access_source_category.go

@@ -0,0 +1,72 @@
+// Copyright (c) 2026 Probo Inc <hello@getprobo.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.
+
+package coredata
+
+import (
+	"database/sql/driver"
+	"fmt"
+)
+
+type AccessSourceCategory string
+
+const (
+	AccessSourceCategorySaaS       AccessSourceCategory = "SAAS"
+	AccessSourceCategoryCloudInfra AccessSourceCategory = "CLOUD_INFRA"
+	AccessSourceCategorySourceCode AccessSourceCategory = "SOURCE_CODE"
+	AccessSourceCategoryOther      AccessSourceCategory = "OTHER"
+)
+
+func AccessSourceCategories() []AccessSourceCategory {
+	return []AccessSourceCategory{
+		AccessSourceCategorySaaS,
+		AccessSourceCategoryCloudInfra,
+		AccessSourceCategorySourceCode,
+		AccessSourceCategoryOther,
+	}
+}
+
+func (c AccessSourceCategory) String() string {
+	return string(c)
+}
+
+func (c *AccessSourceCategory) Scan(value any) error {
+	var str string
+	switch v := value.(type) {
+	case string:
+		str = v
+	case []byte:
+		str = string(v)
+	default:
+		return fmt.Errorf("cannot scan AccessSourceCategory: unsupported type %T", value)
+	}
+
+	switch str {
+	case "SAAS":
+		*c = AccessSourceCategorySaaS
+	case "CLOUD_INFRA":
+		*c = AccessSourceCategoryCloudInfra
+	case "SOURCE_CODE":
+		*c = AccessSourceCategorySourceCode
+	case "OTHER":
+		*c = AccessSourceCategoryOther
+	default:
+		return fmt.Errorf("cannot parse AccessSourceCategory: invalid value %q", str)
+	}
+	return nil
+}
+
+func (c AccessSourceCategory) Value() (driver.Value, error) {
+	return c.String(), nil
+}

pkg/coredata/access_source_order_field.go

@@ -0,0 +1,57 @@
+// Copyright (c) 2025-2026 Probo Inc <hello@getprobo.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.
+
+package coredata
+
+import "fmt"
+
+type (
+	AccessSourceOrderField string
+)
+
+const (
+	AccessSourceOrderFieldCreatedAt AccessSourceOrderField = "CREATED_AT"
+)
+
+func (p AccessSourceOrderField) Column() string {
+	switch p {
+	case AccessSourceOrderFieldCreatedAt:
+		return "created_at"
+	}
+	panic(fmt.Sprintf("unsupported order by: %s", p))
+}
+
+func (p AccessSourceOrderField) IsValid() bool {
+	switch p {
+	case AccessSourceOrderFieldCreatedAt:
+		return true
+	}
+	return false
+}
+
+func (p AccessSourceOrderField) String() string {
+	return string(p)
+}
+
+func (p AccessSourceOrderField) MarshalText() ([]byte, error) {
+	return []byte(p.String()), nil
+}
+
+func (p *AccessSourceOrderField) UnmarshalText(text []byte) error {
+	*p = AccessSourceOrderField(text)
+	if !p.IsValid() {
+		return fmt.Errorf("%s is not a valid AccessSourceOrderField", string(text))
+	}
+	return nil
+}

pkg/coredata/auth_method.go

@@ -0,0 +1,76 @@
+// Copyright (c) 2026 Probo Inc <hello@getprobo.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.
+
+package coredata
+
+import (
+	"database/sql/driver"
+	"fmt"
+)
+
+type AccessEntryAuthMethod string
+
+const (
+	AccessEntryAuthMethodSSO            AccessEntryAuthMethod = "SSO"
+	AccessEntryAuthMethodPassword       AccessEntryAuthMethod = "PASSWORD"
+	AccessEntryAuthMethodAPIKey         AccessEntryAuthMethod = "API_KEY"
+	AccessEntryAuthMethodServiceAccount AccessEntryAuthMethod = "SERVICE_ACCOUNT"
+	AccessEntryAuthMethodUnknown        AccessEntryAuthMethod = "UNKNOWN"
+)
+
+func AccessEntryAuthMethods() []AccessEntryAuthMethod {
+	return []AccessEntryAuthMethod{
+		AccessEntryAuthMethodSSO,
+		AccessEntryAuthMethodPassword,
+		AccessEntryAuthMethodAPIKey,
+		AccessEntryAuthMethodServiceAccount,
+		AccessEntryAuthMethodUnknown,
+	}
+}
+
+func (a AccessEntryAuthMethod) String() string {
+	return string(a)
+}
+
+func (a *AccessEntryAuthMethod) Scan(value any) error {
+	var str string
+	switch v := value.(type) {
+	case string:
+		str = v
+	case []byte:
+		str = string(v)
+	default:
+		return fmt.Errorf("cannot scan AccessEntryAuthMethod: unsupported type %T", value)
+	}
+
+	switch str {
+	case "SSO":
+		*a = AccessEntryAuthMethodSSO
+	case "PASSWORD":
+		*a = AccessEntryAuthMethodPassword
+	case "API_KEY":
+		*a = AccessEntryAuthMethodAPIKey
+	case "SERVICE_ACCOUNT":
+		*a = AccessEntryAuthMethodServiceAccount
+	case "UNKNOWN":
+		*a = AccessEntryAuthMethodUnknown
+	default:
+		return fmt.Errorf("cannot parse AccessEntryAuthMethod: invalid value %q", str)
+	}
+	return nil
+}
+
+func (a AccessEntryAuthMethod) Value() (driver.Value, error) {
+	return a.String(), nil
+}

pkg/coredata/connector.go

@@ -30,13 +30,37 @@ import (
 	"go.probo.inc/probo/pkg/page"
 )
 
+// jsonRawMessageOrNull is a json.RawMessage that scans NULL as an empty
+// slice and serialises an empty/nil value as SQL NULL.  This avoids the
+// need for *json.RawMessage and keeps the zero-value useful.
+type jsonRawMessageOrNull json.RawMessage
+
+func (j *jsonRawMessageOrNull) Scan(src any) error {
+	if src == nil {
+		*j = nil
+		return nil
+	}
+	switch v := src.(type) {
+	case []byte:
+		cp := make(jsonRawMessageOrNull, len(v))
+		copy(cp, v)
+		*j = cp
+		return nil
+	case string:
+		*j = jsonRawMessageOrNull(v)
+		return nil
+	default:
+		return fmt.Errorf("unsupported type for jsonRawMessageOrNull: %T", src)
+	}
+}
+
 type (
 	Connector struct {
 		ID                  gid.GID              `db:"id"`
 		OrganizationID      gid.GID              `db:"organization_id"`
 		Provider            ConnectorProvider    `db:"provider"`
 		Protocol            ConnectorProtocol    `db:"protocol"`
-		Settings            map[string]any       `db:"settings"`
+		RawSettings         jsonRawMessageOrNull `db:"settings"`
 		Connection          connector.Connection `db:"-"`
 		EncryptedConnection []byte               `db:"encrypted_connection"`
 		CreatedAt           time.Time            `db:"created_at"`
@@ -135,7 +159,13 @@ func (c *Connector) LoadByID(
 			return fmt.Errorf("cannot unmarshal connection: %w", err)
 		}
 
-		c.populateSlackSettings()
+		if c.Provider == ConnectorProviderSlack {
+			if slackConn, ok := c.Connection.(*connector.SlackConnection); ok {
+				settings, _ := c.SlackSettings()
+				slackConn.Settings.Channel = settings.Channel
+				slackConn.Settings.ChannelID = settings.ChannelID
+			}
+		}
 	}
 
 	return nil
@@ -250,7 +280,14 @@ INSERT INTO connectors (
 		return fmt.Errorf("connection is nil")
 	}
 
-	c.extractSlackSettings()
+	if c.Provider == ConnectorProviderSlack {
+		if slackConn, ok := c.Connection.(*connector.SlackConnection); ok {
+			_ = c.SetSettings(&SlackConnectorSettings{
+				Channel:   slackConn.Settings.Channel,
+				ChannelID: slackConn.Settings.ChannelID,
+			})
+		}
+	}
 
 	connection, err := json.Marshal(c.Connection)
 	if err != nil {
@@ -262,13 +299,18 @@ INSERT INTO connectors (
 		return fmt.Errorf("cannot encrypt connection: %w", err)
 	}
 
+	var settingsArg any
+	if len(c.RawSettings) > 0 {
+		settingsArg = []byte(c.RawSettings)
+	}
+
 	args := pgx.StrictNamedArgs{
 		"id":                   c.ID,
 		"tenant_id":            scope.GetTenantID(),
 		"organization_id":      c.OrganizationID,
 		"provider":             c.Provider,
 		"protocol":             c.Protocol,
-		"settings":             c.Settings,
+		"settings":             settingsArg,
 		"encrypted_connection": encryptedConnection,
 		"created_at":           c.CreatedAt,
 		"updated_at":           c.UpdatedAt,
@@ -280,48 +322,10 @@ INSERT INTO connectors (
 	}
 
 	c.EncryptedConnection = encryptedConnection
-	c.populateSlackSettings()
 
 	return nil
 }
 
-func (c *Connector) populateSlackSettings() {
-	if c.Provider != ConnectorProviderSlack {
-		return
-	}
-
-	slackConn, ok := c.Connection.(*connector.SlackConnection)
-	if !ok {
-		return
-	}
-
-	if channel, ok := c.Settings["channel"].(string); ok {
-		slackConn.Settings.Channel = channel
-	}
-	if channelID, ok := c.Settings["channel_id"].(string); ok {
-		slackConn.Settings.ChannelID = channelID
-	}
-}
-
-func (c *Connector) extractSlackSettings() {
-	if c.Provider != ConnectorProviderSlack {
-		return
-	}
-
-	slackConn, ok := c.Connection.(*connector.SlackConnection)
-	if !ok {
-		return
-	}
-
-	c.Settings = make(map[string]any)
-	if slackConn.Settings.Channel != "" {
-		c.Settings["channel"] = slackConn.Settings.Channel
-	}
-	if slackConn.Settings.ChannelID != "" {
-		c.Settings["channel_id"] = slackConn.Settings.ChannelID
-	}
-}
-
 func (c *Connectors) loadByOrganizationIDWithPagination(
 	ctx context.Context,
 	conn pg.Conn,
@@ -492,7 +496,14 @@ WHERE
 		return fmt.Errorf("connection is nil")
 	}
 
-	c.extractSlackSettings()
+	if c.Provider == ConnectorProviderSlack {
+		if slackConn, ok := c.Connection.(*connector.SlackConnection); ok {
+			_ = c.SetSettings(&SlackConnectorSettings{
+				Channel:   slackConn.Settings.Channel,
+				ChannelID: slackConn.Settings.ChannelID,
+			})
+		}
+	}
 
 	connection, err := json.Marshal(c.Connection)
 	if err != nil {
@@ -504,9 +515,14 @@ WHERE
 		return fmt.Errorf("cannot encrypt connection: %w", err)
 	}
 
+	var settingsArg any
+	if len(c.RawSettings) > 0 {
+		settingsArg = []byte(c.RawSettings)
+	}
+
 	args := pgx.StrictNamedArgs{
 		"id":                   c.ID,
-		"settings":             c.Settings,
+		"settings":             settingsArg,
 		"encrypted_connection": encryptedConnection,
 		"updated_at":           c.UpdatedAt,
 	}
@@ -522,7 +538,6 @@ WHERE
 	}
 
 	c.EncryptedConnection = encryptedConnection
-	c.populateSlackSettings()
 
 	return nil
 }
@@ -543,7 +558,13 @@ func (c *Connectors) decryptConnections(encryptionKey cipher.EncryptionKey) erro
 			return fmt.Errorf("cannot unmarshal connection for %s: %w", cnnctr.Provider, err)
 		}
 
-		cnnctr.populateSlackSettings()
+		if cnnctr.Provider == ConnectorProviderSlack {
+			if slackConn, ok := cnnctr.Connection.(*connector.SlackConnection); ok {
+				settings, _ := cnnctr.SlackSettings()
+				slackConn.Settings.Channel = settings.Channel
+				slackConn.Settings.ChannelID = settings.ChannelID
+			}
+		}
 	}
 
 	return nil

pkg/coredata/connector_protocol.go

@@ -23,11 +23,13 @@ type ConnectorProtocol string
 
 const (
 	ConnectorProtocolOAuth2 ConnectorProtocol = "OAUTH2"
+	ConnectorProtocolAPIKey ConnectorProtocol = "API_KEY"
 )
 
 func ConnectorProtocols() []ConnectorProtocol {
 	return []ConnectorProtocol{
 		ConnectorProtocolOAuth2,
+		ConnectorProtocolAPIKey,
 	}
 }
 
@@ -49,6 +51,8 @@ func (cp *ConnectorProtocol) Scan(value any) error {
 	switch s {
 	case "OAUTH2":
 		*cp = ConnectorProtocolOAuth2
+	case "API_KEY":
+		*cp = ConnectorProtocolAPIKey
 	default:
 		return fmt.Errorf("invalid ConnectorProtocol value: %q", s)
 	}

pkg/coredata/connector_provider.go

@@ -24,12 +24,41 @@ type ConnectorProvider string
 const (
 	ConnectorProviderSlack           ConnectorProvider = "SLACK"
 	ConnectorProviderGoogleWorkspace ConnectorProvider = "GOOGLE_WORKSPACE"
+	ConnectorProviderLinear          ConnectorProvider = "LINEAR"
+	// _ ConnectorProvider = "FIGMA" — formerly Figma; removed (no driver, no OAuth config, no usage)
+	ConnectorProviderOnePassword ConnectorProvider = "ONE_PASSWORD"
+	ConnectorProviderHubSpot     ConnectorProvider = "HUBSPOT"
+	ConnectorProviderDocuSign    ConnectorProvider = "DOCUSIGN"
+	ConnectorProviderNotion      ConnectorProvider = "NOTION"
+	ConnectorProviderBrex        ConnectorProvider = "BREX"
+	ConnectorProviderTally       ConnectorProvider = "TALLY"
+	ConnectorProviderCloudflare  ConnectorProvider = "CLOUDFLARE"
+	ConnectorProviderOpenAI      ConnectorProvider = "OPENAI"
+	ConnectorProviderSentry      ConnectorProvider = "SENTRY"
+	ConnectorProviderSupabase    ConnectorProvider = "SUPABASE"
+	ConnectorProviderGitHub      ConnectorProvider = "GITHUB"
+	ConnectorProviderIntercom    ConnectorProvider = "INTERCOM"
+	ConnectorProviderResend      ConnectorProvider = "RESEND"
 )
 
 func ConnectorProviders() []ConnectorProvider {
 	return []ConnectorProvider{
 		ConnectorProviderSlack,
 		ConnectorProviderGoogleWorkspace,
+		ConnectorProviderLinear,
+		ConnectorProviderOnePassword,
+		ConnectorProviderHubSpot,
+		ConnectorProviderDocuSign,
+		ConnectorProviderNotion,
+		ConnectorProviderBrex,
+		ConnectorProviderTally,
+		ConnectorProviderCloudflare,
+		ConnectorProviderOpenAI,
+		ConnectorProviderSentry,
+		ConnectorProviderSupabase,
+		ConnectorProviderGitHub,
+		ConnectorProviderIntercom,
+		ConnectorProviderResend,
 	}
 }
 
@@ -53,6 +82,34 @@ func (cp *ConnectorProvider) Scan(value any) error {
 		*cp = ConnectorProviderSlack
 	case "GOOGLE_WORKSPACE":
 		*cp = ConnectorProviderGoogleWorkspace
+	case "LINEAR":
+		*cp = ConnectorProviderLinear
+	case "ONE_PASSWORD":
+		*cp = ConnectorProviderOnePassword
+	case "HUBSPOT":
+		*cp = ConnectorProviderHubSpot
+	case "DOCUSIGN":
+		*cp = ConnectorProviderDocuSign
+	case "NOTION":
+		*cp = ConnectorProviderNotion
+	case "BREX":
+		*cp = ConnectorProviderBrex
+	case "TALLY":
+		*cp = ConnectorProviderTally
+	case "CLOUDFLARE":
+		*cp = ConnectorProviderCloudflare
+	case "OPENAI":
+		*cp = ConnectorProviderOpenAI
+	case "SENTRY":
+		*cp = ConnectorProviderSentry
+	case "SUPABASE":
+		*cp = ConnectorProviderSupabase
+	case "GITHUB":
+		*cp = ConnectorProviderGitHub
+	case "INTERCOM":
+		*cp = ConnectorProviderIntercom
+	case "RESEND":
+		*cp = ConnectorProviderResend
 	default:
 		return fmt.Errorf("invalid ConnectorProvider value: %q", s)
 	}

pkg/coredata/connector_settings.go

@@ -0,0 +1,135 @@
+// Copyright (c) 2026 Probo Inc <hello@getprobo.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.
+
+package coredata
+
+import (
+	"encoding/json"
+	"fmt"
+)
+
+type (
+	SlackConnectorSettings struct {
+		Channel   string `json:"channel,omitempty"`
+		ChannelID string `json:"channel_id,omitempty"`
+	}
+
+	TallyConnectorSettings struct {
+		OrganizationID string `json:"organization_id"`
+	}
+
+	OnePasswordConnectorSettings struct {
+		SCIMBridgeURL string `json:"scim_bridge_url"`
+	}
+
+	SentryConnectorSettings struct {
+		OrganizationSlug string `json:"organization_slug"`
+	}
+
+	SupabaseConnectorSettings struct {
+		OrganizationSlug string `json:"organization_slug"`
+	}
+
+	GitHubConnectorSettings struct {
+		Organization string `json:"organization"`
+	}
+
+	OnePasswordUsersAPISettings struct {
+		AccountID string `json:"account_id"`
+		Region    string `json:"region"`
+	}
+)
+
+// SetSettings marshals a typed settings struct into the connector's RawSettings.
+func (c *Connector) SetSettings(v any) error {
+	data, err := json.Marshal(v)
+	if err != nil {
+		return fmt.Errorf("cannot marshal connector settings: %w", err)
+	}
+	c.RawSettings = data
+	return nil
+}
+
+// SlackSettings unmarshals the connector's RawSettings into SlackConnectorSettings.
+func (c *Connector) SlackSettings() (SlackConnectorSettings, error) {
+	var s SlackConnectorSettings
+	if err := c.unmarshalSettings(&s); err != nil {
+		return s, err
+	}
+	return s, nil
+}
+
+// TallySettings unmarshals the connector's RawSettings into TallyConnectorSettings.
+func (c *Connector) TallySettings() (TallyConnectorSettings, error) {
+	var s TallyConnectorSettings
+	if err := c.unmarshalSettings(&s); err != nil {
+		return s, err
+	}
+	return s, nil
+}
+
+// OnePasswordSettings unmarshals the connector's RawSettings into OnePasswordConnectorSettings.
+func (c *Connector) OnePasswordSettings() (OnePasswordConnectorSettings, error) {
+	var s OnePasswordConnectorSettings
+	if err := c.unmarshalSettings(&s); err != nil {
+		return s, err
+	}
+	return s, nil
+}
+
+// SentrySettings unmarshals the connector's RawSettings into SentryConnectorSettings.
+func (c *Connector) SentrySettings() (SentryConnectorSettings, error) {
+	var s SentryConnectorSettings
+	if err := c.unmarshalSettings(&s); err != nil {
+		return s, err
+	}
+	return s, nil
+}
+
+// SupabaseSettings unmarshals the connector's RawSettings into SupabaseConnectorSettings.
+func (c *Connector) SupabaseSettings() (SupabaseConnectorSettings, error) {
+	var s SupabaseConnectorSettings
+	if err := c.unmarshalSettings(&s); err != nil {
+		return s, err
+	}
+	return s, nil
+}
+
+// GitHubSettings unmarshals the connector's RawSettings into GitHubConnectorSettings.
+func (c *Connector) GitHubSettings() (GitHubConnectorSettings, error) {
+	var s GitHubConnectorSettings
+	if err := c.unmarshalSettings(&s); err != nil {
+		return s, err
+	}
+	return s, nil
+}
+
+// OnePasswordUsersAPISettings unmarshals the connector's RawSettings into OnePasswordUsersAPISettings.
+func (c *Connector) OnePasswordUsersAPISettings() (OnePasswordUsersAPISettings, error) {
+	var s OnePasswordUsersAPISettings
+	if err := c.unmarshalSettings(&s); err != nil {
+		return s, err
+	}
+	return s, nil
+}
+
+func (c *Connector) unmarshalSettings(v any) error {
+	if len(c.RawSettings) == 0 || string(c.RawSettings) == "null" {
+		return nil
+	}
+	if err := json.Unmarshal(c.RawSettings, v); err != nil {
+		return fmt.Errorf("cannot unmarshal connector settings: %w", err)
+	}
+	return nil
+}

pkg/coredata/entity_type_reg.go

@@ -94,6 +94,10 @@ const (
 	AuditLogEntryEntityType                    uint16 = 68
 	DocumentVersionApprovalQuorumEntityType    uint16 = 69
 	DocumentVersionApprovalDecisionEntityType  uint16 = 70
+	AccessSourceEntityType                     uint16 = 71
+	AccessReviewCampaignEntityType             uint16 = 72
+	AccessEntryEntityType                      uint16 = 73
+	AccessEntryDecisionHistoryEntityType       uint16 = 74
 )
 
 func NewEntityFromID(id gid.GID) (any, bool) {
@@ -232,6 +236,14 @@ func NewEntityFromID(id gid.GID) (any, bool) {
 		return &DocumentVersionApprovalDecision{ID: id}, true
 	case DocumentVersionApprovalQuorumEntityType:
 		return &DocumentVersionApprovalQuorum{ID: id}, true
+	case AccessSourceEntityType:
+		return &AccessSource{ID: id}, true
+	case AccessReviewCampaignEntityType:
+		return &AccessReviewCampaign{ID: id}, true
+	case AccessEntryEntityType:
+		return &AccessEntry{ID: id}, true
+	case AccessEntryDecisionHistoryEntityType:
+		return &AccessEntryDecisionHistory{ID: id}, true
 	default:
 		return nil, false
 	}

pkg/coredata/mfa_status.go

@@ -0,0 +1,68 @@
+// Copyright (c) 2026 Probo Inc <hello@getprobo.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.
+
+package coredata
+
+import (
+	"database/sql/driver"
+	"fmt"
+)
+
+type MFAStatus string
+
+const (
+	MFAStatusEnabled  MFAStatus = "ENABLED"
+	MFAStatusDisabled MFAStatus = "DISABLED"
+	MFAStatusUnknown  MFAStatus = "UNKNOWN"
+)
+
+func MFAStatuses() []MFAStatus {
+	return []MFAStatus{
+		MFAStatusEnabled,
+		MFAStatusDisabled,
+		MFAStatusUnknown,
+	}
+}
+
+func (m MFAStatus) String() string {
+	return string(m)
+}
+
+func (m *MFAStatus) Scan(value any) error {
+	var str string
+	switch v := value.(type) {
+	case string:
+		str = v
+	case []byte:
+		str = string(v)
+	default:
+		return fmt.Errorf("cannot scan MFAStatus: unsupported type %T", value)
+	}
+
+	switch str {
+	case "ENABLED":
+		*m = MFAStatusEnabled
+	case "DISABLED":
+		*m = MFAStatusDisabled
+	case "UNKNOWN":
+		*m = MFAStatusUnknown
+	default:
+		return fmt.Errorf("cannot parse MFAStatus: invalid value %q", str)
+	}
+	return nil
+}
+
+func (m MFAStatus) Value() (driver.Value, error) {
+	return m.String(), nil
+}

pkg/coredata/migrations/20260314T200000Z.sql

@@ -0,0 +1,171 @@
+-- Copyright (c) 2026 Probo Inc <hello@getprobo.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.
+
+-- Access review system: enums and tables
+
+-- Enum types
+CREATE TYPE access_review_campaign_status AS ENUM (
+    'DRAFT',
+    'IN_PROGRESS',
+    'PENDING_ACTIONS',
+    'COMPLETED',
+    'CANCELLED',
+    'FAILED'
+);
+
+CREATE TYPE access_source_category AS ENUM (
+    'SAAS',
+    'CLOUD_INFRA',
+    'SOURCE_CODE',
+    'OTHER'
+);
+
+CREATE TYPE access_entry_flag AS ENUM (
+    'NONE',
+    'ORPHANED',
+    'INACTIVE',
+    'EXCESSIVE',
+    'ROLE_MISMATCH',
+    'NEW'
+);
+
+CREATE TYPE access_entry_decision AS ENUM (
+    'PENDING',
+    'APPROVED',
+    'REVOKE',
+    'DEFER',
+    'ESCALATE'
+);
+
+CREATE TYPE mfa_status AS ENUM (
+    'ENABLED',
+    'DISABLED',
+    'UNKNOWN'
+);
+
+CREATE TYPE auth_method AS ENUM (
+    'SSO',
+    'PASSWORD',
+    'API_KEY',
+    'SERVICE_ACCOUNT',
+    'UNKNOWN'
+);
+
+CREATE TYPE access_entry_incremental_tag AS ENUM (
+    'NEW',
+    'REMOVED',
+    'UNCHANGED'
+);
+
+CREATE TYPE access_review_campaign_source_fetch_status AS ENUM (
+    'QUEUED',
+    'FETCHING',
+    'SUCCESS',
+    'FAILED'
+);
+
+-- Connector provider and protocol extensions
+ALTER TYPE connector_provider ADD VALUE IF NOT EXISTS 'LINEAR';
+ALTER TYPE connector_provider ADD VALUE IF NOT EXISTS 'FIGMA';
+ALTER TYPE connector_provider ADD VALUE IF NOT EXISTS 'ONE_PASSWORD';
+ALTER TYPE connector_provider ADD VALUE IF NOT EXISTS 'HUBSPOT';
+ALTER TYPE connector_provider ADD VALUE IF NOT EXISTS 'DOCUSIGN';
+ALTER TYPE connector_provider ADD VALUE IF NOT EXISTS 'NOTION';
+ALTER TYPE connector_provider ADD VALUE IF NOT EXISTS 'BREX';
+ALTER TYPE connector_provider ADD VALUE IF NOT EXISTS 'TALLY';
+ALTER TYPE connector_provider ADD VALUE IF NOT EXISTS 'CLOUDFLARE';
+ALTER TYPE connector_protocol ADD VALUE IF NOT EXISTS 'API_KEY';
+
+-- 1. access_sources: configured data sources
+CREATE TABLE access_sources (
+    id TEXT PRIMARY KEY,
+    tenant_id TEXT NOT NULL,
+    organization_id TEXT NOT NULL REFERENCES organizations(id),
+    connector_id TEXT REFERENCES connectors(id),
+    name TEXT NOT NULL,
+    category access_source_category NOT NULL,
+    csv_data TEXT,
+    created_at TIMESTAMP WITH TIME ZONE NOT NULL,
+    updated_at TIMESTAMP WITH TIME ZONE NOT NULL
+);
+
+-- 2. access_review_campaigns: individual review campaigns
+CREATE TABLE access_review_campaigns (
+    id TEXT PRIMARY KEY,
+    tenant_id TEXT NOT NULL,
+    organization_id TEXT NOT NULL REFERENCES organizations(id),
+    name TEXT NOT NULL,
+    status access_review_campaign_status NOT NULL DEFAULT 'DRAFT',
+    started_at TIMESTAMP WITH TIME ZONE,
+    completed_at TIMESTAMP WITH TIME ZONE,
+    framework_controls TEXT[],
+    created_at TIMESTAMP WITH TIME ZONE NOT NULL,
+    updated_at TIMESTAMP WITH TIME ZONE NOT NULL
+);
+
+-- 3. access_review_campaign_scope_systems: join table campaigns <-> access_sources
+CREATE TABLE access_review_campaign_scope_systems (
+    access_review_campaign_id TEXT NOT NULL REFERENCES access_review_campaigns(id) ON DELETE CASCADE,
+    access_source_id TEXT NOT NULL REFERENCES access_sources(id) ON DELETE CASCADE,
+    tenant_id TEXT NOT NULL,
+    PRIMARY KEY (access_review_campaign_id, access_source_id)
+);
+
+-- 4. access_entries: individual access records per user per system per campaign
+CREATE TABLE access_entries (
+    id TEXT PRIMARY KEY,
+    tenant_id TEXT NOT NULL,
+    access_review_campaign_id TEXT NOT NULL REFERENCES access_review_campaigns(id) ON DELETE CASCADE,
+    access_source_id TEXT NOT NULL REFERENCES access_sources(id) ON DELETE CASCADE,
+    identity_id TEXT REFERENCES identities(id) ON DELETE SET NULL,
+    email TEXT NOT NULL,
+    full_name TEXT NOT NULL,
+    role TEXT NOT NULL,
+    job_title TEXT NOT NULL,
+    is_admin BOOLEAN NOT NULL,
+    mfa_status mfa_status NOT NULL,
+    auth_method auth_method NOT NULL,
+    last_login TIMESTAMP WITH TIME ZONE,
+    account_created_at TIMESTAMP WITH TIME ZONE,
+    external_id TEXT NOT NULL,
+    account_key TEXT NOT NULL,
+    incremental_tag access_entry_incremental_tag NOT NULL,
+    flag access_entry_flag NOT NULL,
+    flag_reason TEXT,
+    decision access_entry_decision NOT NULL,
+    decision_note TEXT,
+    decided_by TEXT,
+    decided_at TIMESTAMP WITH TIME ZONE,
+    created_at TIMESTAMP WITH TIME ZONE NOT NULL,
+    updated_at TIMESTAMP WITH TIME ZONE NOT NULL
+);
+
+CREATE UNIQUE INDEX idx_access_entries_campaign_source_account_key
+    ON access_entries (access_review_campaign_id, access_source_id, account_key);
+
+-- 5. access_review_campaign_source_fetches: tracks per-source fetch lifecycle
+CREATE TABLE access_review_campaign_source_fetches (
+    tenant_id TEXT NOT NULL,
+    access_review_campaign_id TEXT NOT NULL REFERENCES access_review_campaigns(id) ON DELETE CASCADE,
+    access_source_id TEXT NOT NULL REFERENCES access_sources(id) ON DELETE CASCADE,
+    status access_review_campaign_source_fetch_status NOT NULL,
+    fetched_accounts_count INTEGER NOT NULL,
+    attempt_count INTEGER NOT NULL,
+    last_error TEXT,
+    started_at TIMESTAMP WITH TIME ZONE,
+    completed_at TIMESTAMP WITH TIME ZONE,
+    created_at TIMESTAMP WITH TIME ZONE NOT NULL,
+    updated_at TIMESTAMP WITH TIME ZONE NOT NULL,
+    PRIMARY KEY (access_review_campaign_id, access_source_id)
+);

pkg/coredata/migrations/20260324T120000Z.sql

@@ -1,19 +1,6 @@
--- Copyright (c) 2026 Probo Inc <hello@getprobo.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.
-
 ALTER TABLE trust_centers
 ADD COLUMN search_engine_indexing TEXT NOT NULL DEFAULT 'NOT_INDEXABLE';
 
 ALTER TABLE trust_centers
 ALTER COLUMN search_engine_indexing DROP DEFAULT;
+

pkg/coredata/migrations/20260324T130000Z.sql

@@ -0,0 +1,31 @@
+-- Copyright (c) 2026 Probo Inc <hello@getprobo.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 description to campaigns and decision audit trail
+
+-- 1. Campaign description
+ALTER TABLE access_review_campaigns ADD COLUMN description TEXT NOT NULL DEFAULT '';
+ALTER TABLE access_review_campaigns ALTER COLUMN description DROP DEFAULT;
+
+-- 2. Decision audit trail: immutable log of every decision recorded
+CREATE TABLE access_entry_decision_history (
+    id TEXT PRIMARY KEY,
+    tenant_id TEXT NOT NULL,
+    access_entry_id TEXT NOT NULL REFERENCES access_entries(id) ON DELETE CASCADE,
+    decision access_entry_decision NOT NULL,
+    decision_note TEXT,
+    decided_by TEXT,
+    decided_at TIMESTAMP WITH TIME ZONE NOT NULL,
+    created_at TIMESTAMP WITH TIME ZONE NOT NULL
+);

pkg/coredata/migrations/20260325T130000Z.sql

@@ -0,0 +1,16 @@
+-- Copyright (c) 2026 Probo Inc <hello@getprobo.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.
+
+ALTER TABLE access_sources
+    ADD COLUMN name_synced_at TIMESTAMP WITH TIME ZONE;

pkg/coredata/migrations/20260325T140000Z.sql

@@ -0,0 +1,16 @@
+-- Copyright (c) 2026 Probo Inc <hello@getprobo.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.
+
+ALTER TABLE access_entries
+    ADD COLUMN account_type TEXT NOT NULL DEFAULT 'USER';

pkg/coredata/migrations/20260327T130000Z.sql

@@ -0,0 +1,20 @@
+-- Copyright (c) 2026 Probo Inc <hello@getprobo.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.
+
+ALTER TYPE connector_provider ADD VALUE IF NOT EXISTS 'OPENAI';
+ALTER TYPE connector_provider ADD VALUE IF NOT EXISTS 'SENTRY';
+ALTER TYPE connector_provider ADD VALUE IF NOT EXISTS 'SUPABASE';
+ALTER TYPE connector_provider ADD VALUE IF NOT EXISTS 'GITHUB';
+ALTER TYPE connector_provider ADD VALUE IF NOT EXISTS 'INTERCOM';
+ALTER TYPE connector_provider ADD VALUE IF NOT EXISTS 'RESEND';

pkg/coredata/migrations/20260330T100000Z.sql

@@ -0,0 +1,36 @@
+-- Copyright (c) 2026 Probo Inc <hello@getprobo.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.
+
+-- Convert flag from single TEXT to TEXT array
+ALTER TABLE access_entries
+    ADD COLUMN flags TEXT[] NOT NULL DEFAULT '{}';
+
+-- Migrate existing data: copy non-NONE flag values into the array
+UPDATE access_entries
+SET flags = ARRAY[flag]
+WHERE flag != 'NONE';
+
+-- Convert flag_reason to flag_reasons array
+ALTER TABLE access_entries
+    ADD COLUMN flag_reasons TEXT[] NOT NULL DEFAULT '{}';
+
+-- Migrate existing flag_reason
+UPDATE access_entries
+SET flag_reasons = ARRAY[flag_reason]
+WHERE flag_reason IS NOT NULL AND flag_reason != '';
+
+-- Drop old columns in a single statement
+ALTER TABLE access_entries
+    DROP COLUMN flag,
+    DROP COLUMN flag_reason;

pkg/coredata/migrations/20260330T120000Z.sql

@@ -1,36 +1,40 @@
--- Rename priority to rank
-ALTER TABLE tasks RENAME COLUMN priority TO rank;
+-- Copyright (c) 2026 Probo Inc <hello@getprobo.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.
 
-ALTER TABLE tasks DROP CONSTRAINT tasks_organization_id_state_priority_key;
+-- Add organization_id to access_entries and access_entry_decision_history
+-- to avoid JOINs in AuthorizationAttributes lookups.
 
--- Add task priority enum
-CREATE TYPE task_priority AS ENUM ('URGENT', 'HIGH', 'MEDIUM', 'LOW');
+-- 1. access_entries
+ALTER TABLE access_entries
+    ADD COLUMN organization_id TEXT REFERENCES organizations(id);
 
-ALTER TABLE tasks ADD COLUMN priority task_priority NOT NULL DEFAULT 'MEDIUM'::task_priority;
+UPDATE access_entries ae
+SET organization_id = arc.organization_id
+FROM access_review_campaigns arc
+WHERE ae.access_review_campaign_id = arc.id;
 
-ALTER TABLE tasks ALTER COLUMN priority DROP DEFAULT;
+ALTER TABLE access_entries
+    ALTER COLUMN organization_id SET NOT NULL;
 
--- Rank is now scoped to (state, priority) — backfill ranks per group
-WITH ranked AS (
-    SELECT id, ROW_NUMBER() OVER (
-        PARTITION BY organization_id, state, priority
-        ORDER BY rank
-    ) AS new_rank
-    FROM tasks
-)
-UPDATE tasks SET rank = ranked.new_rank FROM ranked WHERE tasks.id = ranked.id;
+-- 2. access_entry_decision_history
+ALTER TABLE access_entry_decision_history
+    ADD COLUMN organization_id TEXT REFERENCES organizations(id);
 
-ALTER TABLE tasks
-ADD CONSTRAINT tasks_organization_id_state_priority_rank_key
-    UNIQUE (organization_id, state, priority, rank)
-    DEFERRABLE INITIALLY DEFERRED;
+UPDATE access_entry_decision_history h
+SET organization_id = ae.organization_id
+FROM access_entries ae
+WHERE h.access_entry_id = ae.id;
 
--- Computed column for composite ordering (priority level then rank)
-ALTER TABLE tasks ADD COLUMN priority_rank int GENERATED ALWAYS AS (
-    (CASE priority
-        WHEN 'URGENT' THEN 1
-        WHEN 'HIGH' THEN 2
-        WHEN 'MEDIUM' THEN 3
-        WHEN 'LOW' THEN 4
-    END) * 1000000 + rank
-) STORED;
+ALTER TABLE access_entry_decision_history
+    ALTER COLUMN organization_id SET NOT NULL;