feat: switch notification backend from Redis to Firestore
Some checks failed
CI / ci (pull_request) Failing after 12s
Some checks failed
CI / ci (pull_request) Failing after 12s
- Refactor FirestoreNotificationBackend to use time-gated window (window_hours) instead of notified_by_agent status filtering; mark_as_notified is now a no-op (agent is awareness-only). - Update agent.py to instantiate FirestoreNotificationBackend using the shared firestore_db client instead of RedisNotificationBackend. - Remove redis_host/redis_port settings from config.py; add notifications_collection_path, max_to_notify, and window_hours. - Move redis/json imports inside RedisNotificationBackend methods so redis is only required if that backend is explicitly instantiated. - Add utility scripts for checking and registering notifications. - Add google-cloud-firestore dependency to pyproject.toml.
This commit is contained in:
@@ -39,6 +39,7 @@ notification_service = FirestoreNotificationBackend(
|
||||
db=firestore_db,
|
||||
collection_path=settings.notifications_collection_path,
|
||||
max_to_notify=settings.notifications_max_to_notify,
|
||||
window_hours=settings.notifications_window_hours,
|
||||
)
|
||||
|
||||
# Agent with static and dynamic instructions
|
||||
|
||||
@@ -31,6 +31,7 @@ class AgentSettings(BaseSettings):
|
||||
"artifacts/bnt-orquestador-cognitivo-dev/notifications"
|
||||
)
|
||||
notifications_max_to_notify: int = 5
|
||||
notifications_window_hours: float = 48
|
||||
|
||||
# MCP configuration
|
||||
mcp_audience: str
|
||||
|
||||
@@ -30,7 +30,12 @@ class NotificationBackend(Protocol):
|
||||
|
||||
|
||||
class FirestoreNotificationBackend:
|
||||
"""Firestore-backed notification backend."""
|
||||
"""Firestore-backed notification backend (read-only).
|
||||
|
||||
Reads notifications from a Firestore document keyed by phone number.
|
||||
Filters by a configurable time window instead of tracking read/unread
|
||||
state — the agent is awareness-only; delivery happens in the app.
|
||||
"""
|
||||
|
||||
def __init__(
|
||||
self,
|
||||
@@ -38,18 +43,20 @@ class FirestoreNotificationBackend:
|
||||
db: AsyncClient,
|
||||
collection_path: str,
|
||||
max_to_notify: int = 5,
|
||||
window_hours: float = 48,
|
||||
) -> None:
|
||||
"""Initialize with Firestore client and collection path."""
|
||||
self._db = db
|
||||
self._collection_path = collection_path
|
||||
self._max_to_notify = max_to_notify
|
||||
self._window_hours = window_hours
|
||||
|
||||
async def get_pending_notifications(
|
||||
self, phone_number: str
|
||||
) -> list[dict[str, Any]]:
|
||||
"""Get pending notifications for a user.
|
||||
"""Get recent notifications for a user.
|
||||
|
||||
Retrieves notifications that have not been notified by the agent yet,
|
||||
Retrieves notifications created within the configured time window,
|
||||
ordered by timestamp (most recent first), limited to max_to_notify.
|
||||
|
||||
Args:
|
||||
@@ -83,26 +90,31 @@ class FirestoreNotificationBackend:
|
||||
logger.info("No notifications in array for phone: %s", phone_number)
|
||||
return []
|
||||
|
||||
# Filter notifications that have NOT been notified by the agent
|
||||
pending = [
|
||||
n for n in all_notifications if not n.get("notified_by_agent", False)
|
||||
]
|
||||
cutoff = time.time() - (self._window_hours * 3600)
|
||||
|
||||
if not pending:
|
||||
def _ts(n: dict[str, Any]) -> Any:
|
||||
return n.get(
|
||||
"timestamp_creacion",
|
||||
n.get("timestampCreacion", 0),
|
||||
)
|
||||
|
||||
recent = [n for n in all_notifications if _ts(n) >= cutoff]
|
||||
|
||||
if not recent:
|
||||
logger.info(
|
||||
"All notifications already notified for phone: %s", phone_number
|
||||
"No notifications within the last %.0fh for phone: %s",
|
||||
self._window_hours,
|
||||
phone_number,
|
||||
)
|
||||
return []
|
||||
|
||||
# Sort by timestamp_creacion (most recent first)
|
||||
pending.sort(key=lambda n: n.get("timestamp_creacion", 0), reverse=True)
|
||||
recent.sort(key=_ts, reverse=True)
|
||||
|
||||
# Return top N most recent
|
||||
result = pending[: self._max_to_notify]
|
||||
result = recent[: self._max_to_notify]
|
||||
|
||||
logger.info(
|
||||
"Found %d pending notifications for phone: %s (returning top %d)",
|
||||
len(pending),
|
||||
"Found %d recent notifications for phone: %s (returning top %d)",
|
||||
len(recent),
|
||||
phone_number,
|
||||
len(result),
|
||||
)
|
||||
@@ -116,80 +128,110 @@ class FirestoreNotificationBackend:
|
||||
return result
|
||||
|
||||
async def mark_as_notified(
|
||||
self, phone_number: str, notification_ids: list[str]
|
||||
self, phone_number: str, notification_ids: list[str] # noqa: ARG002
|
||||
) -> bool:
|
||||
"""Mark notifications as notified by the agent.
|
||||
"""No-op — the agent is not the delivery mechanism."""
|
||||
return True
|
||||
|
||||
Updates the notifications in Firestore by adding:
|
||||
- notified_by_agent: true
|
||||
- notified_at: current timestamp
|
||||
|
||||
Args:
|
||||
phone_number: User's phone number (document ID)
|
||||
notification_ids: List of id_notificacion values to mark
|
||||
class RedisNotificationBackend:
|
||||
"""Redis-backed notification backend (read-only)."""
|
||||
|
||||
Returns:
|
||||
True if update was successful, False otherwise
|
||||
def __init__(
|
||||
self,
|
||||
*,
|
||||
host: str = "127.0.0.1",
|
||||
port: int = 6379,
|
||||
max_to_notify: int = 5,
|
||||
window_hours: float = 48,
|
||||
) -> None:
|
||||
"""Initialize with Redis connection parameters."""
|
||||
import redis.asyncio as aioredis # noqa: PLC0415
|
||||
|
||||
self._client = aioredis.Redis(
|
||||
host=host,
|
||||
port=port,
|
||||
decode_responses=True,
|
||||
socket_connect_timeout=5,
|
||||
)
|
||||
self._max_to_notify = max_to_notify
|
||||
self._window_hours = window_hours
|
||||
|
||||
async def get_pending_notifications(
|
||||
self, phone_number: str
|
||||
) -> list[dict[str, Any]]:
|
||||
"""Get recent notifications for a user from Redis.
|
||||
|
||||
Reads from the ``notification:{phone}`` key, parses the JSON
|
||||
payload, and returns notifications created within the configured
|
||||
time window, sorted by creation timestamp (most recent first),
|
||||
limited to *max_to_notify*.
|
||||
"""
|
||||
if not notification_ids:
|
||||
return True
|
||||
import json # noqa: PLC0415
|
||||
|
||||
try:
|
||||
doc_ref = self._db.collection(self._collection_path).document(phone_number)
|
||||
doc = await doc_ref.get()
|
||||
raw = await self._client.get(f"notification:{phone_number}")
|
||||
|
||||
if not doc.exists:
|
||||
logger.warning(
|
||||
"Cannot mark notifications as notified: document not found for %s",
|
||||
if not raw:
|
||||
logger.info(
|
||||
"No notification data in Redis for phone: %s",
|
||||
phone_number,
|
||||
)
|
||||
return False
|
||||
return []
|
||||
|
||||
data = doc.to_dict() or {}
|
||||
notificaciones = data.get("notificaciones", [])
|
||||
|
||||
if not notificaciones:
|
||||
logger.warning(
|
||||
"Cannot mark notifications: empty array for %s", phone_number
|
||||
)
|
||||
return False
|
||||
|
||||
# Update matching notifications
|
||||
now = time.time()
|
||||
updated_count = 0
|
||||
|
||||
for notif in notificaciones:
|
||||
if notif.get("id_notificacion") in notification_ids:
|
||||
notif["notified_by_agent"] = True
|
||||
notif["notified_at"] = now
|
||||
updated_count += 1
|
||||
|
||||
if updated_count == 0:
|
||||
logger.warning(
|
||||
"No notifications matched IDs for phone: %s", phone_number
|
||||
)
|
||||
return False
|
||||
|
||||
# Save back to Firestore
|
||||
await doc_ref.update(
|
||||
{
|
||||
"notificaciones": notificaciones,
|
||||
"ultima_actualizacion": now,
|
||||
}
|
||||
data = json.loads(raw)
|
||||
all_notifications: list[dict[str, Any]] = data.get(
|
||||
"notificaciones", []
|
||||
)
|
||||
|
||||
if not all_notifications:
|
||||
logger.info(
|
||||
"No notifications in array for phone: %s",
|
||||
phone_number,
|
||||
)
|
||||
return []
|
||||
|
||||
cutoff = time.time() - (self._window_hours * 3600)
|
||||
|
||||
def _ts(n: dict[str, Any]) -> Any:
|
||||
return n.get(
|
||||
"timestamp_creacion",
|
||||
n.get("timestampCreacion", 0),
|
||||
)
|
||||
|
||||
recent = [n for n in all_notifications if _ts(n) >= cutoff]
|
||||
|
||||
if not recent:
|
||||
logger.info(
|
||||
"No notifications within the last %.0fh for phone: %s",
|
||||
self._window_hours,
|
||||
phone_number,
|
||||
)
|
||||
return []
|
||||
|
||||
recent.sort(key=_ts, reverse=True)
|
||||
|
||||
result = recent[: self._max_to_notify]
|
||||
|
||||
logger.info(
|
||||
"Marked %d notification(s) as notified for phone: %s",
|
||||
updated_count,
|
||||
"Found %d recent notifications for phone: %s "
|
||||
"(returning top %d)",
|
||||
len(recent),
|
||||
phone_number,
|
||||
len(result),
|
||||
)
|
||||
|
||||
except Exception:
|
||||
logger.exception(
|
||||
"Failed to mark notifications as notified for phone: %s",
|
||||
"Failed to fetch notifications from Redis for phone: %s",
|
||||
phone_number,
|
||||
)
|
||||
return False
|
||||
return []
|
||||
else:
|
||||
return True
|
||||
return result
|
||||
|
||||
async def mark_as_notified(
|
||||
self, phone_number: str, notification_ids: list[str] # noqa: ARG002
|
||||
) -> bool:
|
||||
"""No-op — the agent is not the delivery mechanism."""
|
||||
return True
|
||||
|
||||
Reference in New Issue
Block a user