dify/api/models/human_input_v2.py

"""Persistence models for retained Human Input v2 Contact, channel, IM sync, and inbox capabilities.

The models use explicit logical owner references where database foreign keys are intentionally absent.
Repository queries must therefore preserve complete owner predicates and explicit loading strategies.
Current IM Identity and Binding rows use their Channel as the persistence boundary.
Historical synchronization rows retain their captured Channel references.
"""

from __future__ import annotations

from collections.abc import Sequence
from datetime import datetime
from enum import StrEnum
from typing import Annotated, Literal

import sqlalchemy as sa
from pydantic import (
    BaseModel,
    ConfigDict,
    Field,
    JsonValue,
    NaiveDatetime,
    RootModel,
    TypeAdapter,
)
from sqlalchemy import orm
from sqlalchemy.orm import Mapped, mapped_column, relationship

from core.human_input_v2.entities import (
    EmailProviderType as _EmailProviderType,
)
from core.human_input_v2.entities import (
    IMProvider as _IMProvider,
)
from core.human_input_v2.entities import (
    IMSyncRemovalReason as _IMSyncRemovalReason,
)
from core.human_input_v2.entities import (
    IMSyncResultType as _IMSyncResultType,
)
from core.human_input_v2.entities import (
    IMSyncRunStatus as _IMSyncRunStatus,
)
from core.human_input_v2.im_integration.adapters.entities import (
    IMEventIngressKind as _IMEventIngressKind,
)
from core.human_input_v2.im_integration.change_log import (
    IMReconciliationOperation as _IMReconciliationOperation,
)
from core.human_input_v2.im_integration.change_log import (
    IMReconciliationSubjectKind as _IMReconciliationSubjectKind,
)
from core.human_input_v2.im_message_inbox import IM_INBOX_PROVIDER_METADATA_MAX_LENGTH
from core.human_input_v2.resolved_form import ResolvedForm
from core.workflow.nodes.human_input.enums import (
    HumanInputFormKind,
    HumanInputFormStatus,
)
from core.workflow.nodes.human_input_v2.entities import RecipientConfig
from libs.datetime_utils import naive_utc_now
from libs.uuid_utils import uuidv7
from repositories.human_input_v2.channel_credentials import ChannelEncryptedCredentials as _ChannelEncryptedCredentials
from repositories.human_input_v2.delivery_attempt_repository import DeliveryStatus
from repositories.human_input_v2.delivery_repository import (
    DeliveryTargetType,
    EmailTargetSnapshot,
    IMUserTargetSnapshot,
    InitiatorSnapshot,
    SubmissionAuthType,
    TargetSnapshot,
)
from repositories.human_input_v2.im_channel_repository import IMChannelStatus

from .base import DefaultFieldsDCMixin, TypeBase
from .types import EnumText, FrozenPydanticModelColumn, LongText, StringUUID


class _ImmutableJSONModel(BaseModel):
    """Strict immutable base for structured JSON persistence values."""

    model_config = ConfigDict(extra="forbid", frozen=True, strict=True, validate_default=True)


class _ImmutableJSONObject(RootModel[dict[str, JsonValue]]):
    """Strict immutable base for intentionally opaque JSON object payloads."""

    model_config = ConfigDict(frozen=True, strict=True, validate_default=True)


class IMIdentityRawPayload(RootModel[dict[str, JsonValue]]):
    """Opaque provider identity payload retained only for diagnostics."""

    model_config = ConfigDict(frozen=True, strict=True, validate_default=True)


class IMSyncDirectoryEntryPayload(_ImmutableJSONObject):
    """Opaque provider directory entry captured by one synchronization run."""


class IMSyncContactSnapshot(_ImmutableJSONModel):
    """Immutable contact values needed to display historical sync results."""

    contact_id: str = Field(description="Contact identifier captured by the sync result.")
    name: str = Field(description="Contact display name captured by the sync result.")
    email: str | None = Field(default=None, description="Contact email captured by the sync result.")
    avatar_file_id: str | None = Field(default=None, description="Avatar file identifier captured by the result.")
    created_at: NaiveDatetime | None = Field(
        default=None,
        description="Contact creation time captured by new results; absent from historical snapshots.",
    )


class IMSyncIdentitySnapshot(_ImmutableJSONModel):
    """Immutable last-known IM identity values retained after removal."""

    identity_id: str = Field(description="IM identity identifier captured by the sync result.")
    provider: _IMProvider = Field(
        strict=False,
        description="Provider that owned the captured IM identity.",
    )
    provider_user_id: str = Field(description="Provider user identifier captured by the sync result.")
    display_name: str | None = Field(default=None, description="Provider display name captured by the result.")
    email: str | None = Field(default=None, description="Provider email captured by the result.")


class IMIdentityReconciliationSnapshot(_ImmutableJSONModel):
    """Minimal current IM identity state retained by the reconciliation change log."""

    subject_kind: Literal[_IMReconciliationSubjectKind.IDENTITY] = _IMReconciliationSubjectKind.IDENTITY
    identity_id: str
    provider: _IMProvider = Field(strict=False)
    provider_user_id: str
    display_name: str | None = None
    email: str | None = None
    normalized_email: str | None = None
    last_seen_sync_run_id: str | None = None


class IMBindingReconciliationSnapshot(_ImmutableJSONModel):
    """Minimal current IM binding state retained by the reconciliation change log."""

    subject_kind: Literal[_IMReconciliationSubjectKind.BINDING] = _IMReconciliationSubjectKind.BINDING
    binding_id: str
    identity_id: str
    contact_id: str


type IMReconciliationChangeSnapshot = Annotated[
    IMIdentityReconciliationSnapshot | IMBindingReconciliationSnapshot,
    Field(discriminator="subject_kind"),
]


_IM_RECONCILIATION_SNAPSHOT_ADAPTER: TypeAdapter[IMReconciliationChangeSnapshot] = TypeAdapter(
    IMReconciliationChangeSnapshot
)


class ContactSubjectType(StrEnum):
    """Immutable Contact subject discriminator, never a current Contact type."""

    ACCOUNT = "account"
    EXTERNAL = "external"


