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 <noreply@anthropic.com>
This commit is contained in:
AFFAANh
@@ -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()
|
||||
|
||||
@@ -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")
|
||||
@@ -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 == []
|
||||
Reference in New Issue
Block a user