matthias/dashboard
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

update backend stuck/stall

Browse Source
This commit is contained in:
Matthias Hochmeister
2026-03-13 08:30:05 +01:00
parent 60488309ca
commit 243da302c7
9 changed files with 199 additions and 156 deletions
Download Patch File Download Diff File Expand all files Collapse all files

283
backend/src/jobs/notification-generation.job.ts
View File

@@ -5,6 +5,7 @@
* 1. Personal atemschutz warnings (untersuchung / leistungstest expiring within 60 days)
* 2. Vehicle issues (for fahrmeister users)
* 3. Equipment issues (for fahrmeister if motorised, zeugmeister if not)
* 4. Nextcloud Talk unread messages
*
* Deduplicates via the unique index on (user_id, quell_typ, quell_id) WHERE NOT gelesen.
* Also cleans up read notifications older than 90 days.
@@ -19,12 +20,18 @@ const INTERVAL_MS = 15 * 60 * 1000; // 15 minutes
const ATEMSCHUTZ_THRESHOLD = 60; // days
let jobInterval: ReturnType<typeof setInterval> | null = null;
let isRunning = false;
// ---------------------------------------------------------------------------
// Core generation function
// ---------------------------------------------------------------------------
export async function runNotificationGeneration(): Promise<void> {
if (isRunning) {
logger.warn('NotificationGenerationJob: previous run still in progress — skipping');
return;
}
isRunning = true;
try {
await generateAtemschutzNotifications();
await generateVehicleNotifications();
@@ -35,6 +42,8 @@ export async function runNotificationGeneration(): Promise<void> {
logger.error('NotificationGenerationJob: unexpected error', {
error: error instanceof Error ? error.message : String(error),
});
} finally {
isRunning = false;
}
}
@@ -100,146 +109,133 @@ async function generateAtemschutzNotifications(): Promise<void> {
}
// ---------------------------------------------------------------------------
// 2. Vehicle issues → fahrmeister users
// 2. Vehicle issues → fahrmeister users (bulk INSERT)
// ---------------------------------------------------------------------------
async function generateVehicleNotifications(): Promise<void> {
try {
// Find vehicles with problems (damaged or not operational, or overdue inspection)
const vehiclesResult = await pool.query(`
SELECT id, bezeichnung, kurzname, status, naechste_pruefung_tage
FROM fahrzeuge
WHERE deleted_at IS NULL
AND (
status IN ('beschaedigt', 'ausser_dienst')
OR (naechste_pruefung_tage IS NOT NULL AND naechste_pruefung_tage::int < 0)
)
await pool.query(`
INSERT INTO notifications (user_id, typ, titel, nachricht, schwere, link, quell_id, quell_typ)
SELECT
u.id,
'fahrzeug_status',
'Fahrzeug nicht einsatzbereit',
CASE WHEN f.kurzname IS NOT NULL
THEN f.bezeichnung || ' (' || f.kurzname || ') hat den Status "' || f.status || '" und ist nicht einsatzbereit.'
ELSE f.bezeichnung || ' hat den Status "' || f.status || '" und ist nicht einsatzbereit.'
END,
'fehler',
'/fahrzeuge/' || f.id::text,
'fahrzeug-status-' || f.id::text,
'fahrzeug_status'
FROM fahrzeuge f
CROSS JOIN users u
WHERE f.deleted_at IS NULL
AND f.status IN ('beschaedigt', 'ausser_dienst')
AND u.is_active = TRUE
AND 'dashboard_fahrmeister' = ANY(u.authentik_groups)
ON CONFLICT (user_id, quell_typ, quell_id)
WHERE NOT gelesen AND quell_typ IS NOT NULL AND quell_id IS NOT NULL
DO NOTHING
`);
if (vehiclesResult.rows.length === 0) return;
// Get all fahrmeister users
const usersResult = await pool.query(`
SELECT id FROM users WHERE is_active = TRUE AND 'dashboard_fahrmeister' = ANY(authentik_groups)
await pool.query(`
INSERT INTO notifications (user_id, typ, titel, nachricht, schwere, link, quell_id, quell_typ)
SELECT
u.id,
'fahrzeug_pruefung',
'Fahrzeugprüfung überfällig',
CASE WHEN f.kurzname IS NOT NULL
THEN 'Die Prüfung von ' || f.bezeichnung || ' (' || f.kurzname || ') ist seit ' || ABS(f.naechste_pruefung_tage::int) || ' Tagen überfällig.'
ELSE 'Die Prüfung von ' || f.bezeichnung || ' ist seit ' || ABS(f.naechste_pruefung_tage::int) || ' Tagen überfällig.'
END,
'fehler',
'/fahrzeuge/' || f.id::text,
'fahrzeug-pruefung-' || f.id::text,
'fahrzeug_pruefung'
FROM fahrzeuge f
CROSS JOIN users u
WHERE f.deleted_at IS NULL
AND f.naechste_pruefung_tage IS NOT NULL
AND f.naechste_pruefung_tage::int < 0
AND u.is_active = TRUE
AND 'dashboard_fahrmeister' = ANY(u.authentik_groups)
ON CONFLICT (user_id, quell_typ, quell_id)
WHERE NOT gelesen AND quell_typ IS NOT NULL AND quell_id IS NOT NULL
DO NOTHING
`);
for (const user of usersResult.rows) {
for (const vehicle of vehiclesResult.rows) {
const label = vehicle.kurzname ? `${vehicle.bezeichnung} (${vehicle.kurzname})` : vehicle.bezeichnung;
const isOverdueInspection = vehicle.naechste_pruefung_tage != null && parseInt(vehicle.naechste_pruefung_tage, 10) < 0;
const isBroken = ['beschaedigt', 'ausser_dienst'].includes(vehicle.status);
if (isBroken) {
await notificationService.createNotification({
user_id: user.id,
typ: 'fahrzeug_status',
titel: `Fahrzeug nicht einsatzbereit`,
nachricht: `${label} hat den Status "${vehicle.status}" und ist nicht einsatzbereit.`,
schwere: 'fehler',
link: `/fahrzeuge/${vehicle.id}`,
quell_id: `fahrzeug-status-${vehicle.id}`,
quell_typ: 'fahrzeug_status',
});
}
if (isOverdueInspection) {
const tage = Math.abs(parseInt(vehicle.naechste_pruefung_tage, 10));
await notificationService.createNotification({
user_id: user.id,
typ: 'fahrzeug_pruefung',
titel: `Fahrzeugprüfung überfällig`,
nachricht: `Die Prüfung von ${label} ist seit ${tage} Tagen überfällig.`,
schwere: 'fehler',
link: `/fahrzeuge/${vehicle.id}`,
quell_id: `fahrzeug-pruefung-${vehicle.id}`,
quell_typ: 'fahrzeug_pruefung',
});
}
}
}
} catch (error) {
logger.error('NotificationGenerationJob: generateVehicleNotifications failed', { error });
}
}
// ---------------------------------------------------------------------------
// 3. Equipment issues → fahrmeister (motorised) or zeugmeister (non-motorised)
// 3. Equipment issues → fahrmeister (motorised) or zeugmeister (bulk INSERT)
// ---------------------------------------------------------------------------
async function generateEquipmentNotifications(): Promise<void> {
try {
// Find equipment with problems (broken, overdue inspection)
const equipmentResult = await pool.query(`
await pool.query(`
INSERT INTO notifications (user_id, typ, titel, nachricht, schwere, link, quell_id, quell_typ)
SELECT
a.id, a.bezeichnung, a.status,
k.motorisiert,
(a.naechste_pruefung_am::date - CURRENT_DATE) AS pruefung_tage
u.id,
'ausruestung_status',
'Ausrüstung nicht einsatzbereit',
a.bezeichnung || ' hat den Status "' || a.status || '" und ist nicht einsatzbereit.',
'fehler',
'/ausruestung/' || a.id::text,
'ausruestung-status-' || a.id::text,
'ausruestung_status'
FROM ausruestung a
JOIN ausruestung_kategorien k ON k.id = a.kategorie_id
JOIN users u ON u.is_active = TRUE AND (
(k.motorisiert = TRUE AND 'dashboard_fahrmeister' = ANY(u.authentik_groups))
OR
(k.motorisiert = FALSE AND 'dashboard_zeugmeister' = ANY(u.authentik_groups))
)
WHERE a.deleted_at IS NULL
AND (
a.status IN ('beschaedigt', 'ausser_dienst')
OR (a.naechste_pruefung_am IS NOT NULL AND a.naechste_pruefung_am::date < CURRENT_DATE)
)
AND a.status IN ('beschaedigt', 'ausser_dienst')
ON CONFLICT (user_id, quell_typ, quell_id)
WHERE NOT gelesen AND quell_typ IS NOT NULL AND quell_id IS NOT NULL
DO NOTHING
`);
if (equipmentResult.rows.length === 0) return;
// Get fahrmeister and zeugmeister users
const [fahrResult, zeugResult] = await Promise.all([
pool.query(`SELECT id FROM users WHERE is_active = TRUE AND 'dashboard_fahrmeister' = ANY(authentik_groups)`),
pool.query(`SELECT id FROM users WHERE is_active = TRUE AND 'dashboard_zeugmeister' = ANY(authentik_groups)`),
]);
const fahrmeisterIds: string[] = fahrResult.rows.map((r: any) => r.id);
const zeugmeisterIds: string[] = zeugResult.rows.map((r: any) => r.id);
for (const item of equipmentResult.rows) {
const targetUsers: string[] = item.motorisiert ? fahrmeisterIds : zeugmeisterIds;
if (targetUsers.length === 0) continue;
const isBroken = ['beschaedigt', 'ausser_dienst'].includes(item.status);
const pruefungTage = item.pruefung_tage != null ? parseInt(item.pruefung_tage, 10) : null;
const isOverdueInspection = pruefungTage !== null && pruefungTage < 0;
for (const userId of targetUsers) {
if (isBroken) {
await notificationService.createNotification({
user_id: userId,
typ: 'ausruestung_status',
titel: `Ausrüstung nicht einsatzbereit`,
nachricht: `${item.bezeichnung} hat den Status "${item.status}" und ist nicht einsatzbereit.`,
schwere: 'fehler',
link: `/ausruestung/${item.id}`,
quell_id: `ausruestung-status-${item.id}`,
quell_typ: 'ausruestung_status',
});
}
if (isOverdueInspection) {
const tage = Math.abs(pruefungTage!);
await notificationService.createNotification({
user_id: userId,
typ: 'ausruestung_pruefung',
titel: `Ausrüstungsprüfung überfällig`,
nachricht: `Die Prüfung von ${item.bezeichnung} ist seit ${tage} Tagen überfällig.`,
schwere: 'fehler',
link: `/ausruestung/${item.id}`,
quell_id: `ausruestung-pruefung-${item.id}`,
quell_typ: 'ausruestung_pruefung',
});
}
}
}
await pool.query(`
INSERT INTO notifications (user_id, typ, titel, nachricht, schwere, link, quell_id, quell_typ)
SELECT
u.id,
'ausruestung_pruefung',
'Ausrüstungsprüfung überfällig',
'Die Prüfung von ' || a.bezeichnung || ' ist seit ' || ABS(a.naechste_pruefung_am::date - CURRENT_DATE) || ' Tagen überfällig.',
'fehler',
'/ausruestung/' || a.id::text,
'ausruestung-pruefung-' || a.id::text,
'ausruestung_pruefung'
FROM ausruestung a
JOIN ausruestung_kategorien k ON k.id = a.kategorie_id
JOIN users u ON u.is_active = TRUE AND (
(k.motorisiert = TRUE AND 'dashboard_fahrmeister' = ANY(u.authentik_groups))
OR
(k.motorisiert = FALSE AND 'dashboard_zeugmeister' = ANY(u.authentik_groups))
)
WHERE a.deleted_at IS NULL
AND a.naechste_pruefung_am IS NOT NULL
AND a.naechste_pruefung_am::date < CURRENT_DATE
ON CONFLICT (user_id, quell_typ, quell_id)
WHERE NOT gelesen AND quell_typ IS NOT NULL AND quell_id IS NOT NULL
DO NOTHING
`);
} catch (error) {
logger.error('NotificationGenerationJob: generateEquipmentNotifications failed', { error });
}
}
// ---------------------------------------------------------------------------
// 4. Nextcloud Talk unread messages → per-user notifications
// 4. Nextcloud Talk unread messages — batched concurrency (3 users at a time)
// ---------------------------------------------------------------------------
const NEXTCLOUD_BATCH_SIZE = 3;
async function generateNextcloudTalkNotifications(): Promise<void> {
const usersResult = await pool.query(`
SELECT id, nextcloud_login_name, nextcloud_app_password
@@ -249,40 +245,47 @@ async function generateNextcloudTalkNotifications(): Promise<void> {
AND nextcloud_app_password IS NOT NULL
`);
for (const user of usersResult.rows) {
try {
const { conversations } = await nextcloudService.getConversations(
user.nextcloud_login_name,
user.nextcloud_app_password,
);
const users = usersResult.rows;
for (const conv of conversations) {
if (conv.unreadMessages <= 0) continue;
await notificationService.createNotification({
user_id: user.id,
typ: 'nextcloud_talk',
titel: conv.displayName,
nachricht: `${conv.unreadMessages} ungelesene Nachrichten`,
schwere: 'info',
link: conv.url,
quell_id: conv.token,
quell_typ: 'nextcloud_talk',
});
}
} catch (error: any) {
if (error?.code === 'NEXTCLOUD_AUTH_INVALID') {
await pool.query(
`UPDATE users SET nextcloud_login_name = NULL, nextcloud_app_password = NULL WHERE id = $1`,
[user.id],
);
logger.warn('NotificationGenerationJob: cleared invalid Nextcloud credentials', { userId: user.id });
continue;
}
logger.error('NotificationGenerationJob: generateNextcloudTalkNotifications failed for user', {
userId: user.id,
error,
for (let i = 0; i < users.length; i += NEXTCLOUD_BATCH_SIZE) {
const batch = users.slice(i, i + NEXTCLOUD_BATCH_SIZE);
await Promise.allSettled(batch.map((user) => processNextcloudUser(user)));
}
}
async function processNextcloudUser(user: { id: string; nextcloud_login_name: string; nextcloud_app_password: string }): Promise<void> {
try {
const { conversations } = await nextcloudService.getConversations(
user.nextcloud_login_name,
user.nextcloud_app_password,
);
for (const conv of conversations) {
if (conv.unreadMessages <= 0) continue;
await notificationService.createNotification({
user_id: user.id,
typ: 'nextcloud_talk',
titel: conv.displayName,
nachricht: `${conv.unreadMessages} ungelesene Nachrichten`,
schwere: 'info',
link: conv.url,
quell_id: conv.token,
quell_typ: 'nextcloud_talk',
});
}
} catch (error: any) {
if (error?.code === 'NEXTCLOUD_AUTH_INVALID') {
await pool.query(
`UPDATE users SET nextcloud_login_name = NULL, nextcloud_app_password = NULL WHERE id = $1`,
[user.id],
);
logger.warn('NotificationGenerationJob: cleared invalid Nextcloud credentials', { userId: user.id });
return;
}
logger.error('NotificationGenerationJob: generateNextcloudTalkNotifications failed for user', {
userId: user.id,
error,
});
}
}