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-11 19:38:00

Added
+13
lines
Removed
-18
lines
Context
314
unchanged
Blobs
from f7969a6939d9
to ad10d46443a1
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 ();
18use LWP::UserAgent;
19use HTTP::Request;
2018use JSON::PP;
2119use MODS::Config;
2220use MODS::DBConnect;
21use MODS::Webhook;
2322
2423$| = 1;
2524my $ts = POSIX::strftime('%Y-%m-%d %H:%M:%S', localtime);
2625
2726my $cfg = MODS::Config->new;
2827my $db = MODS::DBConnect->new;
2928my $dbh = $db->db_connect or die "[$ts] cannot connect to DriftSense storage DB\n";
3029
3130my $public_url = $cfg->settings('public_url') || 'https://watchtower.3dshawn.com';
3231
3332# ---- Fetch active rules ---------------------------------------------
3433my @rules = $db->db_readwrite_multiple($dbh, q~
3534 SELECT alert_rule_id, rule_name, match_kind, match_path_glob,
3635 match_status_list, exclude_ts_only,
3736 delivery_kind, delivery_url, delivery_email,
3837 rate_limit_per_min, coalesce_window_s
3938 FROM ALERT_RULES
4039 WHERE is_active = 1
4140~, __FILE__, __LINE__);
4241
4342unless (@rules) {
4443 print "[$ts] no active alert rules, exiting\n";
4544 $db->db_disconnect($dbh);
4645 exit 0;
4746}
4847
4948# ---- Fetch cursor ---------------------------------------------------
5049my %cursor;
5150foreach my $r ($db->db_readwrite_multiple($dbh, q~
5251 SELECT source_kind, last_seen_id FROM ALERT_CURSOR
5352~, __FILE__, __LINE__)) {
5453 $cursor{$r->{source_kind}} = $r->{last_seen_id};
5554}
5655$cursor{file} //= 0;
5756$cursor{schema} //= 0;
5857
5958# ---- Pull new FILE_CHANGES since cursor ------------------------------
6059my $file_max = $cursor{file};
6160my @file_new = $db->db_readwrite_multiple($dbh, qq~
6261 SELECT fc.file_changes_id AS id, fc.file_name, fc.status, fc.blob_sha,
6362 fc.is_ts_only, fc.date_time,
6463 UNIX_TIMESTAMP(fc.date_time) AS ts_epoch,
6564 s.server_name
6665 FROM FILE_CHANGES fc
6766 LEFT JOIN SERVERS s ON s.server_id = fc.server_id
6867 WHERE fc.file_changes_id > $cursor{file}
6968 ORDER BY fc.file_changes_id ASC
7069 LIMIT 500
7170~, __FILE__, __LINE__);
7271foreach my $r (@file_new) {
7372 $file_max = $r->{id} if $r->{id} > $file_max;
7473}
7574
7675# ---- Pull new SCHEMA_CHANGE since cursor -----------------------------
7776my $schema_max = $cursor{schema};
7877my @schema_new = $db->db_readwrite_multiple($dbh, qq~
7978 SELECT sc.schema_change_id AS id, sc.database_name, sc.table_name,
8079 sc.changes, sc.is_ts_only, sc.change_datetime,
8180 UNIX_TIMESTAMP(sc.change_datetime) AS ts_epoch,
8281 s.server_name
8382 FROM SCHEMA_CHANGE sc
8483 LEFT JOIN SERVERS s ON s.server_id = sc.server_id
8584 WHERE sc.schema_change_id > $cursor{schema}
8685 ORDER BY sc.schema_change_id ASC
8786 LIMIT 500
8887~, __FILE__, __LINE__);
8988foreach my $r (@schema_new) {
9089 $schema_max = $r->{id} if $r->{id} > $schema_max;
9190}
9291
9392my $total_new = scalar(@file_new) + scalar(@schema_new);
9493if ($total_new == 0) {
9594 print "[$ts] no new changes since cursor (file=$cursor{file} schema=$cursor{schema})\n";
9695 $db->db_disconnect($dbh);
9796 exit 0;
9897}
9998
10099# ---- Evaluate + deliver ---------------------------------------------
101my $ua = LWP::UserAgent->new(
102 timeout => 8,
103 agent => 'DriftSense/1.0 (alert-dispatch)',
104);
105$ua->ssl_opts(verify_hostname => 0); # allow self-signed test endpoints
100# HTTP transport is MODS::Webhook (shells out to /usr/bin/curl) so we
101# don't need LWP::Protocol::https / IO::Socket::SSL as CPAN installs.
106102
107103my $delivered = 0;
108104my $skipped = 0;
109105my $failed = 0;
110106
111107foreach my $row (@file_new) { _process_row($row, 'file'); }
112108foreach my $row (@schema_new) { _process_row($row, 'schema'); }
113109
114110# ---- Advance cursor atomically --------------------------------------
115111if ($file_max > $cursor{file}) {
116112 $db->db_readwrite($dbh,
117113 "UPDATE ALERT_CURSOR SET last_seen_id = $file_max WHERE source_kind = 'file'",
118114 __FILE__, __LINE__);
119115}
120116if ($schema_max > $cursor{schema}) {
121117 $db->db_readwrite($dbh,
122118 "UPDATE ALERT_CURSOR SET last_seen_id = $schema_max WHERE source_kind = 'schema'",
123119 __FILE__, __LINE__);
124120}
125121
126122$db->db_disconnect($dbh);
127123print "[$ts] processed +$total_new new changes: $delivered delivered, $skipped skipped, $failed failed\n";
128124exit 0;
129125
130126#---------------------------------------------------------------------
131127sub _process_row {
132128 my ($row, $kind) = @_;
133129 foreach my $rule (@rules) {
134130 next unless _rule_matches($rule, $row, $kind);
135131
136132 # Rate limit check
137133 if ($rule->{rate_limit_per_min} > 0) {
138134 my $rlq = $db->db_readwrite($dbh, qq~
139135 SELECT COUNT(*) AS n FROM ALERT_DELIVERIES
140136 WHERE alert_rule_id = $rule->{alert_rule_id}
141137 AND fired_at >= DATE_SUB(NOW(), INTERVAL 1 MINUTE)
142138 AND delivery_status IN ('sent','pending')
143139 ~, __FILE__, __LINE__);
144140 if ($rlq && $rlq->{n} >= $rule->{rate_limit_per_min}) {
145141 _log_delivery($rule, $row, $kind, 'skipped', undef, undef,
146142 "rate limit ($rule->{rate_limit_per_min}/min) reached");
147143 $skipped++;
148144 next;
149145 }
150146 }
151147
152148 # Build payload + POST
153149 my $payload = _build_payload($rule, $row, $kind);
154150 my ($status, $http_code, $body, $err) = _deliver($rule, $payload);
155151 _log_delivery($rule, $row, $kind, $status, $http_code, $body, $err);
156152
157153 if ($status eq 'sent') { $delivered++; }
158154 elsif ($status eq 'skipped') { $skipped++; }
159155 else { $failed++; }
160156
161157 # Bump the rule's fire_count + last_fired_at
162158 if ($status eq 'sent') {
163159 $db->db_readwrite($dbh, qq~
164160 UPDATE ALERT_RULES
165161 SET fire_count = fire_count + 1,
166162 last_fired_at = NOW()
167163 WHERE alert_rule_id = $rule->{alert_rule_id}
168164 ~, __FILE__, __LINE__);
169165 }
170166 }
171167}
172168
173169#---------------------------------------------------------------------
174170sub _rule_matches {
175171 my ($rule, $row, $kind) = @_;
176172
177173 # Kind filter
178174 return 0 if $rule->{match_kind} ne 'either' && $rule->{match_kind} ne $kind;
179175
180176 # ts-only exclusion (files only)
181177 return 0 if $kind eq 'file' && $rule->{exclude_ts_only} && $row->{is_ts_only};
182178
183179 # Path glob
184180 my $glob = $rule->{match_path_glob} // '';
185181 if (length $glob) {
186182 my $target = $kind eq 'file'
187183 ? ($row->{file_name} // '')
188184 : (($row->{database_name} // '') . '.' . ($row->{table_name} // ''));
189185 my $re = _glob_to_regex($glob);
190186 return 0 unless $target =~ $re;
191187 }
192188
193189 # Status filter (files only)
194190 if ($kind eq 'file' && $rule->{match_status_list}) {
195191 my %allowed = map { s/^\s+|\s+$//g; $_ => 1 }
196192 split /,/, $rule->{match_status_list};
197193 return 0 unless $allowed{ $row->{status} // '' };
198194 }
199195
200196 return 1;
201197}
202198
203199sub _glob_to_regex {
204200 my $g = shift;
205201 my $re = quotemeta $g;
206202 # Handle glob patterns: escaped meta chars back to regex meaning
207203 $re =~ s{\\\*\\\*}{.*}g; # ** -> .*
208204 $re =~ s{\\\*}{[^/]*}g; # * -> [^/]*
209205 $re =~ s{\\\?}{.}g; # ? -> .
210206 return qr/^$re$/;
211207}
212208
213209#---------------------------------------------------------------------
214210sub _build_payload {
215211 my ($rule, $row, $kind) = @_;
216212
217213 my $where = $kind eq 'file'
218214 ? ($row->{file_name} // '(unknown file)')
219215 : (($row->{database_name} // '?') . '.' . ($row->{table_name} // '?'));
220216 my $when_utc = $row->{date_time} || $row->{change_datetime} || '';
221217 my $server = $row->{server_name} || 'local';
222218 my $status_h = $kind eq 'file' ? ($row->{status} // '') : 'DDL change';
223219 my $link_id = $row->{id};
224220 my $link = "$public_url/diff.cgi?kind=$kind&id=$link_id";
225221
226222 my $summary = $kind eq 'file'
227223 ? "$status_h: $where on $server"
228224 : "$where changed on $server";
229225
230226 my $slack = {
231227 text => "*DriftSense*: $summary",
232228 attachments => [{
233229 color => $kind eq 'schema' ? '#f59e0b' : '#14b8a6',
234230 fields => [
235231 { title => 'What', value => $where, short => 0 },
236232 { title => 'When', value => $when_utc, short => 1 },
237233 { title => 'Server', value => $server, short => 1 },
238234 ],
239235 actions => [{
240236 type => 'button', text => 'View diff', url => $link,
241237 }],
242238 }],
243239 };
244240 my $discord = {
245241 content => "**DriftSense**: $summary",
246242 embeds => [{
247243 title => $where,
248244 url => $link,
249245 color => $kind eq 'schema' ? 16101915 : 1349286, # amber, teal
250246 fields => [
251247 { name => 'When', value => $when_utc || '?', inline => \1 },
252248 { name => 'Server', value => $server, inline => \1 },
253249 ],
254250 }],
255251 };
256252 my $generic = {
257253 source => 'drift_sense',
258254 rule_id => $rule->{alert_rule_id},
259255 rule_name => $rule->{rule_name},
260256 kind => $kind,
261257 summary => $summary,
262258 target => $where,
263259 server => $server,
264260 when_utc => $when_utc,
265261 ts_epoch => $row->{ts_epoch},
266262 diff_url => $link,
267263 status => $status_h,
268264 };
269265
270266 return {
271267 slack => $slack,
272268 discord => $discord,
273269 generic_webhook => $generic,
274270 summary => $summary,
275271 };
276272}
277273
278274#---------------------------------------------------------------------
279275sub _deliver {
280276 my ($rule, $payload) = @_;
281277 my $kind = $rule->{delivery_kind};
282278 my $url = $rule->{delivery_url};
283279
284280 if ($kind eq 'email') {
285281 return ('skipped', undef, undef, 'email delivery not yet implemented');
286282 }
287283 unless (length($url // '')) {
288284 return ('failed', undef, undef, "no delivery URL configured");
289285 }
290286
291287 my $body_ref = $kind eq 'slack' ? $payload->{slack}
292288 : $kind eq 'discord' ? $payload->{discord}
293289 : $payload->{generic_webhook};
294 my $json = eval { encode_json($body_ref) };
295 if ($@) { return ('failed', undef, undef, "encode_json: $@"); }
296
297 my $req = HTTP::Request->new(POST => $url);
298 $req->header('Content-Type' => 'application/json');
299 $req->content($json);
300290
301 my $resp = $ua->request($req);
302 my $code = $resp->code;
303 my $rbody = $resp->decoded_content // '';
291 # MODS::Webhook handles JSON encoding + curl transport; returns
292 # (http_code, response_body, error_message).
293 my ($code, $rbody, $err) = MODS::Webhook::post_json(
294 $url, $body_ref,
295 timeout => 8,
296 user_agent => 'DriftSense/1.0 (alert-dispatch)',
297 );
298 $rbody //= '';
304299 $rbody = substr($rbody, 0, 500) if length($rbody) > 500;
305300
306 if ($resp->is_success) {
301 if (!$err && $code >= 200 && $code < 400) {
307302 return ('sent', $code, $rbody, undef);
308303 }
309 return ('failed', $code, $rbody, $resp->status_line);
304 return ('failed', $code, $rbody, $err || "HTTP $code");
310305}
311306
312307#---------------------------------------------------------------------
313308sub _log_delivery {
314309 my ($rule, $row, $kind, $status, $http_code, $body, $err) = @_;
315310 my $summary = substr(($rule->{rule_name} // '') . ' :: ' . ($row->{file_name}
316311 || ($row->{database_name} . '.' . $row->{table_name})
317312 || '?'), 0, 490);
318313 my $q_summary = $dbh->quote($summary);
319314 my $q_status = $dbh->quote($status);
320315 my $q_body = $dbh->quote($body // '');
321316 my $q_err = $dbh->quote($err // '');
322317 my $q_code = defined($http_code) ? int($http_code) : 'NULL';
323318
324319 $db->db_readwrite($dbh, qq~
325320 INSERT INTO ALERT_DELIVERIES
326321 (alert_rule_id, source_kind, source_id, match_summary,
327322 delivery_status, http_status, http_response, error_message)
328323 VALUES
329324 ($rule->{alert_rule_id}, '$kind', $row->{id}, $q_summary,
330325 $q_status, $q_code, $q_body, $q_err)
331326 ~, __FILE__, __LINE__);
332327}