feat(telephony): add WhatsApp calling integration with outbound Calling (Business-Initiated Calling & Permission Management)

api/alembic/versions/b7d2f04c8a15_index_parked_queued_runs.py

@@ -0,0 +1,36 @@
+"""index parked queued runs for whatsapp permission lookups
+
+Revision ID: b7d2f04c8a15
+Revises: e1a2b3c4d5e6
+Create Date: 2026-09-12 10:00:00.000000
+
+"""
+
+from typing import Sequence, Union
+
+import sqlalchemy as sa
+from alembic import op
+
+# revision identifiers, used by Alembic.
+revision: str = "b7d2f04c8a15"
+down_revision: Union[str, None] = "e1a2b3c4d5e6"
+branch_labels: Union[str, Sequence[str], None] = None
+depends_on: Union[str, Sequence[str], None] = None
+
+
+def upgrade() -> None:
+    # get_queued_runs_awaiting_whatsapp_permission filters on retry_reason with
+    # no campaign_id, so none of the existing (campaign_id, ...) indexes apply.
+    # state = 'queued' is not selective on its own, which left the lookup
+    # scanning queued_runs on every inbound permission webhook.
+    op.create_index(
+        "idx_queued_runs_retry_reason_parked",
+        "queued_runs",
+        ["retry_reason"],
+        unique=False,
+        postgresql_where=sa.text("state = 'queued' AND retry_reason IS NOT NULL"),
+    )
+
+
+def downgrade() -> None:
+    op.drop_index("idx_queued_runs_retry_reason_parked", table_name="queued_runs")

api/alembic/versions/c3f5a1b90d47_index_whatsapp_perm_and_queued_run_lookups.py

@@ -0,0 +1,59 @@
+"""index whatsapp permission message id and workflow_runs queued_run_id
+
+Revision ID: c3f5a1b90d47
+Revises: b7d2f04c8a15
+Create Date: 2026-09-12 12:00:00.000000
+
+"""
+
+from typing import Sequence, Union
+
+import sqlalchemy as sa
+from alembic import op
+
+# revision identifiers, used by Alembic.
+revision: str = "c3f5a1b90d47"
+down_revision: Union[str, None] = "b7d2f04c8a15"
+branch_labels: Union[str, Sequence[str], None] = None
+depends_on: Union[str, Sequence[str], None] = None
+
+
+def upgrade() -> None:
+    # update_whatsapp_call_permission_status_by_message_id resolves a row by
+    # meta_message_id alone on every permission-message status and reply
+    # webhook, and none of the indexes created with the table cover that
+    # column. meta_message_id is set once an outbound request message is sent
+    # and is never cleared afterward, so the partial predicate does not limit
+    # this to outstanding requests -- it only excludes rows that never had a
+    # message id attached. Not unique on purpose: the webhook path should
+    # degrade to an extra row rather than a write failure if Meta ever
+    # replays a wamid.
+    op.create_index(
+        "ix_whatsapp_perm_meta_message_id",
+        "whatsapp_call_permissions",
+        ["meta_message_id"],
+        unique=False,
+        postgresql_where=sa.text("meta_message_id IS NOT NULL"),
+    )
+
+    # workflow_runs.queued_run_id was added as a plain foreign key in
+    # fefdd1835b7d; Postgres does not index the referencing side, so every
+    # per-lead lookup (get_workflow_run_by_queued_run_id and the campaign
+    # client's equivalents) scanned workflow_runs. created_at trails the key so
+    # the index also serves the "latest run" ORDER BY without a sort. Partial
+    # because only campaign-dispatched runs carry a queued_run_id and every
+    # caller uses equality, which never matches NULL.
+    op.create_index(
+        "idx_workflow_runs_queued_run_id",
+        "workflow_runs",
+        ["queued_run_id", "created_at"],
+        unique=False,
+        postgresql_where=sa.text("queued_run_id IS NOT NULL"),
+    )
+
+
+def downgrade() -> None:
+    op.drop_index("idx_workflow_runs_queued_run_id", table_name="workflow_runs")
+    op.drop_index(
+        "ix_whatsapp_perm_meta_message_id", table_name="whatsapp_call_permissions"
+    )

api/alembic/versions/e1a2b3c4d5e6_add_whatsapp_call_permissions.py

