operserv: add NOTIFY oper watch list emitting matched-user events to the staff feed

src/engine/db/event.rs

@@ -141,6 +141,10 @@ pub enum Event {
     // "NICK", "CHAN", or "EMAIL".
     ForbidAdded { kind: String, mask: String, setter: String, reason: String, ts: u64 },
     ForbidRemoved { kind: String, mask: String },
+    // Oper watch list (OperServ NOTIFY). Global oper policy; `expires` None =
+    // permanent. `flags` selects which events (cdjpntsS) are announced.
+    NotifyAdded { mask: String, flags: String, reason: String, setter: String, ts: u64, expires: Option<u64> },
+    NotifyRemoved { mask: String },
     // Server jupes (OperServ JUPE). Node-local (each network holds a rogue server
     // by minting a fake one on `sid`), so Local scope but still persisted.
     JupeAdded { name: String, sid: String, reason: String },
@@ -207,6 +211,8 @@ impl Event {
             | Event::FilterRemoved { .. }
             | Event::ForbidAdded { .. }
             | Event::ForbidRemoved { .. }
+            | Event::NotifyAdded { .. }
+            | Event::NotifyRemoved { .. }
             | Event::NewsAdded { .. }
             | Event::NewsDeleted { .. }
             | Event::ReportFiled { .. }
@@ -660,6 +666,13 @@ pub(crate) fn apply(accounts: &mut HashMap<String, Account>, channels: &mut Hash
         Event::ForbidRemoved { kind, mask } => {
             net.forbids.retain(|f| !(f.kind == kind && f.mask.eq_ignore_ascii_case(&mask)));
         }
+        Event::NotifyAdded { mask, flags, reason, setter, ts, expires } => {
+            net.notifies.retain(|n| !n.mask.eq_ignore_ascii_case(&mask));
+            net.notifies.push(Notify { mask, flags, reason, setter, ts, expires });
+        }
+        Event::NotifyRemoved { mask } => {
+            net.notifies.retain(|n| !n.mask.eq_ignore_ascii_case(&mask));
+        }
         Event::JupeAdded { name, sid, reason } => {
             // Idempotent over a snapshot; keep jupe_seq ahead so a fresh add can't
             // reuse a sid.

src/engine/db/mod.rs

@@ -40,7 +40,7 @@ pub(crate) use event::{apply, Scope};
 // echo-api SDK crate; re-exported so the engine keeps naming them locally and
 // modules importing `crate::engine::db::{ChanError, ...}` are unaffected.
 pub use echo_api::{
-    AccountView, AjoinView, AkillView, BotView, Caps, ForbidKind, ForbidView, GroupView, HelpView, IgnoreView, MemoView, NewsKind, NewsView, Privs, ReportView, SuspensionView, ChanAccessView, ChanAkickView, ChanError, ChanSetting, ChannelView, CertError, CodeKind, Kicker, RegError, Store, TriggerView, VhostView, XlineKind,
+    AccountView, AjoinView, AkillView, BotView, Caps, ForbidKind, ForbidView, GroupView, HelpView, IgnoreView, MemoView, NewsKind, NewsView, NotifyView, Privs, ReportView, SuspensionView, ChanAccessView, ChanAkickView, ChanError, ChanSetting, ChannelView, CertError, CodeKind, Kicker, RegError, Store, TriggerView, VhostView, XlineKind,
 };
 
 #[derive(Debug, Clone, Serialize, Deserialize)]
@@ -226,6 +226,20 @@ fn gline_kind() -> String {
     "G".to_string()
 }
 
+// An OperServ NOTIFY watch entry: a mask (nick!user@host#gecos or #channel) whose
+// matching users have flagged events announced to the staff feed. Network-wide
+// oper policy like a forbid, so it gossips and persists; `expires` None = forever.
+#[derive(Debug, Clone, Serialize, Deserialize)]
+pub struct Notify {
+    pub mask: String,
+    pub flags: String,
+    pub reason: String,
+    pub setter: String,
+    pub ts: u64,
+    #[serde(default)]
+    pub expires: Option<u64>,
+}
+
 // A valid 3-char server id for a juped server from a counter: a leading digit
 // (kept high to avoid colliding with typical real SIDs) then two base-36 chars.
 fn jupe_sid(seq: u32) -> String {
@@ -309,6 +323,7 @@ pub struct NetData {
     pub akills: Vec<Akill>,
     pub filters: Vec<Filter>,
     pub forbids: Vec<Forbid>,
+    pub notifies: Vec<Notify>,
     pub news: Vec<News>,
     pub news_seq: u64,
     // Abuse reports (ReportServ), oldest first.
@@ -1257,6 +1272,9 @@ impl Db {
         for f in &self.net.forbids {
             snapshot.push(Event::ForbidAdded { kind: f.kind.clone(), mask: f.mask.clone(), setter: f.setter.clone(), reason: f.reason.clone(), ts: f.ts });
         }
+        for n in self.net.notifies.iter().filter(|n| n.expires.is_none_or(|e| e > now)) {
+            snapshot.push(Event::NotifyAdded { mask: n.mask.clone(), flags: n.flags.clone(), reason: n.reason.clone(), setter: n.setter.clone(), ts: n.ts, expires: n.expires });
+        }
         for a in self.net.akills.iter().filter(|a| a.expires.is_none_or(|e| e > now)) {
             snapshot.push(Event::AkillAdded { kind: a.kind.clone(), mask: a.mask.clone(), setter: a.setter.clone(), reason: a.reason.clone(), ts: a.ts, expires: a.expires });
         }

src/engine/db/network.rs

@@ -130,6 +130,83 @@ impl Db {
             .map(|f| f.reason.clone())
     }
 
+    /// Add (or replace) an OperServ NOTIFY watch. Returns whether it was new.
+    pub fn notify_add(&mut self, mask: &str, flags: &str, reason: &str, setter: &str, expires: Option<u64>) -> Result<bool, RegError> {
+        let ts = now();
+        let same = |n: &Notify| n.mask.eq_ignore_ascii_case(mask);
+        let fresh = !self.net.notifies.iter().any(same);
+        self.log
+            .append(Event::NotifyAdded { mask: mask.to_string(), flags: flags.to_string(), reason: reason.to_string(), setter: setter.to_string(), ts, expires })
+            .map_err(|_| RegError::Internal)?;
+        self.net.notifies.retain(|n| !same(n));
+        self.net.notifies.push(Notify { mask: mask.to_string(), flags: flags.to_string(), reason: reason.to_string(), setter: setter.to_string(), ts, expires });
+        Ok(fresh)
+    }
+
+    /// Remove a NOTIFY watch by exact mask. Returns whether one existed.
+    pub fn notify_del(&mut self, mask: &str) -> Result<bool, RegError> {
+        let same = |n: &Notify| n.mask.eq_ignore_ascii_case(mask);
+        if !self.net.notifies.iter().any(same) {
+            return Ok(false);
+        }
+        self.log.append(Event::NotifyRemoved { mask: mask.to_string() }).map_err(|_| RegError::Internal)?;
+        self.net.notifies.retain(|n| !same(n));
+        Ok(true)
+    }
+
+    /// Drop every NOTIFY watch. Returns how many were removed. One event per entry
+    /// so a peer replaying the log folds to the same empty state.
+    pub fn notify_clear(&mut self) -> Result<usize, RegError> {
+        let masks: Vec<String> = self.net.notifies.iter().map(|n| n.mask.clone()).collect();
+        let count = masks.len();
+        for mask in masks {
+            self.log.append(Event::NotifyRemoved { mask }).map_err(|_| RegError::Internal)?;
+        }
+        self.net.notifies.clear();
+        Ok(count)
+    }
+
+    /// All live (unexpired) NOTIFY watches, oldest first.
+    pub fn notifies(&self) -> Vec<NotifyView> {
+        let now = now();
+        self.net.notifies
+            .iter()
+            .filter(|n| n.expires.is_none_or(|e| e > now))
+            .map(|n| NotifyView { mask: n.mask.clone(), flags: n.flags.clone(), reason: n.reason.clone(), setter: n.setter.clone(), ts: n.ts, expires: n.expires })
+            .collect()
+    }
+
+    /// Whether any live watch exists — a cheap gate so the engine can skip the
+    /// per-event identity lookup when the list is empty.
+    pub fn any_notifies(&self) -> bool {
+        let now = now();
+        self.net.notifies.iter().any(|n| n.expires.is_none_or(|e| e > now))
+    }
+
+    /// Union of the flag letters of every live watch matching this user and/or
+    /// channel: a `#`/`&` mask tests the channel name, any other tests the user's
+    /// identity. The engine tests the result for a given event's letter.
+    pub fn notify_flags(&self, target: Option<&echo_api::BanTarget>, chan: Option<&str>) -> String {
+        let now = now();
+        let mut flags = String::new();
+        for n in self.net.notifies.iter().filter(|n| n.expires.is_none_or(|e| e > now)) {
+            let hit = if n.mask.starts_with('#') || n.mask.starts_with('&') {
+                chan.is_some_and(|c| glob_match(&n.mask, c))
+            } else {
+                // Host/extban masks go through the ban matcher; a bare nick glob
+                // (no '@'/'!') also matches the nick, so `baddie*` works too.
+                target.is_some_and(|t| {
+                    echo_api::akick_matches(&n.mask, t)
+                        || (!n.mask.contains(['@', '!']) && glob_match(&n.mask, t.nick))
+                })
+            };
+            if hit {
+                flags.push_str(&n.flags);
+            }
+        }
+        flags
+    }
+
     /// Add (or replace) a session-limit exception for an IP-mask.
     pub fn session_except_add(&mut self, mask: &str, limit: u32, reason: &str) {
         let _ = self.log.append(Event::SessionExceptionAdded { mask: mask.to_string(), limit, reason: reason.to_string() });

src/engine/db/store.rs

@@ -263,6 +263,24 @@ impl Store for Db {
     fn is_forbidden(&self, kind: ForbidKind, name: &str) -> Option<String> {
         Db::is_forbidden(self, kind, name)
     }
+    fn notify_add(&mut self, mask: &str, flags: &str, reason: &str, setter: &str, expires: Option<u64>) -> Result<bool, RegError> {
+        Db::notify_add(self, mask, flags, reason, setter, expires)
+    }
+    fn notify_del(&mut self, mask: &str) -> Result<bool, RegError> {
+        Db::notify_del(self, mask)
+    }
+    fn notify_clear(&mut self) -> Result<usize, RegError> {
+        Db::notify_clear(self)
+    }
+    fn notifies(&self) -> Vec<NotifyView> {
+        Db::notifies(self)
+    }
+    fn any_notifies(&self) -> bool {
+        Db::any_notifies(self)
+    }
+    fn notify_flags(&self, target: Option<&echo_api::BanTarget>, chan: Option<&str>) -> String {
+        Db::notify_flags(self, target, chan)
+    }
     fn ignore_add(&mut self, mask: &str, reason: &str, expires: Option<u64>) {
         Db::ignore_add(self, mask, reason, expires)
     }

src/engine/dispatch.rs

@@ -82,6 +82,14 @@ impl Engine {
             }
             if let Some(nick) = matched {
                 self.bump(&format!("{nick}.command"));
+                // NOTIFY watch on services use: log the service + verb only (never
+                // the arguments, which may carry a password). SET commands carry the
+                // 'S' flag, everything else 's'.
+                let verb = text.split_whitespace().next().unwrap_or("").to_ascii_uppercase();
+                let flag = if verb == "SET" { 'S' } else { 's' };
+                if let Some(line) = self.notify_line(flag, from, None, &format!("used {nick} {verb}")) {
+                    ctx.actions.push(line);
+                }
             }
         }
         // Fold any counters the command recorded into the shared registry.

src/engine/mod.rs

@@ -377,6 +377,23 @@ impl Engine {
         }
     }
 
+    // OperServ NOTIFY: if `uid` matches a live watch carrying `flag`, build a staff
+    // feed line "nick!ident@host <what>". `chan` scopes a channel-mask watch to the
+    // relevant channel. None when nothing matches, the user is unknown, or there is
+    // no feed sink. The cheap `any_notifies` gate keeps the common (no-watch) path
+    // off the per-event identity lookup.
+    fn notify_line(&self, flag: char, uid: &str, chan: Option<&str>, what: &str) -> Option<NetAction> {
+        if !self.db.any_notifies() {
+            return None;
+        }
+        let target = self.network.ban_target(uid)?;
+        if !self.db.notify_flags(Some(&target), chan).contains(flag) {
+            return None;
+        }
+        let who = format!("\x02{}\x02!{}@{}", target.nick, target.ident, target.host);
+        self.feed("NOTIFY", format!("{who} {what}"))
+    }
+
     // Inactivity-expiry thresholds (seconds); None leaves that kind never
     // expiring. `warn` is the lead time before expiry to email a warning (None =
     // no warning emails).
@@ -1095,8 +1112,10 @@ impl Engine {
                 }
             }
             NetEvent::UserAttrs { uid, ident, realhost, gecos } => {
+                // Full identity (ident/host/gecos) is known now, so this is where a
+                // connect watch can match the whole nick!user@host#gecos.
                 self.network.set_user_attrs(&uid, ident, realhost, gecos);
-                Vec::new()
+                self.notify_line('c', &uid, None, "connected").into_iter().collect()
             }
             NetEvent::UserCert { uid, fp } => {
                 self.network.set_user_cert(&uid, fp);
@@ -1193,9 +1212,14 @@ impl Engine {
             }
             NetEvent::NickChange { uid, nick } => {
                 let new_nick = nick.clone();
+                let old_nick = self.network.nick_of(&uid).unwrap_or("?").to_string();
                 self.network.user_nick_change(&uid, nick);
                 self.pending_enforce.retain(|p| p.uid != uid); // they left the old nick
-                self.enforce_registered_nick(&uid, &new_nick)
+                let mut out = self.enforce_registered_nick(&uid, &new_nick);
+                if let Some(line) = self.notify_line('n', &uid, None, &format!("changed nick (was {old_nick})")) {
+                    out.push(line);
+                }
+                out
             }
             // The uplink finished its initial burst: from here on connections are
             // live, so nick-protection prompts/enforcement are safe to run.
@@ -1232,13 +1256,16 @@ impl Engine {
             // members their status mode, and send the entry message.
             NetEvent::Join { uid, channel, op } => {
                 self.network.channel_join(&channel, &uid, op);
+                // A watched user (or any user of a watched channel) joining: emit a
+                // staff line regardless of what enforcement below does.
+                let mut watch: Vec<NetAction> = self.notify_line('j', &uid, Some(&channel), &format!("joined {channel}")).into_iter().collect();
                 // A join to a registered channel is activity: keep it from expiring.
                 if self.db.channel(&channel).is_some() {
                     self.db.mark_channel_used(&channel, self.now_secs());
                 }
                 // A suspended channel is frozen: no auto-op, akick, or entry message.
                 if self.db.is_channel_suspended(&channel) {
-                    return Vec::new();
+                    return watch;
                 }
                 let account = self.network.account_of(&uid).map(str::to_string);
                 let from = self.channel_mode_source(&channel);
@@ -1251,7 +1278,7 @@ impl Engine {
                 let entrymsg = self.db.channel(&channel).map(|c| c.entrymsg.clone()).filter(|m| !m.is_empty());
                 // Computed before the match below moves `channel` into its actions.
                 let greet = self.greet_on_join(&channel, account.as_deref());
-                let mut out = Vec::new();
+                let mut out = std::mem::take(&mut watch);
                 // A services bot assigned here is opped on join and is never subject
                 // to secureops/restricted — it's staff, not a member.
                 let is_bot = self.bot_uids.values().any(|b| b == &uid);
@@ -1318,6 +1345,8 @@ impl Engine {
                 out
             }
             NetEvent::Part { uid, channel } => {
+                // Match before dropping membership so the identity is still resolvable.
+                let mut out: Vec<NetAction> = self.notify_line('p', &uid, Some(&channel), &format!("left {channel}")).into_iter().collect();
                 self.network.channel_part(&channel, &uid);
                 self.forget_chatter(&channel, &uid);
                 // A services bot kicked/parted from a channel it's still assigned
@@ -1325,10 +1354,10 @@ impl Engine {
                 // reconcile re-adds it.
                 if let Some(bot_lc) = self.bot_uids.iter().find_map(|(lc, u)| (u == &uid).then(|| lc.clone())) {
                     if self.bot_channels.remove(&(bot_lc, channel.clone())) {
-                        return self.reconcile_bots();
+                        out.extend(self.reconcile_bots());
                     }
                 }
-                Vec::new()
+                out
             }
             NetEvent::ChannelOp { channel, uid, op } => {
                 self.network.set_op(&channel, &uid, op);
@@ -1352,10 +1381,11 @@ impl Engine {
                 Vec::new()
             }
             NetEvent::TopicChange { channel, setter, topic } => {
+                let watch = self.notify_line('t', &setter, Some(&channel), &format!("changed the topic of {channel}"));
                 let info = self.db.channel(&channel).map(|c| (c.settings.keeptopic, c.settings.topiclock, c.topic.clone()));
                 // The setter may change a locked topic only with op-level access.
                 let authorized = self.network.account_of(&setter).and_then(|a| self.db.channel(&channel).map(|c| c.is_op(a))).unwrap_or(false);
-                match info {
+                let mut out: Vec<NetAction> = match info {
                     // TOPICLOCK + unauthorised: put the kept topic back.
                     Some((_, true, stored)) if !authorized => {
                         let from = self.chan_service.clone().unwrap_or_default();
@@ -1367,11 +1397,15 @@ impl Engine {
                         Vec::new()
                     }
                     _ => Vec::new(),
-                }
+                };
+                out.extend(watch);
+                out
             }
             NetEvent::Quit { uid } => {
+                // Match before we forget them, so their identity is still resolvable.
+                let out: Vec<NetAction> = self.notify_line('d', &uid, None, "disconnected").into_iter().collect();
                 self.forget_user(&uid);
-                Vec::new()
+                out
             }
             NetEvent::UserKilled { uid } => {
                 // A killed service agent (NickServ, ChanServ, …) has a fixed uid;
@@ -1736,6 +1770,8 @@ fn audit_summary(event: &db::Event) -> Option<String> {
         FilterRemoved { pattern } => format!("removed the spam filter on \x02{pattern}\x02"),
         ForbidAdded { kind, mask, reason, .. } => format!("forbade {} \x02{mask}\x02 ({reason})", kind.to_ascii_lowercase()),
         ForbidRemoved { kind, mask } => format!("un-forbade {} \x02{mask}\x02", kind.to_ascii_lowercase()),
+        NotifyAdded { mask, flags, .. } => format!("added a notify watch on \x02{mask}\x02 [{flags}]"),
+        NotifyRemoved { mask } => format!("removed the notify watch on \x02{mask}\x02"),
         JupeAdded { name, reason, .. } => format!("juped server \x02{name}\x02 ({reason})"),
         JupeRemoved { name } => format!("lifted the jupe on \x02{name}\x02"),
         AccountOperNoteSet { account, note } => match note {

src/engine/tests.rs

@@ -5356,6 +5356,62 @@ fn chanfix_and_diceserv_basic_paths() {
     assert!(svc_ask(&mut e, "42SAAAAAI", "ROLL 2d6").iter().any(|l| l.contains('=')), "dice result");
 }
 
+#[test]
+fn notify_flags_match_user_nick_channel_and_expiry() {
+    fn bt(nick: &'static str, host: &'static str) -> echo_api::BanTarget<'static> {
+        echo_api::BanTarget { nick, ident: "u", host, realhost: host, ip: "1.2.3.4", gecos: "g", account: None, server: "s.net", fingerprint: None, channels: vec![] }
+    }
+    let mut db = svc_db("notifymatch");
+    db.notify_add("baddie*!*@*", "cdn", "nick watch", "admin", None).unwrap();
+    db.notify_add("#warez*", "j", "chan watch", "admin", None).unwrap();
+    db.notify_add("*@spam.net", "c", "host watch", "admin", Some(1)).unwrap(); // expires at epoch 1 = long gone
+
+    // A user mask hits exactly its own flags, not another entry's.
+    let baddie = bt("baddie99", "h.example");
+    assert!(db.notify_flags(Some(&baddie), None).contains('c'), "connect flag");
+    assert!(db.notify_flags(Some(&baddie), None).contains('n'), "nick flag");
+    assert!(!db.notify_flags(Some(&baddie), None).contains('j'), "j belongs to the channel mask only");
+    // An unrelated user matches nothing.
+    assert!(db.notify_flags(Some(&bt("nice", "h.example")), None).is_empty(), "no match for a clean user");
+    // A channel mask matches on the joined channel, not on a different one.
+    assert!(db.notify_flags(Some(&bt("nice", "h.example")), Some("#warez7")).contains('j'), "channel watch hit");
+    assert!(!db.notify_flags(Some(&bt("nice", "h.example")), Some("#chat")).contains('j'), "channel watch miss");
+    // The expired host watch never matches and is hidden from the list.
+    assert!(db.notify_flags(Some(&bt("x", "spam.net")), None).is_empty(), "expired entry never matches");
+    assert_eq!(db.notifies().len(), 2, "expired entry hidden from the list");
+    // DEL by mask removes just that one.
+    assert!(db.notify_del("baddie*!*@*").unwrap());
+    assert_eq!(db.notifies().len(), 1);
+}
+
+#[test]
+fn notify_hook_emits_feed_for_watched_user() {
+    let mut e = engine_with("nfyhook", "foo", "sesame");
+    e.set_sid("42S".into());
+    e.set_log_channel(Some("#services".into()));
+    e.set_debugserv_uid("42SAAAAAO");
+    e.db.notify_add("baddie*!*@*", "cd", "watch", "admin", None).unwrap();
+
+    // A matching user's connect (identity complete at UserAttrs) feeds a line.
+    e.handle(NetEvent::UserConnect { uid: "000AAAAAC".into(), nick: "baddie1".into(), host: "h".into(), ip: "1.2.3.4".into() });
+    let out = e.handle(NetEvent::UserAttrs { uid: "000AAAAAC".into(), ident: "x".into(), realhost: "h".into(), gecos: "g".into() });
+    assert!(
+        out.iter().any(|a| matches!(a, NetAction::Privmsg { from, to, text }
+            if from == "42SAAAAAO" && to == "#services" && text.contains("[NOTIFY]") && text.contains("connected"))),
+        "expected a NOTIFY connect line, got {out:?}"
+    );
+    // A clean user draws nothing.
+    e.handle(NetEvent::UserConnect { uid: "000AAAAAD".into(), nick: "nice".into(), host: "h".into(), ip: "9.9.9.9".into() });
+    let quiet = e.handle(NetEvent::UserAttrs { uid: "000AAAAAD".into(), ident: "y".into(), realhost: "h".into(), gecos: "g".into() });
+    assert!(!quiet.iter().any(|a| matches!(a, NetAction::Privmsg { to, .. } if to == "#services")), "no line for a clean user: {quiet:?}");
+    // Their disconnect is matched before we forget them.
+    let dc = e.handle(NetEvent::Quit { uid: "000AAAAAC".into() });
+    assert!(
+        dc.iter().any(|a| matches!(a, NetAction::Privmsg { text, .. } if text.contains("[NOTIFY]") && text.contains("disconnected"))),
+        "expected a NOTIFY disconnect line, got {dc:?}"
+    );
+}
+
 #[test]
 fn cmd_limiter_bursts_then_throttles_per_host() {
     let mut lim = CmdLimiter::default();

src/grpc.rs

@@ -184,6 +184,8 @@ fn to_wire(entry: &LogEntry) -> Option<ReplicationEvent> {
         | Event::FilterRemoved { .. }
         | Event::ForbidAdded { .. }
         | Event::ForbidRemoved { .. }
+        | Event::NotifyAdded { .. }
+        | Event::NotifyRemoved { .. }
         | Event::JupeAdded { .. }
         | Event::JupeRemoved { .. }
         | Event::StatsSet { .. }