98a4ec305e
Grow event emitter fixes (wrong route / silent failure cannot recur):
- URL: /events/ingest -> /events/ingest/service (canonical service ingest path)
- Add x-growqr-source: matchmaking-service header
- Call resp.raise_for_status() so a 401 is caught and logged, not silently
swallowed (401 is now observable; a wrong route surfaces immediately)
- Action-specific dedupe keys so retries collapse but later distinct user
actions are NOT dropped:
matches.generated -> matchmaking:generated:{user}:{sig}
feed.viewed -> matchmaking:feed:{user}:{updated_at|none}
match.viewed -> matchmaking:viewed:{user}:{opp_id}
match.applied -> matchmaking:applied:{user}:{opp_id} (already present)
All four emit handlers fire the event AFTER the persistent mutation
(set_opportunity_state / save_feed), so no event leaks when a mutation fails.
Tests (tests/test_grow_events.py, 12 cases, red->green TDD):
- URL normalizes to /events/ingest/service (with/without trailing slash)
- x-growqr-source header present
- 401 response raises_for_status -> logged warning; does not escape _post
- emit-after-mutation ordering; no emit when mutation raises
- dedupe keys stable for same evidence, distinct for different evidence
across generated/feed/viewed/applied event types
74 lines
2.3 KiB
Python
74 lines
2.3 KiB
Python
"""Grow event emitter — best-effort POST to the backend /events/ingest/service endpoint.
|
|
|
|
Every call degrades to a no-op when GROW_EVENTS_URL is unset or the POST fails, so
|
|
a Grow outage can never break the search/apply flows. The emitter runs in a
|
|
background thread (via asyncio) so it never blocks the agent session loop.
|
|
"""
|
|
from __future__ import annotations
|
|
|
|
import asyncio
|
|
import logging
|
|
import os
|
|
from datetime import datetime, timezone
|
|
|
|
import httpx
|
|
|
|
logger = logging.getLogger(__name__)
|
|
|
|
SERVICE_NAME = "matchmaking-service"
|
|
|
|
|
|
def _post(event: dict) -> None:
|
|
"""Synchronous POST — run off the event loop via ``asyncio.to_thread``."""
|
|
grow_url = os.getenv("GROW_EVENTS_URL", "").rstrip("/")
|
|
if not grow_url:
|
|
return
|
|
url = grow_url + "/events/ingest/service"
|
|
token = os.getenv("GROW_EVENTS_TOKEN", "") or os.getenv("A2A_SERVICE_KEY", "")
|
|
headers = {"Content-Type": "application/json", "x-growqr-source": SERVICE_NAME}
|
|
if token:
|
|
headers["Authorization"] = f"Bearer {token}"
|
|
try:
|
|
resp = httpx.post(url, json=event, headers=headers, timeout=5.0)
|
|
resp.raise_for_status()
|
|
except Exception as e: # noqa: BLE001 — never fatal
|
|
logger.warning("emit_grow_event failed (%s): %s", event.get("type"), e)
|
|
|
|
|
|
async def emit_grow_event(
|
|
event_type: str,
|
|
user_id: str,
|
|
*,
|
|
org_id: str | None = None,
|
|
payload: dict | None = None,
|
|
correlation: dict | None = None,
|
|
subject: dict | None = None,
|
|
dedupe_key: str | None = None,
|
|
) -> None:
|
|
"""Emit a Grow event to the backend ingest endpoint.
|
|
|
|
Non-blocking and non-fatal: if GROW_EVENTS_URL is unset the call is a no-op,
|
|
and any HTTP error is logged and swallowed. The POST runs in a worker thread
|
|
so the agent session loop is never blocked.
|
|
"""
|
|
if not user_id:
|
|
return
|
|
|
|
event: dict = {
|
|
"source": SERVICE_NAME,
|
|
"type": event_type,
|
|
"category": "service",
|
|
"userId": user_id,
|
|
"orgId": org_id or "growqr",
|
|
"occurredAt": datetime.now(timezone.utc).isoformat(),
|
|
"payload": payload or {},
|
|
}
|
|
if correlation:
|
|
event["correlation"] = correlation
|
|
if subject:
|
|
event["subject"] = subject
|
|
if dedupe_key:
|
|
event["dedupeKey"] = dedupe_key
|
|
|
|
await asyncio.to_thread(_post, event)
|