"""ContactMergeHistory model — tracks contact deduplication/merge operations.""" from __future__ import annotations import uuid from sqlalchemy import ForeignKey, Index, Text, text from sqlalchemy.dialects.postgresql import JSONB from sqlalchemy.dialects.postgresql import UUID as PGUUID from sqlalchemy.orm import Mapped, mapped_column from app.core.db import Base, TenantMixin class ContactMergeHistory(Base, TenantMixin): """Records each contact merge operation (source → target). When two duplicate contacts are merged, the source contact is soft-deleted and its entity_links, tags, etc. are re-pointed to the target contact. This table preserves an audit trail of which fields were merged and by whom. """ __tablename__ = "contact_merge_history" __table_args__ = ( Index("ix_contact_merge_history_tenant", "tenant_id"), Index("ix_contact_merge_history_target", "tenant_id", "target_contact_id"), Index("ix_contact_merge_history_source", "tenant_id", "source_contact_id"), ) id: Mapped[uuid.UUID] = mapped_column( PGUUID(as_uuid=True), primary_key=True, default=uuid.uuid4 ) source_contact_id: Mapped[uuid.UUID] = mapped_column( PGUUID(as_uuid=True), ForeignKey("contacts.id", ondelete="SET NULL"), nullable=False ) target_contact_id: Mapped[uuid.UUID] = mapped_column( PGUUID(as_uuid=True), ForeignKey("contacts.id", ondelete="CASCADE"), nullable=False ) merged_fields: Mapped[dict] = mapped_column( JSONB, nullable=False, server_default=text("'{}'::jsonb") ) merged_by: Mapped[uuid.UUID] = mapped_column( PGUUID(as_uuid=True), ForeignKey("users.id", ondelete="SET NULL"), nullable=True ) note: Mapped[str | None] = mapped_column(Text, nullable=True)