added on local at 2026-07-13 13:59:34
| 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 | #====================================================================== | |
| 14 | use strict; | |
| 15 | use warnings; | |
| 16 | use lib '/var/www/vhosts/3dshawn.com/site1'; | |
| 17 | use POSIX (); | |
| 18 | use Symbol (); | |
| 19 | use JSON::PP; | |
| 20 | use MODS::Config; | |
| 21 | use MODS::DBConnect; | |
| 22 | use MODS::Webhook; | |
| 23 | ||
| 24 | $| = 1; | |
| 25 | my $ts = POSIX::strftime('%Y-%m-%d %H:%M:%S', localtime); | |
| 26 | ||
| 27 | my $cfg = MODS::Config->new; | |
| 28 | my $db = MODS::DBConnect->new; | |
| 29 | my $dbh = $db->db_connect or die "[$ts] cannot connect to DriftSense storage DB\n"; | |
| 30 | ||
| 31 | my $public_url = $cfg->settings('public_url') || 'https://driftsense.3dshawn.com'; | |
| 32 | ||
| 33 | # ---- Quiet hours --------------------------------------------------- | |
| 34 | # Which monitors are in a quiet window right now? A window that straddles | |
| 35 | # midnight (e.g. 22:00 -> 06:00) is active when now >= start OR now < end. | |
| 36 | my %QUIET_MID; | |
| 37 | { | |
| 38 | my $probe = $db->db_readwrite($dbh, q~ | |
| 39 | SELECT 1 AS ok FROM INFORMATION_SCHEMA.COLUMNS | |
| 40 | WHERE TABLE_SCHEMA=DATABASE() | |
| 41 | AND TABLE_NAME='FILE_MONITOR_SETTINGS' | |
| 42 | AND COLUMN_NAME='quiet_hours_start' | |
| 43 | ~, __FILE__, __LINE__); | |
| 44 | if ($probe && $probe->{ok}) { | |
| 45 | my @rows = $db->db_readwrite_multiple($dbh, q~ | |
| 46 | SELECT file_monitor_list_id AS mid, | |
| 47 | TIME_TO_SEC(quiet_hours_start) AS start_s, | |
| 48 | TIME_TO_SEC(quiet_hours_end) AS end_s | |
| 49 | FROM FILE_MONITOR_SETTINGS | |
| 50 | WHERE quiet_hours_start IS NOT NULL | |
| 51 | AND quiet_hours_end IS NOT NULL | |
| 52 | ~, __FILE__, __LINE__); | |
| 53 | my $now_row = $db->db_readwrite($dbh, | |
| 54 | "SELECT TIME_TO_SEC(CURTIME()) AS now_s", __FILE__, __LINE__); | |
| 55 | my $now = $now_row ? $now_row->{now_s} : 0; | |
| 56 | foreach my $r (@rows) { | |
| 57 | my $s = $r->{start_s}; my $e = $r->{end_s}; | |
| 58 | my $quiet = ($s <= $e) ? ($now >= $s && $now < $e) | |
| 59 | : ($now >= $s || $now < $e); | |
| 60 | $QUIET_MID{$r->{mid}} = 1 if $quiet; | |
| 61 | } | |
| 62 | if (%QUIET_MID) { | |
| 63 | print "[$ts] quiet-hours active for monitors: " . join(',', sort keys %QUIET_MID) . "\n"; | |
| 64 | } | |
| 65 | } | |
| 66 | } | |
| 67 | sub _monitor_is_quiet { my $mid = shift; return $QUIET_MID{$mid || 0} ? 1 : 0; } | |
| 68 | ||
| 69 | # ---- Fetch active rules --------------------------------------------- | |
| 70 | my @rules = $db->db_readwrite_multiple($dbh, q~ | |
| 71 | SELECT alert_rule_id, rule_name, match_kind, match_path_glob, | |
| 72 | match_status_list, exclude_ts_only, | |
| 73 | delivery_kind, delivery_url, delivery_email, | |
| 74 | rate_limit_per_min, coalesce_window_s, | |
| 75 | frequency_threshold, frequency_window_min, | |
| 76 | UNIX_TIMESTAMP(last_frequency_fire_at) AS last_freq_fire_epoch | |
| 77 | FROM ALERT_RULES | |
| 78 | WHERE is_active = 1 | |
| 79 | ~, __FILE__, __LINE__); | |
| 80 | ||
| 81 | unless (@rules) { | |
| 82 | print "[$ts] no active alert rules, exiting\n"; | |
| 83 | $db->db_disconnect($dbh); | |
| 84 | exit 0; | |
| 85 | } | |
| 86 | ||
| 87 | # ---- Fetch cursor --------------------------------------------------- | |
| 88 | my %cursor; | |
| 89 | foreach my $r ($db->db_readwrite_multiple($dbh, q~ | |
| 90 | SELECT source_kind, last_seen_id FROM ALERT_CURSOR | |
| 91 | ~, __FILE__, __LINE__)) { | |
| 92 | $cursor{$r->{source_kind}} = $r->{last_seen_id}; | |
| 93 | } | |
| 94 | $cursor{file} //= 0; | |
| 95 | $cursor{schema} //= 0; | |
| 96 | ||
| 97 | # ---- Pull new FILE_CHANGES since cursor ------------------------------ | |
| 98 | my $file_max = $cursor{file}; | |
| 99 | my @file_new = $db->db_readwrite_multiple($dbh, qq~ | |
| 100 | SELECT fc.file_changes_id AS id, fc.file_name, fc.status, fc.blob_sha, | |
| 101 | fc.is_ts_only, fc.date_time, | |
| 102 | UNIX_TIMESTAMP(fc.date_time) AS ts_epoch, | |
| 103 | fc.file_monitor_list_id AS mid, | |
| 104 | s.server_name | |
| 105 | FROM FILE_CHANGES fc | |
| 106 | LEFT JOIN SERVERS s ON s.server_id = fc.server_id | |
| 107 | WHERE fc.file_changes_id > $cursor{file} | |
| 108 | ORDER BY fc.file_changes_id ASC | |
| 109 | LIMIT 500 | |
| 110 | ~, __FILE__, __LINE__); | |
| 111 | foreach my $r (@file_new) { | |
| 112 | $file_max = $r->{id} if $r->{id} > $file_max; | |
| 113 | } | |
| 114 | ||
| 115 | # Quiet-hours filter: still advance cursor past silenced rows so we don't | |
| 116 | # fire them once the window ends, but drop them from the per-rule loop. | |
| 117 | if (%QUIET_MID) { | |
| 118 | my $before = scalar @file_new; | |
| 119 | @file_new = grep { !_monitor_is_quiet($_->{mid}) } @file_new; | |
| 120 | my $dropped = $before - scalar @file_new; | |
| 121 | print "[$ts] quiet-hours filtered $dropped file change(s) out of alert dispatch\n" if $dropped; | |
| 122 | } | |
| 123 | ||
| 124 | # ---- Pull new SCHEMA_CHANGE since cursor ----------------------------- | |
| 125 | my $schema_max = $cursor{schema}; | |
| 126 | my @schema_new = $db->db_readwrite_multiple($dbh, qq~ | |
| 127 | SELECT sc.schema_change_id AS id, sc.database_name, sc.table_name, | |
| 128 | sc.changes, sc.is_ts_only, sc.change_datetime, | |
| 129 | UNIX_TIMESTAMP(sc.change_datetime) AS ts_epoch, | |
| 130 | s.server_name | |
| 131 | FROM SCHEMA_CHANGE sc | |
| 132 | LEFT JOIN SERVERS s ON s.server_id = sc.server_id | |
| 133 | WHERE sc.schema_change_id > $cursor{schema} | |
| 134 | ORDER BY sc.schema_change_id ASC | |
| 135 | LIMIT 500 | |
| 136 | ~, __FILE__, __LINE__); | |
| 137 | foreach my $r (@schema_new) { | |
| 138 | $schema_max = $r->{id} if $r->{id} > $schema_max; | |
| 139 | } | |
| 140 | ||
| 141 | my $total_new = scalar(@file_new) + scalar(@schema_new); | |
| 142 | if ($total_new == 0) { | |
| 143 | print "[$ts] no new changes since cursor (file=$cursor{file} schema=$cursor{schema})\n"; | |
| 144 | $db->db_disconnect($dbh); | |
| 145 | exit 0; | |
| 146 | } | |
| 147 | ||
| 148 | # ---- Evaluate + deliver --------------------------------------------- | |
| 149 | # HTTP transport is MODS::Webhook (shells out to /usr/bin/curl) so we | |
| 150 | # don't need LWP::Protocol::https / IO::Socket::SSL as CPAN installs. | |
| 151 | ||
| 152 | my $delivered = 0; | |
| 153 | my $skipped = 0; | |
| 154 | my $failed = 0; | |
| 155 | ||
| 156 | foreach my $row (@file_new) { _process_row($row, 'file'); } | |
| 157 | foreach my $row (@schema_new) { _process_row($row, 'schema'); } | |
| 158 | ||
| 159 | # ---- Frequency-mode rules ------------------------------------------ | |
| 160 | # For any rule where frequency_threshold > 0, we ignore the per-row loop | |
| 161 | # above and instead: query how many file changes matching the rule's | |
| 162 | # path glob happened in the last window_min minutes. If it crosses the | |
| 163 | # threshold and we haven't fired within the same window, fire once with | |
| 164 | # the top offender. | |
| 165 | foreach my $rule (@rules) { | |
| 166 | next unless ($rule->{frequency_threshold} || 0) > 0; | |
| 167 | my $win = int($rule->{frequency_window_min} || 60); | |
| 168 | my $thr = int($rule->{frequency_threshold}); | |
| 169 | ||
| 170 | # Re-fire cooldown: don't fire more than once per window | |
| 171 | my $last = int($rule->{last_freq_fire_epoch} || 0); | |
| 172 | if ($last && (time - $last) < $win * 60) { | |
| 173 | next; | |
| 174 | } | |
| 175 | ||
| 176 | # Find files whose match_path_glob-hit change count > threshold | |
| 177 | my $glob = $rule->{match_path_glob} // ''; | |
| 178 | my $glob_sql = ''; | |
| 179 | if (length $glob) { | |
| 180 | # Approx: SQL LIKE. Convert glob wildcards to SQL % / _. | |
| 181 | my $like = $glob; | |
| 182 | $like =~ s/\*\*/%/g; | |
| 183 | $like =~ s/\*/%/g; | |
| 184 | $like =~ s/\?/_/g; | |
| 185 | my $q_like = $dbh->quote($like); | |
| 186 | $glob_sql = " AND fc.file_name LIKE $q_like"; | |
| 187 | } | |
| 188 | my @hot = $db->db_readwrite_multiple($dbh, qq~ | |
| 189 | SELECT fc.file_name, | |
| 190 | COUNT(*) AS ct, | |
| 191 | MAX(fc.file_changes_id) AS latest_id, | |
| 192 | s.server_name | |
| 193 | FROM FILE_CHANGES fc | |
| 194 | LEFT JOIN SERVERS s ON s.server_id = fc.server_id | |
| 195 | WHERE fc.date_time >= DATE_SUB(NOW(), INTERVAL $win MINUTE) | |
| 196 | $glob_sql | |
| 197 | GROUP BY fc.file_name | |
| 198 | HAVING ct >= $thr | |
| 199 | ORDER BY ct DESC | |
| 200 | LIMIT 5 | |
| 201 | ~, __FILE__, __LINE__); | |
| 202 | next unless @hot; | |
| 203 | ||
| 204 | # Fire once with the top offender's info | |
| 205 | my $top = $hot[0]; | |
| 206 | my $fake_row = { | |
| 207 | id => $top->{latest_id}, | |
| 208 | file_name => $top->{file_name}, | |
| 209 | server_name => $top->{server_name} // 'local', | |
| 210 | date_time => POSIX::strftime('%Y-%m-%d %H:%M:%S', localtime), | |
| 211 | ts_epoch => time, | |
| 212 | status => 'frequency_alert', | |
| 213 | }; | |
| 214 | my $payload = _build_payload($rule, $fake_row, 'file'); | |
| 215 | # Enrich summary with count + threshold | |
| 216 | my $extra = "frequency alert: $top->{file_name} changed $top->{ct} times in the last $win min (threshold: $thr)"; | |
| 217 | if (ref($payload->{generic_webhook}) eq 'HASH') { | |
| 218 | $payload->{generic_webhook}{frequency_summary} = $extra; | |
| 219 | $payload->{generic_webhook}{frequency_count} = $top->{ct}; | |
| 220 | $payload->{generic_webhook}{frequency_window} = $win; | |
| 221 | $payload->{generic_webhook}{summary} = $extra; | |
| 222 | } | |
| 223 | my ($status, $http_code, $body, $err) = _deliver($rule, $payload); | |
| 224 | _log_delivery($rule, $fake_row, 'file', $status, $http_code, $body, $err); | |
| 225 | if ($status eq 'sent') { | |
| 226 | $delivered++; | |
| 227 | $db->db_readwrite($dbh, qq~ | |
| 228 | UPDATE ALERT_RULES | |
| 229 | SET fire_count = fire_count + 1, | |
| 230 | last_fired_at = NOW(), | |
| 231 | last_frequency_fire_at = NOW() | |
| 232 | WHERE alert_rule_id = $rule->{alert_rule_id} | |
| 233 | ~, __FILE__, __LINE__); | |
| 234 | } else { | |
| 235 | $failed++; | |
| 236 | } | |
| 237 | } | |
| 238 | ||
| 239 | # ---- Advance cursor atomically -------------------------------------- | |
| 240 | if ($file_max > $cursor{file}) { | |
| 241 | $db->db_readwrite($dbh, | |
| 242 | "UPDATE ALERT_CURSOR SET last_seen_id = $file_max WHERE source_kind = 'file'", | |
| 243 | __FILE__, __LINE__); | |
| 244 | } | |
| 245 | if ($schema_max > $cursor{schema}) { | |
| 246 | $db->db_readwrite($dbh, | |
| 247 | "UPDATE ALERT_CURSOR SET last_seen_id = $schema_max WHERE source_kind = 'schema'", | |
| 248 | __FILE__, __LINE__); | |
| 249 | } | |
| 250 | ||
| 251 | # ---- Pin-drift alerts (Feature 5 wave 4) --------------------------- | |
| 252 | # For each Named Release with alert_delivery_url set, check whether any | |
| 253 | # pin's current-latest SHA differs from the pinned SHA. Log the drift | |
| 254 | # to PIN_DRIFT_LOG (with notified=0) and, if not yet notified, POST an | |
| 255 | # alert to the release's delivery URL, then set notified=1. | |
| 256 | foreach my $rel ($db->db_readwrite_multiple($dbh, q~ | |
| 257 | SELECT named_release_id, name, alert_delivery_kind, alert_delivery_url | |
| 258 | FROM NAMED_RELEASES | |
| 259 | WHERE alert_delivery_kind IS NOT NULL | |
| 260 | AND alert_delivery_kind != 'none' | |
| 261 | AND alert_delivery_url IS NOT NULL | |
| 262 | AND alert_delivery_url != '' | |
| 263 | ~, __FILE__, __LINE__)) { | |
| 264 | ||
| 265 | my $rid = $rel->{named_release_id}; | |
| 266 | my @drifts; | |
| 267 | foreach my $pin ($db->db_readwrite_multiple($dbh, qq~ | |
| 268 | SELECT p.pin_id, p.file_name, p.blob_sha AS pinned_sha, | |
| 269 | fc.server_id | |
| 270 | FROM NAMED_RELEASE_PINS p | |
| 271 | LEFT JOIN FILE_CHANGES fc ON fc.file_changes_id = p.file_changes_id | |
| 272 | WHERE p.named_release_id = $rid | |
| 273 | ~, __FILE__, __LINE__)) { | |
| 274 | my $q_file = $dbh->quote($pin->{file_name}); | |
| 275 | my $sid = int($pin->{server_id} || 0); | |
| 276 | my $cur = $db->db_readwrite($dbh, qq~ | |
| 277 | SELECT blob_sha, file_changes_id FROM FILE_CHANGES | |
| 278 | WHERE server_id = $sid AND file_name = $q_file | |
| 279 | ORDER BY file_changes_id DESC LIMIT 1 | |
| 280 | ~, __FILE__, __LINE__); | |
| 281 | my $cur_sha = $cur ? $cur->{blob_sha} : ''; | |
| 282 | next unless $cur_sha && $cur_sha ne $pin->{pinned_sha}; | |
| 283 | ||
| 284 | # Have we already logged + notified this (pin, current_sha) tuple? | |
| 285 | my $q_pinned = $dbh->quote($pin->{pinned_sha}); | |
| 286 | my $q_current = $dbh->quote($cur_sha); | |
| 287 | my $seen = $db->db_readwrite($dbh, qq~ | |
| 288 | SELECT drift_id, notified FROM PIN_DRIFT_LOG | |
| 289 | WHERE pin_id = $pin->{pin_id} | |
| 290 | AND current_sha = $q_current | |
| 291 | ORDER BY drift_id DESC LIMIT 1 | |
| 292 | ~, __FILE__, __LINE__); | |
| 293 | if ($seen && $seen->{notified}) { | |
| 294 | next; # already notified for this exact drift state | |
| 295 | } | |
| 296 | ||
| 297 | # Log the drift | |
| 298 | my $q_fname = $dbh->quote($pin->{file_name}); | |
| 299 | my $cid = int($cur->{file_changes_id} || 0); | |
| 300 | $db->db_readwrite($dbh, qq~ | |
| 301 | INSERT INTO PIN_DRIFT_LOG | |
| 302 | (named_release_id, pin_id, file_name, pinned_sha, current_sha, file_changes_id, notified) | |
| 303 | VALUES | |
| 304 | ($rid, $pin->{pin_id}, $q_fname, $q_pinned, $q_current, $cid, 0) | |
| 305 | ~, __FILE__, __LINE__); | |
| 306 | push @drifts, { | |
| 307 | file_name => $pin->{file_name}, | |
| 308 | pinned_sha => $pin->{pinned_sha}, | |
| 309 | current_sha => $cur_sha, | |
| 310 | change_id => $cid, | |
| 311 | }; | |
| 312 | } | |
| 313 | ||
| 314 | next unless @drifts; | |
| 315 | ||
| 316 | # Build a summary payload | |
| 317 | my $summary = sprintf('Named release "%s" -- %d pinned file(s) have drifted', $rel->{name}, scalar @drifts); | |
| 318 | my $public_url = $cfg->settings('public_url') || 'https://driftsense.3dshawn.com'; | |
| 319 | my $files_list = join("\n", map { | |
| 320 | " * $_->{file_name} (pinned: " . substr($_->{pinned_sha}, 0, 12) . | |
| 321 | " -> current: " . substr($_->{current_sha}, 0, 12) . ")" | |
| 322 | } @drifts); | |
| 323 | my $link = "$public_url/restore_release.cgi?release_id=$rid"; | |
| 324 | ||
| 325 | my $slack_body = { | |
| 326 | text => "*DriftSense pin drift*: $summary", | |
| 327 | attachments => [{ | |
| 328 | color => '#f43f5e', | |
| 329 | fields => [ | |
| 330 | { title => 'Release', value => $rel->{name}, short => 1 }, | |
| 331 | { title => 'Drifted files', value => scalar(@drifts), short => 1 }, | |
| 332 | { title => 'Details', value => $files_list, short => 0 }, | |
| 333 | ], | |
| 334 | actions => [{ type => 'button', text => 'Restore all pinned', url => $link }], | |
| 335 | }], | |
| 336 | }; | |
| 337 | my $discord_body = { | |
| 338 | content => "**DriftSense pin drift**: $summary", | |
| 339 | embeds => [{ | |
| 340 | title => $rel->{name}, | |
| 341 | url => $link, | |
| 342 | color => 16005694, | |
| 343 | description => $files_list, | |
| 344 | }], | |
| 345 | }; | |
| 346 | my $generic_body = { | |
| 347 | source => 'drift_sense', | |
| 348 | alert_type => 'pin_drift', | |
| 349 | release_id => $rid, | |
| 350 | release_name => $rel->{name}, | |
| 351 | drifted_count=> scalar @drifts, | |
| 352 | summary => $summary, | |
| 353 | files => \@drifts, | |
| 354 | restore_url => $link, | |
| 355 | }; | |
| 356 | ||
| 357 | my $body_ref = $rel->{alert_delivery_kind} eq 'slack' ? $slack_body | |
| 358 | : $rel->{alert_delivery_kind} eq 'discord' ? $discord_body | |
| 359 | : $generic_body; | |
| 360 | ||
| 361 | my ($code, $rbody, $err) = MODS::Webhook::post_json( | |
| 362 | $rel->{alert_delivery_url}, $body_ref, | |
| 363 | timeout => 8, | |
| 364 | user_agent => 'DriftSense/1.0 (pin-drift)', | |
| 365 | ); | |
| 366 | ||
| 367 | if (!$err && $code >= 200 && $code < 400) { | |
| 368 | # Mark all just-logged rows as notified | |
| 369 | foreach my $d (@drifts) { | |
| 370 | my $q_current = $dbh->quote($d->{current_sha}); | |
| 371 | $db->db_readwrite($dbh, qq~ | |
| 372 | UPDATE PIN_DRIFT_LOG SET notified = 1 | |
| 373 | WHERE named_release_id = $rid | |
| 374 | AND file_name = @{[ $dbh->quote($d->{file_name}) ]} | |
| 375 | AND current_sha = $q_current | |
| 376 | AND notified = 0 | |
| 377 | ~, __FILE__, __LINE__); | |
| 378 | } | |
| 379 | $delivered++; | |
| 380 | print "[$ts] pin-drift alert delivered for release '$rel->{name}' (" . scalar(@drifts) . " drifted)\n"; | |
| 381 | } else { | |
| 382 | $failed++; | |
| 383 | print "[$ts] pin-drift alert FAILED for release '$rel->{name}': " . ($err // "HTTP $code") . "\n"; | |
| 384 | } | |
| 385 | ||
| 386 | # Bookkeep the check timestamp | |
| 387 | $db->db_readwrite($dbh, qq~ | |
| 388 | UPDATE NAMED_RELEASES SET last_pin_drift_check_at = NOW() | |
| 389 | WHERE named_release_id = $rid | |
| 390 | ~, __FILE__, __LINE__); | |
| 391 | } | |
| 392 | ||
| 393 | $db->db_disconnect($dbh); | |
| 394 | print "[$ts] processed +$total_new new changes: $delivered delivered, $skipped skipped, $failed failed\n"; | |
| 395 | exit 0; | |
| 396 | ||
| 397 | #--------------------------------------------------------------------- | |
| 398 | sub _process_row { | |
| 399 | my ($row, $kind) = @_; | |
| 400 | foreach my $rule (@rules) { | |
| 401 | next unless _rule_matches($rule, $row, $kind); | |
| 402 | ||
| 403 | # Rate limit check | |
| 404 | if ($rule->{rate_limit_per_min} > 0) { | |
| 405 | my $rlq = $db->db_readwrite($dbh, qq~ | |
| 406 | SELECT COUNT(*) AS n FROM ALERT_DELIVERIES | |
| 407 | WHERE alert_rule_id = $rule->{alert_rule_id} | |
| 408 | AND fired_at >= DATE_SUB(NOW(), INTERVAL 1 MINUTE) | |
| 409 | AND delivery_status IN ('sent','pending') | |
| 410 | ~, __FILE__, __LINE__); | |
| 411 | if ($rlq && $rlq->{n} >= $rule->{rate_limit_per_min}) { | |
| 412 | _log_delivery($rule, $row, $kind, 'skipped', undef, undef, | |
| 413 | "rate limit ($rule->{rate_limit_per_min}/min) reached"); | |
| 414 | $skipped++; | |
| 415 | next; | |
| 416 | } | |
| 417 | } | |
| 418 | ||
| 419 | # Build payload + POST | |
| 420 | my $payload = _build_payload($rule, $row, $kind); | |
| 421 | my ($status, $http_code, $body, $err) = _deliver($rule, $payload); | |
| 422 | _log_delivery($rule, $row, $kind, $status, $http_code, $body, $err); | |
| 423 | ||
| 424 | if ($status eq 'sent') { $delivered++; } | |
| 425 | elsif ($status eq 'skipped') { $skipped++; } | |
| 426 | else { $failed++; } | |
| 427 | ||
| 428 | # Bump the rule's fire_count + last_fired_at | |
| 429 | if ($status eq 'sent') { | |
| 430 | $db->db_readwrite($dbh, qq~ | |
| 431 | UPDATE ALERT_RULES | |
| 432 | SET fire_count = fire_count + 1, | |
| 433 | last_fired_at = NOW() | |
| 434 | WHERE alert_rule_id = $rule->{alert_rule_id} | |
| 435 | ~, __FILE__, __LINE__); | |
| 436 | } | |
| 437 | } | |
| 438 | } | |
| 439 | ||
| 440 | #--------------------------------------------------------------------- | |
| 441 | sub _rule_matches { | |
| 442 | my ($rule, $row, $kind) = @_; | |
| 443 | ||
| 444 | # Kind filter | |
| 445 | return 0 if $rule->{match_kind} ne 'either' && $rule->{match_kind} ne $kind; | |
| 446 | ||
| 447 | # ts-only exclusion (files only) | |
| 448 | return 0 if $kind eq 'file' && $rule->{exclude_ts_only} && $row->{is_ts_only}; | |
| 449 | ||
| 450 | # Path glob | |
| 451 | my $glob = $rule->{match_path_glob} // ''; | |
| 452 | if (length $glob) { | |
| 453 | my $target = $kind eq 'file' | |
| 454 | ? ($row->{file_name} // '') | |
| 455 | : (($row->{database_name} // '') . '.' . ($row->{table_name} // '')); | |
| 456 | my $re = _glob_to_regex($glob); | |
| 457 | return 0 unless $target =~ $re; | |
| 458 | } | |
| 459 | ||
| 460 | # Status filter (files only) | |
| 461 | if ($kind eq 'file' && $rule->{match_status_list}) { | |
| 462 | my %allowed = map { s/^\s+|\s+$//g; $_ => 1 } | |
| 463 | split /,/, $rule->{match_status_list}; | |
| 464 | return 0 unless $allowed{ $row->{status} // '' }; | |
| 465 | } | |
| 466 | ||
| 467 | return 1; | |
| 468 | } | |
| 469 | ||
| 470 | sub _glob_to_regex { | |
| 471 | my $g = shift; | |
| 472 | my $re = quotemeta $g; | |
| 473 | # Handle glob patterns: escaped meta chars back to regex meaning | |
| 474 | $re =~ s{\\\*\\\*}{.*}g; # ** -> .* | |
| 475 | $re =~ s{\\\*}{[^/]*}g; # * -> [^/]* | |
| 476 | $re =~ s{\\\?}{.}g; # ? -> . | |
| 477 | return qr/^$re$/; | |
| 478 | } | |
| 479 | ||
| 480 | #--------------------------------------------------------------------- | |
| 481 | sub _build_payload { | |
| 482 | my ($rule, $row, $kind) = @_; | |
| 483 | ||
| 484 | my $where = $kind eq 'file' | |
| 485 | ? ($row->{file_name} // '(unknown file)') | |
| 486 | : (($row->{database_name} // '?') . '.' . ($row->{table_name} // '?')); | |
| 487 | my $when_utc = $row->{date_time} || $row->{change_datetime} || ''; | |
| 488 | my $server = $row->{server_name} || 'local'; | |
| 489 | my $status_h = $kind eq 'file' ? ($row->{status} // '') : 'DDL change'; | |
| 490 | my $link_id = $row->{id}; | |
| 491 | my $link = "$public_url/diff.cgi?kind=$kind&id=$link_id"; | |
| 492 | ||
| 493 | my $summary = $kind eq 'file' | |
| 494 | ? "$status_h: $where on $server" | |
| 495 | : "$where changed on $server"; | |
| 496 | ||
| 497 | my $slack = { | |
| 498 | text => "*DriftSense*: $summary", | |
| 499 | attachments => [{ | |
| 500 | color => $kind eq 'schema' ? '#f59e0b' : '#14b8a6', | |
| 501 | fields => [ | |
| 502 | { title => 'What', value => $where, short => 0 }, | |
| 503 | { title => 'When', value => $when_utc, short => 1 }, | |
| 504 | { title => 'Server', value => $server, short => 1 }, | |
| 505 | ], | |
| 506 | actions => [{ | |
| 507 | type => 'button', text => 'View diff', url => $link, | |
| 508 | }], | |
| 509 | }], | |
| 510 | }; | |
| 511 | my $discord = { | |
| 512 | content => "**DriftSense**: $summary", | |
| 513 | embeds => [{ | |
| 514 | title => $where, | |
| 515 | url => $link, | |
| 516 | color => $kind eq 'schema' ? 16101915 : 1349286, # amber, teal | |
| 517 | fields => [ | |
| 518 | { name => 'When', value => $when_utc || '?', inline => \1 }, | |
| 519 | { name => 'Server', value => $server, inline => \1 }, | |
| 520 | ], | |
| 521 | }], | |
| 522 | }; | |
| 523 | my $generic = { | |
| 524 | source => 'drift_sense', | |
| 525 | rule_id => $rule->{alert_rule_id}, | |
| 526 | rule_name => $rule->{rule_name}, | |
| 527 | kind => $kind, | |
| 528 | summary => $summary, | |
| 529 | target => $where, | |
| 530 | server => $server, | |
| 531 | when_utc => $when_utc, | |
| 532 | ts_epoch => $row->{ts_epoch}, | |
| 533 | diff_url => $link, | |
| 534 | status => $status_h, | |
| 535 | }; | |
| 536 | ||
| 537 | return { | |
| 538 | slack => $slack, | |
| 539 | discord => $discord, | |
| 540 | generic_webhook => $generic, | |
| 541 | summary => $summary, | |
| 542 | }; | |
| 543 | } | |
| 544 | ||
| 545 | #--------------------------------------------------------------------- | |
| 546 | sub _deliver { | |
| 547 | my ($rule, $payload) = @_; | |
| 548 | my $kind = $rule->{delivery_kind}; | |
| 549 | my $url = $rule->{delivery_url}; | |
| 550 | ||
| 551 | if ($kind eq 'email') { | |
| 552 | my $to = $rule->{delivery_email} || ''; | |
| 553 | return ('failed', undef, undef, 'no delivery_email configured') unless length $to; | |
| 554 | my $summary = $payload->{summary} // 'DriftSense alert'; | |
| 555 | my $target = ($payload->{generic_webhook} && $payload->{generic_webhook}{target}) // '(unknown)'; | |
| 556 | my $server = ($payload->{generic_webhook} && $payload->{generic_webhook}{server}) // 'local'; | |
| 557 | my $link = ($payload->{generic_webhook} && $payload->{generic_webhook}{diff_url}) // ''; | |
| 558 | ||
| 559 | my $mail_body = "DriftSense captured a change matching alert rule: $rule->{rule_name}\n\n" | |
| 560 | . "Target : $target\n" | |
| 561 | . "Server : $server\n" | |
| 562 | . "Summary: $summary\n" | |
| 563 | . ($link ? "Diff : $link\n" : '') | |
| 564 | . "\n(Automated notification from DriftSense.)\n"; | |
| 565 | ||
| 566 | my $from = $cfg->settings('alert_from_email') || 'drift_sense@localhost'; | |
| 567 | my $subj = "[DriftSense] " . substr($summary, 0, 120); | |
| 568 | ||
| 569 | my $sendmail = '/usr/sbin/sendmail'; | |
| 570 | $sendmail = '/usr/lib/sendmail' unless -x $sendmail; | |
| 571 | return ('failed', undef, undef, "no sendmail binary") unless -x $sendmail; | |
| 572 | ||
| 573 | require IPC::Open3; | |
| 574 | my ($wtr, $rdr, $err_fh); | |
| 575 | $err_fh = Symbol::gensym(); | |
| 576 | my $pid = eval { IPC::Open3::open3($wtr, $rdr, $err_fh, $sendmail, '-t', '-i') }; | |
| 577 | return ('failed', undef, undef, "spawn: $@") if $@ || !$pid; | |
| 578 | binmode $wtr; | |
| 579 | print {$wtr} "From: $from\r\n"; | |
| 580 | print {$wtr} "To: $to\r\n"; | |
| 581 | print {$wtr} "Subject: $subj\r\n"; | |
| 582 | print {$wtr} "Content-Type: text/plain; charset=utf-8\r\n"; | |
| 583 | print {$wtr} "\r\n"; | |
| 584 | print {$wtr} $mail_body; | |
| 585 | close $wtr; | |
| 586 | do { local $/; <$rdr> // ''; }; | |
| 587 | do { local $/; <$err_fh> // ''; }; | |
| 588 | close $rdr; close $err_fh; | |
| 589 | waitpid $pid, 0; | |
| 590 | my $rc = $? >> 8; | |
| 591 | return $rc == 0 | |
| 592 | ? ('sent', 0, "delivered to $to", undef) | |
| 593 | : ('failed', $rc, undef, "sendmail exit $rc"); | |
| 594 | } | |
| 595 | unless (length($url // '')) { | |
| 596 | return ('failed', undef, undef, "no delivery URL configured"); | |
| 597 | } | |
| 598 | ||
| 599 | my $body_ref = $kind eq 'slack' ? $payload->{slack} | |
| 600 | : $kind eq 'discord' ? $payload->{discord} | |
| 601 | : $payload->{generic_webhook}; | |
| 602 | ||
| 603 | # MODS::Webhook handles JSON encoding + curl transport; returns | |
| 604 | # (http_code, response_body, error_message). | |
| 605 | my ($code, $rbody, $err) = MODS::Webhook::post_json( | |
| 606 | $url, $body_ref, | |
| 607 | timeout => 8, | |
| 608 | user_agent => 'DriftSense/1.0 (alert-dispatch)', | |
| 609 | ); | |
| 610 | $rbody //= ''; | |
| 611 | $rbody = substr($rbody, 0, 500) if length($rbody) > 500; | |
| 612 | ||
| 613 | if (!$err && $code >= 200 && $code < 400) { | |
| 614 | return ('sent', $code, $rbody, undef); | |
| 615 | } | |
| 616 | return ('failed', $code, $rbody, $err || "HTTP $code"); | |
| 617 | } | |
| 618 | ||
| 619 | #--------------------------------------------------------------------- | |
| 620 | sub _log_delivery { | |
| 621 | my ($rule, $row, $kind, $status, $http_code, $body, $err) = @_; | |
| 622 | my $summary = substr(($rule->{rule_name} // '') . ' :: ' . ($row->{file_name} | |
| 623 | || ($row->{database_name} . '.' . $row->{table_name}) | |
| 624 | || '?'), 0, 490); | |
| 625 | my $q_summary = $dbh->quote($summary); | |
| 626 | my $q_status = $dbh->quote($status); | |
| 627 | my $q_body = $dbh->quote($body // ''); | |
| 628 | my $q_err = $dbh->quote($err // ''); | |
| 629 | my $q_code = defined($http_code) ? int($http_code) : 'NULL'; | |
| 630 | ||
| 631 | $db->db_readwrite($dbh, qq~ | |
| 632 | INSERT INTO ALERT_DELIVERIES | |
| 633 | (alert_rule_id, source_kind, source_id, match_summary, | |
| 634 | delivery_status, http_status, http_response, error_message) | |
| 635 | VALUES | |
| 636 | ($rule->{alert_rule_id}, '$kind', $row->{id}, $q_summary, | |
| 637 | $q_status, $q_code, $q_body, $q_err) | |
| 638 | ~, __FILE__, __LINE__); | |
| 639 | } |