feat(campaigns): write status change and audit event atomically

_transition previously called update_campaign then add_event as two
separate commits, so a failed add_event left a status change with no
audit row. Phase 2 launches campaigns that authorise spend, so a
compliance hole like that has to close first.

Add CampaignRepository.update_campaign_with_event, which runs the
UPDATE and the event INSERT on one connection inside one transaction
and commits once. Route campaign transitions (submit/approve/reject)
through it instead of the update_campaign + add_event pair; update_campaign
and add_event themselves are unchanged for other callers.

Also update test_campaign_api.py's FakeRepo/monkeypatch to implement the
new repository method, since production code now calls it in the
submit/approve/reject path.

src/adclaw/app/routers/campaigns.py

@@ -131,18 +131,17 @@ async def _transition(
     if action == "approve":
         campaign.approved_by = payload.actor
     try:
-        saved = await repo.update_campaign(campaign)
+        saved = await repo.update_campaign_with_event(
+            campaign,
+            event_type=f"campaign.{action}",
+            actor=payload.actor,
+            reason=payload.reason,
+        )
     except LookupError as exc:
         raise HTTPException(
             status_code=http_status.HTTP_404_NOT_FOUND,
             detail="Campaign not found.",
         ) from exc
-    await repo.add_event(
-        campaign_id,
-        event_type=f"campaign.{action}",
-        actor=payload.actor,
-        reason=payload.reason,
-    )
     return saved
 
 

src/adclaw/campaigns/repo.py

@@ -183,6 +183,78 @@ WHERE id = %s
             await conn.close()
         return campaign_from_row(row)
 
+    async def update_campaign_with_event(
+        self,
+        spec: CampaignSpec,
+        event_type: str,
+        actor: str | None = None,
+        reason: str | None = None,
+        payload: dict[str, Any] | None = None,
+    ) -> CampaignSpec:
+        """Update a campaign and record its audit event atomically.
+
+        A status change without its audit row is a compliance hole once
+        launches authorise spend, so both statements share one transaction.
+        """
+        conn = await connect_database()
+        try:
+            async with conn.cursor() as cur:
+                await cur.execute(
+                    f"""
+UPDATE maskanx_campaigns SET
+    company_id = %s, name = %s, status = %s, objective = %s,
+    ad_account_id = %s, budget = %s, guardrails = %s, targeting = %s,
+    advanced = %s, channels = %s, schedule = %s, meta_campaign_id = %s,
+    sync_status = %s, sync_error = %s, approved_by = %s, updated_at = NOW()
+WHERE id = %s
+{_RETURNING_CLAUSE}
+""",
+                    (
+                        spec.company_id,
+                        spec.name,
+                        spec.status,
+                        spec.objective,
+                        spec.ad_account_id,
+                        jsonb(spec.budget),
+                        jsonb(spec.guardrails),
+                        jsonb(spec.targeting),
+                        jsonb(spec.advanced),
+                        jsonb(spec.channels),
+                        jsonb(spec.schedule),
+                        spec.meta_campaign_id,
+                        spec.sync_status,
+                        spec.sync_error,
+                        spec.approved_by,
+                        spec.id,
+                    ),
+                )
+                row = await cur.fetchone()
+                if row is None:
+                    raise LookupError(f"Campaign {spec.id} does not exist")
+
+                await cur.execute(
+                    """
+INSERT INTO maskanx_campaign_events (
+    id, campaign_id, event_type, actor, reason, payload
+) VALUES (%s, %s, %s, %s, %s, %s)
+""",
+                    (
+                        str(uuid.uuid4()),
+                        spec.id,
+                        event_type,
+                        actor,
+                        reason,
+                        jsonb(payload or {}),
+                    ),
+                )
+            await conn.commit()
+        except Exception:
+            await conn.rollback()
+            raise
+        finally:
+            await conn.close()
+        return campaign_from_row(row)
+
     async def delete_campaign(self, campaign_id: str) -> bool:
         conn = await connect_database()
         try:

tests/test_campaign_api.py

@@ -28,6 +28,14 @@ class FakeRepo:
         self.items[spec.id] = spec
         return spec
 
+    async def update_campaign_with_event(self, spec, event_type, actor=None,
+                                          reason=None, payload=None):
+        if spec.id not in self.items:
+            raise LookupError(f"Campaign {spec.id} does not exist")
+        self.items[spec.id] = spec
+        self.events.append({"campaign_id": spec.id, "event_type": event_type})
+        return spec
+
     async def delete_campaign(self, campaign_id):
         return self.items.pop(campaign_id, None) is not None
 
@@ -212,10 +220,10 @@ def test_submit_missing_campaign_returns_404(client, monkeypatch):
     created = client.post("/api/campaigns", json=_payload()).json()
     cid = created["id"]
 
-    async def raise_lookup(spec):
+    async def raise_lookup(spec, event_type, actor=None, reason=None, payload=None):
         raise LookupError("gone")
 
-    monkeypatch.setattr(client.repo, "update_campaign", raise_lookup)
+    monkeypatch.setattr(client.repo, "update_campaign_with_event", raise_lookup)
     response = client.post(f"/api/campaigns/{cid}/submit")
     assert response.status_code == 404
 

tests/test_campaign_repo_txn.py

@@ -0,0 +1,41 @@
+# -*- coding: utf-8 -*-
+"""The transactional write must issue both statements on one connection."""
+import inspect
+
+from adclaw.campaigns import repo as repo_module
+
+
+def test_transactional_method_exists():
+    assert hasattr(repo_module.CampaignRepository, "update_campaign_with_event")
+
+
+def test_transactional_method_opens_exactly_one_connection():
+    source = inspect.getsource(
+        repo_module.CampaignRepository.update_campaign_with_event,
+    )
+    assert source.count("connect_database()") == 1
+
+
+def test_transactional_method_commits_once_after_both_statements():
+    source = inspect.getsource(
+        repo_module.CampaignRepository.update_campaign_with_event,
+    )
+    assert source.count("await conn.commit()") == 1
+    assert "maskanx_campaign_events" in source
+    assert "UPDATE maskanx_campaigns" in source
+    assert source.index("UPDATE maskanx_campaigns") < source.index("await conn.commit()")
+    assert source.index("maskanx_campaign_events") < source.index("await conn.commit()")
+
+
+def test_transactional_method_rolls_back_on_error():
+    source = inspect.getsource(
+        repo_module.CampaignRepository.update_campaign_with_event,
+    )
+    assert "await conn.rollback()" in source
+
+
+def test_transactional_method_raises_lookup_error_for_unknown_id():
+    source = inspect.getsource(
+        repo_module.CampaignRepository.update_campaign_with_event,
+    )
+    assert "LookupError" in source