20823830584

Committed 08 Jan 2026 04:24PM UTC coverage: 86.069%. Remained the same

Build # 20823830584

Build Type

push

github

Committed by

goto-bus-stop

Commit Message

Fix AbortSignal handling after Redlock removal

Coverage Stats

1006 of 1198 branches covered (83.97%)

Branch coverage included in aggregate %.

1 of 2 new or added lines in 1 file covered. (50.0%)

10547 of 12225 relevant lines covered (86.27%)

100.89 hits per line

Source File
Press 'n' to go to next uncovered line, 'b' for previous

81.49

/src/plugins/booth.js

import Mutex from 'p-mutex';
import { sql } from 'kysely';
import { EmptyPlaylistError, PlaylistItemNotFoundError } from '../errors/index.js';
import routes from '../routes/booth.js';
import { randomUUID } from 'node:crypto';
import { fromJson, jsonb, jsonGroupArray } from '../utils/sqlite.js';

/**
 * @typedef {import('../schema.js').UserID} UserID
 * @typedef {import('../schema.js').HistoryEntryID} HistoryEntryID
 * @typedef {import('type-fest').JsonObject} JsonObject
 * @typedef {import('../schema.js').User} User
 * @typedef {import('../schema.js').Playlist} Playlist
 * @typedef {import('../schema.js').PlaylistItem} PlaylistItem
 * @typedef {import('../schema.js').HistoryEntry} HistoryEntry
 * @typedef {Omit<import('../schema.js').Media, 'createdAt' | 'updatedAt'>} Media
 */

const KEY_HISTORY_ID = 'booth:historyID';
const KEY_CURRENT_DJ_ID = 'booth:currentDJ';
const KEY_REMOVE_AFTER_CURRENT_PLAY = 'booth:removeAfterCurrentPlay';

class Booth {
  #uw;

  #logger;

  /** @type {ReturnType<typeof setTimeout>|null} */
  #timeout = null;

  #mutex;

  /** @type {Promise<unknown>|null} */
  #awaitAdvance = null;

  /**
   * @param {import('../Uwave.js').Boot} uw
   */
  constructor(uw) {
    this.#uw = uw;
    this.#mutex = new Mutex();
    this.#logger = uw.logger.child({ ns: 'uwave:booth' });
  }

  /** @internal */
  async onStart() {
    const current = await this.getCurrentEntry();
    if (current && this.#timeout === null) {
      // Restart the advance timer after a server restart, if a track was
      // playing before the server restarted.
      const duration = (current.historyEntry.end - current.historyEntry.start) * 1000;
      const endTime = current.historyEntry.createdAt.getTime() + duration;
      if (endTime > Date.now()) {
        this.#timeout = setTimeout(
          () => this.#advanceAutomatically(),
          endTime - Date.now(),
        );
      } else {
        this.#advanceAutomatically();
      }
    }