class HumanInputContactIdentity(DefaultFieldsDCMixin, TypeBase):
    """Immutable Contact identity shared by every durable Contact reference.

    Account profile, membership, Platform visibility, and authorization state
    deliberately remain with their source owners. External mutable facts live
    in exactly one ``HumanInputExternalContactProfile``.
    """

    __tablename__ = "human_input_contact_identities"
    __table_args__ = (
        sa.CheckConstraint(
            "(subject_type = 'account' AND account_id IS NOT NULL) OR "
            "(subject_type = 'external' AND account_id IS NULL)",
            name="human_input_contact_identities_subject_type_ck",
        ),
        sa.UniqueConstraint(
            "account_id",
            name="human_input_contact_identities_account_id_uq",
        ),
        {
            "comment": (
                "Immutable Human Input Contact identities. Mutable Account and External Contact profile facts "
                "live with their source owners."
            )
        },
    )

    subject_type: Mapped[ContactSubjectType] = mapped_column(
        EnumText(ContactSubjectType),
        nullable=False,
        comment="Immutable Account or External subject discriminator.",
    )
    account_id: Mapped[str | None] = mapped_column(
        StringUUID,
        nullable=True,
        default=None,
        comment="Logical accounts.id reference for Account subjects only.",
    )

    external_profile: Mapped[HumanInputExternalContactProfile | None] = relationship(
        lambda: HumanInputExternalContactProfile,
        primaryjoin=lambda: sa.and_(
            HumanInputContactIdentity.id == orm.foreign(HumanInputExternalContactProfile.contact_id),
            HumanInputContactIdentity.subject_type == ContactSubjectType.EXTERNAL,
        ),
        back_populates="identity",
        viewonly=True,
        lazy="raise",
        init=False,
    )


class HumanInputExternalContactProfile(TypeBase):
    """Current workspace-owned profile for one External Contact identity."""

    __tablename__ = "human_input_external_contact_profiles"
    __table_args__ = (
        sa.UniqueConstraint(
            "tenant_id",
            "normalized_email",
            name="hiecp_tenant_normalized_email_uq",
        ),
        sa.Index(
            "hiecp_tenant_normalized_name_idx",
            "tenant_id",
            "normalized_name",
        ),
        {
            "comment": (
                "Current workspace-owned External Contact profiles. Deletion removes both this profile and its "
                "Contact identity."
            )
        },
    )

    contact_id: Mapped[str] = mapped_column(
        StringUUID,
        primary_key=True,
        comment="Logical human_input_contact_identities.id reference for one External subject.",
    )
    tenant_id: Mapped[str] = mapped_column(
        StringUUID,
        nullable=False,
        comment="Owning tenants.id used by every current External Contact lookup.",
    )
    name: Mapped[str] = mapped_column(
        sa.String(255),
        nullable=False,
        comment="Workspace-managed display name.",
    )
    normalized_name: Mapped[str] = mapped_column(
        sa.String(255),
        nullable=False,
        comment="Canonical search value maintained by External Contact writes.",
    )
    email: Mapped[str] = mapped_column(
        sa.String(320),
        nullable=False,
        comment="Workspace-managed deliverable Email address.",
    )
    normalized_email: Mapped[str] = mapped_column(
        sa.String(320),
        nullable=False,
        comment="Canonical Email equality value maintained by External Contact writes.",
    )
    avatar_file_id: Mapped[str | None] = mapped_column(
        StringUUID,
        nullable=True,
        default=None,
        comment="Logical upload_files.id reference owned by the same workspace.",
    )
    created_at: Mapped[datetime] = mapped_column(
        sa.DateTime,
        nullable=False,
        insert_default=naive_utc_now,
        init=False,
        server_default=sa.func.current_timestamp(),
    )
    updated_at: Mapped[datetime] = mapped_column(
        sa.DateTime,
        nullable=False,
        insert_default=naive_utc_now,
        init=False,
        server_default=sa.func.current_timestamp(),
        onupdate=sa.func.current_timestamp(),
    )

    identity: Mapped[HumanInputContactIdentity] = relationship(
        lambda: HumanInputContactIdentity,
        primaryjoin=lambda: orm.foreign(HumanInputExternalContactProfile.contact_id) == HumanInputContactIdentity.id,
        back_populates="external_profile",
        viewonly=True,
        lazy="raise",
        init=False,
    )


class HumanInputPlatformContactWorkspaceEntry(DefaultFieldsDCMixin, TypeBase):
    """EE-only workspace allow-list entry for an Organization Account contact.

    The entry does not own the Contact and does not duplicate workspace
    membership or the externally resolved Contact type. Its existence means that
    an EE Organization Account contact without current membership is explicitly
    available in one workspace. The Enterprise Contact repository must ensure the
    referenced Contact is an Account subject. Membership changes never mutate
    the Contact identity.
    """

    __tablename__ = "human_input_platform_contact_workspace_entries"
    __table_args__ = (
        sa.UniqueConstraint(
            "tenant_id",
            "contact_id",
            name="hipcwe_tenant_contact_uq",
        ),
        sa.Index("hipcwe_tenant_created_at_id_idx", "tenant_id", "created_at", "id"),
        sa.Index("hipcwe_contact_id_idx", "contact_id"),
        {
            "comment": (
                "EE-only workspace allow-list for Organization Account contacts. Workspace membership and External "
                "contact ownership must not create rows in this table."
            )
        },
    )

    tenant_id: Mapped[str] = mapped_column(StringUUID, nullable=False, comment="Logical foreign key to tenants.id.")
    contact_id: Mapped[str] = mapped_column(
        StringUUID,
        nullable=False,
        comment="Logical foreign key to human_input_contact_identities.id.",
    )
    added_by_account_id: Mapped[str] = mapped_column(
        StringUUID,
        nullable=False,
        comment="Logical foreign key to accounts.id for the administrator who added this directory entry.",
    )


