feat(M2-9862): Out-of-Order Flow Submissions with Duplicate Activities (#1954)

* feat(M2-9862): Fix out-of-order flow submissions and handle duplicate activities

- Added FlowSubmissionProgress helper for backend-owned flow tracking
- Replaced order-based logic with Counter-based validation
- Persisted backend-computed flow completion
- Logged mismatches for legacy clients
- Added new out-of-order tests and updated existing ones

* fix: add back create_at and  convert Unix timestamp to datetime in answer existence check

* M2-9862 make the create_at optional

* fix(M2-9862): fix linting using ruff

Author: maxto024

src/apps/answers/api.py

@@ -873,12 +873,12 @@ async def answers_existence_check(
     await AppletService(session, user.id).exist_by_id(schema.applet_id)
     await CheckAccessService(session, user.id).check_answer_check_access(schema.applet_id)
     is_exist = await AnswerService(session, user.id, answer_session).is_answers_uploaded(
-        schema.applet_id, schema.activity_id, schema.submit_id
+        schema.applet_id, schema.activity_id, schema.submit_id, schema.created_at
     )
 
     logger.info(
         f"check-existence: applet_id={schema.applet_id}, activity_id={schema.activity_id}, user_id={user.id}, "
-        f"submit_id={schema.submit_id}, exists={is_exist}, ip={client_ip}"
+        f"submit_id={schema.submit_id}, created_at={schema.created_at}, exists={is_exist}, ip={client_ip}"
     )
 
     return Response[AnswerExistenceResponse](result=AnswerExistenceResponse(exists=is_exist))

src/apps/answers/crud/answers.py

@@ -518,16 +518,22 @@ async def get_by_applet_activity_submit_or_user_id(
         activity_id: str,
         user_id: uuid.UUID | None = None,
         submit_id: uuid.UUID | None = None,
+        created_at: int | None = None,
     ) -> list[AnswerSchema]:
-        # We're not using created_at for filtering as it causes issues with mobile submissions
-        # The combination of applet_id, activity_id, and either user_id or submit_id should be sufficient
+        # created_at is used to distinguish between duplicate activities in flows
         query: Query = select(AnswerSchema)
         query = query.where(AnswerSchema.applet_id == applet_id)
         query = query.filter(AnswerSchema.activity_history_id.startswith(activity_id))
         if submit_id:
             query = query.where(AnswerSchema.submit_id == submit_id)
         if user_id:
             query = query.where(AnswerSchema.respondent_id == user_id)
+        if created_at is not None:
+            # Convert Unix timestamp (milliseconds) to datetime for comparison
+            created_at_datetime = datetime.datetime.fromtimestamp(
+                created_at / 1000.0, tz=datetime.timezone.utc
+            ).replace(tzinfo=None)
+            query = query.where(AnswerSchema.created_at == created_at_datetime)
 
         db_result = await self._execute(query)
         return db_result.scalars().all()

src/apps/answers/domain/answers.py

@@ -673,9 +673,7 @@ class AppletCompletedEntities(InternalModel):
 
 class AnswersCheck(PublicModel):
     applet_id: uuid.UUID
-    # TODO: created_at can be safely removed after
-    # the corresponding mobile PR is merged
-    # https://mindlogger.atlassian.net/browse/M2-9693
+    # Used to distinguish between duplicate activities in flows
     created_at: int | None = None
     activity_id: str
     submit_id: uuid.UUID | None = None

src/apps/answers/flow_submission_progress.py

@@ -0,0 +1,74 @@
+from __future__ import annotations
+
+from collections import Counter
+from typing import Iterable
+
+from sqlalchemy.ext.asyncio import AsyncSession
+
+from apps.activity_flows.crud import FlowsHistoryCRUD
+from apps.answers.crud.answers import AnswersCRUD
+
+
+class FlowSubmissionProgress:
+    """Track flow submission progress using occurrence counting."""
+
+    def __init__(self, session: AsyncSession, answer_session: AsyncSession):
+        self._session = session
+        self._answer_session = answer_session
+        self._expected_counts: Counter[str] = Counter()
+        self._expected_ids: set[str] = set()
+        self._submitted_counts: Counter[str] = Counter()
+        self._has_flow_history: bool = False
+
+    async def load(self, flow_history_id: str, submit_id) -> None:
+        """Load flow structure and existing submissions for the submit id."""
+        flow_histories = await FlowsHistoryCRUD(self._session).load_full([flow_history_id], load_activities=False)
+        if not flow_histories:
+            self._has_flow_history = False
+            return
+
+        flow_history = flow_histories[0]
+        self._expected_counts = Counter(item.activity_id for item in flow_history.items)
+        self._expected_ids = set(self._expected_counts.keys())
+        self._has_flow_history = True
+
+        existing_answers = await AnswersCRUD(self._answer_session).get_by_submit_id(submit_id)
+        self._submitted_counts = Counter(answer.activity_history_id for answer in existing_answers or [])
+
+    @property
+    def has_flow_history(self) -> bool:
+        return self._has_flow_history
+
+    def is_complete_before_current(self) -> bool:
+        return self._all_expected_satisfied(self._submitted_counts.items())
+
+    def can_accept(self, activity_history_id: str) -> bool:
+        expected_total = self._expected_counts.get(activity_history_id, 0)
+        if expected_total == 0:
+            return False
+        return self._submitted_counts.get(activity_history_id, 0) < expected_total
+
+    def completion_state_after_add(self, activity_history_id: str) -> bool:
+        temp_counts = self._submitted_counts.copy()
+        temp_counts[activity_history_id] += 1
+        return self._all_expected_satisfied(temp_counts.items())
+
+    def contains_activity(self, activity_history_id: str) -> bool:
+        if not self._has_flow_history:
+            return False
+        return activity_history_id in self._expected_ids
+
+    @property
+    def expected_total(self) -> int:
+        return sum(self._expected_counts.values())
+
+    @property
+    def submitted_total(self) -> int:
+        return sum(self._submitted_counts.values())
+
+    def _all_expected_satisfied(self, submitted_items: Iterable[tuple[str, int]]) -> bool:
+        submitted_map = dict(submitted_items)
+        for activity_history_id, expected_count in self._expected_counts.items():
+            if submitted_map.get(activity_history_id, 0) < expected_count:
+                return False
+        return True

src/apps/answers/service.py

@@ -6,7 +6,7 @@
 import os
 import time
 import uuid
-from collections import Counter, defaultdict
+from collections import defaultdict
 from json import JSONDecodeError
 from typing import Callable, List, Mapping, Optional
 
@@ -89,6 +89,7 @@
     WrongRespondentForAnswerGroup,
 )
 from apps.answers.filters import AppletSubmitDateFilter, ReviewAppletItemFilter, SummaryActivityFilter
+from apps.answers.flow_submission_progress import FlowSubmissionProgress
 from apps.answers.tasks import create_report
 from apps.applets.crud import AppletsCRUD
 from apps.applets.domain.applet_history import Version
@@ -150,46 +151,54 @@ async def create_answer(self, activity_answer: AppletAnswerCreate, device_id: st
     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, device_id)
+        flow_progress = await self._validate_respondent_answer(activity_answer)
+        return await self._create_answer(activity_answer, device_id, flow_progress)
 
     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, device_id)
