11085094286

Committed 28 Sep 2024 03:39PM UTC coverage: 79.715% (-0.4%) from 80.131%

Build # 11085094286

Build Type

Pull #637

github

Committed by

web-flow

Commit Message

Merge 11ccf3b06 into 14c162f19

Pull Request Pull Request #637: Switch to a relational database, closes #549

Run Details

751 of 918 branches covered (81.81%)

Branch coverage included in aggregate %.

1891 of 2530 new or added lines in 50 files covered. (74.74%)

13 existing lines in 7 files now uncovered.

9191 of 11554 relevant lines covered (79.55%)

68.11 hits per line

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

82.42

/src/plugins/booth.js

import assert from 'node:assert';
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 REDIS_UPVOTES = 'booth:upvotes';
const REDIS_DOWNVOTES = 'booth:downvotes';
const REDIS_FAVORITES = 'booth:favorites';

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() {
    const { db } = this.#uw;

    const historyID = /** @type {HistoryEntryID} */ (await this.#uw.redis.get(REDIS_HISTORY_ID));
    if (!historyID) {
      return null;
    }

    const entry = await db.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',
      ])
      .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,
      },
      // TODO
      upvotes: [],
      downvotes: [],
      favorites: [],
    } : null;
  }

  /**
   * Get vote counts for the currently playing media.
   *
   * @returns {Promise<{ upvotes: UserID[], downvotes: UserID[], favorites: UserID[] }>}
   */
  async getCurrentVoteStats() {
    const { redis } = this.#uw;

    const results = await redis.pipeline()
      .smembers(REDIS_UPVOTES)
      .smembers(REDIS_DOWNVOTES)
      .smembers(REDIS_FAVORITES)
      .exec();
    assert(results);

    const voteStats = {
      upvotes: /** @type {UserID[]} */ (results[0][1]),
      downvotes: /** @type {UserID[]} */ (results[1][1]),
      favorites: /** @type {UserID[]} */ (results[2][1]),
    };

    return voteStats;
  }

  /** @param {{ remove?: boolean }} options */
  async #getNextDJ(options) {
    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);
  }

  /**
   * @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,
      REDIS_UPVOTES,
      REDIS_DOWNVOTES,
      REDIS_FAVORITES,
    );
  }

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

u-wave / core / 11085094286

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