Add event and device tracking to answers (M2-8482)

Pipfile

@@ -11,14 +11,14 @@ alembic = "==1.14.*"
 asyncpg = "==0.30.0"
 azure-storage-blob = "==12.24.*"
 bcrypt = "==4.2.1"
-boto3 = "==1.36.*"
+boto3 = "==1.37.*"
 fastapi = "==0.115.*"
 fastapi-mail = "==1.2.9"
 firebase-admin = "==6.6.*"
 httpx = "==0.28.*"
 jinja2 = "==3.1.*"
 more-itertools = "==10.6.0"
-nh3 = "==0.2.20"
+nh3 = "==0.2.*"
 pydantic = { extras = ["email"], version = "==1.10.18" }
 pyjwt = "==2.10.1"
 pymongo = "==4.11.*"
@@ -34,7 +34,7 @@ taskiq-fastapi = "==0.3.*"
 taskiq-redis = "==1.0.2"
 typer = "==0.15.*"
 uvicorn = { extras = ["standard"], version = "==0.34.*" }
-ddtrace = "==2.20.*"
+ddtrace = "==2.21.*"
 bytecode = "==0.16.*"
 structlog = "==25.1.*"
 asgi-correlation-id = "==4.3.4"

src/apps/answers/api.py

@@ -2,8 +2,9 @@
 import base64
 import datetime
 import uuid
+from typing import Annotated
 
-from fastapi import Body, Depends, Query
+from fastapi import Body, Depends, Header, Query
 from fastapi.responses import Response as FastApiResponse
 from pydantic import parse_obj_as
 
@@ -53,6 +54,8 @@
 from apps.authentication.deps import get_current_user
 from apps.integrations.prolific.domain import ProlificUserInfo
 from apps.integrations.prolific.service.prolific import ProlificIntegrationService
+from apps.schedule.crud.user_device_events_history import UserDeviceEventsHistoryCRUD
+from apps.schedule.service.schedule_history import ScheduleHistoryService
 from apps.shared.deps import get_client_ip, get_i18n
 from apps.shared.domain import Response, ResponseMulti
 from apps.shared.exception import AccessDeniedError, NotFoundError, ValidationError
@@ -78,18 +81,40 @@ async def create_answer(
     tz_offset: int | None = Depends(get_tz_utc_offset()),
     session=Depends(get_session),
     answer_session=Depends(get_answer_session),
+    device_id: Annotated[str | None, Header()] = None,
 ) -> None:
     async with atomic(session):
         await CheckAccessService(session, user.id).check_answer_create_access(schema.applet_id)
         try:
             await AppletHistoryService(session, schema.applet_id, schema.version).get()
         except NotValidAppletHistory:
             raise InvalidVersionError()
+
+        if schema.event_history_id:
+            event = await ScheduleHistoryService(session).get_by_id(schema.event_history_id)
+            if (
+                event is None
+                or event.activity_flow_id != schema.flow_id
+                or event.activity_id != schema.activity_id
+                or (event.user_id is not None and event.user_id != user.id)
+            ):
+                raise NotFoundError("Invalid event_history_id provided")
+
+        device = None
+        if device_id and schema.event_history_id:
+            event_id = uuid.UUID(schema.event_history_id.split("_")[0])
+            event_version = schema.event_history_id.split("_")[1]
+            device = await UserDeviceEventsHistoryCRUD(session).get_device(
+                device_id=device_id, user_id=user.id, event_id=event_id, event_version=event_version
+            )
+            if device is None:
+                raise NotFoundError("Invalid device_id provided")
+
         service = AnswerService(session, user.id, answer_session)
         if tz_offset is not None and schema.answer.tz_offset is None:
             schema.answer.tz_offset = tz_offset // 60  # value in minutes
         async with atomic(answer_session):
-            answer = await service.create_answer(schema)
+            answer = await service.create_answer(schema, device.device_id if device else None)
         await service.create_report_from_answer(answer)
 
 

src/apps/answers/db/schemas.py

@@ -42,6 +42,8 @@ class AnswerSchema(HistoryAware, Base):
     input_subject_id = Column(UUID(as_uuid=True), nullable=True, index=True)
     relation = Column(String(length=20), nullable=True)
     consent_to_share = Column(Boolean(), default=False)
