fix: keep background jobs alive
Add a background job orchestrator that restores profile jobs from saved credentials, keeps pickup and store-watch schedules alive without UI logins, and exposes job status via health/admin endpoints.
This commit is contained in:
260
services/backgroundJobOrchestrator.js
Normal file
260
services/backgroundJobOrchestrator.js
Normal file
@@ -0,0 +1,260 @@
|
||||
const credentialStore = require('./credentialStore');
|
||||
const sessionStore = require('./sessionStore');
|
||||
const foodsharingClient = require('./foodsharingClient');
|
||||
const adminConfig = require('./adminConfig');
|
||||
const { readConfig } = require('./configStore');
|
||||
const { readStoreWatch, listWatcherProfiles } = require('./storeWatchStore');
|
||||
const { readStoresCache } = require('./storeCacheStore');
|
||||
const notificationService = require('./notificationService');
|
||||
const { scheduleConfig } = require('./pickupScheduler');
|
||||
|
||||
const ONE_YEAR_MS = 365 * 24 * 60 * 60 * 1000;
|
||||
const WATCHDOG_INTERVAL_MS = 5 * 60 * 1000;
|
||||
const SESSION_ERROR_COOLDOWN_MS = 6 * 60 * 60 * 1000;
|
||||
|
||||
const adminEmail = (process.env.ADMIN_EMAIL || '').toLowerCase();
|
||||
const profileStates = new Map();
|
||||
const sessionErrorNotifications = new Map();
|
||||
|
||||
let watchdogTimer = null;
|
||||
let ensureAllInFlight = null;
|
||||
|
||||
function isAdmin(profile) {
|
||||
return !!adminEmail && !!profile?.email && profile.email.toLowerCase() === adminEmail;
|
||||
}
|
||||
|
||||
function getProfileIds() {
|
||||
const ids = new Set();
|
||||
Object.keys(credentialStore.loadAll()).forEach((profileId) => ids.add(String(profileId)));
|
||||
listWatcherProfiles().forEach((profileId) => ids.add(String(profileId)));
|
||||
return Array.from(ids).sort();
|
||||
}
|
||||
|
||||
function getActivePickupEntries(config) {
|
||||
return Array.isArray(config) ? config.filter((entry) => entry?.active) : [];
|
||||
}
|
||||
|
||||
function createSignature({ config, watchers, settings }) {
|
||||
return JSON.stringify({
|
||||
pickup: Array.isArray(config)
|
||||
? config.map((entry) => ({
|
||||
id: entry?.id ? String(entry.id) : '',
|
||||
active: !!entry?.active,
|
||||
hidden: !!entry?.hidden,
|
||||
onlyNotify: !!entry?.onlyNotify,
|
||||
desiredDate: entry?.desiredDate || null,
|
||||
desiredDateRange: entry?.desiredDateRange || null,
|
||||
desiredWeekday: entry?.desiredWeekday || null,
|
||||
autoDeactivate: entry?.autoDeactivate !== false
|
||||
}))
|
||||
: [],
|
||||
watch: Array.isArray(watchers)
|
||||
? watchers.map((entry) => ({
|
||||
storeId: entry?.storeId ? String(entry.storeId) : '',
|
||||
lastTeamSearchStatus: entry?.lastTeamSearchStatus ?? null
|
||||
}))
|
||||
: [],
|
||||
settings: {
|
||||
pickupFallbackCron: settings?.pickupFallbackCron || null,
|
||||
pickupWindowOffsetsMinutes: settings?.pickupWindowOffsetsMinutes || [],
|
||||
randomDelayMinSeconds: settings?.randomDelayMinSeconds ?? null,
|
||||
randomDelayMaxSeconds: settings?.randomDelayMaxSeconds ?? null,
|
||||
storeWatchCron: settings?.storeWatchCron || null,
|
||||
storeWatchInitialDelayMinSeconds: settings?.storeWatchInitialDelayMinSeconds ?? null,
|
||||
storeWatchInitialDelayMaxSeconds: settings?.storeWatchInitialDelayMaxSeconds ?? null,
|
||||
storeWatchRequestDelayMs: settings?.storeWatchRequestDelayMs ?? null,
|
||||
storeWatchStatusCacheMaxAgeMinutes: settings?.storeWatchStatusCacheMaxAgeMinutes ?? null
|
||||
}
|
||||
});
|
||||
}
|
||||
|
||||
function shouldNotifySessionError(profileId) {
|
||||
const key = profileId ? String(profileId) : null;
|
||||
if (!key) {
|
||||
return false;
|
||||
}
|
||||
const now = Date.now();
|
||||
const last = sessionErrorNotifications.get(key) || 0;
|
||||
if (now - last < SESSION_ERROR_COOLDOWN_MS) {
|
||||
return false;
|
||||
}
|
||||
sessionErrorNotifications.set(key, now);
|
||||
return true;
|
||||
}
|
||||
|
||||
async function notifySessionError(profileId, credentials, error) {
|
||||
if (!shouldNotifySessionError(profileId)) {
|
||||
return;
|
||||
}
|
||||
try {
|
||||
await notificationService.sendAdminSessionErrorNotification({
|
||||
profileId,
|
||||
profileEmail: credentials?.email,
|
||||
error: error?.message,
|
||||
label: 'background-orchestrator'
|
||||
});
|
||||
} catch (notifyError) {
|
||||
console.error('[BACKGROUND] Admin-Session-Fehler konnte nicht gemeldet werden:', notifyError.message);
|
||||
}
|
||||
}
|
||||
|
||||
async function loginProfile(profileId, credentials) {
|
||||
const auth = await foodsharingClient.login(credentials.email, credentials.password);
|
||||
const profile = {
|
||||
id: String(auth.profile.id),
|
||||
name: auth.profile.name,
|
||||
email: auth.profile.email || credentials.email
|
||||
};
|
||||
const session = sessionStore.create(
|
||||
{
|
||||
cookieHeader: auth.cookieHeader,
|
||||
csrfToken: auth.csrfToken,
|
||||
profile,
|
||||
credentials,
|
||||
isAdmin: isAdmin(profile),
|
||||
storesCache: readStoresCache(profile.id)
|
||||
},
|
||||
credentials.token,
|
||||
ONE_YEAR_MS
|
||||
);
|
||||
credentialStore.save(profile.id, {
|
||||
email: credentials.email,
|
||||
password: credentials.password,
|
||||
token: session.id
|
||||
});
|
||||
if (String(profileId) !== profile.id) {
|
||||
console.warn(`[BACKGROUND] Profil ${profileId} wurde als ${profile.id} angemeldet.`);
|
||||
}
|
||||
return session;
|
||||
}
|
||||
|
||||
async function ensureProfile(profileId, options = {}) {
|
||||
const key = profileId ? String(profileId) : null;
|
||||
if (!key) {
|
||||
return null;
|
||||
}
|
||||
const credentials = credentialStore.get(key);
|
||||
const previous = profileStates.get(key) || {};
|
||||
|
||||
if (!credentials?.email || !credentials?.password) {
|
||||
const state = {
|
||||
...previous,
|
||||
profileId: key,
|
||||
active: false,
|
||||
lastError: 'missing-credentials',
|
||||
checkedAt: Date.now()
|
||||
};
|
||||
profileStates.set(key, state);
|
||||
return state;
|
||||
}
|
||||
|
||||
try {
|
||||
let session = sessionStore.getByProfile(key) || (credentials.token ? sessionStore.get(credentials.token) : null);
|
||||
if (!session?.cookieHeader) {
|
||||
session = await loginProfile(key, credentials);
|
||||
console.log(`[BACKGROUND] Session fuer Profil ${session.profile.id} (${session.profile.name}) bereitgestellt.`);
|
||||
} else {
|
||||
sessionStore.update(session.id, { credentials });
|
||||
session = sessionStore.get(session.id);
|
||||
}
|
||||
|
||||
const profileIdForData = String(session.profile.id);
|
||||
const config = Array.isArray(options.config) ? options.config : readConfig(profileIdForData);
|
||||
const watchers = readStoreWatch(profileIdForData);
|
||||
const settings = options.settings || adminConfig.readSettings();
|
||||
const signature = createSignature({ config, watchers, settings });
|
||||
const hasJobs = Array.isArray(session.jobs) && session.jobs.length > 0;
|
||||
const shouldSchedule = options.force || previous.signature !== signature || !hasJobs;
|
||||
|
||||
if (shouldSchedule) {
|
||||
scheduleConfig(session.id, config, settings);
|
||||
console.log(
|
||||
`[BACKGROUND] Jobs fuer Profil ${profileIdForData} sichergestellt ` +
|
||||
`(${getActivePickupEntries(config).length} aktive Pickup-Eintraege, ${watchers.length} Watch-Stores).`
|
||||
);
|
||||
}
|
||||
|
||||
const state = {
|
||||
profileId: profileIdForData,
|
||||
sessionId: session.id,
|
||||
active: true,
|
||||
signature,
|
||||
checkedAt: Date.now(),
|
||||
lastSuccessAt: Date.now(),
|
||||
lastError: null,
|
||||
activePickupCount: getActivePickupEntries(config).length,
|
||||
watchedStoreCount: watchers.length,
|
||||
jobCount: sessionStore.get(session.id)?.jobs?.length || 0
|
||||
};
|
||||
profileStates.set(profileIdForData, state);
|
||||
return state;
|
||||
} catch (error) {
|
||||
console.error(`[BACKGROUND] Jobs fuer Profil ${key} konnten nicht sichergestellt werden:`, error.message);
|
||||
await notifySessionError(key, credentials, error);
|
||||
const state = {
|
||||
...previous,
|
||||
profileId: key,
|
||||
active: false,
|
||||
checkedAt: Date.now(),
|
||||
lastError: error.message || 'unknown-error'
|
||||
};
|
||||
profileStates.set(key, state);
|
||||
return state;
|
||||
}
|
||||
}
|
||||
|
||||
async function ensureAll(options = {}) {
|
||||
if (ensureAllInFlight) {
|
||||
return ensureAllInFlight;
|
||||
}
|
||||
ensureAllInFlight = (async () => {
|
||||
const profileIds = getProfileIds();
|
||||
const results = [];
|
||||
for (const profileId of profileIds) {
|
||||
results.push(await ensureProfile(profileId, options));
|
||||
}
|
||||
return results;
|
||||
})();
|
||||
try {
|
||||
return await ensureAllInFlight;
|
||||
} finally {
|
||||
ensureAllInFlight = null;
|
||||
}
|
||||
}
|
||||
|
||||
function start(options = {}) {
|
||||
if (watchdogTimer) {
|
||||
return;
|
||||
}
|
||||
const intervalMs = Number(options.intervalMs) || WATCHDOG_INTERVAL_MS;
|
||||
ensureAll({ force: true }).catch((error) => {
|
||||
console.error('[BACKGROUND] Initiale Job-Pruefung fehlgeschlagen:', error.message);
|
||||
});
|
||||
watchdogTimer = setInterval(() => {
|
||||
ensureAll().catch((error) => {
|
||||
console.error('[BACKGROUND] Watchdog-Pruefung fehlgeschlagen:', error.message);
|
||||
});
|
||||
}, intervalMs);
|
||||
}
|
||||
|
||||
function refreshProfile(profileId, options = {}) {
|
||||
return ensureProfile(profileId, { ...options, force: true });
|
||||
}
|
||||
|
||||
function refreshAll(options = {}) {
|
||||
return ensureAll({ ...options, force: true });
|
||||
}
|
||||
|
||||
function getStatus() {
|
||||
return {
|
||||
running: !!watchdogTimer,
|
||||
profiles: Array.from(profileStates.values()).map((state) => ({ ...state }))
|
||||
};
|
||||
}
|
||||
|
||||
module.exports = {
|
||||
start,
|
||||
refreshProfile,
|
||||
refreshAll,
|
||||
getStatus
|
||||
};
|
||||
@@ -1122,7 +1122,7 @@ function scheduleStoreWatchers(sessionId, settings) {
|
||||
return false;
|
||||
}
|
||||
const cronExpression = effectiveSettings.storeWatchCron || DEFAULT_SETTINGS.storeWatchCron;
|
||||
const job = cron.schedule(
|
||||
const cronJob = cron.schedule(
|
||||
cronExpression,
|
||||
() => {
|
||||
checkWatchedStores(sessionId, effectiveSettings).catch((error) => {
|
||||
@@ -1131,14 +1131,22 @@ function scheduleStoreWatchers(sessionId, settings) {
|
||||
},
|
||||
{ timezone: 'Europe/Berlin' }
|
||||
);
|
||||
sessionStore.attachJob(sessionId, job);
|
||||
setTimeout(
|
||||
let initialTimeout = setTimeout(
|
||||
() => checkWatchedStores(sessionId, effectiveSettings),
|
||||
randomDelayMs(
|
||||
effectiveSettings.storeWatchInitialDelayMinSeconds,
|
||||
effectiveSettings.storeWatchInitialDelayMaxSeconds
|
||||
)
|
||||
);
|
||||
sessionStore.attachJob(sessionId, {
|
||||
stop: () => {
|
||||
cronJob.stop();
|
||||
if (initialTimeout) {
|
||||
clearTimeout(initialTimeout);
|
||||
initialTimeout = null;
|
||||
}
|
||||
}
|
||||
});
|
||||
console.log(
|
||||
`[WATCH] Überwache ${watchers.length} Betriebe für Session ${sessionId} (Cron: ${cronExpression}).`
|
||||
);
|
||||
|
||||
@@ -48,6 +48,15 @@ class SessionStore {
|
||||
return session;
|
||||
}
|
||||
|
||||
getByProfile(profileId) {
|
||||
const key = profileId ? String(profileId) : null;
|
||||
if (!key) {
|
||||
return null;
|
||||
}
|
||||
const id = this.profileIndex.get(key);
|
||||
return id ? this.get(id) : null;
|
||||
}
|
||||
|
||||
update(id, patch) {
|
||||
const session = this.get(id);
|
||||
if (!session) {
|
||||
|
||||
Reference in New Issue
Block a user