chore: sync running docker state
Commit the app code currently running in Docker, including scheduler, store status, journal reminder, maintenance mode, and Foodsharing API check updates. Remove generated runtime logs from version control and ignore local runtime captures.
This commit is contained in:
421
services/storeMessageWatcher.js
Normal file
421
services/storeMessageWatcher.js
Normal file
@@ -0,0 +1,421 @@
|
||||
const { io } = require('socket.io-client');
|
||||
const cron = require('node-cron');
|
||||
const foodsharingClient = require('./foodsharingClient');
|
||||
const requestLogStore = require('./requestLogStore');
|
||||
const sessionStore = require('./sessionStore');
|
||||
|
||||
const FOODSHARING_URL = 'https://foodsharing.de';
|
||||
const DEFAULT_FALLBACK_CRON = '0 6-23 * * *';
|
||||
const STORE_DEBOUNCE_MS = 2 * 60 * 1000;
|
||||
|
||||
function extractStoreIdFromHref(href = '') {
|
||||
if (typeof href !== 'string' || !href) {
|
||||
return null;
|
||||
}
|
||||
|
||||
const directMatch = href.match(/\/store\/(\d+)/);
|
||||
if (directMatch) {
|
||||
return directMatch[1];
|
||||
}
|
||||
|
||||
const queryMatch = href.match(/[?&]id=(\d+)/);
|
||||
if (queryMatch && href.includes('fsbetrieb')) {
|
||||
return queryMatch[1];
|
||||
}
|
||||
|
||||
return null;
|
||||
}
|
||||
|
||||
function getMatchingActiveEntries(config, storeId) {
|
||||
if (!Array.isArray(config) || !storeId) {
|
||||
return [];
|
||||
}
|
||||
return config.filter((entry) => entry?.active && String(entry.id) === String(storeId));
|
||||
}
|
||||
|
||||
function summarizePayload(payload, depth = 0) {
|
||||
if (payload === null || payload === undefined) {
|
||||
return payload;
|
||||
}
|
||||
if (typeof payload !== 'object') {
|
||||
return typeof payload === 'string' && payload.length > 240 ? `${payload.slice(0, 240)}…` : payload;
|
||||
}
|
||||
if (depth >= 2) {
|
||||
return Array.isArray(payload) ? `[array:${payload.length}]` : '[object]';
|
||||
}
|
||||
if (Array.isArray(payload)) {
|
||||
return payload.slice(0, 5).map((entry) => summarizePayload(entry, depth + 1));
|
||||
}
|
||||
return Object.fromEntries(
|
||||
Object.entries(payload)
|
||||
.slice(0, 20)
|
||||
.map(([key, value]) => [key, summarizePayload(value, depth + 1)])
|
||||
);
|
||||
}
|
||||
|
||||
function sortConversationsAscending(conversations = []) {
|
||||
return conversations.slice().sort((a, b) => {
|
||||
const left = new Date(a?.lastMessage?.sentAt || 0).getTime() || 0;
|
||||
const right = new Date(b?.lastMessage?.sentAt || 0).getTime() || 0;
|
||||
if (left !== right) {
|
||||
return left - right;
|
||||
}
|
||||
return Number(a?.lastMessage?.id || 0) - Number(b?.lastMessage?.id || 0);
|
||||
});
|
||||
}
|
||||
|
||||
function collectStoreIds(value, ids = new Set(), depth = 0) {
|
||||
if (value === null || value === undefined || depth > 5) {
|
||||
return ids;
|
||||
}
|
||||
if (typeof value === 'string') {
|
||||
const hrefId = extractStoreIdFromHref(value);
|
||||
if (hrefId) {
|
||||
ids.add(hrefId);
|
||||
}
|
||||
return ids;
|
||||
}
|
||||
if (typeof value !== 'object') {
|
||||
return ids;
|
||||
}
|
||||
if (Array.isArray(value)) {
|
||||
value.forEach((entry) => collectStoreIds(entry, ids, depth + 1));
|
||||
return ids;
|
||||
}
|
||||
|
||||
const directStoreId = value.storeId ?? value.store_id ?? value.store?.id ?? value.store?.storeId;
|
||||
if (directStoreId) {
|
||||
ids.add(String(directStoreId));
|
||||
}
|
||||
if (value.href || value.url || value.link) {
|
||||
collectStoreIds(value.href || value.url || value.link, ids, depth + 1);
|
||||
}
|
||||
Object.values(value).forEach((entry) => collectStoreIds(entry, ids, depth + 1));
|
||||
return ids;
|
||||
}
|
||||
|
||||
function getMessageWatcherSettings(settings = {}) {
|
||||
const fallbackCron = settings.messageWatcherFallbackCron || DEFAULT_FALLBACK_CRON;
|
||||
return {
|
||||
socketEnabled: settings.messageWatcherSocketEnabled !== false,
|
||||
fallbackEnabled: settings.messageWatcherFallbackEnabled !== false,
|
||||
fallbackCron: cron.validate(fallbackCron) ? fallbackCron : DEFAULT_FALLBACK_CRON,
|
||||
debugLogging: settings.messageWatcherDebugLogging !== false
|
||||
};
|
||||
}
|
||||
|
||||
function isPotentialMessageSocketEvent(eventName = '') {
|
||||
return /message|conversation|bell|notification|wall|post/i.test(String(eventName));
|
||||
}
|
||||
|
||||
function startStoreMessageWatcher({ sessionId, config, settings, runPickupCheck }) {
|
||||
const watcherSettings = getMessageWatcherSettings(settings);
|
||||
let stopped = false;
|
||||
let fallbackJob = null;
|
||||
let socket = null;
|
||||
let socketConnected = false;
|
||||
let conversationsBaselineReady = false;
|
||||
const lastMessageByConversation = new Map();
|
||||
const lastTriggerByStore = new Map();
|
||||
const inFlightStores = new Set();
|
||||
|
||||
function logWatcherEvent(type, data = {}, options = {}) {
|
||||
const session = sessionStore.get(sessionId);
|
||||
const payload = {
|
||||
event: type,
|
||||
...data
|
||||
};
|
||||
const consoleMessage = `[MESSAGES] ${type}: ${JSON.stringify(payload)}`;
|
||||
const level = options.level || 'info';
|
||||
if (level === 'warn') {
|
||||
console.warn(consoleMessage);
|
||||
} else if (level === 'error') {
|
||||
console.error(consoleMessage);
|
||||
} else {
|
||||
console.log(consoleMessage);
|
||||
}
|
||||
|
||||
if (!watcherSettings.debugLogging && !options.forceLog) {
|
||||
return;
|
||||
}
|
||||
try {
|
||||
requestLogStore.add({
|
||||
direction: 'incoming',
|
||||
target: 'foodsharing.de',
|
||||
method: 'MESSAGE_WATCHER',
|
||||
path: type,
|
||||
status: null,
|
||||
sessionId,
|
||||
profileId: session?.profile?.id ?? null,
|
||||
profileName: session?.profile?.name ?? null,
|
||||
responseBody: payload
|
||||
});
|
||||
} catch (error) {
|
||||
console.warn('[MESSAGES] Watcher-Log konnte nicht geschrieben werden:', error.message);
|
||||
}
|
||||
}
|
||||
|
||||
async function triggerStoreCheck(storeId, source = {}) {
|
||||
const matchingEntries = getMatchingActiveEntries(config, storeId);
|
||||
if (matchingEntries.length === 0) {
|
||||
logWatcherEvent('trigger-skipped-no-active-entry', { storeId, source: source.type });
|
||||
return;
|
||||
}
|
||||
|
||||
const now = Date.now();
|
||||
const lastTrigger = lastTriggerByStore.get(storeId) || 0;
|
||||
if (now - lastTrigger < STORE_DEBOUNCE_MS || inFlightStores.has(storeId)) {
|
||||
logWatcherEvent('trigger-skipped-debounce', { storeId, source: source.type });
|
||||
return;
|
||||
}
|
||||
|
||||
lastTriggerByStore.set(storeId, now);
|
||||
inFlightStores.add(storeId);
|
||||
const storeName = source.storeName || matchingEntries[0]?.name || `Store ${storeId}`;
|
||||
logWatcherEvent(
|
||||
'store-message-pickup-check',
|
||||
{
|
||||
storeId,
|
||||
storeName,
|
||||
checkedEntries: matchingEntries.length,
|
||||
source
|
||||
},
|
||||
{ forceLog: true }
|
||||
);
|
||||
|
||||
try {
|
||||
await runPickupCheck(sessionId, matchingEntries, settings);
|
||||
logWatcherEvent('store-message-pickup-check-complete', {
|
||||
storeId,
|
||||
checkedEntries: matchingEntries.length,
|
||||
source: source.type
|
||||
});
|
||||
} catch (error) {
|
||||
logWatcherEvent(
|
||||
'store-message-pickup-check-error',
|
||||
{ storeId, source: source.type, error: error.message },
|
||||
{ level: 'error', forceLog: true }
|
||||
);
|
||||
} finally {
|
||||
inFlightStores.delete(storeId);
|
||||
}
|
||||
}
|
||||
|
||||
async function processSocketEvent(eventName, args = []) {
|
||||
const summarizedArgs = summarizePayload(args);
|
||||
const storeIds = Array.from(collectStoreIds(args));
|
||||
logWatcherEvent('socket-event', {
|
||||
eventName,
|
||||
storeIds,
|
||||
payload: summarizedArgs
|
||||
});
|
||||
|
||||
if (!isPotentialMessageSocketEvent(eventName)) {
|
||||
return;
|
||||
}
|
||||
|
||||
for (const storeId of storeIds) {
|
||||
await triggerStoreCheck(storeId, {
|
||||
type: 'socket',
|
||||
eventName
|
||||
});
|
||||
}
|
||||
}
|
||||
|
||||
async function pollConversations(reason) {
|
||||
const session = sessionStore.get(sessionId);
|
||||
if (!session?.cookieHeader) {
|
||||
return;
|
||||
}
|
||||
|
||||
const { conversations } = await foodsharingClient.fetchConversations(
|
||||
session.cookieHeader,
|
||||
{ limit: 20, offset: 0 },
|
||||
session
|
||||
);
|
||||
const sorted = sortConversationsAscending(conversations);
|
||||
let newMessages = 0;
|
||||
|
||||
if (!conversationsBaselineReady) {
|
||||
sorted.forEach((conversation) => {
|
||||
if (conversation?.id && conversation?.lastMessage?.id) {
|
||||
lastMessageByConversation.set(String(conversation.id), String(conversation.lastMessage.id));
|
||||
}
|
||||
});
|
||||
conversationsBaselineReady = true;
|
||||
logWatcherEvent('fallback-conversations-baseline', {
|
||||
reason,
|
||||
conversations: sorted.length
|
||||
});
|
||||
return;
|
||||
}
|
||||
|
||||
for (const conversation of sorted) {
|
||||
const conversationId = conversation?.id ? String(conversation.id) : null;
|
||||
const lastMessageId = conversation?.lastMessage?.id ? String(conversation.lastMessage.id) : null;
|
||||
if (!conversationId || !lastMessageId) {
|
||||
continue;
|
||||
}
|
||||
const previousMessageId = lastMessageByConversation.get(conversationId);
|
||||
lastMessageByConversation.set(conversationId, lastMessageId);
|
||||
if (!previousMessageId || previousMessageId === lastMessageId) {
|
||||
continue;
|
||||
}
|
||||
newMessages += 1;
|
||||
if (!conversation.storeId) {
|
||||
logWatcherEvent('conversation-message-without-store', {
|
||||
conversationId,
|
||||
lastMessageId
|
||||
});
|
||||
continue;
|
||||
}
|
||||
await triggerStoreCheck(String(conversation.storeId), {
|
||||
type: 'conversation-fallback',
|
||||
conversationId,
|
||||
messageId: lastMessageId,
|
||||
storeName: conversation.title || null,
|
||||
sentAt: conversation.lastMessage?.sentAt || null,
|
||||
body: summarizePayload(conversation.lastMessage?.body || '')
|
||||
});
|
||||
}
|
||||
|
||||
logWatcherEvent('fallback-conversations-polled', {
|
||||
reason,
|
||||
conversations: sorted.length,
|
||||
newMessages
|
||||
});
|
||||
}
|
||||
|
||||
async function runFallbackPoll(reason) {
|
||||
if (stopped || !watcherSettings.fallbackEnabled) {
|
||||
return;
|
||||
}
|
||||
try {
|
||||
await pollConversations(reason);
|
||||
} catch (error) {
|
||||
logWatcherEvent(
|
||||
'fallback-poll-error',
|
||||
{ reason, error: error.message },
|
||||
{ level: 'warn', forceLog: true }
|
||||
);
|
||||
}
|
||||
}
|
||||
|
||||
function scheduleFallbackPoll() {
|
||||
if (stopped || !watcherSettings.fallbackEnabled) {
|
||||
return;
|
||||
}
|
||||
fallbackJob = cron.schedule(watcherSettings.fallbackCron, async () => {
|
||||
if (!socketConnected) {
|
||||
await runFallbackPoll('socket-disconnected');
|
||||
} else {
|
||||
logWatcherEvent('fallback-poll-skipped-socket-connected', {
|
||||
cron: watcherSettings.fallbackCron
|
||||
});
|
||||
}
|
||||
});
|
||||
}
|
||||
|
||||
function startSocket() {
|
||||
if (!watcherSettings.socketEnabled) {
|
||||
logWatcherEvent('socket-disabled');
|
||||
return;
|
||||
}
|
||||
const session = sessionStore.get(sessionId);
|
||||
if (!session?.cookieHeader) {
|
||||
logWatcherEvent('socket-skipped-missing-cookie', {}, { level: 'warn', forceLog: true });
|
||||
return;
|
||||
}
|
||||
|
||||
socket = io(FOODSHARING_URL, {
|
||||
path: '/websocket/socket.io',
|
||||
transports: ['websocket', 'polling'],
|
||||
extraHeaders: {
|
||||
Cookie: session.cookieHeader,
|
||||
Origin: FOODSHARING_URL,
|
||||
Referer: `${FOODSHARING_URL}/dashboard`
|
||||
},
|
||||
reconnection: true,
|
||||
reconnectionDelay: 10000,
|
||||
reconnectionDelayMax: 60000,
|
||||
timeout: 20000,
|
||||
forceNew: true
|
||||
});
|
||||
|
||||
socket.on('connect', () => {
|
||||
socketConnected = true;
|
||||
logWatcherEvent('socket-connected', {
|
||||
socketId: socket.id,
|
||||
transport: socket.io?.engine?.transport?.name || null
|
||||
}, { forceLog: true });
|
||||
});
|
||||
|
||||
socket.on('disconnect', (reason) => {
|
||||
socketConnected = false;
|
||||
logWatcherEvent('socket-disconnected', { reason }, { level: 'warn', forceLog: true });
|
||||
});
|
||||
|
||||
socket.on('connect_error', (error) => {
|
||||
socketConnected = false;
|
||||
logWatcherEvent(
|
||||
'socket-connect-error',
|
||||
{ error: error.message },
|
||||
{ level: 'warn', forceLog: true }
|
||||
);
|
||||
});
|
||||
|
||||
socket.io.on('reconnect_attempt', (attempt) => {
|
||||
logWatcherEvent('socket-reconnect-attempt', { attempt });
|
||||
});
|
||||
|
||||
socket.io.on('upgrade', (transport) => {
|
||||
logWatcherEvent('socket-transport-upgrade', {
|
||||
transport: transport?.name || null
|
||||
});
|
||||
});
|
||||
|
||||
socket.onAny((eventName, ...args) => {
|
||||
processSocketEvent(eventName, args).catch((error) => {
|
||||
logWatcherEvent(
|
||||
'socket-event-error',
|
||||
{ eventName, error: error.message },
|
||||
{ level: 'error', forceLog: true }
|
||||
);
|
||||
});
|
||||
});
|
||||
}
|
||||
|
||||
logWatcherEvent('watcher-started', {
|
||||
socketEnabled: watcherSettings.socketEnabled,
|
||||
fallbackEnabled: watcherSettings.fallbackEnabled,
|
||||
fallbackCron: watcherSettings.fallbackCron,
|
||||
debugLogging: watcherSettings.debugLogging,
|
||||
activeEntries: Array.isArray(config) ? config.filter((entry) => entry?.active).length : 0
|
||||
}, { forceLog: true });
|
||||
startSocket();
|
||||
if (watcherSettings.fallbackEnabled) {
|
||||
runFallbackPoll('initial-baseline').finally(scheduleFallbackPoll);
|
||||
}
|
||||
|
||||
return {
|
||||
stop() {
|
||||
stopped = true;
|
||||
if (fallbackJob) {
|
||||
fallbackJob.stop();
|
||||
}
|
||||
if (socket) {
|
||||
socket.disconnect();
|
||||
}
|
||||
logWatcherEvent('watcher-stopped', {}, { forceLog: true });
|
||||
}
|
||||
};
|
||||
}
|
||||
|
||||
module.exports = {
|
||||
collectStoreIds,
|
||||
extractStoreIdFromHref,
|
||||
getMatchingActiveEntries,
|
||||
getMessageWatcherSettings,
|
||||
isPotentialMessageSocketEvent,
|
||||
startStoreMessageWatcher
|
||||
};
|
||||
Reference in New Issue
Block a user