diff --git a/alembic/versions/b1c4d7e29a03_add_crm_campaigns.py b/alembic/versions/b1c4d7e29a03_add_crm_campaigns.py new file mode 100644 index 0000000..f9bebef --- /dev/null +++ b/alembic/versions/b1c4d7e29a03_add_crm_campaigns.py @@ -0,0 +1,94 @@ +"""add crm_campaigns + +Mirrors advertising campaigns pushed from MaskanX so the CRM can report +spend and cost per lead alongside the leads those campaigns produced. + +Money columns are integers in minor currency units (paise, cents), matching +what MaskanX sends and what Meta's budget fields use. A Numeric here would +add a second convention and a rounding step between two systems that +currently agree exactly. + +Revision ID: b1c4d7e29a03 +Revises: eca4681c9f66 +Create Date: 2026-08-03 +""" +from typing import Sequence, Union + +from alembic import op +import sqlalchemy as sa + + +revision: str = 'b1c4d7e29a03' +down_revision: Union[str, None] = 'eca4681c9f66' +branch_labels: Union[str, Sequence[str], None] = None +depends_on: Union[str, Sequence[str], None] = None + + +def upgrade() -> None: + op.create_table( + 'crm_campaigns', + sa.Column('id', sa.String(length=36), nullable=False), + sa.Column('tenant_id', sa.String(length=36), nullable=False), + sa.Column('provider', sa.String(length=80), nullable=False), + sa.Column('external_id', sa.String(length=255), nullable=False), + sa.Column('name', sa.String(length=240), nullable=False), + sa.Column('status', sa.String(length=40), nullable=False), + sa.Column('objective', sa.String(length=80), nullable=True), + sa.Column('channel', sa.String(length=80), nullable=True), + sa.Column('currency', sa.String(length=8), nullable=True), + sa.Column('daily_budget', sa.Integer(), nullable=True), + sa.Column('spend', sa.Integer(), nullable=False), + sa.Column('impressions', sa.Integer(), nullable=False), + sa.Column('clicks', sa.Integer(), nullable=False), + sa.Column('leads', sa.Integer(), nullable=False), + sa.Column('cost_per_lead', sa.Integer(), nullable=True), + sa.Column('metrics_from', sa.String(length=20), nullable=True), + sa.Column('metrics_to', sa.String(length=20), nullable=True), + sa.Column('synced_at', sa.DateTime(timezone=True), nullable=False), + sa.Column('metadata_json', sa.JSON(), nullable=False), + sa.Column('created_at', sa.DateTime(timezone=True), nullable=False), + sa.Column('updated_at', sa.DateTime(timezone=True), nullable=False), + sa.ForeignKeyConstraint(['tenant_id'], ['crm_tenants.id'], ondelete='CASCADE'), + sa.PrimaryKeyConstraint('id'), + # MaskanX re-pushes the same campaign as its figures change, so the + # provider's own id is what identifies it, not our surrogate key. + sa.UniqueConstraint( + 'tenant_id', 'provider', 'external_id', + name='uq_crm_campaigns_provider_external', + ), + ) + op.create_index( + 'ix_crm_campaigns_tenant_status', + 'crm_campaigns', + ['tenant_id', 'status'], + ) + + # Ad attribution promoted out of crm_leads.attributes into real columns, + # so counting leads per campaign is an indexed join rather than a JSON + # scan. Nullable: most leads do not come from an ad. + op.add_column( + 'crm_leads', + sa.Column('campaign_external_id', sa.String(length=255), nullable=True), + ) + op.add_column( + 'crm_leads', + sa.Column('adset_external_id', sa.String(length=255), nullable=True), + ) + op.add_column( + 'crm_leads', + sa.Column('ad_external_id', sa.String(length=255), nullable=True), + ) + op.create_index( + 'ix_crm_leads_campaign_external_id', + 'crm_leads', + ['campaign_external_id'], + ) + + +def downgrade() -> None: + op.drop_index('ix_crm_leads_campaign_external_id', table_name='crm_leads') + op.drop_column('crm_leads', 'ad_external_id') + op.drop_column('crm_leads', 'adset_external_id') + op.drop_column('crm_leads', 'campaign_external_id') + op.drop_index('ix_crm_campaigns_tenant_status', table_name='crm_campaigns') + op.drop_table('crm_campaigns') diff --git a/app/api/campaigns.py b/app/api/campaigns.py new file mode 100644 index 0000000..845df3b --- /dev/null +++ b/app/api/campaigns.py @@ -0,0 +1,72 @@ +"""Advertising campaigns, as mirrored from MaskanX. + +Read-only. The CRM does not run campaigns — MaskanX does, and pushes their +figures here through `POST /integrations/campaigns`. This exists so that +"what did this campaign cost us per lead" can be answered next to the leads +themselves, without leaving the CRM. + +Each campaign carries two lead counts: `leads` as Meta attributes them, and +`crm_leads` as this database actually holds them. They routinely differ — +Meta attributes late, and a lead can be entered here by hand — and the gap +is worth seeing rather than hiding behind one number. +""" + +from fastapi import APIRouter +from sqlalchemy import func, select + +from app.core.security import CurrentUser, Database +from app.models import Campaign, Lead +from app.schemas import CampaignOut + +router = APIRouter(prefix="/campaigns", tags=["Campaigns"]) + + +@router.get("", response_model=list[CampaignOut]) +def list_campaigns(user: CurrentUser, db: Database) -> list[CampaignOut]: + campaigns = list( + db.scalars( + select(Campaign) + .where(Campaign.tenant_id == user.tenant_id) + .order_by(Campaign.spend.desc()), + ), + ) + + # One grouped count rather than a query per campaign: a workspace with + # a hundred campaigns should still be one round trip. + counts = dict( + db.execute( + select( + Lead.campaign_external_id, + func.count(Lead.id), + ) + .where( + Lead.tenant_id == user.tenant_id, + Lead.campaign_external_id.is_not(None), + ) + .group_by(Lead.campaign_external_id), + ).all(), + ) + + return [ + CampaignOut( + id=campaign.id, + provider=campaign.provider, + external_id=campaign.external_id, + name=campaign.name, + status=campaign.status, + objective=campaign.objective, + channel=campaign.channel, + currency=campaign.currency, + daily_budget=campaign.daily_budget, + spend=campaign.spend, + impressions=campaign.impressions, + clicks=campaign.clicks, + leads=campaign.leads, + cost_per_lead=campaign.cost_per_lead, + metrics_from=campaign.metrics_from, + metrics_to=campaign.metrics_to, + synced_at=campaign.synced_at, + crm_leads=counts.get(campaign.external_id, 0), + ) + for campaign in campaigns + ] diff --git a/app/api/integrations.py b/app/api/integrations.py index cfd9eb6..6b704f2 100644 --- a/app/api/integrations.py +++ b/app/api/integrations.py @@ -10,6 +10,7 @@ from sqlalchemy.orm import selectinload from app.core.config import get_settings from app.core.security import Admin, Database, hash_service_key from app.models import ( + Campaign, Contact, ExternalLink, IdempotencyRecord, @@ -21,6 +22,8 @@ from app.models import ( Tenant, ) from app.schemas import ( + IntegrationCampaignRequest, + IntegrationCampaignResponse, IntegrationCredentialOut, IntegrationKeyCreate, IntegrationKeyResponse, @@ -155,6 +158,21 @@ def _authenticate_integration(db: Database, raw_key: str | None) -> IntegrationC return credential +def _attribution(campaign: dict, key: str) -> str | None: + """Read one ad id out of a lead's campaign block. + + `campaign` is free-form JSON from the sender, so a value may be absent, + null, or a number. Anything not usable as an id becomes None rather + than being coerced — a lead attributed to the string "None" would be + worse than one attributed to nothing. + """ + value = campaign.get(key) + if value is None: + return None + text = str(value).strip() + return text or None + + @router.get("/status", response_model=IntegrationStatusResponse) def integration_status( db: Database, @@ -333,6 +351,12 @@ def ingest_lead( stage_id=first_stage.id, source_id=source.id, position=next_lead_position(db, credential.tenant_id, first_stage.id), + # Kept in `attributes` as well, so nothing the sender included is + # lost, but the three ids that identify the ad are promoted to + # columns — that is what makes "leads for this campaign" a join. + campaign_external_id=_attribution(payload.campaign, "campaign_id"), + adset_external_id=_attribution(payload.campaign, "adset_id"), + ad_external_id=_attribution(payload.campaign, "ad_id"), attributes={"campaign": payload.campaign, **payload.metadata}, ) db.add(lead) @@ -378,3 +402,65 @@ def ingest_lead( ) db.commit() return result + + +@router.post("/campaigns", response_model=IntegrationCampaignResponse) +def ingest_campaign( + payload: IntegrationCampaignRequest, + db: Database, + integration_key: Annotated[str | None, Header(alias="X-Integration-Key")] = None, +) -> IntegrationCampaignResponse: + """Create or update one advertising campaign's figures. + + Deliberately an upsert keyed on `(provider, external_id)` rather than an + idempotent create like `/leads`. A lead is an event that happened once; + a campaign's figures change every time they are read, and MaskanX pushes + the same campaign repeatedly as its spend grows. An Idempotency-Key here + would either reject the update or return a stale answer. + """ + credential = _authenticate_integration(db, integration_key) + + campaign = db.scalar( + select(Campaign).where( + Campaign.tenant_id == credential.tenant_id, + Campaign.provider == payload.provider, + Campaign.external_id == payload.external_id, + ), + ) + created = campaign is None + if campaign is None: + campaign = Campaign( + tenant_id=credential.tenant_id, + provider=payload.provider, + external_id=payload.external_id, + ) + db.add(campaign) + + campaign.name = payload.name + campaign.status = payload.status + campaign.objective = payload.objective + campaign.channel = payload.channel + campaign.currency = payload.currency + campaign.daily_budget = payload.daily_budget + campaign.spend = payload.spend + campaign.impressions = payload.impressions + campaign.clicks = payload.clicks + campaign.leads = payload.leads + campaign.cost_per_lead = payload.cost_per_lead + campaign.metrics_from = payload.metrics_from + campaign.metrics_to = payload.metrics_to + campaign.metadata_json = payload.metadata + campaign.synced_at = datetime.now(UTC) + + add_event( + db, + tenant_id=credential.tenant_id, + topic="campaign.synced", + payload={ + "provider": payload.provider, + "external_id": payload.external_id, + "spend": payload.spend, + }, + ) + db.commit() + return IntegrationCampaignResponse(campaign_id=campaign.id, created=created) diff --git a/app/api/router.py b/app/api/router.py index ae6dc62..5a786bf 100644 --- a/app/api/router.py +++ b/app/api/router.py @@ -1,6 +1,15 @@ from fastapi import APIRouter -from app.api import activities, auth, catalog, contacts, dashboard, integrations, leads +from app.api import ( + activities, + auth, + campaigns, + catalog, + contacts, + dashboard, + integrations, + leads, +) api_router = APIRouter(prefix="/api/v1") api_router.include_router(auth.router) @@ -9,5 +18,6 @@ api_router.include_router(contacts.router) api_router.include_router(leads.router) api_router.include_router(activities.router) api_router.include_router(catalog.router) +api_router.include_router(campaigns.router) api_router.include_router(integrations.router) diff --git a/app/models.py b/app/models.py index 1a89599..7be9565 100644 --- a/app/models.py +++ b/app/models.py @@ -279,6 +279,14 @@ class Lead(Base, TimestampMixin): type_id: Mapped[str | None] = mapped_column( ForeignKey("crm_lead_types.id", ondelete="SET NULL"), ) + # Ad attribution, promoted out of `attributes` into real columns. + # Meta stamps every lead-ad submission with the campaign, ad set and ad + # that produced it, so "spend and leads for this ad" is a join rather + # than a JSON scan — which also keeps it working on SQLite and + # PostgreSQL alike. Null for leads that did not come from an ad. + campaign_external_id: Mapped[str | None] = mapped_column(String(255), index=True) + adset_external_id: Mapped[str | None] = mapped_column(String(255)) + ad_external_id: Mapped[str | None] = mapped_column(String(255)) attributes: Mapped[dict[str, Any]] = mapped_column(JSON, default=dict, nullable=False) contact: Mapped[Contact | None] = relationship() @@ -558,3 +566,65 @@ class IdempotencyRecord(Base): nullable=False, ) + + +class Campaign(Base, TimestampMixin): + """An advertising campaign mirrored from MaskanX. + + The CRM does not run campaigns; it reports on them next to the leads + they produced. MaskanX pushes each campaign's figures here so that + "what did this campaign cost us per lead" is answerable without leaving + the CRM. + + Money is stored in **minor** currency units (paise, cents), matching + what MaskanX sends and what Meta's own budget fields use. Storing a + decimal here would introduce a second convention and a rounding step + between two systems that currently agree exactly. + """ + + __tablename__ = "crm_campaigns" + __table_args__ = ( + UniqueConstraint( + "tenant_id", + "provider", + "external_id", + name="uq_crm_campaigns_provider_external", + ), + Index("ix_crm_campaigns_tenant_status", "tenant_id", "status"), + ) + + id: Mapped[str] = mapped_column(String(36), primary_key=True, default=new_id) + tenant_id: Mapped[str] = mapped_column( + ForeignKey("crm_tenants.id", ondelete="CASCADE"), + nullable=False, + ) + # "maskanx" for now; kept general so a second ad platform does not need + # a second table. + provider: Mapped[str] = mapped_column(String(80), nullable=False) + external_id: Mapped[str] = mapped_column(String(255), nullable=False) + name: Mapped[str] = mapped_column(String(240), nullable=False) + status: Mapped[str] = mapped_column(String(40), nullable=False, default="unknown") + objective: Mapped[str | None] = mapped_column(String(80), nullable=True) + channel: Mapped[str | None] = mapped_column(String(80), nullable=True) + currency: Mapped[str | None] = mapped_column(String(8), nullable=True) + + daily_budget: Mapped[int | None] = mapped_column(Integer, nullable=True) + spend: Mapped[int] = mapped_column(Integer, nullable=False, default=0) + impressions: Mapped[int] = mapped_column(Integer, nullable=False, default=0) + clicks: Mapped[int] = mapped_column(Integer, nullable=False, default=0) + # Leads as Meta counts them. The CRM's own lead count can differ — a + # lead may arrive here that Meta has not attributed yet, or vice versa — + # so both are kept rather than one being derived from the other. + leads: Mapped[int] = mapped_column(Integer, nullable=False, default=0) + cost_per_lead: Mapped[int | None] = mapped_column(Integer, nullable=True) + + metrics_from: Mapped[str | None] = mapped_column(String(20), nullable=True) + metrics_to: Mapped[str | None] = mapped_column(String(20), nullable=True) + synced_at: Mapped[datetime] = mapped_column( + DateTime(timezone=True), + default=now_utc, + nullable=False, + ) + metadata_json: Mapped[dict[str, Any]] = mapped_column( + JSON, default=dict, nullable=False, + ) diff --git a/app/schemas.py b/app/schemas.py index 9ba1ad4..f2b1b9e 100644 --- a/app/schemas.py +++ b/app/schemas.py @@ -409,3 +409,64 @@ class IntegrationStatusResponse(ApiModel): credential_id: str credential_name: str key_prefix: str + + +class IntegrationCampaignRequest(ApiModel): + """One advertising campaign's current figures, pushed from MaskanX. + + Money is in **minor** currency units (paise, cents), matching what Meta + and MaskanX both use. Sending a decimal would introduce a second + convention and a rounding step between systems that agree exactly. + """ + + provider: str = Field(default="maskanx", min_length=2, max_length=80) + external_id: str = Field(min_length=1, max_length=255) + name: str = Field(min_length=1, max_length=240) + status: str = Field(default="unknown", max_length=40) + objective: str | None = Field(default=None, max_length=80) + channel: str | None = Field(default=None, max_length=80) + currency: str | None = Field(default=None, max_length=8) + + daily_budget: int | None = Field(default=None, ge=0) + spend: int = Field(default=0, ge=0) + impressions: int = Field(default=0, ge=0) + clicks: int = Field(default=0, ge=0) + leads: int = Field(default=0, ge=0) + # Sent rather than derived so the CRM shows exactly the figure MaskanX + # shows. Recomputing it here from spend and leads would drift whenever + # the two systems rounded differently. + cost_per_lead: int | None = Field(default=None, ge=0) + + metrics_from: str | None = Field(default=None, max_length=20) + metrics_to: str | None = Field(default=None, max_length=20) + metadata: dict[str, Any] = Field(default_factory=dict) + + +class IntegrationCampaignResponse(ApiModel): + campaign_id: str + created: bool + + +class CampaignOut(ApiModel): + id: str + provider: str + external_id: str + name: str + status: str + objective: str | None + channel: str | None + currency: str | None + daily_budget: int | None + spend: int + impressions: int + clicks: int + leads: int + cost_per_lead: int | None + metrics_from: str | None + metrics_to: str | None + synced_at: datetime + # Leads the CRM itself holds for this campaign. Kept alongside `leads` + # (Meta's count) rather than replacing it: a lead can exist here that + # Meta has not attributed yet, and the gap between the two is worth + # seeing rather than hiding. + crm_leads: int = 0 diff --git a/tests/test_campaigns.py b/tests/test_campaigns.py new file mode 100644 index 0000000..29bb6b0 --- /dev/null +++ b/tests/test_campaigns.py @@ -0,0 +1,192 @@ +"""Advertising campaigns mirrored from MaskanX. + +The CRM does not run campaigns; MaskanX pushes their figures here so that +spend and cost per lead can be read next to the leads they produced. + +The behaviour that matters is that pushing the same campaign again updates +it rather than duplicating it — MaskanX re-pushes every few minutes as +spend grows — and that lead attribution actually links the two. +""" +from fastapi.testclient import TestClient +import pytest + + +@pytest.fixture() +def service_key(client: TestClient, auth_headers: dict[str, str]) -> str: + created = client.post( + "/api/v1/integrations/credentials", + headers=auth_headers, + json={"name": "MaskanX"}, + ) + assert created.status_code == 201, created.text + return created.json()["key"] + + +def _campaign(**overrides) -> dict: + payload = { + "provider": "maskanx", + "external_id": "camp_1", + "name": "Q3 lead gen", + "status": "live", + "objective": "OUTCOME_LEADS", + "channel": "facebook", + "currency": "INR", + "daily_budget": 50000, + "spend": 46000, + "impressions": 12000, + "clicks": 300, + "leads": 4, + "cost_per_lead": 11500, + "metrics_from": "2026-07-05", + "metrics_to": "2026-08-03", + } + payload.update(overrides) + return payload + + +def _push(client, service_key, **overrides): + return client.post( + "/api/v1/integrations/campaigns", + headers={"X-Integration-Key": service_key}, + json=_campaign(**overrides), + ) + + +def test_a_campaign_can_be_pushed_and_read_back( + client: TestClient, auth_headers: dict, service_key: str, +) -> None: + pushed = _push(client, service_key) + assert pushed.status_code == 200, pushed.text + assert pushed.json()["created"] is True + + listed = client.get("/api/v1/campaigns", headers=auth_headers) + assert listed.status_code == 200, listed.text + body = listed.json() + assert len(body) == 1 + assert body[0]["name"] == "Q3 lead gen" + assert body[0]["spend"] == 46000 + assert body[0]["cost_per_lead"] == 11500 + + +def test_pushing_the_same_campaign_again_updates_it( + client: TestClient, auth_headers: dict, service_key: str, +) -> None: + """MaskanX re-pushes every few minutes as spend grows.""" + _push(client, service_key, spend=46000) + again = _push(client, service_key, spend=90000, leads=9, cost_per_lead=10000) + + assert again.json()["created"] is False + + body = client.get("/api/v1/campaigns", headers=auth_headers).json() + assert len(body) == 1 + assert body[0]["spend"] == 90000 + assert body[0]["leads"] == 9 + + +def test_two_campaigns_are_kept_apart( + client: TestClient, auth_headers: dict, service_key: str, +) -> None: + _push(client, service_key, external_id="camp_1", name="First") + _push(client, service_key, external_id="camp_2", name="Second") + + body = client.get("/api/v1/campaigns", headers=auth_headers).json() + + assert {row["name"] for row in body} == {"First", "Second"} + + +def test_campaigns_are_ordered_by_spend( + client: TestClient, auth_headers: dict, service_key: str, +) -> None: + _push(client, service_key, external_id="small", name="Small", spend=1000) + _push(client, service_key, external_id="big", name="Big", spend=99000) + + body = client.get("/api/v1/campaigns", headers=auth_headers).json() + + assert [row["name"] for row in body] == ["Big", "Small"] + + +def test_pushing_a_campaign_needs_an_integration_key(client: TestClient) -> None: + response = client.post("/api/v1/integrations/campaigns", json=_campaign()) + + assert response.status_code == 401 + + +def test_reading_campaigns_needs_a_logged_in_user(client: TestClient) -> None: + assert client.get("/api/v1/campaigns").status_code == 401 + + +def test_negative_spend_is_rejected(client: TestClient, service_key: str) -> None: + """Money that went backwards is a bug upstream, not a figure to store.""" + assert _push(client, service_key, spend=-1).status_code == 422 + + +# --- attribution --- + + +def _lead(client, service_key, external_id, campaign_id): + return client.post( + "/api/v1/integrations/leads", + headers={ + "X-Integration-Key": service_key, + "Idempotency-Key": f"key-{external_id}", + }, + json={ + "provider": "maskanx", + "external_id": external_id, + "first_name": "Asha", + "last_name": "Menon", + "email": f"{external_id}@example.com", + "campaign": { + "campaign_id": campaign_id, + "adset_id": "meta_set_1", + "ad_id": "meta_ad_1", + }, + }, + ) + + +def test_leads_are_counted_against_the_campaign_that_produced_them( + client: TestClient, auth_headers: dict, service_key: str, +) -> None: + _push(client, service_key, external_id="camp_1", leads=9) + assert _lead(client, service_key, "lead_1", "camp_1").status_code == 200 + assert _lead(client, service_key, "lead_2", "camp_1").status_code == 200 + + body = client.get("/api/v1/campaigns", headers=auth_headers).json() + + # Meta's count and the CRM's own count are kept separately: they + # routinely differ, and the gap is worth seeing. + assert body[0]["leads"] == 9 + assert body[0]["crm_leads"] == 2 + + +def test_a_lead_from_another_campaign_is_not_counted( + client: TestClient, auth_headers: dict, service_key: str, +) -> None: + _push(client, service_key, external_id="camp_1") + _lead(client, service_key, "lead_1", "camp_2") + + body = client.get("/api/v1/campaigns", headers=auth_headers).json() + + assert body[0]["crm_leads"] == 0 + + +def test_a_lead_with_no_campaign_does_not_break_the_count( + client: TestClient, auth_headers: dict, service_key: str, +) -> None: + """Most leads do not come from an ad.""" + _push(client, service_key, external_id="camp_1") + response = client.post( + "/api/v1/integrations/leads", + headers={"X-Integration-Key": service_key, "Idempotency-Key": "manual-1"}, + json={ + "provider": "maskanx", + "external_id": "walk_in_1", + "first_name": "Walk", + "last_name": "In", + }, + ) + assert response.status_code == 200 + + body = client.get("/api/v1/campaigns", headers=auth_headers).json() + assert body[0]["crm_leads"] == 0