Use a heartbeat to allow the client to reconnect after network failures.

3 files changed, 87 insertions, 8 deletions

diff --git a/ui/lib/session.svelte.js b/ui/lib/session.svelte.js
index 67155ab..b953d9c 100644
--- a/ui/lib/session.svelte.js
+++ b/ui/lib/session.svelte.js
@@ -1,8 +1,11 @@
 import { redirect } from '@sveltejs/kit';
 
+import { goto } from '$app/navigation';
+
 import * as api from './apiServer.js';
 import * as r from './state/remote/state.svelte.js';
 import * as l from './state/local/channels.svelte.js';
+import { Watchdog } from './watchdog.js';
 
 class Session {
   remote = $state();
@@ -16,19 +19,32 @@ class Session {
     )
   );
 
-  static boot({ user, users, channels, messages, resume_point }) {
+  static boot({ user, users, channels, messages, resume_point, heartbeat }) {
     const remote = r.State.boot({
       currentUser: user,
       users,
       channels,
       messages,
-      resumePoint: resume_point
+      resumePoint: resume_point,
+      heartbeat
     });
     const local = l.Channels.fromLocalStorage();
     return new Session(remote, local);
   }
 
+  reboot({ user, users, channels, messages, resume_point, heartbeat }) {
+    this.remote = r.State.boot({
+      currentUser: user,
+      users,
+      channels,
+      messages,
+      resumePoint: resume_point,
+      heartbeat
+    });
+  }
+
   constructor(remote, local) {
+    this.watchdog = new Watchdog(this.watchdogExpired.bind(this));
     this.remote = remote;
     this.local = local;
   }
@@ -36,30 +52,62 @@ class Session {
   begin() {
     this.events = api.subscribeToEvents(this.remote.resumePoint);
     this.events.onmessage = this.onMessage.bind(this);
+    this.watchdog.reset(this.heartbeatMillis());
   }
 
   end() {
+    this.watchdog.stop();
     this.events.close();
     this.events = null;
   }
 
+  active() {
+    return this.events !== null;
+  }
+
   onMessage(message) {
     const event = JSON.parse(message.data);
     this.remote.onEvent(event);
     this.local.retainChannels(this.remote.channels.all.keys());
+    this.watchdog.reset(this.heartbeatMillis());
+  }
+
+  heartbeatMillis() {
+    return this.remote.heartbeat /* in seconds */ * 1000 /* millis */;
+  }
+
+  async watchdogExpired() {
+    // We leave `this.events` set here as a marker that the interruption is temporary. That's then
+    // used below, after a potential delay, to decide whether to start the stream back up again or
+    // not.
+    this.events.close();
+    this.watchdog.stop();
+
+    const response = await bootOrNavigate(goto);
+    // Session abandoned; give up here. We need to do this after each await, because that's time in
+    // which the session may have been abandoned.
+    if (!this.active()) return;
+
+    this.reboot(response);
+    this.begin();
   }
 }
 
-export async function boot() {
+async function bootOrNavigate(navigateTo) {
   const response = await api.boot();
   switch (response.status) {
     case 401:
-      redirect(307, '/login');
+      await navigateTo('/login');
       break;
     case 503:
-      redirect(307, '/setup');
+      await navigateTo('/setup');
       break;
     case 200:
-      return Session.boot(response.data);
+      return response.data;
   }
 }
+
+export async function boot() {
+  const response = await bootOrNavigate(async (url) => redirect(307, url));
+  return Session.boot(response);
+}
diff --git a/ui/lib/state/remote/state.svelte.js b/ui/lib/state/remote/state.svelte.js
index 6cbe124..29831a0 100644
--- a/ui/lib/state/remote/state.svelte.js
+++ b/ui/lib/state/remote/state.svelte.js
@@ -8,9 +8,10 @@ export class State {
   channels = $state();
   messages = $state();
 
-  static boot({ currentUser, users, channels, messages, resumePoint }) {
+  static boot({ currentUser, heartbeat, users, channels, messages, resumePoint }) {
     return new State({
       currentUser,
+      heartbeat,
       users: Users.boot(users),
       channels: Channels.boot(channels),
       messages: Messages.boot(messages),
@@ -18,8 +19,9 @@ export class State {
     });
   }
 
-  constructor({ currentUser, users, channels, messages, resumePoint }) {
+  constructor({ currentUser, heartbeat, users, channels, messages, resumePoint }) {
     this.currentUser = currentUser;
+    this.heartbeat = heartbeat;
     this.users = users;
     this.channels = channels;
     this.messages = messages;
@@ -27,6 +29,8 @@ export class State {
   }
 
   onEvent(event) {
+    // Heartbeats are actually completely ignored here. They're handled in `Session`, but not as a
+    // special case; _any_ event is a heartbeat event.
     switch (event.type) {
       case 'channel':
         return this.onChannelEvent(event);
diff --git a/ui/lib/watchdog.js b/ui/lib/watchdog.js
new file mode 100644
index 0000000..c95fd4d
--- /dev/null
+++ b/ui/lib/watchdog.js
@@ -0,0 +1,27 @@
+export class Watchdog {
+  constructor(onExpired) {
+    this.timeout = null;
+    this.onExpired = onExpired;
+  }
+
+  reset(delay) {
+    if (this.timeout !== null) {
+      clearTimeout(this.timeout);
+    }
+    this.timeout = setTimeout(this.expire.bind(this), delay);
+  }
+
+  stop() {
+    if (this.timeout !== null) {
+      clearTimeout(this.timeout);
+      this.timeout = null;
+    }
+  }
+
+  expire() {
+    if (this.timeout !== null) {
+      this.timeout = null;
+    }
+    this.onExpired();
+  }
+}