class IMMessageEvent(TypeBase):
    """Authenticated callback facts and successful processing time."""

    __tablename__ = "im_message_event"
    __table_args__ = (
        sa.UniqueConstraint(
            "provider",
            "event_id",
            name="im_message_event_provider_event_uq",
        ),
        {"comment": "Durable authenticated IM callback records."},
    )

    id: Mapped[str] = mapped_column(
        StringUUID,
        primary_key=True,
        insert_default=lambda: str(uuidv7()),
        default_factory=lambda: str(uuidv7()),
        init=False,
    )
    created_at: Mapped[datetime] = mapped_column(
        sa.DateTime,
        nullable=False,
        insert_default=naive_utc_now,
        default_factory=naive_utc_now,
        init=False,
        server_default=sa.func.current_timestamp(),
    )

    channel_id: Mapped[str] = mapped_column(
        StringUUID,
        nullable=False,
        kw_only=True,
        comment="Logical local Channel routing identifier without physical ownership.",
    )
    provider: Mapped[_IMProvider] = mapped_column(
        EnumText(_IMProvider),
        nullable=False,
        kw_only=True,
        comment="Authenticated Provider discriminator.",
    )
    tenant_id: Mapped[str] = mapped_column(
        sa.String(IM_INBOX_PROVIDER_METADATA_MAX_LENGTH),
        nullable=False,
        kw_only=True,
        comment="Stable authenticated Provider tenant identifier.",
    )
    event_id: Mapped[str | None] = mapped_column(
        sa.String(IM_INBOX_PROVIDER_METADATA_MAX_LENGTH),
        nullable=True,
        default=None,
        kw_only=True,
        comment="Real Provider event ID when supplied.",
    )
    provider_event_time: Mapped[datetime | None] = mapped_column(
        sa.DateTime,
        nullable=True,
        default=None,
        kw_only=True,
        comment="Provider event timestamp when supplied.",
    )
    received_at: Mapped[datetime] = mapped_column(
        sa.DateTime,
        nullable=False,
        kw_only=True,
        comment="Dify receive timestamp for this delivery.",
    )
    provider_event_type: Mapped[str | None] = mapped_column(
        sa.String(IM_INBOX_PROVIDER_METADATA_MAX_LENGTH),
        nullable=True,
        default=None,
        kw_only=True,
        comment="Provider-owned event discriminator.",
    )
    ingress_kind: Mapped[_IMEventIngressKind] = mapped_column(
        EnumText(_IMEventIngressKind),
        nullable=False,
        kw_only=True,
        comment="Ingress contract used to construct the Provider payload snapshot.",
    )
    payload: Mapped[str] = mapped_column(
        LongText,
        nullable=False,
        kw_only=True,
        comment="Authenticated Provider-native payload.",
    )
    request_ip_address: Mapped[str | None] = mapped_column(
        sa.Text,
        nullable=True,
        default=None,
        kw_only=True,
        comment="HTTP request IP when available; absent for stream and historical events.",
    )
    processed_at: Mapped[datetime | None] = mapped_column(
        sa.DateTime,
        nullable=True,
        default=None,
        kw_only=True,
        comment="Timestamp when callback processing completed successfully.",
    )


class HumanInputEmailProvider(DefaultFieldsDCMixin, TypeBase):
    """Single workspace-level email provider used for Human Input delivery."""

    # TODO(QuantumGhost): Rename this model and table to HumanInputEmailChannel.

    __tablename__ = "human_input_email_providers"
    __table_args__ = (
        sa.UniqueConstraint("tenant_id", name="human_input_email_providers_tenant_uq"),
        {"comment": "Workspace-level Human Input email delivery configuration."},
    )

    provider: Mapped[_EmailProviderType] = mapped_column(
        EnumText(_EmailProviderType),
        nullable=False,
        comment="Configured email provider discriminator.",
    )
    sender_email: Mapped[str] = mapped_column(
        sa.String(320), nullable=False, comment="Configured sender email address."
    )
    encrypted_credentials: Mapped[_ChannelEncryptedCredentials] = mapped_column(
        FrozenPydanticModelColumn(_ChannelEncryptedCredentials),
        nullable=False,
        comment="Encrypted complete Channel credentials stored as JSON.",
    )
    tenant_id: Mapped[str] = mapped_column(
        StringUUID,
        nullable=False,
        comment="Logical foreign key to tenants.id.",
    )
    config_version: Mapped[int] = mapped_column(
        sa.Integer,
        nullable=False,
        default=1,
        comment="Monotonic Email configuration revision used for compare-and-swap.",
    )
    sender_name: Mapped[str] = mapped_column(
        sa.String(255),
        nullable=False,
        default="",
        comment="Optional sender display name.",
    )
    configured_by_account_id: Mapped[str | None] = mapped_column(
        StringUUID,
        nullable=True,
        default=None,
        comment="Logical foreign key to accounts.id for the latest configuration write.",
    )


class HumanInputIMChannel(DefaultFieldsDCMixin, TypeBase):
    """One current IM Channel in a repository-generated owner slot.

    ``owner_key`` deliberately replaces a polymorphic database foreign key.
    Its unique constraint serializes concurrent first creation for both
    workspace and deployment owners without Redis or a separate lock row.
    """

    __tablename__ = "human_input_im_channels"
    __table_args__ = (
        sa.UniqueConstraint(
            "owner_key",
            name="human_input_im_channels_owner_key_uq",
        ),
        sa.UniqueConstraint(
            "webhook_id",
            name="human_input_im_channels_webhook_id_uq",
        ),
        sa.CheckConstraint(
            "config_version > 0",
            name=sa.schema.conv("human_input_im_channels_config_version_positive_ck"),
        ),
        {
            "comment": (
                "Current owner-scoped Human Input IM Channel configuration. "
                "Directory, Binding, Sync, and inbox records remain separately owned."
            )
        },
    )

    owner_key: Mapped[str] = mapped_column(
        sa.String(50),
        nullable=False,
        comment="Canonical owner slot: workspace:<tenant_id> or deployment.",
    )
    provider: Mapped[_IMProvider] = mapped_column(
        EnumText(_IMProvider),
        nullable=False,
        comment="Configured IM provider discriminator.",
    )
    provider_tenant_id: Mapped[str] = mapped_column(
        sa.String(255),
        nullable=False,
        comment="Confirmed provider-side organization, tenant, or workspace identifier.",
    )
    encrypted_credentials: Mapped[_ChannelEncryptedCredentials] = mapped_column(
        FrozenPydanticModelColumn(_ChannelEncryptedCredentials),
        nullable=False,
        comment="Versioned opaque encrypted IM Channel credential envelope.",
    )
    app_identifier: Mapped[str] = mapped_column(
        sa.String(255),
        nullable=False,
        comment="Safe provider application identifier used by credential-free projections.",
    )
    webhook_id: Mapped[str] = mapped_column(
        sa.String(32),
        nullable=False,
        comment="Server-generated globally unique route ID used to derive webhook URLs.",
    )
    status: Mapped[IMChannelStatus] = mapped_column(
        EnumText(IMChannelStatus),
        nullable=False,
        default=IMChannelStatus.CONNECTED,
        comment="Stored credential-safe Channel status snapshot.",
    )
    config_version: Mapped[int] = mapped_column(
        sa.Integer,
        nullable=False,
        default=1,
        comment="Monotonic numeric version paired with the Channel ID for CAS.",
    )
    configured_by_account_id: Mapped[str | None] = mapped_column(
        StringUUID,
        nullable=True,
        default=None,
        comment="Latest configuring Dify Account; null for deployment-owned writes.",
    )
    status_reason: Mapped[str | None] = mapped_column(
        LongText,
        nullable=True,
        default=None,
        comment="Operator-safe status explanation without provider payload or credentials.",
    )


