jobs: public IP, trunk and disk watches, retention
apps/api/src/jobs.ts takes over the once-a-minute timer from index.ts (no croner: every schedule in 04 §8 is whole minutes). Each job has a running flag; errors are logged. - Retention (at start and when the date changes, once AMI is up): recordings, voicemail past voicemail.retentionDays (VoicemailRemove, so Asterisk renumbers and updates MWI), expired sessions, webhook deliveries older than 30 days. - publicIpWatch (every 5 min, Settings -> Network): a changed IP is saved, audited, applied (restart required) and emailed to admins. - trunkWatch: a registering trunk rejected/unregistered for 5 minutes -> one email per outage. - diskWatch (hourly): under 10% free -> one email a day. - A failed daily backup emails admins too. Settings gets a Network tab (public IP, Detect, re-check) and voicemail retention on the Recording tab. Task: M11-T4
This commit is contained in:
1 parent
164a5f0990
commit
d229ed2c45
8 files changed
+755
-43
No files matched your search
@@ -0,0 +1,236 @@
|
||||
// SPDX-License-Identifier: AGPL-3.0-or-later
|
||||
// Background jobs (04 §8) on one once-a-minute timer.
|
||||
// ponytail: setInterval instead of croner: every schedule in 04 §8 is a whole number of minutes.
|
||||
import { statfs } from 'node:fs/promises';
|
||||
import { resolve } from 'node:path';
|
||||
import { and, eq, isNull, lt, sql } from 'drizzle-orm';
|
||||
import { z } from 'zod';
|
||||
import type { Config } from './config.ts';
|
||||
import type { Db } from './db/client.ts';
|
||||
import {
|
||||
extensions,
|
||||
sessions,
|
||||
trunks,
|
||||
users,
|
||||
voicemailMessages,
|
||||
webhookDeliveries,
|
||||
} from './db/schema.ts';
|
||||
import type { Generator } from './generator/index.ts';
|
||||
import { todayIn } from './generator/load.ts';
|
||||
import type { LiveState } from './live/state.ts';
|
||||
import { cleanupRecordings } from './routes/recordings.ts';
|
||||
import { audit } from './services/audit.ts';
|
||||
import { backupDue, runBackup } from './services/backup.ts';
|
||||
import type { SendMail } from './services/mail.ts';
|
||||
import { getSetting, setSetting } from './services/settings.ts';
|
||||
|
||||
export type JobDeps = {
|
||||
config: Config;
|
||||
db: Db;
|
||||
ami: { isConnected(): boolean; action(f: Record<string, string>): Promise<unknown> };
|
||||
live: Pick<LiveState, 'snapshot' | 'amiConnected'>;
|
||||
generator: Pick<Generator, 'apply'>;
|
||||
voicemail: { scan(ext: string): Promise<void> };
|
||||
send: SendMail;
|
||||
log: { info(o: object, m: string): void; error(o: object, m: string): void };
|
||||
now?: () => Date;
|
||||
statfs?: typeof statfs;
|
||||
};
|
||||
|
||||
/** The server's public IPv4 as an outside service sees it, or null. */
|
||||
export async function detectPublicIp(): Promise<string | null> {
|
||||
for (const url of ['https://api.ipify.org', 'https://ifconfig.me/ip']) {
|
||||
const ip = await fetch(url, { signal: AbortSignal.timeout(5000) })
|
||||
.then((r) => (r.ok ? r.text() : ''))
|
||||
.catch(() => '');
|
||||
if (z.ipv4().safeParse(ip.trim()).success) return ip.trim();
|
||||
}
|
||||
return null;
|
||||
}
|
||||
|
||||
/** Email every active admin (08 §3.2). Email not set up or failing is logged, not thrown. */
|
||||
export async function mailAdmins(d: JobDeps, subject: string, text: string) {
|
||||
const to = (
|
||||
await d.db
|
||||
.select({ email: users.email })
|
||||
.from(users)
|
||||
.where(and(eq(users.role, 'admin'), eq(users.isActive, true)))
|
||||
).map((r) => r.email);
|
||||
if (to.length)
|
||||
await d.send({ to, subject, text }).catch((err: unknown) => d.log.error({ err }, subject));
|
||||
}
|
||||
|
||||
/** Recordings, voicemail, expired sessions and old webhook deliveries (04 §8 retentionCleanup). */
|
||||
export async function retentionCleanup(d: JobDeps) {
|
||||
const recordings = await cleanupRecordings(
|
||||
d.db,
|
||||
resolve(d.config.RECORDINGS_DIR),
|
||||
await getSetting(d.db, 'recordings.retentionDays'),
|
||||
);
|
||||
let voicemail = 0;
|
||||
const vmDays = await getSetting(d.db, 'voicemail.retentionDays');
|
||||
// Asterisk removes the files and renumbers the folder (the caller waits for AMI).
|
||||
if (vmDays > 0) {
|
||||
const old = await d.db
|
||||
.select({
|
||||
ext: extensions.number,
|
||||
folder: voicemailMessages.folder,
|
||||
msgId: voicemailMessages.msgId,
|
||||
})
|
||||
.from(voicemailMessages)
|
||||
.innerJoin(extensions, eq(extensions.id, voicemailMessages.extensionId))
|
||||
.where(
|
||||
and(
|
||||
isNull(voicemailMessages.deletedAt),
|
||||
lt(voicemailMessages.receivedAt, sql`now() - make_interval(days => ${vmDays})`),
|
||||
),
|
||||
);
|
||||
for (const m of old)
|
||||
await d.ami.action({
|
||||
Action: 'VoicemailRemove',
|
||||
Context: 'default',
|
||||
Mailbox: m.ext,
|
||||
Folder: m.folder,
|
||||
ID: m.msgId,
|
||||
});
|
||||
for (const ext of new Set(old.map((m) => m.ext))) await d.voicemail.scan(ext);
|
||||
voicemail = old.length;
|
||||
}
|
||||
await d.db.delete(sessions).where(lt(sessions.expiresAt, sql`now()`));
|
||||
await d.db
|
||||
.delete(webhookDeliveries)
|
||||
.where(lt(webhookDeliveries.createdAt, sql`now() - interval '30 days'`));
|
||||
if (recordings || voicemail) d.log.info({ recordings, voicemail }, 'old messages deleted');
|
||||
}
|
||||
|
||||
/** The jobs, each keeping its own state (tests call them directly). */
|
||||
export function createJobs(d: JobDeps) {
|
||||
const now = d.now ?? (() => new Date());
|
||||
const tz = () => getSetting(d.db, 'general.timezone');
|
||||
|
||||
// Holidays roll over at midnight (03 §6.10): render again when the local date changes.
|
||||
// Retention runs then too (and at start), not at 02:00.
|
||||
let lastDay = '';
|
||||
let cleanedDay = '';
|
||||
const daily = async () => {
|
||||
const today = todayIn(await tz(), now());
|
||||
if (lastDay && today !== lastDay) await d.generator.apply({ trigger: 'nightly', force: true });
|
||||
lastDay = today;
|
||||
// Voicemail goes through Asterisk: at start, wait the few seconds until AMI is connected.
|
||||
if (cleanedDay !== today && d.ami.isConnected()) {
|
||||
await retentionCleanup(d);
|
||||
cleanedDay = today;
|
||||
}
|
||||
};
|
||||
|
||||
let backupTriedDay = ''; // one try a day: a failing pg_dump shouldn't run every minute
|
||||
const backup = async () => {
|
||||
const today = todayIn(await tz(), now());
|
||||
if (backupTriedDay === today || !(await backupDue(d.config, d.db, now()))) return;
|
||||
backupTriedDay = today;
|
||||
try {
|
||||
const b = await runBackup(d.config, d.db);
|
||||
d.log.info({ backup: b.name, size: b.size }, 'backup made');
|
||||
} catch (err) {
|
||||
await mailAdmins(d, 'Backup failed', `Today's backup failed: ${(err as Error).message}`);
|
||||
throw err;
|
||||
}
|
||||
};
|
||||
|
||||
const publicIpWatch = async () => {
|
||||
if (!(await getSetting(d.db, 'network.publicIpWatch'))) return;
|
||||
const ip = await detectPublicIp();
|
||||
const old = await getSetting(d.db, 'network.publicIp');
|
||||
if (!ip || ip === old) return;
|
||||
await setSetting(d.db, 'network.publicIp', ip);
|
||||
await audit(d.db, {
|
||||
action: 'settings',
|
||||
summary: `Public IP changed from ${old ?? 'none'} to ${ip}`,
|
||||
diff: { before: old, after: ip },
|
||||
});
|
||||
await d.generator.apply({ trigger: 'publicIp', force: true }); // sets restartRequired
|
||||
await mailAdmins(
|
||||
d,
|
||||
`Public IP changed to ${ip}`,
|
||||
`The phone system's public IP changed from ${old ?? 'none'} to ${ip}.\n` +
|
||||
'Restart telephony from the dashboard so calls use the new address (it drops calls in progress).\n' +
|
||||
'Phones and the provider may need the new IP if they were set up with the old one.',
|
||||
);
|
||||
};
|
||||
|
||||
// A registering trunk that stays down for 5 minutes: one email per outage.
|
||||
const downSince = new Map<string, number>();
|
||||
const mailed = new Set<string>();
|
||||
const trunkWatch = async () => {
|
||||
if (!d.live.amiConnected) return; // Asterisk itself being down shows elsewhere
|
||||
const state = new Map((d.live.snapshot(true).trunks ?? []).map((t) => [t.trunkId, t]));
|
||||
const rows = await d.db
|
||||
.select({ id: trunks.id, name: trunks.name })
|
||||
.from(trunks)
|
||||
.where(eq(trunks.enabled, true));
|
||||
for (const t of rows) {
|
||||
const reg = state.get(t.id)?.registration;
|
||||
// ponytail: IP-auth trunks don't register (always 'unknown'), so they're never "down" here.
|
||||
if (reg !== 'rejected' && reg !== 'unregistered') {
|
||||
downSince.delete(t.id);
|
||||
mailed.delete(t.id);
|
||||
continue;
|
||||
}
|
||||
const since = downSince.get(t.id) ?? now().getTime();
|
||||
downSince.set(t.id, since);
|
||||
if (now().getTime() - since < 5 * 60_000 || mailed.has(t.id)) continue;
|
||||
mailed.add(t.id);
|
||||
await mailAdmins(
|
||||
d,
|
||||
`Trunk ${t.name} is down`,
|
||||
`Trunk ${t.name} has not been registered with the provider since ` +
|
||||
`${new Date(since).toISOString()} (${reg}). Calls through it fail.\n` +
|
||||
'Check Trunks → Test, the account password, and the provider status.',
|
||||
);
|
||||
}
|
||||
};
|
||||
|
||||
let diskMailedDay = '';
|
||||
const diskWatch = async () => {
|
||||
const fs = await (d.statfs ?? statfs)(resolve(d.config.RECORDINGS_DIR));
|
||||
const free = fs.bavail / fs.blocks;
|
||||
const today = todayIn(await tz(), now());
|
||||
if (free >= 0.1 || diskMailedDay === today) return;
|
||||
diskMailedDay = today;
|
||||
await mailAdmins(
|
||||
d,
|
||||
`Disk almost full (${Math.round(free * 100)}% free)`,
|
||||
`Only ${Math.round((fs.bavail * fs.bsize) / 1e9)} GB (${Math.round(free * 100)}%) is free ` +
|
||||
'on the phone system. Recordings, voicemail and backups stop being saved when it fills.\n' +
|
||||
'Lower the recording retention (Settings → Recording) or free up space.',
|
||||
);
|
||||
};
|
||||
|
||||
return { daily, backup, publicIpWatch, trunkWatch, diskWatch };
|
||||
}
|
||||
|
||||
/** Starts the jobs; returns a stop function. */
|
||||
export function startJobs(d: JobDeps): () => void {
|
||||
const j = createJobs(d);
|
||||
// Every job: never two runs of itself at once (04 §8), errors logged.
|
||||
const running = new Set<string>();
|
||||
const run = (name: string, fn: () => Promise<void>) => {
|
||||
if (running.has(name)) return;
|
||||
running.add(name);
|
||||
void fn()
|
||||
.catch((err: unknown) => d.log.error({ err }, `job ${name} failed`))
|
||||
.finally(() => running.delete(name));
|
||||
};
|
||||
let minute = 0;
|
||||
const tick = () => {
|
||||
run('daily', j.daily);
|
||||
run('backup', j.backup);
|
||||
run('trunkWatch', j.trunkWatch);
|
||||
if (minute % 5 === 0) run('publicIpWatch', j.publicIpWatch);
|
||||
if (minute % 60 === 0) run('diskWatch', j.diskWatch);
|
||||
minute++;
|
||||
};
|
||||
tick();
|
||||
const timer = setInterval(tick, 60_000);
|
||||
return () => clearInterval(timer);
|
||||
}
|
||||
Reference in new issue
Block a user