rohit/probo
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity
Files
main
probo/pkg/coredata/membership_profile.go
Bryan Frimin 36e038f5ff Join identities in MembershipProfiles.LoadExistingByIDs 
SCIM event CSV exports failed with "cannot parse address :
mail: no address" because this query selected a placeholder
empty string for email_address instead of the real address, and
scanning it into mail.Addr tried to parse it as one.

Join identities the same way every other MembershipProfile query
already does, and drop the redundant identity lookup in the SCIM
export path now that profiles carry their real email address.

Signed-off-by: Bryan Frimin <bryan@probo.com>
2026-07-31 09:39:22 +02:00

1414 lines
33 KiB
Go
Raw Permalink Blame History

// Copyright (c) 2025-2026 Probo Inc <hello@probo.com>.
//
// Permission is hereby granted, free of charge, to any person obtaining a copy
// of this software and associated documentation files (the "Software"), to deal
// in the Software without restriction, including without limitation the rights
// to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
// copies of the Software, and to permit persons to whom the Software is
// furnished to do so, subject to the following conditions:
//
// The above copyright notice and this permission notice shall be included in
// all copies or substantial portions of the Software.
//
// THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
// IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
// FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
// AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
// LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
// OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE
// SOFTWARE.
package coredata
import (
"context"
"errors"
"fmt"
"maps"
"time"
"github.com/jackc/pgx/v5"
"github.com/jackc/pgx/v5/pgconn"
"go.gearno.de/kit/pg"
"go.probo.inc/probo/pkg/gid"
"go.probo.inc/probo/pkg/iam/policy"
"go.probo.inc/probo/pkg/mail"
"go.probo.inc/probo/pkg/page"
)
type (
MembershipProfile struct {
ID gid.GID `db:"id"`
IdentityID gid.GID `db:"identity_id"`
OrganizationID gid.GID `db:"organization_id"`
EmailAddress mail.Addr `db:"email_address"`
Source ProfileSource `db:"source"`
State ProfileState `db:"state"`
FullName string `db:"full_name"`
Kind *string `db:"kind"`
AdditionalEmailAddresses mail.Addrs `db:"additional_email_addresses"`
Position *string `db:"position"`
ContractStartDate *time.Time `db:"contract_start_date"`
ContractEndDate *time.Time `db:"contract_end_date"`
OrganizationName string `db:"organization_name"`
UserName *string `db:"user_name"`
ExternalID *string `db:"external_id"`
Nickname *string `db:"nickname"`
Locale *string `db:"locale"`
Timezone *string `db:"timezone"`
ProfileUrl *string `db:"profile_url"`
PreferredLanguage *string `db:"preferred_language"`
GivenName *string `db:"given_name"`
FamilyName *string `db:"family_name"`
FormattedName *string `db:"formatted_name"`
MiddleName *string `db:"middle_name"`
HonorificPrefix *string `db:"honorific_prefix"`
HonorificSuffix *string `db:"honorific_suffix"`
EmployeeNumber *string `db:"employee_number"`
Department *string `db:"department"`
CostCenter *string `db:"cost_center"`
EnterpriseOrganization *string `db:"enterprise_organization"`
Division *string `db:"division"`
ManagerValue *string `db:"manager_value"`
ActivatedAt *time.Time `db:"activated_at"`
DeactivatedAt *time.Time `db:"deactivated_at"`
CreatedAt time.Time `db:"created_at"`
UpdatedAt time.Time `db:"updated_at"`
}
MembershipProfiles []*MembershipProfile
)
func (p MembershipProfile) CursorKey(orderBy MembershipProfileOrderField) page.CursorKey {
switch orderBy {
case MembershipProfileOrderFieldCreatedAt:
return page.NewCursorKey(p.ID, p.CreatedAt)
case MembershipProfileOrderFieldFullName:
return page.NewCursorKey(p.ID, p.FullName)
case MembershipProfileOrderFieldEmailAddress:
return page.NewCursorKey(p.ID, p.EmailAddress)
case MembershipProfileOrderFieldKind:
return page.NewCursorKey(p.ID, p.Kind)
case MembershipProfileOrderFieldOrganizationName:
return page.NewCursorKey(p.ID, p.OrganizationName)
case MembershipProfileOrderFieldState:
return page.NewCursorKey(p.ID, p.State)
}
panic(fmt.Sprintf("unsupported order by: %s", orderBy))
}
func (p *MembershipProfile) MarkPending(now time.Time) {
p.State = ProfileStatePending
p.ActivatedAt = nil
p.DeactivatedAt = nil
p.UpdatedAt = now
}
func (p *MembershipProfile) MarkActive(now time.Time) {
p.State = ProfileStateActive
p.ActivatedAt = &now
p.DeactivatedAt = nil
p.UpdatedAt = now
}
func (p *MembershipProfile) MarkDeactivated(now time.Time) {
p.State = ProfileStateDeactivated
p.DeactivatedAt = &now
p.UpdatedAt = now
}
func (p *MembershipProfile) AuthorizationAttributes(
ctx context.Context,
conn pg.Querier,
resourceIDs []gid.GID,
) (policy.AttributesByID, error) {
q := `
SELECT
id,
organization_id,
identity_id,
source
FROM
iam_membership_profiles
WHERE
id = ANY(@resource_ids::text[])
`
args := pgx.StrictNamedArgs{
"resource_ids": resourceIDs,
}
rows, err := conn.Query(ctx, q, args)
if err != nil {
return nil, fmt.Errorf("cannot query profile authorization attributes: %w", err)
}
defer rows.Close()
attrsByID := make(policy.AttributesByID, len(resourceIDs))
for rows.Next() {
var (
id gid.GID
organizationID gid.GID
identityID gid.GID
source ProfileSource
)
err = rows.Scan(&id, &organizationID, &identityID, &source)
if err != nil {
return nil, fmt.Errorf("cannot scan profile authorization attributes: %w", err)
}
attrsByID[id] = policy.Attributes{
"organization_id": organizationID.String(),
"identity_id": identityID.String(),
"source": source.String(),
}
}
if err = rows.Err(); err != nil {
return nil, fmt.Errorf("cannot iterate profile authorization attributes: %w", err)
}
return attrsByID, nil
}
func (p *MembershipProfile) LoadByID(
ctx context.Context,
conn pg.Querier,
scope Scoper,
profileID gid.GID,
) error {
q := `
SELECT
p.id,
p.identity_id,
p.organization_id,
i.email_address,
p.source,
p.state,
p.full_name,
p.kind,
p.additional_email_addresses,
p.position,
p.contract_start_date,
p.contract_end_date,
'' AS organization_name,
p.user_name,
p.external_id,
p.nickname,
p.locale,
p.timezone,
p.profile_url,
p.preferred_language,
p.given_name,
p.family_name,
p.formatted_name,
p.middle_name,
p.honorific_prefix,
p.honorific_suffix,
p.employee_number,
p.department,
p.cost_center,
p.enterprise_organization,
p.division,
p.manager_value,
p.activated_at,
p.deactivated_at,
p.created_at,
p.updated_at
FROM
iam_membership_profiles p
INNER JOIN identities i
ON i.id = p.identity_id
WHERE
p.%s
AND p.id = @profile_id
LIMIT 1;
`
q = fmt.Sprintf(q, scope.SQLFragment())
args := pgx.StrictNamedArgs{"profile_id": profileID}
maps.Copy(args, scope.SQLArguments())
rows, err := conn.Query(ctx, q, args)
if err != nil {
return fmt.Errorf("cannot query profile: %w", err)
}
profile, err := pgx.CollectExactlyOneRow(rows, pgx.RowToStructByName[MembershipProfile])
if err != nil {
if errors.Is(err, pgx.ErrNoRows) {
return ErrResourceNotFound
}
return fmt.Errorf("cannot collect profile: %w", err)
}
*p = profile
return nil
}
func (p *MembershipProfile) LoadByIdentityIDAndOrganizationID(
ctx context.Context,
conn pg.Querier,
scope Scoper,
identityID gid.GID,
organizationID gid.GID,
) error {
q := `
SELECT
p.id,
p.identity_id,
p.organization_id,
i.email_address,
p.source,
p.state,
p.full_name,
p.kind,
p.additional_email_addresses,
p.position,
p.contract_start_date,
p.contract_end_date,
'' AS organization_name,
p.user_name,
p.external_id,
p.nickname,
p.locale,
p.timezone,
p.profile_url,
p.preferred_language,
p.given_name,
p.family_name,
p.formatted_name,
p.middle_name,
p.honorific_prefix,
p.honorific_suffix,
p.employee_number,
p.department,
p.cost_center,
p.enterprise_organization,
p.division,
p.manager_value,
p.activated_at,
p.deactivated_at,
p.created_at,
p.updated_at
FROM
iam_membership_profiles p
INNER JOIN identities i
ON i.id = p.identity_id
WHERE
p.%s
AND p.identity_id = @identity_id
AND p.organization_id = @organization_id
LIMIT 1;
`
q = fmt.Sprintf(q, scope.SQLFragment())
args := pgx.StrictNamedArgs{
"identity_id": identityID,
"organization_id": organizationID,
}
maps.Copy(args, scope.SQLArguments())
rows, err := conn.Query(ctx, q, args)
if err != nil {
return fmt.Errorf("cannot query profile: %w", err)
}
profile, err := pgx.CollectExactlyOneRow(rows, pgx.RowToStructByName[MembershipProfile])
if err != nil {
if errors.Is(err, pgx.ErrNoRows) {
return ErrResourceNotFound
}
return fmt.Errorf("cannot collect profile: %w", err)
}
*p = profile
return nil
}
func (p *MembershipProfile) LoadByExternalIDAndOrganizationID(
ctx context.Context,
conn pg.Querier,
scope Scoper,
externalID string,
organizationID gid.GID,
) error {
q := `
SELECT
p.id,
p.identity_id,
p.organization_id,
i.email_address,
p.source,
p.state,
p.full_name,
p.kind,
p.additional_email_addresses,
p.position,
p.contract_start_date,
p.contract_end_date,
'' AS organization_name,
p.user_name,
p.external_id,
p.nickname,
p.locale,
p.timezone,
p.profile_url,
p.preferred_language,
p.given_name,
p.family_name,
p.formatted_name,
p.middle_name,
p.honorific_prefix,
p.honorific_suffix,
p.employee_number,
p.department,
p.cost_center,
p.enterprise_organization,
p.division,
p.manager_value,
p.activated_at,
p.deactivated_at,
p.created_at,
p.updated_at
FROM
iam_membership_profiles p
INNER JOIN identities i
ON i.id = p.identity_id
WHERE
p.%s
AND p.external_id = @external_id
AND p.organization_id = @organization_id
LIMIT 1;
`
q = fmt.Sprintf(q, scope.SQLFragment())
args := pgx.StrictNamedArgs{
"external_id": externalID,
"organization_id": organizationID,
}
maps.Copy(args, scope.SQLArguments())
rows, err := conn.Query(ctx, q, args)
if err != nil {
return fmt.Errorf("cannot query profile: %w", err)
}
profile, err := pgx.CollectExactlyOneRow(rows, pgx.RowToStructByName[MembershipProfile])
if err != nil {
if errors.Is(err, pgx.ErrNoRows) {
return ErrResourceNotFound
}
return fmt.Errorf("cannot collect profile: %w", err)
}
*p = profile
return nil
}
func (p *MembershipProfiles) LoadByIDs(
ctx context.Context,
conn pg.Querier,
scope Scoper,
profileIDs []gid.GID,
) error {
q := `
SELECT
p.id,
p.identity_id,
p.organization_id,
i.email_address,
p.source,
p.state,
p.full_name,
p.kind,
p.additional_email_addresses,
p.position,
p.contract_start_date,
p.contract_end_date,
'' AS organization_name,
p.user_name,
p.external_id,
p.nickname,
p.locale,
p.timezone,
p.profile_url,
p.preferred_language,
p.given_name,
p.family_name,
p.formatted_name,
p.middle_name,
p.honorific_prefix,
p.honorific_suffix,
p.employee_number,
p.department,
p.cost_center,
p.enterprise_organization,
p.division,
p.manager_value,
p.activated_at,
p.deactivated_at,
p.created_at,
p.updated_at
FROM
iam_membership_profiles p
INNER JOIN identities i
ON i.id = p.identity_id
WHERE
p.%s
AND p.id = ANY(@profile_ids)
`
q = fmt.Sprintf(q, scope.SQLFragment())
args := pgx.NamedArgs{"profile_ids": profileIDs}
maps.Copy(args, scope.SQLArguments())
rows, err := conn.Query(ctx, q, args)
if err != nil {
return fmt.Errorf("cannot query profiles: %w", err)
}
profiles, err := pgx.CollectRows(rows, pgx.RowToAddrOfStructByName[MembershipProfile])
if err != nil {
return fmt.Errorf("cannot collect profiles: %w", err)
}
*p = profiles
if len(profiles) != len(gid.NewSet(profileIDs...)) {
return ErrResourceNotFound
}
return nil
}
func (p *MembershipProfiles) LoadExistingByIDs(
ctx context.Context,
conn pg.Querier,
scope Scoper,
profileIDs []gid.GID,
) error {
q := `
SELECT
p.id,
p.identity_id,
p.organization_id,
i.email_address,
p.source,
p.state,
p.full_name,
p.kind,
p.additional_email_addresses,
p.position,
p.contract_start_date,
p.contract_end_date,
'' AS organization_name,
p.user_name,
p.external_id,
p.nickname,
p.locale,
p.timezone,
p.profile_url,
p.preferred_language,
p.given_name,
p.family_name,
p.formatted_name,
p.middle_name,
p.honorific_prefix,
p.honorific_suffix,
p.employee_number,
p.department,
p.cost_center,
p.enterprise_organization,
p.division,
p.manager_value,
p.activated_at,
p.deactivated_at,
p.created_at,
p.updated_at
FROM
iam_membership_profiles p
INNER JOIN identities i
ON i.id = p.identity_id
WHERE
p.%s
AND p.id = ANY(@profile_ids)
`
q = fmt.Sprintf(q, scope.SQLFragment())
args := pgx.NamedArgs{"profile_ids": profileIDs}
maps.Copy(args, scope.SQLArguments())
rows, err := conn.Query(ctx, q, args)
if err != nil {
return fmt.Errorf("cannot query profiles by ids: %w", err)
}
profiles, err := pgx.CollectRows(rows, pgx.RowToAddrOfStructByName[MembershipProfile])
if err != nil {
return fmt.Errorf("cannot collect profiles by ids: %w", err)
}
*p = profiles
return nil
}
func (p *MembershipProfiles) LoadByOrganizationID(
ctx context.Context,
conn pg.Querier,
scope Scoper,
organizationID gid.GID,
cursor *page.Cursor[MembershipProfileOrderField],
filter *MembershipProfileFilter,
) error {
q := `
WITH profiles AS (
SELECT
p.id,
p.identity_id,
p.organization_id,
i.email_address,
p.source,
p.state,
p.full_name,
p.kind,
p.additional_email_addresses,
p.position,
p.contract_start_date,
p.contract_end_date,
p.user_name,
p.external_id,
p.nickname,
p.locale,
p.timezone,
p.profile_url,
p.preferred_language,
p.given_name,
p.family_name,
p.formatted_name,
p.middle_name,
p.honorific_prefix,
p.honorific_suffix,
p.employee_number,
p.department,
p.cost_center,
p.enterprise_organization,
p.division,
p.manager_value,
p.activated_at,
p.deactivated_at,
p.created_at,
p.updated_at
FROM
iam_membership_profiles p
INNER JOIN identities i ON i.id = p.identity_id
WHERE
p.%s
AND p.organization_id = @organization_id
AND %s
)
SELECT
id,
identity_id,
organization_id,
email_address,
source,
state,
full_name,
kind,
additional_email_addresses,
position,
contract_start_date,
contract_end_date,
'' AS organization_name,
user_name,
external_id,
nickname,
locale,
timezone,
profile_url,
preferred_language,
given_name,
family_name,
formatted_name,
middle_name,
honorific_prefix,
honorific_suffix,
employee_number,
department,
cost_center,
enterprise_organization,
division,
manager_value,
activated_at,
deactivated_at,
created_at,
updated_at
FROM profiles
WHERE
%s
`
q = fmt.Sprintf(q, scope.SQLFragment(), filter.SQLFragment(), cursor.SQLFragment())
args := pgx.NamedArgs{"organization_id": organizationID}
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 profiles: %w", err)
}
profiles, err := pgx.CollectRows(rows, pgx.RowToAddrOfStructByName[MembershipProfile])
if err != nil {
return fmt.Errorf("cannot collect profiles: %w", err)
}
*p = profiles
return nil
}
func (p *MembershipProfiles) LoadByIdentityID(
ctx context.Context,
conn pg.Querier,
identityID gid.GID,
cursor *page.Cursor[MembershipProfileOrderField],
filter *MembershipProfileFilter,
) error {
q := `
WITH profiles AS (
SELECT
p.id,
p.identity_id,
p.organization_id,
i.email_address,
p.source,
p.state,
p.full_name,
p.kind,
p.additional_email_addresses,
p.position,
p.contract_start_date,
p.contract_end_date,
p.user_name,
p.external_id,
p.nickname,
p.locale,
p.timezone,
p.profile_url,
p.preferred_language,
p.given_name,
p.family_name,
p.formatted_name,
p.middle_name,
p.honorific_prefix,
p.honorific_suffix,
p.employee_number,
p.department,
p.cost_center,
p.enterprise_organization,
p.division,
p.manager_value,
p.activated_at,
p.deactivated_at,
p.created_at,
p.updated_at
FROM
iam_membership_profiles p
INNER JOIN identities i ON i.id = p.identity_id
WHERE
p.identity_id = @identity_id
AND %s
)
SELECT
p.id,
p.identity_id,
p.organization_id,
p.email_address,
p.source,
p.state,
p.full_name,
p.kind,
p.additional_email_addresses,
p.position,
p.contract_start_date,
p.contract_end_date,
o.name AS organization_name,
p.user_name,
p.external_id,
p.nickname,
p.locale,
p.timezone,
p.profile_url,
p.preferred_language,
p.given_name,
p.family_name,
p.formatted_name,
p.middle_name,
p.honorific_prefix,
p.honorific_suffix,
p.employee_number,
p.department,
p.cost_center,
p.enterprise_organization,
p.division,
p.manager_value,
p.activated_at,
p.deactivated_at,
p.created_at,
p.updated_at
FROM profiles p
INNER JOIN organizations o ON o.id = p.organization_id
WHERE
%s
`
q = fmt.Sprintf(q, filter.SQLFragment(), cursor.SQLFragment())
args := pgx.NamedArgs{"identity_id": identityID}
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 profiles: %w", err)
}
profiles, err := pgx.CollectRows(rows, pgx.RowToAddrOfStructByName[MembershipProfile])
if err != nil {
return fmt.Errorf("cannot collect profiles: %w", err)
}
*p = profiles
return nil
}
func (p *MembershipProfiles) LoadByDocumentVersionID(
ctx context.Context,
conn pg.Querier,
scope Scoper,
documentVersionID gid.GID,
cursor *page.Cursor[MembershipProfileOrderField],
) error {
q := `
WITH latest_quorum AS (
SELECT id
FROM document_version_approval_quorums
WHERE version_id = @version_id
ORDER BY created_at DESC
LIMIT 1
),
version_approvers AS (
SELECT d.approver_id
FROM document_version_approval_decisions d
WHERE d.quorum_id = (SELECT id FROM latest_quorum)
),
profiles AS (
SELECT
mp.id,
mp.identity_id,
mp.organization_id,
mp.full_name,
mp.source,
mp.state,
mp.kind,
mp.additional_email_addresses,
mp.position,
mp.contract_start_date,
mp.contract_end_date,
mp.user_name,
mp.external_id,
mp.nickname,
mp.locale,
mp.timezone,
mp.profile_url,
mp.preferred_language,
mp.given_name,
mp.family_name,
mp.formatted_name,
mp.middle_name,
mp.honorific_prefix,
mp.honorific_suffix,
mp.employee_number,
mp.department,
mp.cost_center,
mp.enterprise_organization,
mp.division,
mp.manager_value,
mp.activated_at,
mp.deactivated_at,
mp.created_at,
mp.updated_at
FROM
iam_membership_profiles mp
INNER JOIN version_approvers va ON va.approver_id = mp.id
WHERE
mp.%s
AND %s
)
SELECT
p.id,
p.identity_id,
p.organization_id,
i.email_address,
p.source,
p.state,
p.full_name,
p.kind,
p.additional_email_addresses,
p.position,
p.contract_start_date,
p.contract_end_date,
'' AS organization_name,
p.user_name,
p.external_id,
p.nickname,
p.locale,
p.timezone,
p.profile_url,
p.preferred_language,
p.given_name,
p.family_name,
p.formatted_name,
p.middle_name,
p.honorific_prefix,
p.honorific_suffix,
p.employee_number,
p.department,
p.cost_center,
p.enterprise_organization,
p.division,
p.manager_value,
p.activated_at,
p.deactivated_at,
p.created_at,
p.updated_at
FROM profiles p
INNER JOIN identities i ON i.id = p.identity_id
`
q = fmt.Sprintf(q, scope.SQLFragment(), cursor.SQLFragment())
args := pgx.NamedArgs{"version_id": documentVersionID}
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 document version approver profiles: %w", err)
}
profiles, err := pgx.CollectRows(rows, pgx.RowToAddrOfStructByName[MembershipProfile])
if err != nil {
return fmt.Errorf("cannot collect document version approver profiles: %w", err)
}
*p = profiles
return nil
}
func (p *MembershipProfiles) CountByDocumentVersionID(
ctx context.Context,
conn pg.Querier,
scope Scoper,
documentVersionID gid.GID,
) (int, error) {
q := `
WITH latest_quorum AS (
SELECT id
FROM document_version_approval_quorums
WHERE version_id = @version_id
ORDER BY created_at DESC
LIMIT 1
)
SELECT
COUNT(DISTINCT mp.id)
FROM
iam_membership_profiles mp
INNER JOIN document_version_approval_decisions dvad ON mp.id = dvad.approver_id
INNER JOIN latest_quorum lq ON lq.id = dvad.quorum_id
WHERE
mp.%s
`
q = fmt.Sprintf(q, scope.SQLFragment())
args := pgx.StrictNamedArgs{"version_id": documentVersionID}
maps.Copy(args, scope.SQLArguments())
var count int
err := conn.QueryRow(ctx, q, args).Scan(&count)
if err != nil {
return 0, fmt.Errorf("cannot query document version approver profiles count: %w", err)
}
return count, nil
}
func (p *MembershipProfiles) CountByIdentityID(
ctx context.Context,
conn pg.Querier,
identityID gid.GID,
filter *MembershipProfileFilter,
) (int, error) {
q := `
SELECT
COUNT(*)
FROM
iam_membership_profiles p
INNER JOIN identities i ON i.id = p.identity_id
WHERE
%s
AND p.identity_id = @identity_id
`
q = fmt.Sprintf(q, filter.SQLFragment())
args := pgx.StrictNamedArgs{"identity_id": identityID}
maps.Copy(args, filter.SQLArguments())
row := conn.QueryRow(ctx, q, args)
var count int
err := row.Scan(&count)
if err != nil {
return 0, fmt.Errorf("cannot collect count: %w", err)
}
return count, nil
}
func (p *MembershipProfiles) CountByOrganizationID(
ctx context.Context,
conn pg.Querier,
scope Scoper,
organizationID gid.GID,
filter *MembershipProfileFilter,
) (int, error) {
q := `
SELECT
COUNT(*)
FROM
iam_membership_profiles p
INNER JOIN identities i ON i.id = p.identity_id
WHERE
p.%s
AND %s
AND p.organization_id = @organization_id
`
q = fmt.Sprintf(q, scope.SQLFragment(), filter.SQLFragment())
args := pgx.StrictNamedArgs{"organization_id": organizationID}
maps.Copy(args, scope.SQLArguments())
maps.Copy(args, filter.SQLArguments())
row := conn.QueryRow(ctx, q, args)
var count int
err := row.Scan(&count)
if err != nil {
return 0, fmt.Errorf("cannot collect count: %w", err)
}
return count, nil
}
func (p *MembershipProfiles) CountActiveOwnerByOrganizationID(
ctx context.Context,
conn pg.Querier,
scope Scoper,
organizationID gid.GID,
) (int, error) {
q := `
SELECT
COUNT(*)
FROM
iam_membership_profiles p
INNER JOIN iam_memberships m ON m.identity_id = p.identity_id AND m.organization_id = p.organization_id
WHERE
p.%s
AND p.organization_id = @organization_id
AND p.state = @state
AND m.role = @role
`
q = fmt.Sprintf(q, scope.SQLFragment())
args := pgx.StrictNamedArgs{
"state": ProfileStateActive,
"role": MembershipRoleOwner,
"organization_id": organizationID,
}
maps.Copy(args, scope.SQLArguments())
row := conn.QueryRow(ctx, q, args)
var count int
err := row.Scan(&count)
if err != nil {
return 0, fmt.Errorf("cannot collect count: %w", err)
}
return count, nil
}
func (p *MembershipProfile) Insert(
ctx context.Context,
conn pg.Tx,
) error {
q := `
INSERT INTO
iam_membership_profiles (
tenant_id,
id,
identity_id,
organization_id,
source,
state,
full_name,
kind,
additional_email_addresses,
position,
contract_start_date,
contract_end_date,
user_name,
external_id,
nickname,
locale,
timezone,
profile_url,
preferred_language,
given_name,
family_name,
formatted_name,
middle_name,
honorific_prefix,
honorific_suffix,
employee_number,
department,
cost_center,
enterprise_organization,
division,
manager_value,
activated_at,
deactivated_at,
created_at,
updated_at
)
VALUES (
@tenant_id,
@id,
@identity_id,
@organization_id,
@source,
@state,
@full_name,
@kind,
COALESCE(@additional_email_addresses, '{}'::CITEXT[]),
@position,
@contract_start_date,
@contract_end_date,
@user_name,
@external_id,
@nickname,
@locale,
@timezone,
@profile_url,
@preferred_language,
@given_name,
@family_name,
@formatted_name,
@middle_name,
@honorific_prefix,
@honorific_suffix,
@employee_number,
@department,
@cost_center,
@enterprise_organization,
@division,
@manager_value,
@activated_at,
@deactivated_at,
@created_at,
@updated_at
)
`
args := pgx.StrictNamedArgs{
"tenant_id": p.ID.TenantID().String(),
"id": p.ID,
"identity_id": p.IdentityID,
"organization_id": p.OrganizationID,
"source": p.Source,
"state": p.State,
"full_name": p.FullName,
"kind": p.Kind,
"additional_email_addresses": p.AdditionalEmailAddresses,
"position": p.Position,
"contract_start_date": p.ContractStartDate,
"contract_end_date": p.ContractEndDate,
"user_name": p.UserName,
"external_id": p.ExternalID,
"nickname": p.Nickname,
"locale": p.Locale,
"timezone": p.Timezone,
"profile_url": p.ProfileUrl,
"preferred_language": p.PreferredLanguage,
"given_name": p.GivenName,
"family_name": p.FamilyName,
"formatted_name": p.FormattedName,
"middle_name": p.MiddleName,
"honorific_prefix": p.HonorificPrefix,
"honorific_suffix": p.HonorificSuffix,
"employee_number": p.EmployeeNumber,
"department": p.Department,
"cost_center": p.CostCenter,
"enterprise_organization": p.EnterpriseOrganization,
"division": p.Division,
"manager_value": p.ManagerValue,
"activated_at": p.ActivatedAt,
"deactivated_at": p.DeactivatedAt,
"created_at": p.CreatedAt,
"updated_at": p.UpdatedAt,
}
_, err := conn.Exec(ctx, q, args)
if err != nil {
if pgErr, ok := errors.AsType[*pgconn.PgError](err); ok && pgErr.Code == "23505" {
switch pgErr.ConstraintName {
case "idx_profiles_identity_id_organization_id",
"idx_profiles_external_id_organization_id",
"idx_profiles_user_name_organization_id":
return ErrResourceAlreadyExists
}
}
return fmt.Errorf("cannot insert profile: %w", err)
}
return nil
}
func (p *MembershipProfile) Update(
ctx context.Context,
conn pg.Tx,
scope Scoper,
) error {
q := `
UPDATE
iam_membership_profiles
SET
identity_id = @identity_id,
source = @source,
state = @state,
full_name = @full_name,
kind = @kind,
additional_email_addresses = COALESCE(@additional_email_addresses, '{}'::CITEXT[]),
position = @position,
contract_start_date = @contract_start_date,
contract_end_date = @contract_end_date,
user_name = @user_name,
external_id = @external_id,
nickname = @nickname,
locale = @locale,
timezone = @timezone,
profile_url = @profile_url,
preferred_language = @preferred_language,
given_name = @given_name,
family_name = @family_name,
formatted_name = @formatted_name,
middle_name = @middle_name,
honorific_prefix = @honorific_prefix,
honorific_suffix = @honorific_suffix,
employee_number = @employee_number,
department = @department,
cost_center = @cost_center,
enterprise_organization = @enterprise_organization,
division = @division,
manager_value = @manager_value,
activated_at = @activated_at,
deactivated_at = @deactivated_at,
updated_at = @updated_at
WHERE
id = @id
AND %s
`
q = fmt.Sprintf(q, scope.SQLFragment())
args := pgx.StrictNamedArgs{
"id": p.ID,
"identity_id": p.IdentityID,
"source": p.Source,
"state": p.State,
"full_name": p.FullName,
"kind": p.Kind,
"additional_email_addresses": p.AdditionalEmailAddresses,
"position": p.Position,
"contract_start_date": p.ContractStartDate,
"contract_end_date": p.ContractEndDate,
"user_name": p.UserName,
"external_id": p.ExternalID,
"nickname": p.Nickname,
"locale": p.Locale,
"timezone": p.Timezone,
"profile_url": p.ProfileUrl,
"preferred_language": p.PreferredLanguage,
"given_name": p.GivenName,
"family_name": p.FamilyName,
"formatted_name": p.FormattedName,
"middle_name": p.MiddleName,
"honorific_prefix": p.HonorificPrefix,
"honorific_suffix": p.HonorificSuffix,
"employee_number": p.EmployeeNumber,
"department": p.Department,
"cost_center": p.CostCenter,
"enterprise_organization": p.EnterpriseOrganization,
"division": p.Division,
"manager_value": p.ManagerValue,
"activated_at": p.ActivatedAt,
"deactivated_at": p.DeactivatedAt,
"updated_at": p.UpdatedAt,
}
maps.Copy(args, scope.SQLArguments())
result, err := conn.Exec(ctx, q, args)
if err != nil {
return fmt.Errorf("cannot update profile: %w", err)
}
if result.RowsAffected() == 0 {
return ErrResourceNotFound
}
return nil
}
func (p *MembershipProfiles) ResetSCIMSources(
ctx context.Context,
conn pg.Querier,
scope Scoper,
organizationID gid.GID,
) error {
q := `
UPDATE iam_membership_profiles
SET
source = 'MANUAL',
external_id = NULL,
user_name = NULL,
updated_at = @updated_at
WHERE
%s
AND organization_id = @organization_id
AND source = 'SCIM'
`
q = fmt.Sprintf(q, scope.SQLFragment())
args := pgx.NamedArgs{
"organization_id": organizationID,
"updated_at": time.Now(),
}
maps.Copy(args, scope.SQLArguments())
_, err := conn.Exec(ctx, q, args)
if err != nil {
return fmt.Errorf("cannot reset SCIM user sources: %w", err)
}
return nil
}
func (p *MembershipProfile) ClearExternalID(
ctx context.Context,
conn pg.Querier,
scope Scoper,
externalID string,
organizationID gid.GID,
) error {
q := `
UPDATE iam_membership_profiles
SET
external_id = NULL,
updated_at = @updated_at
WHERE
%s
AND external_id = @external_id
AND organization_id = @organization_id
AND id != @exclude_id
`
q = fmt.Sprintf(q, scope.SQLFragment())
args := pgx.NamedArgs{
"external_id": externalID,
"organization_id": organizationID,
"exclude_id": p.ID,
"updated_at": time.Now(),
}
maps.Copy(args, scope.SQLArguments())
_, err := conn.Exec(ctx, q, args)
if err != nil {
return fmt.Errorf("cannot clear external id: %w", err)
}
return nil
}
func (p *MembershipProfile) Delete(
ctx context.Context,
conn pg.Tx,
scope Scoper,
profileID gid.GID,
) error {
q := `
DELETE FROM
iam_membership_profiles
WHERE
id = @profile_id
AND %s
`
q = fmt.Sprintf(q, scope.SQLFragment())
args := pgx.StrictNamedArgs{"profile_id": profileID}
maps.Copy(args, scope.SQLArguments())
_, err := conn.Exec(ctx, q, args)
if err != nil {
if pgErr, ok := errors.AsType[*pgconn.PgError](err); ok {
if pgErr.Code == "23503" || pgErr.Code == "23001" {
return ErrResourceInUse
}
}
return fmt.Errorf("cannot delete profile: %w", err)
}
return nil
}
Reference in New Issue View Git Blame Copy Permalink