class HumanInputIMIdentity(DefaultFieldsDCMixin, TypeBase):
    """Current Provider user synchronized through one IM Channel.

    Searchable and match-critical fields are stored in columns. ``raw_payload``
    retains the Provider response for diagnostics without making it query state.
    Channel ownership and Provider configuration remain on the parent Channel.
    """

    __tablename__ = "human_input_im_identities"
    __table_args__ = (
        sa.UniqueConstraint(
            "channel_id",
            "provider_user_id",
            name="human_input_im_identities_channel_provider_user_uq",
        ),
        sa.CheckConstraint(
            "length(trim(provider_user_id)) > 0",
            name="human_input_im_identities_provider_user_nonblank",
        ),
        sa.Index("hiimi_channel_id_idx", "channel_id", "id"),
        sa.Index("hiimi_channel_email_idx", "channel_id", "normalized_email"),
        sa.Index("hiimi_channel_name_idx", "channel_id", "normalized_name"),
        {
            "comment": (
                "Current Provider users synchronized through one IM Channel. "
                "Channel ownership remains outside this table."
            )
        },
    )

    channel_id: Mapped[str] = mapped_column(
        StringUUID,
        nullable=False,
        comment="Logical human_input_im_channels.id reference.",
    )
    provider_user_id: Mapped[str] = mapped_column(
        sa.String(255),
        nullable=False,
        comment="Provider-native user identifier within the owning Channel.",
    )
    raw_payload: Mapped[IMIdentityRawPayload] = mapped_column(
        FrozenPydanticModelColumn(IMIdentityRawPayload),
        nullable=False,
        comment="Latest opaque Provider payload retained for diagnostics.",
    )
    last_seen_sync_run_id: Mapped[str] = mapped_column(
        StringUUID,
        nullable=False,
        comment="Logical human_input_im_sync_runs.id reference for the latest observation.",
    )
    last_seen_at: Mapped[datetime] = mapped_column(
        sa.DateTime,
        nullable=False,
        comment="Timestamp of the latest successful Provider observation.",
    )
    display_name: Mapped[str | None] = mapped_column(
        sa.String(255),
        nullable=True,
        default=None,
        comment="Latest canonical non-blank Provider display name.",
    )
    normalized_name: Mapped[str | None] = mapped_column(
        sa.String(255),
        nullable=True,
        default=None,
        comment="Canonical display name used by persistence queries.",
    )
    email: Mapped[str | None] = mapped_column(
        sa.String(320),
        nullable=True,
        default=None,
        comment="Latest canonical non-blank Provider email.",
    )
    normalized_email: Mapped[str | None] = mapped_column(
        sa.String(320),
        nullable=True,
        default=None,
        comment="Canonical email used by matching and persistence queries.",
    )
    avatar_file_id: Mapped[str | None] = mapped_column(
        StringUUID,
        nullable=True,
        default=None,
        comment="Logical upload_files.id reference for the current IM avatar.",
    )


class HumanInputIMBinding(DefaultFieldsDCMixin, TypeBase):
    """Default Contact-to-IM-identity Binding for one IM Channel."""

    __tablename__ = "human_input_im_bindings"
    __table_args__ = (
        sa.UniqueConstraint(
            "channel_id",
            "contact_id",
            name="human_input_im_bindings_channel_contact_uq",
        ),
        sa.UniqueConstraint(
            "channel_id",
            "im_identity_id",
            name="human_input_im_bindings_channel_identity_uq",
        ),
        {"comment": "Default Contact-to-IM-identity Bindings for one IM Channel."},
    )

    channel_id: Mapped[str] = mapped_column(
        StringUUID,
        nullable=False,
        comment="Logical human_input_im_channels.id reference.",
    )
    contact_id: Mapped[str] = mapped_column(
        StringUUID,
        nullable=False,
        comment="Logical human_input_contact_identities.id reference.",
    )
    im_identity_id: Mapped[str] = mapped_column(
        StringUUID,
        nullable=False,
        comment="Logical human_input_im_identities.id reference.",
    )
    bound_by_account_id: Mapped[str | None] = mapped_column(
        StringUUID,
        nullable=True,
        default=None,
        comment="Latest Dify Account that manually selected this Binding, when available.",
    )


class HumanInputIMBindingWorkspaceOverride(DefaultFieldsDCMixin, TypeBase):
    """Workspace-specific override for one default IM Binding."""

    __tablename__ = "human_input_im_workspace_binding_overrides"
    __table_args__ = (
        sa.UniqueConstraint(
            "channel_id",
            "tenant_id",
            "contact_id",
            name="hiimwbo_channel_tenant_contact_uq",
        ),
        sa.UniqueConstraint(
            "channel_id",
            "tenant_id",
            "im_identity_id",
            name="hiimwbo_channel_tenant_identity_uq",
        ),
        sa.Index("hiimwbo_channel_identity_idx", "channel_id", "im_identity_id"),
        {"comment": "Workspace-specific Binding overrides for one IM Channel."},
    )

    channel_id: Mapped[str] = mapped_column(
        StringUUID,
        nullable=False,
        comment="Logical human_input_im_channels.id reference.",
    )
    tenant_id: Mapped[str] = mapped_column(
        StringUUID,
        nullable=False,
        comment="Target tenants.id whose effective Binding is overridden.",
    )
    contact_id: Mapped[str] = mapped_column(
        StringUUID,
        nullable=False,
        comment="Logical human_input_contact_identities.id reference.",
    )
    im_identity_id: Mapped[str] = mapped_column(
        StringUUID,
        nullable=False,
        comment="Logical human_input_im_identities.id reference.",
    )
    bound_by_account_id: Mapped[str | None] = mapped_column(
        StringUUID,
        nullable=True,
        default=None,
        comment="Dify Account that selected this workspace override.",
    )


