Merge pull request #35 from teams-notifier/feat/ooo-protection

feat: out-of-order protection

Author: babs

.gitignore

@@ -6,3 +6,4 @@ misc
 .coverage
 .pytest_cache
 *.pyc
+gitlab_mr_api.egg-info

.pre-commit-config.yaml

@@ -1,10 +1,12 @@
 # See https://pre-commit.com for more information
 # See https://pre-commit.com/hooks.html for more hooks
 repos:
-  - repo: https://github.com/psf/black
-    rev: 24.10.0
+  - repo: https://github.com/astral-sh/ruff-pre-commit
+    rev: v0.8.3
     hooks:
-      - id: black
+      - id: ruff
+        args: [--fix, --exit-non-zero-on-fix]
+      - id: ruff-format
 
   - repo: https://github.com/pre-commit/pre-commit-hooks
     rev: v5.0.0
@@ -21,19 +23,6 @@ repos:
       - id: debug-statements
         language_version: python3
 
-  - repo: https://github.com/PyCQA/flake8
-    rev: 7.1.1
-    hooks:
-      - id: flake8
-        args: [--max-line-length, "110"]
-        language_version: python3
-
-  - repo: https://github.com/asottile/reorder_python_imports
-    rev: v3.13.0
-    hooks:
-      - id: reorder-python-imports
-        args: ["--application-directories=.:src", "--py36-plus"]
-
   - repo: https://github.com/asottile/pyupgrade
     rev: v3.17.0
     hooks:

app.py

@@ -5,12 +5,14 @@
 import sys
 import traceback
 import uuid
+
 from contextlib import asynccontextmanager
 from typing import Annotated
 
 import asyncpg
 import fastapi_structured_logging
 import httpx
+
 from fastapi import FastAPI
 from fastapi import Header
 from fastapi import HTTPException
@@ -20,12 +22,15 @@
 from fastapi.responses import RedirectResponse
 
 import webhook
+
 from config import DefaultConfig
 from db import database
 from gitlab_model import EmojiPayload
 from gitlab_model import MergeRequestPayload
 from gitlab_model import PipelinePayload
 from periodic_cleanup import periodic_cleanup
+from webhook.merge_request import PartialMessageUpdateError
+
 
 config = DefaultConfig()
 
@@ -163,7 +168,7 @@ async def handle_webhook(
                     raise HTTPException(
                         status_code=400,
                         detail="filter_on_participant_ids must be a list of comma separated integers",
-                    )
+                    ) from None
 
             await webhook.merge_request(
                 payload,
@@ -176,11 +181,22 @@ async def handle_webhook(
         if isinstance(payload, EmojiPayload):
             await webhook.emoji(payload, conversation_tokens)
         return {"status": "ok"}
+    except PartialMessageUpdateError as exc:
+        raise HTTPException(
+            status_code=502,
+            detail={
+                "error": "partial_message_update_failure",
+                "message": str(exc),
+                "failed": exc.failed,
+                "succeeded": exc.succeeded,
+                "total": exc.total,
+            },
+        ) from exc
     except httpx.HTTPStatusError as exc:
         raise HTTPException(
             status_code=exc.response.status_code,
             detail=exc.response.json(),
-        )
+        ) from exc
 
 
 @app.get("/healthz", include_in_schema=False)
@@ -197,7 +213,7 @@ async def healthcheck():
             error_detail=str(e),
             exc_info=True,
         )
-        raise HTTPException(status_code=500, detail=f"{type(e)}: {e}")
+        raise HTTPException(status_code=500, detail=f"{type(e)}: {e}") from e
 
 
 if __name__ == "__main__":

cards/render.py

@@ -1,10 +1,12 @@
 #!/usr/bin/env python3
 import datetime
 import json
+
 from enum import Enum
 from typing import Any
 
 import yaml
+
 from jinja2 import Environment
 from jinja2 import FileSystemLoader
 from jinja2 import Template

config.py

@@ -3,6 +3,7 @@
 
 import dotenv
 
+
 dotenv.load_dotenv()
 
 __all__ = ["DefaultConfig", "config"]
@@ -20,7 +21,7 @@ class DefaultConfig:
     _valid_tokens: list[str]
 
     def __init__(self):
-        self._valid_tokens = list([t.strip() for t in self.VALID_X_GITLAB_TOKEN.lower().split(",")])
+        self._valid_tokens = [t.strip() for t in self.VALID_X_GITLAB_TOKEN.lower().split(",")]
         self.log_queries = False
         if len(self.LOG_QUERIES) and self.LOG_QUERIES[0].lower() in ("y", "t", "1"):
             self.log_queries = True

db.py

@@ -2,19 +2,22 @@
 import asyncio
 import json
 import urllib.parse
+
 from typing import Any
 from typing import Literal
 
 import asyncpg.connect_utils
 import fastapi_structured_logging
+
 from pydantic import BaseModel
 
-from config import config
 from config import DefaultConfig
+from config import config
 from gitlab_model import GLEmojiAttributes
 from gitlab_model import MergeRequestPayload
 from gitlab_model import PipelinePayload
 
+
 log = fastapi_structured_logging.get_logger()
 
 __all__ = ["database", "dbh"]
@@ -138,7 +141,77 @@ async def get_gitlab_instance_id_from_url(self, urlstr: str) -> int:
         assert isinstance(gli_id, int)
         return gli_id
 
