From 643f41d45fa659ba93e8e3853cc4fd30deb872ef Mon Sep 17 00:00:00 2001 From: Gautier DI FOLCO Date: Thu, 20 Aug 2026 13:04:10 +0200 Subject: [PATCH 1/5] WPB-28162: version-aware event delivery Add a Transmit type class and a dispatcher over stored untyped event objects, applied on both delivery channels using the API version of the endpoint the client called. Meeting-related events (conversation.create-meeting, conversation.delete-meeting, meeting.*) are not delivered to clients below V15, the version meetings were introduced in. - wire-api: Transmit class (Wire.API.Notification), family instances (Wire.API.Event.Conversation, Wire.API.Event.Meeting), and dispatcher Wire.API.Event.Transmit (unknown types always delivered, gated types fail closed on decode errors, >= V15 fast path returns stored bytes) - gundeck: APIVersion prefix on all four notification routes; paginate filters survivors with a bounded refill loop so an all-gated page cannot strand the client cursor; fetchId/fetchLast post-filter - cannon: APIVersion prefix on all four websocket routes; per-connection wsApiVersion; legacy push and RMQ paths encode per-target version; filtered drops are acked/reported OK so gundeck does not fall back to native push - tests: unit Test.Wire.API.Event.Transmit incl. encoding drift guards; integration Test.NotificationsVersioned (REST + websocket); Testlib.Cannon can open websockets at an explicit API version Storage, fan-out and native push are unchanged; modern clients receive byte-identical payloads. --- .../wpb-28162-versioned-event-delivery | 9 + integration/integration.cabal | 1 + integration/test/Test/Events.hs | 35 ++- .../test/Test/NotificationsVersioned.hs | 218 ++++++++++++++++++ integration/test/Testlib/Cannon.hs | 18 +- .../src/Wire/API/Event/Conversation.hs | 6 + libs/wire-api/src/Wire/API/Event/Meeting.hs | 7 + libs/wire-api/src/Wire/API/Event/Transmit.hs | 106 +++++++++ libs/wire-api/src/Wire/API/Notification.hs | 9 + .../src/Wire/API/Routes/Public/Cannon.hs | 5 + .../src/Wire/API/Routes/Public/Gundeck.hs | 6 + .../test/unit/Test/Wire/API/Event/Transmit.hs | 159 +++++++++++++ libs/wire-api/test/unit/Test/Wire/API/Run.hs | 2 + libs/wire-api/wire-api.cabal | 2 + services/cannon/src/Cannon/API/Internal.hs | 43 ++-- services/cannon/src/Cannon/API/Public.hs | 15 +- services/cannon/src/Cannon/App.hs | 7 +- .../cannon/src/Cannon/RabbitMqConsumerApp.hs | 29 ++- services/cannon/src/Cannon/WS.hs | 11 +- services/gundeck/src/Gundeck/API/Public.hs | 20 +- services/gundeck/src/Gundeck/Notification.hs | 50 +++- 21 files changed, 692 insertions(+), 66 deletions(-) create mode 100644 changelog.d/1-api-changes/wpb-28162-versioned-event-delivery create mode 100644 integration/test/Test/NotificationsVersioned.hs create mode 100644 libs/wire-api/src/Wire/API/Event/Transmit.hs create mode 100644 libs/wire-api/test/unit/Test/Wire/API/Event/Transmit.hs diff --git a/changelog.d/1-api-changes/wpb-28162-versioned-event-delivery b/changelog.d/1-api-changes/wpb-28162-versioned-event-delivery new file mode 100644 index 00000000000..94fc0afc469 --- /dev/null +++ b/changelog.d/1-api-changes/wpb-28162-versioned-event-delivery @@ -0,0 +1,9 @@ +Event delivery is now version-aware: gundeck REST endpoints (`GET +/notifications`, by-id, last) and cannon websockets filter stored events by +the API version of the requesting client. Meeting-related events +(`conversation.create-meeting`, `conversation.delete-meeting`, +`meeting.create`, `meeting.update`, `meeting.delete`, `meeting.member-add`) +are not delivered to clients that called the API below V15, the version +meetings were introduced in. Storage, fan-out and native push are unchanged; +modern clients receive byte-identical payloads. Unknown event types are always +delivered for forward compatibility. diff --git a/integration/integration.cabal b/integration/integration.cabal index d124256a118..a95974c2fe5 100644 --- a/integration/integration.cabal +++ b/integration/integration.cabal @@ -196,6 +196,7 @@ library Test.MLS.Unreachable Test.NginxZAuthModule Test.Notifications + Test.NotificationsVersioned Test.OAuth Test.One2OneTeamConv Test.PasswordReset diff --git a/integration/test/Test/Events.hs b/integration/test/Test/Events.hs index a172cf2ff0f..50cdaa6b20a 100644 --- a/integration/test/Test/Events.hs +++ b/integration/test/Test/Events.hs @@ -595,7 +595,7 @@ testChannelLimit = withModifiedBackend -- the first client fails to connect because the server runs out of channels do - eithWS <- createEventsWebSocketEither alice (Just client0) Nothing + eithWS <- createEventsWebSocketEither alice (Just client0) Nothing Nothing case eithWS of Left (WS.MalformedResponse respHead _) -> lift $ respHead.responseCode `shouldMatchInt` 503 @@ -958,16 +958,21 @@ createEventWebSockets :: Codensity App [EventWebSocket] createEventWebSockets = traverse (uncurry createEventsWebSocket) +requireConnectedWebSocket :: + (HasCallStack) => + Either WS.HandshakeException EventWebSocket -> + Codensity App EventWebSocket +requireConnectedWebSocket = \case + Left e -> lift $ assertFailure $ "Websocket failed to connect due to handshake exception: " <> displayException e + Right ws -> pure ws + createEventsWebSocket :: (HasCallStack, MakesValue user) => user -> Maybe String -> Codensity App EventWebSocket -createEventsWebSocket user cid = do - eithWS <- createEventsWebSocketEither user cid Nothing - case eithWS of - Left e -> lift $ assertFailure $ "Websocket failed to connect due to handshake exception: " <> displayException e - Right ws -> pure ws +createEventsWebSocket user cid = + createEventsWebSocketEither user cid Nothing Nothing >>= requireConnectedWebSocket createEventsWebSocketWithSync :: (HasCallStack, MakesValue user) => @@ -976,22 +981,32 @@ createEventsWebSocketWithSync :: Codensity App (String, EventWebSocket) createEventsWebSocketWithSync user cid = do syncMarker <- lift randomId - eithWS <- createEventsWebSocketEither user cid (Just syncMarker) - case eithWS of + createEventsWebSocketEither user cid (Just syncMarker) Nothing >>= \case Left e -> lift $ assertFailure $ "Websocket failed to connect due to handshake exception: " <> displayException e Right ws -> pure (syncMarker, ws) +-- | 'createEventsWebSocket', but connecting at an explicit API version. +createEventsWebSocketAtVersion :: + (HasCallStack, MakesValue user) => + user -> + Maybe String -> + Int -> + Codensity App EventWebSocket +createEventsWebSocketAtVersion user cid v = + createEventsWebSocketEither user cid Nothing (Just v) >>= requireConnectedWebSocket + createEventsWebSocketEither :: (HasCallStack, MakesValue user) => user -> Maybe String -> Maybe String -> + Maybe Int -> Codensity App (Either WS.HandshakeException EventWebSocket) -createEventsWebSocketEither user cid mSyncMarker = do +createEventsWebSocketEither user cid mSyncMarker mApiVersion = do eventsChan <- liftIO newChan ackChan <- liftIO newEmptyMVar serviceMap <- lift $ getServiceMap =<< objDomain user - apiVersion <- lift $ getAPIVersionFor $ objDomain user + apiVersion <- maybe (lift $ getAPIVersionFor $ objDomain user) pure mApiVersion wsStarted <- newEmptyMVar let minAPIVersion = 8 lift diff --git a/integration/test/Test/NotificationsVersioned.hs b/integration/test/Test/NotificationsVersioned.hs new file mode 100644 index 00000000000..1cfcdeb2747 --- /dev/null +++ b/integration/test/Test/NotificationsVersioned.hs @@ -0,0 +1,218 @@ +-- This file is part of the Wire Server implementation. +-- +-- Copyright (C) 2026 Wire Swiss GmbH +-- +-- This program is free software: you can redistribute it and/or modify it under +-- the terms of the GNU Affero General Public License as published by the Free +-- Software Foundation, either version 3 of the License, or (at your option) any +-- later version. +-- +-- This program is distributed in the hope that it will be useful, but WITHOUT +-- ANY WARRANTY; without even the implied warranty of MERCHANTABILITY or FITNESS +-- FOR A PARTICULAR PURPOSE. See the GNU Affero General Public License for more +-- details. +-- +-- You should have received a copy of the GNU Affero General Public License along +-- with this program. If not, see . + +module Test.NotificationsVersioned where + +import API.Brig (addClient, putHandle) +import API.BrigCommon (AddClient (..)) +import API.Common (randomHandle) +import API.Galley +import Control.Monad.Codensity (runCodensity) +import Data.Time.Clock +import Notifications (isConvCreateMeetingNotif, isMeetingCreateNotif) +import SetupHelpers +import Test.Events (ackEvent, assertFindsEvent, consumeAllEventsNoAck, createEventsWebSocketAtVersion, enableConsumableNotifications) +import Test.Meetings (defaultMeetingJson) +import Testlib.Cannon +import Testlib.Prelude +import UnliftIO.Concurrent (threadDelay) + +gatedTypes :: [String] +gatedTypes = + [ "conversation.create-meeting", + "conversation.delete-meeting", + "meeting.create", + "meeting.update", + "meeting.delete", + "meeting.member-add" + ] + +isGated :: String -> Bool +isGated t = t `elem` gatedTypes + +-- | Drain all notification pages at the given API version and return the +-- payload event types seen. Also asserts that pagination terminates (no +-- endless empty 'has_more=true' pages). +drainNotificationsAt :: (HasCallStack, MakesValue user) => user -> Int -> App [String] +drainNotificationsAt user v = go Nothing [] + where + go since acc = do + req <- baseRequest user Gundeck (ExplicitVersion v) "/notifications" + let req' = + req + & addQueryParams + ( [("since", s) | s <- toList since] + <> [("size", "100")] + ) + r <- submit "GET" req' + r.status `shouldMatchInt` 200 + body <- getJSON 200 r + notifications <- body %. "notifications" & asList + types <- + mconcat + <$> for + notifications + ( \n -> do + payload <- n %. "payload" & asList + for payload (\e -> e %. "type" >>= asString) + ) + lastId <- case reverse notifications of + [] -> pure Nothing + (n : _) -> Just <$> (n %. "id" >>= asString) + hasMore <- body %. "has_more" & asBool + if hasMore + then case lastId of + Just l -> go (Just l) (acc <> types) + Nothing -> assertFailure "has_more=true but no notification id for cursor" + else pure (acc <> types) + +mkMeeting :: App Value +mkMeeting = do + now <- liftIO getCurrentTime + let startTime = addUTCTime 3600 now + endTime = addUTCTime 7200 now + pure $ defaultMeetingJson "Versioned meeting" startTime endTime [] + +-- | The meeting creator (who receives the meeting events) must not see them +-- via a V14 fetch, while a current-version fetch of the same window shows +-- them; V14 pagination terminates. +testVersionedNotificationsHideMeetingEvents :: (HasCallStack) => App () +testVersionedNotificationsHideMeetingEvents = do + (alice, _tid, _members) <- createTeam OwnDomain 1 + meeting <- mkMeeting + + withWebSocket alice $ \wsAlice -> do + resp <- postMeetings alice meeting + assertSuccess resp + -- the current-version websocket sees the meeting events + void $ awaitMatch isConvCreateMeetingNotif wsAlice + void $ awaitMatch isMeetingCreateNotif wsAlice + + v14Types <- drainNotificationsAt alice 14 + filter isGated v14Types `shouldMatch` ([] :: [String]) + + curTypes <- drainNotificationsAt alice 17 + curTypes `shouldContain` ["conversation.create-meeting"] + curTypes `shouldContain` ["meeting.create"] + +-- | A V14 client still sees non-meeting events (e.g. conversation.create) +-- while meeting events are filtered from the same window. +testVersionedNotificationsKeepNonMeetingEvents :: (HasCallStack) => App () +testVersionedNotificationsKeepNonMeetingEvents = do + (alice, tid, [bob]) <- createTeam OwnDomain 2 + resp <- + postConversation + alice + defProteus + { qualifiedUsers = [bob], + name = Just "plain conv", + team = Just tid + } + assertSuccess resp + meeting <- mkMeeting + mresp <- postMeetings alice meeting + assertSuccess mresp + + -- bob sees the plain conversation event, but no meeting events (the + -- meeting's gated events go to alice, and none leak to bob at V14). + bobTypes <- drainNotificationsAt bob 14 + bobTypes `shouldContain` ["conversation.create"] + filter isGated bobTypes `shouldMatch` ([] :: [String]) + + aliceTypes <- drainNotificationsAt alice 14 + aliceTypes `shouldContain` ["conversation.create"] + filter isGated aliceTypes `shouldMatch` ([] :: [String]) + +-- | A websocket connected at a low version receives no meeting event frames, +-- while a current-version connection of the same user does. +testVersionedWebSocketFiltersMeetingEvents :: (HasCallStack) => App () +testVersionedWebSocketFiltersMeetingEvents = do + (alice, _tid, _members) <- createTeam OwnDomain 1 + aliceId <- alice %. "id" >>= asString + aliceDomain <- objDomain alice + meeting <- mkMeeting + + let lowV = WSConnect aliceId aliceDomain Nothing (Just "lowconn") (Just 14) + highV = WSConnect aliceId aliceDomain Nothing (Just "highconn") Nothing + + withWebSocket lowV $ \wsLow -> + withWebSocket highV $ \wsHigh -> do + resp <- postMeetings alice meeting + assertSuccess resp + -- current version gets the meeting events ... + void $ awaitMatch isConvCreateMeetingNotif wsHigh + void $ awaitMatch isMeetingCreateNotif wsHigh + -- ... the low version does not (allow some time for delivery) + liftIO $ threadDelay 1_000_000 + assertNoEvent 1 wsLow + +-- | The rabbitmq-backed /events websocket of a low-version client skips (and +-- server-side acks) meeting event frames, while a current-version connection +-- of the same user receives them. Tolerant drain: stray ungated events are +-- allowed on the low socket, gated ones are not. +testVersionedEventsSocketFiltersMeetingEvents :: (HasCallStack) => App () +testVersionedEventsSocketFiltersMeetingEvents = + withModifiedBackend (enableConsumableNotifications def) $ \domain -> do + (alice, _tid, _members) <- createTeam domain 1 + -- mirror the other temp-/events tests in Test.Events: create a + -- consumable-notifications client for alice + void $ addClient alice def {acapabilities = Just ["consumable-notifications"]} >>= getJSON 201 + -- Two temp queues with no client id (each binds userRoutingKey and gets its + -- own version-filtered consumer); sharing a client id would round-robin a + -- single queue and race. + runCodensity (createEventsWebSocketAtVersion alice Nothing 14) $ \wsLow -> + runCodensity (createEventsWebSocketAtVersion alice Nothing 17) $ \wsHigh -> do + meeting <- mkMeeting + postMeetings alice meeting >>= assertSuccess + assertFindsEvent wsHigh $ \e -> do + e %. "type" `shouldMatch` "event" + t <- e %. "data.event.payload.0.type" >>= asString + unless (isGated t) + $ assertFailure ("expected a gated meeting event on the V17 socket, got: " <> t) + ackEvent wsHigh e + -- allow some time for delivery, then tolerate stray ungated events on + -- the low socket but assert that none of them is gated + liftIO $ threadDelay 1_000_000 + drained <- consumeAllEventsNoAck wsLow + types <- traverse (\e -> e %. "data.event.payload.0.type" >>= asString) drained + filter isGated types `shouldMatch` ([] :: [String]) + +-- | A V14 notification cursor is not stranded on an all-gated page: the +-- gundeck refill loop must skip past a fully-gated page (server minimum page +-- size is 100) and still deliver a later ungated event. +testVersionedNotificationsRefillPastGatedBacklog :: (HasCallStack) => App () +testVersionedNotificationsRefillPastGatedBacklog = do + (alice, _tid, _members) <- createTeam OwnDomain 1 + -- 101 meetings (~30-60 s by design) guarantee > 100 gated notification rows + -- for the creator even if galley batches conversation.create-meeting and + -- meeting.create into a single row. + replicateM_ 101 $ do + meeting <- mkMeeting + postMeetings alice meeting >>= assertSuccess + -- The ungated event must land strictly after the gated backlog (gundeck + -- persists notifications synchronously in the request path). If that ever + -- becomes async, user.update could land inside the first 100 rows and this + -- test would silently degrade to never exercising the refill loop. + handle <- randomHandle + putHandle alice handle >>= assertSuccess + + v14 <- drainNotificationsAt alice 14 + v14 `shouldContain` ["user.update"] + filter isGated v14 `shouldMatch` ([] :: [String]) + + v17 <- drainNotificationsAt alice 17 + v17 `shouldContain` ["meeting.create"] diff --git a/integration/test/Testlib/Cannon.hs b/integration/test/Testlib/Cannon.hs index 4e7371a154a..8c86deae202 100644 --- a/integration/test/Testlib/Cannon.hs +++ b/integration/test/Testlib/Cannon.hs @@ -110,7 +110,10 @@ data WSConnect = WSConnect domain :: String, client :: Maybe String, -- | If this is Nothing then a random Z-Connection will be used - conn :: Maybe String + conn :: Maybe String, + -- | Explicit API version prefix for the websocket endpoint (affects + -- versioned event delivery). 'Nothing' = current maximum version. + version :: Maybe Int } class ToWSConnect a where @@ -124,20 +127,20 @@ instance {-# OVERLAPPABLE #-} (MakesValue user) => ToWSConnect user where (domain, uid) <- objQid u mc <- lookupField u "client_id" c <- traverse asString mc - pure (WSConnect uid domain c Nothing) + pure (WSConnect uid domain c Nothing Nothing) instance (MakesValue user, MakesValue conn) => ToWSConnect (user, conn) where toWSConnect (u, c) = do (domain, uid) <- objQid u conn <- make c & asString - pure (WSConnect uid domain Nothing (Just conn)) + pure (WSConnect uid domain Nothing (Just conn) Nothing) instance (MakesValue user, MakesValue conn, MakesValue client) => ToWSConnect (user, conn, client) where toWSConnect (u, c, cl) = do (domain, uid) <- objQid u client <- make cl & asString conn <- make c & asString - pure (WSConnect uid domain (Just client) (Just conn)) + pure (WSConnect uid domain (Just client) (Just conn) Nothing) connect :: (HasCallStack) => WSConnect -> App WebSocket connect wsConnect = do @@ -178,6 +181,8 @@ run wsConnect app = do connId <- case wsConnect.conn of Just c -> pure c Nothing -> show <$> liftIO (randomIO :: IO Word32) + apiV <- maybe (getAPIVersionFor domain) pure wsConnect.version + let versionPrefix = "/v" <> show apiV let path = "/await" @@ -185,12 +190,13 @@ run wsConnect app = do Nothing -> "" Just client -> fromJust . fromByteString $ Http.queryString (Http.setQueryString [("client", Just (toByteString' client))] Http.defaultRequest) ) + wsPath = versionPrefix <> path caHdrs = [ ("Z-User", toByteString' (wsConnect.user)), ("Z-Connection", toByteString' connId) ] request <- do - r <- rawBaseRequest domain Cannon Versioned path + r <- rawBaseRequest domain Cannon (ExplicitVersion apiV) path pure r {HTTP.requestHeaders = caHdrs} wsapp <- @@ -200,7 +206,7 @@ run wsConnect app = do ( WS.runClientWith caHost (fromIntegral caPort) - path + wsPath WS.defaultConnectionOptions caHdrs app diff --git a/libs/wire-api/src/Wire/API/Event/Conversation.hs b/libs/wire-api/src/Wire/API/Event/Conversation.hs index f39a745fcf8..a9936b695c9 100644 --- a/libs/wire-api/src/Wire/API/Event/Conversation.hs +++ b/libs/wire-api/src/Wire/API/Event/Conversation.hs @@ -118,6 +118,7 @@ import Wire.API.Conversation.Typing import Wire.API.Event.LeaveReason import Wire.API.History import Wire.API.MLS.SubConversation +import Wire.API.Notification (Transmit (..)) import Wire.API.Routes.MultiVerb import Wire.API.Routes.Version import Wire.API.User (QualifiedUserIdList (..), qualifiedUserIdListObjectSchema) @@ -170,6 +171,11 @@ data Event = Event evtType :: Event -> EventType evtType = eventDataType . evtData +instance Transmit Event where + transmit e v + | v < V15, evtType e `elem` [ConvCreateMeeting, ConvDeleteMeeting] = Nothing + | otherwise = Just e + instance Arbitrary Event where arbitrary = do typ <- arbitrary diff --git a/libs/wire-api/src/Wire/API/Event/Meeting.hs b/libs/wire-api/src/Wire/API/Event/Meeting.hs index 90fd6b25c1e..b76e317b843 100644 --- a/libs/wire-api/src/Wire/API/Event/Meeting.hs +++ b/libs/wire-api/src/Wire/API/Event/Meeting.hs @@ -37,6 +37,8 @@ import Data.Schema import Data.Time (UTCTime) import Imports import Wire.API.Event.Conversation (EventFrom (..), eventFromUserId, eventVia, mkEventFrom) +import Wire.API.Notification (Transmit (..)) +import Wire.API.Routes.Version (Version (..)) import Wire.Arbitrary (Arbitrary (arbitrary), GenericUniform (..)) -------------------------------------------------------------------------------- @@ -57,6 +59,11 @@ instance ToSchema EventType where element "meeting.member-add" MemberAdd ] +instance Transmit Event where + transmit e v + | v < V15 = Nothing -- all meeting.* event types were introduced with V15 + | otherwise = Just e + -------------------------------------------------------------------------------- -- Event diff --git a/libs/wire-api/src/Wire/API/Event/Transmit.hs b/libs/wire-api/src/Wire/API/Event/Transmit.hs new file mode 100644 index 00000000000..77a4670d1fe --- /dev/null +++ b/libs/wire-api/src/Wire/API/Event/Transmit.hs @@ -0,0 +1,106 @@ +-- This file is part of the Wire Server implementation. +-- +-- Copyright (C) 2026 Wire Swiss GmbH +-- +-- This program is free software: you can redistribute it and/or modify it under +-- the terms of the GNU Affero General Public License as published by the Free +-- Software Foundation, either version 3 of the License, or (at your option) any +-- later version. +-- +-- This program is distributed in the hope that it will be useful, but WITHOUT +-- ANY WARRANTY; without even the implied warranty of MERCHANTABILITY or FITNESS +-- FOR A PARTICULAR PURPOSE. See the GNU Affero General Public License for more +-- details. +-- +-- You should have received a copy of the GNU Affero General Public License along +-- with this program. If not, see . + +-- | Dispatcher that applies the 'Transmit' instances of the event family +-- modules to the untyped event objects stored and relayed by gundeck and +-- cannon. Only the JSON \"type\" strings governed by a family instance are +-- ever decoded; everything else is passed through unchanged. +module Wire.API.Event.Transmit + ( transmitEvent, + transmitQueuedNotification, + transmitInternalNotification, + ) +where + +import Control.Lens ((.~), (^.)) +import Data.Aeson qualified as A +import Data.Aeson.KeyMap qualified as KeyMap +import Data.List.NonEmpty qualified as NonEmpty +import Data.Set qualified as Set +import Imports +import Wire.API.Event.Conversation qualified as Conversation +import Wire.API.Event.Meeting qualified as Meeting +import Wire.API.Internal.Notification qualified as InternalNotification +import Wire.API.Notification +import Wire.API.Routes.Version + +-- | Highest version for which any gate applies; requests at >= this version +-- are passed through byte-identical without decoding. Bump when adding a +-- gate above V15. +maxGateVersion :: Version +maxGateVersion = V15 + +-- | Exact JSON \"type\" strings governed by the family instances above. Only +-- these are ever decoded; every other event type is passed through unchanged. +conversationMeetingEventTypes, meetingEventTypes :: Set Text +conversationMeetingEventTypes = + Set.fromList + [ "conversation.create-meeting", + "conversation.delete-meeting" + ] +meetingEventTypes = + Set.fromList + [ "meeting.create", + "meeting.update", + "meeting.delete", + "meeting.member-add" + ] + +-- | Adjust a stored event object for delivery to a client that called the API +-- at the given 'Version'. 'Nothing' = do not deliver. +-- +-- Unknown event types and objects without a \"type\" key are always delivered +-- (forward compatibility). A gated type that cannot be decoded below its gate +-- is dropped (fail closed). +transmitEvent :: Version -> Event -> Maybe Event +transmitEvent v o + | v >= maxGateVersion = Just o + | otherwise = case KeyMap.lookup "type" o of + Just (A.String t) + | t `Set.member` meetingEventTypes -> reTransmit @Meeting.Event + | t `Set.member` conversationMeetingEventTypes -> reTransmit @Conversation.Event + _ -> Just o + where + reTransmit :: forall e. (Transmit e, Eq e, A.FromJSON e, A.ToJSON e) => Maybe Event + reTransmit = case A.fromJSON @e (A.Object o) of + A.Success e -> case transmit e v of + Nothing -> Nothing + Just e' + -- byte-preserve the stored object when the event is unchanged + | e == e' -> Just o + | otherwise -> case A.toJSON e' of + A.Object o' -> Just o' + _ -> Just o + -- a gated type we cannot decode must not leak below its gate: + A.Error _ -> Nothing + +-- | 'transmitEvent' over the payload of a 'QueuedNotification'; 'Nothing' +-- when the payload empties. +transmitQueuedNotification :: Version -> QueuedNotification -> Maybe QueuedNotification +transmitQueuedNotification v n = do + payload <- NonEmpty.nonEmpty (mapMaybe (transmitEvent v) (NonEmpty.toList (n ^. queuedNotificationPayload))) + pure (n & queuedNotificationPayload .~ payload) + +-- | 'transmitEvent' over the payload of an internal 'Notification'; 'Nothing' +-- when the payload empties. +transmitInternalNotification :: + Version -> + InternalNotification.Notification -> + Maybe InternalNotification.Notification +transmitInternalNotification v n = do + payload <- NonEmpty.nonEmpty (mapMaybe (transmitEvent v) (NonEmpty.toList (InternalNotification.ntfPayload n))) + pure n {InternalNotification.ntfPayload = payload} diff --git a/libs/wire-api/src/Wire/API/Notification.hs b/libs/wire-api/src/Wire/API/Notification.hs index 69b53bd9c9f..186002e795f 100644 --- a/libs/wire-api/src/Wire/API/Notification.hs +++ b/libs/wire-api/src/Wire/API/Notification.hs @@ -25,6 +25,7 @@ module Wire.API.Notification RawNotificationId (..), Event, ServerTime (..), + Transmit (..), -- * QueuedNotification QueuedNotification, @@ -75,6 +76,7 @@ import Network.HTTP.Types import Network.Wai.Utilities (mkError) import Servant import Wire.API.Routes.MultiVerb +import Wire.API.Routes.Version (Version) import Wire.Arbitrary (Arbitrary, GenericUniform (..)) type NotificationId = Id QueuedNotification @@ -94,6 +96,13 @@ mkNotificationId = do -- (e.g. visible in 'modelEvent'). Can we specify it in a better way? type Event = Aeson.Object +-- | Adjust an event for delivery to a client that called the API at the given +-- 'Version'. 'Nothing' = do not deliver. The default is pass-through; +-- instances are provided by event family modules. +class Transmit e where + transmit :: e -> Version -> Maybe e + transmit e _ = Just e + -- | Schema for an `Event` object. -- -- This is basically a schema for a JSON object with some pre-defined structure. diff --git a/libs/wire-api/src/Wire/API/Routes/Public/Cannon.hs b/libs/wire-api/src/Wire/API/Routes/Public/Cannon.hs index 6fefbcf80ae..91ff714c17f 100644 --- a/libs/wire-api/src/Wire/API/Routes/Public/Cannon.hs +++ b/libs/wire-api/src/Wire/API/Routes/Public/Cannon.hs @@ -25,6 +25,7 @@ import Wire.API.Routes.Named import Wire.API.Routes.Public (ZConn, ZUser) import Wire.API.Routes.Version import Wire.API.Routes.WebSocket +import Wire.API.VersionInfo (APIVersion) type CannonAPI = Named @@ -32,6 +33,7 @@ type CannonAPI = ( Summary "Establish websocket connection" -- Description "This is the legacy variant of \"consume-events\"" :> "await" + :> APIVersion Version :> ZUser :> ZConn :> QueryParam' @@ -49,6 +51,7 @@ type CannonAPI = ( Summary "Establish websocket connection" :> Description "This is a temporary copy of await, please do not use it" :> "websocket" + :> APIVersion Version :> ZUser :> ZConn :> QueryParam' @@ -68,6 +71,7 @@ type CannonAPI = :> From 'V8 :> Until 'V9 :> "events" + :> APIVersion Version :> ZUser :> QueryParam' [ Optional, @@ -85,6 +89,7 @@ type CannonAPI = :> Description "This is the rabbitMQ-based variant of \"await-notifications\"" :> From 'V9 :> "events" + :> APIVersion Version :> ZUser :> QueryParam' [ Optional, diff --git a/libs/wire-api/src/Wire/API/Routes/Public/Gundeck.hs b/libs/wire-api/src/Wire/API/Routes/Public/Gundeck.hs index f37500dd196..5a32b33d487 100644 --- a/libs/wire-api/src/Wire/API/Routes/Public/Gundeck.hs +++ b/libs/wire-api/src/Wire/API/Routes/Public/Gundeck.hs @@ -30,6 +30,7 @@ import Wire.API.Routes.MultiVerb import Wire.API.Routes.Named import Wire.API.Routes.Public import Wire.API.Routes.Version +import Wire.API.VersionInfo (APIVersion) type GundeckAPI = PushAPI :<|> NotificationAPI :<|> TimeAPI @@ -68,6 +69,7 @@ type NotificationAPI = Named "get-notification-by-id" ( Summary "Fetch a notification by ID" + :> APIVersion Version :> ZUser :> "notifications" :> Capture' '[Description "Notification ID"] "id" NotificationId @@ -83,6 +85,8 @@ type NotificationAPI = :<|> Named "get-last-notification" ( Summary "Fetch the last notification" + :> Description "If the most recent notification contains only events not deliverable at the client's API version, 404 is returned." + :> APIVersion Version :> ZUser :> "notifications" :> "last" @@ -99,6 +103,7 @@ type NotificationAPI = "get-notifications@v2" ( Summary "Fetch notifications" :> Until 'V3 + :> APIVersion Version :> ZUser :> "notifications" :> QueryParam' [Optional, Strict, Description "Only return notifications more recent than this"] "since" RawNotificationId @@ -117,6 +122,7 @@ type NotificationAPI = ( Summary "Fetch notifications" :> Description "See also: GET /teams/notifications" :> From 'V3 + :> APIVersion Version :> ZUser :> "notifications" :> QueryParam' [Optional, Strict, Description "Only return notifications more recent than this"] "since" NotificationId diff --git a/libs/wire-api/test/unit/Test/Wire/API/Event/Transmit.hs b/libs/wire-api/test/unit/Test/Wire/API/Event/Transmit.hs new file mode 100644 index 00000000000..a5fd7769e26 --- /dev/null +++ b/libs/wire-api/test/unit/Test/Wire/API/Event/Transmit.hs @@ -0,0 +1,159 @@ +-- This file is part of the Wire Server implementation. +-- +-- Copyright (C) 2026 Wire Swiss GmbH +-- +-- This program is free software: you can redistribute it and/or modify it under +-- the terms of the GNU Affero General Public License as published by the Free +-- Software Foundation, either version 3 of the License, or (at your option) any +-- later version. +-- +-- This program is distributed in the hope that it will be useful, but WITHOUT +-- ANY WARRANTY; without even the implied warranty of MERCHANTABILITY or FITNESS +-- FOR A PARTICULAR PURPOSE. See the GNU Affero General Public License for more +-- details. +-- +-- You should have received a copy of the GNU Affero General Public License along +-- with this program. If not, see . + +module Test.Wire.API.Event.Transmit (tests) where + +import Data.Aeson qualified as Aeson +import Data.Aeson.KeyMap qualified as KeyMap +import Data.Id +import Data.List.NonEmpty (NonEmpty ((:|))) +import Data.Set qualified as Set +import Data.Text qualified as T +import Data.UUID qualified as UUID +import Imports +import Test.Tasty +import Test.Tasty.HUnit +import Wire.API.Event.Conversation qualified as Conv +import Wire.API.Event.Meeting qualified as Mtg +import Wire.API.Event.Transmit +import Wire.API.Internal.Notification qualified as Internal +import Wire.API.Notification +import Wire.API.Routes.Version + +tests :: TestTree +tests = + testGroup + "Transmit" + [ gatedTypesTests, + passthroughTests, + malformedTests, + notificationTests, + driftGuardTests + ] + +typedEvent :: Text -> Event +typedEvent t = KeyMap.singleton "type" (Aeson.String t) + +gatedTypes :: [Text] +gatedTypes = + [ "conversation.create-meeting", + "conversation.delete-meeting", + "meeting.create", + "meeting.update", + "meeting.delete", + "meeting.member-add" + ] + +passthroughTypes :: [Text] +passthroughTypes = ["conversation.create", "user.update", "team.member-join"] + +gatedTypesTests :: TestTree +gatedTypesTests = + testGroup + "gated types are dropped below V15 and delivered at V15" + ( [ testCase ("dropped at V14: " <> T.unpack t) $ + transmitEvent V14 (typedEvent t) @?= Nothing + | t <- gatedTypes + ] + ++ [ testCase ("delivered at V15: " <> T.unpack t) $ + transmitEvent V15 (typedEvent t) @?= Just (typedEvent t) + | t <- gatedTypes + ] + ) + +passthroughTests :: TestTree +passthroughTests = + testGroup + "non-gated types and objects without a type key are always delivered" + ( [ testCase ("delivered at V14: " <> T.unpack t) $ + transmitEvent V14 (typedEvent t) @?= Just (typedEvent t) + | t <- passthroughTypes + ] + ++ [ testCase "no type key delivered at V14" $ + transmitEvent V14 (KeyMap.fromList []) @?= Just (KeyMap.fromList []), + testCase "no type key delivered at V15" $ + transmitEvent V15 (KeyMap.fromList []) @?= Just (KeyMap.fromList []), + testCase "non-string type delivered at V14" $ + transmitEvent V14 (KeyMap.singleton "type" (Aeson.Number 42)) @?= Just (KeyMap.singleton "type" (Aeson.Number 42)) + ] + ) + +malformedTests :: TestTree +malformedTests = + testGroup + "gated type that fails to decode is dropped below its gate, passed at/above it" + [ testCase "malformed meeting.create at V14" $ + transmitEvent V14 (typedEvent "meeting.create") @?= Nothing, + testCase "malformed meeting.create at V15" $ + transmitEvent V15 (typedEvent "meeting.create") @?= Just (typedEvent "meeting.create") + ] + +dummyId :: Id a +dummyId = Id (fromJust (UUID.fromString "7c2dc4e0-1bd0-11e4-8c21-0800200c9a66")) + +notificationTests :: TestTree +notificationTests = + testGroup + "transmitQueuedNotification / transmitInternalNotification" + [ testCase "all-gated payload becomes Nothing" $ do + let qn = queuedNotification dummyId (typedEvent "meeting.create" :| [typedEvent "conversation.create-meeting"]) + transmitQueuedNotification V14 qn @?= Nothing, + testCase "mixed payload keeps ungated events" $ do + let keep = typedEvent "conversation.create" + qn = queuedNotification dummyId (typedEvent "meeting.delete" :| [keep]) + transmitQueuedNotification V14 qn @?= Just (queuedNotification dummyId (keep :| [])), + testCase "internal all-gated payload becomes Nothing" $ do + let n = Internal.Notification dummyId False (typedEvent "meeting.update" :| []) + transmitInternalNotification V14 n @?= Nothing, + testCase "internal mixed payload keeps ungated events" $ do + let keep = typedEvent "user.update" + n = Internal.Notification dummyId False (typedEvent "meeting.update" :| [keep]) + transmitInternalNotification V14 n @?= Just (Internal.Notification dummyId False (keep :| [])) + ] + +driftGuardTests :: TestTree +driftGuardTests = + testGroup + "dispatcher type strings match the family event type encodings" + [ testCase "conversation meeting event types encode to the gated strings" $ do + Aeson.toJSON Conv.ConvCreateMeeting @?= Aeson.String "conversation.create-meeting" + Aeson.toJSON Conv.ConvDeleteMeeting @?= Aeson.String "conversation.delete-meeting", + testCase "meeting event types encode to the gated strings" $ + Set.fromList (Aeson.toJSON @Mtg.EventType <$> [minBound .. maxBound]) + @?= Set.fromList (Aeson.String <$> ["meeting.create", "meeting.update", "meeting.delete", "meeting.member-add"]), + testCase "no other conversation event type is gated" $ + Set.fromList + [ Aeson.toJSON t + | t <- [minBound .. maxBound] :: [Conv.EventType], + t `notElem` [Conv.ConvCreateMeeting, Conv.ConvDeleteMeeting] + ] + `disjointFrom` gatedStrings, + testCase "meeting event types are exactly the gated meeting strings" $ + Set.fromList (Aeson.toJSON @Mtg.EventType <$> [minBound .. maxBound]) + `disjointFrom` conversationMeetingStrings + ] + where + gatedStrings :: Set Aeson.Value + gatedStrings = + Set.map Aeson.String (Set.fromList gatedTypes) + + conversationMeetingStrings :: Set Aeson.Value + conversationMeetingStrings = + Set.map Aeson.String (Set.fromList ["conversation.create-meeting", "conversation.delete-meeting"]) + + disjointFrom :: Set Aeson.Value -> Set Aeson.Value -> Assertion + disjointFrom a b = True @?= Set.null (Set.intersection a b) diff --git a/libs/wire-api/test/unit/Test/Wire/API/Run.hs b/libs/wire-api/test/unit/Test/Wire/API/Run.hs index 3e30589b08c..bb6628cfd42 100644 --- a/libs/wire-api/test/unit/Test/Wire/API/Run.hs +++ b/libs/wire-api/test/unit/Test/Wire/API/Run.hs @@ -22,6 +22,7 @@ import System.IO.Unsafe (unsafePerformIO) import Test.Tasty import Test.Wire.API.Call.Config qualified as Call.Config import Test.Wire.API.Conversation qualified as Conversation +import Test.Wire.API.Event.Transmit qualified as Event.Transmit import Test.Wire.API.MLS qualified as MLS import Test.Wire.API.MLS.Group qualified as Group import Test.Wire.API.Meeting qualified as Meeting @@ -52,6 +53,7 @@ main = "Tests" [ Call.Config.tests, Team.Member.tests, + Event.Transmit.tests, Team.Export.tests, User.tests, User.Search.tests, diff --git a/libs/wire-api/wire-api.cabal b/libs/wire-api/wire-api.cabal index d6d83c2fd4b..6980685d2c0 100644 --- a/libs/wire-api/wire-api.cabal +++ b/libs/wire-api/wire-api.cabal @@ -107,6 +107,7 @@ library Wire.API.Event.LeaveReason Wire.API.Event.Meeting Wire.API.Event.Team + Wire.API.Event.Transmit Wire.API.Event.WebSocketProtocol Wire.API.FederationStatus Wire.API.FederationUpdate @@ -715,6 +716,7 @@ test-suite wire-api-tests Paths_wire_api Test.Wire.API.Call.Config Test.Wire.API.Conversation + Test.Wire.API.Event.Transmit Test.Wire.API.Meeting Test.Wire.API.MLS Test.Wire.API.MLS.Group diff --git a/services/cannon/src/Cannon/API/Internal.hs b/services/cannon/src/Cannon/API/Internal.hs index ce44b90a549..f735532055b 100644 --- a/services/cannon/src/Cannon/API/Internal.hs +++ b/services/cannon/src/Cannon/API/Internal.hs @@ -25,14 +25,15 @@ import Cannon.Dict qualified as D import Cannon.Types import Cannon.WS import Control.Monad.Catch -import Data.Aeson (encode) +import Data.Aeson (eitherDecode', encode) import Data.Id +import Data.Text qualified as T import Imports -import Network.WebSockets import Servant import Servant.Conduit () import System.Logger.Class (msg, val) import System.Logger.Class qualified as LC +import Wire.API.Event.Transmit (transmitInternalNotification) import Wire.API.Internal.BulkPush import Wire.API.Internal.Notification import Wire.API.RawJson @@ -48,13 +49,25 @@ internalServer = pushHandler :: UserId -> ConnId -> RawJson -> Cannon (Maybe ()) pushHandler user conn body = - singlePush (rawJsonBytes body) (PushTarget user conn) >>= \case + singlePush (eitherDecode' (rawJsonBytes body)) (PushTarget user conn) >>= \case PushStatusOk -> pure $ Just () PushStatusGone -> pure Nothing --- | Take notification @n@ and send it to the 'PushTarget'. -singlePush :: (WebSocketsData a) => a -> PushTarget -> Cannon PushStatus -singlePush n (PushTarget usrid conid) = do +-- | Take notification @n@ and send it to the 'PushTarget', encoding it for +-- the target connection's API version. A notification that is fully +-- filtered out for that version is skipped but still reported as +-- 'PushStatusOk' so gundeck neither falls back to native push nor treats the +-- client as gone. +singlePush :: Either String Notification -> PushTarget -> Cannon PushStatus +singlePush (Left err) (PushTarget usrid conid) = do + -- fail closed: a body we cannot decode cannot be version-gated, but the + -- drop must be observable (schema skew between gundeck and cannon would + -- otherwise silently discard all legacy /await pushes). + LC.err $ + client (key2bytes (mkKey usrid conid)) + . msg ("push: failed to decode notification: " <> T.pack err) + pure PushStatusOk +singlePush (Right n) (PushTarget usrid conid) = do let k = mkKey usrid conid d <- clients LC.debug $ client (key2bytes k) . msg (val "push") @@ -63,20 +76,22 @@ singlePush n (PushTarget usrid conid) = do Nothing -> do LC.debug $ client (key2bytes k) . msg (val "push: client gone") pure PushStatusGone - Just x -> do - e <- wsenv - runWS e $ do - catchAll - (runWS e (sendMsg n k x) >> pure PushStatusOk) - (const (terminate k x >> pure PushStatusGone)) + Just x -> case transmitInternalNotification (wsApiVersion x) n of + Nothing -> pure PushStatusOk + Just n' -> do + e <- wsenv + runWS e $ do + catchAll + (runWS e (sendMsg (encode n') k x) >> pure PushStatusOk) + (const (terminate k x >> pure PushStatusGone)) bulkPushHandler :: BulkPushRequest -> Cannon BulkPushResponse bulkPushHandler (BulkPushRequest ns) = BulkPushResponse . mconcat . zipWith compileResp ns <$> (uncurry doNotify `Imports.mapM` ns) where doNotify :: Notification -> [PushTarget] -> Cannon [PushStatus] - doNotify (encode -> notification) = - mapConcurrentlyCannon (singlePush notification) + doNotify notif = + mapConcurrentlyCannon (singlePush (Right notif)) compileResp :: (Notification, [PushTarget]) -> [PushStatus] -> diff --git a/services/cannon/src/Cannon/API/Public.hs b/services/cannon/src/Cannon/API/Public.hs index e80eb7aa4f5..8ef1eb53391 100644 --- a/services/cannon/src/Cannon/API/Public.hs +++ b/services/cannon/src/Cannon/API/Public.hs @@ -32,20 +32,21 @@ import Network.WebSockets.Connection import Servant import Wire.API.Routes.Named import Wire.API.Routes.Public.Cannon +import Wire.API.Routes.Version (Version) publicAPIServer :: ServerT CannonAPI Cannon publicAPIServer = Named @"await-notifications" streamData :<|> Named @"websocket" streamData - :<|> Named @"consume-events@v8" (\userId mClientId -> consumeEvents userId mClientId Nothing) + :<|> Named @"consume-events@v8" (\v userId mClientId -> consumeEvents v userId mClientId Nothing) :<|> Named @"consume-events" consumeEvents -streamData :: UserId -> ConnId -> Maybe ClientId -> PendingConnection -> Cannon () -streamData userId connId clientId con = do +streamData :: Version -> UserId -> ConnId -> Maybe ClientId -> PendingConnection -> Cannon () +streamData v userId connId clientId con = do e <- wsenv - liftIO $ wsapp (mkKey userId connId) clientId e con + liftIO $ wsapp v (mkKey userId connId) clientId e con -consumeEvents :: UserId -> Maybe ClientId -> Maybe Text -> PendingConnection -> Cannon () -consumeEvents userId mClientId mSyncMarker con = do +consumeEvents :: Version -> UserId -> Maybe ClientId -> Maybe Text -> PendingConnection -> Cannon () +consumeEvents v userId mClientId mSyncMarker con = do e <- wsenv - liftIO $ rabbitMQWebSocketApp userId mClientId mSyncMarker e con + liftIO $ rabbitMQWebSocketApp v userId mClientId mSyncMarker e con diff --git a/services/cannon/src/Cannon/App.hs b/services/cannon/src/Cannon/App.hs index 538e9b14acd..b0b237c68e9 100644 --- a/services/cannon/src/Cannon/App.hs +++ b/services/cannon/src/Cannon/App.hs @@ -33,19 +33,20 @@ import System.Logger qualified as Log import System.Logger.Class hiding (Error, close) import System.Logger.Class qualified as Logger import UnliftIO (throwIO, timeout) +import Wire.API.Routes.Version (Version) -- | Maximum lifetime of a websocket in seconds. maxLifetime :: Int maxLifetime = 3 * 24 * 3600 -wsapp :: Key -> Maybe ClientId -> Env -> ServerApp -wsapp k c e pc = runWS e (go `catches` ioErrors k c) +wsapp :: Version -> Key -> Maybe ClientId -> Env -> ServerApp +wsapp v k c e pc = runWS e (go `catches` ioErrors k c) where go = do runInIO <- askRunInIO conn0 <- liftIO (acceptRequest pc `catch` rejectOnError pc) liftIO . withPingPong defaultPingPongOptions conn0 $ \conn -> runInIO $ do - ws <- mkWebSocket conn + ws <- mkWebSocket v conn debug $ client (key2bytes k) ~~ "websocket" .= connIdent ws registerLocal k ws registerRemote k c diff --git a/services/cannon/src/Cannon/RabbitMqConsumerApp.hs b/services/cannon/src/Cannon/RabbitMqConsumerApp.hs index 13f7f1950cb..d57e51d8a26 100644 --- a/services/cannon/src/Cannon/RabbitMqConsumerApp.hs +++ b/services/cannon/src/Cannon/RabbitMqConsumerApp.hs @@ -42,16 +42,18 @@ import Network.WebSockets qualified as WS import Network.WebSockets.Connection import System.Logger qualified as Log import System.Timeout +import Wire.API.Event.Transmit (transmitQueuedNotification) import Wire.API.Event.WebSocketProtocol import Wire.API.Notification +import Wire.API.Routes.Version qualified as V data InactivityTimeout = InactivityTimeout deriving (Show) instance Exception InactivityTimeout -rabbitMQWebSocketApp :: UserId -> Maybe ClientId -> Maybe Text -> Env -> ServerApp -rabbitMQWebSocketApp uid mcid mSyncMarkerId e pendingConn = +rabbitMQWebSocketApp :: V.Version -> UserId -> Maybe ClientId -> Maybe Text -> Env -> ServerApp +rabbitMQWebSocketApp apiVersion uid mcid mSyncMarkerId e pendingConn = handle handleTooManyChannels . lowerCodensity $ do (chan, queueInfo) <- createChannel uid mcid e.pool createQueue @@ -132,14 +134,21 @@ rabbitMQWebSocketApp uid mcid mSyncMarkerId e pendingConn = Q.rejectEnv envelope False -- try again getEventData chan - Right notif -> do - logEvent notif - pure $ - Left $ - EventData - { event = notif, - deliveryTag = envelope.envDeliveryTag - } + Right notif -> case transmitQueuedNotification apiVersion notif of + Nothing -> do + -- The event is gated for this client's API version: deliver + -- nothing and ack server-side so the message is not + -- redelivered. + ackMessage chan envelope.envDeliveryTag False + getEventData chan + Just notif' -> do + logEvent notif' + pure $ + Left $ + EventData + { event = notif', + deliveryTag = envelope.envDeliveryTag + } handleWebSocketExceptions wsConn = Handler $ diff --git a/services/cannon/src/Cannon/WS.hs b/services/cannon/src/Cannon/WS.hs index 21d215104bb..c88a24372ff 100644 --- a/services/cannon/src/Cannon/WS.hs +++ b/services/cannon/src/Cannon/WS.hs @@ -39,6 +39,7 @@ module Cannon.WS Websocket, connection, connIdent, + wsApiVersion, Key, mkKey, key2bytes, @@ -83,6 +84,7 @@ import System.Logger.Class hiding (Error, Settings, close, (.=)) import System.Random.MWC (GenIO, uniform) import UnliftIO.Async (async, cancel, pooledMapConcurrentlyN_) import Wire.API.Presence +import Wire.API.Routes.Version (Version) ----------------------------------------------------------------------------- -- Key @@ -112,13 +114,14 @@ keyConnBytes = snd . _key data Websocket = Websocket { connection :: Connection, - connIdent :: !Word + connIdent :: !Word, + wsApiVersion :: !Version } -mkWebSocket :: Connection -> WS Websocket -mkWebSocket c = do +mkWebSocket :: Version -> Connection -> WS Websocket +mkWebSocket v c = do g <- WS $ asks rand - Websocket c <$> liftIO (uniform g) + Websocket c <$> liftIO (uniform g) <*> pure v ----------------------------------------------------------------------------- -- Clock diff --git a/services/gundeck/src/Gundeck/API/Public.hs b/services/gundeck/src/Gundeck/API/Public.hs index 90337190bb4..bd002243b8d 100644 --- a/services/gundeck/src/Gundeck/API/Public.hs +++ b/services/gundeck/src/Gundeck/API/Public.hs @@ -30,9 +30,11 @@ import Gundeck.Notification.Data qualified as Data import Gundeck.Push qualified as Push import Imports import Servant (HasServer (..), (:<|>) (..)) +import Wire.API.Event.Transmit (transmitQueuedNotification) import Wire.API.Notification qualified as Public import Wire.API.Routes.Named (Named (Named)) import Wire.API.Routes.Public.Gundeck +import Wire.API.Routes.Version (Version) ------------------------------------------------------------------------------- -- Servant API @@ -46,11 +48,15 @@ servantSitemap = pushAPI :<|> notificationAPI :<|> timeAPI :<|> Named @"get-push-tokens" Push.listTokens notificationAPI = - Named @"get-notification-by-id" Data.fetchId - :<|> Named @"get-last-notification" Data.fetchLast + Named @"get-notification-by-id" fetchIdH + :<|> Named @"get-last-notification" fetchLastH :<|> Named @"get-notifications@v2" paginateUntilV2 :<|> Named @"get-notifications" paginate + fetchIdH v u n c = (>>= transmitQueuedNotification v) <$> Data.fetchId u n c + + fetchLastH v u c = (>>= transmitQueuedNotification v) <$> Data.fetchLast u c + timeAPI = Named @"get-server-time" getServerTime @@ -87,14 +93,15 @@ servantSitemap = pushAPI :<|> notificationAPI :<|> timeAPI -- (arianvp): I am not sure why it is convenient for clients to distinguish -- between these two cases. paginateUntilV2 :: + Version -> UserId -> Maybe Public.RawNotificationId -> Maybe ClientId -> Maybe (Range 100 10000 Int32) -> Gundeck Public.GetNotificationsResponse -paginateUntilV2 uid mbSince mbClient mbSize = do +paginateUntilV2 v uid mbSince mbClient mbSize = do let size = fromMaybe (unsafeRange 1000) mbSize - Notification.PaginateResult gap page <- Notification.paginate uid (join since) mbClient size + Notification.PaginateResult gap page <- Notification.paginate v uid (join since) mbClient size pure $ if gap then Public.GetNotificationsWithStatusNotFound page @@ -113,14 +120,15 @@ paginateUntilV2 uid mbSince mbClient mbSize = do isV1UUID u = if UUID.version u == 1 then Just u else Nothing paginate :: + Version -> UserId -> Maybe Public.NotificationId -> Maybe ClientId -> Maybe (Range 100 10000 Int32) -> Gundeck (Maybe Public.QueuedNotificationList) -paginate uid mbSince mbClient mbSize = do +paginate v uid mbSince mbClient mbSize = do let size = fromMaybe (unsafeRange 1000) mbSize - Notification.PaginateResult gap page <- Notification.paginate uid mbSince mbClient size + Notification.PaginateResult gap page <- Notification.paginate v uid mbSince mbClient size pure $ if gap then Nothing else Just page getServerTime :: UserId -> Gundeck Public.ServerTime diff --git a/services/gundeck/src/Gundeck/Notification.hs b/services/gundeck/src/Gundeck/Notification.hs index 9c982696e19..06eb01712f5 100644 --- a/services/gundeck/src/Gundeck/Notification.hs +++ b/services/gundeck/src/Gundeck/Notification.hs @@ -40,29 +40,67 @@ import Network.Wai.Utilities.Error import System.Logger.Class import System.Logger.Class qualified as Log import Util.Options (Endpoint (Endpoint)) +import Wire.API.Event.Transmit (transmitQueuedNotification) import Wire.API.Internal.Notification import Wire.API.Notification +import Wire.API.Routes.Version (Version) data PaginateResult = PaginateResult { paginateResultGap :: Bool, paginateResultPage :: QueuedNotificationList } -paginate :: UserId -> Maybe NotificationId -> Maybe ClientId -> Range 100 10000 Int32 -> Gundeck PaginateResult -paginate uid since mclt size = do +paginate :: Version -> UserId -> Maybe NotificationId -> Maybe ClientId -> Range 100 10000 Int32 -> Gundeck PaginateResult +paginate v uid since mclt size = do traverse_ validateNotificationId since for_ mclt $ \clt -> updateActivity uid clt time <- posixTime rs <- Data.fetch uid mclt since size - pure $ PaginateResult (Data.resultGap rs) (resultList time rs) + -- 'gap' semantics come from the first fetch only; pages fetched during the + -- refill loop start exactly at the previous page's last cursor. + (page, hasMore) <- refill rs (1 :: Int) + pure $ PaginateResult (Data.resultGap rs) (resultList time hasMore page) where - resultList time rs = + resultList time more ns = queuedNotificationList - (toList (Data.resultSeq rs)) - (Data.resultHasMore rs) + (toList ns) + more (Just (msToUTCSecs time)) + -- Keep fetching while the client-visible survivors of the last page are + -- empty but the store says there is more, so an all-gated page cannot + -- leave the client's 'since' cursor stuck on an empty 'has_more=true' + -- page. Bounded to keep requests finite. + refill rs pages = do + let survivors = filtered rs + more = Data.resultHasMore rs + if not (null survivors) || not more + then pure (survivors, more) + else + if pages >= refillPageLimit + then do + -- one survivor is enough to advance the client cursor, so refill + -- pages are fetched at a small size to bound read amplification + Log.warn $ + Log.msg (val "notification refill limit reached (all-gated backlog?)") + ~~ "user" + .= UUID.toASCIIBytes (toUUID uid) + pure (survivors, more) + else case listToMaybe (reverse (toList (Data.resultSeq rs))) of + -- Nothing must stay terminal: 'hasMore' set on an empty page would + -- otherwise loop forever. + Nothing -> pure (survivors, more) + Just lastRaw -> do + rs' <- Data.fetch uid mclt (Just (view queuedNotificationId lastRaw)) refillSize + refill rs' (pages + 1) + + filtered rs = mapMaybe (transmitQueuedNotification v) (toList (Data.resultSeq rs)) + + refillSize = unsafeRange 100 :: Range 100 10000 Int32 + + refillPageLimit = 32 :: Int + validateNotificationId :: NotificationId -> Gundeck () validateNotificationId n = unless (isValidNotificationId n) $ From 0f4a1252e5edb9ba94115fe6d82cc2e1a132ca8b Mon Sep 17 00:00:00 2001 From: Gautier DI FOLCO Date: Thu, 20 Aug 2026 20:40:12 +0200 Subject: [PATCH 2/5] Hello CI From e6dc2844d586025642ede5c93531cd7cc1469c8a Mon Sep 17 00:00:00 2001 From: Gautier DI FOLCO Date: Thu, 20 Aug 2026 21:57:00 +0200 Subject: [PATCH 3/5] Hello CI From 32c5b7c7fd1a7949c4bc239caff2705b572b5223 Mon Sep 17 00:00:00 2001 From: Gautier DI FOLCO Date: Thu, 20 Aug 2026 23:13:50 +0200 Subject: [PATCH 4/5] Hello CI From ad1ad9ba3fd9bafc67ab4a0d109911107d4ba67a Mon Sep 17 00:00:00 2001 From: Gautier DI FOLCO Date: Fri, 21 Aug 2026 00:07:46 +0200 Subject: [PATCH 5/5] Hello CI