+        flow_progress = await self._validate_anonymous_answer(activity_answer)
+        return await self._create_answer(activity_answer, device_id, flow_progress)
 
-    async def _validate_respondent_answer(self, activity_answer: AppletAnswerCreate) -> None:
-        await self._validate_answer(activity_answer)
+    async def _validate_respondent_answer(self, activity_answer: AppletAnswerCreate) -> FlowSubmissionProgress | None:
+        flow_progress = await self._validate_answer(activity_answer)
         await self._validate_applet_for_user_response(activity_answer.applet_id)
 
-    async def _validate_anonymous_answer(self, activity_answer: AppletAnswerCreate) -> None:
+        return flow_progress
+
+    async def _validate_anonymous_answer(self, activity_answer: AppletAnswerCreate) -> FlowSubmissionProgress | None:
         await self._validate_applet_for_anonymous_response(activity_answer.applet_id, activity_answer.version)
-        await self._validate_answer(activity_answer)
+        return await self._validate_answer(activity_answer)
 
-    async def _validate_answer(self, applet_answer: AppletAnswerCreate) -> None:  # noqa: C901
+    async def _validate_answer(self, applet_answer: AppletAnswerCreate) -> FlowSubmissionProgress | None:  # noqa: C901
         pk = self._generate_history_id(applet_answer.version)
