Rehydrate shortly upcoming talks from PentaDb to ensure short-notice changes are picked up (#168)

Author: reivilibre

src/Scheduler.ts

@@ -191,6 +191,16 @@ export class Scheduler {
             LogService.info("Scheduler", "Scheduling tasks");
             try {
                 const minVar = config.conference.lookaheadMinutes;
+                try {
+                    // Refresh upcoming parts of our schedule to ensure it's really up to date.
+                    // Rationale: Sometimes schedules get changed at short notice, so we try our best to accommodate that.
+                    // Rationale for adding 1 minute: so we don't cut it too close to the wire; whilst processing the refresh,
+                    //     time may slip forward.
+                    await this.conference.backend.refreshShortTerm((minVar + 1) * 60);
+                } catch (e) {
+                    LogService.error("Scheduler", `Failed short-term schedule refresh: ${e.message ?? e}\n${e.stack ?? '?'}`);
+                }
+
                 const upcomingTalks = await this.conference.getUpcomingTalkStarts(minVar, minVar);
                 const upcomingQA = await this.conference.getUpcomingQAStarts(minVar, minVar);
                 const upcomingEnds = await this.conference.getUpcomingTalkEnds(minVar, minVar);

src/backends/CachingBackend.ts

@@ -42,6 +42,12 @@ export class CachingBackend implements IScheduleBackend {
 
     private wasCached: boolean = true;
 
+    /**
+     * We hang on to the last backend for refreshing short-notice alterations.
+     * (I don't particularly like it, but there isn't time to reason through a better approach at the moment.)
+     */
+    private lastUsedBackend: IScheduleBackend | null = null;
+
     /**
      * @param underlyingBackend A factory for the underlying backend, which will be reconstructed each time we try to refresh.
      */
@@ -76,6 +82,7 @@ export class CachingBackend implements IScheduleBackend {
 
     async refresh(): Promise<void> {
         const backend = await this.underlyingBackend();
+        this.lastUsedBackend = backend;
 
         this.conference = backend.conference;
         this.talks = backend.talks;
@@ -91,6 +98,15 @@ export class CachingBackend implements IScheduleBackend {
         }
     }
 
+    async refreshShortTerm(lookaheadSeconds: number): Promise<void> {
+        if (this.lastUsedBackend !== null) {
+            // It's notable that we don't save any changes to disk.
+            // It wouldn't be a bad idea to persist the changes, but introducing a lot of disk I/O into this frequent operation
+            // made me uncomfortable.
+            await this.lastUsedBackend.refreshShortTerm(lookaheadSeconds);
+        }
+    }
+
     private async saveCacheToDisk(): Promise<void> {
         // Save a cached copy.
         // Do it atomically so that there's very little chance of anything going wrong: write to a file first, then move into place.

src/backends/IScheduleBackend.ts

@@ -19,6 +19,15 @@ export interface IScheduleBackend {
      */
     refresh(): Promise<void>;
 
+    /**
+     * Calling this function requests the schedule backend to look ahead `lookaheadSeconds` seconds into the future
+     * and try its best to ensure whatever talks in that window of time are up-to-date in the backend's view of the schedule.
+     *
+     * This is an ugly hack to support short-notice changes to the conference schedule, as happens in real life.
+     * It is principally expected to be called by the Scheduler when scheduling tasks in the short-term future.
+     */
+    refreshShortTerm(lookaheadSeconds: number): Promise<void>;
+
     /**
      * Returns true iff the current schedule was loaded from cache, rather than from the intended source.
      * This happens if there was a problem loading the schedule from the intended source for some reason.

src/backends/json/JsonScheduleBackend.ts

@@ -71,6 +71,12 @@ export class JsonScheduleBackend implements IScheduleBackend {
         this.wasFromCache = false;
     }
 
+    async refreshShortTerm(_lookaheadSeconds: number): Promise<void> {
+        // NOP: There's no way to partially refresh a JSON schedule.
+        // Short-term changes to a JSON schedule are therefore currently unimplemented.
+        // This hack was intended for Penta anyway.
+    }
+
     get conference(): IConference {
         return this.loader.conference;
     };

src/backends/penta/PentaBackend.ts

@@ -5,6 +5,7 @@ import { PentaDb } from "./db/PentaDb";
 import { PentabarfParser } from "./PentabarfParser";
 import * as fetch from "node-fetch";
 import { LogService } from "matrix-bot-sdk";
+import { IDbTalk } from "./db/DbTalk";
 
 
 export class PentaBackend implements IScheduleBackend {
@@ -69,12 +70,23 @@ export class PentaBackend implements IScheduleBackend {
     private async hydrateTalk(talk: ITalk): Promise<void> {
         const dbTalk = await this.db.getTalk(talk.id);
         if (dbTalk === null) return;
+        this.rehydrateTalkFrom(talk, dbTalk);
+    }
 
+    private rehydrateTalkFrom(talk: ITalk, dbTalk: IDbTalk): void {
         if (talk.qa_startTime !== null) {
             // hydrate Q&A time if enabled
+            // Rationale for hydrating Q&A time: it's not available in the Pentabarf XML.
             talk.qa_startTime = dbTalk.qa_start_datetime;
         }
+
         talk.livestream_endTime = dbTalk.livestream_end_datetime;
+
+        // Rationale for hydrating talk start & end time: there can be short-notice alterations to the schedule
+        //     (and rehydrating talks is how `refreshShortTerm` is implemented)
+        //     and during testing, the PentaDB can have a time shift set which changes the time of talks compared to the XML.
+        talk.startTime = dbTalk.start_datetime;
+        talk.endTime = dbTalk.end_datetime;
     }
 
     private async hydratePerson(person: IPerson): Promise<void> {
@@ -114,6 +126,23 @@ export class PentaBackend implements IScheduleBackend {
         throw new Error("refresh() not implemented for Penta backend.");
     }
 
+    /**
+     * See description on `IScheduleBackend`.
+     *
+     * For the penta backend, we consult the database for short-notice alterations and rehydrate any affected talks.
+     */
+    async refreshShortTerm(lookaheadSeconds: number): Promise<void> {
+        const talksOfInterest = await this.db.getTalksWithUpcomingEvents(lookaheadSeconds / 60);
+        for (const dbTalk of talksOfInterest) {
+            const talk = this.talks.get(dbTalk.event_id);
+            if (talk === undefined) {
+                LogService.warn("PentaBackend", `refreshShortTerm: DB talk '${dbTalk.event_id}' is upcoming but has no talk entry to hydrate.`);
+                continue;
+            }
+            this.rehydrateTalkFrom(talk, dbTalk);
+        }
+    }
+
     conference: IConference;
     talks: Map<TalkId, ITalk>;
     auditoriums: Map<AuditoriumId, IAuditorium>;

src/backends/penta/db/PentaDb.ts

@@ -115,6 +115,32 @@ export class PentaDb {
         return this.getTalksWithin(END_QUERY, inNextMinutes, minBefore);
     }
 
+    /**
+     * Returns a list of database entries for all talks who have some kind of event in the next `lookaheadSeconds` seconds.
+     *
+     * A suggested implementation might be to return all talks overlapping this window, but for now the simple implementation
+     * is just to return talks with a start, end or Q&A start within the time window.
+     *
+     * @param lookaheadMinutes Number of minutes to look ahead into the future.
+     */
+    public async getTalksWithUpcomingEvents(lookaheadMinutes: number): Promise<IDbTalk[]> {
+        const talksStarting = await this.getUpcomingTalkStarts(lookaheadMinutes, lookaheadMinutes);
+        const talksEnding = await this.getUpcomingTalkEnds(lookaheadMinutes, lookaheadMinutes);
+        const talksQaStarting = await this.getUpcomingQAStarts(lookaheadMinutes, lookaheadMinutes);
+        const result: IDbTalk[] = [];
+        const seenTalkIds = new Set();
+
+        for (const talk of talksStarting.concat(talksEnding, talksQaStarting)) {
+            if (seenTalkIds.has(talk.event_id)) {
+                continue;
+            }
+            seenTalkIds.add(talk.event_id);
+            result.push(talk);
+        }
+
+        return result;
+    }
+
     /**
      * Gets the record for a talk.
      * @param talkId The talk ID.
@@ -136,7 +162,6 @@ export class PentaDb {
     }
 
     private postprocessDbTalk(talk: IRawDbTalk): IDbTalk {
-
         const qaStartDatetime = talk.qa_start_datetime + this.config.schedulePreBufferSeconds * 1000;
         let livestreamStartDatetime: number;
         if (talk.prerecorded) {
@@ -159,7 +184,7 @@ export class PentaDb {
     }
 
     private postprocessDbTalks(rows: IRawDbTalk[]): IDbTalk[] {
-        return rows.map(this.postprocessDbTalk);
+        return rows.map(this.postprocessDbTalk.bind(this));
     }
 
     private sanitizeRecords(rows: IDbPerson[]): IDbPerson[] {