12018572686

Committed 25 Nov 2024 08:51PM UTC coverage: 82.056% (-0.03%) from 82.089%

Build # 12018572686

Build Type

Pull #657

github

Committed by

web-flow

Commit Message

Merge 3742dbfd0 into cbb61457b

Pull Request Pull Request #657: SQLite Key-value store to replace Redis

Run Details

840 of 1010 branches covered (83.17%)

Branch coverage included in aggregate %.

26 of 37 new or added lines in 2 files covered. (70.27%)

109 existing lines in 4 files now uncovered.

9115 of 11122 relevant lines covered (81.95%)

89.39 hits per line

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

82.8

/src/plugins/booth.js

import RedLock from 'redlock';
import { EmptyPlaylistError, PlaylistItemNotFoundError } from '../errors/index.js';
import routes from '../routes/booth.js';
import { randomUUID } from 'node:crypto';
import { jsonb } 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 REDIS_ADVANCING = 'booth:advancing';
const REDIS_HISTORY_ID = 'booth:historyID';
const REDIS_CURRENT_DJ_ID = 'booth:currentDJ';
const REDIS_REMOVE_AFTER_CURRENT_PLAY = 'booth:removeAfterCurrentPlay';

const REMOVE_AFTER_CURRENT_PLAY_SCRIPT = {
  keys: [REDIS_CURRENT_DJ_ID, REDIS_REMOVE_AFTER_CURRENT_PLAY],
  lua: `
    local k_dj = KEYS[1]
    local k_remove = KEYS[2]
    local user_id = ARGV[1]
    local value = ARGV[2]
    local current_dj_id = redis.call('GET', k_dj)
    if current_dj_id == user_id then
      if value == 'true' then
        redis.call('SET', k_remove, 'true')
        return 1
      else
        redis.call('DEL', k_remove)
        return 0
      end
    else
      return redis.error_reply('You are not currently playing')
    end
  `,
};

class Booth {
  #uw;

  #logger;

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

  #locker;

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

  /**
   * @param {import('../Uwave.js').Boot} uw
   */
  constructor(uw) {
    this.#uw = uw;
    this.#locker = new RedLock([this.#uw.redis]);
    this.#logger = uw.logger.child({ ns: 'uwave:booth' });

    uw.redis.defineCommand('uw:removeAfterCurrentPlay', {
      numberOfKeys: REMOVE_AFTER_CURRENT_PLAY_SCRIPT.keys.length,
      lua: REMOVE_AFTER_CURRENT_PLAY_SCRIPT.lua,
    });
  }

  /** @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 historyID = /** @type {HistoryEntryID} */ (await this.#uw.redis.get(REDIS_HISTORY_ID));
    if (!historyID) {
      return null;
    }

    const entry = await tx.selectFrom('historyEntries')
      .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) => eb.fn.agg('json_group_array', ['userID']).as('userIDs'))
          .as('upvotes'),
        (eb) => eb.selectFrom('feedback')
          .where('historyEntryID', '=', eb.ref('historyEntries.id'))
          .where('vote', '=', -1)
          .select((eb) => eb.fn.agg('json_group_array', ['userID']).as('userIDs'))
          .as('downvotes'),
        (eb) => eb.selectFrom('feedback')
          .where('historyEntryID', '=', eb.ref('historyEntries.id'))
          .where('favorite', '=', 1)
          .select((eb) => eb.fn.agg('json_group_array', ['userID']).as('userIDs'))
          .as('favorites'),
      ])
      .where('historyEntries.id', '=', historyID)
      .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: /** @type {UserID[]} */ (JSON.parse(entry.upvotes)),
      downvotes: /** @type {UserID[]} */ (JSON.parse(entry.downvotes)),
      favorites: /** @type {UserID[]} */ (JSON.parse(entry.favorites)),
    } : null;
  }

  /**
   * @param {{ remove?: boolean }} options
   */
  async #getNextDJ(options, tx = this.#uw.db) {
    let userID = /** @type {UserID|null} */ (await this.#uw.redis.lindex('waitlist', 0));
    if (!userID && !options.remove) {
      // If the waitlist is empty, the current DJ will play again immediately.
      userID = /** @type {UserID|null} */ (await this.#uw.redis.get(REDIS_CURRENT_DJ_ID));
    }
    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,
      },
    };
  }

  /**
   * @param {UserID|null} previous
   * @param {{ remove?: boolean }} options
   */
  async #cycleWaitlist(previous, options) {
    const waitlistLen = await this.#uw.redis.llen('waitlist');
    if (waitlistLen > 0) {
      await this.#uw.redis.lpop('waitlist');
      if (previous && !options.remove) {
        // The previous DJ should only be added to the waitlist again if it was
        // not empty. If it was empty, the previous DJ is already in the booth.
        await this.#uw.redis.rpush('waitlist', previous);
      }
    }
  }

  async clear() {
    await this.#uw.redis.del(
      REDIS_HISTORY_ID,
      REDIS_CURRENT_DJ_ID,
      REDIS_REMOVE_AFTER_CURRENT_PLAY,
    );
  }

  /**
   * @param {{ historyEntry: { id: HistoryEntryID }, user: { id: UserID } }} next
   */
  async #update(next) {
    await this.#uw.redis.multi()
      .del(REDIS_REMOVE_AFTER_CURRENT_PLAY)
      .set(REDIS_HISTORY_ID, next.historyEntry.id)
      .set(REDIS_CURRENT_DJ_ID, next.user.id)
      .exec();
  }

  #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 {import('redlock').RedlockAbortSignal} [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.redis.del(REDIS_REMOVE_AFTER_CURRENT_PLAY)) === 1;
    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.error;
    }

    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);
      if (sourceData) {
        next.historyEntry.sourceData = sourceData;
      }
      const historyEntry = await tx.insertInto('historyEntries')
        .returningAll()
        .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();

      result = {
        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.#locker.using(
      [REDIS_ADVANCING],
      10_000,
      (signal) => this.#advanceLocked({ ...opts, signal }),
    );
    this.#awaitAdvance = result;
    return result;
  }

  /**
   * @param {User} user
   * @param {boolean} remove
   */
  async setRemoveAfterCurrentPlay(user, remove) {
    const newValue = await this.#uw.redis['uw:removeAfterCurrentPlay'](
      ...REMOVE_AFTER_CURRENT_PLAY_SCRIPT.keys,
      user.id,
      remove,
    );
    return newValue === 1;
  }

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

u-wave / core / 12018572686

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