From 2522701ec4886893396fa64a2dbd940490f93c4a Mon Sep 17 00:00:00 2001 From: AFFAANh Date: Mon, 3 Aug 2026 16:50:16 +0530 Subject: [PATCH] feat(campaigns): reconcile local records with Meta on a background loop Meta publishes no webhooks for campaign create or delete, so this polls every 120s by default (MASKANX_CAMPAIGN_RECONCILE_SECONDS; 0 disables). Meta is authoritative for delivery state; MaskanX keeps its own metadata. Guardrails, approvals and audit history are never touched by reconciliation, and campaigns still in a local-only status are skipped entirely so a draft that has never reached Meta cannot be overwritten. A campaign that disappears from Meta is archived, not deleted: its spend history has to stay reportable. The loop swallows and logs a failed cycle rather than dying, and is cancelled on shutdown alongside the watchdog. It only starts when META_ADS_ACCOUNT_ID is set. Co-Authored-By: Claude Opus 5 --- src/adclaw/app/_app.py | 25 +++++ src/adclaw/campaigns/reconcile.py | 159 +++++++++++++++++++++++++++ tests/test_campaign_reconcile.py | 173 ++++++++++++++++++++++++++++++ 3 files changed, 357 insertions(+) create mode 100644 src/adclaw/campaigns/reconcile.py create mode 100644 tests/test_campaign_reconcile.py diff --git a/src/adclaw/app/_app.py b/src/adclaw/app/_app.py index e7be1a7..b52cb9b 100644 --- a/src/adclaw/app/_app.py +++ b/src/adclaw/app/_app.py @@ -259,6 +259,29 @@ async def lifespan(app: FastAPI): # pylint: disable=too-many-statements watchdog_task = asyncio.create_task(watchdog.start()) app.state.watchdog = watchdog + # --- Campaign reconciliation (Meta has no webhooks for ad objects) --- + reconcile_task = None + reconcile_account = (os.environ.get("META_ADS_ACCOUNT_ID") or "").strip() + if reconcile_account: + from ..campaigns.reconcile import reconcile_interval_seconds, reconcile_loop + + if reconcile_interval_seconds() > 0: + from ..campaigns.repo import CampaignRepository + from ..meta.client import MetaClient, access_token_from_env + + def _meta_client(): + return MetaClient(access_token=access_token_from_env()) + + reconcile_task = asyncio.create_task( + reconcile_loop(CampaignRepository, _meta_client, reconcile_account), + name="campaign_reconcile", + ) + app.state.reconcile_task = reconcile_task + else: + logger.debug( + "META_ADS_ACCOUNT_ID not set; campaign reconciliation not started", + ) + try: if mcp_initial_config is not None: mcp_init_task = _schedule_mcp_initialization( @@ -270,6 +293,8 @@ async def lifespan(app: FastAPI): # pylint: disable=too-many-statements finally: if hasattr(app.state, "watchdog"): app.state.watchdog.stop() + if reconcile_task is not None: + reconcile_task.cancel() # stop order: watchers -> cron -> channels -> mcp -> runner try: await config_watcher.stop() diff --git a/src/adclaw/campaigns/reconcile.py b/src/adclaw/campaigns/reconcile.py new file mode 100644 index 0000000..46a8716 --- /dev/null +++ b/src/adclaw/campaigns/reconcile.py @@ -0,0 +1,159 @@ +# -*- coding: utf-8 -*- +"""Keep local campaign records in step with Meta. + +Meta publishes no webhooks for campaign create or delete, so this polls. + +Conflict rule, from the design spec: + Meta is authoritative for delivery state (status). + MaskanX is authoritative for its own metadata (guardrails, approvals, + audit history), which this loop never touches. + +A campaign that disappears from Meta is archived, never hard-deleted: its +spend history has to stay reportable. +""" +from __future__ import annotations + +import asyncio +import logging +import os +from dataclasses import dataclass, field +from datetime import datetime, timezone + +from .adopt import local_status_for, spec_from_meta + +logger = logging.getLogger(__name__) + +INTERVAL_ENV = "MASKANX_CAMPAIGN_RECONCILE_SECONDS" +DEFAULT_INTERVAL_SECONDS = 120.0 + +# Statuses MaskanX owns outright. A local campaign that has not reached Meta +# yet must not be touched by reconciliation. +_LOCAL_ONLY_STATUSES = frozenset( + {"draft", "pending_approval", "approved", "archived"}, +) + + +@dataclass +class ReconcileResult: + imported: list[str] = field(default_factory=list) + status_changed: list[str] = field(default_factory=list) + archived: list[str] = field(default_factory=list) + + @property + def changed(self) -> bool: + return bool(self.imported or self.status_changed or self.archived) + + +def reconcile_interval_seconds() -> float: + """Return the poll interval; 0 or less disables the loop.""" + raw = os.environ.get(INTERVAL_ENV) + if raw is None: + return DEFAULT_INTERVAL_SECONDS + try: + return max(0.0, float(raw)) + except ValueError: + logger.warning( + "%s is not a number (%r); using the default of %.0fs", + INTERVAL_ENV, + raw, + DEFAULT_INTERVAL_SECONDS, + ) + return DEFAULT_INTERVAL_SECONDS + + +async def reconcile_once(repo, meta, ad_account_id: str) -> ReconcileResult: + """Bring local records in step with one ad account. Returns what changed.""" + result = ReconcileResult() + + remote = await meta.list_campaigns(ad_account_id) + remote_by_id = {c["id"]: c for c in remote if c.get("id")} + + local = await repo.list_campaigns() + local_by_meta_id = {c.meta_campaign_id: c for c in local if c.meta_campaign_id} + + # Campaigns that exist in Meta but not here. + for meta_id, remote_campaign in remote_by_id.items(): + if meta_id in local_by_meta_id: + continue + spec = spec_from_meta(remote_campaign, ad_account_id) + created = await repo.create_campaign(spec) + await repo.add_event( + created.id, + event_type="campaign.reconciled_import", + reason=f"Discovered in Meta as {meta_id}", + ) + result.imported.append(created.id) + + for meta_id, campaign in local_by_meta_id.items(): + if campaign.ad_account_id != ad_account_id: + continue + + remote_campaign = remote_by_id.get(meta_id) + + if remote_campaign is None: + # Gone from Meta. Archive rather than delete: the insight history + # attached to this campaign must stay reportable. + if campaign.status == "archived": + continue + campaign.status = "archived" + campaign.advanced["deleted_in_meta_at"] = datetime.now( + timezone.utc, + ).isoformat() + await repo.update_campaign_with_event( + campaign, + event_type="campaign.deleted_in_meta", + reason="No longer present in the Meta ad account", + ) + result.archived.append(campaign.id) + continue + + # Meta owns delivery state. Local-only statuses are left alone: a + # draft has never been near Meta, and an archived campaign is done. + if campaign.status in _LOCAL_ONLY_STATUSES: + continue + + remote_status = local_status_for(remote_campaign.get("status")) + if remote_status != campaign.status: + previous = campaign.status + campaign.status = remote_status + await repo.update_campaign_with_event( + campaign, + event_type="campaign.status_reconciled", + reason=f"Meta reports '{remote_campaign.get('status')}' " + f"(was '{previous}')", + ) + result.status_changed.append(campaign.id) + + return result + + +async def reconcile_loop(repo_factory, meta_factory, ad_account_id: str) -> None: + """Poll Meta forever. Never raises: a bad cycle is logged and retried.""" + interval = reconcile_interval_seconds() + if interval <= 0: + logger.info("Campaign reconciliation disabled (%s=0)", INTERVAL_ENV) + return + + logger.info( + "Campaign reconciliation started for %s every %.0fs", + ad_account_id, + interval, + ) + while True: + try: + await asyncio.sleep(interval) + result = await reconcile_once(repo_factory(), meta_factory(), ad_account_id) + if result.changed: + logger.info( + "Reconciled %s: %d imported, %d status changes, %d archived", + ad_account_id, + len(result.imported), + len(result.status_changed), + len(result.archived), + ) + except asyncio.CancelledError: + logger.info("Campaign reconciliation stopped") + raise + except Exception: + # A failed cycle must not kill the loop; the next one retries. + logger.exception("Campaign reconciliation cycle failed") diff --git a/tests/test_campaign_reconcile.py b/tests/test_campaign_reconcile.py new file mode 100644 index 0000000..8a3ae7e --- /dev/null +++ b/tests/test_campaign_reconcile.py @@ -0,0 +1,173 @@ +# -*- coding: utf-8 -*- +"""Two-way reconciliation between local records and Meta.""" +import pytest + +from adclaw.campaigns.models import CampaignSpec +from adclaw.campaigns.reconcile import ( + DEFAULT_INTERVAL_SECONDS, + reconcile_interval_seconds, + reconcile_once, +) + + +class FakeRepo: + def __init__(self, campaigns=None): + self.items: dict[str, CampaignSpec] = {c.id: c for c in (campaigns or [])} + self.events: list[dict] = [] + + async def list_campaigns(self, company_id=None, status=None): + return list(self.items.values()) + + async def create_campaign(self, spec): + self.items[spec.id] = spec + return spec + + async def update_campaign_with_event( + self, spec, event_type, actor=None, reason=None, payload=None, + ): + self.items[spec.id] = spec + self.events.append({"id": spec.id, "event_type": event_type}) + return spec + + async def add_event( + self, campaign_id, event_type, actor=None, reason=None, payload=None, + ): + self.events.append({"id": campaign_id, "event_type": event_type}) + + +class FakeMeta: + def __init__(self, campaigns): + self.campaigns = campaigns + + async def list_campaigns(self, ad_account_id): + return self.campaigns + + +def _local(**overrides) -> CampaignSpec: + data = { + "id": "camp_1", + "name": "Local", + "status": "live", + "origin": "maskanx", + "ad_account_id": "act_1", + "meta_campaign_id": "meta_1", + "guardrails": {"max_cost_per_lead": 150}, + "approved_by": "owner", + } + data.update(overrides) + return CampaignSpec(**data) + + +def test_interval_defaults_when_unset(monkeypatch): + monkeypatch.delenv("MASKANX_CAMPAIGN_RECONCILE_SECONDS", raising=False) + assert reconcile_interval_seconds() == DEFAULT_INTERVAL_SECONDS + + +def test_interval_zero_disables_the_loop(monkeypatch): + monkeypatch.setenv("MASKANX_CAMPAIGN_RECONCILE_SECONDS", "0") + assert reconcile_interval_seconds() == 0 + + +def test_interval_falls_back_on_a_bad_value(monkeypatch): + monkeypatch.setenv("MASKANX_CAMPAIGN_RECONCILE_SECONDS", "not-a-number") + assert reconcile_interval_seconds() == DEFAULT_INTERVAL_SECONDS + + +@pytest.mark.asyncio +async def test_unknown_meta_campaign_is_imported(): + repo = FakeRepo() + meta = FakeMeta([{"id": "meta_9", "name": "Remote", "status": "ACTIVE"}]) + + result = await reconcile_once(repo, meta, "act_1") + + assert len(result.imported) == 1 + imported = repo.items[result.imported[0]] + assert imported.origin == "imported" + assert imported.meta_campaign_id == "meta_9" + assert imported.status == "live" + + +@pytest.mark.asyncio +async def test_status_change_in_meta_updates_the_local_record(): + repo = FakeRepo([_local(status="live")]) + meta = FakeMeta([{"id": "meta_1", "name": "Local", "status": "PAUSED"}]) + + result = await reconcile_once(repo, meta, "act_1") + + assert result.status_changed == ["camp_1"] + assert repo.items["camp_1"].status == "paused" + assert repo.events[-1]["event_type"] == "campaign.status_reconciled" + + +@pytest.mark.asyncio +async def test_campaign_deleted_in_meta_is_archived_not_removed(): + """Spend history must stay reportable, so the record survives.""" + repo = FakeRepo([_local(status="live")]) + meta = FakeMeta([]) + + result = await reconcile_once(repo, meta, "act_1") + + assert result.archived == ["camp_1"] + assert "camp_1" in repo.items + assert repo.items["camp_1"].status == "archived" + assert repo.items["camp_1"].advanced["deleted_in_meta_at"] + assert repo.events[-1]["event_type"] == "campaign.deleted_in_meta" + + +@pytest.mark.asyncio +async def test_already_archived_campaign_is_not_rearchived(): + repo = FakeRepo([_local(status="archived")]) + meta = FakeMeta([]) + + result = await reconcile_once(repo, meta, "act_1") + + assert result.archived == [] + assert repo.events == [] + + +@pytest.mark.asyncio +async def test_reconciliation_never_touches_maskanx_owned_metadata(): + """Meta owns delivery state; guardrails and approvals stay ours.""" + repo = FakeRepo([_local(status="live")]) + meta = FakeMeta([{"id": "meta_1", "name": "Renamed in Meta", "status": "PAUSED"}]) + + await reconcile_once(repo, meta, "act_1") + + campaign = repo.items["camp_1"] + assert campaign.guardrails == {"max_cost_per_lead": 150} + assert campaign.approved_by == "owner" + + +@pytest.mark.parametrize("status", ["draft", "pending_approval", "approved"]) +@pytest.mark.asyncio +async def test_local_only_statuses_are_left_alone(status): + """A campaign that never reached Meta must not be reconciled.""" + repo = FakeRepo([_local(status=status)]) + meta = FakeMeta([{"id": "meta_1", "name": "Local", "status": "ACTIVE"}]) + + result = await reconcile_once(repo, meta, "act_1") + + assert result.status_changed == [] + assert repo.items["camp_1"].status == status + + +@pytest.mark.asyncio +async def test_campaigns_from_other_ad_accounts_are_ignored(): + repo = FakeRepo([_local(ad_account_id="act_other")]) + meta = FakeMeta([]) + + result = await reconcile_once(repo, meta, "act_1") + + assert result.archived == [] + assert repo.items["camp_1"].status == "live" + + +@pytest.mark.asyncio +async def test_matching_status_produces_no_write(): + repo = FakeRepo([_local(status="live")]) + meta = FakeMeta([{"id": "meta_1", "name": "Local", "status": "ACTIVE"}]) + + result = await reconcile_once(repo, meta, "act_1") + + assert result.changed is False + assert repo.events == []