Diff -- /var/www/vhosts/3dshawn.com/site1/_alert_dispatch.pl

O Operator
Diff

/var/www/vhosts/3dshawn.com/site1/_alert_dispatch.pl

modified on local at 2026-07-12 00:19:42

Added
+142
lines
Removed
-0
lines
Context
409
unchanged
Blobs
from 50aef6cbce02
to 6bb32e8ab07d
Restore this content →
Unified diff
Naive LCS-based line diff. Additions in emerald, deletions in rose. Both sides decompressed live from BLOB_STORE.
11#!/usr/bin/perl
22#======================================================================
33# DriftSense -- alert dispatcher
44#
55# Cron entry point (NOT a CGI). Walks FILE_CHANGES + SCHEMA_CHANGE rows
66# inserted since the last cursor tick, evaluates every ACTIVE rule
77# against each new row, and POSTs matching payloads to the configured
88# webhook (Slack / Discord / generic HTTP). Every attempt is logged
99# to ALERT_DELIVERIES for the operator's audit trail.
1010#
1111# Config: /etc/drift_sense/drift_sense.conf (via MODS::Config).
1212# Log: /var/log/drift_sense/alert_dispatch.log (via cron redirect).
1313#======================================================================
1414use strict;
1515use warnings;
1616use lib '/var/www/vhosts/3dshawn.com/site1';
1717use POSIX ();
1818use JSON::PP;
1919use MODS::Config;
2020use MODS::DBConnect;
2121use MODS::Webhook;
2222
2323$| = 1;
2424my $ts = POSIX::strftime('%Y-%m-%d %H:%M:%S', localtime);
2525
2626my $cfg = MODS::Config->new;
2727my $db = MODS::DBConnect->new;
2828my $dbh = $db->db_connect or die "[$ts] cannot connect to DriftSense storage DB\n";
2929
3030my $public_url = $cfg->settings('public_url') || 'https://watchtower.3dshawn.com';
3131
3232# ---- Fetch active rules ---------------------------------------------
3333my @rules = $db->db_readwrite_multiple($dbh, q~
3434 SELECT alert_rule_id, rule_name, match_kind, match_path_glob,
3535 match_status_list, exclude_ts_only,
3636 delivery_kind, delivery_url, delivery_email,
3737 rate_limit_per_min, coalesce_window_s,
3838 frequency_threshold, frequency_window_min,
3939 UNIX_TIMESTAMP(last_frequency_fire_at) AS last_freq_fire_epoch
4040 FROM ALERT_RULES
4141 WHERE is_active = 1
4242~, __FILE__, __LINE__);
4343
4444unless (@rules) {
4545 print "[$ts] no active alert rules, exiting\n";
4646 $db->db_disconnect($dbh);
4747 exit 0;
4848}
4949
5050# ---- Fetch cursor ---------------------------------------------------
5151my %cursor;
5252foreach my $r ($db->db_readwrite_multiple($dbh, q~
5353 SELECT source_kind, last_seen_id FROM ALERT_CURSOR
5454~, __FILE__, __LINE__)) {
5555 $cursor{$r->{source_kind}} = $r->{last_seen_id};
5656}
5757$cursor{file} //= 0;
5858$cursor{schema} //= 0;
5959
6060# ---- Pull new FILE_CHANGES since cursor ------------------------------
6161my $file_max = $cursor{file};
6262my @file_new = $db->db_readwrite_multiple($dbh, qq~
6363 SELECT fc.file_changes_id AS id, fc.file_name, fc.status, fc.blob_sha,
6464 fc.is_ts_only, fc.date_time,
6565 UNIX_TIMESTAMP(fc.date_time) AS ts_epoch,
6666 s.server_name
6767 FROM FILE_CHANGES fc
6868 LEFT JOIN SERVERS s ON s.server_id = fc.server_id
6969 WHERE fc.file_changes_id > $cursor{file}
7070 ORDER BY fc.file_changes_id ASC
7171 LIMIT 500
7272~, __FILE__, __LINE__);
7373foreach my $r (@file_new) {
7474 $file_max = $r->{id} if $r->{id} > $file_max;
7575}
7676
7777# ---- Pull new SCHEMA_CHANGE since cursor -----------------------------
7878my $schema_max = $cursor{schema};
7979my @schema_new = $db->db_readwrite_multiple($dbh, qq~
8080 SELECT sc.schema_change_id AS id, sc.database_name, sc.table_name,
8181 sc.changes, sc.is_ts_only, sc.change_datetime,
8282 UNIX_TIMESTAMP(sc.change_datetime) AS ts_epoch,
8383 s.server_name
8484 FROM SCHEMA_CHANGE sc
8585 LEFT JOIN SERVERS s ON s.server_id = sc.server_id
8686 WHERE sc.schema_change_id > $cursor{schema}
8787 ORDER BY sc.schema_change_id ASC
8888 LIMIT 500
8989~, __FILE__, __LINE__);
9090foreach my $r (@schema_new) {
9191 $schema_max = $r->{id} if $r->{id} > $schema_max;
9292}
9393
9494my $total_new = scalar(@file_new) + scalar(@schema_new);
9595if ($total_new == 0) {
9696 print "[$ts] no new changes since cursor (file=$cursor{file} schema=$cursor{schema})\n";
9797 $db->db_disconnect($dbh);
9898 exit 0;
9999}
100100
101101# ---- Evaluate + deliver ---------------------------------------------
102102# HTTP transport is MODS::Webhook (shells out to /usr/bin/curl) so we
103103# don't need LWP::Protocol::https / IO::Socket::SSL as CPAN installs.
104104
105105my $delivered = 0;
106106my $skipped = 0;
107107my $failed = 0;
108108
109109foreach my $row (@file_new) { _process_row($row, 'file'); }
110110foreach my $row (@schema_new) { _process_row($row, 'schema'); }
111111
112112# ---- Frequency-mode rules ------------------------------------------
113113# For any rule where frequency_threshold > 0, we ignore the per-row loop
114114# above and instead: query how many file changes matching the rule's
115115# path glob happened in the last window_min minutes. If it crosses the
116116# threshold and we haven't fired within the same window, fire once with
117117# the top offender.
118118foreach my $rule (@rules) {
119119 next unless ($rule->{frequency_threshold} || 0) > 0;
120120 my $win = int($rule->{frequency_window_min} || 60);
121121 my $thr = int($rule->{frequency_threshold});
122122
123123 # Re-fire cooldown: don't fire more than once per window
124124 my $last = int($rule->{last_freq_fire_epoch} || 0);
125125 if ($last && (time - $last) < $win * 60) {
126126 next;
127127 }
128128
129129 # Find files whose match_path_glob-hit change count > threshold
130130 my $glob = $rule->{match_path_glob} // '';
131131 my $glob_sql = '';
132132 if (length $glob) {
133133 # Approx: SQL LIKE. Convert glob wildcards to SQL % / _.
134134 my $like = $glob;
135135 $like =~ s/\*\*/%/g;
136136 $like =~ s/\*/%/g;
137137 $like =~ s/\?/_/g;
138138 my $q_like = $dbh->quote($like);
139139 $glob_sql = " AND fc.file_name LIKE $q_like";
140140 }
141141 my @hot = $db->db_readwrite_multiple($dbh, qq~
142142 SELECT fc.file_name,
143143 COUNT(*) AS ct,
144144 MAX(fc.file_changes_id) AS latest_id,
145145 s.server_name
146146 FROM FILE_CHANGES fc
147147 LEFT JOIN SERVERS s ON s.server_id = fc.server_id
148148 WHERE fc.date_time >= DATE_SUB(NOW(), INTERVAL $win MINUTE)
149149 $glob_sql
150150 GROUP BY fc.file_name
151151 HAVING ct >= $thr
152152 ORDER BY ct DESC
153153 LIMIT 5
154154 ~, __FILE__, __LINE__);
155155 next unless @hot;
156156
157157 # Fire once with the top offender's info
158158 my $top = $hot[0];
159159 my $fake_row = {
160160 id => $top->{latest_id},
161161 file_name => $top->{file_name},
162162 server_name => $top->{server_name} // 'local',
163163 date_time => POSIX::strftime('%Y-%m-%d %H:%M:%S', localtime),
164164 ts_epoch => time,
165165 status => 'frequency_alert',
166166 };
167167 my $payload = _build_payload($rule, $fake_row, 'file');
168168 # Enrich summary with count + threshold
169169 my $extra = "frequency alert: $top->{file_name} changed $top->{ct} times in the last $win min (threshold: $thr)";
170170 if (ref($payload->{generic_webhook}) eq 'HASH') {
171171 $payload->{generic_webhook}{frequency_summary} = $extra;
172172 $payload->{generic_webhook}{frequency_count} = $top->{ct};
173173 $payload->{generic_webhook}{frequency_window} = $win;
174174 $payload->{generic_webhook}{summary} = $extra;
175175 }
176176 my ($status, $http_code, $body, $err) = _deliver($rule, $payload);
177177 _log_delivery($rule, $fake_row, 'file', $status, $http_code, $body, $err);
178178 if ($status eq 'sent') {
179179 $delivered++;
180180 $db->db_readwrite($dbh, qq~
181181 UPDATE ALERT_RULES
182182 SET fire_count = fire_count + 1,
183183 last_fired_at = NOW(),
184184 last_frequency_fire_at = NOW()
185185 WHERE alert_rule_id = $rule->{alert_rule_id}
186186 ~, __FILE__, __LINE__);
187187 } else {
188188 $failed++;
189189 }
190190}
191191
192192# ---- Advance cursor atomically --------------------------------------
193193if ($file_max > $cursor{file}) {
194194 $db->db_readwrite($dbh,
195195 "UPDATE ALERT_CURSOR SET last_seen_id = $file_max WHERE source_kind = 'file'",
196196 __FILE__, __LINE__);
197197}
198198if ($schema_max > $cursor{schema}) {
199199 $db->db_readwrite($dbh,
200200 "UPDATE ALERT_CURSOR SET last_seen_id = $schema_max WHERE source_kind = 'schema'",
201201 __FILE__, __LINE__);
202}
203
204# ---- Pin-drift alerts (Feature 5 wave 4) ---------------------------
205# For each Named Release with alert_delivery_url set, check whether any
206# pin's current-latest SHA differs from the pinned SHA. Log the drift
207# to PIN_DRIFT_LOG (with notified=0) and, if not yet notified, POST an
208# alert to the release's delivery URL, then set notified=1.
209foreach my $rel ($db->db_readwrite_multiple($dbh, q~
210 SELECT named_release_id, name, alert_delivery_kind, alert_delivery_url
211 FROM NAMED_RELEASES
212 WHERE alert_delivery_kind IS NOT NULL
213 AND alert_delivery_kind != 'none'
214 AND alert_delivery_url IS NOT NULL
215 AND alert_delivery_url != ''
216~, __FILE__, __LINE__)) {
217
218 my $rid = $rel->{named_release_id};
219 my @drifts;
220 foreach my $pin ($db->db_readwrite_multiple($dbh, qq~
221 SELECT p.pin_id, p.file_name, p.blob_sha AS pinned_sha,
222 fc.server_id
223 FROM NAMED_RELEASE_PINS p
224 LEFT JOIN FILE_CHANGES fc ON fc.file_changes_id = p.file_changes_id
225 WHERE p.named_release_id = $rid
226 ~, __FILE__, __LINE__)) {
227 my $q_file = $dbh->quote($pin->{file_name});
228 my $sid = int($pin->{server_id} || 0);
229 my $cur = $db->db_readwrite($dbh, qq~
230 SELECT blob_sha, file_changes_id FROM FILE_CHANGES
231 WHERE server_id = $sid AND file_name = $q_file
232 ORDER BY file_changes_id DESC LIMIT 1
233 ~, __FILE__, __LINE__);
234 my $cur_sha = $cur ? $cur->{blob_sha} : '';
235 next unless $cur_sha && $cur_sha ne $pin->{pinned_sha};
236
237 # Have we already logged + notified this (pin, current_sha) tuple?
238 my $q_pinned = $dbh->quote($pin->{pinned_sha});
239 my $q_current = $dbh->quote($cur_sha);
240 my $seen = $db->db_readwrite($dbh, qq~
241 SELECT drift_id, notified FROM PIN_DRIFT_LOG
242 WHERE pin_id = $pin->{pin_id}
243 AND current_sha = $q_current
244 ORDER BY drift_id DESC LIMIT 1
245 ~, __FILE__, __LINE__);
246 if ($seen && $seen->{notified}) {
247 next; # already notified for this exact drift state
248 }
249
250 # Log the drift
251 my $q_fname = $dbh->quote($pin->{file_name});
252 my $cid = int($cur->{file_changes_id} || 0);
253 $db->db_readwrite($dbh, qq~
254 INSERT INTO PIN_DRIFT_LOG
255 (named_release_id, pin_id, file_name, pinned_sha, current_sha, file_changes_id, notified)
256 VALUES
257 ($rid, $pin->{pin_id}, $q_fname, $q_pinned, $q_current, $cid, 0)
258 ~, __FILE__, __LINE__);
259 push @drifts, {
260 file_name => $pin->{file_name},
261 pinned_sha => $pin->{pinned_sha},
262 current_sha => $cur_sha,
263 change_id => $cid,
264 };
265 }
266
267 next unless @drifts;
268
269 # Build a summary payload
270 my $summary = sprintf('Named release "%s" -- %d pinned file(s) have drifted', $rel->{name}, scalar @drifts);
271 my $public_url = $cfg->settings('public_url') || 'https://watchtower.3dshawn.com';
272 my $files_list = join("\n", map {
273 " * $_->{file_name} (pinned: " . substr($_->{pinned_sha}, 0, 12) .
274 " -> current: " . substr($_->{current_sha}, 0, 12) . ")"
275 } @drifts);
276 my $link = "$public_url/restore_release.cgi?release_id=$rid";
277
278 my $slack_body = {
279 text => "*DriftSense pin drift*: $summary",
280 attachments => [{
281 color => '#f43f5e',
282 fields => [
283 { title => 'Release', value => $rel->{name}, short => 1 },
284 { title => 'Drifted files', value => scalar(@drifts), short => 1 },
285 { title => 'Details', value => $files_list, short => 0 },
286 ],
287 actions => [{ type => 'button', text => 'Restore all pinned', url => $link }],
288 }],
289 };
290 my $discord_body = {
291 content => "**DriftSense pin drift**: $summary",
292 embeds => [{
293 title => $rel->{name},
294 url => $link,
295 color => 16005694,
296 description => $files_list,
297 }],
298 };
299 my $generic_body = {
300 source => 'drift_sense',
301 alert_type => 'pin_drift',
302 release_id => $rid,
303 release_name => $rel->{name},
304 drifted_count=> scalar @drifts,
305 summary => $summary,
306 files => \@drifts,
307 restore_url => $link,
308 };
309
310 my $body_ref = $rel->{alert_delivery_kind} eq 'slack' ? $slack_body
311 : $rel->{alert_delivery_kind} eq 'discord' ? $discord_body
312 : $generic_body;
313
314 my ($code, $rbody, $err) = MODS::Webhook::post_json(
315 $rel->{alert_delivery_url}, $body_ref,
316 timeout => 8,
317 user_agent => 'DriftSense/1.0 (pin-drift)',
318 );
319
320 if (!$err && $code >= 200 && $code < 400) {
321 # Mark all just-logged rows as notified
322 foreach my $d (@drifts) {
323 my $q_current = $dbh->quote($d->{current_sha});
324 $db->db_readwrite($dbh, qq~
325 UPDATE PIN_DRIFT_LOG SET notified = 1
326 WHERE named_release_id = $rid
327 AND file_name = @{[ $dbh->quote($d->{file_name}) ]}
328 AND current_sha = $q_current
329 AND notified = 0
330 ~, __FILE__, __LINE__);
331 }
332 $delivered++;
333 print "[$ts] pin-drift alert delivered for release '$rel->{name}' (" . scalar(@drifts) . " drifted)\n";
334 } else {
335 $failed++;
336 print "[$ts] pin-drift alert FAILED for release '$rel->{name}': " . ($err // "HTTP $code") . "\n";
337 }
338
339 # Bookkeep the check timestamp
340 $db->db_readwrite($dbh, qq~
341 UPDATE NAMED_RELEASES SET last_pin_drift_check_at = NOW()
342 WHERE named_release_id = $rid
343 ~, __FILE__, __LINE__);
202344}
203345
204346$db->db_disconnect($dbh);
205347print "[$ts] processed +$total_new new changes: $delivered delivered, $skipped skipped, $failed failed\n";
206348exit 0;
207349
208350#---------------------------------------------------------------------
209351sub _process_row {
210352 my ($row, $kind) = @_;
211353 foreach my $rule (@rules) {
212354 next unless _rule_matches($rule, $row, $kind);
213355
214356 # Rate limit check
215357 if ($rule->{rate_limit_per_min} > 0) {
216358 my $rlq = $db->db_readwrite($dbh, qq~
217359 SELECT COUNT(*) AS n FROM ALERT_DELIVERIES
218360 WHERE alert_rule_id = $rule->{alert_rule_id}
219361 AND fired_at >= DATE_SUB(NOW(), INTERVAL 1 MINUTE)
220362 AND delivery_status IN ('sent','pending')
221363 ~, __FILE__, __LINE__);
222364 if ($rlq && $rlq->{n} >= $rule->{rate_limit_per_min}) {
223365 _log_delivery($rule, $row, $kind, 'skipped', undef, undef,
224366 "rate limit ($rule->{rate_limit_per_min}/min) reached");
225367 $skipped++;
226368 next;
227369 }
228370 }
229371
230372 # Build payload + POST
231373 my $payload = _build_payload($rule, $row, $kind);
232374 my ($status, $http_code, $body, $err) = _deliver($rule, $payload);
233375 _log_delivery($rule, $row, $kind, $status, $http_code, $body, $err);
234376
235377 if ($status eq 'sent') { $delivered++; }
236378 elsif ($status eq 'skipped') { $skipped++; }
237379 else { $failed++; }
238380
239381 # Bump the rule's fire_count + last_fired_at
240382 if ($status eq 'sent') {
241383 $db->db_readwrite($dbh, qq~
242384 UPDATE ALERT_RULES
243385 SET fire_count = fire_count + 1,
244386 last_fired_at = NOW()
245387 WHERE alert_rule_id = $rule->{alert_rule_id}
246388 ~, __FILE__, __LINE__);
247389 }
248390 }
249391}
250392
251393#---------------------------------------------------------------------
252394sub _rule_matches {
253395 my ($rule, $row, $kind) = @_;
254396
255397 # Kind filter
256398 return 0 if $rule->{match_kind} ne 'either' && $rule->{match_kind} ne $kind;
257399
258400 # ts-only exclusion (files only)
259401 return 0 if $kind eq 'file' && $rule->{exclude_ts_only} && $row->{is_ts_only};
260402
261403 # Path glob
262404 my $glob = $rule->{match_path_glob} // '';
263405 if (length $glob) {
264406 my $target = $kind eq 'file'
265407 ? ($row->{file_name} // '')
266408 : (($row->{database_name} // '') . '.' . ($row->{table_name} // ''));
267409 my $re = _glob_to_regex($glob);
268410 return 0 unless $target =~ $re;
269411 }
270412
271413 # Status filter (files only)
272414 if ($kind eq 'file' && $rule->{match_status_list}) {
273415 my %allowed = map { s/^\s+|\s+$//g; $_ => 1 }
274416 split /,/, $rule->{match_status_list};
275417 return 0 unless $allowed{ $row->{status} // '' };
276418 }
277419
278420 return 1;
279421}
280422
281423sub _glob_to_regex {
282424 my $g = shift;
283425 my $re = quotemeta $g;
284426 # Handle glob patterns: escaped meta chars back to regex meaning
285427 $re =~ s{\\\*\\\*}{.*}g; # ** -> .*
286428 $re =~ s{\\\*}{[^/]*}g; # * -> [^/]*
287429 $re =~ s{\\\?}{.}g; # ? -> .
288430 return qr/^$re$/;
289431}
290432
291433#---------------------------------------------------------------------
292434sub _build_payload {
293435 my ($rule, $row, $kind) = @_;
294436
295437 my $where = $kind eq 'file'
296438 ? ($row->{file_name} // '(unknown file)')
297439 : (($row->{database_name} // '?') . '.' . ($row->{table_name} // '?'));
298440 my $when_utc = $row->{date_time} || $row->{change_datetime} || '';
299441 my $server = $row->{server_name} || 'local';
300442 my $status_h = $kind eq 'file' ? ($row->{status} // '') : 'DDL change';
301443 my $link_id = $row->{id};
302444 my $link = "$public_url/diff.cgi?kind=$kind&id=$link_id";
303445
304446 my $summary = $kind eq 'file'
305447 ? "$status_h: $where on $server"
306448 : "$where changed on $server";
307449
308450 my $slack = {
309451 text => "*DriftSense*: $summary",
310452 attachments => [{
311453 color => $kind eq 'schema' ? '#f59e0b' : '#14b8a6',
312454 fields => [
313455 { title => 'What', value => $where, short => 0 },
314456 { title => 'When', value => $when_utc, short => 1 },
315457 { title => 'Server', value => $server, short => 1 },
316458 ],
317459 actions => [{
318460 type => 'button', text => 'View diff', url => $link,
319461 }],
320462 }],
321463 };
322464 my $discord = {
323465 content => "**DriftSense**: $summary",
324466 embeds => [{
325467 title => $where,
326468 url => $link,
327469 color => $kind eq 'schema' ? 16101915 : 1349286, # amber, teal
328470 fields => [
329471 { name => 'When', value => $when_utc || '?', inline => \1 },
330472 { name => 'Server', value => $server, inline => \1 },
331473 ],
332474 }],
333475 };
334476 my $generic = {
335477 source => 'drift_sense',
336478 rule_id => $rule->{alert_rule_id},
337479 rule_name => $rule->{rule_name},
338480 kind => $kind,
339481 summary => $summary,
340482 target => $where,
341483 server => $server,
342484 when_utc => $when_utc,
343485 ts_epoch => $row->{ts_epoch},
344486 diff_url => $link,
345487 status => $status_h,
346488 };
347489
348490 return {
349491 slack => $slack,
350492 discord => $discord,
351493 generic_webhook => $generic,
352494 summary => $summary,
353495 };
354496}
355497
356498#---------------------------------------------------------------------
357499sub _deliver {
358500 my ($rule, $payload) = @_;
359501 my $kind = $rule->{delivery_kind};
360502 my $url = $rule->{delivery_url};
361503
362504 if ($kind eq 'email') {
363505 return ('skipped', undef, undef, 'email delivery not yet implemented');
364506 }
365507 unless (length($url // '')) {
366508 return ('failed', undef, undef, "no delivery URL configured");
367509 }
368510
369511 my $body_ref = $kind eq 'slack' ? $payload->{slack}
370512 : $kind eq 'discord' ? $payload->{discord}
371513 : $payload->{generic_webhook};
372514
373515 # MODS::Webhook handles JSON encoding + curl transport; returns
374516 # (http_code, response_body, error_message).
375517 my ($code, $rbody, $err) = MODS::Webhook::post_json(
376518 $url, $body_ref,
377519 timeout => 8,
378520 user_agent => 'DriftSense/1.0 (alert-dispatch)',
379521 );
380522 $rbody //= '';
381523 $rbody = substr($rbody, 0, 500) if length($rbody) > 500;
382524
383525 if (!$err && $code >= 200 && $code < 400) {
384526 return ('sent', $code, $rbody, undef);
385527 }
386528 return ('failed', $code, $rbody, $err || "HTTP $code");
387529}
388530
389531#---------------------------------------------------------------------
390532sub _log_delivery {
391533 my ($rule, $row, $kind, $status, $http_code, $body, $err) = @_;
392534 my $summary = substr(($rule->{rule_name} // '') . ' :: ' . ($row->{file_name}
393535 || ($row->{database_name} . '.' . $row->{table_name})
394536 || '?'), 0, 490);
395537 my $q_summary = $dbh->quote($summary);
396538 my $q_status = $dbh->quote($status);
397539 my $q_body = $dbh->quote($body // '');
398540 my $q_err = $dbh->quote($err // '');
399541 my $q_code = defined($http_code) ? int($http_code) : 'NULL';
400542
401543 $db->db_readwrite($dbh, qq~
402544 INSERT INTO ALERT_DELIVERIES
403545 (alert_rule_id, source_kind, source_id, match_summary,
404546 delivery_status, http_status, http_response, error_message)
405547 VALUES
406548 ($rule->{alert_rule_id}, '$kind', $row->{id}, $q_summary,
407549 $q_status, $q_code, $q_body, $q_err)
408550 ~, __FILE__, __LINE__);
409551}