Add a cleanup mechanism for old db entries

flottform/forms/src/flottform-channel-host.ts

@@ -11,6 +11,9 @@ import {
 
 export class FlottformChannelHost extends EventEmitter<FlottformEventMap> {
 	private flottformApi: string | URL;
+	private baseApi: string;
+	private endpointId: string = '';
+	private hostKey: string = '';
 	private createClientUrl: (params: { endpointId: string }) => Promise<string>;
 	private rtcConfiguration: RTCConfiguration;
 	private pollTimeForIceInMs: number;
@@ -35,6 +38,11 @@ export class FlottformChannelHost extends EventEmitter<FlottformEventMap> {
 	}) {
 		super();
 		this.flottformApi = flottformApi;
+		this.baseApi = (
+			this.flottformApi instanceof URL ? this.flottformApi : new URL(this.flottformApi)
+		)
+			.toString()
+			.replace(/\/$/, '');
 		this.createClientUrl = createClientUrl;
 		this.rtcConfiguration = DEFAULT_WEBRTC_CONFIG;
 		this.pollTimeForIceInMs = pollTimeForIceInMs;
@@ -55,14 +63,9 @@ export class FlottformChannelHost extends EventEmitter<FlottformEventMap> {
 		if (this.openPeerConnection) {
 			this.close();
 		}
-		const baseApi = (
-			this.flottformApi instanceof URL ? this.flottformApi : new URL(this.flottformApi)
-		)
-			.toString()
-			.replace(/\/$/, '');
 
 		try {
-			this.rtcConfiguration.iceServers = await this.fetchIceServers(baseApi);
+			this.rtcConfiguration.iceServers = await this.fetchIceServers(this.baseApi);
 		} catch (error) {
 			// Use the default configuration as a fallback
 			this.logger.error(error);
@@ -74,11 +77,13 @@ export class FlottformChannelHost extends EventEmitter<FlottformEventMap> {
 		const session = await this.openPeerConnection.createOffer();
 		await this.openPeerConnection.setLocalDescription(session);
 
-		const { endpointId, hostKey } = await this.createEndpoint(baseApi, session);
+		const { endpointId, hostKey } = await this.createEndpoint(this.baseApi, session);
+		this.hostKey = hostKey;
+		this.endpointId = endpointId;
 		this.logger.log('Created endpoint', { endpointId, hostKey });
 
-		const getEndpointInfoUrl = `${baseApi}/${endpointId}`;
-		const putHostInfoUrl = `${baseApi}/${endpointId}/host`;
+		const getEndpointInfoUrl = `${this.baseApi}/${endpointId}`;
+		const putHostInfoUrl = `${this.baseApi}/${endpointId}/host`;
 
 		const hostIceCandidates = new Set<RTCIceCandidateInit>();
 		await this.putHostInfo(putHostInfoUrl, hostKey, hostIceCandidates, session);
@@ -103,6 +108,8 @@ export class FlottformChannelHost extends EventEmitter<FlottformEventMap> {
 			this.openPeerConnection = null;
 		}
 		this.changeState('disconnected');
+		// Cleanup old entries.
+		this.deleteEndpoint(this.baseApi, this.endpointId, this.hostKey);
 	};
 
 	private setupDataChannelListener = () => {
@@ -223,6 +230,19 @@ export class FlottformChannelHost extends EventEmitter<FlottformEventMap> {
 		return response.json();
 	};
 
+	private deleteEndpoint = async (baseApi: string, endpointId: string, hostKey: string) => {
+		const response = await fetch(`${baseApi}/${endpointId}`, {
+			method: 'DELETE',
+			headers: {
+				Accept: 'application/json',
+				'Content-Type': 'application/json'
+			},
+			body: JSON.stringify({ hostKey })
+		});
+
+		return response.json();
+	};
+
 	private fetchIceServers = async (baseApi: string) => {
 		const response = await fetch(`${baseApi}/ice-server-credentials`, {
 			method: 'GET',

flottform/server/src/database.spec.ts

@@ -168,9 +168,68 @@ describe('Flottform database', () => {
 						session: answer,
 						iceCandidates: []
 					})
-			).rejects.toThrow(/peerkey/i);
+			).rejects.toThrow(
+				/clientKey is wrong: Another peer is already connected and you cannot change this info without the correct key anymore. If you lost your key, initiate a new Flottform connection./i
+			);
 			const infoAfter = await db.getEndpoint({ endpointId });
 			expect(infoBefore).toStrictEqual(infoAfter);
 		});
 	});
+
+	describe('startCleanup()', () => {
+		function sleep(ms: number) {
+			return new Promise((resolve) => setTimeout(resolve, ms));
+		}
+
+		it('Should clean up stale entries after entryTTL', async () => {
+			const db = await createFlottformDatabase(1000, 500);
+			const conn = new RTCPeerConnection();
+			const offer = await conn.createOffer();
+			const { endpointId } = await db.createEndpoint({ session: offer });
+
+			const connPeer = new RTCPeerConnection();
+			await connPeer.setRemoteDescription(offer);
+			const answer = await connPeer.createAnswer();
+			const clientKey = 'random-key';
+
+			await db.putClientInfo({
+				endpointId,
+				clientKey,
+				session: answer,
+				iceCandidates: [],
+				lastUpdate: Date.now()
+			});
+
+			// Sleep for enough time to trigger the first cleanup
+			await sleep(1100);
+
+			// The endpoint should be cleaned by now
+			expect(async () => await db.getEndpoint({ endpointId })).rejects.toThrow(/endpoint/i);
+		});
+
+		it("Shouldn't clean up entries before entryTTL is expired", async () => {
+			const db = await createFlottformDatabase(1000, 500);
+			const conn = new RTCPeerConnection();
+			const offer = await conn.createOffer();
+			const { endpointId } = await db.createEndpoint({ session: offer });
+
+			const connPeer = new RTCPeerConnection();
+			await connPeer.setRemoteDescription(offer);
+			const answer = await connPeer.createAnswer();
+			const clientKey = 'random-key';
+
+			await db.putClientInfo({
+				endpointId,
+				clientKey,
+				session: answer,
+				iceCandidates: [],
+				lastUpdate: Date.now()
+			});
+
+			// The endpoint shouldn't be cleaned by now
+			const retrievedInfo = await db.getEndpoint({ endpointId });
+			expect(retrievedInfo).toBeDefined();
+			expect(retrievedInfo?.hostInfo.session).toStrictEqual(offer);
+		});
+	});
 });