    this.#uw.onClose(async () => {
      this.#onStop();
      await this.#awaitAdvance;
    });
  }

  async #advanceAutomatically() {
    try {
      await this.advance();
    } catch (error) {
      this.#logger.error({ err: error }, 'advance failed');
    }
  }

  #onStop() {
    this.#maybeStop();
  }

  async getCurrentEntry(tx = this.#uw.db) {
    const entry = await tx.selectFrom('keyval')
      .where('key', '=', KEY_HISTORY_ID)
      .innerJoin('historyEntries', (join) => join.on(
        (eb) => sql`${eb.ref('value')}->>'$'`,
        '=',
        (eb) => eb.ref('historyEntries.id'),
      ))
      .innerJoin('media', 'historyEntries.mediaID', 'media.id')
      .innerJoin('users', 'historyEntries.userID', 'users.id')
      .select([
        'historyEntries.id as id',
        'media.id as media.id',
        'media.sourceID as media.sourceID',
        'media.sourceType as media.sourceType',
        'media.sourceData as media.sourceData',
        'media.artist as media.artist',
        'media.title as media.title',
        'media.duration as media.duration',
        'media.thumbnail as media.thumbnail',
        'users.id as users.id',
        'users.username as users.username',
        'users.avatar as users.avatar',
        'users.createdAt as users.createdAt',
        'historyEntries.artist',
        'historyEntries.title',
        'historyEntries.start',
        'historyEntries.end',
        'historyEntries.createdAt',
        (eb) => eb.selectFrom('feedback')
          .where('historyEntryID', '=', eb.ref('historyEntries.id'))
          .where('vote', '=', 1)
          .select((eb) => jsonGroupArray(eb.ref('userID')).as('userIDs'))
          .as('upvotes'),
        (eb) => eb.selectFrom('feedback')
          .where('historyEntryID', '=', eb.ref('historyEntries.id'))
          .where('vote', '=', -1)
          .select((eb) => jsonGroupArray(eb.ref('userID')).as('userIDs'))
          .as('downvotes'),
        (eb) => eb.selectFrom('feedback')
          .where('historyEntryID', '=', eb.ref('historyEntries.id'))
          .where('favorite', '=', 1)
          .select((eb) => jsonGroupArray(eb.ref('userID')).as('userIDs'))
          .as('favorites'),
      ])
      .executeTakeFirst();

    return entry ? {
      media: {
        id: entry['media.id'],
        artist: entry['media.artist'],
        title: entry['media.title'],
        duration: entry['media.duration'],
        thumbnail: entry['media.thumbnail'],
        sourceID: entry['media.sourceID'],
        sourceType: entry['media.sourceType'],
        sourceData: entry['media.sourceData'] ?? {},
      },
      user: {
        id: entry['users.id'],
        username: entry['users.username'],
        avatar: entry['users.avatar'],
        createdAt: entry['users.createdAt'],
      },
      historyEntry: {
        id: entry.id,
        userID: entry['users.id'],
        mediaID: entry['media.id'],
        artist: entry.artist,
        title: entry.title,
        start: entry.start,
        end: entry.end,
        createdAt: entry.createdAt,
      },
      upvotes: entry.upvotes != null ? fromJson(entry.upvotes) : [],
      downvotes: entry.downvotes != null ? fromJson(entry.downvotes) : [],
      favorites: entry.favorites != null ? fromJson(entry.favorites) : [],
    } : null;
  }

  /**
   * @param {{ remove?: boolean }} options
   */
  async #getNextDJ(options, tx = this.#uw.db) {
    const waitlist = await this.#uw.waitlist.getUserIDs();
    let userID = waitlist.at(0) ?? null;
    if (!userID && !options.remove) {
      // If the waitlist is empty, the current DJ will play again immediately.
      userID = /** @type {UserID|null} */ (await this.#uw.keyv.get(KEY_CURRENT_DJ_ID, tx));
    }
    if (!userID) {
      return null;
    }

    return this.#uw.users.getUser(userID, tx);
  }

  /**
   * @param {{ remove?: boolean }} options
   */
  async #getNextEntry(options) {
    const { playlists } = this.#uw;

    const user = await this.#getNextDJ(options);
    if (!user || !user.activePlaylistID) {
      return null;
    }
    const playlist = await playlists.getUserPlaylist(user, user.activePlaylistID);
    if (playlist.size === 0) {
      throw new EmptyPlaylistError();
    }

    const { playlistItem, media } = await playlists.getPlaylistItemAt(playlist, 0);
    if (!playlistItem) {
      throw new PlaylistItemNotFoundError();
    }

    return {
      user,
      playlist,
      playlistItem,
      media,
      historyEntry: {
        id: /** @type {HistoryEntryID} */ (randomUUID()),
        userID: user.id,
        mediaID: media.id,
        artist: playlistItem.artist,
        title: playlistItem.title,
        start: playlistItem.start,
        end: playlistItem.end,
        /** @type {null | JsonObject} */
        sourceData: null,
        createdAt: new Date(),
      },
    };
  }

  /**
   * @param {UserID|null} previous
   * @param {{ remove?: boolean }} options
   */
  async #cycleWaitlist(previous, options) {
    await this.#uw.waitlist.cycle(previous, options);
  }

  async clear(tx = this.#uw.db) {
    await this.#uw.keyv.delete(KEY_REMOVE_AFTER_CURRENT_PLAY, tx);
    await this.#uw.keyv.delete(KEY_HISTORY_ID, tx);
    await this.#uw.keyv.delete(KEY_CURRENT_DJ_ID, tx);
  }

  /**
   * @param {{ historyEntry: { id: HistoryEntryID }, user: { id: UserID } }} next
   */
  async #update(next, tx = this.#uw.db) {
    await this.#uw.keyv.delete(KEY_REMOVE_AFTER_CURRENT_PLAY, tx);
    await this.#uw.keyv.set(KEY_HISTORY_ID, next.historyEntry.id, tx);
    await this.#uw.keyv.set(KEY_CURRENT_DJ_ID, next.user.id, tx);
  }

  #maybeStop() {
    if (this.#timeout) {
      clearTimeout(this.#timeout);
      this.#timeout = null;
    }
  }

  /**
   * @param {Pick<HistoryEntry, 'start' | 'end'>} entry
   */
  #play(entry) {
    this.#maybeStop();
    this.#timeout = setTimeout(
      () => this.#advanceAutomatically(),
      (entry.end - entry.start) * 1000,
    );
  }

  /**
   * This method creates a `media` object that clients can understand from a
   * history entry object.
   *
   * We present the playback-specific `sourceData` as if it is
   * a property of the media model for backwards compatibility.
   * Old clients don't expect `sourceData` directly on a history entry object.
   *
   * @param {{ user: User, media: Media, historyEntry: HistoryEntry }} next
   */
  getMediaForPlayback(next) {
    return {
      artist: next.historyEntry.artist,
      title: next.historyEntry.title,
      start: next.historyEntry.start,
      end: next.historyEntry.end,
      media: {
        sourceType: next.media.sourceType,
        sourceID: next.media.sourceID,
        artist: next.media.artist,
        title: next.media.title,
        duration: next.media.duration,
        sourceData: {
          ...next.media.sourceData,
          ...next.historyEntry.sourceData,
        },
      },
    };
  }

  /**
   * @param {{
   *   user: User,
   *   playlist: Playlist,
   *   media: Media,
   *   historyEntry: HistoryEntry
   * } | null} next
   */
  async #publishAdvanceComplete(next) {
    const { waitlist } = this.#uw;

    if (next != null) {
      this.#uw.publish('advance:complete', {
        historyID: next.historyEntry.id,
        userID: next.user.id,
        playlistID: next.playlist.id,
        media: this.getMediaForPlayback(next),
        playedAt: next.historyEntry.createdAt.getTime(),
      });
      this.#uw.publish('playlist:cycle', {
        userID: next.user.id,
        playlistID: next.playlist.id,
      });
    } else {
      this.#uw.publish('advance:complete', null);
    }
    this.#uw.publish('waitlist:update', await waitlist.getUserIDs());
  }

  /**
   * @param {{ user: User, media: { sourceID: string, sourceType: string } }} entry
   */
  async #getSourceDataForPlayback(entry) {
    const { sourceID, sourceType } = entry.media;
    const source = this.#uw.source(sourceType);
    if (source) {
      this.#logger.trace({ sourceType: source.type, sourceID }, 'running pre-play hook');
      /** @type {JsonObject | undefined} */
      let sourceData;
      try {
        sourceData = await source.play(entry.user, entry.media);
        this.#logger.trace({ sourceType: source.type, sourceID, sourceData }, 'pre-play hook result');
      } catch (error) {
        this.#logger.error({ sourceType: source.type, sourceID, err: error }, 'pre-play hook failed');
      }
      return sourceData;
    }

    return undefined;
  }

  /**
   * @typedef {object} AdvanceOptions
   * @prop {boolean} [remove]
   * @prop {boolean} [publish]
   * @prop {AbortSignal} [signal]
   * @param {AdvanceOptions} [opts]
   * @returns {Promise<{
   *   historyEntry: HistoryEntry,
   *   user: User,
   *   media: Media,
   *   playlist: Playlist,
   * }|null>}
   */
  async #advanceLocked(opts = {}, tx = this.#uw.db) {
    const { playlists } = this.#uw;

    const publish = opts.publish ?? true;
    const removeAfterCurrent = (await this.#uw.keyv.delete(KEY_REMOVE_AFTER_CURRENT_PLAY)) === true;
    const remove = opts.remove || removeAfterCurrent || (
      !await this.#uw.waitlist.isCycleEnabled()
    );

    const previous = await this.getCurrentEntry(tx);
    let next;
    try {
      next = await this.#getNextEntry({ remove });
    } catch (err) {
      // If the next user's playlist was empty, remove them from the waitlist
      // and try advancing again.
      if (err instanceof EmptyPlaylistError) {
        this.#logger.info('user has empty playlist, skipping on to the next');
        const previousDJ = previous != null ? previous.historyEntry.userID : null;
        await this.#cycleWaitlist(previousDJ, { remove });
        return this.#advanceLocked({ publish, remove: true }, tx);
      }
      throw err;
    }

    if (opts.signal?.aborted) {
      throw opts.signal.reason;
    }

    if (previous) {
      this.#logger.info({
        id: previous.historyEntry.id,
        artist: previous.media.artist,
        title: previous.media.title,
        upvotes: previous.upvotes.length,
        favorites: previous.favorites.length,
        downvotes: previous.downvotes.length,
      }, 'previous track stats');
    }

    let result = null;
    if (next != null) {
      this.#logger.info({
        id: next.playlistItem.id,
        artist: next.playlistItem.artist,
        title: next.playlistItem.title,
      }, 'next track');
      const sourceData = await this.#getSourceDataForPlayback(next);

      // Conservatively, we should take *all* the data from the inserted values.
      // But then we need to reparse the source data... It's easier to only take
      // the actually generated value from there :')
      const { createdAt } = await tx.insertInto('historyEntries')
        .returning('createdAt')
        .values({
          id: next.historyEntry.id,
          userID: next.user.id,
          mediaID: next.media.id,
          artist: next.historyEntry.artist,
          title: next.historyEntry.title,
          start: next.historyEntry.start,
          end: next.historyEntry.end,
          sourceData: sourceData != null ? jsonb(sourceData) : null,
        })
        .executeTakeFirstOrThrow();

      if (sourceData != null) {
        next.historyEntry.sourceData = sourceData;
      }
      next.historyEntry.createdAt = createdAt;

      result = {
        historyEntry: next.historyEntry,
        playlist: next.playlist,
        user: next.user,
        media: next.media,
      };
    } else {
      this.#maybeStop();
    }

    await this.#cycleWaitlist(previous != null ? previous.historyEntry.userID : null, { remove });

    if (next) {
      await this.#update(next);
      await playlists.cyclePlaylist(next.playlist, tx);
      this.#play(next.historyEntry);
    } else {
      await this.clear();
    }

    if (publish !== false) {
      await this.#publishAdvanceComplete(result);
    }

    return result;
  }

  /**
   * @param {AdvanceOptions} [opts]
   */
  advance(opts = {}) {
    const result = this.#mutex.withLock(() => {
      const signal = AbortSignal.timeout(10_000);
      return this.#advanceLocked({ ...opts, signal });
    });
    this.#awaitAdvance = result;
    return result;
  }

  /**
   * @param {User} user
   * @param {boolean} remove
   */
  async setRemoveAfterCurrentPlay(user, remove) {
    const newValue = await this.#uw.db.transaction().execute(async (tx) => {
      const currentDJ = /** @type {UserID|undefined} */ (
        await this.#uw.keyv.get(KEY_CURRENT_DJ_ID, tx)
      );
      if (currentDJ === user.id) {
        if (remove) {
          await this.#uw.keyv.set(KEY_REMOVE_AFTER_CURRENT_PLAY, true, tx);
          return true;
        }
        await this.#uw.keyv.delete(KEY_REMOVE_AFTER_CURRENT_PLAY, tx);
        return false;
      } else {
        throw new Error('You are not currently playing');
      }
    });
    return newValue;
  }

  /**
   * @param {User} user
   */
  async getRemoveAfterCurrentPlay(user, tx = this.#uw.db) {
    const currentDJ = /** @type {UserID|undefined} */ (
      await this.#uw.keyv.get(KEY_CURRENT_DJ_ID, tx)
    );
    const removeAfterCurrentPlay = /** @type {boolean|undefined} */ (
      await this.#uw.keyv.get(KEY_REMOVE_AFTER_CURRENT_PLAY, tx)
    );

    if (currentDJ === user.id) {
      return removeAfterCurrentPlay != null;
    }
    return null;
  }
}

