ntfy-bridge/dedup.v

module main

import time
import sources

pub struct DedupEntry {
pub mut:
	last_seen time.Time
	count     int
}

pub struct Dedup {
pub mut:
	enabled        bool
	ttl_seconds    int
	max_per_minute int
	max_per_source int
	cache          map[string]DedupEntry
	topic_history  map[string][]time.Time
	source_history map[string][]time.Time
}

pub fn new_dedup(cfg DedupConfig) Dedup {
	return Dedup{
		enabled:        cfg.enabled
		ttl_seconds:    cfg.ttl_seconds
		max_per_minute: cfg.rate_limit.max_per_minute
		max_per_source: cfg.rate_limit.max_per_source
		cache:          map[string]DedupEntry{}
		topic_history:  map[string][]time.Time{}
		source_history: map[string][]time.Time{}
	}
}

fn simple_hash(s string) u64 {
	mut h := u64(14695981039346656037)
	for c in s.bytes() {
		h ^= u64(c)
		h *= u64(1099511628211)
	}
	return h
}

fn dedup_key(event sources.Event) string {
	h := simple_hash(event.message)
	return '${event.source}:${event.name}:${event.topic}:${h}'
}

fn (mut d Dedup) cleanup(now time.Time) {
	now_unix := now.unix()
	for key, entry in d.cache {
		if now_unix - entry.last_seen.unix() > d.ttl_seconds {
			d.cache.delete(key)
		}
	}
	for topic, times in d.topic_history {
		mut kept := []time.Time{}
		for t in times {
			if now_unix - t.unix() <= 60 {
				kept << t
			}
		}
		if kept.len == 0 {
			d.topic_history.delete(topic)
		} else {
			d.topic_history[topic] = kept
		}
	}
	for source, times in d.source_history {
		mut kept := []time.Time{}
		for t in times {
			if now_unix - t.unix() <= 60 {
				kept << t
			}
		}
		if kept.len == 0 {
			d.source_history.delete(source)
		} else {
			d.source_history[source] = kept
		}
	}
}

pub fn (mut d Dedup) allow(event sources.Event) bool {
	if !d.enabled {
		return true
	}
	now := time.now()
	d.cleanup(now)

	// Rate limit per topic
	topic_times := d.topic_history[event.topic] or { []time.Time{} }
	if topic_times.len >= d.max_per_minute && d.max_per_minute > 0 {
		return false
	}

	// Rate limit per source
	source_key := '${event.source}:${event.name}'
	source_times := d.source_history[source_key] or { []time.Time{} }
	if source_times.len >= d.max_per_source && d.max_per_source > 0 {
		return false
	}

	// Dedup check
	key := dedup_key(event)
	entry := d.cache[key] or { DedupEntry{} }
	if entry.count > 0 {
		d.cache[key] = DedupEntry{
			last_seen: now
			count:     entry.count + 1
		}
		return false
	}

	d.cache[key] = DedupEntry{
		last_seen: now
		count:     1
	}

	mut new_topic_times := topic_times.clone()
	new_topic_times << now
	d.topic_history[event.topic] = new_topic_times

	mut new_source_times := source_times.clone()
	new_source_times << now
	d.source_history[source_key] = new_source_times

	return true
}