bruno/ntfy-bridge
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

feat: phase 3 + phase 4 complete — v0.4.0
CI / test (push) Failing after 24s
Details

Browse Source 
Phase 3:
- Config reload via SIGHUP (handle_signals loop: .hup reloads config, .int/.term shutdown)
- Hot reload: cfg, webhook_routes, ntfy client, dedup updated without restart

Phase 4:
- Silence rules: POST /api/silence?duration=30m, GET /api/silence, DELETE /api/silence
- Grouping: buffer 10s, events grouped with (×N in 10s) suffix
- CI/CD: .gitea/workflows/ci.yml (build + test on push/PR)

Docs:
- ROADMAP.md: phases 3 & 4 marked DONE, version table updated
- ARCHITECTURE.md: synced with actual code

Version: 0.1.0 → 0.4.0 (main.v, v.mod)
This commit is contained in:
bruno
2026-08-03 14:35:59 -04:00
parent 4e1d7cd94a
commit 72b797704f
6 changed files with 503 additions and 200 deletions
Download Patch File Download Diff File Expand all files Collapse all files
server.v
+233 -41
View File
@@ -8,6 +8,7 @@ import crypto.sha256
import encoding.hex
import sources
import os
import strconv
pub struct Stats {
@@ -19,17 +20,21 @@ pub mut:
pub struct App {
pub mut:
cfg Config
logger Logger
ntfy NtfyClient
dedup Dedup
stats Stats
start_time time.Time
poll_state map[string]bool // url -> is_down
webhook_routes map[string]WebhookRoute
recent_events []RecentEvent
shutdown_flag bool
dashboard_html string // cached dashboard HTML
cfg Config
config_path string // path to config file for SIGHUP reload
logger Logger
ntfy NtfyClient
dedup Dedup
stats Stats
start_time time.Time
poll_state map[string]bool // url -> is_down
webhook_routes map[string]WebhookRoute
recent_events []RecentEvent
shutdown_flag bool
dashboard_html string // cached dashboard HTML
silence_until time.Time // silence all notifications until this time
group_buffer map[string][]GroupEntry // key: source:name:topic
group_last_flush time.Time
}
pub struct RecentEvent {
@@ -42,27 +47,38 @@ pub struct RecentEvent {
success bool
}
pub struct GroupEntry {
event sources.Event
mut:
received time.Time
count int
}
const recent_events_max = 100
pub fn new_app(cfg Config, quiet bool) App {
pub fn new_app(cfg Config, config_path string, quiet bool) App {
// json_mode = quiet — when running as a systemd service (quiet mode),
// use JSON structured logging; otherwise use human-readable console output
logger := new_logger(.info, quiet)
dashboard := load_dashboard_html(logger)
return App{
cfg: cfg
logger: logger
ntfy: new_ntfy_client(cfg.server, logger)
dedup: new_dedup(cfg.dedup)
stats: Stats{
cfg: cfg
config_path: config_path
logger: logger
ntfy: new_ntfy_client(cfg.server, logger)
dedup: new_dedup(cfg.dedup)
stats: Stats{
by_source: map[string]int{}
}
start_time: time.now()
poll_state: map[string]bool{}
webhook_routes: build_webhook_routes(cfg)
recent_events: []RecentEvent{}
shutdown_flag: false
dashboard_html: dashboard
start_time: time.now()
poll_state: map[string]bool{}
webhook_routes: build_webhook_routes(cfg)
recent_events: []RecentEvent{}
shutdown_flag: false
dashboard_html: dashboard
silence_until: time.Time{}
group_buffer: map[string][]GroupEntry{}
group_last_flush: time.now()
}
}
@@ -104,12 +120,30 @@ pub fn (mut app App) run() ! {
$if linux || macos {
fn (mut app App) handle_signals() {
sigch := os.new_signal_channel(.int, .term)
_ := <-sigch
app.logger.info('server', 'shutdown', 'received signal, shutting down...')
app.shutdown_flag = true
time.sleep(500 * time.millisecond)
exit(0)
sigch := os.new_signal_channel(.int, .term, .hup)
for {
sig := <-sigch
match sig {
.hup {
app.logger.info('server', 'reload', 'SIGHUP received, reloading config...')
new_cfg := load_config(app.config_path) or {
app.logger.error('server', 'reload', 'failed to reload config: ${err}')
continue
}
app.cfg = new_cfg
app.webhook_routes = build_webhook_routes(new_cfg)
app.ntfy = new_ntfy_client(new_cfg.server, app.logger)
app.dedup = new_dedup(new_cfg.dedup)
app.logger.info('server', 'reload', 'config reloaded successfully')
}
.int, .term {
app.logger.info('server', 'shutdown', 'received signal, shutting down...')
app.shutdown_flag = true
time.sleep(500 * time.millisecond)
exit(0)
}
}
}
}
}
@@ -120,6 +154,8 @@ fn (mut app App) start_background_tasks() {
if app.cfg.sources.docker.len > 0 {
spawn app.docker_watch_loop()
}
// Grouping flush goroutine
spawn app.group_flush_loop()
}
// ---- HTTP Handler ----
@@ -145,10 +181,21 @@ pub fn (mut app App) handle(req http.Request) http.Response {
if req.url == '/api/history' {
return app.history_api_response()
}
if req.url == '/api/silence' {
return app.silence_status_response()
}
}
.post {
if req.url.starts_with('/api/silence') {
return app.silence_create_response(req.url)
}
return app.handle_webhook(req)
}
.delete {
if req.url == '/api/silence' {
return app.silence_clear_response()
}
}
else {}
}
@@ -286,24 +333,44 @@ fn (mut app App) verify_hmac(payload string, sig_header string) bool {
pub fn (mut app App) publish(event sources.Event) {
mut e := apply_defaults(event, app.cfg.defaults)
// Check silence
now := time.now()
if now < app.silence_until {
app.logger.debug('server', 'publish', 'silenced: ${e.source}:${e.name}')
return
}
if !app.dedup.allow(e) {
app.logger.debug('server', 'dedup', 'dropped event ${e.source}:${e.name}')
return
}
now := time.now()
app.ntfy.publish(e) or {
app.logger.error('server', 'publish', 'failed to publish: ${err}')
app.stats.errors++
app.add_recent_event(now, e, false)
return
// Grouping: buffer the event
key := '${e.source}:${e.name}:${e.topic}'
mut entries := app.group_buffer[key] or { []GroupEntry{} }
// Check if there's a matching event in the buffer (same source/name/topic)
mut found_idx := -1
for i, entry in entries {
if entry.event.message == e.message {
found_idx = i
break
}
}
app.stats.total_notifications++
mut count := app.stats.by_source[e.source] or { 0 }
count++
app.stats.by_source[e.source] = count
app.logger.info('server', 'publish', '${e.source}:${e.name} → ${e.topic}')
app.add_recent_event(now, e, true)
if found_idx >= 0 {
// Rebuild to mutate count/received
mut updated := entries[found_idx]
updated.count++
updated.received = now
entries[found_idx] = updated
} else {
entries << GroupEntry{
event: e
received: now
count: 1
}
}
app.group_buffer[key] = entries
}
fn (mut app App) add_recent_event(ts time.Time, event sources.Event, success bool) {
@@ -388,3 +455,128 @@ fn (mut app App) dashboard_response() http.Response {
header: header
})
}
// ---- Silence Rules ----
fn (mut app App) silence_status_response() http.Response {
is_silenced := time.now() < app.silence_until
remaining := if is_silenced { app.silence_until.unix() - time.now().unix() } else { 0 }
body := '{"silenced":${is_silenced},"remaining_seconds":${remaining}}'
return new_json_response(.ok, body)
}
fn (mut app App) silence_create_response(url string) http.Response {
// Parse duration from query string: /api/silence?duration=30m
duration_str := parse_query_param(url, 'duration')
if duration_str == '' {
return new_json_response(.bad_request, '{"error":"missing duration parameter (e.g. ?duration=30m)"}')
}
seconds := parse_duration(duration_str)
if seconds <= 0 {
return new_json_response(.bad_request, '{"error":"invalid duration: ${duration_str}"}')
}
app.silence_until = time.now().add_seconds(seconds)
app.logger.info('server', 'silence', 'notifications silenced for ${duration_str} (${seconds}s)')
body := '{"silenced":true,"duration":"${duration_str}","seconds":${seconds}}'
return new_json_response(.ok, body)
}
fn (mut app App) silence_clear_response() http.Response {
app.silence_until = time.Time{}
app.logger.info('server', 'silence', 'silence cleared')
return new_json_response(.ok, '{"silenced":false}')
}
// parse_query_param extracts a query parameter value from a URL string.
fn parse_query_param(url string, key string) string {
prefix := '/api/silence?'
if !url.starts_with(prefix) {
return ''
}
qs := url[prefix.len..]
parts := qs.split('&')
for part in parts {
kv := part.split('=')
if kv.len == 2 && kv[0] == key {
return kv[1]
}
}
return ''
}
// parse_duration converts a human duration string (e.g. "30m", "2h", "1h30m") to seconds.
fn parse_duration(s string) int {
mut total := 0
mut current := ''
for c in s {
if c >= `0` && c <= `9` {
current += c.ascii_str()
} else {
val := strconv.atoi(current) or { 0 }
match c {
`s` { total += val }
`m` { total += val * 60 }
`h` { total += val * 3600 }
`d` { total += val * 86400 }
else {}
}
current = ''
}
}
return total
}
// ---- Grouping ----
const group_flush_interval = 10 // seconds
fn (mut app App) group_flush_loop() {
for {
if app.shutdown_flag {
break
}
time.sleep(group_flush_interval * time.second)
app.flush_group_buffer()
}
}
fn (mut app App) flush_group_buffer() {
now := time.now()
for _, entries in app.group_buffer {
if entries.len == 0 {
continue
}
mut total_count := 0
mut latest_entry := entries[0]
for i, entry in entries {
total_count += entry.count
if entry.received > latest_entry.received {
latest_entry = entries[i]
}
}
mut msg := latest_entry.event.message
if total_count > 1 {
msg = '${msg} (×${total_count} in ${group_flush_interval}s)'
}
mut e := latest_entry.event.clone()
e.message = msg
app.ntfy.publish(e) or {
app.logger.error('server', 'publish', 'failed to publish: ${err}')
app.stats.errors++
app.add_recent_event(now, e, false)
continue
}
app.stats.total_notifications++
mut count := app.stats.by_source[e.source] or { 0 }
count++
app.stats.by_source[e.source] = count
app.logger.info('server', 'publish',
'${e.source}:${e.name} → ${e.topic} (grouped ×${total_count})')
app.add_recent_event(now, e, true)
}
app.group_buffer = map[string][]GroupEntry{}
app.group_last_flush = time.now()
}