[Issue #3536] Add saved opportunity notifications to backend job

api/src/db/models/user_models.py

@@ -110,7 +110,9 @@ class UserSavedSearch(ApiSchemaTable, TimestampMixin):
 class UserNotificationLog(ApiSchemaTable, TimestampMixin):
     __tablename__ = "user_notification_log"
 
-    user_notification_log_id: Mapped[uuid.UUID] = mapped_column(UUID, primary_key=True)
+    user_notification_log_id: Mapped[uuid.UUID] = mapped_column(
+        UUID, primary_key=True, default=uuid.uuid4
+    )
 
     user_id: Mapped[uuid.UUID] = mapped_column(ForeignKey(User.user_id), index=True)
     user: Mapped[User] = relationship(User)

api/src/search/backend/load_opportunities_to_index.py

@@ -182,7 +182,7 @@ def _handle_incremental_upserts(self, existing_opportunity_ids: set[int]) -> Non
             self.db_session.execute(
                 update(OpportunityChangeAudit)
                 .where(OpportunityChangeAudit.opportunity_id.in_(processed_opportunity_ids))
-                .values(updated_at=get_now_us_eastern_datetime())
+                .values(updated_at=datetime_util.utcnow())
             )
 
     def _handle_incremental_delete(self, existing_opportunity_ids: set[int]) -> None:

api/src/task/notifications/generate_notifications.py

@@ -7,8 +7,8 @@
 
 import src.adapters.db as db
 import src.adapters.db.flask_db as flask_db
-from src.db.models.opportunity_models import Opportunity
-from src.db.models.user_models import User, UserSavedOpportunity
+from src.db.models.opportunity_models import OpportunityChangeAudit
+from src.db.models.user_models import UserNotificationLog, UserSavedOpportunity
 from src.task.ecs_background_task import ecs_background_task
 from src.task.task import Task
 from src.task.task_blueprint import task_blueprint
@@ -32,8 +32,7 @@ def run_notification_task(db_session: db.Session) -> None:
 class NotificationContainer:
     """Container for collecting notifications for a single user"""
 
-    user: User
-    updated_opportunity_ids: list[int] = field(default_factory=list)
+    saved_opportunities: list[UserSavedOpportunity] = field(default_factory=list)
     # TODO: Change from str to something else
     updated_searches: list[str] = field(default_factory=list)
 
@@ -61,33 +60,29 @@ def run_task(self) -> None:
     def _collect_opportunity_notifications(self) -> None:
         """Collect notifications for changed opportunities that users are tracking"""
         stmt = (
-            select(User.user_id, UserSavedOpportunity.opportunity_id)
+            select(UserSavedOpportunity)
             .join(
-                UserSavedOpportunity,
-                User.user_id == UserSavedOpportunity.user_id,
+                OpportunityChangeAudit,
+                OpportunityChangeAudit.opportunity_id == UserSavedOpportunity.opportunity_id,
             )
-            .join(
-                Opportunity,
-                UserSavedOpportunity.opportunity_id == Opportunity.opportunity_id,
-            )
-            .where(Opportunity.updated_at > UserSavedOpportunity.last_notified_at)
+            .where(OpportunityChangeAudit.updated_at > UserSavedOpportunity.last_notified_at)
+            .distinct()
         )
 
         results = self.db_session.execute(stmt)
 
-        for row in results.mappings():
-            user_id = row["user_id"]
-            opportunity_id = row["opportunity_id"]
+        for row in results.scalars():
+            user_id = row.user_id
             if user_id not in self.user_notification_map:
-                self.user_notification_map[user_id] = NotificationContainer(user=user_id)
-            self.user_notification_map[user_id].updated_opportunity_ids.append(opportunity_id)
+                self.user_notification_map[user_id] = NotificationContainer()
+            self.user_notification_map[user_id].saved_opportunities.append(row)
 
         logger.info(
             "Collected opportunity notifications",
             extra={
                 "user_count": len(self.user_notification_map),
                 "total_notifications": sum(
-                    len(container.updated_opportunity_ids)
+                    len(container.saved_opportunities)
                     for container in self.user_notification_map.values()
                 ),
             },
@@ -103,32 +98,42 @@ def _collect_search_notifications(self) -> None:
     def _send_notifications(self) -> None:
         """Send collected notifications to users"""
         for user_id, container in self.user_notification_map.items():
-            if not container.updated_opportunity_ids and not container.updated_searches:
+            if not container.saved_opportunities and not container.updated_searches:
                 continue
 
             # TODO: Implement actual notification sending in future ticket
             logger.info(
                 "Would send notification to user",
                 extra={
                     "user_id": user_id,
-                    "opportunity_count": len(container.updated_opportunity_ids),
+                    "opportunity_count": len(container.saved_opportunities),
                     "search_count": len(container.updated_searches),
                 },
             )
 
+            # Create notification log entry
+            # TODO: Use enum for notification reason?
+            notification_log = UserNotificationLog(
+                user_id=user_id,
+                notification_reason="opportunity_updates",
+                notification_sent=True,
+            )
+            self.db_session.add(notification_log)
+
             # Update last_notified_at for all opportunities we just notified about
+            opportunity_ids = [
+                saved_opp.opportunity_id for saved_opp in container.saved_opportunities
+            ]
             self.db_session.execute(
                 update(UserSavedOpportunity)
                 .where(
                     UserSavedOpportunity.user_id == user_id,
-                    UserSavedOpportunity.opportunity_id.in_(container.updated_opportunity_ids),
+                    UserSavedOpportunity.opportunity_id.in_(opportunity_ids),
                 )
                 .values(last_notified_at=datetime_util.utcnow())
             )
 
-            self.increment(
-                self.Metrics.OPPORTUNITIES_TRACKED, len(container.updated_opportunity_ids)
-            )
+            self.increment(self.Metrics.OPPORTUNITIES_TRACKED, len(container.saved_opportunities))
             self.increment(self.Metrics.SEARCHES_TRACKED, len(container.updated_searches))
             self.increment(self.Metrics.NOTIFICATIONS_SENT)
             self.increment(self.Metrics.USERS_NOTIFIED)

api/tests/src/task/notifications/test_generate_notifications.py

@@ -1,6 +1,15 @@
 from datetime import timedelta
 
+import pytest
+
 import tests.src.db.models.factories as factories
+from src.db.models.user_models import UserNotificationLog
+
+
+@pytest.fixture
+def clear_notification_logs(db_session):
+    """Clear all notification logs"""
+    db_session.query(UserNotificationLog).delete()
 
 
 def test_via_cli(cli_runner, db_session, enable_factory_create, user):
@@ -14,11 +23,15 @@ def test_collect_notifications_cli(cli_runner, db_session, enable_factory_create
     """Simple test that verifies we can invoke the notification task via CLI"""
     # Create a saved opportunity that needs notification
     opportunity = factories.OpportunityFactory.create()
-    factories.UserSavedOpportunityFactory.create(
+    saved_opportunity = factories.UserSavedOpportunityFactory.create(
         user=user,
         opportunity=opportunity,
         last_notified_at=opportunity.updated_at - timedelta(days=1),
     )
+    factories.OpportunityChangeAuditFactory.create(
+        opportunity=opportunity,
+        updated_at=saved_opportunity.last_notified_at + timedelta(minutes=1),
+    )
 
     result = cli_runner.invoke(args=["task", "generate-notifications"])
 
@@ -46,7 +59,10 @@ def test_last_notified_at_updates(cli_runner, db_session, enable_factory_create,
         opportunity=opportunity,
         last_notified_at=opportunity.updated_at - timedelta(days=1),
     )
-
+    factories.OpportunityChangeAuditFactory.create(
+        opportunity=opportunity,
+        updated_at=saved_opp.last_notified_at + timedelta(minutes=1),
+    )
     # Store the original notification time
     original_notification_time = saved_opp.last_notified_at
 
@@ -61,3 +77,55 @@ def test_last_notified_at_updates(cli_runner, db_session, enable_factory_create,
     assert saved_opp.last_notified_at > original_notification_time
     # Verify last_notified_at is now after the opportunity's updated_at
     assert saved_opp.last_notified_at > opportunity.updated_at
+
+
+def test_notification_log_creation(
+    cli_runner, db_session, enable_factory_create, clear_notification_logs, user
+):
+    """Test that notification logs are created when notifications are sent"""
+    # Create a saved opportunity that needs notification
+    opportunity = factories.OpportunityFactory.create()
+    saved_opportunity = factories.UserSavedOpportunityFactory.create(
+        user=user,
+        opportunity=opportunity,
+        last_notified_at=opportunity.updated_at - timedelta(days=1),
+    )
+
+    factories.OpportunityChangeAuditFactory.create(
+        opportunity=opportunity,
+        updated_at=saved_opportunity.last_notified_at + timedelta(minutes=1),
+    )
+
+    # Run the notification task
+    result = cli_runner.invoke(args=["task", "generate-notifications"])
+    assert result.exit_code == 0
+
+    # Verify notification log was created
+    notification_logs = db_session.query(UserNotificationLog).all()
+    assert len(notification_logs) == 1
+
+    log = notification_logs[0]
+    assert log.user_id == user.user_id
+    assert log.notification_reason == "opportunity_updates"
+    assert log.notification_sent is True
+
+
+def test_no_notification_log_when_no_updates(
+    cli_runner, db_session, enable_factory_create, clear_notification_logs, user
+):
+    """Test that no notification log is created when there are no updates"""
+    # Create a saved opportunity that doesn't need notification
+    opportunity = factories.OpportunityFactory.create()
+    factories.UserSavedOpportunityFactory.create(
+        user=user,
+        opportunity=opportunity,
+        last_notified_at=opportunity.updated_at + timedelta(minutes=1),  # After the update
+    )
+
+    # Run the notification task
+    result = cli_runner.invoke(args=["task", "generate-notifications"])
+    assert result.exit_code == 0
+
+    # Verify no notification log was created
+    notification_logs = db_session.query(UserNotificationLog).all()
+    assert len(notification_logs) == 0