@@ -0,0 +1,125 @@
+"""add whatsapp_call_permissions table
+
+Revision ID: e1a2b3c4d5e6
+Revises: f3a1c47b9e02
+Create Date: 2026-09-10 15:00:00.000000
+
+"""
+
+from typing import Sequence, Union
+
+import sqlalchemy as sa
+from alembic import op
+
+# revision identifiers, used by Alembic.
+revision: str = "e1a2b3c4d5e6"
+down_revision: Union[str, None] = "f3a1c47b9e02"
+branch_labels: Union[str, Sequence[str], None] = None
+depends_on: Union[str, Sequence[str], None] = None
+
+
+def upgrade() -> None:
+    op.create_table(
+        "whatsapp_call_permissions",
+        sa.Column(
+            "id", sa.Integer(), primary_key=True, autoincrement=True, nullable=False
+        ),
+        sa.Column("organization_id", sa.Integer(), nullable=False),
+        sa.Column("telephony_configuration_id", sa.Integer(), nullable=False),
+        sa.Column("phone_number_id", sa.String(length=64), nullable=False),
+        sa.Column("recipient_phone_number", sa.String(length=32), nullable=False),
+        sa.Column(
+            "status", sa.String(length=32), nullable=False, server_default="pending"
+        ),
+        sa.Column("permission_type", sa.String(length=32), nullable=True),
+        sa.Column("meta_message_id", sa.String(length=128), nullable=True),
+        sa.Column(
+            "requested_at",
+            sa.DateTime(timezone=True),
+            nullable=False,
+            server_default=sa.func.now(),
+        ),
+        sa.Column("granted_at", sa.DateTime(timezone=True), nullable=True),
+        sa.Column("expires_at", sa.DateTime(timezone=True), nullable=True),
+        sa.Column(
+            "updated_at",
+            sa.DateTime(timezone=True),
+            nullable=False,
+            server_default=sa.func.now(),
+        ),
+        sa.ForeignKeyConstraint(
+            ["organization_id"],
+            ["organizations.id"],
+            ondelete="CASCADE",
+        ),
+        sa.ForeignKeyConstraint(
+            ["telephony_configuration_id"],
+            ["telephony_configurations.id"],
+            ondelete="CASCADE",
+        ),
+        sa.UniqueConstraint(
+            "telephony_configuration_id",
+            "recipient_phone_number",
+            name="uq_whatsapp_perm_config_recipient",
+        ),
+    )
+    op.create_index(
+        "ix_whatsapp_call_permissions_id",
+        "whatsapp_call_permissions",
+        ["id"],
+        unique=False,
+    )
+    op.create_index(
+        "ix_whatsapp_call_permissions_organization_id",
+        "whatsapp_call_permissions",
+        ["organization_id"],
+        unique=False,
+    )
+    op.create_index(
+        "ix_whatsapp_call_permissions_telephony_configuration_id",
+        "whatsapp_call_permissions",
+        ["telephony_configuration_id"],
+        unique=False,
+    )
+    op.create_index(
+        "ix_whatsapp_call_permissions_phone_number_id",
+        "whatsapp_call_permissions",
+        ["phone_number_id"],
+        unique=False,
+    )
+    op.create_index(
+        "ix_whatsapp_call_permissions_recipient_phone_number",
+        "whatsapp_call_permissions",
+        ["recipient_phone_number"],
+        unique=False,
+    )
+    op.create_index(
+        "ix_whatsapp_perm_lookup",
+        "whatsapp_call_permissions",
+        ["phone_number_id", "recipient_phone_number"],
+        unique=False,
+    )
+
+
+def downgrade() -> None:
+    op.drop_index("ix_whatsapp_perm_lookup", table_name="whatsapp_call_permissions")
+    op.drop_index(
+        "ix_whatsapp_call_permissions_recipient_phone_number",
+        table_name="whatsapp_call_permissions",
+    )
+    op.drop_index(
+        "ix_whatsapp_call_permissions_phone_number_id",
+        table_name="whatsapp_call_permissions",
+    )
+    op.drop_index(
+        "ix_whatsapp_call_permissions_telephony_configuration_id",
+        table_name="whatsapp_call_permissions",
+    )
+    op.drop_index(
+        "ix_whatsapp_call_permissions_organization_id",
+        table_name="whatsapp_call_permissions",
+    )
+    op.drop_index(
+        "ix_whatsapp_call_permissions_id", table_name="whatsapp_call_permissions"
+    )
+    op.drop_table("whatsapp_call_permissions")

api/constants.py

@@ -78,6 +78,7 @@
 DOGRAH_MPS_SECRET_KEY = os.getenv("DOGRAH_MPS_SECRET_KEY", None)
 MPS_API_URL = os.getenv("MPS_API_URL", "https://services.dograh.com")
 DOGRAH_DEVOPS_SECRET = os.getenv("DOGRAH_DEVOPS_SECRET") or None
+WHATSAPP_WEBHOOK_VERIFY_TOKEN = os.getenv("WHATSAPP_WEBHOOK_VERIFY_TOKEN") or None
 
 # Storage Configuration
 ENABLE_AWS_S3 = os.getenv("ENABLE_AWS_S3", "false").lower() == "true"