Skip to content

Commit 8b94b16

Browse files
committed
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.
1 parent f2a9c1d commit 8b94b16

21 files changed

Lines changed: 693 additions & 66 deletions

File tree

Lines changed: 9 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,9 @@
1+
Event delivery is now version-aware: gundeck REST endpoints (`GET
2+
/notifications`, by-id, last) and cannon websockets filter stored events by
3+
the API version of the requesting client. Meeting-related events
4+
(`conversation.create-meeting`, `conversation.delete-meeting`,
5+
`meeting.create`, `meeting.update`, `meeting.delete`, `meeting.member-add`)
6+
are not delivered to clients that called the API below V15, the version
7+
meetings were introduced in. Storage, fan-out and native push are unchanged;
8+
modern clients receive byte-identical payloads. Unknown event types are always
9+
delivered for forward compatibility.

integration/integration.cabal

Lines changed: 1 addition & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -196,6 +196,7 @@ library
196196
Test.MLS.Unreachable
197197
Test.NginxZAuthModule
198198
Test.Notifications
199+
Test.NotificationsVersioned
199200
Test.OAuth
200201
Test.One2OneTeamConv
201202
Test.PasswordReset

integration/test/Test/Events.hs

Lines changed: 25 additions & 10 deletions
Original file line numberDiff line numberDiff line change
@@ -595,7 +595,7 @@ testChannelLimit = withModifiedBackend
595595

596596
-- the first client fails to connect because the server runs out of channels
597597
do
598-
eithWS <- createEventsWebSocketEither alice (Just client0) Nothing
598+
eithWS <- createEventsWebSocketEither alice (Just client0) Nothing Nothing
599599
case eithWS of
600600
Left (WS.MalformedResponse respHead _) ->
601601
lift $ respHead.responseCode `shouldMatchInt` 503
@@ -958,16 +958,21 @@ createEventWebSockets ::
958958
Codensity App [EventWebSocket]
959959
createEventWebSockets = traverse (uncurry createEventsWebSocket)
960960

961+
requireConnectedWebSocket ::
962+
(HasCallStack) =>
963+
Either WS.HandshakeException EventWebSocket ->
964+
Codensity App EventWebSocket
965+
requireConnectedWebSocket = \case
966+
Left e -> lift $ assertFailure $ "Websocket failed to connect due to handshake exception: " <> displayException e
967+
Right ws -> pure ws
968+
961969
createEventsWebSocket ::
962970
(HasCallStack, MakesValue user) =>
963971
user ->
964972
Maybe String ->
965973
Codensity App EventWebSocket
966-
createEventsWebSocket user cid = do
967-
eithWS <- createEventsWebSocketEither user cid Nothing
968-
case eithWS of
969-
Left e -> lift $ assertFailure $ "Websocket failed to connect due to handshake exception: " <> displayException e
970-
Right ws -> pure ws
974+
createEventsWebSocket user cid =
975+
createEventsWebSocketEither user cid Nothing Nothing >>= requireConnectedWebSocket
971976

972977
createEventsWebSocketWithSync ::
973978
(HasCallStack, MakesValue user) =>
@@ -976,22 +981,32 @@ createEventsWebSocketWithSync ::
976981
Codensity App (String, EventWebSocket)
977982
createEventsWebSocketWithSync user cid = do
978983
syncMarker <- lift randomId
979-
eithWS <- createEventsWebSocketEither user cid (Just syncMarker)
980-
case eithWS of
984+
createEventsWebSocketEither user cid (Just syncMarker) Nothing >>= \case
981985
Left e -> lift $ assertFailure $ "Websocket failed to connect due to handshake exception: " <> displayException e
982986
Right ws -> pure (syncMarker, ws)
983987