+
+        # Timestamp-based duplicate detection: when created_at provided, check for exact duplicate
+        # Each unique timestamp represents a distinct submission, bypassing occurrence limits
+        if applet_answer.created_at is not None:
+            created_at_ms = int(applet_answer.created_at.timestamp() * 1000)
+            existing_with_timestamp = await AnswersCRUD(self.answer_session).get_by_applet_activity_submit_or_user_id(
+                applet_answer.applet_id,
+                str(applet_answer.activity_id),
+                None,
+                applet_answer.submit_id,
+                created_at_ms,
+            )
+            if existing_with_timestamp:
+                raise ValidationError("Duplicate answer with same timestamp already exists")
+
         existed_answers = await AnswersCRUD(self.answer_session).get_by_submit_id(applet_answer.submit_id)
 
         activity_history_id = pk(applet_answer.activity_id)
         flow_history_id = pk(applet_answer.flow_id) if applet_answer.flow_id else None
-
-        activity_indexes = set()  # same activity is allowed multiple times in flow
-        latest_activity_index = None
-        if flow_history_id:
-            flow_histories = await FlowsHistoryCRUD(self.session).load_full(
-                [pk(applet_answer.flow_id)], load_activities=False
-            )
-            if not flow_histories:
+        flow_progress: FlowSubmissionProgress | None = None
+        # Only use occurrence-based flow validation when created_at is NOT provided
+        if flow_history_id and applet_answer.created_at is None:
+            flow_progress = FlowSubmissionProgress(self.session, self.answer_session)
+            await flow_progress.load(flow_history_id, applet_answer.submit_id)
+            if not flow_progress.has_flow_history:
                 raise ValidationError("Flow not found")
-            flow_history = next(iter(flow_histories))
-
-            # check activity in the flow
-            for i, item in enumerate(flow_history.items):
-                if item.activity_id == activity_history_id:
-                    activity_indexes.add(i)
-            latest_activity_index = len(flow_history.items) - 1
-            if not activity_indexes:
+            if not flow_progress.contains_activity(activity_history_id):
                 raise ValidationError("Activity not found in the flow")
 
         if existed_answers:
@@ -208,60 +217,28 @@ async def _validate_answer(self, applet_answer: AppletAnswerCreate) -> None:  #
             if flow_history_id != existed_answer.flow_history_id:
                 raise ValidationError("Submit id duplicate error")
 
-            # check current answer is provided in right order in the flow, so prev activities already answered
-            prev_answers_count = len(existed_answers)
-            is_flow_completed = any(answer.is_flow_completed for answer in existed_answers)
-
-            # Smart flow completion check
-            if is_flow_completed:
-                # Count expected and already persisted activity occurrences
-                flow_activity_counts = Counter(item.activity_id for item in flow_history.items)
-                submitted_counts = Counter(answer.activity_history_id for answer in existed_answers)
-
-                # If all activities already persisted before this submission, the flow truly finished
-                if all(submitted_counts.get(act_id, 0) >= count for act_id, count in flow_activity_counts.items()):
-                    raise ValidationError("Flow is already completed")
+        if flow_progress:
+            if flow_progress.is_complete_before_current():
+                raise ValidationError("Flow is already completed")
 
-                current_expected_total = flow_activity_counts.get(activity_history_id, 0)
-                current_submitted = submitted_counts.get(activity_history_id, 0)
-
-                # Reject duplicates that exceed expected occurrences for the activity
-                if current_submitted >= current_expected_total:
-                    raise ValidationError("Flow is already completed")
+            if not flow_progress.can_accept(activity_history_id):
+                raise ValidationError("Activity submission exceeds expected occurrences for this flow")
 
+            if existed_answers:
                 logger.info(
-                    "Allowing late submission for flow %s, activity %s, submit_id %s",
+                    "Allowing flow submission for flow %s, activity %s, submit_id %s",
                     flow_history_id,
                     activity_history_id,
                     applet_answer.submit_id,
                 )
 
-            # Continue with existing order validation only if flow not marked complete
-            # When is_flow_completed=True, we allow out-of-order submissions
-            if not is_flow_completed and prev_answers_count not in activity_indexes:
-                assert latest_activity_index is not None
-                # allow latest activity for flow autocompletion FE logic
-                if not (
-                    prev_answers_count < latest_activity_index + 1
-                    and max(activity_indexes) == latest_activity_index
-                    and applet_answer.is_flow_completed
-                ):
-                    raise ValidationError("Wrong activity order in the flow")
-
-        elif flow_history_id and 0 not in activity_indexes:
-            # check first flow answer - but allow if flow is marked as complete
-            # Check both existing answers and current answer for is_flow_completed
-            if not (
-                (existed_answers and any(answer.is_flow_completed for answer in existed_answers))
-                or applet_answer.is_flow_completed
-            ):
-                raise ValidationError("Wrong activity order in the flow")
-
         activity_history = await ActivityHistoriesCRUD(self.session).get_by_id(activity_history_id)
 
         if not activity_history.applet_id.startswith(f"{applet_answer.applet_id}"):
             raise ActivityHistoryDoeNotExist()
 
