maskantech/maskanx_cm_backend
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

perf(crm): track a lead cursor instead of re-sending the whole window

Browse Source 
Every lead from the last seven days was re-sent on every cycle. It was
correct — the CRM deduplicates on Meta's lead id — but at a five-minute
interval a campaign with fifty leads a week meant fourteen thousand CRM
requests a day to re-send leads the CRM already had.

Each campaign now records how far its leads have been sent, and the next
run reads from there minus an hour. The overlap matters: Meta filters on
creation time, so without it a lead created moments before a run started
falls between two runs and is never sent.

The cursor advances only after every lead in the batch has been accepted,
so a run that fails half-way is retried in full by the next one — the
self-healing the naive version had, without the traffic. Nothing is
written when there were no leads, since that would mean an event row per
campaign every five minutes.

Also batches insight upserts into one executemany. A single campaign sync
writes a month of days times six breakdowns, which was a few hundred
round trips per campaign per cycle.

Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>
This commit is contained in:
AFFAANh
2026-08-04 11:11:11 +05:30
co-authored by Claude Opus 5
parent b63b852cbe
commit 1ea73c1621
3 changed files with 179 additions and 49 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/adclaw/campaigns/crm_sync.py
+62 -11
View File
@@ -49,10 +49,14 @@ LEAD_WINDOW_DAYS_ENV = "MASKANX_CRM_LEAD_WINDOW_DAYS"
# history every five minutes.
DEFAULT_LEAD_WINDOW_DAYS = 7
# Where the last successful push is recorded on the campaign, so the next
# run can report what changed rather than what exists.
# Where the lead cursor is recorded on the campaign.
CRM_STATE_KEY = "crm_sync"
# How far back before the cursor to re-read. Covers a lead created moments
# before a run started: Meta filters on creation time, so without an
# overlap such a lead falls between two runs and is never sent.
LEAD_OVERLAP = timedelta(hours=1)
@dataclass
class CRMSyncResult:
@@ -98,18 +102,45 @@ async def _in_thread(func, *args, **kwargs):
return await loop.run_in_executor(None, lambda: func(*args, **kwargs))
async def push_campaign_leads(crm, meta, campaign) -> int:
"""Send this campaign's recent leads to the CRM. Returns how many.
def _lead_cursor(campaign) -> datetime:
"""When leads were last successfully sent for this campaign.
Every lead in the window is re-sent, not only new ones. The CRM
deduplicates on Meta's lead id, so this costs a request and gains
self-healing: a run that failed half-way is fixed by the next one
without any bookkeeping of where it stopped.
Falls back to the full window for a campaign never synced before, so a
campaign adopted today still brings its recent leads across.
"""
state = campaign.advanced.get(CRM_STATE_KEY)
raw = state.get("leads_synced_through") if isinstance(state, dict) else None
if raw:
try:
return datetime.fromisoformat(raw)
except ValueError:
pass
return datetime.now(timezone.utc) - timedelta(days=lead_window_days())
async def push_campaign_leads(repo, crm, meta, campaign) -> int:
"""Send this campaign's new leads to the CRM. Returns how many.
Reads from where the last successful run finished, minus an overlap.
Re-reading the whole window every time would be correct — the CRM
deduplicates on Meta's lead id — but at a five-minute interval it means
thousands of requests a day to re-send leads the CRM already has.
The cursor only advances after every lead in the batch has been
accepted, so a run that fails half-way is retried in full by the next
one. That is what keeps the self-healing the naive version had, without
the traffic.
The overlap covers a lead created moments before a run started: Meta
filters on creation time, and without it such a lead would fall between
two runs and never be sent.
"""
if not campaign.meta_campaign_id:
return 0
since = datetime.now(timezone.utc) - timedelta(days=lead_window_days())
since = _lead_cursor(campaign) - LEAD_OVERLAP
started = datetime.now(timezone.utc)
leads = await meta.get_leads(
campaign.meta_campaign_id,
since=str(int(since.timestamp())),
@@ -123,8 +154,28 @@ async def push_campaign_leads(crm, meta, campaign) -> int:
# it would create a duplicate on every run.
continue
payload = crm_payload(lead, campaign_name=campaign.name)
await _in_thread(crm.create_lead, payload, idempotency_key=f"meta-lead-{lead_id}")
await _in_thread(
crm.create_lead, payload, idempotency_key=f"meta-lead-{lead_id}",
)
sent += 1
if not sent:
# Nothing to record, and writing anyway would mean an event row per
# campaign every five minutes. With no cursor stored the fallback
# window slides forward on its own, so nothing is lost.
return 0
# Stamped with when the read started, not when it finished: a lead
# created during the run would otherwise be skipped next time.
state = dict(campaign.advanced.get(CRM_STATE_KEY) or {})
state["leads_synced_through"] = started.isoformat()
campaign.advanced[CRM_STATE_KEY] = state
await repo.update_campaign_with_event(
campaign,
event_type="campaign.crm_leads_synced",
actor="crm-sync",
payload={"leads_sent": sent},
)
return sent
@@ -182,7 +233,7 @@ async def sync_to_crm(repo, insights_repo, meta, crm) -> CRMSyncResult:
try:
if await push_campaign_metrics(crm, insights_repo, campaign):
result.campaigns_sent += 1
result.leads_sent += await push_campaign_leads(crm, meta, campaign)
result.leads_sent += await push_campaign_leads(repo, crm, meta, campaign)
except Exception as exc:
# One campaign failing must not stop the rest reaching the CRM,
# for the same reason it does not in guardrail enforcement.
src/adclaw/campaigns/insights_repo.py
+35 -25
View File
@@ -51,36 +51,46 @@ class InsightsRepository:
if not rows:
return 0
# One executemany rather than a loop of execute: a single campaign
# sync writes a month of days times six breakdowns, so a round trip
# per row is a few hundred per campaign per cycle.
params = [
(
f"ins_{uuid.uuid4().hex[:16]}",
row["level"],
row["object_id"],
row.get("campaign_id"),
row["date"],
row.get("breakdown_key") or NO_BREAKDOWN,
row.get("breakdown_value") or NO_BREAKDOWN,
jsonb(row.get("metrics") or {}),
)
for row in rows
]
conn = await connect_database()
try:
async with conn.cursor() as cur:
for row in rows:
await cur.execute(
"""
INSERT INTO maskanx_campaign_insights (
id, level, object_id, campaign_id, date,
breakdown_key, breakdown_value, metrics, fetched_at
)
VALUES (%s, %s, %s, %s, %s, %s, %s, %s, NOW())
ON CONFLICT (level, object_id, date, breakdown_key,
breakdown_value)
DO UPDATE SET
metrics = EXCLUDED.metrics,
campaign_id = EXCLUDED.campaign_id,
fetched_at = NOW()
""",
(
f"ins_{uuid.uuid4().hex[:16]}",
row["level"],
row["object_id"],
row.get("campaign_id"),
row["date"],
row.get("breakdown_key") or NO_BREAKDOWN,
row.get("breakdown_value") or NO_BREAKDOWN,
jsonb(row.get("metrics") or {}),
),
await cur.executemany(
"""
INSERT INTO maskanx_campaign_insights (
id, level, object_id, campaign_id, date,
breakdown_key, breakdown_value, metrics, fetched_at
)
VALUES (%s, %s, %s, %s, %s, %s, %s, %s, NOW())
ON CONFLICT (level, object_id, date, breakdown_key,
breakdown_value)
DO UPDATE SET
metrics = EXCLUDED.metrics,
campaign_id = EXCLUDED.campaign_id,
fetched_at = NOW()
""",
params,
)
await conn.commit()
except Exception:
await conn.rollback()
raise
finally:
await conn.close()
return len(rows)
tests/test_campaign_crm_sync.py
+82 -13
View File
@@ -6,6 +6,8 @@ half-failed run heals itself, a lead without Meta's id is never sent (it
would duplicate on every run), campaign figures come from stored insights
rather than Meta, and one campaign failing does not stop the rest.
"""
from datetime import datetime, timedelta, timezone
import pytest
from adclaw.campaigns.crm_sync import (
@@ -64,10 +66,17 @@ class FakeInsightsRepo:
class FakeRepo:
def __init__(self, campaigns=()):
self.items = list(campaigns)
self.events: list[dict] = []
async def list_campaigns(self, company_id=None, status=None):
return list(self.items)
async def update_campaign_with_event(
self, spec, event_type, actor=None, reason=None, payload=None,
):
self.events.append({"event_type": event_type, "payload": payload})
return spec
def _campaign(**overrides) -> CampaignSpec:
data = {
@@ -125,9 +134,10 @@ def test_the_lead_window_is_never_shorter_than_a_day(monkeypatch):
async def test_each_lead_is_sent_keyed_on_metas_lead_id():
"""That key is what lets the CRM ignore a lead it already has."""
repo = FakeRepo()
crm, meta = FakeCRM(), FakeMeta([_meta_lead("lead_1"), _meta_lead("lead_2")])
sent = await push_campaign_leads(crm, meta, _campaign())
sent = await push_campaign_leads(repo, crm, meta, _campaign())
assert sent == 2
assert [key for _, key in crm.leads] == ["meta-lead-lead_1", "meta-lead-lead_2"]
@@ -136,7 +146,7 @@ async def test_each_lead_is_sent_keyed_on_metas_lead_id():
async def test_the_lead_carries_its_campaign_attribution():
crm, meta = FakeCRM(), FakeMeta([_meta_lead()])
await push_campaign_leads(crm, meta, _campaign())
await push_campaign_leads(FakeRepo(), crm, meta, _campaign())
payload, _ = crm.leads[0]
assert payload["campaign"]["ad_id"] == "meta_ad_1"
@@ -148,26 +158,85 @@ async def test_a_lead_without_an_id_is_never_sent():
"""With no stable external id it would duplicate on every run."""
crm, meta = FakeCRM(), FakeMeta([_meta_lead(lead_id="")])
assert await push_campaign_leads(crm, meta, _campaign()) == 0
assert await push_campaign_leads(FakeRepo(), crm, meta, _campaign()) == 0
assert crm.leads == []
async def test_leads_are_requested_from_a_trailing_window():
crm, meta = FakeCRM(), FakeMeta()
await push_campaign_leads(crm, meta, _campaign())
assert meta.calls[0]["object_id"] == "meta_camp_1"
assert meta.calls[0]["since"] is not None
async def test_an_unsynced_campaign_asks_meta_for_nothing():
crm, meta = FakeCRM(), FakeMeta()
assert await push_campaign_leads(crm, meta, _campaign(meta_campaign_id=None)) == 0
sent = await push_campaign_leads(
FakeRepo(), crm, meta, _campaign(meta_campaign_id=None),
)
assert sent == 0
assert meta.calls == []
# --- the cursor ---
async def test_a_campaign_never_synced_reads_the_whole_window():
"""A campaign adopted today should still bring its recent leads across."""
crm, meta = FakeCRM(), FakeMeta()
await push_campaign_leads(FakeRepo(), crm, meta, _campaign())
since = datetime.fromtimestamp(int(meta.calls[0]["since"]), tz=timezone.utc)
age = datetime.now(timezone.utc) - since
assert timedelta(days=6) < age < timedelta(days=9)
async def test_the_next_run_reads_from_where_the_last_one_finished():
"""Re-sending the whole window every five minutes is thousands of
requests a day for leads the CRM already has."""
repo = FakeRepo()
campaign = _campaign()
crm, meta = FakeCRM(), FakeMeta([_meta_lead()])
await push_campaign_leads(repo, crm, meta, campaign)
await push_campaign_leads(repo, crm, meta, campaign)
second_since = datetime.fromtimestamp(
int(meta.calls[1]["since"]), tz=timezone.utc,
)
# Roughly an hour back, the deliberate overlap — not seven days.
age = datetime.now(timezone.utc) - second_since
assert timedelta(minutes=50) < age < timedelta(hours=2)
async def test_the_cursor_only_advances_after_every_lead_is_accepted():
"""A run that fails half-way must be retried in full by the next one."""
repo = FakeRepo()
campaign = _campaign()
crm = FakeCRM()
crm.fail_on_lead = "lead_2"
meta = FakeMeta([_meta_lead("lead_1"), _meta_lead("lead_2")])
with pytest.raises(RuntimeError):
await push_campaign_leads(repo, crm, meta, campaign)
assert "crm_sync" not in campaign.advanced
async def test_nothing_is_written_when_there_were_no_leads():
"""Otherwise every campaign gets an event row every five minutes."""
repo = FakeRepo()
await push_campaign_leads(repo, FakeCRM(), FakeMeta([]), _campaign())
assert repo.events == []
async def test_a_successful_send_is_recorded():
repo = FakeRepo()
await push_campaign_leads(repo, FakeCRM(), FakeMeta([_meta_lead()]), _campaign())
assert repo.events[-1]["event_type"] == "campaign.crm_leads_synced"
assert repo.events[-1]["payload"]["leads_sent"] == 1
# --- campaign figures ---