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

O Operator
Diff

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

added on local at 2026-07-11 18:27:42

Added
+332
lines
Removed
-0
lines
Context
0
unchanged
Blobs
from
to f7969a6939d9
Restore this content →
Unified diff
Naive LCS-based line diff. Additions in emerald, deletions in rose. Both sides decompressed live from BLOB_STORE.
1#!/usr/bin/perl
2#======================================================================
3# DriftSense -- alert dispatcher
4#
5# Cron entry point (NOT a CGI). Walks FILE_CHANGES + SCHEMA_CHANGE rows
6# inserted since the last cursor tick, evaluates every ACTIVE rule
7# against each new row, and POSTs matching payloads to the configured
8# webhook (Slack / Discord / generic HTTP). Every attempt is logged
9# to ALERT_DELIVERIES for the operator's audit trail.
10#
11# Config: /etc/drift_sense/drift_sense.conf (via MODS::Config).
12# Log: /var/log/drift_sense/alert_dispatch.log (via cron redirect).
13#======================================================================
14use strict;
15use warnings;
16use lib '/var/www/vhosts/3dshawn.com/site1';
17use POSIX ();
18use LWP::UserAgent;
19use HTTP::Request;
20use JSON::PP;
21use MODS::Config;
22use MODS::DBConnect;
23
24$| = 1;
25my $ts = POSIX::strftime('%Y-%m-%d %H:%M:%S', localtime);
26
27my $cfg = MODS::Config->new;
28my $db = MODS::DBConnect->new;
29my $dbh = $db->db_connect or die "[$ts] cannot connect to DriftSense storage DB\n";
30
31my $public_url = $cfg->settings('public_url') || 'https://watchtower.3dshawn.com';
32
33# ---- Fetch active rules ---------------------------------------------
34my @rules = $db->db_readwrite_multiple($dbh, q~
35 SELECT alert_rule_id, rule_name, match_kind, match_path_glob,
36 match_status_list, exclude_ts_only,
37 delivery_kind, delivery_url, delivery_email,
38 rate_limit_per_min, coalesce_window_s
39 FROM ALERT_RULES
40 WHERE is_active = 1
41~, __FILE__, __LINE__);
42
43unless (@rules) {
44 print "[$ts] no active alert rules, exiting\n";
45 $db->db_disconnect($dbh);
46 exit 0;
47}
48
49# ---- Fetch cursor ---------------------------------------------------
50my %cursor;
51foreach my $r ($db->db_readwrite_multiple($dbh, q~
52 SELECT source_kind, last_seen_id FROM ALERT_CURSOR
53~, __FILE__, __LINE__)) {
54 $cursor{$r->{source_kind}} = $r->{last_seen_id};
55}
56$cursor{file} //= 0;
57$cursor{schema} //= 0;
58
59# ---- Pull new FILE_CHANGES since cursor ------------------------------
60my $file_max = $cursor{file};
61my @file_new = $db->db_readwrite_multiple($dbh, qq~
62 SELECT fc.file_changes_id AS id, fc.file_name, fc.status, fc.blob_sha,
63 fc.is_ts_only, fc.date_time,
64 UNIX_TIMESTAMP(fc.date_time) AS ts_epoch,
65 s.server_name
66 FROM FILE_CHANGES fc
67 LEFT JOIN SERVERS s ON s.server_id = fc.server_id
68 WHERE fc.file_changes_id > $cursor{file}
69 ORDER BY fc.file_changes_id ASC
70 LIMIT 500
71~, __FILE__, __LINE__);
72foreach my $r (@file_new) {
73 $file_max = $r->{id} if $r->{id} > $file_max;
74}
75
76# ---- Pull new SCHEMA_CHANGE since cursor -----------------------------
77my $schema_max = $cursor{schema};
78my @schema_new = $db->db_readwrite_multiple($dbh, qq~
79 SELECT sc.schema_change_id AS id, sc.database_name, sc.table_name,
80 sc.changes, sc.is_ts_only, sc.change_datetime,
81 UNIX_TIMESTAMP(sc.change_datetime) AS ts_epoch,
82 s.server_name
83 FROM SCHEMA_CHANGE sc
84 LEFT JOIN SERVERS s ON s.server_id = sc.server_id
85 WHERE sc.schema_change_id > $cursor{schema}
86 ORDER BY sc.schema_change_id ASC
87 LIMIT 500
88~, __FILE__, __LINE__);
89foreach my $r (@schema_new) {
90 $schema_max = $r->{id} if $r->{id} > $schema_max;
91}
92
93my $total_new = scalar(@file_new) + scalar(@schema_new);
94if ($total_new == 0) {
95 print "[$ts] no new changes since cursor (file=$cursor{file} schema=$cursor{schema})\n";
96 $db->db_disconnect($dbh);
97 exit 0;
98}
99
100# ---- 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
106
107my $delivered = 0;
108my $skipped = 0;
109my $failed = 0;
110
111foreach my $row (@file_new) { _process_row($row, 'file'); }
112foreach my $row (@schema_new) { _process_row($row, 'schema'); }
113
114# ---- Advance cursor atomically --------------------------------------
115if ($file_max > $cursor{file}) {
116 $db->db_readwrite($dbh,
117 "UPDATE ALERT_CURSOR SET last_seen_id = $file_max WHERE source_kind = 'file'",
118 __FILE__, __LINE__);
119}
120if ($schema_max > $cursor{schema}) {
121 $db->db_readwrite($dbh,
122 "UPDATE ALERT_CURSOR SET last_seen_id = $schema_max WHERE source_kind = 'schema'",
123 __FILE__, __LINE__);
124}
125
126$db->db_disconnect($dbh);
127print "[$ts] processed +$total_new new changes: $delivered delivered, $skipped skipped, $failed failed\n";
128exit 0;
129
130#---------------------------------------------------------------------
131sub _process_row {
132 my ($row, $kind) = @_;
133 foreach my $rule (@rules) {
134 next unless _rule_matches($rule, $row, $kind);
135
136 # Rate limit check
137 if ($rule->{rate_limit_per_min} > 0) {
138 my $rlq = $db->db_readwrite($dbh, qq~
139 SELECT COUNT(*) AS n FROM ALERT_DELIVERIES
140 WHERE alert_rule_id = $rule->{alert_rule_id}
141 AND fired_at >= DATE_SUB(NOW(), INTERVAL 1 MINUTE)
142 AND delivery_status IN ('sent','pending')
143 ~, __FILE__, __LINE__);
144 if ($rlq && $rlq->{n} >= $rule->{rate_limit_per_min}) {
145 _log_delivery($rule, $row, $kind, 'skipped', undef, undef,
146 "rate limit ($rule->{rate_limit_per_min}/min) reached");
147 $skipped++;
148 next;
149 }
150 }
151
152 # Build payload + POST
153 my $payload = _build_payload($rule, $row, $kind);
154 my ($status, $http_code, $body, $err) = _deliver($rule, $payload);
155 _log_delivery($rule, $row, $kind, $status, $http_code, $body, $err);
156
157 if ($status eq 'sent') { $delivered++; }
158 elsif ($status eq 'skipped') { $skipped++; }
159 else { $failed++; }
160
161 # Bump the rule's fire_count + last_fired_at
162 if ($status eq 'sent') {
163 $db->db_readwrite($dbh, qq~
164 UPDATE ALERT_RULES
165 SET fire_count = fire_count + 1,
166 last_fired_at = NOW()
167 WHERE alert_rule_id = $rule->{alert_rule_id}
168 ~, __FILE__, __LINE__);
169 }
170 }
171}
172
173#---------------------------------------------------------------------
174sub _rule_matches {
175 my ($rule, $row, $kind) = @_;
176
177 # Kind filter
178 return 0 if $rule->{match_kind} ne 'either' && $rule->{match_kind} ne $kind;
179
180 # ts-only exclusion (files only)
181 return 0 if $kind eq 'file' && $rule->{exclude_ts_only} && $row->{is_ts_only};
182
183 # Path glob
184 my $glob = $rule->{match_path_glob} // '';
185 if (length $glob) {
186 my $target = $kind eq 'file'
187 ? ($row->{file_name} // '')
188 : (($row->{database_name} // '') . '.' . ($row->{table_name} // ''));
189 my $re = _glob_to_regex($glob);
190 return 0 unless $target =~ $re;
191 }
192
193 # Status filter (files only)
194 if ($kind eq 'file' && $rule->{match_status_list}) {
195 my %allowed = map { s/^\s+|\s+$//g; $_ => 1 }
196 split /,/, $rule->{match_status_list};
197 return 0 unless $allowed{ $row->{status} // '' };
198 }
199
200 return 1;
201}
202
203sub _glob_to_regex {
204 my $g = shift;
205 my $re = quotemeta $g;
206 # Handle glob patterns: escaped meta chars back to regex meaning
207 $re =~ s{\\\*\\\*}{.*}g; # ** -> .*
208 $re =~ s{\\\*}{[^/]*}g; # * -> [^/]*
209 $re =~ s{\\\?}{.}g; # ? -> .
210 return qr/^$re$/;
211}
212
213#---------------------------------------------------------------------
214sub _build_payload {
215 my ($rule, $row, $kind) = @_;
216
217 my $where = $kind eq 'file'
218 ? ($row->{file_name} // '(unknown file)')
219 : (($row->{database_name} // '?') . '.' . ($row->{table_name} // '?'));
220 my $when_utc = $row->{date_time} || $row->{change_datetime} || '';
221 my $server = $row->{server_name} || 'local';
222 my $status_h = $kind eq 'file' ? ($row->{status} // '') : 'DDL change';
223 my $link_id = $row->{id};
224 my $link = "$public_url/diff.cgi?kind=$kind&id=$link_id";
225
226 my $summary = $kind eq 'file'
227 ? "$status_h: $where on $server"
228 : "$where changed on $server";
229
230 my $slack = {
231 text => "*DriftSense*: $summary",
232 attachments => [{
233 color => $kind eq 'schema' ? '#f59e0b' : '#14b8a6',
234 fields => [
235 { title => 'What', value => $where, short => 0 },
236 { title => 'When', value => $when_utc, short => 1 },
237 { title => 'Server', value => $server, short => 1 },
238 ],
239 actions => [{
240 type => 'button', text => 'View diff', url => $link,
241 }],
242 }],
243 };
244 my $discord = {
245 content => "**DriftSense**: $summary",
246 embeds => [{
247 title => $where,
248 url => $link,
249 color => $kind eq 'schema' ? 16101915 : 1349286, # amber, teal
250 fields => [
251 { name => 'When', value => $when_utc || '?', inline => \1 },
252 { name => 'Server', value => $server, inline => \1 },
253 ],
254 }],
255 };
256 my $generic = {
257 source => 'drift_sense',
258 rule_id => $rule->{alert_rule_id},
259 rule_name => $rule->{rule_name},
260 kind => $kind,
261 summary => $summary,
262 target => $where,
263 server => $server,
264 when_utc => $when_utc,
265 ts_epoch => $row->{ts_epoch},
266 diff_url => $link,
267 status => $status_h,
268 };
269
270 return {
271 slack => $slack,
272 discord => $discord,
273 generic_webhook => $generic,
274 summary => $summary,
275 };
276}
277
278#---------------------------------------------------------------------
279sub _deliver {
280 my ($rule, $payload) = @_;
281 my $kind = $rule->{delivery_kind};
282 my $url = $rule->{delivery_url};
283
284 if ($kind eq 'email') {
285 return ('skipped', undef, undef, 'email delivery not yet implemented');
286 }
287 unless (length($url // '')) {
288 return ('failed', undef, undef, "no delivery URL configured");
289 }
290
291 my $body_ref = $kind eq 'slack' ? $payload->{slack}
292 : $kind eq 'discord' ? $payload->{discord}
293 : $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);
300
301 my $resp = $ua->request($req);
302 my $code = $resp->code;
303 my $rbody = $resp->decoded_content // '';
304 $rbody = substr($rbody, 0, 500) if length($rbody) > 500;
305
306 if ($resp->is_success) {
307 return ('sent', $code, $rbody, undef);
308 }
309 return ('failed', $code, $rbody, $resp->status_line);
310}
311
312#---------------------------------------------------------------------
313sub _log_delivery {
314 my ($rule, $row, $kind, $status, $http_code, $body, $err) = @_;
315 my $summary = substr(($rule->{rule_name} // '') . ' :: ' . ($row->{file_name}
316 || ($row->{database_name} . '.' . $row->{table_name})
317 || '?'), 0, 490);
318 my $q_summary = $dbh->quote($summary);
319 my $q_status = $dbh->quote($status);
320 my $q_body = $dbh->quote($body // '');
321 my $q_err = $dbh->quote($err // '');
322 my $q_code = defined($http_code) ? int($http_code) : 'NULL';
323
324 $db->db_readwrite($dbh, qq~
325 INSERT INTO ALERT_DELIVERIES
326 (alert_rule_id, source_kind, source_id, match_summary,
327 delivery_status, http_status, http_response, error_message)
328 VALUES
329 ($rule->{alert_rule_id}, '$kind', $row->{id}, $q_summary,
330 $q_status, $q_code, $q_body, $q_err)
331 ~, __FILE__, __LINE__);
332}