From eb988ae7824b84bfd69d35458c5c7ef254f7608d Mon Sep 17 00:00:00 2001 From: andrewDoing Date: Wed, 13 Mar 2024 09:27:45 -0500 Subject: [PATCH 01/21] working refactor --- .../statefulClient/GraphNotificationClient.ts | 395 +++++++++++++----- .../src/statefulClient/ThreadEventEmitter.ts | 6 + packages/mgt-chat/src/utils/Timer.ts | 10 +- 3 files changed, 306 insertions(+), 105 deletions(-) diff --git a/packages/mgt-chat/src/statefulClient/GraphNotificationClient.ts b/packages/mgt-chat/src/statefulClient/GraphNotificationClient.ts index 054bf9742c..c5522f18a8 100644 --- a/packages/mgt-chat/src/statefulClient/GraphNotificationClient.ts +++ b/packages/mgt-chat/src/statefulClient/GraphNotificationClient.ts @@ -6,7 +6,13 @@ */ import { BetaGraph, IGraph, Providers, createFromProvider, error, log } from '@microsoft/mgt-element'; -import { HubConnection, HubConnectionBuilder, IHttpConnectionOptions, LogLevel } from '@microsoft/signalr'; +import { + HubConnection, + HubConnectionBuilder, + HubConnectionState, + IHttpConnectionOptions, + LogLevel +} from '@microsoft/signalr'; import { ThreadEventEmitter } from './ThreadEventEmitter'; import type { Entity, @@ -19,6 +25,7 @@ import { GraphConfig } from './GraphConfig'; import { SubscriptionsCache } from './Caching/SubscriptionCache'; import { Timer } from '../utils/Timer'; import { getOrGenerateGroupId } from './getOrGenerateGroupId'; +import { v4 as uuid } from 'uuid'; export const appSettings = { defaultSubscriptionLifetimeInMinutes: 10, @@ -51,12 +58,16 @@ const isMembershipNotification = (o: Notification): o is Notification[] = []; - // Cache the subscription in storage for re-hydration on page refreshes - awaits.push(this.cacheSubscription(subscription)); - - // create a connection to the web socket if one does not exist - if (!this.connection) awaits.push(this.createSignalRConnection(subscription.notificationUrl)); - log('Invoked CreateSubscription'); - return Promise.all(awaits); + return subscription; } - private readonly startRenewalTimer = () => { - if (this.renewalTimeout) this.timer.clearTimeout(this.renewalTimeout); - this.renewalTimeout = this.timer.setTimeout(this.syncRenewalTimerWrapper, appSettings.renewalTimerInterval * 1000); - log(`Start renewal timer . Id: ${this.renewalTimeout}`); - }; - - private readonly syncRenewalTimerWrapper = () => void this.renewalTimer(); - - private readonly renewalTimer = async () => { - log(`running subscription renewal timer for chatId: ${this.chatId} sessionId: ${this.sessionId}`); - const subscriptions = (await this.subscriptionCache.loadSubscriptions(this.chatId))?.subscriptions || []; - if (subscriptions.length === 0) { - log(`No subscriptions found in session state. Stop renewal timer ${this.renewalTimeout}.`); - if (this.renewalTimeout) this.timer.clearTimeout(this.renewalTimeout); - return; - } - - for (const subscription of subscriptions) { - if (!subscription.expirationDateTime) continue; - const expirationTime = new Date(subscription.expirationDateTime); - const now = new Date(); - const diff = Math.round((expirationTime.getTime() - now.getTime()) / 1000); - - if (diff <= appSettings.renewalThreshold) { - this.renewalCount++; - log(`Renewing Graph subscription. RenewalCount: ${this.renewalCount}`); - // stop interval to prevent new invokes until refresh is ready. - if (this.renewalTimeout) this.timer.clearTimeout(this.renewalTimeout); - this.renewalTimeout = undefined; - await this.renewChatSubscriptions(); - // There is one subscription that need expiration, all subscriptions will be renewed - break; - } - } - this.renewalTimeout = this.timer.setTimeout(this.syncRenewalTimerWrapper, appSettings.renewalTimerInterval * 1000); - }; - - public renewChatSubscriptions = async () => { - const expirationTime = new Date( - new Date().getTime() + appSettings.defaultSubscriptionLifetimeInMinutes * 60 * 1000 - ); - - const subscriptionCache = await this.subscriptionCache.loadSubscriptions(this.chatId); - const awaits: Promise[] = []; - for (const subscription of subscriptionCache?.subscriptions || []) { - if (!subscription.id) continue; - // the renewSubscription method caches the updated subscription to track the new expiration time - awaits.push(this.renewSubscription(subscription.id, expirationTime.toISOString())); - log(`Invoked RenewSubscription ${subscription.id}`); - } - await Promise.all(awaits); - if (!this.renewalTimeout) { - this.renewalTimeout = this.timer.setTimeout( - this.syncRenewalTimerWrapper, - appSettings.renewalTimerInterval * 1000 - ); - } - }; + // private readonly startRenewalTimer = () => { + // if (this.renewalTimeout) this.timer.clearTimeout(this.renewalTimeout); + // this.renewalTimeout = this.timer.setTimeout('renewal:' + this.instanceId, this.syncRenewalTimerWrapper, appSettings.renewalTimerInterval * 1000); + // log(`Start renewal timer . Id: ${this.renewalTimeout}`); + // }; + + // private readonly syncRenewalTimerWrapper = () => void this.renewalTimer(); + + // private readonly renewalTimer = async () => { + // log(`running subscription renewal timer for chatId: ${this.chatId} sessionId: ${this.sessionId}`); + // const subscriptions = (await this.subscriptionCache.loadSubscriptions(this.chatId))?.subscriptions || []; + // if (subscriptions.length === 0) { + // log(`No subscriptions found in session state. Stop renewal timer ${this.renewalTimeout}.`); + // if (this.renewalTimeout) this.timer.clearTimeout(this.renewalTimeout); + // return; + // } + + // for (const subscription of subscriptions) { + // if (!subscription.expirationDateTime) continue; + // const expirationTime = new Date(subscription.expirationDateTime); + // const now = new Date(); + // const diff = Math.round((expirationTime.getTime() - now.getTime()) / 1000); + + // if (diff <= appSettings.renewalThreshold) { + // this.renewalCount++; + // log(`Renewing Graph subscription. RenewalCount: ${this.renewalCount}`); + // // stop interval to prevent new invokes until refresh is ready. + // if (this.renewalTimeout) this.timer.clearTimeout(this.renewalTimeout); + // this.renewalTimeout = undefined; + // await this.renewChatSubscriptions(); + // // There is one subscription that need expiration, all subscriptions will be renewed + // break; + // } + // } + // this.renewalTimeout = this.timer.setTimeout('renewal:' + this.instanceId, this.syncRenewalTimerWrapper, appSettings.renewalTimerInterval * 1000); + // }; + + // public renewChatSubscriptions = async () => { + // const expirationTime = new Date( + // new Date().getTime() + appSettings.defaultSubscriptionLifetimeInMinutes * 60 * 1000 + // ); + + // const subscriptionCache = await this.subscriptionCache.loadSubscriptions(this.chatId); + // const awaits: Promise[] = []; + // for (const subscription of subscriptionCache?.subscriptions || []) { + // if (!subscription.id) continue; + // // the renewSubscription method caches the updated subscription to track the new expiration time + // awaits.push(this.renewSubscription(subscription.id)); + // log(`Invoked RenewSubscription ${subscription.id}`); + // } + // await Promise.all(awaits); + // if (!this.renewalTimeout) { + // this.renewalTimeout = this.timer.setTimeout( + // 'renewal:' + this.instanceId, + // this.syncRenewalTimerWrapper, + // appSettings.renewalTimerInterval * 1000 + // ); + // } + // }; public renewSubscription = async (subscriptionId: string, expirationDateTime: string): Promise => { // PATCH /subscriptions/{id} @@ -330,7 +335,11 @@ export class GraphNotificationClient { } private startCleanupTimer() { - this.cleanupTimeout = this.timer.setTimeout(this.cleanupTimerSync, appSettings.removalTimerInterval * 1000); + this.cleanupTimeout = this.timer.setTimeout( + 'renewal:' + this.instanceId, + this.cleanupTimerSync, + appSettings.removalTimerInterval * 1000 + ); } private readonly cleanupTimerSync = () => { @@ -363,46 +372,228 @@ export class GraphNotificationClient { this.connection = undefined; } - private async unsubscribeFromChatNotifications(chatId: string) { - await this.closeSignalRConnection(); - const cacheData = await this.subscriptionCache.loadSubscriptions(chatId); - if (cacheData) { - await Promise.all([ - this.removeSubscriptions(cacheData.subscriptions), - this.subscriptionCache.deleteCachedSubscriptions(chatId) - ]); - } - } + // private async unsubscribeFromChatNotifications(chatId: string) { + // await this.closeSignalRConnection(); + // const cacheData = await this.subscriptionCache.loadSubscriptions(chatId); + // if (cacheData) { + // await Promise.all([ + // this.removeSubscriptions(cacheData.subscriptions), + // this.subscriptionCache.deleteCachedSubscriptions(chatId) + // ]); + // } + // } public async subscribeToChatNotifications(chatId: string, sessionId: string) { + log(`Chat subscription with id: ${chatId} and session id: ${sessionId}`); + this.wasConnected = undefined; this.chatId = chatId; this.sessionId = sessionId; - // MGT uses a per-user cache, so no concerns of loading the cached data for another user. - const cacheData = await this.subscriptionCache.loadSubscriptions(chatId); - if (cacheData) { - // check subscription validity & renew if all still valid otherwise recreate - const someExpired = cacheData.subscriptions.some( - s => s.expirationDateTime && new Date(s.expirationDateTime) <= new Date() - ); - // for a given user + app + chatId + sessionId they only get one websocket and receive all notifications via that websocket. - const webSocketUrl = cacheData.subscriptions.find(s => s.notificationUrl)?.notificationUrl; - if (!someExpired && webSocketUrl) { - // if we have a websocket url and all the subscriptions are valid, we can reuse the websocket and return before recreating subscriptions. - await this.createSignalRConnection(webSocketUrl); - await this.renewChatSubscriptions(); - return; - } else if (someExpired) { - // if some are expired, remove them and continue to recreate the subscription - await this.removeSubscriptions(cacheData.subscriptions); - } - await this.subscriptionCache.deleteCachedSubscriptions(chatId); - } - const promises: Promise[] = []; + this.renewalTimeout = this.timer.setTimeout('renewal:' + this.instanceId, this.renewalSync, 0); + } + + // private async createSubscriptions(chatId: string, sessionId: string) { + // const changeTypes: ChangeTypes[] = ['created', 'updated', 'deleted']; + + // // build subscription request + // const expirationDateTime = new Date( + // new Date().getTime() + appSettings.defaultSubscriptionLifetimeInMinutes * 60 * 1000 + // ).toISOString(); + // const subscriptionDefinition: Subscription = { + // changeType: changeTypes.join(','), + // notificationUrl: `${GraphConfig.webSocketsPrefix}?groupId=${groupId}`, + // resource: resourcePath, + // expirationDateTime, + // includeResourceData: true, + // clientState: 'wsssecret' + // }; + + // log('subscribing to changes for ' + resourcePath); + // const subscriptionEndpoint = GraphConfig.subscriptionEndpoint; + // // send subscription POST to Graph + // const subscription: Subscription = (await this.subscriptionGraph + // .api(subscriptionEndpoint) + // .post(subscriptionDefinition)) as Subscription; + // if (!subscription?.notificationUrl) throw new Error('Subscription not created'); + // log(subscription); + + // this.subscriptionId = subscription.id!; + // await this.cacheSubscription(userId, subscription); + + // log('Subscription created.'); + + // const promises: Promise[] = []; + // promises.push(this.subscribeToResource(`/chats/${chatId}/messages`, ['created', 'updated', 'deleted'])); + // promises.push(this.subscribeToResource(`/chats/${chatId}/members`, ['created', 'deleted'])); + // promises.push(this.subscribeToResource(`/chats/${chatId}`, ['updated', 'deleted'])); + // await Promise.all(promises).catch((e: Error) => { + // this?.emitter.graphNotificationClientError(e); + // }); + // } + + private createSubscriptions = async (chatId: string, sessionId: string) => { + const promises: Promise[] = []; promises.push(this.subscribeToResource(`/chats/${chatId}/messages`, ['created', 'updated', 'deleted'])); promises.push(this.subscribeToResource(`/chats/${chatId}/members`, ['created', 'deleted'])); promises.push(this.subscribeToResource(`/chats/${chatId}`, ['updated', 'deleted'])); - await Promise.all(promises).catch((e: Error) => { + const subscriptions = await Promise.all(promises).catch((e: Error) => { + this?.emitter.graphNotificationClientError(e); + }); + + // Cache the subscriptions in storage for re-hydration on page refreshes + const awaits: Promise[] = []; + // ensure that subscriptions is not void + if (subscriptions) { + for (const subscription of subscriptions.filter(Boolean)) { + awaits.push(this.cacheSubscription(subscription!)); + } + } + + await Promise.all(awaits).catch((e: Error) => { this?.emitter.graphNotificationClientError(e); }); + + return subscriptions; + }; + + private async deleteCachedSubscription(subscriptionId: string) { + try { + log(`Removing subscription ${subscriptionId} from cache...`); + await this.subscriptionCache.deleteCachedSubscriptions(subscriptionId); + log(`Successfully removed subscription ${subscriptionId} from cache.`); + } catch (e) { + error(`Failed to remove subscription ${subscriptionId} from cache.`, e); + } + } + + private async getSubscriptions(chatId: string): Promise { + const subscriptions = (await this.subscriptionCache.loadSubscriptions(chatId))?.subscriptions || []; + return subscriptions.length > 0 ? subscriptions : undefined; + } + + private trySwitchToConnected() { + if (this.wasConnected !== true) { + log('The user will receive notifications from the chat subscriptions.'); + this.wasConnected = true; + this.emitter?.connected(); + } + } + + private trySwitchToDisconnected(ignoreIfUndefined = false) { + if (ignoreIfUndefined && this.wasConnected === undefined) return; + if (this.wasConnected !== false) { + log('The user will NOT receive notifications from the chat subscriptions.'); + this.wasConnected = false; + this.emitter?.disconnected(); + } } + + private readonly renewalSync = () => { + void this.renewal(); + }; + + private readonly renewal = async () => { + let nextRenewalTimeInSec = appSettings.renewalTimerInterval; + try { + // if there are current subscriptions for this chat id... + let subscriptions = await this.getSubscriptions(this.chatId); + if (subscriptions) { + // attempt a renewal if necessary + try { + const expirationTime = new Date( + new Date().getTime() + appSettings.defaultSubscriptionLifetimeInMinutes * 60 * 1000 + ); + for (let subscription of subscriptions) { + const expirationTime = new Date(subscription.expirationDateTime!); + const diff = Math.round((expirationTime.getTime() - new Date().getTime()) / 1000); + if (diff <= 0) { + log(`Renewing chat subscription ${subscription.id} that has already expired...`); + this.trySwitchToDisconnected(true); + await this.renewSubscription(this.chatId, expirationTime.toISOString()); + log(`Successfully renewed chat subscription ${subscription.id}.`); + } else if (diff <= appSettings.renewalThreshold) { + log(`Renewing chat subscription ${subscription.id} that will expire in ${diff} seconds...`); + await this.renewSubscription(this.chatId, expirationTime.toISOString()); + log(`Successfully renewed chat subscription ${subscription.id}.`); + } + } + } catch (e) { + error(`Failed to renew subscriptions.`, e); + const promises: Promise[] = []; + for (let subscription of subscriptions) { + promises.push(this.deleteCachedSubscription(this.chatId)); + } + await Promise.all(promises); + subscriptions = undefined; + } + } + + // if there are no subscriptions, try to create them + if (!subscriptions) { + try { + this.trySwitchToDisconnected(true); + const subscriptions = await this.createSubscriptions(this.chatId, this.sessionId); + } catch (e) { + const err = e as { statusCode?: number; message: string }; + if (err.statusCode === 403 && err.message.indexOf('has reached its limit') > 0) { + // if the limit is reached, back-off (NOTE: this should probably be a 429) + nextRenewalTimeInSec = appSettings.renewalTimerInterval * 3; + throw new Error( + `Failed to create new subscriptions due to a limitation; retrying in ${nextRenewalTimeInSec} seconds: ${err.message}.` + ); + } else if (err.statusCode === 403 || err.statusCode === 402) { + // permanent error, stop renewal + error('Failed to create new subscriptions due to a permanent condition; stopping renewals.', e); + return; // exit without setting the next renewal timer + } else { + // transient error, retry + throw new Error( + `Failed to create new subscriptions due to a transient condition; retrying in ${nextRenewalTimeInSec} seconds: ${err.message}.` + ); + } + } + } + + // create or reconnect the SignalR connection + // notificationUrl comes in the form of websockets:https://graph.microsoft.com/beta/subscriptions/notificationChannel/websockets/?groupid=&sessionid=default + // if changes, we need to create a new connection + const subscriptionIds = subscriptions?.map(s => { + if (s) { + return s.id; + } + }); + if (this.connection?.state === HubConnectionState.Connected) { + await this.connection?.send('ping'); // ensure the connection is still alive + } + if (!this.connection) { + // log(`Creating a new SignalR connection for subscription ${subscription.id!}...`); + this.trySwitchToDisconnected(true); + this.lastNotificationUrl = subscriptions![0]?.notificationUrl!; + await this.createSignalRConnection(subscriptions![0]?.notificationUrl!); + log(`Successfully created a new SignalR connection for subscriptions: ${subscriptionIds}`); + } else if (this.connection.state !== HubConnectionState.Connected) { + log(`Reconnecting SignalR connection for subscriptions: ${subscriptionIds}...`); + this.trySwitchToDisconnected(true); + await this.connection.start(); + log(`Successfully reconnected SignalR connection for subscriptions: ${subscriptionIds}`); + } else if (this.lastNotificationUrl !== subscriptions![0]?.notificationUrl) { + log(`Updating SignalR connection for subscriptions: ${subscriptionIds} due to new notification URL...`); + this.trySwitchToDisconnected(true); + await this.closeSignalRConnection(); + this.lastNotificationUrl = subscriptions![0]?.notificationUrl!; + await this.createSignalRConnection(subscriptions![0]?.notificationUrl!); + log(`Successfully updated SignalR connection for subscriptions: ${subscriptionIds}`); + } + + // emit the new connection event if necessary + this.trySwitchToConnected(); + } catch (e) { + error('Error in chat subscription connection process.', e); + this.trySwitchToDisconnected(); + } + this.renewalTimeout = this.timer.setTimeout( + 'renewal:' + this.instanceId, + this.renewalSync, + nextRenewalTimeInSec * 1000 + ); + }; } diff --git a/packages/mgt-chat/src/statefulClient/ThreadEventEmitter.ts b/packages/mgt-chat/src/statefulClient/ThreadEventEmitter.ts index fcaf8a2193..60db0e537a 100644 --- a/packages/mgt-chat/src/statefulClient/ThreadEventEmitter.ts +++ b/packages/mgt-chat/src/statefulClient/ThreadEventEmitter.ts @@ -66,6 +66,12 @@ export class ThreadEventEmitter { notificationsSubscribedForResource(resouce: string) { this.emitter.emit('notificationsSubscribedForResource', resouce); } + disconnected() { + this.emitter.emit('disconnected'); + } + connected() { + this.emitter.emit('connected'); + } graphNotificationClientError(error: Error) { this.emitter.emit('graphNotificationClientError', error); } diff --git a/packages/mgt-chat/src/utils/Timer.ts b/packages/mgt-chat/src/utils/Timer.ts index bb3c9b86a0..0ddf3745f2 100644 --- a/packages/mgt-chat/src/utils/Timer.ts +++ b/packages/mgt-chat/src/utils/Timer.ts @@ -5,7 +5,6 @@ * ------------------------------------------------------------------------------------------- */ -import { v4 as uuid } from 'uuid'; import { TimerWork } from './timerWorker'; export interface Work { @@ -21,10 +20,12 @@ export class Timer { this.worker.port.onmessage = this.onMessage; } - public setTimeout(callback: () => void, delay: number): string { + // NOTE: this.work was increasing in size every time there was a new timeout because id was uuid() + // but clearTimeout was only being called on teardown. + public setTimeout(id: string, callback: () => void, delay: number): string { const timeoutWork: TimerWork = { type: 'setTimeout', - id: uuid(), + id, delay }; @@ -46,6 +47,8 @@ export class Timer { } } + /* + // NOTE: setInterval is not used in the library, but it is here for reference public setInterval(callback: () => void, delay: number): string { const intervalWork: TimerWork = { type: 'setInterval', @@ -70,6 +73,7 @@ export class Timer { this.work.delete(id); } } + */ private readonly onMessage = (event: MessageEvent): void => { const intervalWork = event.data; From 0470770f29114e1bbd6e434330e340a2bdbee885 Mon Sep 17 00:00:00 2001 From: andrewDoing Date: Wed, 13 Mar 2024 11:59:20 -0500 Subject: [PATCH 02/21] progress --- .../statefulClient/GraphNotificationClient.ts | 185 ++++-------------- 1 file changed, 36 insertions(+), 149 deletions(-) diff --git a/packages/mgt-chat/src/statefulClient/GraphNotificationClient.ts b/packages/mgt-chat/src/statefulClient/GraphNotificationClient.ts index c5522f18a8..c23b1901e0 100644 --- a/packages/mgt-chat/src/statefulClient/GraphNotificationClient.ts +++ b/packages/mgt-chat/src/statefulClient/GraphNotificationClient.ts @@ -186,11 +186,7 @@ export class GraphNotificationClient { private readonly cacheSubscription = async (subscriptionRecord: Subscription): Promise => { log(subscriptionRecord); - await this.subscriptionCache.cacheSubscription(this.chatId, subscriptionRecord); - - // // only start timer once. undefined for renewalInterval is semaphore it has stopped. - // if (this.renewalTimeout === undefined) this.startRenewalTimer(); }; private async subscribeToResource(resourcePath: string, changeTypes: ChangeTypes[]) { @@ -207,14 +203,19 @@ export class GraphNotificationClient { clientState: 'wsssecret' }; - log('subscribing to changes for ' + resourcePath); + log(`subscribing to changes for ${resourcePath}`); const subscriptionEndpoint = GraphConfig.subscriptionEndpoint; const subscriptionGraph = this.subscriptionGraph; + if (!subscriptionGraph) return; // send subscription POST to Graph - const subscription: Subscription = (await subscriptionGraph - .api(subscriptionEndpoint) - .post(subscriptionDefinition)) as Subscription; + let subscription: Subscription; + try { + subscription = (await subscriptionGraph.api(subscriptionEndpoint).post(subscriptionDefinition)) as Subscription; + } catch (error) { + log(`Error creating subscription: ${error}`); + throw error; + } if (!subscription?.notificationUrl) throw new Error('Subscription not created'); log(subscription); @@ -222,68 +223,9 @@ export class GraphNotificationClient { return subscription; } - // private readonly startRenewalTimer = () => { - // if (this.renewalTimeout) this.timer.clearTimeout(this.renewalTimeout); - // this.renewalTimeout = this.timer.setTimeout('renewal:' + this.instanceId, this.syncRenewalTimerWrapper, appSettings.renewalTimerInterval * 1000); - // log(`Start renewal timer . Id: ${this.renewalTimeout}`); - // }; - - // private readonly syncRenewalTimerWrapper = () => void this.renewalTimer(); - - // private readonly renewalTimer = async () => { - // log(`running subscription renewal timer for chatId: ${this.chatId} sessionId: ${this.sessionId}`); - // const subscriptions = (await this.subscriptionCache.loadSubscriptions(this.chatId))?.subscriptions || []; - // if (subscriptions.length === 0) { - // log(`No subscriptions found in session state. Stop renewal timer ${this.renewalTimeout}.`); - // if (this.renewalTimeout) this.timer.clearTimeout(this.renewalTimeout); - // return; - // } - - // for (const subscription of subscriptions) { - // if (!subscription.expirationDateTime) continue; - // const expirationTime = new Date(subscription.expirationDateTime); - // const now = new Date(); - // const diff = Math.round((expirationTime.getTime() - now.getTime()) / 1000); - - // if (diff <= appSettings.renewalThreshold) { - // this.renewalCount++; - // log(`Renewing Graph subscription. RenewalCount: ${this.renewalCount}`); - // // stop interval to prevent new invokes until refresh is ready. - // if (this.renewalTimeout) this.timer.clearTimeout(this.renewalTimeout); - // this.renewalTimeout = undefined; - // await this.renewChatSubscriptions(); - // // There is one subscription that need expiration, all subscriptions will be renewed - // break; - // } - // } - // this.renewalTimeout = this.timer.setTimeout('renewal:' + this.instanceId, this.syncRenewalTimerWrapper, appSettings.renewalTimerInterval * 1000); - // }; - - // public renewChatSubscriptions = async () => { - // const expirationTime = new Date( - // new Date().getTime() + appSettings.defaultSubscriptionLifetimeInMinutes * 60 * 1000 - // ); - - // const subscriptionCache = await this.subscriptionCache.loadSubscriptions(this.chatId); - // const awaits: Promise[] = []; - // for (const subscription of subscriptionCache?.subscriptions || []) { - // if (!subscription.id) continue; - // // the renewSubscription method caches the updated subscription to track the new expiration time - // awaits.push(this.renewSubscription(subscription.id)); - // log(`Invoked RenewSubscription ${subscription.id}`); - // } - // await Promise.all(awaits); - // if (!this.renewalTimeout) { - // this.renewalTimeout = this.timer.setTimeout( - // 'renewal:' + this.instanceId, - // this.syncRenewalTimerWrapper, - // appSettings.renewalTimerInterval * 1000 - // ); - // } - // }; - public renewSubscription = async (subscriptionId: string, expirationDateTime: string): Promise => { - // PATCH /subscriptions/{id} + // this.renewalCount++; + log(`Renewing Graph subscription for ChatId. RenewalCount: ${this.renewalCount}.`); const renewedSubscription = (await this.graph?.api(`${GraphConfig.subscriptionEndpoint}/${subscriptionId}`).patch({ expirationDateTime })) as Subscription | undefined; @@ -366,88 +308,21 @@ export class GraphNotificationClient { this.startCleanupTimer(); }; - public async closeSignalRConnection() { - // stop the connection and set it to undefined so it will reconnect when next subscription is created. - await this.connection?.stop(); - this.connection = undefined; - } - - // private async unsubscribeFromChatNotifications(chatId: string) { - // await this.closeSignalRConnection(); - // const cacheData = await this.subscriptionCache.loadSubscriptions(chatId); - // if (cacheData) { - // await Promise.all([ - // this.removeSubscriptions(cacheData.subscriptions), - // this.subscriptionCache.deleteCachedSubscriptions(chatId) - // ]); - // } - // } - - public async subscribeToChatNotifications(chatId: string, sessionId: string) { - log(`Chat subscription with id: ${chatId} and session id: ${sessionId}`); - this.wasConnected = undefined; - this.chatId = chatId; - this.sessionId = sessionId; - this.renewalTimeout = this.timer.setTimeout('renewal:' + this.instanceId, this.renewalSync, 0); - } - - // private async createSubscriptions(chatId: string, sessionId: string) { - // const changeTypes: ChangeTypes[] = ['created', 'updated', 'deleted']; - - // // build subscription request - // const expirationDateTime = new Date( - // new Date().getTime() + appSettings.defaultSubscriptionLifetimeInMinutes * 60 * 1000 - // ).toISOString(); - // const subscriptionDefinition: Subscription = { - // changeType: changeTypes.join(','), - // notificationUrl: `${GraphConfig.webSocketsPrefix}?groupId=${groupId}`, - // resource: resourcePath, - // expirationDateTime, - // includeResourceData: true, - // clientState: 'wsssecret' - // }; - - // log('subscribing to changes for ' + resourcePath); - // const subscriptionEndpoint = GraphConfig.subscriptionEndpoint; - // // send subscription POST to Graph - // const subscription: Subscription = (await this.subscriptionGraph - // .api(subscriptionEndpoint) - // .post(subscriptionDefinition)) as Subscription; - // if (!subscription?.notificationUrl) throw new Error('Subscription not created'); - // log(subscription); - - // this.subscriptionId = subscription.id!; - // await this.cacheSubscription(userId, subscription); - - // log('Subscription created.'); - - // const promises: Promise[] = []; - // promises.push(this.subscribeToResource(`/chats/${chatId}/messages`, ['created', 'updated', 'deleted'])); - // promises.push(this.subscribeToResource(`/chats/${chatId}/members`, ['created', 'deleted'])); - // promises.push(this.subscribeToResource(`/chats/${chatId}`, ['updated', 'deleted'])); - // await Promise.all(promises).catch((e: Error) => { - // this?.emitter.graphNotificationClientError(e); - // }); - // } - private createSubscriptions = async (chatId: string, sessionId: string) => { const promises: Promise[] = []; promises.push(this.subscribeToResource(`/chats/${chatId}/messages`, ['created', 'updated', 'deleted'])); promises.push(this.subscribeToResource(`/chats/${chatId}/members`, ['created', 'deleted'])); promises.push(this.subscribeToResource(`/chats/${chatId}`, ['updated', 'deleted'])); - const subscriptions = await Promise.all(promises).catch((e: Error) => { + const results = await Promise.all(promises).catch((e: Error) => { this?.emitter.graphNotificationClientError(e); }); // Cache the subscriptions in storage for re-hydration on page refreshes const awaits: Promise[] = []; - // ensure that subscriptions is not void - if (subscriptions) { - for (const subscription of subscriptions.filter(Boolean)) { - awaits.push(this.cacheSubscription(subscription!)); - } + const subscriptions: Subscription[] = (results as (Subscription | undefined)[]).filter(Boolean) as Subscription[]; + for (let subscription of subscriptions) { + awaits.push(this.cacheSubscription(subscription)); } - await Promise.all(awaits).catch((e: Error) => { this?.emitter.graphNotificationClientError(e); }); @@ -489,13 +364,15 @@ export class GraphNotificationClient { private readonly renewalSync = () => { void this.renewal(); - }; + } - private readonly renewal = async () => { + private readonly renewal = async () =>{ let nextRenewalTimeInSec = appSettings.renewalTimerInterval; try { // if there are current subscriptions for this chat id... + log(`Retrieving chat subscriptions for chat ${this.chatId}...`); let subscriptions = await this.getSubscriptions(this.chatId); + log(`Retrieved chat subscriptions for chat ${this.chatId}: ${subscriptions?.map(s => s?.id).join(', ')}.`); if (subscriptions) { // attempt a renewal if necessary try { @@ -531,7 +408,7 @@ export class GraphNotificationClient { if (!subscriptions) { try { this.trySwitchToDisconnected(true); - const subscriptions = await this.createSubscriptions(this.chatId, this.sessionId); + subscriptions = await this.createSubscriptions(this.chatId, this.sessionId); } catch (e) { const err = e as { statusCode?: number; message: string }; if (err.statusCode === 403 && err.message.indexOf('has reached its limit') > 0) { @@ -556,11 +433,7 @@ export class GraphNotificationClient { // create or reconnect the SignalR connection // notificationUrl comes in the form of websockets:https://graph.microsoft.com/beta/subscriptions/notificationChannel/websockets/?groupid=&sessionid=default // if changes, we need to create a new connection - const subscriptionIds = subscriptions?.map(s => { - if (s) { - return s.id; - } - }); + const subscriptionIds = subscriptions?.map(s => s?.id).filter(Boolean); if (this.connection?.state === HubConnectionState.Connected) { await this.connection?.send('ping'); // ensure the connection is still alive } @@ -595,5 +468,19 @@ export class GraphNotificationClient { this.renewalSync, nextRenewalTimeInSec * 1000 ); - }; + } + + public async closeSignalRConnection() { + // stop the connection and set it to undefined so it will reconnect when next subscription is created. + await this.connection?.stop(); + this.connection = undefined; + } + + public async subscribeToChatNotifications(chatId: string, sessionId: string) { + log(`Chat subscription with id: ${chatId} and session id: ${sessionId}`); + this.wasConnected = undefined; + this.chatId = chatId; + this.sessionId = sessionId; + this.renewalTimeout = this.timer.setTimeout('renewal:' + this.instanceId, this.renewalSync, 0); + } } From ef3ba50f2639115e273076c099d3ef9dc726d427 Mon Sep 17 00:00:00 2001 From: TechnicallyWilliams Date: Wed, 13 Mar 2024 15:20:54 -0500 Subject: [PATCH 03/21] fix renewal loop --- .../statefulClient/GraphNotificationClient.ts | 25 +++++++++++-------- 1 file changed, 14 insertions(+), 11 deletions(-) diff --git a/packages/mgt-chat/src/statefulClient/GraphNotificationClient.ts b/packages/mgt-chat/src/statefulClient/GraphNotificationClient.ts index c23b1901e0..c8676a9e9d 100644 --- a/packages/mgt-chat/src/statefulClient/GraphNotificationClient.ts +++ b/packages/mgt-chat/src/statefulClient/GraphNotificationClient.ts @@ -223,9 +223,15 @@ export class GraphNotificationClient { return subscription; } - public renewSubscription = async (subscriptionId: string, expirationDateTime: string): Promise => { - // this.renewalCount++; + public renewSubscription = async (subscriptionId: string): Promise => { log(`Renewing Graph subscription for ChatId. RenewalCount: ${this.renewalCount}.`); + + const newExpirationTime = new Date( + new Date().getTime() + appSettings.defaultSubscriptionLifetimeInMinutes * 60 * 1000 + ); + + // this.renewalCount++; + const expirationDateTime = newExpirationTime.toISOString(); const renewedSubscription = (await this.graph?.api(`${GraphConfig.subscriptionEndpoint}/${subscriptionId}`).patch({ expirationDateTime })) as Subscription | undefined; @@ -364,9 +370,9 @@ export class GraphNotificationClient { private readonly renewalSync = () => { void this.renewal(); - } + }; - private readonly renewal = async () =>{ + private readonly renewal = async () => { let nextRenewalTimeInSec = appSettings.renewalTimerInterval; try { // if there are current subscriptions for this chat id... @@ -376,20 +382,17 @@ export class GraphNotificationClient { if (subscriptions) { // attempt a renewal if necessary try { - const expirationTime = new Date( - new Date().getTime() + appSettings.defaultSubscriptionLifetimeInMinutes * 60 * 1000 - ); for (let subscription of subscriptions) { const expirationTime = new Date(subscription.expirationDateTime!); const diff = Math.round((expirationTime.getTime() - new Date().getTime()) / 1000); if (diff <= 0) { log(`Renewing chat subscription ${subscription.id} that has already expired...`); this.trySwitchToDisconnected(true); - await this.renewSubscription(this.chatId, expirationTime.toISOString()); + await this.renewSubscription(subscription.id!); log(`Successfully renewed chat subscription ${subscription.id}.`); } else if (diff <= appSettings.renewalThreshold) { log(`Renewing chat subscription ${subscription.id} that will expire in ${diff} seconds...`); - await this.renewSubscription(this.chatId, expirationTime.toISOString()); + await this.renewSubscription(subscription.id!); log(`Successfully renewed chat subscription ${subscription.id}.`); } } @@ -438,7 +441,7 @@ export class GraphNotificationClient { await this.connection?.send('ping'); // ensure the connection is still alive } if (!this.connection) { - // log(`Creating a new SignalR connection for subscription ${subscription.id!}...`); + log(`Creating a new SignalR connection for subscriptions: ${subscriptionIds}...`); this.trySwitchToDisconnected(true); this.lastNotificationUrl = subscriptions![0]?.notificationUrl!; await this.createSignalRConnection(subscriptions![0]?.notificationUrl!); @@ -468,7 +471,7 @@ export class GraphNotificationClient { this.renewalSync, nextRenewalTimeInSec * 1000 ); - } + }; public async closeSignalRConnection() { // stop the connection and set it to undefined so it will reconnect when next subscription is created. From c762b497769f7e5539e880b3315f3d615fb6a484 Mon Sep 17 00:00:00 2001 From: TechnicallyWilliams Date: Wed, 13 Mar 2024 15:45:38 -0500 Subject: [PATCH 04/21] update cache cleanup in renewal loop --- .../statefulClient/GraphNotificationClient.ts | 20 +++++++++---------- 1 file changed, 9 insertions(+), 11 deletions(-) diff --git a/packages/mgt-chat/src/statefulClient/GraphNotificationClient.ts b/packages/mgt-chat/src/statefulClient/GraphNotificationClient.ts index c8676a9e9d..251f8e514e 100644 --- a/packages/mgt-chat/src/statefulClient/GraphNotificationClient.ts +++ b/packages/mgt-chat/src/statefulClient/GraphNotificationClient.ts @@ -294,6 +294,7 @@ export class GraphNotificationClient { void this.cleanupTimer(); }; + // used to remove inactive chatId subscriptions (different from the renewal timer, which renews active subscriptions) private readonly cleanupTimer = async () => { log(`running cleanup timer`); const offset = Math.min( @@ -336,13 +337,13 @@ export class GraphNotificationClient { return subscriptions; }; - private async deleteCachedSubscription(subscriptionId: string) { + private async deleteCachedSubscriptions(chatId: string) { try { - log(`Removing subscription ${subscriptionId} from cache...`); - await this.subscriptionCache.deleteCachedSubscriptions(subscriptionId); - log(`Successfully removed subscription ${subscriptionId} from cache.`); + log('Removing all chat subscriptions from cache for chatId:', chatId); + await this.subscriptionCache.deleteCachedSubscriptions(chatId); + log('Successfully removed all chat subscriptions from cache.'); } catch (e) { - error(`Failed to remove subscription ${subscriptionId} from cache.`, e); + error(`Failed to remove chat subscription for ${chatId} from cache.`, e); } } @@ -381,6 +382,7 @@ export class GraphNotificationClient { log(`Retrieved chat subscriptions for chat ${this.chatId}: ${subscriptions?.map(s => s?.id).join(', ')}.`); if (subscriptions) { // attempt a renewal if necessary + const existingSubscriptionIds = subscriptions?.map(s => s?.id).filter(Boolean); try { for (let subscription of subscriptions) { const expirationTime = new Date(subscription.expirationDateTime!); @@ -397,12 +399,8 @@ export class GraphNotificationClient { } } } catch (e) { - error(`Failed to renew subscriptions.`, e); - const promises: Promise[] = []; - for (let subscription of subscriptions) { - promises.push(this.deleteCachedSubscription(this.chatId)); - } - await Promise.all(promises); + error(`Failed to renew subscriptions for ${existingSubscriptionIds}.`, e); + await this.deleteCachedSubscriptions(this.chatId); subscriptions = undefined; } } From 27cb6464acd778223501a5d507f60092bfc3ce49 Mon Sep 17 00:00:00 2001 From: TechnicallyWilliams Date: Wed, 13 Mar 2024 15:48:48 -0500 Subject: [PATCH 05/21] update logging --- packages/mgt-chat/src/statefulClient/GraphNotificationClient.ts | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/packages/mgt-chat/src/statefulClient/GraphNotificationClient.ts b/packages/mgt-chat/src/statefulClient/GraphNotificationClient.ts index 251f8e514e..b11e11eb38 100644 --- a/packages/mgt-chat/src/statefulClient/GraphNotificationClient.ts +++ b/packages/mgt-chat/src/statefulClient/GraphNotificationClient.ts @@ -478,7 +478,7 @@ export class GraphNotificationClient { } public async subscribeToChatNotifications(chatId: string, sessionId: string) { - log(`Chat subscription with id: ${chatId} and session id: ${sessionId}`); + log(`Chat subscription with chat id: ${chatId} and session id: ${sessionId}`); this.wasConnected = undefined; this.chatId = chatId; this.sessionId = sessionId; From 57f662db593138798796a27edd8bb0e60592ebc0 Mon Sep 17 00:00:00 2001 From: andrewDoing Date: Wed, 13 Mar 2024 15:51:47 -0500 Subject: [PATCH 06/21] merge --- .../src/statefulClient/GraphNotificationClient.ts | 15 ++++----------- 1 file changed, 4 insertions(+), 11 deletions(-) diff --git a/packages/mgt-chat/src/statefulClient/GraphNotificationClient.ts b/packages/mgt-chat/src/statefulClient/GraphNotificationClient.ts index c8676a9e9d..63e5e68483 100644 --- a/packages/mgt-chat/src/statefulClient/GraphNotificationClient.ts +++ b/packages/mgt-chat/src/statefulClient/GraphNotificationClient.ts @@ -213,7 +213,6 @@ export class GraphNotificationClient { try { subscription = (await subscriptionGraph.api(subscriptionEndpoint).post(subscriptionDefinition)) as Subscription; } catch (error) { - log(`Error creating subscription: ${error}`); throw error; } if (!subscription?.notificationUrl) throw new Error('Subscription not created'); @@ -314,14 +313,12 @@ export class GraphNotificationClient { this.startCleanupTimer(); }; - private createSubscriptions = async (chatId: string, sessionId: string) => { + private createSubscriptions = async (chatId: string) => { const promises: Promise[] = []; promises.push(this.subscribeToResource(`/chats/${chatId}/messages`, ['created', 'updated', 'deleted'])); promises.push(this.subscribeToResource(`/chats/${chatId}/members`, ['created', 'deleted'])); promises.push(this.subscribeToResource(`/chats/${chatId}`, ['updated', 'deleted'])); - const results = await Promise.all(promises).catch((e: Error) => { - this?.emitter.graphNotificationClientError(e); - }); + const results = await Promise.all(promises); // Cache the subscriptions in storage for re-hydration on page refreshes const awaits: Promise[] = []; @@ -329,9 +326,7 @@ export class GraphNotificationClient { for (let subscription of subscriptions) { awaits.push(this.cacheSubscription(subscription)); } - await Promise.all(awaits).catch((e: Error) => { - this?.emitter.graphNotificationClientError(e); - }); + await Promise.all(awaits); return subscriptions; }; @@ -376,9 +371,7 @@ export class GraphNotificationClient { let nextRenewalTimeInSec = appSettings.renewalTimerInterval; try { // if there are current subscriptions for this chat id... - log(`Retrieving chat subscriptions for chat ${this.chatId}...`); let subscriptions = await this.getSubscriptions(this.chatId); - log(`Retrieved chat subscriptions for chat ${this.chatId}: ${subscriptions?.map(s => s?.id).join(', ')}.`); if (subscriptions) { // attempt a renewal if necessary try { @@ -411,7 +404,7 @@ export class GraphNotificationClient { if (!subscriptions) { try { this.trySwitchToDisconnected(true); - subscriptions = await this.createSubscriptions(this.chatId, this.sessionId); + subscriptions = await this.createSubscriptions(this.chatId); } catch (e) { const err = e as { statusCode?: number; message: string }; if (err.statusCode === 403 && err.message.indexOf('has reached its limit') > 0) { From ddeb35b3e9b577352acb0a4f12c79d5419b48366 Mon Sep 17 00:00:00 2001 From: andrewDoing Date: Wed, 13 Mar 2024 15:54:13 -0500 Subject: [PATCH 07/21] fix renewal count --- .../mgt-chat/src/statefulClient/GraphNotificationClient.ts | 6 ++++-- 1 file changed, 4 insertions(+), 2 deletions(-) diff --git a/packages/mgt-chat/src/statefulClient/GraphNotificationClient.ts b/packages/mgt-chat/src/statefulClient/GraphNotificationClient.ts index 63e5e68483..6dc5e617af 100644 --- a/packages/mgt-chat/src/statefulClient/GraphNotificationClient.ts +++ b/packages/mgt-chat/src/statefulClient/GraphNotificationClient.ts @@ -229,12 +229,14 @@ export class GraphNotificationClient { new Date().getTime() + appSettings.defaultSubscriptionLifetimeInMinutes * 60 * 1000 ); - // this.renewalCount++; const expirationDateTime = newExpirationTime.toISOString(); const renewedSubscription = (await this.graph?.api(`${GraphConfig.subscriptionEndpoint}/${subscriptionId}`).patch({ expirationDateTime })) as Subscription | undefined; - if (renewedSubscription) return this.cacheSubscription(renewedSubscription); + if (renewedSubscription) { + this.renewalCount++; + return this.cacheSubscription(renewedSubscription); + } }; public async createSignalRConnection(notificationUrl: string) { From 3c6659e1668c0c863de59a7ce8efcde82756193e Mon Sep 17 00:00:00 2001 From: TechnicallyWilliams Date: Wed, 13 Mar 2024 16:16:46 -0500 Subject: [PATCH 08/21] copy over logic for ignoring messages from other subscriptions --- .../src/statefulClient/GraphNotificationClient.ts | 13 +++++++++++-- 1 file changed, 11 insertions(+), 2 deletions(-) diff --git a/packages/mgt-chat/src/statefulClient/GraphNotificationClient.ts b/packages/mgt-chat/src/statefulClient/GraphNotificationClient.ts index 510382d321..36bb895ee7 100644 --- a/packages/mgt-chat/src/statefulClient/GraphNotificationClient.ts +++ b/packages/mgt-chat/src/statefulClient/GraphNotificationClient.ts @@ -67,6 +67,7 @@ export class GraphNotificationClient { private chatId = ''; private sessionId = ''; private lastNotificationUrl = ''; + private subscriptionIds: string[] = []; private readonly subscriptionCache: SubscriptionsCache = new SubscriptionsCache(); private readonly timer = new Timer(); @@ -119,8 +120,15 @@ export class GraphNotificationClient { private readonly receiveNotificationMessage = (message: string) => { if (typeof message !== 'string') throw new Error('Expected string from receivenotificationmessageasync'); + const ackMessage: unknown = { StatusCode: '200' }; const notification: ReceivedNotification = JSON.parse(message) as ReceivedNotification; - log('received notification message', notification); + // only process notifications for the current subscription + if (this.subscriptionIds.length > 0 && !this.subscriptionIds.includes(notification.subscriptionId)) { + log('Received notification for a different subscription', notification); + return ackMessage; + } + + log('received chat notification message', notification); const emitter: ThreadEventEmitter | undefined = this.emitter; if (!notification.resourceData) throw new Error('Message did not contain resourceData'); if (isMessageNotification(notification)) { @@ -131,7 +139,6 @@ export class GraphNotificationClient { this.processChatPropertiesNotification(notification, emitter); } // Need to return a status code string of 200 so that graph knows the message was received and doesn't re-send the notification - const ackMessage: unknown = { StatusCode: '200' }; return GraphConfig.ackAsString ? JSON.stringify(ackMessage) : ackMessage; }; @@ -328,6 +335,7 @@ export class GraphNotificationClient { const subscriptions: Subscription[] = (results as (Subscription | undefined)[]).filter(Boolean) as Subscription[]; for (let subscription of subscriptions) { awaits.push(this.cacheSubscription(subscription)); + this.subscriptionIds.push(subscription.id!); } await Promise.all(awaits); @@ -339,6 +347,7 @@ export class GraphNotificationClient { log('Removing all chat subscriptions from cache for chatId:', chatId); await this.subscriptionCache.deleteCachedSubscriptions(chatId); log('Successfully removed all chat subscriptions from cache.'); + this.subscriptionIds = []; } catch (e) { error(`Failed to remove chat subscription for ${chatId} from cache.`, e); } From 7d49ac1a0dcf101c2dc7e261dc620746f6312fb7 Mon Sep 17 00:00:00 2001 From: TechnicallyWilliams Date: Wed, 13 Mar 2024 17:03:02 -0500 Subject: [PATCH 09/21] generate groupid once --- .../statefulClient/GraphNotificationClient.ts | 17 +++++++---------- 1 file changed, 7 insertions(+), 10 deletions(-) diff --git a/packages/mgt-chat/src/statefulClient/GraphNotificationClient.ts b/packages/mgt-chat/src/statefulClient/GraphNotificationClient.ts index 36bb895ee7..a2e971e44a 100644 --- a/packages/mgt-chat/src/statefulClient/GraphNotificationClient.ts +++ b/packages/mgt-chat/src/statefulClient/GraphNotificationClient.ts @@ -196,14 +196,14 @@ export class GraphNotificationClient { await this.subscriptionCache.cacheSubscription(this.chatId, subscriptionRecord); }; - private async subscribeToResource(resourcePath: string, changeTypes: ChangeTypes[]) { + private async subscribeToResource(resourcePath: string, groupId: string, changeTypes: ChangeTypes[]) { // build subscription request const expirationDateTime = new Date( new Date().getTime() + appSettings.defaultSubscriptionLifetimeInMinutes * 60 * 1000 ).toISOString(); const subscriptionDefinition: Subscription = { changeType: changeTypes.join(','), - notificationUrl: `${GraphConfig.webSocketsPrefix}?groupId=${getOrGenerateGroupId(this.chatId)}`, + notificationUrl: `${GraphConfig.webSocketsPrefix}?groupId=${groupId}`, resource: resourcePath, expirationDateTime, includeResourceData: true, @@ -217,11 +217,7 @@ export class GraphNotificationClient { if (!subscriptionGraph) return; // send subscription POST to Graph let subscription: Subscription; - try { - subscription = (await subscriptionGraph.api(subscriptionEndpoint).post(subscriptionDefinition)) as Subscription; - } catch (error) { - throw error; - } + subscription = (await subscriptionGraph.api(subscriptionEndpoint).post(subscriptionDefinition)) as Subscription; if (!subscription?.notificationUrl) throw new Error('Subscription not created'); log(subscription); @@ -325,9 +321,10 @@ export class GraphNotificationClient { private createSubscriptions = async (chatId: string) => { const promises: Promise[] = []; - promises.push(this.subscribeToResource(`/chats/${chatId}/messages`, ['created', 'updated', 'deleted'])); - promises.push(this.subscribeToResource(`/chats/${chatId}/members`, ['created', 'deleted'])); - promises.push(this.subscribeToResource(`/chats/${chatId}`, ['updated', 'deleted'])); + const groupId = getOrGenerateGroupId(this.chatId); + promises.push(this.subscribeToResource(`/chats/${chatId}/messages`, groupId, ['created', 'updated', 'deleted'])); + promises.push(this.subscribeToResource(`/chats/${chatId}/members`, groupId, ['created', 'deleted'])); + promises.push(this.subscribeToResource(`/chats/${chatId}`, groupId, ['updated', 'deleted'])); const results = await Promise.all(promises); // Cache the subscriptions in storage for re-hydration on page refreshes From b963a363a786f7317f644309609a077baf4faf1a Mon Sep 17 00:00:00 2001 From: andrewDoing Date: Thu, 14 Mar 2024 10:53:06 -0500 Subject: [PATCH 10/21] fix a lot --- .../statefulClient/GraphNotificationClient.ts | 128 +++++++++++------- .../statefulClient/StatefulGraphChatClient.ts | 13 +- .../src/statefulClient/useGraphChatClient.ts | 15 +- 3 files changed, 83 insertions(+), 73 deletions(-) diff --git a/packages/mgt-chat/src/statefulClient/GraphNotificationClient.ts b/packages/mgt-chat/src/statefulClient/GraphNotificationClient.ts index 510382d321..e3aa22b98d 100644 --- a/packages/mgt-chat/src/statefulClient/GraphNotificationClient.ts +++ b/packages/mgt-chat/src/statefulClient/GraphNotificationClient.ts @@ -30,9 +30,8 @@ import { v4 as uuid } from 'uuid'; export const appSettings = { defaultSubscriptionLifetimeInMinutes: 10, renewalThreshold: 75, // The number of seconds before subscription expires it will be renewed - renewalTimerInterval: 20, // The number of seconds between executions of the renewal timer - removalThreshold: 1 * 60, // number of seconds after the last update of a subscription to consider in inactive - removalTimerInterval: 1 * 60, // the number of seconds between executions of the timer to remove inactive subscriptions + renewalTimerInterval: 3, // The number of seconds between executions of the renewal timer + fastRenewalInterval: 500, // The number of milliseconds between executions of the fast renewal timer useCanary: GraphConfig.useCanary }; @@ -65,7 +64,14 @@ export class GraphNotificationClient { private cleanupTimeout?: string; private renewalCount = 0; private chatId = ''; - private sessionId = ''; + private previousChatId = ''; + private renewalTimerAccumulator = 0; + /** + * Provides a stable sessionId for the lifetime of the browser tab. + * @returns a string that is either read from session storage or generated and placed in session storage + */ + // use session storage to store the session id so that it is stable for the lifetime of the browser tab + private sessionId = sessionStorage.getItem('mgt-chat-session-id') || uuid(); private lastNotificationUrl = ''; private readonly subscriptionCache: SubscriptionsCache = new SubscriptionsCache(); @@ -89,10 +95,45 @@ export class GraphNotificationClient { private readonly emitter: ThreadEventEmitter, private readonly _graph: IGraph | undefined ) { - // start the cleanup timer when we create the notification client. - this.startCleanupTimer(); + // ensure the session id is stored in session storage + if (!sessionStorage.getItem('mgt-chat-session-id')) { + sessionStorage.setItem('mgt-chat-session-id', this.sessionId); + } } + // private startCleanupTimer() { + // this.cleanupTimeout = this.timer.setTimeout( + // 'renewal:' + this.instanceId, + // this.cleanupTimerSync, + // appSettings.removalTimerInterval * 1000 + // ); + // } + + // private readonly cleanupTimerSync = () => { + // void this.cleanupTimer(); + // }; + + // // used to remove inactive chatId subscriptions (different from the renewal timer, which renews active subscriptions) + // private readonly cleanupTimer = async () => { + // log(`running cleanup timer`); + // const offset = Math.min( + // appSettings.removalThreshold * 1000, + // appSettings.defaultSubscriptionLifetimeInMinutes * 60 * 1000 + // ); + // const threshold = new Date(new Date().getTime() - offset).toISOString(); + // const inactiveSubs = await this.subscriptionCache.loadInactiveSubscriptions(threshold); + // let tasks: Promise[] = []; + // for (const inactive of inactiveSubs) { + // tasks.push(this.removeSubscriptions(inactive.subscriptions)); + // } + // await Promise.all(tasks); + // tasks = []; + // for (const inactive of inactiveSubs) { + // tasks.push(this.subscriptionCache.deleteCachedSubscriptions(inactive.chatId)); + // } + // this.startCleanupTimer(); + // }; + /** * Removes any active timers that may exist to prevent memory leaks and perf issues. * Call this method when the component that depends an instance of this class is being removed from the DOM @@ -267,7 +308,9 @@ export class GraphNotificationClient { private async deleteSubscription(id: string) { try { + log(`Deleting subscription with id: ${id}...`) await this.graph?.api(`${GraphConfig.subscriptionEndpoint}/${id}`).delete(); + log(`Successfully deleted subscription with id: ${id}.`) } catch (e) { error(e); } @@ -283,39 +326,6 @@ export class GraphNotificationClient { return Promise.all(tasks); } - private startCleanupTimer() { - this.cleanupTimeout = this.timer.setTimeout( - 'renewal:' + this.instanceId, - this.cleanupTimerSync, - appSettings.removalTimerInterval * 1000 - ); - } - - private readonly cleanupTimerSync = () => { - void this.cleanupTimer(); - }; - - // used to remove inactive chatId subscriptions (different from the renewal timer, which renews active subscriptions) - private readonly cleanupTimer = async () => { - log(`running cleanup timer`); - const offset = Math.min( - appSettings.removalThreshold * 1000, - appSettings.defaultSubscriptionLifetimeInMinutes * 60 * 1000 - ); - const threshold = new Date(new Date().getTime() - offset).toISOString(); - const inactiveSubs = await this.subscriptionCache.loadInactiveSubscriptions(threshold); - let tasks: Promise[] = []; - for (const inactive of inactiveSubs) { - tasks.push(this.removeSubscriptions(inactive.subscriptions)); - } - await Promise.all(tasks); - tasks = []; - for (const inactive of inactiveSubs) { - tasks.push(this.subscriptionCache.deleteCachedSubscriptions(inactive.chatId)); - } - this.startCleanupTimer(); - }; - private createSubscriptions = async (chatId: string) => { const promises: Promise[] = []; promises.push(this.subscribeToResource(`/chats/${chatId}/messages`, ['created', 'updated', 'deleted'])); @@ -373,8 +383,16 @@ export class GraphNotificationClient { private readonly renewal = async () => { let nextRenewalTimeInSec = appSettings.renewalTimerInterval; try { + // this allows us to renew on chatId change much faster than the normal renewal interval + this.renewalTimerAccumulator += appSettings.fastRenewalInterval; + const chatId = this.chatId; + if (this.renewalTimerAccumulator < appSettings.renewalTimerInterval * 1000 && this.chatId === this.previousChatId) { + return; + } + this.previousChatId = chatId; + // if there are current subscriptions for this chat id... - let subscriptions = await this.getSubscriptions(this.chatId); + let subscriptions = await this.getSubscriptions(chatId); if (subscriptions) { // attempt a renewal if necessary const existingSubscriptionIds = subscriptions?.map(s => s?.id).filter(Boolean); @@ -395,7 +413,7 @@ export class GraphNotificationClient { } } catch (e) { error(`Failed to renew subscriptions for ${existingSubscriptionIds}.`, e); - await this.deleteCachedSubscriptions(this.chatId); + await this.deleteCachedSubscriptions(chatId); subscriptions = undefined; } } @@ -404,7 +422,7 @@ export class GraphNotificationClient { if (!subscriptions) { try { this.trySwitchToDisconnected(true); - subscriptions = await this.createSubscriptions(this.chatId); + subscriptions = await this.createSubscriptions(chatId); } catch (e) { const err = e as { statusCode?: number; message: string }; if (err.statusCode === 403 && err.message.indexOf('has reached its limit') > 0) { @@ -416,7 +434,8 @@ export class GraphNotificationClient { } else if (err.statusCode === 403 || err.statusCode === 402) { // permanent error, stop renewal error('Failed to create new subscriptions due to a permanent condition; stopping renewals.', e); - return; // exit without setting the next renewal timer + nextRenewalTimeInSec = -1; + return; // exit and don't reschedule the next renewal } else { // transient error, retry throw new Error( @@ -458,12 +477,15 @@ export class GraphNotificationClient { } catch (e) { error('Error in chat subscription connection process.', e); this.trySwitchToDisconnected(); + } finally { + if (nextRenewalTimeInSec >= 0) { + this.renewalTimeout = this.timer.setTimeout( + 'renewal:' + this.instanceId, + this.renewalSync, + nextRenewalTimeInSec * 1000 + ); + } } - this.renewalTimeout = this.timer.setTimeout( - 'renewal:' + this.instanceId, - this.renewalSync, - nextRenewalTimeInSec * 1000 - ); }; public async closeSignalRConnection() { @@ -472,11 +494,13 @@ export class GraphNotificationClient { this.connection = undefined; } - public async subscribeToChatNotifications(chatId: string, sessionId: string) { - log(`Chat subscription with chat id: ${chatId} and session id: ${sessionId}`); + public async subscribeToChatNotifications(chatId: string) { + log(`Chat subscription with chat id: ${chatId}`); this.wasConnected = undefined; this.chatId = chatId; - this.sessionId = sessionId; - this.renewalTimeout = this.timer.setTimeout('renewal:' + this.instanceId, this.renewalSync, 0); + // start the renewal timer only once + if (!this.renewalTimeout) { + this.renewalTimeout = this.timer.setTimeout('renewal:' + this.instanceId, this.renewalSync, 0); + } } } diff --git a/packages/mgt-chat/src/statefulClient/StatefulGraphChatClient.ts b/packages/mgt-chat/src/statefulClient/StatefulGraphChatClient.ts index dddf3a7bac..d0498639e7 100644 --- a/packages/mgt-chat/src/statefulClient/StatefulGraphChatClient.ts +++ b/packages/mgt-chat/src/statefulClient/StatefulGraphChatClient.ts @@ -405,16 +405,13 @@ class StatefulGraphChatClient extends BaseStatefulClient { return; } this._chatId = value; - if (this._chatId && this._sessionId) { + if (this._chatId) { void this.updateFollowedChat(); } } - private _sessionId: string | undefined; - - public subscribeToChat(chatId: string, sessionId: string) { - if (chatId && sessionId) { - this._sessionId = sessionId; + public subscribeToChat(chatId: string) { + if (chatId) { this.chatId = chatId; } } @@ -434,7 +431,7 @@ class StatefulGraphChatClient extends BaseStatefulClient { */ private async updateFollowedChat() { // avoid subscribing to a resource with an empty chatId - if (this.chatId && this._sessionId) { + if (this.chatId) { // reset state to initial this.notifyStateChange((draft: GraphChatClient) => { draft.status = 'initial'; @@ -456,7 +453,7 @@ class StatefulGraphChatClient extends BaseStatefulClient { // subscribing to notifications will trigger the chatMessageNotificationsSubscribed event // this client will then load the chat and messages when that event listener is called if (this._notificationClient) { - tasks.push(this._notificationClient.subscribeToChatNotifications(this._chatId, this._sessionId)); + tasks.push(this._notificationClient.subscribeToChatNotifications(this._chatId)); await Promise.all(tasks); } } catch (e) { diff --git a/packages/mgt-chat/src/statefulClient/useGraphChatClient.ts b/packages/mgt-chat/src/statefulClient/useGraphChatClient.ts index e7bb49688b..f91c988843 100644 --- a/packages/mgt-chat/src/statefulClient/useGraphChatClient.ts +++ b/packages/mgt-chat/src/statefulClient/useGraphChatClient.ts @@ -10,32 +10,21 @@ import { v4 as uuid } from 'uuid'; import { StatefulGraphChatClient } from './StatefulGraphChatClient'; import { log } from '@microsoft/mgt-element'; -/** - * Provides a stable sessionId for the lifetime of the browser tab. - * @returns a string that is either read from session storage or generated and placed in session storage - */ -const useSessionId = (): string => { - const [sessionId] = useState(() => uuid()); - - return sessionId; -}; - /** * Custom hook to abstract the creation of a stateful graph chat client. * @param {string} chatId the current chatId to be rendered * @returns {StatefulGraphChatClient} a stateful graph chat client that is subscribed to the given chatId */ export const useGraphChatClient = (chatId: string): StatefulGraphChatClient => { - const sessionId = useSessionId(); const [chatClient] = useState(() => new StatefulGraphChatClient()); // when chatId or sessionId changes this effect subscribes or unsubscribes // the component to/from web socket based notifications for the given chatId useEffect(() => { // we must have both a chatId & sessionId to subscribe. - if (chatId && sessionId) chatClient.subscribeToChat(chatId, sessionId); + if (chatId) chatClient.subscribeToChat(chatId); else chatClient.setStatus('no chat id'); - }, [chatId, sessionId, chatClient]); + }, [chatId, chatClient]); // Returns a cleanup function to call tearDown on the chatClient // This allows us to clean up when the consuming component is being unmounted from the DOM From 685cb89eb0c3b62059fd732e5cbbf92eb104eb4a Mon Sep 17 00:00:00 2001 From: andrewDoing Date: Thu, 14 Mar 2024 10:55:19 -0500 Subject: [PATCH 11/21] remove commented code --- .../statefulClient/GraphNotificationClient.ts | 33 ------------------- 1 file changed, 33 deletions(-) diff --git a/packages/mgt-chat/src/statefulClient/GraphNotificationClient.ts b/packages/mgt-chat/src/statefulClient/GraphNotificationClient.ts index c10c0b4264..70b3a7223f 100644 --- a/packages/mgt-chat/src/statefulClient/GraphNotificationClient.ts +++ b/packages/mgt-chat/src/statefulClient/GraphNotificationClient.ts @@ -102,39 +102,6 @@ export class GraphNotificationClient { } } - // private startCleanupTimer() { - // this.cleanupTimeout = this.timer.setTimeout( - // 'renewal:' + this.instanceId, - // this.cleanupTimerSync, - // appSettings.removalTimerInterval * 1000 - // ); - // } - - // private readonly cleanupTimerSync = () => { - // void this.cleanupTimer(); - // }; - - // // used to remove inactive chatId subscriptions (different from the renewal timer, which renews active subscriptions) - // private readonly cleanupTimer = async () => { - // log(`running cleanup timer`); - // const offset = Math.min( - // appSettings.removalThreshold * 1000, - // appSettings.defaultSubscriptionLifetimeInMinutes * 60 * 1000 - // ); - // const threshold = new Date(new Date().getTime() - offset).toISOString(); - // const inactiveSubs = await this.subscriptionCache.loadInactiveSubscriptions(threshold); - // let tasks: Promise[] = []; - // for (const inactive of inactiveSubs) { - // tasks.push(this.removeSubscriptions(inactive.subscriptions)); - // } - // await Promise.all(tasks); - // tasks = []; - // for (const inactive of inactiveSubs) { - // tasks.push(this.subscriptionCache.deleteCachedSubscriptions(inactive.chatId)); - // } - // this.startCleanupTimer(); - // }; - /** * Removes any active timers that may exist to prevent memory leaks and perf issues. * Call this method when the component that depends an instance of this class is being removed from the DOM From b7e420eb07041c0740d7cf6e36ec6eab3cfd80bc Mon Sep 17 00:00:00 2001 From: TechnicallyWilliams Date: Thu, 14 Mar 2024 11:41:13 -0500 Subject: [PATCH 12/21] cache correct subscription --- .../statefulClient/GraphNotificationClient.ts | 32 +++++++++---------- 1 file changed, 16 insertions(+), 16 deletions(-) diff --git a/packages/mgt-chat/src/statefulClient/GraphNotificationClient.ts b/packages/mgt-chat/src/statefulClient/GraphNotificationClient.ts index 70b3a7223f..6c9b0b82b5 100644 --- a/packages/mgt-chat/src/statefulClient/GraphNotificationClient.ts +++ b/packages/mgt-chat/src/statefulClient/GraphNotificationClient.ts @@ -67,10 +67,10 @@ export class GraphNotificationClient { private previousChatId = ''; private renewalTimerAccumulator = 0; /** - * Provides a stable sessionId for the lifetime of the browser tab. - * @returns a string that is either read from session storage or generated and placed in session storage - */ - // use session storage to store the session id so that it is stable for the lifetime of the browser tab + * Provides a stable sessionId for the lifetime of the browser tab. + * @returns a string that is either read from session storage or generated and placed in session storage + */ + // use session storage to store the session id so that it is stable for the lifetime of the browser tab private sessionId = sessionStorage.getItem('mgt-chat-session-id') || uuid(); private lastNotificationUrl = ''; private subscriptionIds: string[] = []; @@ -199,9 +199,9 @@ export class GraphNotificationClient { } } - private readonly cacheSubscription = async (subscriptionRecord: Subscription): Promise => { + private readonly cacheSubscription = async (chatId: string, subscriptionRecord: Subscription): Promise => { log(subscriptionRecord); - await this.subscriptionCache.cacheSubscription(this.chatId, subscriptionRecord); + await this.subscriptionCache.cacheSubscription(chatId, subscriptionRecord); }; private async subscribeToResource(resourcePath: string, groupId: string, changeTypes: ChangeTypes[]) { @@ -233,7 +233,7 @@ export class GraphNotificationClient { return subscription; } - public renewSubscription = async (subscriptionId: string): Promise => { + public renewSubscription = async (chatId: string, subscriptionId: string): Promise => { log(`Renewing Graph subscription for ChatId. RenewalCount: ${this.renewalCount}.`); const newExpirationTime = new Date( @@ -246,7 +246,7 @@ export class GraphNotificationClient { })) as Subscription | undefined; if (renewedSubscription) { this.renewalCount++; - return this.cacheSubscription(renewedSubscription); + return this.cacheSubscription(chatId, renewedSubscription); } }; @@ -278,9 +278,9 @@ export class GraphNotificationClient { private async deleteSubscription(id: string) { try { - log(`Deleting subscription with id: ${id}...`) + log(`Deleting subscription with id: ${id}...`); await this.graph?.api(`${GraphConfig.subscriptionEndpoint}/${id}`).delete(); - log(`Successfully deleted subscription with id: ${id}.`) + log(`Successfully deleted subscription with id: ${id}.`); } catch (e) { error(e); } @@ -298,7 +298,7 @@ export class GraphNotificationClient { private createSubscriptions = async (chatId: string) => { const promises: Promise[] = []; - const groupId = getOrGenerateGroupId(this.chatId); + const groupId = getOrGenerateGroupId(chatId); promises.push(this.subscribeToResource(`/chats/${chatId}/messages`, groupId, ['created', 'updated', 'deleted'])); promises.push(this.subscribeToResource(`/chats/${chatId}/members`, groupId, ['created', 'deleted'])); promises.push(this.subscribeToResource(`/chats/${chatId}`, groupId, ['updated', 'deleted'])); @@ -308,7 +308,7 @@ export class GraphNotificationClient { const awaits: Promise[] = []; const subscriptions: Subscription[] = (results as (Subscription | undefined)[]).filter(Boolean) as Subscription[]; for (let subscription of subscriptions) { - awaits.push(this.cacheSubscription(subscription)); + awaits.push(this.cacheSubscription(chatId, subscription)); this.subscriptionIds.push(subscription.id!); } await Promise.all(awaits); @@ -356,10 +356,10 @@ export class GraphNotificationClient { private readonly renewal = async () => { let nextRenewalTimeInSec = appSettings.renewalTimerInterval; try { + const chatId = this.chatId; // this allows us to renew on chatId change much faster than the normal renewal interval this.renewalTimerAccumulator += appSettings.fastRenewalInterval; - const chatId = this.chatId; - if (this.renewalTimerAccumulator < appSettings.renewalTimerInterval * 1000 && this.chatId === this.previousChatId) { + if (this.renewalTimerAccumulator < appSettings.renewalTimerInterval * 1000 && chatId === this.previousChatId) { return; } this.previousChatId = chatId; @@ -376,11 +376,11 @@ export class GraphNotificationClient { if (diff <= 0) { log(`Renewing chat subscription ${subscription.id} that has already expired...`); this.trySwitchToDisconnected(true); - await this.renewSubscription(subscription.id!); + await this.renewSubscription(chatId, subscription.id!); log(`Successfully renewed chat subscription ${subscription.id}.`); } else if (diff <= appSettings.renewalThreshold) { log(`Renewing chat subscription ${subscription.id} that will expire in ${diff} seconds...`); - await this.renewSubscription(subscription.id!); + await this.renewSubscription(chatId, subscription.id!); log(`Successfully renewed chat subscription ${subscription.id}.`); } } From f559402f0fb2a5fb87aac3bfcc6cad9cbe300425 Mon Sep 17 00:00:00 2001 From: TechnicallyWilliams Date: Thu, 14 Mar 2024 11:55:16 -0500 Subject: [PATCH 13/21] adding back notification client error emitter --- packages/mgt-chat/src/statefulClient/GraphNotificationClient.ts | 1 + 1 file changed, 1 insertion(+) diff --git a/packages/mgt-chat/src/statefulClient/GraphNotificationClient.ts b/packages/mgt-chat/src/statefulClient/GraphNotificationClient.ts index 6c9b0b82b5..e193eb6b81 100644 --- a/packages/mgt-chat/src/statefulClient/GraphNotificationClient.ts +++ b/packages/mgt-chat/src/statefulClient/GraphNotificationClient.ts @@ -450,6 +450,7 @@ export class GraphNotificationClient { } catch (e) { error('Error in chat subscription connection process.', e); this.trySwitchToDisconnected(); + this?.emitter.graphNotificationClientError(e as Error); } finally { if (nextRenewalTimeInSec >= 0) { this.renewalTimeout = this.timer.setTimeout( From c1a6dc7235979c7866718897c3de487267673f2a Mon Sep 17 00:00:00 2001 From: TechnicallyWilliams Date: Thu, 14 Mar 2024 13:02:01 -0500 Subject: [PATCH 14/21] restore shared expirationtime for renewals --- .../statefulClient/GraphNotificationClient.ts | 47 ++++++++++++++----- 1 file changed, 35 insertions(+), 12 deletions(-) diff --git a/packages/mgt-chat/src/statefulClient/GraphNotificationClient.ts b/packages/mgt-chat/src/statefulClient/GraphNotificationClient.ts index e193eb6b81..a60c735e67 100644 --- a/packages/mgt-chat/src/statefulClient/GraphNotificationClient.ts +++ b/packages/mgt-chat/src/statefulClient/GraphNotificationClient.ts @@ -233,20 +233,41 @@ export class GraphNotificationClient { return subscription; } - public renewSubscription = async (chatId: string, subscriptionId: string): Promise => { + public renewSubscriptions = async (chatId: string) => { log(`Renewing Graph subscription for ChatId. RenewalCount: ${this.renewalCount}.`); const newExpirationTime = new Date( new Date().getTime() + appSettings.defaultSubscriptionLifetimeInMinutes * 60 * 1000 - ); - - const expirationDateTime = newExpirationTime.toISOString(); - const renewedSubscription = (await this.graph?.api(`${GraphConfig.subscriptionEndpoint}/${subscriptionId}`).patch({ - expirationDateTime - })) as Subscription | undefined; - if (renewedSubscription) { - this.renewalCount++; - return this.cacheSubscription(chatId, renewedSubscription); + ).toISOString(); + + let subscriptions = await this.getSubscriptions(chatId); + const awaits: Promise[] = []; + for (const subscription of subscriptions || []) { + // the renewSubscription method caches the updated subscription to track the new expiration time + awaits.push(this.renewSubscription(chatId, subscription.id!, newExpirationTime)); + log(`Invoked RenewSubscription ${subscription.id}`); + } + await Promise.all(awaits); + }; + + private renewSubscription = async ( + chatId: string, + subscriptionId: string, + expirationDateTime: string + ): Promise => { + // PATCH /subscriptions/{id} + try { + const renewedSubscription = (await this.graph + ?.api(`${GraphConfig.subscriptionEndpoint}/${subscriptionId}`) + .patch({ + expirationDateTime + })) as Subscription | undefined; + if (renewedSubscription) { + this.renewalCount++; + return this.cacheSubscription(chatId, renewedSubscription); + } + } catch (e) { + return Promise.reject(e); } }; @@ -376,12 +397,14 @@ export class GraphNotificationClient { if (diff <= 0) { log(`Renewing chat subscription ${subscription.id} that has already expired...`); this.trySwitchToDisconnected(true); - await this.renewSubscription(chatId, subscription.id!); + await this.renewSubscriptions(chatId); log(`Successfully renewed chat subscription ${subscription.id}.`); + break; } else if (diff <= appSettings.renewalThreshold) { log(`Renewing chat subscription ${subscription.id} that will expire in ${diff} seconds...`); - await this.renewSubscription(chatId, subscription.id!); + await this.renewSubscriptions(chatId); log(`Successfully renewed chat subscription ${subscription.id}.`); + break; } } } catch (e) { From b348cd55e2585a06783ec638b8f215c48477002b Mon Sep 17 00:00:00 2001 From: TechnicallyWilliams Date: Thu, 14 Mar 2024 18:59:25 -0500 Subject: [PATCH 15/21] synchronize current active chat with it's subscriptionIds --- .../statefulClient/GraphNotificationClient.ts | 27 ++++++++++--------- 1 file changed, 14 insertions(+), 13 deletions(-) diff --git a/packages/mgt-chat/src/statefulClient/GraphNotificationClient.ts b/packages/mgt-chat/src/statefulClient/GraphNotificationClient.ts index a60c735e67..6913e6bb33 100644 --- a/packages/mgt-chat/src/statefulClient/GraphNotificationClient.ts +++ b/packages/mgt-chat/src/statefulClient/GraphNotificationClient.ts @@ -130,7 +130,7 @@ export class GraphNotificationClient { const ackMessage: unknown = { StatusCode: '200' }; const notification: ReceivedNotification = JSON.parse(message) as ReceivedNotification; - // only process notifications for the current subscription + // only process notifications for the current chat's subscriptions if (this.subscriptionIds.length > 0 && !this.subscriptionIds.includes(notification.subscriptionId)) { log('Received notification for a different subscription', notification); return ackMessage; @@ -276,6 +276,9 @@ export class GraphNotificationClient { accessTokenFactory: this.getToken, withCredentials: false }; + // log the notification url and session id + log(`Creating SignalR connection for notification url: ${notificationUrl}`); + log(`Session Id: ${this.sessionId}`); const connection = new HubConnectionBuilder() .withUrl(GraphConfig.adjustNotificationUrl(notificationUrl, this.sessionId), connectionOptions) .withAutomaticReconnect() @@ -330,10 +333,8 @@ export class GraphNotificationClient { const subscriptions: Subscription[] = (results as (Subscription | undefined)[]).filter(Boolean) as Subscription[]; for (let subscription of subscriptions) { awaits.push(this.cacheSubscription(chatId, subscription)); - this.subscriptionIds.push(subscription.id!); } await Promise.all(awaits); - return subscriptions; }; @@ -342,7 +343,6 @@ export class GraphNotificationClient { log('Removing all chat subscriptions from cache for chatId:', chatId); await this.subscriptionCache.deleteCachedSubscriptions(chatId); log('Successfully removed all chat subscriptions from cache.'); - this.subscriptionIds = []; } catch (e) { error(`Failed to remove chat subscription for ${chatId} from cache.`, e); } @@ -383,13 +383,14 @@ export class GraphNotificationClient { if (this.renewalTimerAccumulator < appSettings.renewalTimerInterval * 1000 && chatId === this.previousChatId) { return; } + this.renewalTimerAccumulator = 0; this.previousChatId = chatId; // if there are current subscriptions for this chat id... let subscriptions = await this.getSubscriptions(chatId); if (subscriptions) { // attempt a renewal if necessary - const existingSubscriptionIds = subscriptions?.map(s => s?.id).filter(Boolean); + const subscriptionIds = subscriptions?.map(s => s?.id).filter(Boolean); try { for (let subscription of subscriptions) { const expirationTime = new Date(subscription.expirationDateTime!); @@ -408,7 +409,7 @@ export class GraphNotificationClient { } } } catch (e) { - error(`Failed to renew subscriptions for ${existingSubscriptionIds}.`, e); + error(`Failed to renew subscriptions for ${subscriptionIds}.`, e); await this.deleteCachedSubscriptions(chatId); subscriptions = undefined; } @@ -444,28 +445,28 @@ export class GraphNotificationClient { // create or reconnect the SignalR connection // notificationUrl comes in the form of websockets:https://graph.microsoft.com/beta/subscriptions/notificationChannel/websockets/?groupid=&sessionid=default // if changes, we need to create a new connection - const subscriptionIds = subscriptions?.map(s => s?.id).filter(Boolean); + this.subscriptionIds = subscriptions.map(s => s.id!); if (this.connection?.state === HubConnectionState.Connected) { await this.connection?.send('ping'); // ensure the connection is still alive } if (!this.connection) { - log(`Creating a new SignalR connection for subscriptions: ${subscriptionIds}...`); + log(`Creating a new SignalR connection for subscriptions: ${this.subscriptionIds}...`); this.trySwitchToDisconnected(true); this.lastNotificationUrl = subscriptions![0]?.notificationUrl!; await this.createSignalRConnection(subscriptions![0]?.notificationUrl!); - log(`Successfully created a new SignalR connection for subscriptions: ${subscriptionIds}`); + log(`Successfully created a new SignalR connection for subscriptions: ${this.subscriptionIds}`); } else if (this.connection.state !== HubConnectionState.Connected) { - log(`Reconnecting SignalR connection for subscriptions: ${subscriptionIds}...`); + log(`Reconnecting SignalR connection for subscriptions: ${this.subscriptionIds}...`); this.trySwitchToDisconnected(true); await this.connection.start(); - log(`Successfully reconnected SignalR connection for subscriptions: ${subscriptionIds}`); + log(`Successfully reconnected SignalR connection for subscriptions: ${this.subscriptionIds}`); } else if (this.lastNotificationUrl !== subscriptions![0]?.notificationUrl) { - log(`Updating SignalR connection for subscriptions: ${subscriptionIds} due to new notification URL...`); + log(`Updating SignalR connection for subscriptions: ${this.subscriptionIds} due to new notification URL...`); this.trySwitchToDisconnected(true); await this.closeSignalRConnection(); this.lastNotificationUrl = subscriptions![0]?.notificationUrl!; await this.createSignalRConnection(subscriptions![0]?.notificationUrl!); - log(`Successfully updated SignalR connection for subscriptions: ${subscriptionIds}`); + log(`Successfully updated SignalR connection for subscriptions: ${this.subscriptionIds}`); } // emit the new connection event if necessary From af8a9a7ec71d141b94f52dae9c825fc0b76c8690 Mon Sep 17 00:00:00 2001 From: TechnicallyWilliams Date: Fri, 15 Mar 2024 10:03:42 -0500 Subject: [PATCH 16/21] move renewalcount; only set previousChatId if renewal was successfull --- .../mgt-chat/src/statefulClient/GraphNotificationClient.ts | 7 ++++--- 1 file changed, 4 insertions(+), 3 deletions(-) diff --git a/packages/mgt-chat/src/statefulClient/GraphNotificationClient.ts b/packages/mgt-chat/src/statefulClient/GraphNotificationClient.ts index 6913e6bb33..43581a9811 100644 --- a/packages/mgt-chat/src/statefulClient/GraphNotificationClient.ts +++ b/packages/mgt-chat/src/statefulClient/GraphNotificationClient.ts @@ -246,6 +246,7 @@ export class GraphNotificationClient { // the renewSubscription method caches the updated subscription to track the new expiration time awaits.push(this.renewSubscription(chatId, subscription.id!, newExpirationTime)); log(`Invoked RenewSubscription ${subscription.id}`); + this.renewalCount++; } await Promise.all(awaits); }; @@ -263,7 +264,6 @@ export class GraphNotificationClient { expirationDateTime })) as Subscription | undefined; if (renewedSubscription) { - this.renewalCount++; return this.cacheSubscription(chatId, renewedSubscription); } } catch (e) { @@ -383,8 +383,6 @@ export class GraphNotificationClient { if (this.renewalTimerAccumulator < appSettings.renewalTimerInterval * 1000 && chatId === this.previousChatId) { return; } - this.renewalTimerAccumulator = 0; - this.previousChatId = chatId; // if there are current subscriptions for this chat id... let subscriptions = await this.getSubscriptions(chatId); @@ -471,6 +469,9 @@ export class GraphNotificationClient { // emit the new connection event if necessary this.trySwitchToConnected(); + // set if renewal was successful + this.renewalTimerAccumulator = 0; + this.previousChatId = chatId; } catch (e) { error('Error in chat subscription connection process.', e); this.trySwitchToDisconnected(); From e143fa51c40e09f79f0844af9d233c0836eb8b4d Mon Sep 17 00:00:00 2001 From: TechnicallyWilliams Date: Fri, 15 Mar 2024 10:55:58 -0500 Subject: [PATCH 17/21] synchronize current active chat with it's subscriptionIds --- .../statefulClient/GraphNotificationClient.ts | 19 +++++++++++-------- 1 file changed, 11 insertions(+), 8 deletions(-) diff --git a/packages/mgt-chat/src/statefulClient/GraphNotificationClient.ts b/packages/mgt-chat/src/statefulClient/GraphNotificationClient.ts index 43581a9811..29d9e7c8d6 100644 --- a/packages/mgt-chat/src/statefulClient/GraphNotificationClient.ts +++ b/packages/mgt-chat/src/statefulClient/GraphNotificationClient.ts @@ -226,7 +226,9 @@ export class GraphNotificationClient { // send subscription POST to Graph let subscription: Subscription; subscription = (await subscriptionGraph.api(subscriptionEndpoint).post(subscriptionDefinition)) as Subscription; - if (!subscription?.notificationUrl) throw new Error('Subscription not created'); + if (!subscription?.notificationUrl) { + throw new Error('Subscription not created'); + } log(subscription); log('Invoked CreateSubscription'); @@ -443,28 +445,28 @@ export class GraphNotificationClient { // create or reconnect the SignalR connection // notificationUrl comes in the form of websockets:https://graph.microsoft.com/beta/subscriptions/notificationChannel/websockets/?groupid=&sessionid=default // if changes, we need to create a new connection - this.subscriptionIds = subscriptions.map(s => s.id!); + const subscriptionIds = subscriptions.map(s => s.id!); if (this.connection?.state === HubConnectionState.Connected) { await this.connection?.send('ping'); // ensure the connection is still alive } if (!this.connection) { - log(`Creating a new SignalR connection for subscriptions: ${this.subscriptionIds}...`); + log(`Creating a new SignalR connection for subscriptions: ${subscriptionIds}...`); this.trySwitchToDisconnected(true); this.lastNotificationUrl = subscriptions![0]?.notificationUrl!; await this.createSignalRConnection(subscriptions![0]?.notificationUrl!); - log(`Successfully created a new SignalR connection for subscriptions: ${this.subscriptionIds}`); + log(`Successfully created a new SignalR connection for subscriptions: ${subscriptionIds}`); } else if (this.connection.state !== HubConnectionState.Connected) { - log(`Reconnecting SignalR connection for subscriptions: ${this.subscriptionIds}...`); + log(`Reconnecting SignalR connection for subscriptions: ${subscriptionIds}...`); this.trySwitchToDisconnected(true); await this.connection.start(); - log(`Successfully reconnected SignalR connection for subscriptions: ${this.subscriptionIds}`); + log(`Successfully reconnected SignalR connection for subscriptions: ${subscriptionIds}`); } else if (this.lastNotificationUrl !== subscriptions![0]?.notificationUrl) { - log(`Updating SignalR connection for subscriptions: ${this.subscriptionIds} due to new notification URL...`); + log(`Updating SignalR connection for subscriptions: ${subscriptionIds} due to new notification URL...`); this.trySwitchToDisconnected(true); await this.closeSignalRConnection(); this.lastNotificationUrl = subscriptions![0]?.notificationUrl!; await this.createSignalRConnection(subscriptions![0]?.notificationUrl!); - log(`Successfully updated SignalR connection for subscriptions: ${this.subscriptionIds}`); + log(`Successfully updated SignalR connection for subscriptions: ${subscriptionIds}`); } // emit the new connection event if necessary @@ -472,6 +474,7 @@ export class GraphNotificationClient { // set if renewal was successful this.renewalTimerAccumulator = 0; this.previousChatId = chatId; + this.subscriptionIds = subscriptionIds } catch (e) { error('Error in chat subscription connection process.', e); this.trySwitchToDisconnected(); From bdc196386658d6b9ae5d6c109d95f6d06de8c0a7 Mon Sep 17 00:00:00 2001 From: TechnicallyWilliams Date: Fri, 15 Mar 2024 15:49:59 -0500 Subject: [PATCH 18/21] update log msgs; rm methods from prev implem --- .../statefulClient/GraphNotificationClient.ts | 51 +++++-------------- 1 file changed, 13 insertions(+), 38 deletions(-) diff --git a/packages/mgt-chat/src/statefulClient/GraphNotificationClient.ts b/packages/mgt-chat/src/statefulClient/GraphNotificationClient.ts index 29d9e7c8d6..a6925a1ade 100644 --- a/packages/mgt-chat/src/statefulClient/GraphNotificationClient.ts +++ b/packages/mgt-chat/src/statefulClient/GraphNotificationClient.ts @@ -61,7 +61,6 @@ export class GraphNotificationClient { private connection?: HubConnection = undefined; private wasConnected?: boolean | undefined; private renewalTimeout?: string; - private cleanupTimeout?: string; private renewalCount = 0; private chatId = ''; private previousChatId = ''; @@ -108,7 +107,6 @@ export class GraphNotificationClient { */ public tearDown() { log('cleaning up graph notification resources'); - if (this.cleanupTimeout) this.timer.clearTimeout(this.cleanupTimeout); if (this.renewalTimeout) this.timer.clearTimeout(this.renewalTimeout); this.timer.close(); } @@ -119,10 +117,8 @@ export class GraphNotificationClient { return token; }; - // TODO: understand if this is needed under the native model private readonly onReconnect = (connectionId: string | undefined) => { log(`Reconnected. ConnectionId: ${connectionId || 'undefined'}`); - // void this.renewChatSubscriptions(); }; private readonly receiveNotificationMessage = (message: string) => { @@ -132,7 +128,7 @@ export class GraphNotificationClient { const notification: ReceivedNotification = JSON.parse(message) as ReceivedNotification; // only process notifications for the current chat's subscriptions if (this.subscriptionIds.length > 0 && !this.subscriptionIds.includes(notification.subscriptionId)) { - log('Received notification for a different subscription', notification); + log('Received chat notification message for a different chat subscription', notification); return ackMessage; } @@ -279,8 +275,7 @@ export class GraphNotificationClient { withCredentials: false }; // log the notification url and session id - log(`Creating SignalR connection for notification url: ${notificationUrl}`); - log(`Session Id: ${this.sessionId}`); + log(`Creating SignalR connection for notification url: ${notificationUrl} with session id: ${this.sessionId}`); const connection = new HubConnectionBuilder() .withUrl(GraphConfig.adjustNotificationUrl(notificationUrl, this.sessionId), connectionOptions) .withAutomaticReconnect() @@ -302,26 +297,6 @@ export class GraphNotificationClient { } } - private async deleteSubscription(id: string) { - try { - log(`Deleting subscription with id: ${id}...`); - await this.graph?.api(`${GraphConfig.subscriptionEndpoint}/${id}`).delete(); - log(`Successfully deleted subscription with id: ${id}.`); - } catch (e) { - error(e); - } - } - - private async removeSubscriptions(subscriptions: Subscription[]): Promise { - const tasks: Promise[] = []; - for (const s of subscriptions) { - // if there is no id or the subscription is expired, skip - if (!s.id || (s.expirationDateTime && new Date(s.expirationDateTime) <= new Date())) continue; - tasks.push(this.deleteSubscription(s.id)); - } - return Promise.all(tasks); - } - private createSubscriptions = async (chatId: string) => { const promises: Promise[] = []; const groupId = getOrGenerateGroupId(chatId); @@ -409,7 +384,7 @@ export class GraphNotificationClient { } } } catch (e) { - error(`Failed to renew subscriptions for ${subscriptionIds}.`, e); + error(`Failed to renew chat subscriptions for ${subscriptionIds}.`, e); await this.deleteCachedSubscriptions(chatId); subscriptions = undefined; } @@ -426,17 +401,17 @@ export class GraphNotificationClient { // if the limit is reached, back-off (NOTE: this should probably be a 429) nextRenewalTimeInSec = appSettings.renewalTimerInterval * 3; throw new Error( - `Failed to create new subscriptions due to a limitation; retrying in ${nextRenewalTimeInSec} seconds: ${err.message}.` + `Failed to create new chat subscriptions due to a limitation; retrying in ${nextRenewalTimeInSec} seconds: ${err.message}.` ); } else if (err.statusCode === 403 || err.statusCode === 402) { // permanent error, stop renewal - error('Failed to create new subscriptions due to a permanent condition; stopping renewals.', e); + error('Failed to create new chat subscriptions due to a permanent condition; stopping renewals.', e); nextRenewalTimeInSec = -1; return; // exit and don't reschedule the next renewal } else { // transient error, retry throw new Error( - `Failed to create new subscriptions due to a transient condition; retrying in ${nextRenewalTimeInSec} seconds: ${err.message}.` + `Failed to create new chat subscriptions due to a transient condition; retrying in ${nextRenewalTimeInSec} seconds: ${err.message}.` ); } } @@ -450,23 +425,23 @@ export class GraphNotificationClient { await this.connection?.send('ping'); // ensure the connection is still alive } if (!this.connection) { - log(`Creating a new SignalR connection for subscriptions: ${subscriptionIds}...`); + log(`Creating a new SignalR connection for chat subscriptions: ${subscriptionIds}...`); this.trySwitchToDisconnected(true); this.lastNotificationUrl = subscriptions![0]?.notificationUrl!; await this.createSignalRConnection(subscriptions![0]?.notificationUrl!); - log(`Successfully created a new SignalR connection for subscriptions: ${subscriptionIds}`); + log(`Successfully created a new SignalR connection for chat subscriptions: ${subscriptionIds}`); } else if (this.connection.state !== HubConnectionState.Connected) { - log(`Reconnecting SignalR connection for subscriptions: ${subscriptionIds}...`); + log(`Reconnecting SignalR connection for chat subscriptions: ${subscriptionIds}...`); this.trySwitchToDisconnected(true); await this.connection.start(); - log(`Successfully reconnected SignalR connection for subscriptions: ${subscriptionIds}`); + log(`Successfully reconnected SignalR connection for chat subscriptions: ${subscriptionIds}`); } else if (this.lastNotificationUrl !== subscriptions![0]?.notificationUrl) { - log(`Updating SignalR connection for subscriptions: ${subscriptionIds} due to new notification URL...`); + log(`Updating SignalR connection for chat subscriptions: ${subscriptionIds} due to new notification URL...`); this.trySwitchToDisconnected(true); await this.closeSignalRConnection(); this.lastNotificationUrl = subscriptions![0]?.notificationUrl!; await this.createSignalRConnection(subscriptions![0]?.notificationUrl!); - log(`Successfully updated SignalR connection for subscriptions: ${subscriptionIds}`); + log(`Successfully updated SignalR connection for chat subscriptions: ${subscriptionIds}`); } // emit the new connection event if necessary @@ -497,7 +472,7 @@ export class GraphNotificationClient { } public async subscribeToChatNotifications(chatId: string) { - log(`Chat subscription with chat id: ${chatId}`); + log(`Subscribing to chat notifications for chatId: ${chatId}`); this.wasConnected = undefined; this.chatId = chatId; // start the renewal timer only once From 0f604b0cdb6893120266d54fa9f1e6959aeee7c7 Mon Sep 17 00:00:00 2001 From: andrewDoing Date: Fri, 15 Mar 2024 16:04:57 -0500 Subject: [PATCH 19/21] add doc comments to public methods --- .../mgt-chat/src/statefulClient/GraphNotificationClient.ts | 7 +++++++ 1 file changed, 7 insertions(+) diff --git a/packages/mgt-chat/src/statefulClient/GraphNotificationClient.ts b/packages/mgt-chat/src/statefulClient/GraphNotificationClient.ts index a6925a1ade..007e992589 100644 --- a/packages/mgt-chat/src/statefulClient/GraphNotificationClient.ts +++ b/packages/mgt-chat/src/statefulClient/GraphNotificationClient.ts @@ -465,12 +465,19 @@ export class GraphNotificationClient { } }; + /** + * Closes the SignalR connection + */ public async closeSignalRConnection() { // stop the connection and set it to undefined so it will reconnect when next subscription is created. await this.connection?.stop(); this.connection = undefined; } + /** + * Subscribes to chat notifications for the given chatId + * @param chatId chat id to subscribe to + */ public async subscribeToChatNotifications(chatId: string) { log(`Subscribing to chat notifications for chatId: ${chatId}`); this.wasConnected = undefined; From c32c7694b883c9a7fac91cd1dfebe203a55d89bb Mon Sep 17 00:00:00 2001 From: TechnicallyWilliams Date: Mon, 18 Mar 2024 11:12:15 -0500 Subject: [PATCH 20/21] replace renewal accumulator with last renewal time --- .../src/statefulClient/GraphNotificationClient.ts | 12 +++++++----- 1 file changed, 7 insertions(+), 5 deletions(-) diff --git a/packages/mgt-chat/src/statefulClient/GraphNotificationClient.ts b/packages/mgt-chat/src/statefulClient/GraphNotificationClient.ts index a6925a1ade..2c5e963245 100644 --- a/packages/mgt-chat/src/statefulClient/GraphNotificationClient.ts +++ b/packages/mgt-chat/src/statefulClient/GraphNotificationClient.ts @@ -62,9 +62,9 @@ export class GraphNotificationClient { private wasConnected?: boolean | undefined; private renewalTimeout?: string; private renewalCount = 0; + private lastRenewalTime = new Date(); private chatId = ''; private previousChatId = ''; - private renewalTimerAccumulator = 0; /** * Provides a stable sessionId for the lifetime of the browser tab. * @returns a string that is either read from session storage or generated and placed in session storage @@ -355,9 +355,11 @@ export class GraphNotificationClient { let nextRenewalTimeInSec = appSettings.renewalTimerInterval; try { const chatId = this.chatId; + // this allows us to renew on chatId change much faster than the normal renewal interval - this.renewalTimerAccumulator += appSettings.fastRenewalInterval; - if (this.renewalTimerAccumulator < appSettings.renewalTimerInterval * 1000 && chatId === this.previousChatId) { + const timeElapsed = new Date().getTime() - this.lastRenewalTime.getTime(); + if (timeElapsed < appSettings.renewalTimerInterval * 1000 && chatId === this.previousChatId) { + this.lastRenewalTime = new Date(); return; } @@ -447,9 +449,9 @@ export class GraphNotificationClient { // emit the new connection event if necessary this.trySwitchToConnected(); // set if renewal was successful - this.renewalTimerAccumulator = 0; + this.lastRenewalTime = new Date(); this.previousChatId = chatId; - this.subscriptionIds = subscriptionIds + this.subscriptionIds = subscriptionIds; } catch (e) { error('Error in chat subscription connection process.', e); this.trySwitchToDisconnected(); From f38bf32907efc7e9b635eed4615ca2a07ed3707d Mon Sep 17 00:00:00 2001 From: TechnicallyWilliams Date: Mon, 18 Mar 2024 13:05:48 -0500 Subject: [PATCH 21/21] remove whitespace --- packages/mgt-chat/src/statefulClient/GraphNotificationClient.ts | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/packages/mgt-chat/src/statefulClient/GraphNotificationClient.ts b/packages/mgt-chat/src/statefulClient/GraphNotificationClient.ts index 141a9385ce..d328cc2dd9 100644 --- a/packages/mgt-chat/src/statefulClient/GraphNotificationClient.ts +++ b/packages/mgt-chat/src/statefulClient/GraphNotificationClient.ts @@ -355,7 +355,7 @@ export class GraphNotificationClient { let nextRenewalTimeInSec = appSettings.renewalTimerInterval; try { const chatId = this.chatId; - + // this allows us to renew on chatId change much faster than the normal renewal interval const timeElapsed = new Date().getTime() - this.lastRenewalTime.getTime(); if (timeElapsed < appSettings.renewalTimerInterval * 1000 && chatId === this.previousChatId) {