from datetime import UTC, datetime from secrets import token_urlsafe from typing import Annotated from fastapi import APIRouter, Header, HTTPException, status from sqlalchemy import func, select from sqlalchemy.exc import IntegrityError 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, IntegrationCredential, Lead, LeadSource, Organization, Pipeline, Tenant, ) from app.schemas import ( IntegrationCampaignRequest, IntegrationCampaignResponse, IntegrationCredentialOut, IntegrationKeyCreate, IntegrationKeyResponse, IntegrationLeadRequest, IntegrationLeadResponse, IntegrationStatusResponse, ) from app.services import add_audit, add_event, next_lead_position router = APIRouter(prefix="/integrations", tags=["Integrations"]) @router.get( "/credentials", response_model=list[IntegrationCredentialOut], ) def list_integration_keys( user: Admin, db: Database, ) -> list[IntegrationCredential]: return list( db.scalars( select(IntegrationCredential) .where(IntegrationCredential.tenant_id == user.tenant_id) .order_by(IntegrationCredential.created_at.desc()), ).all(), ) @router.post( "/credentials", response_model=IntegrationKeyResponse, status_code=status.HTTP_201_CREATED, ) def create_integration_key( payload: IntegrationKeyCreate, user: Admin, db: Database, ) -> IntegrationKeyResponse: raw_key = f"mcrm_{token_urlsafe(36)}" credential = IntegrationCredential( tenant_id=user.tenant_id, name=payload.name, key_prefix=raw_key[:12], key_hash=hash_service_key(raw_key), ) db.add(credential) try: db.flush() except IntegrityError as exc: # Credential names are unique per tenant. Reusing one is a client # mistake, not a server fault, so report it as a conflict instead of # letting the constraint surface as a 500. db.rollback() raise HTTPException( status_code=status.HTTP_409_CONFLICT, detail=( f"An integration credential named '{payload.name}' already " "exists. Choose a different name, or delete the existing one." ), ) from exc add_audit( db, actor=user, tenant_id=user.tenant_id, action="integration_credential.created", entity_type="integration_credential", entity_id=credential.id, ) db.commit() db.refresh(credential) return IntegrationKeyResponse( id=credential.id, name=credential.name, key=raw_key, key_prefix=credential.key_prefix, created_at=credential.created_at, ) @router.delete( "/credentials/{credential_id}", status_code=status.HTTP_204_NO_CONTENT, ) def revoke_integration_key( credential_id: str, user: Admin, db: Database, ) -> None: credential = db.scalar( select(IntegrationCredential).where( IntegrationCredential.id == credential_id, IntegrationCredential.tenant_id == user.tenant_id, ), ) if credential is None: raise HTTPException( status_code=status.HTTP_404_NOT_FOUND, detail="The integration credential was not found.", ) if credential.is_active: credential.is_active = False add_audit( db, actor=user, tenant_id=user.tenant_id, action="integration_credential.revoked", entity_type="integration_credential", entity_id=credential.id, ) db.commit() def _authenticate_integration(db: Database, raw_key: str | None) -> IntegrationCredential: if not raw_key: raise HTTPException( status_code=status.HTTP_401_UNAUTHORIZED, detail="An integration key is required.", ) credential = db.scalar( select(IntegrationCredential).where( IntegrationCredential.key_hash == hash_service_key(raw_key), IntegrationCredential.is_active.is_(True), ), ) if credential is None: raise HTTPException( status_code=status.HTTP_401_UNAUTHORIZED, detail="The integration key is invalid.", ) credential.last_used_at = datetime.now(UTC) 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, integration_key: Annotated[str | None, Header(alias="X-Integration-Key")] = None, ) -> IntegrationStatusResponse: credential = _authenticate_integration(db, integration_key) tenant = db.scalar( select(Tenant).where(Tenant.id == credential.tenant_id), ) if tenant is None: raise HTTPException( status_code=status.HTTP_503_SERVICE_UNAVAILABLE, detail="The integration workspace is unavailable.", ) settings = get_settings() response = IntegrationStatusResponse( status="ready", product=settings.app_name, mode=settings.mode, tenant_id=tenant.id, tenant_name=tenant.name, workspace=tenant.slug, credential_id=credential.id, credential_name=credential.name, key_prefix=credential.key_prefix, ) db.commit() return response @router.post("/leads", response_model=IntegrationLeadResponse) def ingest_lead( payload: IntegrationLeadRequest, db: Database, integration_key: Annotated[str | None, Header(alias="X-Integration-Key")] = None, idempotency_key: Annotated[str | None, Header(alias="Idempotency-Key")] = None, ) -> IntegrationLeadResponse: credential = _authenticate_integration(db, integration_key) if not idempotency_key: raise HTTPException( status_code=status.HTTP_400_BAD_REQUEST, detail="Idempotency-Key is required.", ) previous = db.scalar( select(IdempotencyRecord).where( IdempotencyRecord.tenant_id == credential.tenant_id, IdempotencyRecord.scope == "integration.lead", IdempotencyRecord.idempotency_key == idempotency_key, ), ) if previous: return IntegrationLeadResponse.model_validate(previous.response_body) existing_link = db.scalar( select(ExternalLink).where( ExternalLink.tenant_id == credential.tenant_id, ExternalLink.provider == payload.provider, ExternalLink.entity_type == "lead", ExternalLink.external_id == payload.external_id, ), ) if existing_link: existing_lead = db.scalar( select(Lead) .where( Lead.id == existing_link.entity_id, Lead.tenant_id == credential.tenant_id, ) .options(selectinload(Lead.contact)), ) if existing_lead: result = IntegrationLeadResponse( contact_id=existing_lead.contact_id or "", lead_id=existing_lead.id, created=False, ) db.add( IdempotencyRecord( tenant_id=credential.tenant_id, scope="integration.lead", idempotency_key=idempotency_key, response_code=200, response_body=result.model_dump(), ), ) db.commit() return result organization = None if payload.company_name: organization = db.scalar( select(Organization).where( Organization.tenant_id == credential.tenant_id, func.lower(Organization.name) == payload.company_name.lower(), ), ) if organization is None: organization = Organization( tenant_id=credential.tenant_id, name=payload.company_name, ) db.add(organization) db.flush() contact = None if payload.email: contact = db.scalar( select(Contact).where( Contact.tenant_id == credential.tenant_id, func.lower(Contact.primary_email) == payload.email.lower(), ), ) if contact is None: phones = ( [{"label": "work", "value": payload.phone, "primary": True}] if payload.phone else [] ) emails = ( [{"label": "work", "value": str(payload.email), "primary": True}] if payload.email else [] ) contact = Contact( tenant_id=credential.tenant_id, first_name=payload.first_name, last_name=payload.last_name, job_title=payload.job_title, primary_email=str(payload.email) if payload.email else None, emails=emails, phones=phones, lead_source=payload.source, score=payload.score, organization_id=organization.id if organization else None, attributes={"campaign": payload.campaign, **payload.metadata}, ) db.add(contact) db.flush() pipeline = db.scalar( select(Pipeline) .where( Pipeline.tenant_id == credential.tenant_id, Pipeline.is_default.is_(True), ) .options(selectinload(Pipeline.stages)), ) if pipeline is None or not pipeline.stages: raise HTTPException( status_code=status.HTTP_503_SERVICE_UNAVAILABLE, detail="The CRM workspace does not have a default pipeline.", ) source = db.scalar( select(LeadSource).where( LeadSource.tenant_id == credential.tenant_id, func.lower(LeadSource.name) == payload.source.lower(), ), ) if source is None: source = LeadSource(tenant_id=credential.tenant_id, name=payload.source) db.add(source) db.flush() first_stage = sorted(pipeline.stages, key=lambda item: item.position)[0] lead = Lead( tenant_id=credential.tenant_id, title=payload.lead_title or f"{contact.name} opportunity", description="Lead received through MaskanX integration.", score=payload.score, contact_id=contact.id, organization_id=organization.id if organization else None, pipeline_id=pipeline.id, 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) db.flush() db.add( ExternalLink( tenant_id=credential.tenant_id, provider=payload.provider, entity_type="lead", entity_id=lead.id, external_id=payload.external_id, metadata_json=payload.metadata, ), ) add_audit( db, actor=None, tenant_id=credential.tenant_id, action="lead.ingested", entity_type="lead", entity_id=lead.id, payload={"provider": payload.provider, "external_id": payload.external_id}, ) add_event( db, tenant_id=credential.tenant_id, topic="crm.lead.ingested", payload={"lead_id": lead.id, "provider": payload.provider}, ) result = IntegrationLeadResponse( contact_id=contact.id, lead_id=lead.id, created=True, ) db.add( IdempotencyRecord( tenant_id=credential.tenant_id, scope="integration.lead", idempotency_key=idempotency_key, response_code=200, response_body=result.model_dump(), ), ) 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)