+    event_history_id = Column(String(), nullable=True, index=True)
+    device_id = Column(Text(), nullable=True, index=True)
 
     answer_item = relationship(
         "AnswerItemSchema",

src/apps/answers/domain/answers.py

@@ -115,6 +115,7 @@ class AppletAnswerCreate(InternalModel):
     input_subject_id: uuid.UUID | None = None
     consent_to_share: bool | None = False
     prolific_params: ProlificParamsActivityAnswer | None = None
+    event_history_id: str | None = None
 
     _dates_from_ms = validator("created_at", pre=True, allow_reuse=True)(datetime_from_ms)
 

src/apps/answers/service.py

@@ -134,21 +134,25 @@ def key_generator(pk: uuid.UUID) -> str:
 
         return key_generator
 
-    async def create_answer(self, activity_answer: AppletAnswerCreate) -> AnswerSchema:
+    async def create_answer(self, activity_answer: AppletAnswerCreate, device_id: str | None = None) -> AnswerSchema:
         # Check for prolific parameters in the answer helping to identify whether the respondent comes from prolific
         is_prolific_respondent = activity_answer.prolific_params is not None
         if self.user_id and not is_prolific_respondent:
-            return await self._create_respondent_answer(activity_answer)
+            return await self._create_respondent_answer(activity_answer, device_id)
         else:
-            return await self._create_anonymous_answer(activity_answer)
+            return await self._create_anonymous_answer(activity_answer, device_id)
 
-    async def _create_respondent_answer(self, activity_answer: AppletAnswerCreate) -> AnswerSchema:
+    async def _create_respondent_answer(
+        self, activity_answer: AppletAnswerCreate, device_id: str | None
+    ) -> AnswerSchema:
         await self._validate_respondent_answer(activity_answer)
-        return await self._create_answer(activity_answer)
+        return await self._create_answer(activity_answer, device_id)
 
-    async def _create_anonymous_answer(self, activity_answer: AppletAnswerCreate) -> AnswerSchema:
+    async def _create_anonymous_answer(
+        self, activity_answer: AppletAnswerCreate, device_id: str | None
+    ) -> AnswerSchema:
         await self._validate_anonymous_answer(activity_answer)
-        return await self._create_answer(activity_answer)
+        return await self._create_answer(activity_answer, device_id)
 
     async def _validate_respondent_answer(self, activity_answer: AppletAnswerCreate) -> None:
         await self._validate_answer(activity_answer)
@@ -307,7 +311,7 @@ async def _get_answer_relation(
 
         return relation.relation
 
-    async def _create_answer(self, applet_answer: AppletAnswerCreate) -> AnswerSchema:
+    async def _create_answer(self, applet_answer: AppletAnswerCreate, device_id: str | None) -> AnswerSchema:
         assert self.user_id
         pk = self._generate_history_id(applet_answer.version)
         created_at = applet_answer.created_at or datetime.datetime.now(datetime.UTC).replace(tzinfo=None)
@@ -384,6 +388,8 @@ async def _create_answer(self, applet_answer: AppletAnswerCreate) -> AnswerSchem
                 input_subject_id=input_subject.id,
                 relation=relation,
                 consent_to_share=applet_answer.consent_to_share,
+                event_history_id=applet_answer.event_history_id,
+                device_id=device_id,
             )
         )
         item_answer = applet_answer.answer

src/apps/answers/tests/test_answers.py

@@ -25,6 +25,8 @@
 from apps.applets.errors import InvalidVersionError
 from apps.applets.service import AppletService
 from apps.mailing.services import TestMail
+from apps.schedule.domain.schedule import PublicEvent
+from apps.schedule.service import ScheduleService
 from apps.shared.test import BaseTest
 from apps.shared.test.client import TestClient
 from apps.subjects.constants import Relation
@@ -33,7 +35,7 @@
 from apps.subjects.services import SubjectsService
 from apps.users import User
 from apps.users.cruds.user import UsersCRUD
-from apps.users.domain import UserCreate
+from apps.users.domain import AppInfoOS, UserCreate, UserDeviceCreate
 from apps.users.tests.fixtures.users import _get_or_create_user
 from apps.workspaces.crud.user_applet_access import UserAppletAccessCRUD
 from apps.workspaces.db.schemas import UserAppletAccessSchema