+        return flow_progress
+
     async def _validate_applet_for_anonymous_response(self, applet_id: uuid.UUID, version: str) -> None:
         await AppletHistoryService(self.session, applet_id, version).get()
         # Validate applet for anonymous answer
@@ -346,12 +323,41 @@ async def _get_answer_relation(
 
         return relation.relation
 
-    async def _create_answer(self, applet_answer: AppletAnswerCreate, device_id: str | None) -> AnswerSchema:
+    async def _create_answer(
+        self,
+        applet_answer: AppletAnswerCreate,
+        device_id: str | None,
+        flow_progress: FlowSubmissionProgress | 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)
         subject_crud = SubjectsCrud(self.session)
 
+        activity_history_id = pk(applet_answer.activity_id)
+        flow_history_id = pk(applet_answer.flow_id) if applet_answer.flow_id else None
+        client_flow_completed_flag = bool(applet_answer.is_flow_completed) if applet_answer.flow_id else None
+
+        is_flow_completed_backend = None
+        if applet_answer.flow_id:
+            if flow_progress:
+                is_flow_completed_backend = flow_progress.completion_state_after_add(activity_history_id)
+                if client_flow_completed_flag is not None and client_flow_completed_flag != is_flow_completed_backend:
+                    logger.info(
+                        "Flow completion mismatch for flow %s, activity %s, submit_id %s: client=%s backend=%s",
+                        flow_history_id,
+                        activity_history_id,
+                        applet_answer.submit_id,
+                        client_flow_completed_flag,
+                        is_flow_completed_backend,
+                    )
+            else:
+                is_flow_completed_backend = client_flow_completed_flag
+
+        migrated_data = None
+        if client_flow_completed_flag is not None:
+            migrated_data = {"client_flow_completed_flag": client_flow_completed_flag}
+
         respondent_subject = await subject_crud.get_user_subject(
             user_id=self.user_id, applet_id=applet_answer.applet_id
         )
@@ -413,18 +419,19 @@ async def _create_answer(self, applet_answer: AppletAnswerCreate, device_id: str
                 applet_id=applet_answer.applet_id,
                 version=applet_answer.version,
                 applet_history_id=pk(applet_answer.applet_id),
-                flow_history_id=pk(applet_answer.flow_id) if applet_answer.flow_id else None,
-                activity_history_id=pk(applet_answer.activity_id),
+                flow_history_id=flow_history_id,
+                activity_history_id=activity_history_id,
                 respondent_id=self.user_id,
                 client=applet_answer.client.dict(),
-                is_flow_completed=bool(applet_answer.is_flow_completed) if applet_answer.flow_id else None,
+                is_flow_completed=is_flow_completed_backend,
                 target_subject_id=target_subject.id,
                 source_subject_id=source_subject.id,
                 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,
+                migrated_data=migrated_data,
             )
         )
         item_answer = applet_answer.answer
@@ -1724,11 +1731,11 @@ async def get_completed_answers_data_list(
         return result
 
     async def is_answers_uploaded(
-        self, applet_id: uuid.UUID, activity_id: str, submit_id: uuid.UUID | None = None
+        self, applet_id: uuid.UUID, activity_id: str, submit_id: uuid.UUID | None = None, created_at: int | None = None
     ) -> bool:
         # check by submit id if provided otherwise by user_id
         answers = await AnswersCRUD(self.answer_session).get_by_applet_activity_submit_or_user_id(
-            applet_id, activity_id, self.user_id if not submit_id else None, submit_id
+            applet_id, activity_id, self.user_id if not submit_id else None, submit_id, created_at
         )
         if not answers:
             return False

src/apps/answers/tests/conftest.py

@@ -235,7 +235,7 @@ def answer_create(
         submit_id=uuid.uuid4(),
         activity_id=applet.activities[0].id,
         answer=answer_item_create,
-        created_at=datetime.datetime.now(datetime.UTC).replace(microsecond=0),
+        created_at=None,  # None by default - uses occurrence-based validation
         client=client_meta,
         consent_to_share=False,
     )