class HumanInputIMSyncRun(DefaultFieldsDCMixin, TypeBase):
    """One manually triggered organization IM directory synchronization.

    Preventing parallel queued or running rows is a service transaction invariant:
    lock the owning channel row before creating a run. A portable partial
    unique index is deliberately not used. A worker may apply current identities
    and bindings only while ``channel_config_version`` still matches the
    owning channel.
    """

    __tablename__ = "human_input_im_sync_runs"
    __table_args__ = (
        sa.CheckConstraint("channel_config_version > 0", name="captured_version_positive"),
        sa.CheckConstraint(
            "(started_by_account_id IS NOT NULL AND started_by_sys_user_id IS NULL) OR "
            "(started_by_account_id IS NULL AND started_by_sys_user_id IS NOT NULL)",
            name="initiator_exclusive",
        ),
        sa.CheckConstraint(
            "added_count >= 0 AND not_matched_count >= 0 AND failed_count >= 0 AND removed_count >= 0 "
            "AND skipped_count >= 0",
            name="result_counts_nonnegative",
        ),
        sa.Index("hiimsr_channel_created_idx", "channel_id", "created_at", "id"),
        sa.Index(
            "hiimsr_channel_status_created_idx",
            "channel_id",
            "status",
            "created_at",
        ),
        {"comment": "Manual IM directory synchronization lifecycle and aggregate counts."},
    )
    channel_id: Mapped[str] = mapped_column(
        StringUUID,
        nullable=False,
        comment="Logical foreign key to human_input_im_channels.id.",
    )
    channel_config_version: Mapped[int] = mapped_column(
        sa.Integer,
        nullable=False,
        comment="Channel configuration revision captured when this synchronization was created.",
    )
    provider: Mapped[_IMProvider] = mapped_column(
        EnumText(_IMProvider),
        nullable=False,
        comment="Provider snapshot used by this synchronization.",
    )
    status: Mapped[_IMSyncRunStatus] = mapped_column(
        EnumText(_IMSyncRunStatus),
        nullable=False,
        comment="Current synchronization lifecycle state.",
    )
    added_count: Mapped[int] = mapped_column(
        sa.Integer,
        nullable=False,
        default=0,
        comment="Number of entries newly matched and bound.",
    )
    not_matched_count: Mapped[int] = mapped_column(
        sa.Integer,
        nullable=False,
        default=0,
        comment="Number of entries requiring later manual handling.",
    )
    failed_count: Mapped[int] = mapped_column(
        sa.Integer,
        nullable=False,
        default=0,
        comment="Number of entries that failed reconciliation.",
    )
    removed_count: Mapped[int] = mapped_column(
        sa.Integer,
        nullable=False,
        default=0,
        comment="Number of removed binding facts, including one unbound-identity fact when applicable.",
    )
    skipped_count: Mapped[int] = mapped_column(
        sa.Integer,
        nullable=False,
        default=0,
        comment="Number of entries intentionally skipped.",
    )
    started_by_account_id: Mapped[str | None] = mapped_column(
        StringUUID,
        nullable=True,
        default=None,
        comment="Logical accounts.id reference for CE/SaaS initiators; null only for EE sys_user initiators.",
    )
    started_by_sys_user_id: Mapped[str | None] = mapped_column(
        StringUUID,
        nullable=True,
        default=None,
        comment="Logical reference to upstream EE sys_users.id; mutually exclusive with started_by_account_id.",
    )
    started_at: Mapped[datetime | None] = mapped_column(
        sa.DateTime,
        nullable=True,
        default=None,
        comment="Timestamp when a worker started processing the run.",
    )
    finished_at: Mapped[datetime | None] = mapped_column(
        sa.DateTime,
        nullable=True,
        default=None,
        comment="Terminal completion timestamp.",
    )
    error_code: Mapped[str | None] = mapped_column(
        sa.String(100),
        nullable=True,
        default=None,
        comment="Machine-readable terminal failure code.",
    )
    error_message: Mapped[str | None] = mapped_column(
        LongText,
        nullable=True,
        default=None,
        comment="Operator-safe terminal failure summary.",
    )

    results: Mapped[list[HumanInputIMSyncResult]] = relationship(
        lambda: HumanInputIMSyncResult,
        primaryjoin=lambda: HumanInputIMSyncRun.id == orm.foreign(HumanInputIMSyncResult.sync_run_id),
        back_populates="sync_run",
        viewonly=True,
        lazy="raise",
        init=False,
    )