+    async def get_or_create_merge_request_ref_id(self, merge_request: MergeRequestPayload) -> int:
+        """
+        Get or create an MR ref without updating payload.
+
+        Used for OOO check - we need the ID to query message refs,
+        but we don't want to corrupt the payload with stale data.
+        Only sets initial state on INSERT, never updates existing records.
+        """
+        gitlab_instance_id = await self.get_gitlab_instance_id_from_url(merge_request.object_attributes.url)
+
+        merge_ref_id = await self._generic_norm_upsert(
+            table="merge_request_ref",
+            identity_col="merge_request_ref_id",
+            select_attrs={
+                "gitlab_instance_id": gitlab_instance_id,
+                "gitlab_project_id": merge_request.object_attributes.target_project_id,
+                "gitlab_merge_request_iid": merge_request.object_attributes.iid,
+            },
+            insert_only_vals={
+                "gitlab_merge_request_id": merge_request.object_attributes.id,
+                "head_pipeline_id": merge_request.object_attributes.head_pipeline_id,
+                "merge_request_payload": merge_request.model_dump(),
+                "merge_request_extra_state": {
+                    "version": 1,
+                    "opener": {
+                        "id": merge_request.user.id,
+                        "name": merge_request.user.name,
+                        "username": merge_request.user.username,
+                    },
+                    "approvers": {},
+                    "pipeline_statuses": {},
+                    "emojis": {},
+                },
+            },
+        )
+        assert isinstance(merge_ref_id, int)
+        return merge_ref_id
+
+    async def update_merge_request_ref_payload(
+        self, merge_request_ref_id: int, merge_request: MergeRequestPayload
+    ) -> MergeRequestInfos:
+        """
+        Update MR ref payload after OOO check has passed.
+
+        Called only for non-OOO events to update the stored payload.
+        """
+        connection: asyncpg.Connection
+        async with await database.acquire() as connection:
+            row = await connection.fetchrow(
+                """UPDATE merge_request_ref
+                   SET gitlab_merge_request_id = $1,
+                       head_pipeline_id = $2,
+                       merge_request_payload = $3
+                   WHERE merge_request_ref_id = $4
+                   RETURNING merge_request_ref_id, merge_request_payload,
+                             merge_request_extra_state, head_pipeline_id""",
+                merge_request.object_attributes.id,
+                merge_request.object_attributes.head_pipeline_id,
+                merge_request.model_dump(),
+                merge_request_ref_id,
+            )
+            assert row is not None
+            return MergeRequestInfos(**row)
+
     async def get_merge_request_ref_infos(self, merge_request: MergeRequestPayload) -> MergeRequestInfos:
+        """
+        Get or create MR ref AND update payload (legacy behavior).
+
+        Note: This updates payload on every call. For OOO-safe behavior,
+        use get_or_create_merge_request_ref_id() + update_merge_request_ref_payload().
+        """
         gitlab_instance_id = await self.get_gitlab_instance_id_from_url(merge_request.object_attributes.url)
 
         merge_ref = await self._generic_norm_upsert(
@@ -212,7 +285,6 @@ async def _generic_norm_upsert(
         insert_only_vals: dict[str, Any] | None = None,
         extra_sel_cols: list[str] | None = None,
     ) -> Any:
-
         if extra_insert_and_update_vals is None:
             extra_insert_and_update_vals = {}
 

db/migrations/20251210000000_add_last_processed_updated_at.sql

@@ -0,0 +1,7 @@
+-- migrate:up
+ALTER TABLE gitlab_mr_api.merge_request_message_ref
+ADD COLUMN last_processed_updated_at timestamp with time zone;
+
+-- migrate:down
+ALTER TABLE gitlab_mr_api.merge_request_message_ref
+DROP COLUMN last_processed_updated_at;

db/schema.sql

@@ -82,7 +82,8 @@ CREATE TABLE gitlab_mr_api.merge_request_message_ref (
     failure jsonb,
     created_at timestamp with time zone DEFAULT now() NOT NULL,
     updated_at timestamp with time zone,
-    last_processed_fingerprint character varying(64)
+    last_processed_fingerprint character varying(64),
+    last_processed_updated_at timestamp with time zone
 );
 
 

gitlab_model.py

@@ -23,6 +23,7 @@ class GLUser(BaseModel, extra="allow"):
 
 
 class GLProject(BaseModel, extra="allow"):
+    id: int
     path_with_namespace: str
     web_url: str
 
@@ -36,7 +37,7 @@ class GLMRAttributes(BaseModel, extra="allow"):
     state: str
     url: str
     action: str
-    updated_at: str | None
+    updated_at: str
     oldrev: str | None = None
 
     # https://docs.gitlab.com/ee/api/merge_requests.html#merge-status
@@ -80,6 +81,7 @@ class PipelinePayload(BaseModel, extra="allow"):
     object_kind: Literal["pipeline"]
     object_attributes: GLPipelineAttributes
     builds: list[GLPipelineBuild]
+    project: GLProject
 
 
 class GLEmojiAttributes(BaseModel, extra="allow"):

periodic_cleanup.py

@@ -9,6 +9,7 @@
 from config import DefaultConfig
 from db import DatabaseLifecycleHandler
 
+
 logger = fastapi_structured_logging.get_logger()
 
 signal = asyncio.Event()