flottform/server/src/database.ts

@@ -13,6 +13,7 @@ type EndpointInfo = {
 		session: RTCSessionDescriptionInit;
 		iceCandidates: RTCIceCandidateInit[];
 	};
+	lastUpdate: number;
 };
 type SafeEndpointInfo = Omit<EndpointInfo, 'hostKey' | 'clientKey'>;
 
@@ -25,8 +26,46 @@ function createRandomEndpointId(): string {
 
 class FlottformDatabase {
 	private map = new Map<EndpointId, EndpointInfo>();
+	private cleanupTimeoutId: NodeJS.Timeout | null = null;
+	private cleanupPeriod: number;
+	private entryTTL: number; // Time-to-Live for each entry
+
+	constructor(cleanupPeriod = 30 * 60 * 1000, entryTTL = 25 * 60 * 1000) {
+		this.cleanupPeriod = cleanupPeriod;
+		this.entryTTL = entryTTL;
+		this.startCleanup();
+	}
+
+	private startCleanup() {
+		this.cleanupTimeoutId = setTimeout(this.cleanupFn.bind(this), this.cleanupPeriod);
+	}
+
+	private cleanupFn() {
+		if (!this.map || this.map.size === 0) return;
+		const now = Date.now();
+		// Loop over all entries and delete the stale ones.
+		for (const [endpointId, endpointInfo] of this.map) {
+			const lastUpdated = endpointInfo.lastUpdate;
+			if (now - lastUpdated > this.entryTTL) {
+				this.map.delete(endpointId);
+				console.log(`Cleaned up stale entry: ${endpointId}`);
+			}
+		}
+		this.cleanupTimeoutId = setTimeout(this.startCleanup.bind(this), this.cleanupPeriod);
+	}
+
+	private stopCleanup() {
+		// Clear the interval to stop cleanup
+		if (this.cleanupTimeoutId) {
+			clearTimeout(this.cleanupTimeoutId);
+			this.cleanupTimeoutId = null;
+		}
+	}
 
-	constructor() {}
+	// Stop the cleanup when the database is no longer needed
+	destroy() {
+		this.stopCleanup();
+	}
 
 	async createEndpoint({ session }: { session: RTCSessionDescriptionInit }): Promise<EndpointInfo> {
 		const entry = {
@@ -35,7 +74,8 @@ class FlottformDatabase {
 			hostInfo: {
 				session,
 				iceCandidates: []
-			}
+			},
+			lastUpdate: Date.now()
 		};
 		this.map.set(entry.endpointId, entry);
 		return entry;
@@ -46,6 +86,7 @@ class FlottformDatabase {
 		if (!entry) {
 			throw Error('Endpoint not found');
 		}
+		entry.lastUpdate = Date.now();
 		const { hostKey: _ignore1, clientKey: _ignore2, ...endpoint } = entry;
 
 		return endpoint;
@@ -55,12 +96,14 @@ class FlottformDatabase {
 		endpointId,
 		hostKey,
 		session,
-		iceCandidates
+		iceCandidates,
+		lastUpdate = Date.now()
 	}: {
 		endpointId: EndpointId;
 		hostKey: HostKey;
 		session: RTCSessionDescriptionInit;
 		iceCandidates: RTCIceCandidateInit[];
+		lastUpdate?: number;
 	}): Promise<SafeEndpointInfo> {
 		const existingSession = this.map.get(endpointId);
 		if (!existingSession) {
@@ -72,7 +115,8 @@ class FlottformDatabase {
 
 		const newInfo = {
 			...existingSession,
-			hostInfo: { ...existingSession.hostInfo, session, iceCandidates }
+			hostInfo: { ...existingSession.hostInfo, session, iceCandidates },
+			lastUpdate
 		};
 		this.map.set(endpointId, newInfo);
 
@@ -85,12 +129,14 @@ class FlottformDatabase {
 		endpointId,
 		clientKey,
 		session,
-		iceCandidates
+		iceCandidates,
+		lastUpdate = Date.now()
 	}: {
 		endpointId: EndpointId;
 		clientKey: ClientKey;
 		session: RTCSessionDescriptionInit;
 		iceCandidates: RTCIceCandidateInit[];
+		lastUpdate?: number;
 	}): Promise<Required<SafeEndpointInfo>> {
 		const existingSession = this.map.get(endpointId);
 		if (!existingSession) {
@@ -105,7 +151,8 @@ class FlottformDatabase {
 		const newInfo = {
 			...existingSession,
 			clientKey,
-			clientInfo: { session, iceCandidates }
+			clientInfo: { session, iceCandidates },
+			lastUpdate
 		};
 		this.map.set(endpointId, newInfo);
 
@@ -130,8 +177,11 @@ class FlottformDatabase {
 	}
 }
 
-export async function createFlottformDatabase(): Promise<FlottformDatabase> {
-	return new FlottformDatabase();
+export async function createFlottformDatabase(
+	cleanupPeriod = 30 * 60 * 1000,
+	entryTTL = 25 * 60 * 1000
+): Promise<FlottformDatabase> {
+	return new FlottformDatabase(cleanupPeriod, entryTTL);
 }
 
 export type { FlottformDatabase };