17
17
from typing import (
18
18
TYPE_CHECKING ,
19
19
Any ,
20
+ Awaitable ,
21
+ Callable ,
20
22
Dict ,
21
23
Iterable ,
22
24
List ,
23
25
Mapping ,
24
26
Optional ,
25
27
Sequence ,
26
28
Tuple ,
29
+ TypeVar ,
27
30
)
28
31
29
32
from prometheus_client import Counter
30
- from typing_extensions import TypeGuard
33
+ from typing_extensions import Concatenate , ParamSpec , TypeGuard
31
34
32
35
from synapse .api .constants import EventTypes , Membership , ThirdPartyEntityKind
33
36
from synapse .api .errors import CodeMessageException , HttpResponseException
78
81
HOUR_IN_MS = 60 * 60 * 1000
79
82
80
83
81
- APP_SERVICE_PREFIX = "/_matrix/app/unstable"
84
+ APP_SERVICE_PREFIX = "/_matrix/app/v1"
85
+ APP_SERVICE_UNSTABLE_PREFIX = "/_matrix/app/unstable"
86
+
87
+ P = ParamSpec ("P" )
88
+ R = TypeVar ("R" )
82
89
83
90
84
91
def _is_valid_3pe_metadata (info : JsonDict ) -> bool :
@@ -121,17 +128,60 @@ def __init__(self, hs: "HomeServer"):
121
128
hs .get_clock (), "as_protocol_meta" , timeout_ms = HOUR_IN_MS
122
129
)
123
130
131
+ async def _send_with_fallbacks (
132
+ self ,
133
+ service : "ApplicationService" ,
134
+ prefixes : List [str ],
135
+ path : str ,
136
+ func : Callable [Concatenate [str , P ], Awaitable [R ]],
137
+ * args : P .args ,
138
+ ** kwargs : P .kwargs ,
139
+ ) -> R :
140
+ """
141
+ Attempt to call an application service with multiple paths, falling back
142
+ until one succeeds.
143
+
144
+ Args:
145
+ service: The appliacation service, this provides the base URL.
146
+ prefixes: A last of paths to try in order for the requests.
147
+ path: A suffix to append to each prefix.
148
+ func: The function to call, the first argument will be the full
149
+ endpoint to fetch. Other arguments are provided by args/kwargs.
150
+
151
+ Returns:
152
+ The return value of func.
153
+ """
154
+ for i , prefix in enumerate (prefixes , start = 1 ):
155
+ uri = f"{ service .url } { prefix } { path } "
156
+ try :
157
+ return await func (uri , * args , ** kwargs )
158
+ except HttpResponseException as e :
159
+ # If an error is received that is due to an unrecognised path,
160
+ # fallback to next path (if one exists). Otherwise, consider it
161
+ # a legitimate error and raise.
162
+ if i < len (prefixes ) and is_unknown_endpoint (e ):
163
+ continue
164
+ raise
165
+ except Exception :
166
+ # Unexpected exceptions get sent to the caller.
167
+ raise
168
+
169
+ # The function should always exit via the return or raise above this.
170
+ raise RuntimeError ("Unexpected fallback behaviour. This should never be seen." )
171
+
124
172
async def query_user (self , service : "ApplicationService" , user_id : str ) -> bool :
125
173
if service .url is None :
126
174
return False
127
175
128
176
# This is required by the configuration.
129
177
assert service .hs_token is not None
130
178
131
- uri = service .url + ("/users/%s" % urllib .parse .quote (user_id ))
132
179
try :
133
- response = await self .get_json (
134
- uri ,
180
+ response = await self ._send_with_fallbacks (
181
+ service ,
182
+ [APP_SERVICE_PREFIX , "" ],
183
+ f"/users/{ urllib .parse .quote (user_id )} " ,
184
+ self .get_json ,
135
185
{"access_token" : service .hs_token },
136
186
headers = {"Authorization" : [f"Bearer { service .hs_token } " ]},
137
187
)
@@ -140,9 +190,9 @@ async def query_user(self, service: "ApplicationService", user_id: str) -> bool:
140
190
except CodeMessageException as e :
141
191
if e .code == 404 :
142
192
return False
143
- logger .warning ("query_user to %s received %s" , uri , e .code )
193
+ logger .warning ("query_user to %s received %s" , service . url , e .code )
144
194
except Exception as ex :
145
- logger .warning ("query_user to %s threw exception %s" , uri , ex )
195
+ logger .warning ("query_user to %s threw exception %s" , service . url , ex )
146
196
return False
147
197
148
198
async def query_alias (self , service : "ApplicationService" , alias : str ) -> bool :
@@ -152,21 +202,23 @@ async def query_alias(self, service: "ApplicationService", alias: str) -> bool:
152
202
# This is required by the configuration.
153
203
assert service .hs_token is not None
154
204
155
- uri = service .url + ("/rooms/%s" % urllib .parse .quote (alias ))
156
205
try :
157
- response = await self .get_json (
158
- uri ,
206
+ response = await self ._send_with_fallbacks (
207
+ service ,
208
+ [APP_SERVICE_PREFIX , "" ],
209
+ f"/rooms/{ urllib .parse .quote (alias )} " ,
210
+ self .get_json ,
159
211
{"access_token" : service .hs_token },
160
212
headers = {"Authorization" : [f"Bearer { service .hs_token } " ]},
161
213
)
162
214
if response is not None : # just an empty json object
163
215
return True
164
216
except CodeMessageException as e :
165
- logger .warning ("query_alias to %s received %s" , uri , e .code )
217
+ logger .warning ("query_alias to %s received %s" , service . url , e .code )
166
218
if e .code == 404 :
167
219
return False
168
220
except Exception as ex :
169
- logger .warning ("query_alias to %s threw exception %s" , uri , ex )
221
+ logger .warning ("query_alias to %s threw exception %s" , service . url , ex )
170
222
return False
171
223
172
224
async def query_3pe (
@@ -188,25 +240,24 @@ async def query_3pe(
188
240
# This is required by the configuration.
189
241
assert service .hs_token is not None
190
242
191
- uri = "%s%s/thirdparty/%s/%s" % (
192
- service .url ,
193
- APP_SERVICE_PREFIX ,
194
- kind ,
195
- urllib .parse .quote (protocol ),
196
- )
197
243
try :
198
244
args : Mapping [Any , Any ] = {
199
245
** fields ,
200
246
b"access_token" : service .hs_token ,
201
247
}
202
- response = await self .get_json (
203
- uri ,
248
+ response = await self ._send_with_fallbacks (
249
+ service ,
250
+ [APP_SERVICE_PREFIX , APP_SERVICE_UNSTABLE_PREFIX ],
251
+ f"/thirdparty/{ kind } /{ urllib .parse .quote (protocol )} " ,
252
+ self .get_json ,
204
253
args = args ,
205
254
headers = {"Authorization" : [f"Bearer { service .hs_token } " ]},
206
255
)
207
256
if not isinstance (response , list ):
208
257
logger .warning (
209
- "query_3pe to %s returned an invalid response %r" , uri , response
258
+ "query_3pe to %s returned an invalid response %r" ,
259
+ service .url ,
260
+ response ,
210
261
)
211
262
return []
212
263
@@ -216,12 +267,12 @@ async def query_3pe(
216
267
ret .append (r )
217
268
else :
218
269
logger .warning (
219
- "query_3pe to %s returned an invalid result %r" , uri , r
270
+ "query_3pe to %s returned an invalid result %r" , service . url , r
220
271
)
221
272
222
273
return ret
223
274
except Exception as ex :
224
- logger .warning ("query_3pe to %s threw exception %s" , uri , ex )
275
+ logger .warning ("query_3pe to %s threw exception %s" , service . url , ex )
225
276
return []
226
277
227
278
async def get_3pe_protocol (
@@ -233,21 +284,20 @@ async def get_3pe_protocol(
233
284
async def _get () -> Optional [JsonDict ]:
234
285
# This is required by the configuration.
235
286
assert service .hs_token is not None
236
- uri = "%s%s/thirdparty/protocol/%s" % (
237
- service .url ,
238
- APP_SERVICE_PREFIX ,
239
- urllib .parse .quote (protocol ),
240
- )
241
287
try :
242
- info = await self .get_json (
243
- uri ,
288
+ info = await self ._send_with_fallbacks (
289
+ service ,
290
+ [APP_SERVICE_PREFIX , APP_SERVICE_UNSTABLE_PREFIX ],
291
+ f"/thirdparty/protocol/{ urllib .parse .quote (protocol )} " ,
292
+ self .get_json ,
244
293
{"access_token" : service .hs_token },
245
294
headers = {"Authorization" : [f"Bearer { service .hs_token } " ]},
246
295
)
247
296
248
297
if not _is_valid_3pe_metadata (info ):
249
298
logger .warning (
250
- "query_3pe_protocol to %s did not return a valid result" , uri
299
+ "query_3pe_protocol to %s did not return a valid result" ,
300
+ service .url ,
251
301
)
252
302
return None
253
303
@@ -260,7 +310,9 @@ async def _get() -> Optional[JsonDict]:
260
310
261
311
return info
262
312
except Exception as ex :
263
- logger .warning ("query_3pe_protocol to %s threw exception %s" , uri , ex )
313
+ logger .warning (
314
+ "query_3pe_protocol to %s threw exception %s" , service .url , ex
315
+ )
264
316
return None
265
317
266
318
key = (service .id , protocol )
@@ -274,7 +326,7 @@ async def ping(self, service: "ApplicationService", txn_id: Optional[str]) -> No
274
326
assert service .hs_token is not None
275
327
276
328
await self .post_json_get_json (
277
- uri = service .url + "/_matrix/app/unstable /fi.mau.msc2659/ping" ,
329
+ uri = f" { service .url } { APP_SERVICE_UNSTABLE_PREFIX } /fi.mau.msc2659/ping" ,
278
330
post_json = {"transaction_id" : txn_id },
279
331
headers = {"Authorization" : [f"Bearer { service .hs_token } " ]},
280
332
)
@@ -318,8 +370,6 @@ async def push_bulk(
318
370
)
319
371
txn_id = 0
320
372
321
- uri = service .url + ("/transactions/%s" % urllib .parse .quote (str (txn_id )))
322
-
323
373
# Never send ephemeral events to appservices that do not support it
324
374
body : JsonDict = {"events" : serialized_events }
325
375
if service .supports_ephemeral :
@@ -351,16 +401,19 @@ async def push_bulk(
351
401
}
352
402
353
403
try :
354
- await self .put_json (
355
- uri = uri ,
404
+ await self ._send_with_fallbacks (
405
+ service ,
406
+ [APP_SERVICE_PREFIX , "" ],
407
+ f"/transactions/{ urllib .parse .quote (str (txn_id ))} " ,
408
+ self .put_json ,
356
409
json_body = body ,
357
410
args = {"access_token" : service .hs_token },
358
411
headers = {"Authorization" : [f"Bearer { service .hs_token } " ]},
359
412
)
360
413
if logger .isEnabledFor (logging .DEBUG ):
361
414
logger .debug (
362
415
"push_bulk to %s succeeded! events=%s" ,
363
- uri ,
416
+ service . url ,
364
417
[event .get ("event_id" ) for event in events ],
365
418
)
366
419
sent_transactions_counter .labels (service .id ).inc ()
@@ -371,15 +424,15 @@ async def push_bulk(
371
424
except CodeMessageException as e :
372
425
logger .warning (
373
426
"push_bulk to %s received code=%s msg=%s" ,
374
- uri ,
427
+ service . url ,
375
428
e .code ,
376
429
e .msg ,
377
430
exc_info = logger .isEnabledFor (logging .DEBUG ),
378
431
)
379
432
except Exception as ex :
380
433
logger .warning (
381
434
"push_bulk to %s threw exception(%s) %s args=%s" ,
382
- uri ,
435
+ service . url ,
383
436
type (ex ).__name__ ,
384
437
ex ,
385
438
ex .args ,
0 commit comments