token-consumer / storage.js
qwen2api's picture
Upload 11 files
e5f68c9 verified
Raw
History Blame Contribute Delete
4.35 kB
const fs = require('fs');
const path = require('path');
const DATA_DIR = path.join(__dirname, 'data');
const TASKS_FILE = path.join(DATA_DIR, 'tasks.json');
const CONFIG_FILE = path.join(DATA_DIR, 'config.json');
// In-memory cache to avoid read-modify-write race conditions
let tasksCache = null;
// Write throttling
let writePending = false;
let writeTimer = null;
function ensureDir() {
if (!fs.existsSync(DATA_DIR)) {
fs.mkdirSync(DATA_DIR, { recursive: true });
}
}
// Load tasks into memory cache (lazy load)
function loadTasksCache() {
if (tasksCache === null) {
try {
if (fs.existsSync(TASKS_FILE)) {
const content = fs.readFileSync(TASKS_FILE, 'utf-8');
const data = JSON.parse(content);
tasksCache = data.tasks || {};
} else {
tasksCache = {};
}
} catch (e) {
console.error('Error loading tasks:', e.message);
tasksCache = {};
}
}
return tasksCache;
}
// Throttled write - batch writes within 100ms
function scheduleWrite() {
if (!writePending) {
writePending = true;
if (writeTimer) {
clearTimeout(writeTimer);
}
writeTimer = setTimeout(() => {
writePending = false;
writeTimer = null;
try {
ensureDir();
fs.writeFileSync(TASKS_FILE, JSON.stringify({ tasks: tasksCache }, null, 2), 'utf-8');
} catch (e) {
console.error('Write error:', e.message);
}
}, 100);
}
}
// Immediate write (for critical operations)
function flushWrite() {
if (writeTimer) {
clearTimeout(writeTimer);
writeTimer = null;
}
if (tasksCache !== null) {
try {
ensureDir();
fs.writeFileSync(TASKS_FILE, JSON.stringify({ tasks: tasksCache }, null, 2), 'utf-8');
} catch (e) {
console.error('Flush write error:', e.message);
}
}
writePending = false;
}
// Read JSON helper for other files
function readJSON(file, defaultValue = {}) {
try {
if (fs.existsSync(file)) {
const content = fs.readFileSync(file, 'utf-8');
return JSON.parse(content);
}
} catch (e) {
console.error(`Error reading ${file}:`, e.message);
}
return defaultValue;
}
// Task persistence - all operations use in-memory cache
function loadTasks() {
return { tasks: loadTasksCache() };
}
function saveTasks(tasks) {
tasksCache = tasks;
scheduleWrite();
}
function getTask(taskId) {
const cache = loadTasksCache();
return cache[taskId] || null;
}
function saveTask(taskId, taskData) {
const cache = loadTasksCache();
cache[taskId] = taskData;
scheduleWrite();
}
function deleteTask(taskId) {
const cache = loadTasksCache();
delete cache[taskId];
scheduleWrite();
}
function getAllTasks() {
return loadTasksCache();
}
// Config persistence
function loadConfig() {
return readJSON(CONFIG_FILE, {});
}
function saveConfig(config) {
ensureDir();
fs.writeFileSync(CONFIG_FILE, JSON.stringify(config, null, 2), 'utf-8');
}
// Raw tasks.json operations (for config editor)
function getTasksRaw() {
flushWrite(); // Ensure latest data is written
return readJSON(TASKS_FILE, { tasks: {} });
}
function saveTasksRaw(content) {
// Parse and validate
const data = JSON.parse(content);
// Update cache and write immediately
tasksCache = data.tasks || {};
flushWrite();
}
// Check if any task is running
function hasRunningTasks() {
const tasks = getAllTasks();
return Object.values(tasks).some(t => t.status === 'running' || t.status === 'paused');
}
// Check if there's pending data to write
function hasPendingWrite() {
return writePending || writeTimer !== null;
}
// Graceful shutdown - flush all pending data
function shutdown() {
if (writeTimer) {
clearTimeout(writeTimer);
writeTimer = null;
}
if (tasksCache !== null && writePending) {
try {
ensureDir();
fs.writeFileSync(TASKS_FILE, JSON.stringify({ tasks: tasksCache }, null, 2), 'utf-8');
console.log('[Storage] Data flushed on shutdown');
} catch (e) {
console.error('[Storage] Shutdown flush error:', e.message);
}
}
writePending = false;
}
module.exports = {
loadTasks,
saveTasks,
getTask,
saveTask,
deleteTask,
getAllTasks,
loadConfig,
saveConfig,
getTasksRaw,
saveTasksRaw,
hasRunningTasks,
flushTasks: flushWrite,
hasPendingWrite,
shutdown
};