class HumanInputIMSyncResult(DefaultFieldsDCMixin, TypeBase):
    """Immutable reconciliation result for one entry, removed binding, or diagnostic.

    Provider identifiers and normalized email remain queryable columns. JSON is
    limited to the raw provider input and immutable display snapshots.
    """

    __tablename__ = "human_input_im_sync_results"
    __table_args__ = (
        sa.UniqueConstraint(
            "sync_run_id",
            "operation_key",
            name="human_input_im_sync_results_run_operation_uq",
        ),
        sa.Index(
            "hiimsres_run_type_created_idx",
            "sync_run_id",
            "result_type",
            "created_at",
            "id",
        ),
        sa.Index(
            "hiimsres_channel_contact_created_idx",
            "channel_id",
            "contact_id",
            "created_at",
        ),
        sa.Index(
            "hiimsres_channel_identity_created_idx",
            "channel_id",
            "im_identity_id",
            "created_at",
        ),
        {"comment": "Append-only per-entry, removed-binding, and diagnostic IM synchronization outcomes."},
    )

    channel_id: Mapped[str] = mapped_column(
        StringUUID,
        nullable=False,
        comment=(
            "Denormalized logical foreign key to human_input_im_channels.id; "
            "must match the channel referenced by sync_run_id."
        ),
    )
    sync_run_id: Mapped[str] = mapped_column(
        StringUUID,
        nullable=False,
        comment="Logical foreign key to human_input_im_sync_runs.id.",
    )
    result_type: Mapped[_IMSyncResultType] = mapped_column(
        EnumText(_IMSyncResultType),
        nullable=False,
        comment="Stable result bucket used by pagination.",
    )
    operation_key: Mapped[str | None] = mapped_column(
        sa.String(255),
        nullable=True,
        default=None,
        comment="Deterministic run-local idempotency key; null only for historical results.",
    )
    provider_user_id: Mapped[str | None] = mapped_column(
        sa.String(255),
        nullable=True,
        default=None,
        comment="External provider user identifier observed for this result; not a logical foreign key.",
    )
    display_name: Mapped[str | None] = mapped_column(
        sa.String(255),
        nullable=True,
        default=None,
        comment="Provider display name observed for this result.",
    )
    email: Mapped[str | None] = mapped_column(
        sa.String(320),
        nullable=True,
        default=None,
        comment="Provider email observed for this result.",
    )
    normalized_email: Mapped[str | None] = mapped_column(
        sa.String(320),
        nullable=True,
        default=None,
        comment="Lower-cased provider email used during matching.",
    )
    contact_id: Mapped[str | None] = mapped_column(
        StringUUID,
        nullable=True,
        default=None,
        comment="Logical foreign key to human_input_contact_identities.id, when a contact was matched.",
    )
    im_identity_id: Mapped[str | None] = mapped_column(
        StringUUID,
        nullable=True,
        default=None,
        comment=(
            "Historical logical foreign key to human_input_im_identities.id; the target row may be deleted after sync."
        ),
    )
    im_binding_id: Mapped[str | None] = mapped_column(
        StringUUID,
        nullable=True,
        default=None,
        comment=(
            "Historical logical foreign key to human_input_im_bindings.id; the target row may be deleted or replaced."
        ),
    )
    removal_reason: Mapped[_IMSyncRemovalReason | None] = mapped_column(
        EnumText(_IMSyncRemovalReason),
        nullable=True,
        default=None,
        comment="Stable removal reason for removed results only.",
    )
    reason_code: Mapped[str | None] = mapped_column(
        sa.String(100),
        nullable=True,
        default=None,
        comment="Machine-readable failure, skip, or mismatch reason.",
    )
    reason_message: Mapped[str | None] = mapped_column(
        LongText,
        nullable=True,
        default=None,
        comment="Operator-safe detail for diagnostics.",
    )
    directory_entry_payload: Mapped[IMSyncDirectoryEntryPayload | None] = mapped_column(
        FrozenPydanticModelColumn(IMSyncDirectoryEntryPayload),
        nullable=True,
        default=None,
        comment="Immutable provider entry payload Pydantic model observed during this run.",
    )
    contact_snapshot: Mapped[IMSyncContactSnapshot | None] = mapped_column(
        FrozenPydanticModelColumn(IMSyncContactSnapshot),
        nullable=True,
        default=None,
        comment="Immutable contact snapshot Pydantic model used for historical display.",
    )
    identity_snapshot: Mapped[IMSyncIdentitySnapshot | None] = mapped_column(
        FrozenPydanticModelColumn(IMSyncIdentitySnapshot),
        nullable=True,
        default=None,
        comment="Immutable last-known IM identity Pydantic model for removed results.",
    )

    sync_run: Mapped[HumanInputIMSyncRun] = relationship(
        lambda: HumanInputIMSyncRun,
        primaryjoin=lambda: orm.foreign(HumanInputIMSyncResult.sync_run_id) == HumanInputIMSyncRun.id,
        back_populates="results",
        viewonly=True,
        lazy="raise",
        init=False,
    )
    identity: Mapped[HumanInputIMIdentity | None] = relationship(
        lambda: HumanInputIMIdentity,
        primaryjoin=lambda: orm.foreign(HumanInputIMSyncResult.im_identity_id) == HumanInputIMIdentity.id,
        viewonly=True,
        lazy="raise",
        init=False,
    )
    binding: Mapped[HumanInputIMBinding | None] = relationship(
        lambda: HumanInputIMBinding,
        primaryjoin=lambda: orm.foreign(HumanInputIMSyncResult.im_binding_id) == HumanInputIMBinding.id,
        viewonly=True,
        lazy="raise",
        init=False,
    )


class HumanInputIMReconciliationChange(DefaultFieldsDCMixin, TypeBase):
    """Append-only before/after fact for one committed identity or IM binding mutation."""

    __tablename__ = "human_input_im_reconciliation_changes"
    __table_args__ = (
        sa.UniqueConstraint(
            "sync_run_id",
            "operation_key",
            name="human_input_im_reconciliation_changes_run_operation_uq",
        ),
        sa.CheckConstraint(
            "before_snapshot IS NOT NULL OR after_snapshot IS NOT NULL",
            name="snapshot_present",
        ),
        sa.CheckConstraint(
            "(operation = 'create' AND before_snapshot IS NULL AND after_snapshot IS NOT NULL) OR "
            "(operation = 'delete' AND before_snapshot IS NOT NULL AND after_snapshot IS NULL) OR "
            "(operation NOT IN ('create', 'delete') AND before_snapshot IS NOT NULL AND after_snapshot IS NOT NULL)",
            name="snapshot_operation_shape",
        ),
        sa.CheckConstraint(
            "(subject_kind = 'identity' AND im_binding_id IS NULL) OR "
            "(subject_kind = 'binding' AND im_binding_id IS NOT NULL)",
            name="subject_identifier_shape",
        ),
        sa.Index(
            "hiimrc_run_subject_committed_idx",
            "sync_run_id",
            "subject_kind",
            "committed_at",
            "id",
        ),
        sa.Index("hiimrc_channel_committed_idx", "channel_id", "committed_at", "id"),
        {"comment": "Append-only IM identity and IM binding reconciliation mutation history."},
    )

    channel_id: Mapped[str] = mapped_column(
        StringUUID,
        nullable=False,
        comment="Logical foreign key to human_input_im_channels.id.",
    )
    sync_run_id: Mapped[str] = mapped_column(
        StringUUID,
        nullable=False,
        comment="Logical foreign key to human_input_im_sync_runs.id.",
    )
    operation_key: Mapped[str] = mapped_column(
        sa.String(255),
        nullable=False,
        comment="Deterministic run-local idempotency key.",
    )
    subject_kind: Mapped[_IMReconciliationSubjectKind] = mapped_column(
        EnumText(_IMReconciliationSubjectKind), nullable=False
    )
    operation: Mapped[_IMReconciliationOperation] = mapped_column(EnumText(_IMReconciliationOperation), nullable=False)
    reason_code: Mapped[str] = mapped_column(sa.String(100), nullable=False)
    im_identity_id: Mapped[str] = mapped_column(StringUUID, nullable=False)
    committed_at: Mapped[datetime] = mapped_column(sa.DateTime, nullable=False)
    im_binding_id: Mapped[str | None] = mapped_column(StringUUID, nullable=True, default=None)
    contact_id: Mapped[str | None] = mapped_column(StringUUID, nullable=True, default=None)
    before_snapshot: Mapped[IMReconciliationChangeSnapshot | None] = mapped_column(
        FrozenPydanticModelColumn(
            _IM_RECONCILIATION_SNAPSHOT_ADAPTER,
            model_types=(
                IMIdentityReconciliationSnapshot,
                IMBindingReconciliationSnapshot,
            ),
        ),
        nullable=True,
        default=None,
    )
    after_snapshot: Mapped[IMReconciliationChangeSnapshot | None] = mapped_column(
        FrozenPydanticModelColumn(
            _IM_RECONCILIATION_SNAPSHOT_ADAPTER,
            model_types=(
                IMIdentityReconciliationSnapshot,
                IMBindingReconciliationSnapshot,
            ),
        ),
        nullable=True,
        default=None,
    )