@@ -673,6 +675,18 @@ async def applet_one_user_subject(session: AsyncSession, applet_one: AppletFull,
     )
 
 
+@pytest.fixture
+async def applet_default_events(session: AsyncSession, applet: AppletFull) -> list[PublicEvent]:
+    srv = ScheduleService(session)
+    events = await srv.get_all_schedules(applet_id=applet.id)
+    return events
+
+
+@pytest.fixture
+def device_create_data() -> UserDeviceCreate:
+    return UserDeviceCreate(os=AppInfoOS(name="os1", version="1.0.0"), app_version="51.0.0", device_id="device_id")
+
+
 @pytest.mark.usefixtures("mock_kiq_report")
 class TestAnswerActivityItems(BaseTest):
     fixtures = [
@@ -812,6 +826,60 @@ async def test_create_answer__wrong_applet_version(
         assert response.status_code == http.HTTPStatus.BAD_REQUEST
         assert response.json()["result"][0]["message"] == InvalidVersionError.message
 
+    async def test_create_answer__with_device_id_and_event_history_id(
+        self,
+        client: TestClient,
+        tom: User,
+        answer_create: AppletAnswerCreate,
+        applet_default_events,
+        device_create_data,
+    ):
+        client.login(tom)
+
+        response = await client.get(
+            url="/users/me/respondent/current_events",
+            headers={
+                "Device-Id": device_create_data.device_id,
+                "OS-Name": device_create_data.os.name,
+                "OS-Version": device_create_data.os.version,
+                "App-Version": device_create_data.app_version,
+            },
+        )
+
+        assert response.status_code == http.HTTPStatus.OK
+
+        data = answer_create.copy(deep=True)
+        event = next((event for event in applet_default_events if event.activity_id == answer_create.activity_id), None)
+        assert event
+        data.event_history_id = f"{event.id}_{event.version}"
+        response = await client.post(self.answer_url, data=data, headers={"Device-Id": device_create_data.device_id})
+
+        assert response.status_code == http.HTTPStatus.CREATED
+
+    async def test_create_answer_with_wrong_event_history_id(
+        self, client: TestClient, tom: User, answer_create: AppletAnswerCreate
+    ):
+        client.login(tom)
+        data = answer_create.copy(deep=True)
+        data.event_history_id = str(uuid.uuid4())
+        response = await client.post(self.answer_url, data=data)
+
+        assert response.status_code == http.HTTPStatus.NOT_FOUND
+        assert response.json()["result"][0]["message"] == "Invalid event_history_id provided"
+
+    async def test_create_answer_with_wrong_device_id(
+        self, client: TestClient, tom: User, answer_create: AppletAnswerCreate, applet_default_events
+    ):
+        client.login(tom)
+        data = answer_create.copy(deep=True)
+        event = next((event for event in applet_default_events if event.activity_id == answer_create.activity_id), None)
+        assert event
+        data.event_history_id = f"{event.id}_{event.version}"
+        response = await client.post(self.answer_url, data=data, headers={"Device-Id": "wrong_device_id"})
+
+        assert response.status_code == http.HTTPStatus.NOT_FOUND
+        assert response.json()["result"][0]["message"] == "Invalid device_id provided"
+
     async def test_create_activity_answers__submit_duplicate(
         self,
         client: TestClient,

src/apps/schedule/crud/schedule_history.py

@@ -15,6 +15,11 @@
 
 
 class ScheduleHistoryCRUD(BaseCRUD[EventHistorySchema]):
+    schema_class = EventHistorySchema
+
+    async def get_by_id(self, id_version: str) -> EventHistorySchema | None:
+        return await self._get("id_version", id_version)
+
     async def add(self, event: EventHistorySchema) -> EventHistorySchema:
         return await self._create(event)
 

src/apps/schedule/crud/user_device_events_history.py

@@ -17,6 +17,22 @@ def __init__(self, session):
         super().__init__(session)
         self.schema_class = UserDeviceEventsHistorySchema
 
+    async def get_device(
+        self,
+        device_id: str,
+        user_id: uuid.UUID,
+        event_id: uuid.UUID,
+        event_version: str,
+    ) -> UserDeviceEventsHistorySchema:
+        query: Query = select(UserDeviceEventsHistorySchema)
+        query = query.where(UserDeviceEventsHistorySchema.device_id == device_id)
+        query = query.where(UserDeviceEventsHistorySchema.user_id == user_id)
+        query = query.where(UserDeviceEventsHistorySchema.event_id == event_id)
+        query = query.where(UserDeviceEventsHistorySchema.event_version == event_version)
+
+        result = await self._execute(query)
+        return result.scalars().first()
+
     async def record_event_versions(
         self,
         user_id: uuid.UUID,

src/apps/schedule/service/schedule_history.py

@@ -24,6 +24,9 @@ class ScheduleHistoryService:
     def __init__(self, session):
         self.session = session
 
+    async def get_by_id(self, id_version: str) -> EventHistorySchema | None:
+        return await ScheduleHistoryCRUD(self.session).get_by_id(id_version)
+
     async def add_history(self, applet_id: uuid.UUID, event: ScheduleEvent):
         applet = await AppletsCRUD(self.session).get_by_id(applet_id)
 

src/apps/users/services/user_device.py

@@ -9,6 +9,11 @@ def __init__(self, session, user_id: uuid.UUID) -> None:
         self.session = session
         self.user_id = user_id
 
+    async def get_by_device_id(self, device_id: str) -> UserDevice | None:
+        schema = await UserDevicesCRUD(self.session).get_by_device_id(device_id)
+
+        return UserDevice.from_schema(schema) if schema else None
+
     async def add_device(self, data: UserDeviceCreate) -> UserDevice:
         app_data = dict(
             app_version=data.app_version,

src/infrastructure/database/migrations/versions/2025_02_24_16_10-add_event_history_id_and_device_id_to_.py

@@ -0,0 +1,26 @@
+"""Add event_history_id and device_id to answers table
+
+Revision ID: 5af378151328
+Revises: 70987d489b17
+Create Date: 2025-02-24 16:10:44.661177
+
+"""
+
+import sqlalchemy as sa
+from alembic import op
+
+# revision identifiers, used by Alembic.
+revision = "5af378151328"
+down_revision = "4e2b42e69c39"
+branch_labels = None
+depends_on = None
+
+
+def upgrade() -> None:
+    op.add_column("answers", sa.Column("event_history_id", sa.String(), nullable=True))
+    op.add_column("answers", sa.Column("device_id", sa.Text(), nullable=True))
+
+
+def downgrade() -> None:
+    op.drop_column("answers", "device_id")
+    op.drop_column("answers", "event_history_id")

src/infrastructure/database/migrations/versions/2025_02_27_11_56-merge_heads.py

@@ -0,0 +1,21 @@
+"""Merge heads after prolific branching
+
+Revision ID: a620e11dda13
+Revises: 5af378151328, 11b5cd083450
+Create Date: 2025-02-27 11:56:06.642160
+
+"""
+
+# revision identifiers, used by Alembic.
+revision = "a620e11dda13"
+down_revision = ("5af378151328", "11b5cd083450")
+branch_labels = None
+depends_on = None
+
+
+def upgrade() -> None:
+    pass
+
+
+def downgrade() -> None:
+    pass

src/infrastructure/database/migrations_arbitrary/versions/2025_02_24_16_10-add_event_history_id_and_device_id_to_.py

@@ -0,0 +1,27 @@
+"""Add event_history_id and device_id to answers table
+
+Revision ID: 5af378151328
+Revises: 70987d489b17
+Create Date: 2025-02-24 16:10:44.661177
+
+"""
+
+import sqlalchemy as sa
+from alembic import op
+from sqlalchemy.dialects import postgresql
+
+# revision identifiers, used by Alembic.
+revision = "5af378151328"
+down_revision = "3d8602537b1d"
+branch_labels = None
+depends_on = None
+
+
+def upgrade() -> None:
+    op.add_column("answers", sa.Column("event_history_id", sa.String(), nullable=True))
+    op.add_column("answers", sa.Column("device_id", sa.Text(), nullable=True))
+
+
+def downgrade() -> None:
+    op.drop_column("answers", "device_id")
+    op.drop_column("answers", "event_history_id")