988+
-- | 'createEventsWebSocket', but connecting at an explicit API version.
989+
createEventsWebSocketAtVersion ::
990+
(HasCallStack, MakesValue user) =>
991+
user ->
992+
Maybe String ->
993+
Int ->
994+
Codensity App EventWebSocket
995+
createEventsWebSocketAtVersion user cid v =
996+
createEventsWebSocketEither user cid Nothing (Just v) >>= requireConnectedWebSocket
997+
984998
createEventsWebSocketEither ::
985999
(HasCallStack, MakesValue user) =>
9861000
user ->
9871001
Maybe String ->
9881002
Maybe String ->
1003+
Maybe Int ->
9891004
Codensity App (Either WS.HandshakeException EventWebSocket)
990-
createEventsWebSocketEither user cid mSyncMarker = do
1005+
createEventsWebSocketEither user cid mSyncMarker mApiVersion = do
9911006
eventsChan <- liftIO newChan
9921007
ackChan <- liftIO newEmptyMVar
9931008
serviceMap <- lift $ getServiceMap =<< objDomain user
994-
apiVersion <- lift $ getAPIVersionFor $ objDomain user
1009+
apiVersion <- maybe (lift $ getAPIVersionFor $ objDomain user) pure mApiVersion
9951010
wsStarted <- newEmptyMVar
9961011
let minAPIVersion = 8
9971012
lift
Lines changed: 218 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,218 @@
1+
-- This file is part of the Wire Server implementation.
2+
--
3+
-- Copyright (C) 2026 Wire Swiss GmbH <opensource@wire.com>
4+
--
5+
-- This program is free software: you can redistribute it and/or modify it under
6+
-- the terms of the GNU Affero General Public License as published by the Free
7+
-- Software Foundation, either version 3 of the License, or (at your option) any
8+
-- later version.
9+
--
10+
-- This program is distributed in the hope that it will be useful, but WITHOUT
11+
-- ANY WARRANTY; without even the implied warranty of MERCHANTABILITY or FITNESS
12+
-- FOR A PARTICULAR PURPOSE. See the GNU Affero General Public License for more
13+
-- details.
14+
--
15+
-- You should have received a copy of the GNU Affero General Public License along
16+
-- with this program. If not, see <https://www.gnu.org/licenses/>.
17+
18+
module Test.NotificationsVersioned where
19+
20+
import API.Brig (addClient, putHandle)
21+
import API.BrigCommon (AddClient (..))
22+
import API.Common (randomHandle)
23+
import API.Galley
24+
25+
import Control.Monad.Codensity (runCodensity)
26+
import Data.Time.Clock
27+
import Notifications (isConvCreateMeetingNotif, isMeetingCreateNotif)
28+
import SetupHelpers
29+
import Test.Events (ackEvent, assertFindsEvent, consumeAllEventsNoAck, createEventsWebSocketAtVersion, enableConsumableNotifications)
30+
import Test.Meetings (defaultMeetingJson)
31+
import Testlib.Cannon
32+
import Testlib.Prelude
33+
import UnliftIO.Concurrent (threadDelay)
34+
35+
gatedTypes :: [String]
36+
gatedTypes =
37+
[ "conversation.create-meeting",
38+
"conversation.delete-meeting",
39+
"meeting.create",
40+
"meeting.update",
41+
"meeting.delete",
42+
"meeting.member-add"
43+
]
44+
45+
isGated :: String -> Bool
46+
isGated t = t `elem` gatedTypes
47+
48+
-- | Drain all notification pages at the given API version and return the
49+
-- payload event types seen. Also asserts that pagination terminates (no
50+
-- endless empty 'has_more=true' pages).
51+
drainNotificationsAt :: (HasCallStack, MakesValue user) => user -> Int -> App [String]
52+
drainNotificationsAt user v = go Nothing []
53+
where
54+
go since acc = do
55+
req <- baseRequest user Gundeck (ExplicitVersion v) "/notifications"
56+
let req' =
57+
req
58+
& addQueryParams
59+
( [("since", s) | s <- toList since]
60+
<> [("size", "100")]
61+
)
62+
r <- submit "GET" req'
63+
r.status `shouldMatchInt` 200
64+
body <- getJSON 200 r
65+
notifications <- body %. "notifications" & asList
66+
types <-
67+
mconcat <$> for
68+
notifications
69+
( \n -> do
70+
payload <- n %. "payload" & asList
71+
for payload (\e -> e %. "type" >>= asString)
72+
)
73+
lastId <- case reverse notifications of
74+
[] -> pure Nothing
75+
(n : _) -> Just <$> (n %. "id" >>= asString)
76+
hasMore <- body %. "has_more" & asBool
77+
if hasMore
78+
then case lastId of
79+
Just l -> go (Just l) (acc <> types)
80+
Nothing -> assertFailure "has_more=true but no notification id for cursor"
81+
else pure (acc <> types)
82+
83+
mkMeeting :: App Value
84+
mkMeeting = do
85+
now <- liftIO getCurrentTime
86+
let startTime = addUTCTime 3600 now
87+
endTime = addUTCTime 7200 now
88+
pure $ defaultMeetingJson "Versioned meeting" startTime endTime []
89+
90+
-- | The meeting creator (who receives the meeting events) must not see them
91+
-- via a V14 fetch, while a current-version fetch of the same window shows
92+
-- them; V14 pagination terminates.
93+
testVersionedNotificationsHideMeetingEvents :: (HasCallStack) => App ()
94+
testVersionedNotificationsHideMeetingEvents = do
95+
(alice, _tid, _members) <- createTeam OwnDomain 1
96+
meeting <- mkMeeting
97+
98+
withWebSocket alice $ \wsAlice -> do
99+
resp <- postMeetings alice meeting
100+
assertSuccess resp
101+
-- the current-version websocket sees the meeting events
102+
void $ awaitMatch isConvCreateMeetingNotif wsAlice
103+
void $ awaitMatch isMeetingCreateNotif wsAlice
104+
105+
v14Types <- drainNotificationsAt alice 14
106+
filter isGated v14Types `shouldMatch` ([] :: [String])
107+
108+
curTypes <- drainNotificationsAt alice 17
109+
curTypes `shouldContain` ["conversation.create-meeting"]
110+
curTypes `shouldContain` ["meeting.create"]
111+
112+
-- | A V14 client still sees non-meeting events (e.g. conversation.create)
113+
-- while meeting events are filtered from the same window.
114+
testVersionedNotificationsKeepNonMeetingEvents :: (HasCallStack) => App ()
115+
testVersionedNotificationsKeepNonMeetingEvents = do
116+
(alice, tid, [bob]) <- createTeam OwnDomain 2
117+
resp <-
118+
postConversation
119+
alice
120+
defProteus
121+
{ qualifiedUsers = [bob],
122+
name = Just "plain conv",
123+
team = Just tid
124+
}
125+
assertSuccess resp
126+
meeting <- mkMeeting
127+
mresp <- postMeetings alice meeting
128+
assertSuccess mresp
129+
130+
-- bob sees the plain conversation event, but no meeting events (the
131+
-- meeting's gated events go to alice, and none leak to bob at V14).
132+
bobTypes <- drainNotificationsAt bob 14
133+
bobTypes `shouldContain` ["conversation.create"]
134+
filter isGated bobTypes `shouldMatch` ([] :: [String])
135+
136+
aliceTypes <- drainNotificationsAt alice 14
137+
aliceTypes `shouldContain` ["conversation.create"]
138+
filter isGated aliceTypes `shouldMatch` ([] :: [String])
139+
140+
-- | A websocket connected at a low version receives no meeting event frames,
141+
-- while a current-version connection of the same user does.
142+
testVersionedWebSocketFiltersMeetingEvents :: (HasCallStack) => App ()
143+
testVersionedWebSocketFiltersMeetingEvents = do
144+
(alice, _tid, _members) <- createTeam OwnDomain 1
145+
aliceId <- alice %. "id" >>= asString
146+
aliceDomain <- objDomain alice
147+
meeting <- mkMeeting
148+
149+
let lowV = WSConnect aliceId aliceDomain Nothing (Just "lowconn") (Just 14)
150+
highV = WSConnect aliceId aliceDomain Nothing (Just "highconn") Nothing
151+
152+
withWebSocket lowV $ \wsLow ->
153+
withWebSocket highV $ \wsHigh -> do
154+
resp <- postMeetings alice meeting
155+
assertSuccess resp
156+
-- current version gets the meeting events ...
157+
void $ awaitMatch isConvCreateMeetingNotif wsHigh
158+
void $ awaitMatch isMeetingCreateNotif wsHigh
159+
-- ... the low version does not (allow some time for delivery)
160+
liftIO $ threadDelay 1_000_000
161+
assertNoEvent 1 wsLow
162+
163+
-- | The rabbitmq-backed /events websocket of a low-version client skips (and
164+
-- server-side acks) meeting event frames, while a current-version connection
165+
-- of the same user receives them. Tolerant drain: stray ungated events are
166+
-- allowed on the low socket, gated ones are not.
167+
testVersionedEventsSocketFiltersMeetingEvents :: (HasCallStack) => App ()
168+
testVersionedEventsSocketFiltersMeetingEvents =
169+
withModifiedBackend (enableConsumableNotifications def) $ \domain -> do
170+
(alice, _tid, _members) <- createTeam domain 1
171+
-- mirror the other temp-/events tests in Test.Events: create a
172+
-- consumable-notifications client for alice
173+
void $ addClient alice def {acapabilities = Just ["consumable-notifications"]} >>= getJSON 201
174+
-- Two temp queues with no client id (each binds userRoutingKey and gets its
175+
-- own version-filtered consumer); sharing a client id would round-robin a
176+
-- single queue and race.
177+
runCodensity (createEventsWebSocketAtVersion alice Nothing 14) $ \wsLow ->
178+
runCodensity (createEventsWebSocketAtVersion alice Nothing 17) $ \wsHigh -> do
179+
meeting <- mkMeeting
180+
postMeetings alice meeting >>= assertSuccess
181+
assertFindsEvent wsHigh $ \e -> do
182+
e %. "type" `shouldMatch` "event"
183+
t <- e %. "data.event.payload.0.type" >>= asString
184+
unless (isGated t) $
185+
assertFailure ("expected a gated meeting event on the V17 socket, got: " <> t)
186+
ackEvent wsHigh e
187+
-- allow some time for delivery, then tolerate stray ungated events on
188+
-- the low socket but assert that none of them is gated
189+
liftIO $ threadDelay 1_000_000
190+
drained <- consumeAllEventsNoAck wsLow
191+
types <- traverse (\e -> e %. "data.event.payload.0.type" >>= asString) drained
192+
filter isGated types `shouldMatch` ([] :: [String])
193+
194+
-- | A V14 notification cursor is not stranded on an all-gated page: the
195+
-- gundeck refill loop must skip past a fully-gated page (server minimum page
196+
-- size is 100) and still deliver a later ungated event.
197+
testVersionedNotificationsRefillPastGatedBacklog :: (HasCallStack) => App ()
198+
testVersionedNotificationsRefillPastGatedBacklog = do
199+
(alice, _tid, _members) <- createTeam OwnDomain 1
200+
-- 101 meetings (~30-60 s by design) guarantee > 100 gated notification rows
201+
-- for the creator even if galley batches conversation.create-meeting and
202+
-- meeting.create into a single row.
203+
replicateM_ 101 $ do
204+
meeting <- mkMeeting
205+
postMeetings alice meeting >>= assertSuccess
206+
-- The ungated event must land strictly after the gated backlog (gundeck
207+
-- persists notifications synchronously in the request path). If that ever
208+
-- becomes async, user.update could land inside the first 100 rows and this
209+
-- test would silently degrade to never exercising the refill loop.
210+
handle <- randomHandle
211+
putHandle alice handle >>= assertSuccess
212+
213+
v14 <- drainNotificationsAt alice 14
214+
v14 `shouldContain` ["user.update"]
215+
filter isGated v14 `shouldMatch` ([] :: [String])
216+
217+
v17 <- drainNotificationsAt alice 17
218+
v17 `shouldContain` ["meeting.create"]

integration/test/Testlib/Cannon.hs

Lines changed: 12 additions & 6 deletions
Original file line numberDiff line numberDiff line change
@@ -110,7 +110,10 @@ data WSConnect = WSConnect
110110
domain :: String,
111111
client :: Maybe String,
112112
-- | If this is Nothing then a random Z-Connection will be used
113-
conn :: Maybe String
113+
conn :: Maybe String,
114+
-- | Explicit API version prefix for the websocket endpoint (affects
115+
-- versioned event delivery). 'Nothing' = current maximum version.
116+
version :: Maybe Int
114117
}
115118