class HumanInputForm(DefaultFieldsDCMixin, TypeBase):
    __tablename__ = "hitlv2_forms"
    __table_args__ = (
        sa.UniqueConstraint(
            "workflow_run_id",
            "node_execution_id",
            name="hitlv2_forms_execution_uq",
        ),
        sa.Index("hitlv2_forms_run_id_idx", "workflow_run_id", "id"),
        # Equality predicates precede the ID cursor used by expiration scans.
        sa.Index("hitlv2_forms_expiration_scan_idx", "tenant_id", "app_id", "form_kind", "status", "id"),
        sa.Index("hitlv2_forms_node_deadline_idx", "form_kind", "status", "expiration_time", "id"),
        sa.Index("hitlv2_forms_global_deadline_idx", "form_kind", "status", "global_timeout_deadline", "id"),
    )

    tenant_id: Mapped[str] = mapped_column(StringUUID, nullable=False)
    app_id: Mapped[str] = mapped_column(StringUUID, nullable=False)
    workflow_run_id: Mapped[str | None] = mapped_column(StringUUID, nullable=True)
    # correspond to WorkflowNodeExecutionModel.id
    node_execution_id: Mapped[str | None] = mapped_column(StringUUID, nullable=True)
    form_kind: Mapped[HumanInputFormKind] = mapped_column(
        EnumText(HumanInputFormKind),
        nullable=False,
        default=HumanInputFormKind.RUNTIME,
        kw_only=True,
    )
    status: Mapped[HumanInputFormStatus] = mapped_column(
        EnumText(HumanInputFormStatus),
        nullable=False,
        default=HumanInputFormStatus.WAITING,
        server_default=HumanInputFormStatus.WAITING.value,
        kw_only=True,
        comment="Persisted form outcome; terminal states are not recomputed from deadlines.",
    )
    # Form level expiration time.
    expiration_time: Mapped[datetime] = mapped_column(
        sa.DateTime,
        nullable=False,
    )
    # global timeout deadline for this form / workflow execution.
    global_timeout_deadline: Mapped[datetime] = mapped_column(
        sa.DateTime,
        nullable=False,
    )
    # The form definition after resolving all runtime related variables.
    resolved_form: Mapped[ResolvedForm] = mapped_column(FrozenPydanticModelColumn(ResolvedForm), nullable=False)

    # The following fields are not `None` once the form has been submitted.
    submitted_at: Mapped[datetime | None] = mapped_column(sa.DateTime, nullable=True)
    submitted_by_recipient_id: Mapped[str | None] = mapped_column(StringUUID, nullable=True)
    submitted_by_delivery_id: Mapped[str | None] = mapped_column(StringUUID, nullable=True, default=None, kw_only=True)
    # submission_data
    selected_action_id: Mapped[str | None] = mapped_column(sa.String(200), nullable=True)
    # original submission inputs data without any transformation or sanitization.
    raw_submission_inputs: Mapped[str | None] = mapped_column(sa.Text, nullable=True)
    # validated and sanitized submission data, a mapping from submitted form data to
    # variable segments.
    #
    # For special reference types such as File / FileSegment, the File ownership has been
    # validated.
    normalized_submission_data: Mapped[str | None] = mapped_column(sa.Text, nullable=True)


class RecipientSubjectType(StrEnum):
    CONTACT = "contact"
    EMAIL = "email"
    END_USER = "end_user"


class RecipientSources(RootModel[Sequence[RecipientConfig]]):
    model_config = ConfigDict(frozen=True, strict=True, validate_default=True)


class RecipientSnapshot(BaseModel):
    model_config = ConfigDict(frozen=True, strict=True, validate_default=True)
    # The name of the recipeint when the recipient is created.
    #
    # For recipeint without a name (Dyanmic Email / Onetime Email / EndUser), this
    # Field is set to `None`.
    name: str | None = None


class HumanInputRecipient(DefaultFieldsDCMixin, TypeBase):
    __tablename__ = "hitlv2_recipients"
    __table_args__ = (
        sa.UniqueConstraint(
            "form_id",
            "subject_type",
            "subject_value",
            name="hitlv2_recipients_subject_uq",
        ),
    )

    tenant_id: Mapped[str] = mapped_column(StringUUID, nullable=False)
    form_id: Mapped[str] = mapped_column(StringUUID, nullable=False)

    subject_type: Mapped[RecipientSubjectType] = mapped_column(
        EnumText(RecipientSubjectType),
        nullable=False,
    )
    # The interpretation of identity_value depends on `subject_value`:
    #
    # - For RecipientSubjectType.CONTACT, this field stores the value of HumanInputContactIdentity.id
    # - For RecipientSubjectType.END_USER, this field stores the value of EndUser.id
    # - For RecipientSubjectType.EMAIL, this field stores the normalized email of the recipient.
    subject_value: Mapped[str] = mapped_column(sa.String(255), nullable=False)
    sources: Mapped[RecipientSources] = mapped_column(
        FrozenPydanticModelColumn(RecipientSources),
        nullable=False,
    )
    snapshot: Mapped[RecipientSnapshot] = mapped_column(
        FrozenPydanticModelColumn(RecipientSnapshot),
        nullable=False,
    )


