fix(db): serialize complete JSON mutations and preserve concurrent edits

Adapted from MikaelKW/nodecast-tv-plus commit 55cb8de23a37eb5683459bffed9b64c6c5dd73ae. Queue reads, validations, and writes together; return write errors to callers; preserve existing IDs and metadata; route source/user edits through the atomic operations. Cover concurrent writes, stale edits, queue recovery, and administrator races.

Co-authored-by: Mikael <mikaelkwessel@gmail.com>
This commit is contained in:
Technomancer702
2026-10-12 16:01:58 -07:00
parent 8127d26f29
commit 12080ba0a6
5 changed files with 442 additions and 404 deletions

View File

@ -3,53 +3,66 @@ const path = require('path');
const { existsSync, mkdirSync } = require('fs');
// Ensure data directory exists (sync is fine for startup)
const dataDir = path.join(__dirname, '..', 'data');
const dataDir = process.env.NODECAST_DATA_DIR
? path.resolve(process.env.NODECAST_DATA_DIR)
: path.join(__dirname, '..', 'data');
if (!existsSync(dataDir)) {
mkdirSync(dataDir, { recursive: true });
}
const dbPath = path.join(dataDir, 'db.json');
// Initialize database structure
function getEmptyDb() {
return {
sources: [],
hiddenItems: [],
favorites: [],
settings: getDefaultSettings(),
users: [],
nextId: 1
};
}
// Writes use this strict reader so a damaged or temporarily unreadable file
// cannot be replaced with an empty database after an apparently successful edit.
async function readDbForMutation() {
try {
const fileContent = await fs.readFile(dbPath, 'utf-8');
const data = JSON.parse(fileContent);
if (!data || typeof data !== 'object' || Array.isArray(data)) {
throw new Error('Invalid database document');
}
let nextId = Number.isSafeInteger(data.nextId) && data.nextId > 0 ? data.nextId : 1;
for (const records of [data.sources, data.hiddenItems, data.favorites, data.users]) {
if (!records) continue;
if (!Array.isArray(records)) throw new Error('Invalid database collection');
for (const record of records) {
const id = Number(record.id);
if (Number.isSafeInteger(id) && id >= nextId) nextId = id + 1;
}
}
return {
...data,
sources: data.sources || [],
hiddenItems: data.hiddenItems || [],
favorites: data.favorites || [],
settings: data.settings || getDefaultSettings(),
users: data.users || [],
nextId
};
} catch (error) {
if (error.code === 'ENOENT') return getEmptyDb();
throw error;
}
}
// Preserve the existing read-only fallback behavior. Mutations never use it.
async function loadDb() {
try {
// Check if file exists (using fs.access is better for async, but we can catch ENOENT)
try {
const fileContent = await fs.readFile(dbPath, 'utf-8');
const data = JSON.parse(fileContent);
return {
sources: data.sources || [],
hiddenItems: data.hiddenItems || [],
favorites: data.favorites || [],
settings: data.settings || getDefaultSettings(),
users: data.users || [],
nextId: data.nextId || 1
};
} catch (error) {
if (error.code === 'ENOENT') {
// File doesn't exist, return default
return {
sources: [],
hiddenItems: [],
favorites: [],
settings: getDefaultSettings(),
users: [],
nextId: 1
};
}
throw error;
}
return await readDbForMutation();
} catch (err) {
console.error('Error loading database:', err);
// Return safe default on error to prevent crashing, but log it
return {
sources: [],
hiddenItems: [],
favorites: [],
settings: getDefaultSettings(),
users: [],
nextId: 1
};
return getEmptyDb();
}
}
@ -101,30 +114,54 @@ function getUserAgent(settings) {
return USER_AGENT_PRESETS[settings.userAgentPreset] || USER_AGENT_PRESETS.chrome;
}
// Write lock to prevent concurrent writes from corrupting db.json
let writeQueue = Promise.resolve();
// Queue the entire read-modify-write transaction, not only the file write.
let mutationQueue = Promise.resolve();
const tmpPath = dbPath + '.tmp';
async function saveDb(data) {
// Queue this write operation - each write waits for the previous one
writeQueue = writeQueue.then(async () => {
try {
const jsonString = JSON.stringify(data, null, 2);
// Atomic write: write to temp file, then rename
// Rename is atomic on most filesystems, preventing corruption on crash
await fs.writeFile(tmpPath, jsonString);
await fs.rename(tmpPath, dbPath);
} catch (err) {
console.error('Error writing database:', err);
// Clean up temp file if it exists
try { await fs.unlink(tmpPath); } catch { /* ignore */ }
throw err;
async function writeDb(data) {
try {
const jsonString = JSON.stringify(data, null, 2);
// Rename keeps readers from observing a partially written JSON file.
await fs.writeFile(tmpPath, jsonString);
// Windows can briefly lock the destination while another request reads it.
// Retry only transient rename errors; persistent failures still reach the caller.
for (let attempt = 0; ; attempt++) {
try {
await fs.rename(tmpPath, dbPath);
break;
} catch (error) {
if (attempt >= 4 || !['EPERM', 'EACCES', 'EBUSY'].includes(error.code)) throw error;
await new Promise(resolve => setTimeout(resolve, 10 * (2 ** attempt)));
}
}
}).catch(err => {
console.error('Database write failed:', err);
});
} catch (err) {
console.error('Error writing database:', err);
try { await fs.unlink(tmpPath); } catch { /* ignore */ }
throw err;
}
}
return writeQueue;
function queueMutation(operation) {
const result = mutationQueue.then(operation);
// A rejected operation must not poison later writes, but its own caller
// must receive the error instead of an incorrect success response.
mutationQueue = result.catch(() => {});
return result;
}
async function mutateDb(mutator) {
return queueMutation(async () => {
const data = await readDbForMutation();
let changed = false;
const result = await mutator(data, () => { changed = true; });
if (changed) await writeDb(data);
return result;
});
}
// Full-document saves remain serialized for compatibility with external callers.
function saveDb(data) {
return queueMutation(() => writeDb(data));
}
// Source CRUD operations
@ -145,51 +182,55 @@ const sources = {
},
async create(source) {
const db = await loadDb();
const newSource = {
id: db.nextId++,
...source,
enabled: true,
created_at: new Date().toISOString(),
updated_at: new Date().toISOString()
};
db.sources.push(newSource);
await saveDb(db);
return newSource;
return mutateDb((db, markChanged) => {
const newSource = {
id: db.nextId++,
...source,
enabled: true,
created_at: new Date().toISOString(),
updated_at: new Date().toISOString()
};
db.sources.push(newSource);
markChanged();
return newSource;
});
},
async update(id, updates) {
const db = await loadDb();
const index = db.sources.findIndex(s => s.id === parseInt(id));
if (index === -1) return null;
return mutateDb((db, markChanged) => {
const index = db.sources.findIndex(s => s.id === parseInt(id));
if (index === -1) return null;
db.sources[index] = {
...db.sources[index],
...updates,
updated_at: new Date().toISOString()
};
await saveDb(db);
return db.sources[index];
db.sources[index] = {
...db.sources[index],
...updates,
updated_at: new Date().toISOString()
};
markChanged();
return db.sources[index];
});
},
async delete(id) {
const db = await loadDb();
db.sources = db.sources.filter(s => s.id !== parseInt(id));
// Also delete related hidden items and favorites
db.hiddenItems = db.hiddenItems.filter(h => h.source_id !== parseInt(id));
db.favorites = db.favorites.filter(f => f.source_id !== parseInt(id));
await saveDb(db);
return mutateDb((db, markChanged) => {
db.sources = db.sources.filter(s => s.id !== parseInt(id));
// Also delete related hidden items and favorites
db.hiddenItems = db.hiddenItems.filter(h => h.source_id !== parseInt(id));
db.favorites = db.favorites.filter(f => f.source_id !== parseInt(id));
markChanged();
});
},
async toggleEnabled(id) {
const db = await loadDb();
const source = db.sources.find(s => s.id === parseInt(id));
if (source) {
source.enabled = !source.enabled;
source.updated_at = new Date().toISOString();
await saveDb(db);
}
return source;
return mutateDb((db, markChanged) => {
const source = db.sources.find(s => s.id === parseInt(id));
if (source) {
source.enabled = !source.enabled;
source.updated_at = new Date().toISOString();
markChanged();
}
return source;
});
}
};
@ -204,28 +245,30 @@ const hiddenItems = {
},
async hide(sourceId, itemType, itemId) {
const db = await loadDb();
// Check if already hidden
const exists = db.hiddenItems.find(
h => h.source_id === parseInt(sourceId) && h.item_type === itemType && h.item_id === itemId
);
if (!exists) {
db.hiddenItems.push({
id: db.nextId++,
source_id: parseInt(sourceId),
item_type: itemType,
item_id: itemId
});
await saveDb(db);
}
return mutateDb((db, markChanged) => {
// Check if already hidden
const exists = db.hiddenItems.find(
h => h.source_id === parseInt(sourceId) && h.item_type === itemType && h.item_id === itemId
);
if (!exists) {
db.hiddenItems.push({
id: db.nextId++,
source_id: parseInt(sourceId),
item_type: itemType,
item_id: itemId
});
markChanged();
}
});
},
async show(sourceId, itemType, itemId) {
const db = await loadDb();
db.hiddenItems = db.hiddenItems.filter(
h => !(h.source_id === parseInt(sourceId) && h.item_type === itemType && h.item_id === itemId)
);
await saveDb(db);
return mutateDb((db, markChanged) => {
db.hiddenItems = db.hiddenItems.filter(
h => !(h.source_id === parseInt(sourceId) && h.item_type === itemType && h.item_id === itemId)
);
markChanged();
});
},
async isHidden(sourceId, itemType, itemId) {
@ -236,47 +279,49 @@ const hiddenItems = {
},
async bulkHide(items) {
const db = await loadDb();
let modified = false;
return mutateDb((db, markChanged) => {
let modified = false;
items.forEach(item => {
const { sourceId, itemType, itemId } = item;
const exists = db.hiddenItems.find(
h => h.source_id === parseInt(sourceId) && h.item_type === itemType && h.item_id === itemId
);
items.forEach(item => {
const { sourceId, itemType, itemId } = item;
const exists = db.hiddenItems.find(
h => h.source_id === parseInt(sourceId) && h.item_type === itemType && h.item_id === itemId
);
if (!exists) {
db.hiddenItems.push({
id: db.nextId++,
source_id: parseInt(sourceId),
item_type: itemType,
item_id: itemId
});
modified = true;
if (!exists) {
db.hiddenItems.push({
id: db.nextId++,
source_id: parseInt(sourceId),
item_type: itemType,
item_id: itemId
});
modified = true;
}
});
if (modified) {
markChanged();
}
return true;
});
if (modified) {
await saveDb(db);
}
return true;
},
async bulkShow(items) {
const db = await loadDb();
const initialLength = db.hiddenItems.length;
return mutateDb((db, markChanged) => {
const initialLength = db.hiddenItems.length;
// Create a set of "signatures" for O(1) lookup of items to remove
const toRemove = new Set(items.map(i => `${i.sourceId}:${i.itemType}:${i.itemId}`));
// Create a set of "signatures" for O(1) lookup of items to remove
const toRemove = new Set(items.map(i => `${i.sourceId}:${i.itemType}:${i.itemId}`));
db.hiddenItems = db.hiddenItems.filter(h =>
!toRemove.has(`${h.source_id}:${h.item_type}:${h.item_id}`)
);
db.hiddenItems = db.hiddenItems.filter(h =>
!toRemove.has(`${h.source_id}:${h.item_type}:${h.item_id}`)
);
if (db.hiddenItems.length !== initialLength) {
await saveDb(db);
}
return true;
if (db.hiddenItems.length !== initialLength) {
markChanged();
}
return true;
});
}
};
@ -295,31 +340,33 @@ const favorites = {
},
async add(sourceId, itemId, itemType = 'channel') {
const db = await loadDb();
// Check if already favorited
const exists = db.favorites.find(
f => f.source_id === parseInt(sourceId) && f.item_id === String(itemId) && f.item_type === itemType
);
if (!exists) {
db.favorites.push({
id: db.nextId++,
source_id: parseInt(sourceId),
item_id: String(itemId),
item_type: itemType, // 'channel', 'movie', 'series'
created_at: new Date().toISOString()
});
await saveDb(db);
}
return true;
return mutateDb((db, markChanged) => {
// Check if already favorited
const exists = db.favorites.find(
f => f.source_id === parseInt(sourceId) && f.item_id === String(itemId) && f.item_type === itemType
);
if (!exists) {
db.favorites.push({
id: db.nextId++,
source_id: parseInt(sourceId),
item_id: String(itemId),
item_type: itemType, // 'channel', 'movie', 'series'
created_at: new Date().toISOString()
});
markChanged();
}
return true;
});
},
async remove(sourceId, itemId, itemType = 'channel') {
const db = await loadDb();
db.favorites = db.favorites.filter(
f => !(f.source_id === parseInt(sourceId) && f.item_id === String(itemId) && f.item_type === itemType)
);
await saveDb(db);
return true;
return mutateDb((db, markChanged) => {
db.favorites = db.favorites.filter(
f => !(f.source_id === parseInt(sourceId) && f.item_id === String(itemId) && f.item_type === itemType)
);
markChanged();
return true;
});
},
async isFavorite(sourceId, itemId, itemType = 'channel') {
@ -338,17 +385,19 @@ const settings = {
},
async update(newSettings) {
const db = await loadDb();
db.settings = { ...db.settings, ...newSettings };
await saveDb(db);
return db.settings;
return mutateDb((db, markChanged) => {
db.settings = { ...db.settings, ...newSettings };
markChanged();
return db.settings;
});
},
async reset() {
const db = await loadDb();
db.settings = getDefaultSettings();
await saveDb(db);
return db.settings;
return mutateDb((db, markChanged) => {
db.settings = getDefaultSettings();
markChanged();
return db.settings;
});
}
};
@ -380,83 +429,92 @@ const users = {
},
async create(userData) {
const db = await loadDb();
if (!db.users) {
db.users = [];
}
return mutateDb((db, markChanged) => {
if (!db.users) {
db.users = [];
}
// Check if username already exists
if (db.users.some(u => u.username === userData.username)) {
throw new Error('Username already exists');
}
// Check if username already exists
if (db.users.some(u => u.username === userData.username)) {
throw new Error('Username already exists');
}
const newUser = {
id: db.nextId++,
username: userData.username,
// For OIDC users, passwordHash is optional
passwordHash: userData.passwordHash || null,
role: userData.role || 'viewer',
oidcId: userData.oidcId || null,
email: userData.email || null,
createdAt: new Date().toISOString()
};
const newUser = {
id: db.nextId++,
username: userData.username,
// For OIDC users, passwordHash is optional
passwordHash: userData.passwordHash || null,
role: userData.role || 'viewer',
oidcId: userData.oidcId || null,
email: userData.email || null,
createdAt: new Date().toISOString()
};
db.users.push(newUser);
await saveDb(db);
db.users.push(newUser);
markChanged();
// Return user without password hash
const { passwordHash, ...userWithoutPassword } = newUser;
return userWithoutPassword;
// Return user without password hash
const { passwordHash, ...userWithoutPassword } = newUser;
return userWithoutPassword;
});
},
async update(id, updates) {
const db = await loadDb();
const userIndex = db.users?.findIndex(u => u.id === parseInt(id));
return mutateDb((db, markChanged) => {
const userIndex = db.users?.findIndex(u => u.id === parseInt(id));
if (userIndex === -1 || userIndex === undefined) {
throw new Error('User not found');
}
// Check if username is being changed and if it already exists
if (updates.username && updates.username !== db.users[userIndex].username) {
if (db.users.some(u => u.username === updates.username)) {
throw new Error('Username already exists');
if (userIndex === -1 || userIndex === undefined) {
throw new Error('User not found');
}
}
db.users[userIndex] = {
...db.users[userIndex],
...updates,
updatedAt: new Date().toISOString()
};
// Check if username is being changed and if it already exists
if (updates.username && updates.username !== db.users[userIndex].username) {
if (db.users.some(u => u.username === updates.username)) {
throw new Error('Username already exists');
}
}
await saveDb(db);
if (db.users[userIndex].role === 'admin' && updates.role && updates.role !== 'admin') {
if (db.users.filter(user => user.role === 'admin').length <= 1) {
throw new Error('Cannot remove admin role from the last admin user');
}
}
// Return user without password hash
const { passwordHash, ...userWithoutPassword } = db.users[userIndex];
return userWithoutPassword;
db.users[userIndex] = {
...db.users[userIndex],
...updates,
updatedAt: new Date().toISOString()
};
markChanged();
// Return user without password hash
const { passwordHash, ...userWithoutPassword } = db.users[userIndex];
return userWithoutPassword;
});
},
async delete(id) {
const db = await loadDb();
const userIndex = db.users?.findIndex(u => u.id === parseInt(id));
return mutateDb((db, markChanged) => {
const userIndex = db.users?.findIndex(u => u.id === parseInt(id));
if (userIndex === -1 || userIndex === undefined) {
throw new Error('User not found');
}
// Prevent deleting the last admin
const user = db.users[userIndex];
if (user.role === 'admin') {
const adminCount = db.users.filter(u => u.role === 'admin').length;
if (adminCount <= 1) {
throw new Error('Cannot delete the last admin user');
if (userIndex === -1 || userIndex === undefined) {
throw new Error('User not found');
}
}
db.users.splice(userIndex, 1);
await saveDb(db);
return true;
// Prevent deleting the last admin
const user = db.users[userIndex];
if (user.role === 'admin') {
const adminCount = db.users.filter(u => u.role === 'admin').length;
if (adminCount <= 1) {
throw new Error('Cannot delete the last admin user');
}
}
db.users.splice(userIndex, 1);
markChanged();
return true;
});
},
async count() {