"""Identity stitching service for matching community members across platforms.""" from typing import Any from django.db import transaction from members.models import Identity, Member class IdentityStitchingService: """ Service for stitching identities across different platforms to create unified member profiles. """ @classmethod def find_or_create_member_for_identity( cls, *, workspace_id: int, provider: str, external_id: str, handle: str, email: str | None = None, display_name: str | None = None, ) -> tuple[Member, Identity, bool]: """ Find an existing member that matches the given identity, or create a new one. Returns: Tuple of (member, identity, created) where created indicates if a new member was created. """ with transaction.atomic(): # First, check if this exact identity already exists existing_identity = ( Identity.objects.filter(provider=provider, external_id=external_id).select_related("member").first() ) if existing_identity: return existing_identity.member, existing_identity, False # Try to find an existing member to stitch to member = cls._find_matching_member(workspace_id=workspace_id, provider=provider, handle=handle, email=email) if member: # Create identity linked to existing member identity = Identity.objects.create( member=member, provider=provider, external_id=external_id, handle=handle, ) return member, identity, False # Create new member member = Member.objects.create( workspace_id=workspace_id, display_name=display_name or handle, email=email or "", ) identity = Identity.objects.create( member=member, provider=provider, external_id=external_id, handle=handle, ) return member, identity, True @classmethod def _find_matching_member( cls, *, workspace_id: int, provider: str, handle: str, email: str | None, ) -> Member | None: """ Find a member that likely matches the given identity based on: 1. Email match (strongest signal) 2. Same handle on a related provider """ # Email matching is the strongest signal if email: member = Member.objects.filter( workspace_id=workspace_id, email__iexact=email, ).first() if member: return member # Try to find a member with the same handle on a related provider related_providers = cls._get_related_providers(provider=provider) if related_providers: identity = ( Identity.objects.filter( member__workspace_id=workspace_id, provider__in=related_providers, handle__iexact=handle, ) .select_related("member") .first() ) if identity: return identity.member return None @classmethod def _get_related_providers(cls, *, provider: str) -> list[str]: """ Get providers that are likely to share handles. E.g., GitHub and Discord often have the same username. """ related_groups = [ {"github", "discord"}, # Developers often use same handle {"twitter", "mastodon"}, # Social platforms {"slack", "discord"}, # Chat platforms ] for group in related_groups: if provider in group: return list(group - {provider}) return [] @classmethod def merge_members(cls, *, primary_member: Member, duplicate_member: Member) -> Member: """ Merge two members, moving all identities from the duplicate to the primary. The duplicate member will be deleted. """ if primary_member.workspace_id != duplicate_member.workspace_id: raise ValueError("Cannot merge members from different workspaces") if primary_member.id == duplicate_member.id: raise ValueError("Cannot merge member with itself") with transaction.atomic(): # Move all identities to the primary member Identity.objects.filter(member=duplicate_member).update(member=primary_member) # If primary has no email but duplicate does, copy it if not primary_member.email and duplicate_member.email: primary_member.email = duplicate_member.email primary_member.save(update_fields=["email"]) # Delete the duplicate duplicate_member.delete() return primary_member @classmethod def get_member_identities_summary(cls, *, member: Member) -> dict: """Get a summary of all identities for a member.""" identities = Identity.objects.filter(member=member).values( "id", "provider", "handle", "external_id", "created_at" ) providers: dict[str, list[dict[str, Any]]] = {} for identity in identities: provider = identity["provider"] if provider not in providers: providers[provider] = [] providers[provider].append( { "id": identity["id"], "handle": identity["handle"], "external_id": identity["external_id"], "created_at": identity["created_at"], } ) return { "member_id": member.id, "display_name": member.display_name, "email": member.email, "providers": providers, "total_identities": sum(len(ids) for ids in providers.values()), } @classmethod def suggest_potential_merges(cls, *, workspace_id: int, limit: int = 20) -> list[dict]: """ Find potential duplicate members in a workspace that could be merged. Returns a list of potential merge pairs with confidence scores. """ suggestions: list[dict[str, Any]] = [] # Find members with similar emails members_with_email = ( Member.objects.filter(workspace_id=workspace_id).exclude(email="").values("id", "display_name", "email") ) email_map: dict[str, list[dict[str, Any]]] = {} for member in members_with_email: email_base = member["email"].split("@")[0].lower() if email_base not in email_map: email_map[email_base] = [] email_map[email_base].append(dict(member)) for _email_base, members in email_map.items(): if len(members) > 1: for i, m1 in enumerate(members): for m2 in members[i + 1 :]: suggestions.append( { "member_1": {"id": m1["id"], "display_name": m1["display_name"], "email": m1["email"]}, "member_2": {"id": m2["id"], "display_name": m2["display_name"], "email": m2["email"]}, "reason": "similar_email", "confidence": 0.7, } ) # Find members with same handles across providers identities = ( Identity.objects.filter(member__workspace_id=workspace_id) .select_related("member") .values("member_id", "member__display_name", "handle", "provider") ) handle_map: dict[str, list[dict]] = {} for identity in identities: handle_lower = identity["handle"].lower() if handle_lower not in handle_map: handle_map[handle_lower] = [] handle_map[handle_lower].append( { "member_id": identity["member_id"], "display_name": identity["member__display_name"], "provider": identity["provider"], } ) for handle, members in handle_map.items(): unique_members = {} for m in members: if m["member_id"] not in unique_members: unique_members[m["member_id"]] = m if len(unique_members) > 1: member_list = list(unique_members.values()) for i, m1 in enumerate(member_list): for m2 in member_list[i + 1 :]: suggestions.append( { "member_1": {"id": m1["member_id"], "display_name": m1["display_name"]}, "member_2": {"id": m2["member_id"], "display_name": m2["display_name"]}, "reason": f"same_handle_{handle}", "confidence": 0.5, } ) # Sort by confidence and limit suggestions.sort(key=lambda x: -x["confidence"]) return suggestions[:limit]