diff --git a/pkg/coredata/access_entry.go b/pkg/coredata/access_entry.go new file mode 100644 index 000000000..d2213167a --- /dev/null +++ b/pkg/coredata/access_entry.go @@ -0,0 +1,803 @@ +// Copyright (c) 2026 Probo Inc . +// +// Permission to use, copy, modify, and/or distribute this software for any +// purpose with or without fee is hereby granted, provided that the above +// copyright notice and this permission notice appear in all copies. +// +// THE SOFTWARE IS PROVIDED "AS IS" AND THE AUTHOR DISCLAIMS ALL WARRANTIES WITH +// REGARD TO THIS SOFTWARE INCLUDING ALL IMPLIED WARRANTIES OF MERCHANTABILITY +// AND FITNESS. IN NO EVENT SHALL THE AUTHOR BE LIABLE FOR ANY SPECIAL, DIRECT, +// INDIRECT, OR CONSEQUENTIAL DAMAGES OR ANY DAMAGES WHATSOEVER RESULTING FROM +// LOSS OF USE, DATA OR PROFITS, WHETHER IN AN ACTION OF CONTRACT, NEGLIGENCE OR +// OTHER TORTIOUS ACTION, ARISING OUT OF OR IN CONNECTION WITH THE USE OR +// PERFORMANCE OF THIS SOFTWARE. + +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 +} diff --git a/pkg/coredata/access_entry_account_type.go b/pkg/coredata/access_entry_account_type.go new file mode 100644 index 000000000..599820d78 --- /dev/null +++ b/pkg/coredata/access_entry_account_type.go @@ -0,0 +1,64 @@ +// Copyright (c) 2026 Probo Inc . +// +// Permission to use, copy, modify, and/or distribute this software for any +// purpose with or without fee is hereby granted, provided that the above +// copyright notice and this permission notice appear in all copies. +// +// THE SOFTWARE IS PROVIDED "AS IS" AND THE AUTHOR DISCLAIMS ALL WARRANTIES WITH +// REGARD TO THIS SOFTWARE INCLUDING ALL IMPLIED WARRANTIES OF MERCHANTABILITY +// AND FITNESS. IN NO EVENT SHALL THE AUTHOR BE LIABLE FOR ANY SPECIAL, DIRECT, +// INDIRECT, OR CONSEQUENTIAL DAMAGES OR ANY DAMAGES WHATSOEVER RESULTING FROM +// LOSS OF USE, DATA OR PROFITS, WHETHER IN AN ACTION OF CONTRACT, NEGLIGENCE OR +// OTHER TORTIOUS ACTION, ARISING OUT OF OR IN CONNECTION WITH THE USE OR +// PERFORMANCE OF THIS SOFTWARE. + +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 +} diff --git a/pkg/coredata/access_entry_decision.go b/pkg/coredata/access_entry_decision.go new file mode 100644 index 000000000..e1a2de7fe --- /dev/null +++ b/pkg/coredata/access_entry_decision.go @@ -0,0 +1,66 @@ +// Copyright (c) 2025-2026 Probo Inc . +// +// Permission to use, copy, modify, and/or distribute this software for any +// purpose with or without fee is hereby granted, provided that the above +// copyright notice and this permission notice appear in all copies. +// +// THE SOFTWARE IS PROVIDED "AS IS" AND THE AUTHOR DISCLAIMS ALL WARRANTIES WITH +// REGARD TO THIS SOFTWARE INCLUDING ALL IMPLIED WARRANTIES OF MERCHANTABILITY +// AND FITNESS. IN NO EVENT SHALL THE AUTHOR BE LIABLE FOR ANY SPECIAL, DIRECT, +// INDIRECT, OR CONSEQUENTIAL DAMAGES OR ANY DAMAGES WHATSOEVER RESULTING FROM +// LOSS OF USE, DATA OR PROFITS, WHETHER IN AN ACTION OF CONTRACT, NEGLIGENCE OR +// OTHER TORTIOUS ACTION, ARISING OUT OF OR IN CONNECTION WITH THE USE OR +// PERFORMANCE OF THIS SOFTWARE. + +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 +} diff --git a/pkg/coredata/access_entry_decision_history.go b/pkg/coredata/access_entry_decision_history.go new file mode 100644 index 000000000..528202462 --- /dev/null +++ b/pkg/coredata/access_entry_decision_history.go @@ -0,0 +1,150 @@ +// Copyright (c) 2026 Probo Inc . +// +// Permission to use, copy, modify, and/or distribute this software for any +// purpose with or without fee is hereby granted, provided that the above +// copyright notice and this permission notice appear in all copies. +// +// THE SOFTWARE IS PROVIDED "AS IS" AND THE AUTHOR DISCLAIMS ALL WARRANTIES WITH +// REGARD TO THIS SOFTWARE INCLUDING ALL IMPLIED WARRANTIES OF MERCHANTABILITY +// AND FITNESS. IN NO EVENT SHALL THE AUTHOR BE LIABLE FOR ANY SPECIAL, DIRECT, +// INDIRECT, OR CONSEQUENTIAL DAMAGES OR ANY DAMAGES WHATSOEVER RESULTING FROM +// LOSS OF USE, DATA OR PROFITS, WHETHER IN AN ACTION OF CONTRACT, NEGLIGENCE OR +// OTHER TORTIOUS ACTION, ARISING OUT OF OR IN CONNECTION WITH THE USE OR +// PERFORMANCE OF THIS SOFTWARE. + +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 +} diff --git a/pkg/coredata/access_entry_filter.go b/pkg/coredata/access_entry_filter.go new file mode 100644 index 000000000..c6c0d87a8 --- /dev/null +++ b/pkg/coredata/access_entry_filter.go @@ -0,0 +1,109 @@ +// Copyright (c) 2026 Probo Inc . +// +// Permission to use, copy, modify, and/or distribute this software for any +// purpose with or without fee is hereby granted, provided that the above +// copyright notice and this permission notice appear in all copies. +// +// THE SOFTWARE IS PROVIDED "AS IS" AND THE AUTHOR DISCLAIMS ALL WARRANTIES WITH +// REGARD TO THIS SOFTWARE INCLUDING ALL IMPLIED WARRANTIES OF MERCHANTABILITY +// AND FITNESS. IN NO EVENT SHALL THE AUTHOR BE LIABLE FOR ANY SPECIAL, DIRECT, +// INDIRECT, OR CONSEQUENTIAL DAMAGES OR ANY DAMAGES WHATSOEVER RESULTING FROM +// LOSS OF USE, DATA OR PROFITS, WHETHER IN AN ACTION OF CONTRACT, NEGLIGENCE OR +// OTHER TORTIOUS ACTION, ARISING OUT OF OR IN CONNECTION WITH THE USE OR +// PERFORMANCE OF THIS SOFTWARE. + +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 +} diff --git a/pkg/coredata/access_entry_flag.go b/pkg/coredata/access_entry_flag.go new file mode 100644 index 000000000..4ac41990f --- /dev/null +++ b/pkg/coredata/access_entry_flag.go @@ -0,0 +1,96 @@ +// Copyright (c) 2026 Probo Inc . +// +// Permission to use, copy, modify, and/or distribute this software for any +// purpose with or without fee is hereby granted, provided that the above +// copyright notice and this permission notice appear in all copies. +// +// THE SOFTWARE IS PROVIDED "AS IS" AND THE AUTHOR DISCLAIMS ALL WARRANTIES WITH +// REGARD TO THIS SOFTWARE INCLUDING ALL IMPLIED WARRANTIES OF MERCHANTABILITY +// AND FITNESS. IN NO EVENT SHALL THE AUTHOR BE LIABLE FOR ANY SPECIAL, DIRECT, +// INDIRECT, OR CONSEQUENTIAL DAMAGES OR ANY DAMAGES WHATSOEVER RESULTING FROM +// LOSS OF USE, DATA OR PROFITS, WHETHER IN AN ACTION OF CONTRACT, NEGLIGENCE OR +// OTHER TORTIOUS ACTION, ARISING OUT OF OR IN CONNECTION WITH THE USE OR +// PERFORMANCE OF THIS SOFTWARE. + +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 +} diff --git a/pkg/coredata/access_entry_incremental_tag.go b/pkg/coredata/access_entry_incremental_tag.go new file mode 100644 index 000000000..dddfc54d1 --- /dev/null +++ b/pkg/coredata/access_entry_incremental_tag.go @@ -0,0 +1,61 @@ +// Copyright (c) 2026 Probo Inc . +// +// Permission to use, copy, modify, and/or distribute this software for any +// purpose with or without fee is hereby granted, provided that the above +// copyright notice and this permission notice appear in all copies. +// +// THE SOFTWARE IS PROVIDED "AS IS" AND THE AUTHOR DISCLAIMS ALL WARRANTIES WITH +// REGARD TO THIS SOFTWARE INCLUDING ALL IMPLIED WARRANTIES OF MERCHANTABILITY +// AND FITNESS. IN NO EVENT SHALL THE AUTHOR BE LIABLE FOR ANY SPECIAL, DIRECT, +// INDIRECT, OR CONSEQUENTIAL DAMAGES OR ANY DAMAGES WHATSOEVER RESULTING FROM +// LOSS OF USE, DATA OR PROFITS, WHETHER IN AN ACTION OF CONTRACT, NEGLIGENCE OR +// OTHER TORTIOUS ACTION, ARISING OUT OF OR IN CONNECTION WITH THE USE OR +// PERFORMANCE OF THIS SOFTWARE. + +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 +} diff --git a/pkg/coredata/access_entry_order_field.go b/pkg/coredata/access_entry_order_field.go new file mode 100644 index 000000000..1e6b40903 --- /dev/null +++ b/pkg/coredata/access_entry_order_field.go @@ -0,0 +1,57 @@ +// Copyright (c) 2026 Probo Inc . +// +// Permission to use, copy, modify, and/or distribute this software for any +// purpose with or without fee is hereby granted, provided that the above +// copyright notice and this permission notice appear in all copies. +// +// THE SOFTWARE IS PROVIDED "AS IS" AND THE AUTHOR DISCLAIMS ALL WARRANTIES WITH +// REGARD TO THIS SOFTWARE INCLUDING ALL IMPLIED WARRANTIES OF MERCHANTABILITY +// AND FITNESS. IN NO EVENT SHALL THE AUTHOR BE LIABLE FOR ANY SPECIAL, DIRECT, +// INDIRECT, OR CONSEQUENTIAL DAMAGES OR ANY DAMAGES WHATSOEVER RESULTING FROM +// LOSS OF USE, DATA OR PROFITS, WHETHER IN AN ACTION OF CONTRACT, NEGLIGENCE OR +// OTHER TORTIOUS ACTION, ARISING OUT OF OR IN CONNECTION WITH THE USE OR +// PERFORMANCE OF THIS SOFTWARE. + +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 +} diff --git a/pkg/coredata/access_entry_statistics.go b/pkg/coredata/access_entry_statistics.go new file mode 100644 index 000000000..5f3075b61 --- /dev/null +++ b/pkg/coredata/access_entry_statistics.go @@ -0,0 +1,243 @@ +// Copyright (c) 2026 Probo Inc . +// +// Permission to use, copy, modify, and/or distribute this software for any +// purpose with or without fee is hereby granted, provided that the above +// copyright notice and this permission notice appear in all copies. +// +// THE SOFTWARE IS PROVIDED "AS IS" AND THE AUTHOR DISCLAIMS ALL WARRANTIES WITH +// REGARD TO THIS SOFTWARE INCLUDING ALL IMPLIED WARRANTIES OF MERCHANTABILITY +// AND FITNESS. IN NO EVENT SHALL THE AUTHOR BE LIABLE FOR ANY SPECIAL, DIRECT, +// INDIRECT, OR CONSEQUENTIAL DAMAGES OR ANY DAMAGES WHATSOEVER RESULTING FROM +// LOSS OF USE, DATA OR PROFITS, WHETHER IN AN ACTION OF CONTRACT, NEGLIGENCE OR +// OTHER TORTIOUS ACTION, ARISING OUT OF OR IN CONNECTION WITH THE USE OR +// PERFORMANCE OF THIS SOFTWARE. + +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 +} diff --git a/pkg/coredata/access_review_campaign.go b/pkg/coredata/access_review_campaign.go new file mode 100644 index 000000000..0d8db056e --- /dev/null +++ b/pkg/coredata/access_review_campaign.go @@ -0,0 +1,366 @@ +// Copyright (c) 2026 Probo Inc . +// +// Permission to use, copy, modify, and/or distribute this software for any +// purpose with or without fee is hereby granted, provided that the above +// copyright notice and this permission notice appear in all copies. +// +// THE SOFTWARE IS PROVIDED "AS IS" AND THE AUTHOR DISCLAIMS ALL WARRANTIES WITH +// REGARD TO THIS SOFTWARE INCLUDING ALL IMPLIED WARRANTIES OF MERCHANTABILITY +// AND FITNESS. IN NO EVENT SHALL THE AUTHOR BE LIABLE FOR ANY SPECIAL, DIRECT, +// INDIRECT, OR CONSEQUENTIAL DAMAGES OR ANY DAMAGES WHATSOEVER RESULTING FROM +// LOSS OF USE, DATA OR PROFITS, WHETHER IN AN ACTION OF CONTRACT, NEGLIGENCE OR +// OTHER TORTIOUS ACTION, ARISING OUT OF OR IN CONNECTION WITH THE USE OR +// PERFORMANCE OF THIS SOFTWARE. + +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 +} diff --git a/pkg/coredata/access_review_campaign_order_field.go b/pkg/coredata/access_review_campaign_order_field.go new file mode 100644 index 000000000..8a8e296e6 --- /dev/null +++ b/pkg/coredata/access_review_campaign_order_field.go @@ -0,0 +1,57 @@ +// Copyright (c) 2026 Probo Inc . +// +// Permission to use, copy, modify, and/or distribute this software for any +// purpose with or without fee is hereby granted, provided that the above +// copyright notice and this permission notice appear in all copies. +// +// THE SOFTWARE IS PROVIDED "AS IS" AND THE AUTHOR DISCLAIMS ALL WARRANTIES WITH +// REGARD TO THIS SOFTWARE INCLUDING ALL IMPLIED WARRANTIES OF MERCHANTABILITY +// AND FITNESS. IN NO EVENT SHALL THE AUTHOR BE LIABLE FOR ANY SPECIAL, DIRECT, +// INDIRECT, OR CONSEQUENTIAL DAMAGES OR ANY DAMAGES WHATSOEVER RESULTING FROM +// LOSS OF USE, DATA OR PROFITS, WHETHER IN AN ACTION OF CONTRACT, NEGLIGENCE OR +// OTHER TORTIOUS ACTION, ARISING OUT OF OR IN CONNECTION WITH THE USE OR +// PERFORMANCE OF THIS SOFTWARE. + +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 +} diff --git a/pkg/coredata/access_review_campaign_scope_system.go b/pkg/coredata/access_review_campaign_scope_system.go new file mode 100644 index 000000000..bfe1c4913 --- /dev/null +++ b/pkg/coredata/access_review_campaign_scope_system.go @@ -0,0 +1,214 @@ +// Copyright (c) 2026 Probo Inc . +// +// Permission to use, copy, modify, and/or distribute this software for any +// purpose with or without fee is hereby granted, provided that the above +// copyright notice and this permission notice appear in all copies. +// +// THE SOFTWARE IS PROVIDED "AS IS" AND THE AUTHOR DISCLAIMS ALL WARRANTIES WITH +// REGARD TO THIS SOFTWARE INCLUDING ALL IMPLIED WARRANTIES OF MERCHANTABILITY +// AND FITNESS. IN NO EVENT SHALL THE AUTHOR BE LIABLE FOR ANY SPECIAL, DIRECT, +// INDIRECT, OR CONSEQUENTIAL DAMAGES OR ANY DAMAGES WHATSOEVER RESULTING FROM +// LOSS OF USE, DATA OR PROFITS, WHETHER IN AN ACTION OF CONTRACT, NEGLIGENCE OR +// OTHER TORTIOUS ACTION, ARISING OUT OF OR IN CONNECTION WITH THE USE OR +// PERFORMANCE OF THIS SOFTWARE. + +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 +} diff --git a/pkg/coredata/access_review_campaign_source_fetch.go b/pkg/coredata/access_review_campaign_source_fetch.go new file mode 100644 index 000000000..8b929517e --- /dev/null +++ b/pkg/coredata/access_review_campaign_source_fetch.go @@ -0,0 +1,302 @@ +// Copyright (c) 2026 Probo Inc . +// +// Permission to use, copy, modify, and/or distribute this software for any +// purpose with or without fee is hereby granted, provided that the above +// copyright notice and this permission notice appear in all copies. +// +// THE SOFTWARE IS PROVIDED "AS IS" AND THE AUTHOR DISCLAIMS ALL WARRANTIES WITH +// REGARD TO THIS SOFTWARE INCLUDING ALL IMPLIED WARRANTIES OF MERCHANTABILITY +// AND FITNESS. IN NO EVENT SHALL THE AUTHOR BE LIABLE FOR ANY SPECIAL, DIRECT, +// INDIRECT, OR CONSEQUENTIAL DAMAGES OR ANY DAMAGES WHATSOEVER RESULTING FROM +// LOSS OF USE, DATA OR PROFITS, WHETHER IN AN ACTION OF CONTRACT, NEGLIGENCE OR +// OTHER TORTIOUS ACTION, ARISING OUT OF OR IN CONNECTION WITH THE USE OR +// PERFORMANCE OF THIS SOFTWARE. + +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 +} diff --git a/pkg/coredata/access_review_campaign_source_fetch_status.go b/pkg/coredata/access_review_campaign_source_fetch_status.go new file mode 100644 index 000000000..cd09685f7 --- /dev/null +++ b/pkg/coredata/access_review_campaign_source_fetch_status.go @@ -0,0 +1,68 @@ +// Copyright (c) 2026 Probo Inc . +// +// Permission to use, copy, modify, and/or distribute this software for any +// purpose with or without fee is hereby granted, provided that the above +// copyright notice and this permission notice appear in all copies. +// +// THE SOFTWARE IS PROVIDED "AS IS" AND THE AUTHOR DISCLAIMS ALL WARRANTIES WITH +// REGARD TO THIS SOFTWARE INCLUDING ALL IMPLIED WARRANTIES OF MERCHANTABILITY +// AND FITNESS. IN NO EVENT SHALL THE AUTHOR BE LIABLE FOR ANY SPECIAL, DIRECT, +// INDIRECT, OR CONSEQUENTIAL DAMAGES OR ANY DAMAGES WHATSOEVER RESULTING FROM +// LOSS OF USE, DATA OR PROFITS, WHETHER IN AN ACTION OF CONTRACT, NEGLIGENCE OR +// OTHER TORTIOUS ACTION, ARISING OUT OF OR IN CONNECTION WITH THE USE OR +// PERFORMANCE OF THIS SOFTWARE. + +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 +} diff --git a/pkg/coredata/access_review_campaign_source_fetch_status_test.go b/pkg/coredata/access_review_campaign_source_fetch_status_test.go new file mode 100644 index 000000000..28d9fd147 --- /dev/null +++ b/pkg/coredata/access_review_campaign_source_fetch_status_test.go @@ -0,0 +1,83 @@ +// Copyright (c) 2026 Probo Inc . +// +// Permission to use, copy, modify, and/or distribute this software for any +// purpose with or without fee is hereby granted, provided that the above +// copyright notice and this permission notice appear in all copies. +// +// THE SOFTWARE IS PROVIDED "AS IS" AND THE AUTHOR DISCLAIMS ALL WARRANTIES WITH +// REGARD TO THIS SOFTWARE INCLUDING ALL IMPLIED WARRANTIES OF MERCHANTABILITY +// AND FITNESS. IN NO EVENT SHALL THE AUTHOR BE LIABLE FOR ANY SPECIAL, DIRECT, +// INDIRECT, OR CONSEQUENTIAL DAMAGES OR ANY DAMAGES WHATSOEVER RESULTING FROM +// LOSS OF USE, DATA OR PROFITS, WHETHER IN AN ACTION OF CONTRACT, NEGLIGENCE OR +// OTHER TORTIOUS ACTION, ARISING OUT OF OR IN CONNECTION WITH THE USE OR +// PERFORMANCE OF THIS SOFTWARE. + +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) + } + }) + } +} diff --git a/pkg/coredata/access_review_campaign_status.go b/pkg/coredata/access_review_campaign_status.go new file mode 100644 index 000000000..27eccab50 --- /dev/null +++ b/pkg/coredata/access_review_campaign_status.go @@ -0,0 +1,69 @@ +// Copyright (c) 2026 Probo Inc . +// +// Permission to use, copy, modify, and/or distribute this software for any +// purpose with or without fee is hereby granted, provided that the above +// copyright notice and this permission notice appear in all copies. +// +// THE SOFTWARE IS PROVIDED "AS IS" AND THE AUTHOR DISCLAIMS ALL WARRANTIES WITH +// REGARD TO THIS SOFTWARE INCLUDING ALL IMPLIED WARRANTIES OF MERCHANTABILITY +// AND FITNESS. IN NO EVENT SHALL THE AUTHOR BE LIABLE FOR ANY SPECIAL, DIRECT, +// INDIRECT, OR CONSEQUENTIAL DAMAGES OR ANY DAMAGES WHATSOEVER RESULTING FROM +// LOSS OF USE, DATA OR PROFITS, WHETHER IN AN ACTION OF CONTRACT, NEGLIGENCE OR +// OTHER TORTIOUS ACTION, ARISING OUT OF OR IN CONNECTION WITH THE USE OR +// PERFORMANCE OF THIS SOFTWARE. + +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 +} diff --git a/pkg/coredata/access_source.go b/pkg/coredata/access_source.go new file mode 100644 index 000000000..26f946c18 --- /dev/null +++ b/pkg/coredata/access_source.go @@ -0,0 +1,408 @@ +// Copyright (c) 2025-2026 Probo Inc . +// +// Permission to use, copy, modify, and/or distribute this software for any +// purpose with or without fee is hereby granted, provided that the above +// copyright notice and this permission notice appear in all copies. +// +// THE SOFTWARE IS PROVIDED "AS IS" AND THE AUTHOR DISCLAIMS ALL WARRANTIES WITH +// REGARD TO THIS SOFTWARE INCLUDING ALL IMPLIED WARRANTIES OF MERCHANTABILITY +// AND FITNESS. IN NO EVENT SHALL THE AUTHOR BE LIABLE FOR ANY SPECIAL, DIRECT, +// INDIRECT, OR CONSEQUENTIAL DAMAGES OR ANY DAMAGES WHATSOEVER RESULTING FROM +// LOSS OF USE, DATA OR PROFITS, WHETHER IN AN ACTION OF CONTRACT, NEGLIGENCE OR +// OTHER TORTIOUS ACTION, ARISING OUT OF OR IN CONNECTION WITH THE USE OR +// PERFORMANCE OF THIS SOFTWARE. + +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 +} diff --git a/pkg/coredata/access_source_category.go b/pkg/coredata/access_source_category.go new file mode 100644 index 000000000..2a2b248bd --- /dev/null +++ b/pkg/coredata/access_source_category.go @@ -0,0 +1,72 @@ +// Copyright (c) 2026 Probo Inc . +// +// Permission to use, copy, modify, and/or distribute this software for any +// purpose with or without fee is hereby granted, provided that the above +// copyright notice and this permission notice appear in all copies. +// +// THE SOFTWARE IS PROVIDED "AS IS" AND THE AUTHOR DISCLAIMS ALL WARRANTIES WITH +// REGARD TO THIS SOFTWARE INCLUDING ALL IMPLIED WARRANTIES OF MERCHANTABILITY +// AND FITNESS. IN NO EVENT SHALL THE AUTHOR BE LIABLE FOR ANY SPECIAL, DIRECT, +// INDIRECT, OR CONSEQUENTIAL DAMAGES OR ANY DAMAGES WHATSOEVER RESULTING FROM +// LOSS OF USE, DATA OR PROFITS, WHETHER IN AN ACTION OF CONTRACT, NEGLIGENCE OR +// OTHER TORTIOUS ACTION, ARISING OUT OF OR IN CONNECTION WITH THE USE OR +// PERFORMANCE OF THIS SOFTWARE. + +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 +} diff --git a/pkg/coredata/access_source_order_field.go b/pkg/coredata/access_source_order_field.go new file mode 100644 index 000000000..944c414e9 --- /dev/null +++ b/pkg/coredata/access_source_order_field.go @@ -0,0 +1,57 @@ +// Copyright (c) 2025-2026 Probo Inc . +// +// Permission to use, copy, modify, and/or distribute this software for any +// purpose with or without fee is hereby granted, provided that the above +// copyright notice and this permission notice appear in all copies. +// +// THE SOFTWARE IS PROVIDED "AS IS" AND THE AUTHOR DISCLAIMS ALL WARRANTIES WITH +// REGARD TO THIS SOFTWARE INCLUDING ALL IMPLIED WARRANTIES OF MERCHANTABILITY +// AND FITNESS. IN NO EVENT SHALL THE AUTHOR BE LIABLE FOR ANY SPECIAL, DIRECT, +// INDIRECT, OR CONSEQUENTIAL DAMAGES OR ANY DAMAGES WHATSOEVER RESULTING FROM +// LOSS OF USE, DATA OR PROFITS, WHETHER IN AN ACTION OF CONTRACT, NEGLIGENCE OR +// OTHER TORTIOUS ACTION, ARISING OUT OF OR IN CONNECTION WITH THE USE OR +// PERFORMANCE OF THIS SOFTWARE. + +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 +} diff --git a/pkg/coredata/auth_method.go b/pkg/coredata/auth_method.go new file mode 100644 index 000000000..15c276c91 --- /dev/null +++ b/pkg/coredata/auth_method.go @@ -0,0 +1,76 @@ +// Copyright (c) 2026 Probo Inc . +// +// Permission to use, copy, modify, and/or distribute this software for any +// purpose with or without fee is hereby granted, provided that the above +// copyright notice and this permission notice appear in all copies. +// +// THE SOFTWARE IS PROVIDED "AS IS" AND THE AUTHOR DISCLAIMS ALL WARRANTIES WITH +// REGARD TO THIS SOFTWARE INCLUDING ALL IMPLIED WARRANTIES OF MERCHANTABILITY +// AND FITNESS. IN NO EVENT SHALL THE AUTHOR BE LIABLE FOR ANY SPECIAL, DIRECT, +// INDIRECT, OR CONSEQUENTIAL DAMAGES OR ANY DAMAGES WHATSOEVER RESULTING FROM +// LOSS OF USE, DATA OR PROFITS, WHETHER IN AN ACTION OF CONTRACT, NEGLIGENCE OR +// OTHER TORTIOUS ACTION, ARISING OUT OF OR IN CONNECTION WITH THE USE OR +// PERFORMANCE OF THIS SOFTWARE. + +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 +} diff --git a/pkg/coredata/connector.go b/pkg/coredata/connector.go index 3631966d9..4465409a2 100644 --- a/pkg/coredata/connector.go +++ b/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 diff --git a/pkg/coredata/connector_protocol.go b/pkg/coredata/connector_protocol.go index b81a19785..d0f435eea 100644 --- a/pkg/coredata/connector_protocol.go +++ b/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) } diff --git a/pkg/coredata/connector_provider.go b/pkg/coredata/connector_provider.go index 5d41af011..5d085cbbb 100644 --- a/pkg/coredata/connector_provider.go +++ b/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) } diff --git a/pkg/coredata/connector_settings.go b/pkg/coredata/connector_settings.go new file mode 100644 index 000000000..571444c71 --- /dev/null +++ b/pkg/coredata/connector_settings.go @@ -0,0 +1,135 @@ +// Copyright (c) 2026 Probo Inc . +// +// Permission to use, copy, modify, and/or distribute this software for any +// purpose with or without fee is hereby granted, provided that the above +// copyright notice and this permission notice appear in all copies. +// +// THE SOFTWARE IS PROVIDED "AS IS" AND THE AUTHOR DISCLAIMS ALL WARRANTIES WITH +// REGARD TO THIS SOFTWARE INCLUDING ALL IMPLIED WARRANTIES OF MERCHANTABILITY +// AND FITNESS. IN NO EVENT SHALL THE AUTHOR BE LIABLE FOR ANY SPECIAL, DIRECT, +// INDIRECT, OR CONSEQUENTIAL DAMAGES OR ANY DAMAGES WHATSOEVER RESULTING FROM +// LOSS OF USE, DATA OR PROFITS, WHETHER IN AN ACTION OF CONTRACT, NEGLIGENCE OR +// OTHER TORTIOUS ACTION, ARISING OUT OF OR IN CONNECTION WITH THE USE OR +// PERFORMANCE OF THIS SOFTWARE. + +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 +} diff --git a/pkg/coredata/entity_type_reg.go b/pkg/coredata/entity_type_reg.go index f204d1c1f..e83c98428 100644 --- a/pkg/coredata/entity_type_reg.go +++ b/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 } diff --git a/pkg/coredata/mfa_status.go b/pkg/coredata/mfa_status.go new file mode 100644 index 000000000..4e71dc7c0 --- /dev/null +++ b/pkg/coredata/mfa_status.go @@ -0,0 +1,68 @@ +// Copyright (c) 2026 Probo Inc . +// +// Permission to use, copy, modify, and/or distribute this software for any +// purpose with or without fee is hereby granted, provided that the above +// copyright notice and this permission notice appear in all copies. +// +// THE SOFTWARE IS PROVIDED "AS IS" AND THE AUTHOR DISCLAIMS ALL WARRANTIES WITH +// REGARD TO THIS SOFTWARE INCLUDING ALL IMPLIED WARRANTIES OF MERCHANTABILITY +// AND FITNESS. IN NO EVENT SHALL THE AUTHOR BE LIABLE FOR ANY SPECIAL, DIRECT, +// INDIRECT, OR CONSEQUENTIAL DAMAGES OR ANY DAMAGES WHATSOEVER RESULTING FROM +// LOSS OF USE, DATA OR PROFITS, WHETHER IN AN ACTION OF CONTRACT, NEGLIGENCE OR +// OTHER TORTIOUS ACTION, ARISING OUT OF OR IN CONNECTION WITH THE USE OR +// PERFORMANCE OF THIS SOFTWARE. + +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 +} diff --git a/pkg/coredata/migrations/20260314T200000Z.sql b/pkg/coredata/migrations/20260314T200000Z.sql new file mode 100644 index 000000000..26b055117 --- /dev/null +++ b/pkg/coredata/migrations/20260314T200000Z.sql @@ -0,0 +1,171 @@ +-- Copyright (c) 2026 Probo Inc . +-- +-- Permission to use, copy, modify, and/or distribute this software for any +-- purpose with or without fee is hereby granted, provided that the above +-- copyright notice and this permission notice appear in all copies. +-- +-- THE SOFTWARE IS PROVIDED "AS IS" AND THE AUTHOR DISCLAIMS ALL WARRANTIES WITH +-- REGARD TO THIS SOFTWARE INCLUDING ALL IMPLIED WARRANTIES OF MERCHANTABILITY +-- AND FITNESS. IN NO EVENT SHALL THE AUTHOR BE LIABLE FOR ANY SPECIAL, DIRECT, +-- INDIRECT, OR CONSEQUENTIAL DAMAGES OR ANY DAMAGES WHATSOEVER RESULTING FROM +-- LOSS OF USE, DATA OR PROFITS, WHETHER IN AN ACTION OF CONTRACT, NEGLIGENCE OR +-- OTHER TORTIOUS ACTION, ARISING OUT OF OR IN CONNECTION WITH THE USE OR +-- PERFORMANCE OF THIS SOFTWARE. + +-- 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) +); diff --git a/pkg/coredata/migrations/20260324T120000Z.sql b/pkg/coredata/migrations/20260324T120000Z.sql index 17328ba3c..3f8c99257 100644 --- a/pkg/coredata/migrations/20260324T120000Z.sql +++ b/pkg/coredata/migrations/20260324T120000Z.sql @@ -1,19 +1,6 @@ --- Copyright (c) 2026 Probo Inc . --- --- Permission to use, copy, modify, and/or distribute this software for any --- purpose with or without fee is hereby granted, provided that the above --- copyright notice and this permission notice appear in all copies. --- --- THE SOFTWARE IS PROVIDED "AS IS" AND THE AUTHOR DISCLAIMS ALL WARRANTIES WITH --- REGARD TO THIS SOFTWARE INCLUDING ALL IMPLIED WARRANTIES OF MERCHANTABILITY --- AND FITNESS. IN NO EVENT SHALL THE AUTHOR BE LIABLE FOR ANY SPECIAL, DIRECT, --- INDIRECT, OR CONSEQUENTIAL DAMAGES OR ANY DAMAGES WHATSOEVER RESULTING FROM --- LOSS OF USE, DATA OR PROFITS, WHETHER IN AN ACTION OF CONTRACT, NEGLIGENCE OR --- OTHER TORTIOUS ACTION, ARISING OUT OF OR IN CONNECTION WITH THE USE OR --- PERFORMANCE OF THIS SOFTWARE. - 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; + diff --git a/pkg/coredata/migrations/20260324T130000Z.sql b/pkg/coredata/migrations/20260324T130000Z.sql new file mode 100644 index 000000000..4d8daea59 --- /dev/null +++ b/pkg/coredata/migrations/20260324T130000Z.sql @@ -0,0 +1,31 @@ +-- Copyright (c) 2026 Probo Inc . +-- +-- Permission to use, copy, modify, and/or distribute this software for any +-- purpose with or without fee is hereby granted, provided that the above +-- copyright notice and this permission notice appear in all copies. +-- +-- THE SOFTWARE IS PROVIDED "AS IS" AND THE AUTHOR DISCLAIMS ALL WARRANTIES WITH +-- REGARD TO THIS SOFTWARE INCLUDING ALL IMPLIED WARRANTIES OF MERCHANTABILITY +-- AND FITNESS. IN NO EVENT SHALL THE AUTHOR BE LIABLE FOR ANY SPECIAL, DIRECT, +-- INDIRECT, OR CONSEQUENTIAL DAMAGES OR ANY DAMAGES WHATSOEVER RESULTING FROM +-- LOSS OF USE, DATA OR PROFITS, WHETHER IN AN ACTION OF CONTRACT, NEGLIGENCE OR +-- OTHER TORTIOUS ACTION, ARISING OUT OF OR IN CONNECTION WITH THE USE OR +-- PERFORMANCE OF THIS SOFTWARE. + +-- Add 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 +); diff --git a/pkg/coredata/migrations/20260325T130000Z.sql b/pkg/coredata/migrations/20260325T130000Z.sql new file mode 100644 index 000000000..7d82140d9 --- /dev/null +++ b/pkg/coredata/migrations/20260325T130000Z.sql @@ -0,0 +1,16 @@ +-- Copyright (c) 2026 Probo Inc . +-- +-- Permission to use, copy, modify, and/or distribute this software for any +-- purpose with or without fee is hereby granted, provided that the above +-- copyright notice and this permission notice appear in all copies. +-- +-- THE SOFTWARE IS PROVIDED "AS IS" AND THE AUTHOR DISCLAIMS ALL WARRANTIES WITH +-- REGARD TO THIS SOFTWARE INCLUDING ALL IMPLIED WARRANTIES OF MERCHANTABILITY +-- AND FITNESS. IN NO EVENT SHALL THE AUTHOR BE LIABLE FOR ANY SPECIAL, DIRECT, +-- INDIRECT, OR CONSEQUENTIAL DAMAGES OR ANY DAMAGES WHATSOEVER RESULTING FROM +-- LOSS OF USE, DATA OR PROFITS, WHETHER IN AN ACTION OF CONTRACT, NEGLIGENCE OR +-- OTHER TORTIOUS ACTION, ARISING OUT OF OR IN CONNECTION WITH THE USE OR +-- PERFORMANCE OF THIS SOFTWARE. + +ALTER TABLE access_sources + ADD COLUMN name_synced_at TIMESTAMP WITH TIME ZONE; diff --git a/pkg/coredata/migrations/20260325T140000Z.sql b/pkg/coredata/migrations/20260325T140000Z.sql new file mode 100644 index 000000000..512c1e1e6 --- /dev/null +++ b/pkg/coredata/migrations/20260325T140000Z.sql @@ -0,0 +1,16 @@ +-- Copyright (c) 2026 Probo Inc . +-- +-- Permission to use, copy, modify, and/or distribute this software for any +-- purpose with or without fee is hereby granted, provided that the above +-- copyright notice and this permission notice appear in all copies. +-- +-- THE SOFTWARE IS PROVIDED "AS IS" AND THE AUTHOR DISCLAIMS ALL WARRANTIES WITH +-- REGARD TO THIS SOFTWARE INCLUDING ALL IMPLIED WARRANTIES OF MERCHANTABILITY +-- AND FITNESS. IN NO EVENT SHALL THE AUTHOR BE LIABLE FOR ANY SPECIAL, DIRECT, +-- INDIRECT, OR CONSEQUENTIAL DAMAGES OR ANY DAMAGES WHATSOEVER RESULTING FROM +-- LOSS OF USE, DATA OR PROFITS, WHETHER IN AN ACTION OF CONTRACT, NEGLIGENCE OR +-- OTHER TORTIOUS ACTION, ARISING OUT OF OR IN CONNECTION WITH THE USE OR +-- PERFORMANCE OF THIS SOFTWARE. + +ALTER TABLE access_entries + ADD COLUMN account_type TEXT NOT NULL DEFAULT 'USER'; diff --git a/pkg/coredata/migrations/20260327T130000Z.sql b/pkg/coredata/migrations/20260327T130000Z.sql new file mode 100644 index 000000000..a835d7e55 --- /dev/null +++ b/pkg/coredata/migrations/20260327T130000Z.sql @@ -0,0 +1,20 @@ +-- Copyright (c) 2026 Probo Inc . +-- +-- Permission to use, copy, modify, and/or distribute this software for any +-- purpose with or without fee is hereby granted, provided that the above +-- copyright notice and this permission notice appear in all copies. +-- +-- THE SOFTWARE IS PROVIDED "AS IS" AND THE AUTHOR DISCLAIMS ALL WARRANTIES WITH +-- REGARD TO THIS SOFTWARE INCLUDING ALL IMPLIED WARRANTIES OF MERCHANTABILITY +-- AND FITNESS. IN NO EVENT SHALL THE AUTHOR BE LIABLE FOR ANY SPECIAL, DIRECT, +-- INDIRECT, OR CONSEQUENTIAL DAMAGES OR ANY DAMAGES WHATSOEVER RESULTING FROM +-- LOSS OF USE, DATA OR PROFITS, WHETHER IN AN ACTION OF CONTRACT, NEGLIGENCE OR +-- OTHER TORTIOUS ACTION, ARISING OUT OF OR IN CONNECTION WITH THE USE OR +-- PERFORMANCE OF THIS SOFTWARE. + +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'; diff --git a/pkg/coredata/migrations/20260330T100000Z.sql b/pkg/coredata/migrations/20260330T100000Z.sql new file mode 100644 index 000000000..f935d6138 --- /dev/null +++ b/pkg/coredata/migrations/20260330T100000Z.sql @@ -0,0 +1,36 @@ +-- Copyright (c) 2026 Probo Inc . +-- +-- Permission to use, copy, modify, and/or distribute this software for any +-- purpose with or without fee is hereby granted, provided that the above +-- copyright notice and this permission notice appear in all copies. +-- +-- THE SOFTWARE IS PROVIDED "AS IS" AND THE AUTHOR DISCLAIMS ALL WARRANTIES WITH +-- REGARD TO THIS SOFTWARE INCLUDING ALL IMPLIED WARRANTIES OF MERCHANTABILITY +-- AND FITNESS. IN NO EVENT SHALL THE AUTHOR BE LIABLE FOR ANY SPECIAL, DIRECT, +-- INDIRECT, OR CONSEQUENTIAL DAMAGES OR ANY DAMAGES WHATSOEVER RESULTING FROM +-- LOSS OF USE, DATA OR PROFITS, WHETHER IN AN ACTION OF CONTRACT, NEGLIGENCE OR +-- OTHER TORTIOUS ACTION, ARISING OUT OF OR IN CONNECTION WITH THE USE OR +-- PERFORMANCE OF THIS SOFTWARE. + +-- 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; diff --git a/pkg/coredata/migrations/20260330T120000Z.sql b/pkg/coredata/migrations/20260330T120000Z.sql index 7a53ba206..2e04a80e2 100644 --- a/pkg/coredata/migrations/20260330T120000Z.sql +++ b/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 . +-- +-- 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;