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:02:08

Added
+83
lines
Removed
-1
lines
Context
326
unchanged
Blobs
from ad10d46443a1
to 50aef6cbce02
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,
37 rate_limit_per_min, coalesce_window_s
37 rate_limit_per_min, coalesce_window_s,
38 frequency_threshold, frequency_window_min,
39 UNIX_TIMESTAMP(last_frequency_fire_at) AS last_freq_fire_epoch
3840 FROM ALERT_RULES
3941 WHERE is_active = 1
4042~, __FILE__, __LINE__);
4143
4244unless (@rules) {
4345 print "[$ts] no active alert rules, exiting\n";
4446 $db->db_disconnect($dbh);
4547 exit 0;
4648}
4749
4850# ---- Fetch cursor ---------------------------------------------------
4951my %cursor;
5052foreach my $r ($db->db_readwrite_multiple($dbh, q~
5153 SELECT source_kind, last_seen_id FROM ALERT_CURSOR
5254~, __FILE__, __LINE__)) {
5355 $cursor{$r->{source_kind}} = $r->{last_seen_id};
5456}
5557$cursor{file} //= 0;
5658$cursor{schema} //= 0;
5759
5860# ---- Pull new FILE_CHANGES since cursor ------------------------------
5961my $file_max = $cursor{file};
6062my @file_new = $db->db_readwrite_multiple($dbh, qq~
6163 SELECT fc.file_changes_id AS id, fc.file_name, fc.status, fc.blob_sha,
6264 fc.is_ts_only, fc.date_time,
6365 UNIX_TIMESTAMP(fc.date_time) AS ts_epoch,
6466 s.server_name
6567 FROM FILE_CHANGES fc
6668 LEFT JOIN SERVERS s ON s.server_id = fc.server_id
6769 WHERE fc.file_changes_id > $cursor{file}
6870 ORDER BY fc.file_changes_id ASC
6971 LIMIT 500
7072~, __FILE__, __LINE__);
7173foreach my $r (@file_new) {
7274 $file_max = $r->{id} if $r->{id} > $file_max;
7375}
7476
7577# ---- Pull new SCHEMA_CHANGE since cursor -----------------------------
7678my $schema_max = $cursor{schema};
7779my @schema_new = $db->db_readwrite_multiple($dbh, qq~
7880 SELECT sc.schema_change_id AS id, sc.database_name, sc.table_name,
7981 sc.changes, sc.is_ts_only, sc.change_datetime,
8082 UNIX_TIMESTAMP(sc.change_datetime) AS ts_epoch,
8183 s.server_name
8284 FROM SCHEMA_CHANGE sc
8385 LEFT JOIN SERVERS s ON s.server_id = sc.server_id
8486 WHERE sc.schema_change_id > $cursor{schema}
8587 ORDER BY sc.schema_change_id ASC
8688 LIMIT 500
8789~, __FILE__, __LINE__);
8890foreach my $r (@schema_new) {
8991 $schema_max = $r->{id} if $r->{id} > $schema_max;
9092}
9193
9294my $total_new = scalar(@file_new) + scalar(@schema_new);
9395if ($total_new == 0) {
9496 print "[$ts] no new changes since cursor (file=$cursor{file} schema=$cursor{schema})\n";
9597 $db->db_disconnect($dbh);
9698 exit 0;
9799}
98100
99101# ---- Evaluate + deliver ---------------------------------------------
100102# HTTP transport is MODS::Webhook (shells out to /usr/bin/curl) so we
101103# don't need LWP::Protocol::https / IO::Socket::SSL as CPAN installs.
102104
103105my $delivered = 0;
104106my $skipped = 0;
105107my $failed = 0;
106108
107109foreach my $row (@file_new) { _process_row($row, 'file'); }
108110foreach my $row (@schema_new) { _process_row($row, 'schema'); }
111
112# ---- Frequency-mode rules ------------------------------------------
113# For any rule where frequency_threshold > 0, we ignore the per-row loop
114# above and instead: query how many file changes matching the rule's
115# path glob happened in the last window_min minutes. If it crosses the
116# threshold and we haven't fired within the same window, fire once with
117# the top offender.
118foreach my $rule (@rules) {
119 next unless ($rule->{frequency_threshold} || 0) > 0;
120 my $win = int($rule->{frequency_window_min} || 60);
121 my $thr = int($rule->{frequency_threshold});
122
123 # Re-fire cooldown: don't fire more than once per window
124 my $last = int($rule->{last_freq_fire_epoch} || 0);
125 if ($last && (time - $last) < $win * 60) {
126 next;
127 }
128
129 # Find files whose match_path_glob-hit change count > threshold
130 my $glob = $rule->{match_path_glob} // '';
131 my $glob_sql = '';
132 if (length $glob) {
133 # Approx: SQL LIKE. Convert glob wildcards to SQL % / _.
134 my $like = $glob;
135 $like =~ s/\*\*/%/g;
136 $like =~ s/\*/%/g;
137 $like =~ s/\?/_/g;
138 my $q_like = $dbh->quote($like);
139 $glob_sql = " AND fc.file_name LIKE $q_like";
140 }
141 my @hot = $db->db_readwrite_multiple($dbh, qq~
142 SELECT fc.file_name,
143 COUNT(*) AS ct,
144 MAX(fc.file_changes_id) AS latest_id,
145 s.server_name
146 FROM FILE_CHANGES fc
147 LEFT JOIN SERVERS s ON s.server_id = fc.server_id
148 WHERE fc.date_time >= DATE_SUB(NOW(), INTERVAL $win MINUTE)
149 $glob_sql
150 GROUP BY fc.file_name
151 HAVING ct >= $thr
152 ORDER BY ct DESC
153 LIMIT 5
154 ~, __FILE__, __LINE__);
155 next unless @hot;
156
157 # Fire once with the top offender's info
158 my $top = $hot[0];
159 my $fake_row = {
160 id => $top->{latest_id},
161 file_name => $top->{file_name},
162 server_name => $top->{server_name} // 'local',
163 date_time => POSIX::strftime('%Y-%m-%d %H:%M:%S', localtime),
164 ts_epoch => time,
165 status => 'frequency_alert',
166 };
167 my $payload = _build_payload($rule, $fake_row, 'file');
168 # Enrich summary with count + threshold
169 my $extra = "frequency alert: $top->{file_name} changed $top->{ct} times in the last $win min (threshold: $thr)";
170 if (ref($payload->{generic_webhook}) eq 'HASH') {
171 $payload->{generic_webhook}{frequency_summary} = $extra;
172 $payload->{generic_webhook}{frequency_count} = $top->{ct};
173 $payload->{generic_webhook}{frequency_window} = $win;
174 $payload->{generic_webhook}{summary} = $extra;
175 }
176 my ($status, $http_code, $body, $err) = _deliver($rule, $payload);
177 _log_delivery($rule, $fake_row, 'file', $status, $http_code, $body, $err);
178 if ($status eq 'sent') {
179 $delivered++;
180 $db->db_readwrite($dbh, qq~
181 UPDATE ALERT_RULES
182 SET fire_count = fire_count + 1,
183 last_fired_at = NOW(),
184 last_frequency_fire_at = NOW()
185 WHERE alert_rule_id = $rule->{alert_rule_id}
186 ~, __FILE__, __LINE__);
187 } else {
188 $failed++;
189 }
190}
109191
110192# ---- Advance cursor atomically --------------------------------------
111193if ($file_max > $cursor{file}) {
112194 $db->db_readwrite($dbh,
113195 "UPDATE ALERT_CURSOR SET last_seen_id = $file_max WHERE source_kind = 'file'",
114196 __FILE__, __LINE__);
115197}
116198if ($schema_max > $cursor{schema}) {
117199 $db->db_readwrite($dbh,
118200 "UPDATE ALERT_CURSOR SET last_seen_id = $schema_max WHERE source_kind = 'schema'",
119201 __FILE__, __LINE__);
120202}
121203
122204$db->db_disconnect($dbh);
123205print "[$ts] processed +$total_new new changes: $delivered delivered, $skipped skipped, $failed failed\n";
124206exit 0;
125207
126208#---------------------------------------------------------------------
127209sub _process_row {
128210 my ($row, $kind) = @_;
129211 foreach my $rule (@rules) {
130212 next unless _rule_matches($rule, $row, $kind);
131213
132214 # Rate limit check
133215 if ($rule->{rate_limit_per_min} > 0) {
134216 my $rlq = $db->db_readwrite($dbh, qq~
135217 SELECT COUNT(*) AS n FROM ALERT_DELIVERIES
136218 WHERE alert_rule_id = $rule->{alert_rule_id}
137219 AND fired_at >= DATE_SUB(NOW(), INTERVAL 1 MINUTE)
138220 AND delivery_status IN ('sent','pending')
139221 ~, __FILE__, __LINE__);
140222 if ($rlq && $rlq->{n} >= $rule->{rate_limit_per_min}) {
141223 _log_delivery($rule, $row, $kind, 'skipped', undef, undef,
142224 "rate limit ($rule->{rate_limit_per_min}/min) reached");
143225 $skipped++;
144226 next;
145227 }
146228 }
147229
148230 # Build payload + POST
149231 my $payload = _build_payload($rule, $row, $kind);
150232 my ($status, $http_code, $body, $err) = _deliver($rule, $payload);
151233 _log_delivery($rule, $row, $kind, $status, $http_code, $body, $err);
152234
153235 if ($status eq 'sent') { $delivered++; }
154236 elsif ($status eq 'skipped') { $skipped++; }
155237 else { $failed++; }
156238
157239 # Bump the rule's fire_count + last_fired_at
158240 if ($status eq 'sent') {
159241 $db->db_readwrite($dbh, qq~
160242 UPDATE ALERT_RULES
161243 SET fire_count = fire_count + 1,
162244 last_fired_at = NOW()
163245 WHERE alert_rule_id = $rule->{alert_rule_id}
164246 ~, __FILE__, __LINE__);
165247 }
166248 }
167249}
168250
169251#---------------------------------------------------------------------
170252sub _rule_matches {
171253 my ($rule, $row, $kind) = @_;
172254
173255 # Kind filter
174256 return 0 if $rule->{match_kind} ne 'either' && $rule->{match_kind} ne $kind;
175257
176258 # ts-only exclusion (files only)
177259 return 0 if $kind eq 'file' && $rule->{exclude_ts_only} && $row->{is_ts_only};
178260
179261 # Path glob
180262 my $glob = $rule->{match_path_glob} // '';
181263 if (length $glob) {
182264 my $target = $kind eq 'file'
183265 ? ($row->{file_name} // '')
184266 : (($row->{database_name} // '') . '.' . ($row->{table_name} // ''));
185267 my $re = _glob_to_regex($glob);
186268 return 0 unless $target =~ $re;
187269 }
188270
189271 # Status filter (files only)
190272 if ($kind eq 'file' && $rule->{match_status_list}) {
191273 my %allowed = map { s/^\s+|\s+$//g; $_ => 1 }
192274 split /,/, $rule->{match_status_list};
193275 return 0 unless $allowed{ $row->{status} // '' };
194276 }
195277
196278 return 1;
197279}
198280
199281sub _glob_to_regex {
200282 my $g = shift;
201283 my $re = quotemeta $g;
202284 # Handle glob patterns: escaped meta chars back to regex meaning
203285 $re =~ s{\\\*\\\*}{.*}g; # ** -> .*
204286 $re =~ s{\\\*}{[^/]*}g; # * -> [^/]*
205287 $re =~ s{\\\?}{.}g; # ? -> .
206288 return qr/^$re$/;
207289}
208290
209291#---------------------------------------------------------------------
210292sub _build_payload {
211293 my ($rule, $row, $kind) = @_;
212294
213295 my $where = $kind eq 'file'
214296 ? ($row->{file_name} // '(unknown file)')
215297 : (($row->{database_name} // '?') . '.' . ($row->{table_name} // '?'));
216298 my $when_utc = $row->{date_time} || $row->{change_datetime} || '';
217299 my $server = $row->{server_name} || 'local';
218300 my $status_h = $kind eq 'file' ? ($row->{status} // '') : 'DDL change';
219301 my $link_id = $row->{id};
220302 my $link = "$public_url/diff.cgi?kind=$kind&id=$link_id";
221303
222304 my $summary = $kind eq 'file'
223305 ? "$status_h: $where on $server"
224306 : "$where changed on $server";
225307
226308 my $slack = {
227309 text => "*DriftSense*: $summary",
228310 attachments => [{
229311 color => $kind eq 'schema' ? '#f59e0b' : '#14b8a6',
230312 fields => [
231313 { title => 'What', value => $where, short => 0 },
232314 { title => 'When', value => $when_utc, short => 1 },
233315 { title => 'Server', value => $server, short => 1 },
234316 ],
235317 actions => [{
236318 type => 'button', text => 'View diff', url => $link,
237319 }],
238320 }],
239321 };
240322 my $discord = {
241323 content => "**DriftSense**: $summary",
242324 embeds => [{
243325 title => $where,
244326 url => $link,
245327 color => $kind eq 'schema' ? 16101915 : 1349286, # amber, teal
246328 fields => [
247329 { name => 'When', value => $when_utc || '?', inline => \1 },
248330 { name => 'Server', value => $server, inline => \1 },
249331 ],
250332 }],
251333 };
252334 my $generic = {
253335 source => 'drift_sense',
254336 rule_id => $rule->{alert_rule_id},
255337 rule_name => $rule->{rule_name},
256338 kind => $kind,
257339 summary => $summary,
258340 target => $where,
259341 server => $server,
260342 when_utc => $when_utc,
261343 ts_epoch => $row->{ts_epoch},
262344 diff_url => $link,
263345 status => $status_h,
264346 };
265347
266348 return {
267349 slack => $slack,
268350 discord => $discord,
269351 generic_webhook => $generic,
270352 summary => $summary,
271353 };
272354}
273355
274356#---------------------------------------------------------------------
275357sub _deliver {
276358 my ($rule, $payload) = @_;
277359 my $kind = $rule->{delivery_kind};
278360 my $url = $rule->{delivery_url};
279361
280362 if ($kind eq 'email') {
281363 return ('skipped', undef, undef, 'email delivery not yet implemented');
282364 }
283365 unless (length($url // '')) {
284366 return ('failed', undef, undef, "no delivery URL configured");
285367 }
286368
287369 my $body_ref = $kind eq 'slack' ? $payload->{slack}
288370 : $kind eq 'discord' ? $payload->{discord}
289371 : $payload->{generic_webhook};
290372
291373 # MODS::Webhook handles JSON encoding + curl transport; returns
292374 # (http_code, response_body, error_message).
293375 my ($code, $rbody, $err) = MODS::Webhook::post_json(
294376 $url, $body_ref,
295377 timeout => 8,
296378 user_agent => 'DriftSense/1.0 (alert-dispatch)',
297379 );
298380 $rbody //= '';
299381 $rbody = substr($rbody, 0, 500) if length($rbody) > 500;
300382
301383 if (!$err && $code >= 200 && $code < 400) {
302384 return ('sent', $code, $rbody, undef);
303385 }
304386 return ('failed', $code, $rbody, $err || "HTTP $code");
305387}
306388
307389#---------------------------------------------------------------------
308390sub _log_delivery {
309391 my ($rule, $row, $kind, $status, $http_code, $body, $err) = @_;
310392 my $summary = substr(($rule->{rule_name} // '') . ' :: ' . ($row->{file_name}
311393 || ($row->{database_name} . '.' . $row->{table_name})
312394 || '?'), 0, 490);
313395 my $q_summary = $dbh->quote($summary);
314396 my $q_status = $dbh->quote($status);
315397 my $q_body = $dbh->quote($body // '');
316398 my $q_err = $dbh->quote($err // '');
317399 my $q_code = defined($http_code) ? int($http_code) : 'NULL';
318400
319401 $db->db_readwrite($dbh, qq~
320402 INSERT INTO ALERT_DELIVERIES
321403 (alert_rule_id, source_kind, source_id, match_summary,
322404 delivery_status, http_status, http_response, error_message)
323405 VALUES
324406 ($rule->{alert_rule_id}, '$kind', $row->{id}, $q_summary,
325407 $q_status, $q_code, $q_body, $q_err)
326408 ~, __FILE__, __LINE__);
327409}