-
-
Notifications
You must be signed in to change notification settings - Fork 589
/
sync.js
1704 lines (1536 loc) · 62.6 KB
/
sync.js
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
343
344
345
346
347
348
349
350
351
352
353
354
355
356
357
358
359
360
361
362
363
364
365
366
367
368
369
370
371
372
373
374
375
376
377
378
379
380
381
382
383
384
385
386
387
388
389
390
391
392
393
394
395
396
397
398
399
400
401
402
403
404
405
406
407
408
409
410
411
412
413
414
415
416
417
418
419
420
421
422
423
424
425
426
427
428
429
430
431
432
433
434
435
436
437
438
439
440
441
442
443
444
445
446
447
448
449
450
451
452
453
454
455
456
457
458
459
460
461
462
463
464
465
466
467
468
469
470
471
472
473
474
475
476
477
478
479
480
481
482
483
484
485
486
487
488
489
490
491
492
493
494
495
496
497
498
499
500
501
502
503
504
505
506
507
508
509
510
511
512
513
514
515
516
517
518
519
520
521
522
523
524
525
526
527
528
529
530
531
532
533
534
535
536
537
538
539
540
541
542
543
544
545
546
547
548
549
550
551
552
553
554
555
556
557
558
559
560
561
562
563
564
565
566
567
568
569
570
571
572
573
574
575
576
577
578
579
580
581
582
583
584
585
586
587
588
589
590
591
592
593
594
595
596
597
598
599
600
601
602
603
604
605
606
607
608
609
610
611
612
613
614
615
616
617
618
619
620
621
622
623
624
625
626
627
628
629
630
631
632
633
634
635
636
637
638
639
640
641
642
643
644
645
646
647
648
649
650
651
652
653
654
655
656
657
658
659
660
661
662
663
664
665
666
667
668
669
670
671
672
673
674
675
676
677
678
679
680
681
682
683
684
685
686
687
688
689
690
691
692
693
694
695
696
697
698
699
700
701
702
703
704
705
706
707
708
709
710
711
712
713
714
715
716
717
718
719
720
721
722
723
724
725
726
727
728
729
730
731
732
733
734
735
736
737
738
739
740
741
742
743
744
745
746
747
748
749
750
751
752
753
754
755
756
757
758
759
760
761
762
763
764
765
766
767
768
769
770
771
772
773
774
775
776
777
778
779
780
781
782
783
784
785
786
787
788
789
790
791
792
793
794
795
796
797
798
799
800
801
802
803
804
805
806
807
808
809
810
811
812
813
814
815
816
817
818
819
820
821
822
823
824
825
826
827
828
829
830
831
832
833
834
835
836
837
838
839
840
841
842
843
844
845
846
847
848
849
850
851
852
853
854
855
856
857
858
859
860
861
862
863
864
865
866
867
868
869
870
871
872
873
874
875
876
877
878
879
880
881
882
883
884
885
886
887
888
889
890
891
892
893
894
895
896
897
898
899
900
901
902
903
904
905
906
907
908
909
910
911
912
913
914
915
916
917
918
919
920
921
922
923
924
925
926
927
928
929
930
931
932
933
934
935
936
937
938
939
940
941
942
943
944
945
946
947
948
949
950
951
952
953
954
955
956
957
958
959
960
961
962
963
964
965
966
967
968
969
970
971
972
973
974
975
976
977
978
979
980
981
982
983
984
985
986
987
988
989
990
991
992
993
994
995
996
997
998
999
1000
/*
Copyright 2015, 2016 OpenMarket Ltd
Copyright 2017 Vector Creations Ltd
Copyright 2018 New Vector Ltd
Copyright 2019 The Matrix.org Foundation C.I.C.
Licensed under the Apache License, Version 2.0 (the "License");
you may not use this file except in compliance with the License.
You may obtain a copy of the License at
http://www.apache.org/licenses/LICENSE-2.0
Unless required by applicable law or agreed to in writing, software
distributed under the License is distributed on an "AS IS" BASIS,
WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
See the License for the specific language governing permissions and
limitations under the License.
*/
/*
* TODO:
* This class mainly serves to take all the syncing logic out of client.js and
* into a separate file. It's all very fluid, and this class gut wrenches a lot
* of MatrixClient props (e.g. _http). Given we want to support WebSockets as
* an alternative syncing API, we may want to have a proper syncing interface
* for HTTP and WS at some point.
*/
import {User} from "./models/user";
import {Room} from "./models/room";
import {Group} from "./models/group";
import * as utils from "./utils";
import {Filter} from "./filter";
import {EventTimeline} from "./models/event-timeline";
import {PushProcessor} from "./pushprocessor";
import {logger} from './logger';
import {InvalidStoreError} from './errors';
const DEBUG = true;
// /sync requests allow you to set a timeout= but the request may continue
// beyond that and wedge forever, so we need to track how long we are willing
// to keep open the connection. This constant is *ADDED* to the timeout= value
// to determine the max time we're willing to wait.
const BUFFER_PERIOD_MS = 80 * 1000;
// Number of consecutive failed syncs that will lead to a syncState of ERROR as opposed
// to RECONNECTING. This is needed to inform the client of server issues when the
// keepAlive is successful but the server /sync fails.
const FAILED_SYNC_ERROR_THRESHOLD = 3;
function getFilterName(userId, suffix) {
// scope this on the user ID because people may login on many accounts
// and they all need to be stored!
return "FILTER_SYNC_" + userId + (suffix ? "_" + suffix : "");
}
function debuglog(...params) {
if (!DEBUG) {
return;
}
logger.log(...params);
}
/**
* <b>Internal class - unstable.</b>
* Construct an entity which is able to sync with a homeserver.
* @constructor
* @param {MatrixClient} client The matrix client instance to use.
* @param {Object} opts Config options
* @param {module:crypto=} opts.crypto Crypto manager
* @param {Function=} opts.canResetEntireTimeline A function which is called
* with a room ID and returns a boolean. It should return 'true' if the SDK can
* SAFELY remove events from this room. It may not be safe to remove events if
* there are other references to the timelines for this room.
* Default: returns false.
* @param {Boolean=} opts.disablePresence True to perform syncing without automatically
* updating presence.
*/
export function SyncApi(client, opts) {
this.client = client;
opts = opts || {};
opts.initialSyncLimit = (
opts.initialSyncLimit === undefined ? 8 : opts.initialSyncLimit
);
opts.resolveInvitesToProfiles = opts.resolveInvitesToProfiles || false;
opts.pollTimeout = opts.pollTimeout || (30 * 1000);
opts.pendingEventOrdering = opts.pendingEventOrdering || "chronological";
if (!opts.canResetEntireTimeline) {
opts.canResetEntireTimeline = function(roomId) {
return false;
};
}
this.opts = opts;
this._peekRoomId = null;
this._currentSyncRequest = null;
this._syncState = null;
this._syncStateData = null; // additional data (eg. error object for failed sync)
this._catchingUp = false;
this._running = false;
this._keepAliveTimer = null;
this._connectionReturnedDefer = null;
this._notifEvents = []; // accumulator of sync events in the current sync response
this._failedSyncCount = 0; // Number of consecutive failed /sync requests
this._storeIsInvalid = false; // flag set if the store needs to be cleared before we can start
if (client.getNotifTimelineSet()) {
client.reEmitter.reEmit(client.getNotifTimelineSet(),
["Room.timeline", "Room.timelineReset"]);
}
}
/**
* @param {string} roomId
* @return {Room}
*/
SyncApi.prototype.createRoom = function(roomId) {
const client = this.client;
const {
timelineSupport,
unstableClientRelationAggregation,
} = client;
const room = new Room(roomId, client, client.getUserId(), {
lazyLoadMembers: this.opts.lazyLoadMembers,
pendingEventOrdering: this.opts.pendingEventOrdering,
timelineSupport,
unstableClientRelationAggregation,
});
client.reEmitter.reEmit(room, ["Room.name", "Room.timeline",
"Room.redaction",
"Room.redactionCancelled",
"Room.receipt", "Room.tags",
"Room.timelineReset",
"Room.localEchoUpdated",
"Room.accountData",
"Room.myMembership",
"Room.replaceEvent",
]);
this._registerStateListeners(room);
return room;
};
/**
* @param {string} groupId
* @return {Group}
*/
SyncApi.prototype.createGroup = function(groupId) {
const client = this.client;
const group = new Group(groupId);
client.reEmitter.reEmit(group, ["Group.profile", "Group.myMembership"]);
client.store.storeGroup(group);
return group;
};
/**
* @param {Room} room
* @private
*/
SyncApi.prototype._registerStateListeners = function(room) {
const client = this.client;
// we need to also re-emit room state and room member events, so hook it up
// to the client now. We need to add a listener for RoomState.members in
// order to hook them correctly. (TODO: find a better way?)
client.reEmitter.reEmit(room.currentState, [
"RoomState.events", "RoomState.members", "RoomState.newMember",
]);
room.currentState.on("RoomState.newMember", function(event, state, member) {
member.user = client.getUser(member.userId);
client.reEmitter.reEmit(
member,
[
"RoomMember.name", "RoomMember.typing", "RoomMember.powerLevel",
"RoomMember.membership",
],
);
});
};
/**
* @param {Room} room
* @private
*/
SyncApi.prototype._deregisterStateListeners = function(room) {
// could do with a better way of achieving this.
room.currentState.removeAllListeners("RoomState.events");
room.currentState.removeAllListeners("RoomState.members");
room.currentState.removeAllListeners("RoomState.newMember");
};
/**
* Sync rooms the user has left.
* @return {Promise} Resolved when they've been added to the store.
*/
SyncApi.prototype.syncLeftRooms = function() {
const client = this.client;
const self = this;
// grab a filter with limit=1 and include_leave=true
const filter = new Filter(this.client.credentials.userId);
filter.setTimelineLimit(1);
filter.setIncludeLeaveRooms(true);
const localTimeoutMs = this.opts.pollTimeout + BUFFER_PERIOD_MS;
const qps = {
timeout: 0, // don't want to block since this is a single isolated req
};
return client.getOrCreateFilter(
getFilterName(client.credentials.userId, "LEFT_ROOMS"), filter,
).then(function(filterId) {
qps.filter = filterId;
return client._http.authedRequest(
undefined, "GET", "/sync", qps, undefined, localTimeoutMs,
);
}).then(function(data) {
let leaveRooms = [];
if (data.rooms && data.rooms.leave) {
leaveRooms = self._mapSyncResponseToRoomArray(data.rooms.leave);
}
const rooms = [];
leaveRooms.forEach(function(leaveObj) {
const room = leaveObj.room;
rooms.push(room);
if (!leaveObj.isBrandNewRoom) {
// the intention behind syncLeftRooms is to add in rooms which were
// *omitted* from the initial /sync. Rooms the user were joined to
// but then left whilst the app is running will appear in this list
// and we do not want to bother with them since they will have the
// current state already (and may get dupe messages if we add
// yet more timeline events!), so skip them.
// NB: When we persist rooms to localStorage this will be more
// complicated...
return;
}
leaveObj.timeline = leaveObj.timeline || {};
const timelineEvents =
self._mapSyncEventsFormat(leaveObj.timeline, room);
const stateEvents = self._mapSyncEventsFormat(leaveObj.state, room);
// set the back-pagination token. Do this *before* adding any
// events so that clients can start back-paginating.
room.getLiveTimeline().setPaginationToken(leaveObj.timeline.prev_batch,
EventTimeline.BACKWARDS);
self._processRoomEvents(room, stateEvents, timelineEvents);
room.recalculate();
client.store.storeRoom(room);
client.emit("Room", room);
self._processEventsForNotifs(room, timelineEvents);
});
return rooms;
});
};
/**
* Peek into a room. This will result in the room in question being synced so it
* is accessible via getRooms(). Live updates for the room will be provided.
* @param {string} roomId The room ID to peek into.
* @return {Promise} A promise which resolves once the room has been added to the
* store.
*/
SyncApi.prototype.peek = function(roomId) {
const self = this;
const client = this.client;
this._peekRoomId = roomId;
return this.client.roomInitialSync(roomId, 20).then(function(response) {
// make sure things are init'd
response.messages = response.messages || {};
response.messages.chunk = response.messages.chunk || [];
response.state = response.state || [];
const peekRoom = self.createRoom(roomId);
// FIXME: Mostly duplicated from _processRoomEvents but not entirely
// because "state" in this API is at the BEGINNING of the chunk
const oldStateEvents = utils.map(
utils.deepCopy(response.state), client.getEventMapper(),
);
const stateEvents = utils.map(
response.state, client.getEventMapper(),
);
const messages = utils.map(
response.messages.chunk, client.getEventMapper(),
);
// XXX: copypasted from /sync until we kill off this
// minging v1 API stuff)
// handle presence events (User objects)
if (response.presence && utils.isArray(response.presence)) {
response.presence.map(client.getEventMapper()).forEach(
function(presenceEvent) {
let user = client.store.getUser(presenceEvent.getContent().user_id);
if (user) {
user.setPresenceEvent(presenceEvent);
} else {
user = createNewUser(client, presenceEvent.getContent().user_id);
user.setPresenceEvent(presenceEvent);
client.store.storeUser(user);
}
client.emit("event", presenceEvent);
});
}
// set the pagination token before adding the events in case people
// fire off pagination requests in response to the Room.timeline
// events.
if (response.messages.start) {
peekRoom.oldState.paginationToken = response.messages.start;
}
// set the state of the room to as it was after the timeline executes
peekRoom.oldState.setStateEvents(oldStateEvents);
peekRoom.currentState.setStateEvents(stateEvents);
self._resolveInvites(peekRoom);
peekRoom.recalculate();
// roll backwards to diverge old state. addEventsToTimeline
// will overwrite the pagination token, so make sure it overwrites
// it with the right thing.
peekRoom.addEventsToTimeline(messages.reverse(), true,
peekRoom.getLiveTimeline(),
response.messages.start);
client.store.storeRoom(peekRoom);
client.emit("Room", peekRoom);
self._peekPoll(peekRoom);
return peekRoom;
});
};
/**
* Stop polling for updates in the peeked room. NOPs if there is no room being
* peeked.
*/
SyncApi.prototype.stopPeeking = function() {
this._peekRoomId = null;
};
/**
* Do a peek room poll.
* @param {Room} peekRoom
* @param {string} token from= token
*/
SyncApi.prototype._peekPoll = function(peekRoom, token) {
if (this._peekRoomId !== peekRoom.roomId) {
debuglog("Stopped peeking in room %s", peekRoom.roomId);
return;
}
const self = this;
// FIXME: gut wrenching; hard-coded timeout values
this.client._http.authedRequest(undefined, "GET", "/events", {
room_id: peekRoom.roomId,
timeout: 30 * 1000,
from: token,
}, undefined, 50 * 1000).then(function(res) {
if (self._peekRoomId !== peekRoom.roomId) {
debuglog("Stopped peeking in room %s", peekRoom.roomId);
return;
}
// We have a problem that we get presence both from /events and /sync
// however, /sync only returns presence for users in rooms
// you're actually joined to.
// in order to be sure to get presence for all of the users in the
// peeked room, we handle presence explicitly here. This may result
// in duplicate presence events firing for some users, which is a
// performance drain, but such is life.
// XXX: copypasted from /sync until we can kill this minging v1 stuff.
res.chunk.filter(function(e) {
return e.type === "m.presence";
}).map(self.client.getEventMapper()).forEach(function(presenceEvent) {
let user = self.client.store.getUser(presenceEvent.getContent().user_id);
if (user) {
user.setPresenceEvent(presenceEvent);
} else {
user = createNewUser(self.client, presenceEvent.getContent().user_id);
user.setPresenceEvent(presenceEvent);
self.client.store.storeUser(user);
}
self.client.emit("event", presenceEvent);
});
// strip out events which aren't for the given room_id (e.g presence)
// and also ephemeral events (which we're assuming is anything without
// and event ID because the /events API doesn't separate them).
const events = res.chunk.filter(function(e) {
return e.room_id === peekRoom.roomId && e.event_id;
}).map(self.client.getEventMapper());
peekRoom.addLiveEvents(events);
self._peekPoll(peekRoom, res.end);
}, function(err) {
logger.error("[%s] Peek poll failed: %s", peekRoom.roomId, err);
setTimeout(function() {
self._peekPoll(peekRoom, token);
}, 30 * 1000);
});
};
/**
* Returns the current state of this sync object
* @see module:client~MatrixClient#event:"sync"
* @return {?String}
*/
SyncApi.prototype.getSyncState = function() {
return this._syncState;
};
/**
* Returns the additional data object associated with
* the current sync state, or null if there is no
* such data.
* Sync errors, if available, are put in the 'error' key of
* this object.
* @return {?Object}
*/
SyncApi.prototype.getSyncStateData = function() {
return this._syncStateData;
};
SyncApi.prototype.recoverFromSyncStartupError = async function(savedSyncPromise, err) {
// Wait for the saved sync to complete - we send the pushrules and filter requests
// before the saved sync has finished so they can run in parallel, but only process
// the results after the saved sync is done. Equivalently, we wait for it to finish
// before reporting failures from these functions.
await savedSyncPromise;
const keepaliveProm = this._startKeepAlives();
this._updateSyncState("ERROR", { error: err });
await keepaliveProm;
};
/**
* Is the lazy loading option different than in previous session?
* @param {bool} lazyLoadMembers current options for lazy loading
* @return {bool} whether or not the option has changed compared to the previous session */
SyncApi.prototype._wasLazyLoadingToggled = async function(lazyLoadMembers) {
lazyLoadMembers = !!lazyLoadMembers;
// assume it was turned off before
// if we don't know any better
let lazyLoadMembersBefore = false;
const isStoreNewlyCreated = await this.client.store.isNewlyCreated();
if (!isStoreNewlyCreated) {
const prevClientOptions = await this.client.store.getClientOptions();
if (prevClientOptions) {
lazyLoadMembersBefore = !!prevClientOptions.lazyLoadMembers;
}
return lazyLoadMembersBefore !== lazyLoadMembers;
}
return false;
};
SyncApi.prototype._shouldAbortSync = function(error) {
if (error.errcode === "M_UNKNOWN_TOKEN") {
// The logout already happened, we just need to stop.
logger.warn("Token no longer valid - assuming logout");
this.stop();
return true;
}
return false;
};
/**
* Main entry point
*/
SyncApi.prototype.sync = function() {
const client = this.client;
const self = this;
this._running = true;
if (global.document) {
this._onOnlineBound = this._onOnline.bind(this);
global.document.addEventListener("online", this._onOnlineBound, false);
}
let savedSyncPromise = Promise.resolve();
let savedSyncToken = null;
// We need to do one-off checks before we can begin the /sync loop.
// These are:
// 1) We need to get push rules so we can check if events should bing as we get
// them from /sync.
// 2) We need to get/create a filter which we can use for /sync.
// 3) We need to check the lazy loading option matches what was used in the
// stored sync. If it doesn't, we can't use the stored sync.
async function getPushRules() {
try {
debuglog("Getting push rules...");
const result = await client.getPushRules();
debuglog("Got push rules");
client.pushRules = result;
} catch (err) {
logger.error("Getting push rules failed", err);
if (self._shouldAbortSync(err)) return;
// wait for saved sync to complete before doing anything else,
// otherwise the sync state will end up being incorrect
debuglog("Waiting for saved sync before retrying push rules...");
await self.recoverFromSyncStartupError(savedSyncPromise, err);
getPushRules();
return;
}
checkLazyLoadStatus(); // advance to the next stage
}
const checkLazyLoadStatus = async () => {
debuglog("Checking lazy load status...");
if (this.opts.lazyLoadMembers && client.isGuest()) {
this.opts.lazyLoadMembers = false;
}
if (this.opts.lazyLoadMembers) {
debuglog("Checking server lazy load support...");
const supported = await client.doesServerSupportLazyLoading();
if (supported) {
try {
debuglog("Creating and storing lazy load sync filter...");
this.opts.filter = await client.createFilter(
Filter.LAZY_LOADING_SYNC_FILTER,
);
debuglog("Created and stored lazy load sync filter");
} catch (err) {
logger.error(
"Creating and storing lazy load sync filter failed",
err,
);
throw err;
}
} else {
debuglog("LL: lazy loading requested but not supported " +
"by server, so disabling");
this.opts.lazyLoadMembers = false;
}
}
// need to vape the store when enabling LL and wasn't enabled before
debuglog("Checking whether lazy loading has changed in store...");
const shouldClear = await this._wasLazyLoadingToggled(this.opts.lazyLoadMembers);
if (shouldClear) {
this._storeIsInvalid = true;
const reason = InvalidStoreError.TOGGLED_LAZY_LOADING;
const error = new InvalidStoreError(reason, !!this.opts.lazyLoadMembers);
this._updateSyncState("ERROR", { error });
// bail out of the sync loop now: the app needs to respond to this error.
// we leave the state as 'ERROR' which isn't great since this normally means
// we're retrying. The client must be stopped before clearing the stores anyway
// so the app should stop the client, clear the store and start it again.
logger.warn("InvalidStoreError: store is not usable: stopping sync.");
return;
}
if (this.opts.lazyLoadMembers && this.opts.crypto) {
this.opts.crypto.enableLazyLoading();
}
try {
debuglog("Storing client options...");
await this.client._storeClientOptions();
debuglog("Stored client options");
} catch (err) {
logger.error("Storing client options failed", err);
throw err;
}
getFilter(); // Now get the filter and start syncing
};
async function getFilter() {
debuglog("Getting filter...");
let filter;
if (self.opts.filter) {
filter = self.opts.filter;
} else {
filter = new Filter(client.credentials.userId);
filter.setTimelineLimit(self.opts.initialSyncLimit);
}
let filterId;
try {
filterId = await client.getOrCreateFilter(
getFilterName(client.credentials.userId), filter,
);
} catch (err) {
logger.error("Getting filter failed", err);
if (self._shouldAbortSync(err)) return;
// wait for saved sync to complete before doing anything else,
// otherwise the sync state will end up being incorrect
debuglog("Waiting for saved sync before retrying filter...");
await self.recoverFromSyncStartupError(savedSyncPromise, err);
getFilter();
return;
}
// reset the notifications timeline to prepare it to paginate from
// the current point in time.
// The right solution would be to tie /sync pagination tokens into
// /notifications API somehow.
client.resetNotifTimelineSet();
if (self._currentSyncRequest === null) {
// Send this first sync request here so we can then wait for the saved
// sync data to finish processing before we process the results of this one.
debuglog("Sending first sync request...");
self._currentSyncRequest = self._doSyncRequest({ filterId }, savedSyncToken);
}
// Now wait for the saved sync to finish...
debuglog("Waiting for saved sync before starting sync processing...");
await savedSyncPromise;
self._sync({ filterId });
}
if (client.isGuest()) {
// no push rules for guests, no access to POST filter for guests.
self._sync({});
} else {
// Pull the saved sync token out first, before the worker starts sending
// all the sync data which could take a while. This will let us send our
// first incremental sync request before we've processed our saved data.
debuglog("Getting saved sync token...");
savedSyncPromise = client.store.getSavedSyncToken().then((tok) => {
debuglog("Got saved sync token");
savedSyncToken = tok;
debuglog("Getting saved sync...");
return client.store.getSavedSync();
}).then((savedSync) => {
debuglog(`Got reply from saved sync, exists? ${!!savedSync}`);
if (savedSync) {
return self._syncFromCache(savedSync);
}
}).catch(err => {
logger.error("Getting saved sync failed", err);
});
// Now start the first incremental sync request: this can also
// take a while so if we set it going now, we can wait for it
// to finish while we process our saved sync data.
getPushRules();
}
};
/**
* Stops the sync object from syncing.
*/
SyncApi.prototype.stop = function() {
debuglog("SyncApi.stop");
if (global.document) {
global.document.removeEventListener("online", this._onOnlineBound, false);
this._onOnlineBound = undefined;
}
this._running = false;
if (this._currentSyncRequest) {
this._currentSyncRequest.abort();
}
if (this._keepAliveTimer) {
clearTimeout(this._keepAliveTimer);
this._keepAliveTimer = null;
}
};
/**
* Retry a backed off syncing request immediately. This should only be used when
* the user <b>explicitly</b> attempts to retry their lost connection.
* @return {boolean} True if this resulted in a request being retried.
*/
SyncApi.prototype.retryImmediately = function() {
if (!this._connectionReturnedDefer) {
return false;
}
this._startKeepAlives(0);
return true;
};
/**
* Process a single set of cached sync data.
* @param {Object} savedSync a saved sync that was persisted by a store. This
* should have been acquired via client.store.getSavedSync().
*/
SyncApi.prototype._syncFromCache = async function(savedSync) {
debuglog("sync(): not doing HTTP hit, instead returning stored /sync data");
const nextSyncToken = savedSync.nextBatch;
// Set sync token for future incremental syncing
this.client.store.setSyncToken(nextSyncToken);
// No previous sync, set old token to null
const syncEventData = {
oldSyncToken: null,
nextSyncToken,
catchingUp: false,
fromCache: true,
};
const data = {
next_batch: nextSyncToken,
rooms: savedSync.roomsData,
groups: savedSync.groupsData,
account_data: {
events: savedSync.accountData,
},
};
try {
await this._processSyncResponse(syncEventData, data);
} catch(e) {
logger.error("Error processing cached sync", e.stack || e);
}
// Don't emit a prepared if we've bailed because the store is invalid:
// in this case the client will not be usable until stopped & restarted
// so this would be useless and misleading.
if (!this._storeIsInvalid) {
this._updateSyncState("PREPARED", syncEventData);
}
};
/**
* Invoke me to do /sync calls
* @param {Object} syncOptions
* @param {string} syncOptions.filterId
* @param {boolean} syncOptions.hasSyncedBefore
*/
SyncApi.prototype._sync = async function(syncOptions) {
const client = this.client;
if (!this._running) {
debuglog("Sync no longer running: exiting.");
if (this._connectionReturnedDefer) {
this._connectionReturnedDefer.reject();
this._connectionReturnedDefer = null;
}
this._updateSyncState("STOPPED");
return;
}
const syncToken = client.store.getSyncToken();
let data;
try {
//debuglog('Starting sync since=' + syncToken);
if (this._currentSyncRequest === null) {
this._currentSyncRequest = this._doSyncRequest(syncOptions, syncToken);
}
data = await this._currentSyncRequest;
} catch (e) {
this._onSyncError(e, syncOptions);
return;
} finally {
this._currentSyncRequest = null;
}
//debuglog('Completed sync, next_batch=' + data.next_batch);
// set the sync token NOW *before* processing the events. We do this so
// if something barfs on an event we can skip it rather than constantly
// polling with the same token.
client.store.setSyncToken(data.next_batch);
// Reset after a successful sync
this._failedSyncCount = 0;
await client.store.setSyncData(data);
const syncEventData = {
oldSyncToken: syncToken,
nextSyncToken: data.next_batch,
catchingUp: this._catchingUp,
};
if (this.opts.crypto) {
// tell the crypto module we're about to process a sync
// response
await this.opts.crypto.onSyncWillProcess(syncEventData);
}
try {
await this._processSyncResponse(syncEventData, data);
} catch(e) {
// log the exception with stack if we have it, else fall back
// to the plain description
logger.error("Caught /sync error", e.stack || e);
// Emit the exception for client handling
this.client.emit("sync.unexpectedError", e);
}
// update this as it may have changed
syncEventData.catchingUp = this._catchingUp;
// emit synced events
if (!syncOptions.hasSyncedBefore) {
this._updateSyncState("PREPARED", syncEventData);
syncOptions.hasSyncedBefore = true;
}
// tell the crypto module to do its processing. It may block (to do a
// /keys/changes request).
if (this.opts.crypto) {
await this.opts.crypto.onSyncCompleted(syncEventData);
}
// keep emitting SYNCING -> SYNCING for clients who want to do bulk updates
this._updateSyncState("SYNCING", syncEventData);
if (client.store.wantsSave()) {
// We always save the device list (if it's dirty) before saving the sync data:
// this means we know the saved device list data is at least as fresh as the
// stored sync data which means we don't have to worry that we may have missed
// device changes. We can also skip the delay since we're not calling this very
// frequently (and we don't really want to delay the sync for it).
if (this.opts.crypto) {
await this.opts.crypto.saveDeviceList(0);
}
// tell databases that everything is now in a consistent state and can be saved.
client.store.save();
}
// Begin next sync
this._sync(syncOptions);
};
SyncApi.prototype._doSyncRequest = function(syncOptions, syncToken) {
const qps = this._getSyncParams(syncOptions, syncToken);
return this.client._http.authedRequest(
undefined, "GET", "/sync", qps, undefined,
qps.timeout + BUFFER_PERIOD_MS,
);
};
SyncApi.prototype._getSyncParams = function(syncOptions, syncToken) {
let pollTimeout = this.opts.pollTimeout;
if (this.getSyncState() !== 'SYNCING' || this._catchingUp) {
// unless we are happily syncing already, we want the server to return
// as quickly as possible, even if there are no events queued. This
// serves two purposes:
//
// * When the connection dies, we want to know asap when it comes back,
// so that we can hide the error from the user. (We don't want to
// have to wait for an event or a timeout).
//
// * We want to know if the server has any to_device messages queued up
// for us. We do that by calling it with a zero timeout until it
// doesn't give us any more to_device messages.
this._catchingUp = true;
pollTimeout = 0;
}
let filterId = syncOptions.filterId;
if (this.client.isGuest() && !filterId) {
filterId = this._getGuestFilter();
}
const qps = {
filter: filterId,
timeout: pollTimeout,
};
if (this.opts.disablePresence) {
qps.set_presence = "offline";
}
if (syncToken) {
qps.since = syncToken;
} else {
// use a cachebuster for initialsyncs, to make sure that
// we don't get a stale sync
// (https://github.com/vector-im/vector-web/issues/1354)
qps._cacheBuster = Date.now();
}
if (this.getSyncState() == 'ERROR' || this.getSyncState() == 'RECONNECTING') {
// we think the connection is dead. If it comes back up, we won't know
// about it till /sync returns. If the timeout= is high, this could
// be a long time. Set it to 0 when doing retries so we don't have to wait
// for an event or a timeout before emiting the SYNCING event.
qps.timeout = 0;
}
return qps;
};
SyncApi.prototype._onSyncError = function(err, syncOptions) {
if (!this._running) {
debuglog("Sync no longer running: exiting");
if (this._connectionReturnedDefer) {
this._connectionReturnedDefer.reject();
this._connectionReturnedDefer = null;
}
this._updateSyncState("STOPPED");
return;
}
logger.error("/sync error %s", err);
logger.error(err);
if(this._shouldAbortSync(err)) {
return;
}
this._failedSyncCount++;
logger.log('Number of consecutive failed sync requests:', this._failedSyncCount);
debuglog("Starting keep-alive");
// Note that we do *not* mark the sync connection as
// lost yet: we only do this if a keepalive poke
// fails, since long lived HTTP connections will
// go away sometimes and we shouldn't treat this as
// erroneous. We set the state to 'reconnecting'
// instead, so that clients can observe this state
// if they wish.
this._startKeepAlives().then((connDidFail) => {
// Only emit CATCHUP if we detected a connectivity error: if we didn't,
// it's quite likely the sync will fail again for the same reason and we
// want to stay in ERROR rather than keep flip-flopping between ERROR
// and CATCHUP.
if (connDidFail && this.getSyncState() === 'ERROR') {
this._updateSyncState("CATCHUP", {
oldSyncToken: null,
nextSyncToken: null,
catchingUp: true,
});
}
this._sync(syncOptions);
});
this._currentSyncRequest = null;
// Transition from RECONNECTING to ERROR after a given number of failed syncs
this._updateSyncState(
this._failedSyncCount >= FAILED_SYNC_ERROR_THRESHOLD ?
"ERROR" : "RECONNECTING",
{ error: err },
);
};
/**
* Process data returned from a sync response and propagate it
* into the model objects
*
* @param {Object} syncEventData Object containing sync tokens associated with this sync
* @param {Object} data The response from /sync
*/
SyncApi.prototype._processSyncResponse = async function(
syncEventData, data,
) {
const client = this.client;
const self = this;
// data looks like:
// {
// next_batch: $token,
// presence: { events: [] },
// account_data: { events: [] },
// device_lists: { changed: ["@user:server", ... ]},
// to_device: { events: [] },
// device_one_time_keys_count: { signed_curve25519: 42 },
// rooms: {
// invite: {
// $roomid: {
// invite_state: { events: [] }
// }
// },
// join: {
// $roomid: {
// state: { events: [] },
// timeline: { events: [], prev_batch: $token, limited: true },
// ephemeral: { events: [] },
// summary: {
// m.heroes: [ $user_id ],
// m.joined_member_count: $count,
// m.invited_member_count: $count
// },
// account_data: { events: [] },
// unread_notifications: {
// highlight_count: 0,
// notification_count: 0,
// }
// }
// },
// leave: {
// $roomid: {
// state: { events: [] },
// timeline: { events: [], prev_batch: $token }
// }
// }
// },
// groups: {
// invite: {
// $groupId: {
// inviter: $inviter,
// profile: {
// avatar_url: $avatarUrl,
// name: $groupName,
// },
// },
// },
// join: {},