116119
class ToWSConnect a where
@@ -124,20 +127,20 @@ instance {-# OVERLAPPABLE #-} (MakesValue user) => ToWSConnect user where
124127
(domain, uid) <- objQid u
125128
mc <- lookupField u "client_id"
126129
c <- traverse asString mc
127-
pure (WSConnect uid domain c Nothing)
130+
pure (WSConnect uid domain c Nothing Nothing)
128131

129132
instance (MakesValue user, MakesValue conn) => ToWSConnect (user, conn) where
130133
toWSConnect (u, c) = do
131134
(domain, uid) <- objQid u
132135
conn <- make c & asString
133-
pure (WSConnect uid domain Nothing (Just conn))
136+
pure (WSConnect uid domain Nothing (Just conn) Nothing)
134137

135138
instance (MakesValue user, MakesValue conn, MakesValue client) => ToWSConnect (user, conn, client) where
136139
toWSConnect (u, c, cl) = do
137140
(domain, uid) <- objQid u
138141
client <- make cl & asString
139142
conn <- make c & asString
140-
pure (WSConnect uid domain (Just client) (Just conn))
143+
pure (WSConnect uid domain (Just client) (Just conn) Nothing)
141144

142145
connect :: (HasCallStack) => WSConnect -> App WebSocket
143146
connect wsConnect = do
@@ -178,19 +181,22 @@ run wsConnect app = do
178181
connId <- case wsConnect.conn of
179182
Just c -> pure c
180183
Nothing -> show <$> liftIO (randomIO :: IO Word32)
184+
apiV <- maybe (getAPIVersionFor domain) pure wsConnect.version
185+
let versionPrefix = "/v" <> show apiV
181186

182187
let path =
183188
"/await"
184189
<> ( case wsConnect.client of
185190
Nothing -> ""
186191
Just client -> fromJust . fromByteString $ Http.queryString (Http.setQueryString [("client", Just (toByteString' client))] Http.defaultRequest)
187192
)
193+
wsPath = versionPrefix <> path
188194
caHdrs =
189195
[ ("Z-User", toByteString' (wsConnect.user)),
190196
("Z-Connection", toByteString' connId)
191197
]
192198
request <- do
193-
r <- rawBaseRequest domain Cannon Versioned path
199+
r <- rawBaseRequest domain Cannon (ExplicitVersion apiV) path
194200
pure r {HTTP.requestHeaders = caHdrs}
195201

196202
wsapp <-
@@ -200,7 +206,7 @@ run wsConnect app = do
200206
( WS.runClientWith
201207
caHost
202208
(fromIntegral caPort)
203-
path
209+
wsPath
204210
WS.defaultConnectionOptions
205211
caHdrs
206212
app

libs/wire-api/src/Wire/API/Event/Conversation.hs

Lines changed: 6 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -118,6 +118,7 @@ import Wire.API.Conversation.Typing
118118
import Wire.API.Event.LeaveReason
119119
import Wire.API.History
120120
import Wire.API.MLS.SubConversation
121+
import Wire.API.Notification (Transmit (..))
121122
import Wire.API.Routes.MultiVerb
122123
import Wire.API.Routes.Version
123124
import Wire.API.User (QualifiedUserIdList (..), qualifiedUserIdListObjectSchema)
@@ -170,6 +171,11 @@ data Event = Event
170171
evtType :: Event -> EventType
171172
evtType = eventDataType . evtData
172173

174+
instance Transmit Event where
175+
transmit e v
176+
| v < V15, evtType e `elem` [ConvCreateMeeting, ConvDeleteMeeting] = Nothing
177+
| otherwise = Just e
178+
173179
instance Arbitrary Event where
174180
arbitrary = do
175181
typ <- arbitrary

0 commit comments

Comments
 (0)