/**
 * @param {import('../Uwave.js').Boot} uw
 */
async function boothPlugin(uw) {
  uw.booth = new Booth(uw);
  uw.httpApi.use('/booth', routes());

  uw.after(async (err) => {
    if (!err) {
      await uw.booth.onStart();
    }
  });
}

export default boothPlugin;
export { Booth };

1	import Mutex from 'p-mutex';	1✔
2	import { sql } from 'kysely';	1✔
3	import { EmptyPlaylistError, PlaylistItemNotFoundError } from '../errors/index.js';	1✔
4	import routes from '../routes/booth.js';	1✔
5	import { randomUUID } from 'node:crypto';	1✔
6	import { fromJson, jsonb, jsonGroupArray } from '../utils/sqlite.js';	1✔
7		1✔
8	/**	1✔
9	* @typedef {import('../schema.js').UserID} UserID	1✔
10	* @typedef {import('../schema.js').HistoryEntryID} HistoryEntryID	1✔
11	* @typedef {import('type-fest').JsonObject} JsonObject	1✔
12	* @typedef {import('../schema.js').User} User	1✔
13	* @typedef {import('../schema.js').Playlist} Playlist	1✔
14	* @typedef {import('../schema.js').PlaylistItem} PlaylistItem	1✔
15	* @typedef {import('../schema.js').HistoryEntry} HistoryEntry	1✔
16	* @typedef {Omit<import('../schema.js').Media, 'createdAt' \| 'updatedAt'>} Media	1✔
17	*/	1✔
18		1✔
19	const KEY_HISTORY_ID = 'booth:historyID';	1✔
20	const KEY_CURRENT_DJ_ID = 'booth:currentDJ';	1✔
21	const KEY_REMOVE_AFTER_CURRENT_PLAY = 'booth:removeAfterCurrentPlay';	1✔
22		1✔
23	class Booth {	1✔
24	#uw;	161✔
25		161✔
26	#logger;	161✔
27		161✔
28	/** @type {ReturnType<typeof setTimeout>\|null} */	161✔
29	#timeout = null;	161✔
30		161✔
31	#mutex;	161✔
32		161✔
33	/** @type {Promise<unknown>\|null} */	161✔
34	#awaitAdvance = null;	161✔
35		161✔
36	/**	161✔
37	* @param {import('../Uwave.js').Boot} uw	161✔
38	*/	161✔
39	constructor(uw) {	161✔
40	this.#uw = uw;	161✔
41	this.#mutex = new Mutex();	161✔
42	this.#logger = uw.logger.child({ ns: 'uwave:booth' });	161✔
43	}	161✔
44		161✔
45	/** @internal */	161✔
46	async onStart() {	161✔
47	const current = await this.getCurrentEntry();	161✔
48	if (current && this.#timeout === null) {	161!
49	// Restart the advance timer after a server restart, if a track was	×
50	// playing before the server restarted.	×
51	const duration = (current.historyEntry.end - current.historyEntry.start) * 1000;	×
52	const endTime = current.historyEntry.createdAt.getTime() + duration;	×
53	if (endTime > Date.now()) {	×
54	this.#timeout = setTimeout(	×
55	() => this.#advanceAutomatically(),	×
56	endTime - Date.now(),	×
57	);	×
58	} else {	×
59	this.#advanceAutomatically();	×
60	}	×
61	}	×
62		161✔
63	this.#uw.onClose(async () => {	161✔
64	this.#onStop();	161✔
65	await this.#awaitAdvance;	161✔
66	});	161✔
67	}	161✔
68		161✔
69	async #advanceAutomatically() {	161✔
70	try {	×
71	await this.advance();	×
72	} catch (error) {	×
73	this.#logger.error({ err: error }, 'advance failed');	×
74	}	×
75	}	×
76		161✔
77	#onStop() {	161✔
78	this.#maybeStop();	161✔
79	}	161✔
80		161✔
81	async getCurrentEntry(tx = this.#uw.db) {	161✔
82	const entry = await tx.selectFrom('keyval')	191✔
83	.where('key', '=', KEY_HISTORY_ID)	191✔
84	.innerJoin('historyEntries', (join) => join.on(	191✔
85	(eb) => sql`${eb.ref('value')}->>'$'`,	191✔
86	'=',	191✔
87	(eb) => eb.ref('historyEntries.id'),	191✔
88	))	191✔
89	.innerJoin('media', 'historyEntries.mediaID', 'media.id')	191✔
90	.innerJoin('users', 'historyEntries.userID', 'users.id')	191✔
91	.select([	191✔
92	'historyEntries.id as id',	191✔
93	'media.id as media.id',	191✔
94	'media.sourceID as media.sourceID',	191✔
95	'media.sourceType as media.sourceType',	191✔
96	'media.sourceData as media.sourceData',	191✔
97	'media.artist as media.artist',	191✔
98	'media.title as media.title',	191✔
99	'media.duration as media.duration',	191✔
100	'media.thumbnail as media.thumbnail',	191✔
101	'users.id as users.id',	191✔
102	'users.username as users.username',	191✔
103	'users.avatar as users.avatar',	191✔
104	'users.createdAt as users.createdAt',	191✔
105	'historyEntries.artist',	191✔
106	'historyEntries.title',	191✔
107	'historyEntries.start',	191✔
108	'historyEntries.end',	191✔
109	'historyEntries.createdAt',	191✔
110	(eb) => eb.selectFrom('feedback')	191✔
111	.where('historyEntryID', '=', eb.ref('historyEntries.id'))	191✔
112	.where('vote', '=', 1)	191✔
113	.select((eb) => jsonGroupArray(eb.ref('userID')).as('userIDs'))	191✔
114	.as('upvotes'),	191✔
115	(eb) => eb.selectFrom('feedback')	191✔
116	.where('historyEntryID', '=', eb.ref('historyEntries.id'))	191✔
117	.where('vote', '=', -1)	191✔
118	.select((eb) => jsonGroupArray(eb.ref('userID')).as('userIDs'))	191✔
119	.as('downvotes'),	191✔
120	(eb) => eb.selectFrom('feedback')	191✔
121	.where('historyEntryID', '=', eb.ref('historyEntries.id'))	191✔
122	.where('favorite', '=', 1)	191✔
123	.select((eb) => jsonGroupArray(eb.ref('userID')).as('userIDs'))	191✔
124	.as('favorites'),	191✔
125	])	191✔
126	.executeTakeFirst();	191✔
127		191✔
128	return entry ? {	191✔
129	media: {	7✔
130	id: entry['media.id'],	7✔
131	artist: entry['media.artist'],	7✔
132	title: entry['media.title'],	7✔
133	duration: entry['media.duration'],	7✔
134	thumbnail: entry['media.thumbnail'],	7✔
135	sourceID: entry['media.sourceID'],	7✔
136	sourceType: entry['media.sourceType'],	7✔
137	sourceData: entry['media.sourceData'] ?? {},	7✔
138	},	7✔
139	user: {	7✔
140	id: entry['users.id'],	7✔
141	username: entry['users.username'],	7✔
142	avatar: entry['users.avatar'],	7✔
143	createdAt: entry['users.createdAt'],	7✔
144	},	7✔
145	historyEntry: {	7✔
146	id: entry.id,	7✔
147	userID: entry['users.id'],	7✔
148	mediaID: entry['media.id'],	7✔
149	artist: entry.artist,	7✔
150	title: entry.title,	7✔
151	start: entry.start,	7✔
152	end: entry.end,	7✔
153	createdAt: entry.createdAt,	7✔
154	},	7✔
155	upvotes: entry.upvotes != null ? fromJson(entry.upvotes) : [],	7!
156	downvotes: entry.downvotes != null ? fromJson(entry.downvotes) : [],	7!
157	favorites: entry.favorites != null ? fromJson(entry.favorites) : [],	7!
158	} : null;	191✔
159	}	191✔
160		161✔
161	/**	161✔
162	* @param {{ remove?: boolean }} options	161✔
163	*/	161✔
164	async #getNextDJ(options, tx = this.#uw.db) {	161✔
165	const waitlist = await this.#uw.waitlist.getUserIDs();	18✔
166	let userID = waitlist.at(0) ?? null;	18!
167	if (!userID && !options.remove) {	18!
168	// If the waitlist is empty, the current DJ will play again immediately.	×
169	userID = /** @type {UserID\|null} */ (await this.#uw.keyv.get(KEY_CURRENT_DJ_ID, tx));	×
170	}	×
171	if (!userID) {	18!
172	return null;	×
173	}	×
174		18✔
175	return this.#uw.users.getUser(userID, tx);	18✔
176	}	18✔
177		161✔
178	/**	161✔
179	* @param {{ remove?: boolean }} options	161✔
180	*/	161✔
181	async #getNextEntry(options) {	161✔
182	const { playlists } = this.#uw;	18✔
183		18✔
184	const user = await this.#getNextDJ(options);	18✔
185	if (!user \|\| !user.activePlaylistID) {	18!
186	return null;	×
187	}	×
188	const playlist = await playlists.getUserPlaylist(user, user.activePlaylistID);	18✔
189	if (playlist.size === 0) {	18!
190	throw new EmptyPlaylistError();	×
191	}	×
192		18✔
193	const { playlistItem, media } = await playlists.getPlaylistItemAt(playlist, 0);	18✔
194	if (!playlistItem) {	18!
195	throw new PlaylistItemNotFoundError();	×
196	}	×
197		18✔
198	return {	18✔
199	user,	18✔
200	playlist,	18✔
201	playlistItem,	18✔
202	media,	18✔
203	historyEntry: {	18✔
204	id: /** @type {HistoryEntryID} */ (randomUUID()),	18✔
205	userID: user.id,	18✔
206	mediaID: media.id,	18✔
207	artist: playlistItem.artist,	18✔
208	title: playlistItem.title,	18✔
209	start: playlistItem.start,	18✔
210	end: playlistItem.end,	18✔
211	/** @type {null \| JsonObject} */	18✔
212	sourceData: null,	18✔
213	createdAt: new Date(),	18✔
214	},	18✔
215	};	18✔
216	}	18✔
217		161✔
218	/**	161✔
219	* @param {UserID\|null} previous	161✔
220	* @param {{ remove?: boolean }} options	161✔
221	*/	161✔
222	async #cycleWaitlist(previous, options) {	161✔
223	await this.#uw.waitlist.cycle(previous, options);	18✔
224	}	18✔
225		161✔
226	async clear(tx = this.#uw.db) {	161✔
227	await this.#uw.keyv.delete(KEY_REMOVE_AFTER_CURRENT_PLAY, tx);	×
228	await this.#uw.keyv.delete(KEY_HISTORY_ID, tx);	×
229	await this.#uw.keyv.delete(KEY_CURRENT_DJ_ID, tx);	×
230	}	×
231		161✔
232	/**	161✔
233	* @param {{ historyEntry: { id: HistoryEntryID }, user: { id: UserID } }} next	161✔
234	*/	161✔
235	async #update(next, tx = this.#uw.db) {	161✔
236	await this.#uw.keyv.delete(KEY_REMOVE_AFTER_CURRENT_PLAY, tx);	18✔
237	await this.#uw.keyv.set(KEY_HISTORY_ID, next.historyEntry.id, tx);	18✔
238	await this.#uw.keyv.set(KEY_CURRENT_DJ_ID, next.user.id, tx);	18✔
239	}	18✔
240		161✔
241	#maybeStop() {	161✔
242	if (this.#timeout) {	179✔
243	clearTimeout(this.#timeout);	18✔
244	this.#timeout = null;	18✔
245	}	18✔
246	}	179✔
247		161✔
248	/**	161✔
249	* @param {Pick<HistoryEntry, 'start' \| 'end'>} entry	161✔
250	*/	161✔
251	#play(entry) {	161✔
252	this.#maybeStop();	18✔
253	this.#timeout = setTimeout(	18✔
254	() => this.#advanceAutomatically(),	18✔
255	(entry.end - entry.start) * 1000,	18✔
256	);	18✔
257	}	18✔
258		161✔
259	/**	161✔
260	* This method creates a `media` object that clients can understand from a	161✔
261	* history entry object.	161✔
262	*	161✔
263	* We present the playback-specific `sourceData` as if it is	161✔
264	* a property of the media model for backwards compatibility.	161✔
265	* Old clients don't expect `sourceData` directly on a history entry object.	161✔
266	*	161✔
267	* @param {{ user: User, media: Media, historyEntry: HistoryEntry }} next	161✔
268	*/	161✔
269	getMediaForPlayback(next) {	161✔
270	return {	25✔
271	artist: next.historyEntry.artist,	25✔
272	title: next.historyEntry.title,	25✔
273	start: next.historyEntry.start,	25✔
274	end: next.historyEntry.end,	25✔
275	media: {	25✔
276	sourceType: next.media.sourceType,	25✔
277	sourceID: next.media.sourceID,	25✔
278	artist: next.media.artist,	25✔
279	title: next.media.title,	25✔
280	duration: next.media.duration,	25✔
281	sourceData: {	25✔
282	...next.media.sourceData,	25✔
283	...next.historyEntry.sourceData,	25✔
284	},	25✔
285	},	25✔
286	};	25✔
287	}	25✔
288		161✔
289	/**	161✔
290	* @param {{	161✔
291	* user: User,	161✔
292	* playlist: Playlist,	161✔
293	* media: Media,	161✔
294	* historyEntry: HistoryEntry	161✔
295	* } \| null} next	161✔
296	*/	161✔
297	async #publishAdvanceComplete(next) {	161✔
298	const { waitlist } = this.#uw;	18✔
299		18✔
300	if (next != null) {	18✔
301	this.#uw.publish('advance:complete', {	18✔
302	historyID: next.historyEntry.id,	18✔
303	userID: next.user.id,	18✔
304	playlistID: next.playlist.id,	18✔
305	media: this.getMediaForPlayback(next),	18✔
306	playedAt: next.historyEntry.createdAt.getTime(),	18✔
307	});	18✔
308	this.#uw.publish('playlist:cycle', {	18✔
309	userID: next.user.id,	18✔
310	playlistID: next.playlist.id,	18✔
311	});	18✔
312	} else {	18!
313	this.#uw.publish('advance:complete', null);	×
314	}	×
315	this.#uw.publish('waitlist:update', await waitlist.getUserIDs());	18✔
316	}	18✔
317		161✔
318	/**	161✔
319	* @param {{ user: User, media: { sourceID: string, sourceType: string } }} entry	161✔
320	*/	161✔
321	async #getSourceDataForPlayback(entry) {	161✔
322	const { sourceID, sourceType } = entry.media;	18✔
323	const source = this.#uw.source(sourceType);	18✔
324	if (source) {	18✔
325	this.#logger.trace({ sourceType: source.type, sourceID }, 'running pre-play hook');	18✔
326	/** @type {JsonObject \| undefined} */	18✔
327	let sourceData;	18✔
328	try {	18✔
329	sourceData = await source.play(entry.user, entry.media);	18✔
330	this.#logger.trace({ sourceType: source.type, sourceID, sourceData }, 'pre-play hook result');	18✔
331	} catch (error) {	18!
332	this.#logger.error({ sourceType: source.type, sourceID, err: error }, 'pre-play hook failed');	×
333	}	×
334	return sourceData;	18✔
335	}	18✔
336		×
337	return undefined;	×
338	}	18✔
339		161✔
340	/**	161✔
341	* @typedef {object} AdvanceOptions	161✔
342	* @prop {boolean} [remove]	161✔
343	* @prop {boolean} [publish]	161✔
344	* @prop {AbortSignal} [signal]	161✔
345	* @param {AdvanceOptions} [opts]	161✔
346	* @returns {Promise<{	161✔
347	* historyEntry: HistoryEntry,	161✔
348	* user: User,	161✔
349	* media: Media,	161✔
350	* playlist: Playlist,	161✔
351	* }\|null>}	161✔
352	*/	161✔
353	async #advanceLocked(opts = {}, tx = this.#uw.db) {	161✔
354	const { playlists } = this.#uw;	18✔
355		18✔
356	const publish = opts.publish ?? true;	18✔
357	const removeAfterCurrent = (await this.#uw.keyv.delete(KEY_REMOVE_AFTER_CURRENT_PLAY)) === true;	18✔
358	const remove = opts.remove \|\| removeAfterCurrent \|\| (	18✔
359	!await this.#uw.waitlist.isCycleEnabled()	18✔
360	);	18✔
361		18✔
362	const previous = await this.getCurrentEntry(tx);	18✔
363	let next;	18✔
364	try {	18✔
365	next = await this.#getNextEntry({ remove });	18✔
366	} catch (err) {	18!
367	// If the next user's playlist was empty, remove them from the waitlist	×
368	// and try advancing again.	×
369	if (err instanceof EmptyPlaylistError) {	×
370	this.#logger.info('user has empty playlist, skipping on to the next');	×
371	const previousDJ = previous != null ? previous.historyEntry.userID : null;	×
372	await this.#cycleWaitlist(previousDJ, { remove });	×
373	return this.#advanceLocked({ publish, remove: true }, tx);	×
374	}	×
375	throw err;	×
376	}	×
377		18✔
378	if (opts.signal?.aborted) {	18!
NEW 379	throw opts.signal.reason;	×
380	}	×
381		18✔
382	if (previous) {	18!
383	this.#logger.info({	×
384	id: previous.historyEntry.id,	×
385	artist: previous.media.artist,	×
386	title: previous.media.title,	×
387	upvotes: previous.upvotes.length,	×
388	favorites: previous.favorites.length,	×
389	downvotes: previous.downvotes.length,	×
390	}, 'previous track stats');	×
391	}	×
392		18✔
393	let result = null;	18✔
394	if (next != null) {	18✔
395	this.#logger.info({	18✔
396	id: next.playlistItem.id,	18✔
397	artist: next.playlistItem.artist,	18✔
398	title: next.playlistItem.title,	18✔
399	}, 'next track');	18✔
400	const sourceData = await this.#getSourceDataForPlayback(next);	18✔
401		18✔
402	// Conservatively, we should take all the data from the inserted values.	18✔
403	// But then we need to reparse the source data... It's easier to only take	18✔
404	// the actually generated value from there :')	18✔
405	const { createdAt } = await tx.insertInto('historyEntries')	18✔
406	.returning('createdAt')	18✔
407	.values({	18✔
408	id: next.historyEntry.id,	18✔
409	userID: next.user.id,	18✔
410	mediaID: next.media.id,	18✔
411	artist: next.historyEntry.artist,	18✔
412	title: next.historyEntry.title,	18✔
413	start: next.historyEntry.start,	18✔
414	end: next.historyEntry.end,	18✔
415	sourceData: sourceData != null ? jsonb(sourceData) : null,	18!
416	})	18✔
417	.executeTakeFirstOrThrow();	18✔
418		18✔
419	if (sourceData != null) {	18!
420	next.historyEntry.sourceData = sourceData;	×
421	}	×
422	next.historyEntry.createdAt = createdAt;	18✔
423		18✔
424	result = {	18✔
425	historyEntry: next.historyEntry,	18✔
426	playlist: next.playlist,	18✔
427	user: next.user,	18✔
428	media: next.media,	18✔
429	};	18✔
430	} else {	18!
431	this.#maybeStop();	×
432	}	×
433		18✔
434	await this.#cycleWaitlist(previous != null ? previous.historyEntry.userID : null, { remove });	18!
435		18✔
436	if (next) {	18✔
437	await this.#update(next);	18✔
438	await playlists.cyclePlaylist(next.playlist, tx);	18✔
439	this.#play(next.historyEntry);	18✔
440	} else {	18!
441	await this.clear();	×
442	}	×
443		18✔
444	if (publish !== false) {	18✔
445	await this.#publishAdvanceComplete(result);	18✔
446	}	18✔
447		18✔
448	return result;	18✔
449	}	18✔
450		161✔
451	/**	161✔
452	* @param {AdvanceOptions} [opts]	161✔
453	*/	161✔
454	advance(opts = {}) {	161✔
455	const result = this.#mutex.withLock(() => {	18✔
456	const signal = AbortSignal.timeout(10_000);	18✔
457	return this.#advanceLocked({ ...opts, signal });	18✔
458	});	18✔
459	this.#awaitAdvance = result;	18✔
460	return result;	18✔
461	}	18✔
462		161✔
463	/**	161✔
464	* @param {User} user	161✔
465	* @param {boolean} remove	161✔
466	*/	161✔
467	async setRemoveAfterCurrentPlay(user, remove) {	161✔
468	const newValue = await this.#uw.db.transaction().execute(async (tx) => {	×
469	const currentDJ = /** @type {UserID\|undefined} */ (	×
470	await this.#uw.keyv.get(KEY_CURRENT_DJ_ID, tx)	×
471	);	×
472	if (currentDJ === user.id) {	×
473	if (remove) {	×
474	await this.#uw.keyv.set(KEY_REMOVE_AFTER_CURRENT_PLAY, true, tx);	×
475	return true;	×
476	}	×
477	await this.#uw.keyv.delete(KEY_REMOVE_AFTER_CURRENT_PLAY, tx);	×
478	return false;	×
479	} else {	×
480	throw new Error('You are not currently playing');	×
481	}	×
482	});	×
483	return newValue;	×
484	}	×
485		161✔
486	/**	161✔
487	* @param {User} user	161✔
488	*/	161✔
489	async getRemoveAfterCurrentPlay(user, tx = this.#uw.db) {	161✔
490	const currentDJ = /** @type {UserID\|undefined} */ (	3✔
491	await this.#uw.keyv.get(KEY_CURRENT_DJ_ID, tx)	3✔
492	);	3✔
493	const removeAfterCurrentPlay = /** @type {boolean\|undefined} */ (	3✔
494	await this.#uw.keyv.get(KEY_REMOVE_AFTER_CURRENT_PLAY, tx)	3✔
495	);	3✔
496		3✔
497	if (currentDJ === user.id) {	3✔
498	return removeAfterCurrentPlay != null;	1✔
499	}	1✔
500	return null;	2✔
501	}	3✔
502	}	161✔
503		1✔
504	/**	1✔
505	* @param {import('../Uwave.js').Boot} uw	1✔
506	*/	1✔
507	async function boothPlugin(uw) {	161✔
508	uw.booth = new Booth(uw);	161✔
509	uw.httpApi.use('/booth', routes());	161✔
510		161✔
511	uw.after(async (err) => {	161✔
512	if (!err) {	161✔
513	await uw.booth.onStart();	161✔
514	}	161✔
515	});	161✔
516	}	161✔
517		1✔
518	export default boothPlugin;	1✔
519	export { Booth };	1✔

u-wave / core / 20823830584

Source File Press 'n' to go to next uncovered line, 'b' for previous

Source File
Press 'n' to go to next uncovered line, 'b' for previous