feat(campaigns): report advertising spend and cost per lead

MaskanX runs the campaigns; this stores what they cost and what they
produced so the question "what did this campaign cost us per lead" is
answerable next to the leads themselves.

POST /integrations/campaigns is an upsert keyed on (provider,
external_id), not 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 re-pushes the same campaign as its spend grows. An
Idempotency-Key here would pin the CRM to the first numbers it ever saw.

Money is stored as integers in minor currency units, matching what
MaskanX sends and what Meta uses. A Numeric would add a second convention
and a rounding step between systems that currently agree exactly.

Ad attribution is promoted out of crm_leads.attributes into indexed
columns, so counting leads per campaign is a join rather than a JSON scan
— which also keeps it working on both SQLite and PostgreSQL.

Meta's lead count and the CRM's own are both kept. They routinely differ,
since Meta attributes late and leads can be entered by hand, and the gap
is worth seeing rather than hiding behind one number.

Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>
This commit is contained in:
AFFAANh
2026-08-04 10:55:46 +05:30
co-authored by Claude Opus 5
parent f6a54d6f93
commit 55d5af849b
7 changed files with 586 additions and 1 deletions
@@ -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')
+72
View File
@@ -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
]
+86
View File
@@ -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)
+11 -1
View File
@@ -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)
+70
View File
@@ -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,
)
+61
View File
@@ -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
+192
View File
@@ -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