class DeliveryType(StrEnum):
    IM_CARD = "im_card"
    COSNOLE_WEB = "console_web"
    STANDALONE_WEB = "standalone_web"


_TARGET_SNAPSHOT_ADAPTER: TypeAdapter[TargetSnapshot] = TypeAdapter(TargetSnapshot)


class HumanInputDelivery(DefaultFieldsDCMixin, TypeBase):
    __tablename__ = "hitlv2_deliveries"
    __table_args__ = (
        sa.Index("hitlv2_deliveries_token_idx", "token_hash"),
        sa.Index("hitlv2_deliveries_form_id_idx", "form_id", "id"),
        sa.Index("hitlv2_deliveries_form_target_idx", "form_id", "target_type"),
    )

    tenant_id: Mapped[str] = mapped_column(StringUUID, nullable=False)
    form_id: Mapped[str] = mapped_column(StringUUID, nullable=False)

    token_hash: Mapped[str] = mapped_column(sa.String(64), nullable=False)

    recipient_id: Mapped[str] = mapped_column(StringUUID, nullable=False)
    auth_type: Mapped[SubmissionAuthType] = mapped_column(
        EnumText(SubmissionAuthType),
        nullable=False,
    )
    target_type: Mapped[DeliveryTargetType] = mapped_column(EnumText(DeliveryTargetType), nullable=False)
    target_snapshot: Mapped[TargetSnapshot] = mapped_column(
        FrozenPydanticModelColumn(
            _TARGET_SNAPSHOT_ADAPTER,
            model_types=(IMUserTargetSnapshot, EmailTargetSnapshot, InitiatorSnapshot),
        ),
        nullable=False,
    )


class DeliveryResponse(RootModel[dict[str, JsonValue]]):
    model_config = ConfigDict(frozen=True, strict=True, validate_default=True)


class HumanInputDeliveryAttempt(DefaultFieldsDCMixin, TypeBase):
    __tablename__ = "hitlv2_delivery_attempts"

    tenant_id: Mapped[str] = mapped_column(StringUUID, nullable=False)
    form_id: Mapped[str] = mapped_column(StringUUID, nullable=False)

    delivery_id: Mapped[str] = mapped_column(StringUUID, nullable=False)
    status: Mapped[DeliveryStatus] = mapped_column(EnumText(DeliveryStatus), nullable=False)
    error_message: Mapped[str | None] = mapped_column(sa.String(200), nullable=True)
    response: Mapped[DeliveryResponse] = mapped_column(FrozenPydanticModelColumn(DeliveryResponse), nullable=False)


class AuditEventType(StrEnum):
    # The form is opened and viewed.
    FORM_VIEW = "form_view"
    # An Email OTP send operation
    OTP_SEND = "otp_send"
    # An Email OTP verification
    OTP_VERIFICATION = "otp_verification"
    # This seems to break the IM abstration???
    IM_CALLBACK_VERIFICATION = "im_callback_verification"
    #
    FORM_SUBMISSION = "form_submission"


class FormViewEvent(_ImmutableJSONModel):
    type: Literal[AuditEventType.FORM_VIEW] = AuditEventType.FORM_VIEW
    # None means the request IP was unavailable.
    viewer_ip_address: str | None


class OtpSendEvent(_ImmutableJSONModel):
    type: Literal[AuditEventType.OTP_SEND] = AuditEventType.OTP_SEND

    recipient_id: str
    recipient_email: str
    sender_ip_address: str | None


class OtpVerificationEvent(_ImmutableJSONModel):
    type: Literal[AuditEventType.OTP_VERIFICATION] = AuditEventType.OTP_VERIFICATION

    recipient_id: str
    recipient_email: str
    succeeded: bool
    submitter_ip_address: str | None
    reason: str  # wrong verification code or wrong ticket.


class IMCallbackVerificationEvent(_ImmutableJSONModel):
    # We might need to break the abstration.
    type: Literal[AuditEventType.IM_CALLBACK_VERIFICATION] = AuditEventType.IM_CALLBACK_VERIFICATION

    succeeded: bool
    request_ip_address: str | None
    reason: str  # fail reasons


class ServiceAPIFormSubmissionEvent(_ImmutableJSONModel):
    """Distinguish the approving recipient grant from the Service API caller."""

    # TODO(QuantumGhost): we need to check what other information needed to be stored
    # in the audit event later.

    type: Literal[AuditEventType.FORM_SUBMISSION] = AuditEventType.FORM_SUBMISSION
    recipient_id: str
    end_user_id: str


type AuditEvent = Annotated[
    FormViewEvent | OtpSendEvent | OtpVerificationEvent | IMCallbackVerificationEvent | ServiceAPIFormSubmissionEvent,
    Field(discriminator="type"),
]

AUDIT_EVENT_ADAPTER: TypeAdapter[AuditEvent] = TypeAdapter(AuditEvent)


class HumanInputAuditEvent(DefaultFieldsDCMixin, TypeBase):
    __tablename__ = "hitlv2_audit_events"

    tenant_id: Mapped[str] = mapped_column(StringUUID, nullable=False)
    form_id: Mapped[str] = mapped_column(StringUUID, nullable=False)

    payload: Mapped[AuditEvent] = mapped_column(
        FrozenPydanticModelColumn(
            AUDIT_EVENT_ADAPTER,
            model_types=(
                FormViewEvent,
                OtpSendEvent,
                OtpVerificationEvent,
                IMCallbackVerificationEvent,
                ServiceAPIFormSubmissionEvent,
            ),
        ),
        nullable=False,
    )


__all__ = [
    "ContactSubjectType",
    "HumanInputContactIdentity",
    "HumanInputEmailProvider",
    "HumanInputExternalContactProfile",
    "HumanInputIMBinding",
    "HumanInputIMBindingWorkspaceOverride",
    "HumanInputIMChannel",
    "HumanInputIMIdentity",
    "HumanInputIMReconciliationChange",
    "HumanInputIMSyncResult",
    "HumanInputIMSyncRun",
    "HumanInputPlatformContactWorkspaceEntry",
    "IMBindingReconciliationSnapshot",
    "IMIdentityRawPayload",
    "IMIdentityReconciliationSnapshot",
    "IMMessageEvent",
    "IMReconciliationChangeSnapshot",
    "IMSyncContactSnapshot",
    "IMSyncDirectoryEntryPayload",
    "IMSyncIdentitySnapshot",
]