diff --git a/server/models/User.js b/server/models/User.js index 9ad4772d5b..35d71ebd18 100644 --- a/server/models/User.js +++ b/server/models/User.js @@ -53,6 +53,14 @@ class UserCache { const userCache = new UserCache() +/** + * Pending media progress updates keyed by user id and media item + * Used to run updates for the same media item one at a time so concurrent requests + * (e.g. /api/session/local and /api/me/progress/batch/update) cannot each create a mediaProgress + * @type {Map>} + */ +const mediaProgressUpdateQueues = new Map() + const { DataTypes, Model } = sequelize /** @@ -721,13 +729,37 @@ class User extends Model { return mediaProgress?.getOldMediaProgress() || null } + /** + * Updates for the same media item are queued so that concurrent requests do not create duplicate mediaProgress rows + * + * @param {ProgressUpdatePayload} progressPayload + * @returns {Promise<{ mediaProgress: import('./MediaProgress'), error: [string], statusCode: [number] }>} + */ + createUpdateMediaProgressFromPayload(progressPayload) { + const queueKey = `${this.id}:${progressPayload.episodeId || progressPayload.libraryItemId}` + const previousUpdate = mediaProgressUpdateQueues.get(queueKey) || Promise.resolve() + const update = previousUpdate.then(() => this.applyMediaProgressPayload(progressPayload)) + + // Keep the queue going if this update fails and remove the key once nothing else is queued + const queueTail = update.catch(() => {}) + mediaProgressUpdateQueues.set(queueKey, queueTail) + queueTail.then(() => { + if (mediaProgressUpdateQueues.get(queueKey) === queueTail) { + mediaProgressUpdateQueues.delete(queueKey) + } + }) + + return update + } + /** * TODO: Uses old model and should account for the different between ebook/audiobook progress + * Should only be called from createUpdateMediaProgressFromPayload * * @param {ProgressUpdatePayload} progressPayload * @returns {Promise<{ mediaProgress: import('./MediaProgress'), error: [string], statusCode: [number] }>} */ - async createUpdateMediaProgressFromPayload(progressPayload) { + async applyMediaProgressPayload(progressPayload) { /** @type {import('./MediaProgress')|null} */ let mediaProgress = null let mediaItemId = null diff --git a/test/server/models/User.test.js b/test/server/models/User.test.js new file mode 100644 index 0000000000..b4a8e65161 --- /dev/null +++ b/test/server/models/User.test.js @@ -0,0 +1,63 @@ +const { expect } = require('chai') +const { Sequelize } = require('sequelize') +const sinon = require('sinon') + +const Database = require('../../../server/Database') +const Logger = require('../../../server/Logger') + +describe('User', () => { + beforeEach(async () => { + global.ServerSettings = {} + Database.sequelize = new Sequelize({ dialect: 'sqlite', storage: ':memory:', logging: false }) + Database.sequelize.uppercaseFirst = (str) => (str ? `${str[0].toUpperCase()}${str.substr(1)}` : '') + await Database.buildModels() + + sinon.stub(Logger, 'info') + }) + + afterEach(async () => { + sinon.restore() + + // Clear all tables + await Database.sequelize.sync({ force: true }) + }) + + describe('createUpdateMediaProgressFromPayload', () => { + let user + let book + let libraryItem + + beforeEach(async () => { + const library = await Database.libraryModel.create({ name: 'Test Library', mediaType: 'book' }) + const libraryFolder = await Database.libraryFolderModel.create({ path: '/test', libraryId: library.id }) + book = await Database.bookModel.create({ title: 'Test Book', audioFiles: [], tags: [], narrators: [], genres: [], chapters: [] }) + libraryItem = await Database.libraryItemModel.create({ libraryFiles: [], mediaId: book.id, mediaType: 'book', libraryId: library.id, libraryFolderId: libraryFolder.id }) + + await Database.userModel.create({ username: 'testuser', type: 'user', isActive: true }) + user = await Database.userModel.findOne({ where: { username: 'testuser' }, include: Database.mediaProgressModel }) + }) + + it('should create a single media progress when concurrent updates create progress for the same item', async () => { + const results = await Promise.all([user.createUpdateMediaProgressFromPayload({ libraryItemId: libraryItem.id, currentTime: 100, duration: 3600 }), user.createUpdateMediaProgressFromPayload({ libraryItemId: libraryItem.id, currentTime: 200, duration: 3600 }), user.createUpdateMediaProgressFromPayload({ libraryItemId: libraryItem.id, currentTime: 300, duration: 3600 })]) + + const mediaProgresses = await Database.mediaProgressModel.findAll({ where: { userId: user.id, mediaItemId: book.id } }) + expect(mediaProgresses).to.have.length(1) + expect(mediaProgresses[0].currentTime).to.equal(300) + expect(results.map((r) => r.mediaProgress.id)).to.deep.equal([mediaProgresses[0].id, mediaProgresses[0].id, mediaProgresses[0].id]) + + expect(user.mediaProgresses).to.have.length(1) + expect(user.getMediaProgress(book.id).currentTime).to.equal(300) + }) + + it('should keep processing queued updates for the same item after an update fails', async () => { + const applyStub = sinon.stub(user, 'applyMediaProgressPayload').callThrough() + applyStub.onFirstCall().rejects(new Error('db error')) + + const [failedUpdate, update] = await Promise.allSettled([user.createUpdateMediaProgressFromPayload({ libraryItemId: libraryItem.id, currentTime: 100, duration: 3600 }), user.createUpdateMediaProgressFromPayload({ libraryItemId: libraryItem.id, currentTime: 200, duration: 3600 })]) + + expect(failedUpdate.status).to.equal('rejected') + expect(update.status).to.equal('fulfilled') + expect(update.value.mediaProgress.currentTime).to.equal(200) + }) + }) +})