From 02961af54fafa5ca442d5955b306a4ffefafbbfa Mon Sep 17 00:00:00 2001 From: PeterChrz Date: Tue, 25 Aug 2026 00:24:49 -0400 Subject: [PATCH] fix(buzz): suppress replayed events after restart --- pkg/channels/buzz/buzz.go | 44 +++--- pkg/channels/buzz/buzz_test.go | 269 ++++++++++++++++++++++++++++++++ pkg/channels/buzz/handler.go | 22 +++ pkg/channels/buzz/seen_store.go | 185 ++++++++++++++++++++++ pkg/config/config.go | 5 + 5 files changed, 504 insertions(+), 21 deletions(-) create mode 100644 pkg/channels/buzz/seen_store.go diff --git a/pkg/channels/buzz/buzz.go b/pkg/channels/buzz/buzz.go index 370f6d2c..58d4d516 100644 --- a/pkg/channels/buzz/buzz.go +++ b/pkg/channels/buzz/buzz.go @@ -38,6 +38,13 @@ type BuzzChannel struct { ctx context.Context cancel context.CancelFunc wg sync.WaitGroup + + // seen persists per-channel high-water state so a service restart does + // not redeliver already-handled kind:9 events to HandleInboundContext. + // It tracks the max seen CreatedAt (used to set a Since filter on + // resubscribe) and a bounded ring of recent event IDs (the authoritative + // dedup guard, since NIP-01 Since is inclusive and some relays ignore it). + seen *seenStore } // NewBuzzChannel creates a new Buzz channel. @@ -73,6 +80,7 @@ func NewBuzzChannel( config: cfg, secretKey: sk, publicKey: pk, + seen: newSeenStore(cfg.SeenStatePath, bc.Name()), }, nil } @@ -118,33 +126,27 @@ func (c *BuzzChannel) Start(ctx context.Context) error { } c.relay = relay - // NIP-42: the relay sends an AUTH challenge asynchronously after the - // WebSocket handshake completes. go-nostr stores the challenge in an - // unexported field populated by the background read loop. If we call - // Auth() before the challenge arrives, the auth event carries an empty - // challenge tag and the relay rejects it permanently ("verification - // failed" → "authentication already failed" on all subsequent attempts). - // - // We cannot inspect the challenge field directly (unexported), so we wait - // briefly to give the read loop time to receive and store it. The relay - // typically delivers the challenge within a few hundred milliseconds. - time.Sleep(2 * time.Second) - - if err := relay.Auth(c.ctx, func(evt *nostr.Event) error { - return evt.Sign(c.secretKey) - }); err != nil { - _ = relay.Close() - c.cancel() - return fmt.Errorf("buzz NIP-42 auth failed: %w", err) - } - channelIDs := []string(c.config.Channels) filters := nostr.Filters{{ Kinds: []int{kindStreamMessage}, Tags: nostr.TagMap{"h": channelIDs}, }} - sub, err := relay.Subscribe(c.ctx, filters) + // Replay suppression: skip events older than the highest CreatedAt we have + // already handled across all channels. NIP-01 Since is inclusive, so events + // at exactly the high-water timestamp may still be redelivered; the seen + // store's recent-ID ring drops those client-side. A zero high-water mark + // (first run, or empty store) leaves Since nil so we receive full history + // once — exactly as before. + if hw := c.seen.highWaterMark(); hw > 0 { + since := hw + filters[0].Since = &since + logger.InfoCF("buzz", "Replay filter from high-water mark", map[string]any{ + "since": since.Time().Format(time.RFC3339), + }) + } + + sub, err := c.subscribeAuthed(c.ctx, filters) if err != nil { _ = relay.Close() c.cancel() diff --git a/pkg/channels/buzz/buzz_test.go b/pkg/channels/buzz/buzz_test.go index ef8ebff5..ab945939 100644 --- a/pkg/channels/buzz/buzz_test.go +++ b/pkg/channels/buzz/buzz_test.go @@ -1,11 +1,18 @@ package buzz import ( + "context" + "os" + "path/filepath" "strings" "testing" + "time" "github.com/nbd-wtf/go-nostr" "github.com/nbd-wtf/go-nostr/nip19" + + "github.com/sipeed/picoclaw/pkg/bus" + "github.com/sipeed/picoclaw/pkg/config" ) func TestNormalizeSecretKeyAcceptsHex(t *testing.T) { @@ -125,3 +132,265 @@ func TestShortPubkey(t *testing.T) { t.Fatalf("expected short input unchanged, got %s", got) } } + +// newTestBuzzChannel builds a real BuzzChannel wired to a temp-disk seen store +// and a live message bus, without touching the network. The caller's pubkey +// (c.publicKey) is the bot identity; events are signed with a different key so +// they are not treated as self-messages. +func newTestBuzzChannel(t *testing.T, seenPath string) (*BuzzChannel, *bus.MessageBus, context.Context) { + t.Helper() + + botSK := nostr.GeneratePrivateKey() + if _, err := nostr.GetPublicKey(botSK); err != nil { + t.Fatalf("bot pubkey: %v", err) + } + + bc := &config.Channel{} + bc.SetName("buzz-test") + cfg := &config.BuzzSettings{ + RelayURL: "wss://relay.invalid", + PrivateKey: *config.NewSecureString(botSK), + Channels: config.FlexibleStringSlice{"test-channel"}, + SeenStatePath: seenPath, + } + + msgBus := bus.NewMessageBus() + t.Cleanup(msgBus.Close) + + ch, err := NewBuzzChannel(bc, cfg, msgBus) + if err != nil { + t.Fatalf("NewBuzzChannel: %v", err) + } + // onEvent reads c.ctx; Start would set it but we don't want a relay. + ch.ctx, ch.cancel = context.WithCancel(context.Background()) + t.Cleanup(func() { + if ch.cancel != nil { + ch.cancel() + } + }) + return ch, msgBus, ch.ctx +} + +// makeEvent builds and signs a kind:9 event from senderSK scoped to chatID, +// returning a real nostr.Event with a valid ID. +func makeEvent(t *testing.T, senderSK, chatID, content string, createdAt nostr.Timestamp) *nostr.Event { + t.Helper() + senderPK, err := nostr.GetPublicKey(senderSK) + if err != nil { + t.Fatalf("sender pubkey: %v", err) + } + evt := &nostr.Event{ + PubKey: senderPK, + CreatedAt: createdAt, + Kind: kindStreamMessage, + Tags: nostr.Tags{nostr.Tag{"h", chatID}}, + Content: content, + } + if err := evt.Sign(senderSK); err != nil { + t.Fatalf("sign: %v", err) + } + return evt +} + +// expectInbound drains one inbound message from the bus, failing if none +// arrives within the timeout or if the bus is closed empty. +func expectInbound(t *testing.T, msgBus *bus.MessageBus, wantContent string) { + t.Helper() + select { + case msg, ok := <-msgBus.InboundChan(): + if !ok { + t.Fatal("expected an inbound message but bus channel was closed") + } + if msg.Content != wantContent { + t.Fatalf("inbound content = %q, want %q", msg.Content, wantContent) + } + case <-time.After(time.Second): + t.Fatal("expected an inbound message but none arrived within 1s") + } +} + +// expectNoInbound asserts that no inbound message arrives within the timeout, +// i.e. the event was suppressed by the seen store. +func expectNoInbound(t *testing.T, msgBus *bus.MessageBus) { + t.Helper() + select { + case msg, ok := <-msgBus.InboundChan(): + if ok { + t.Fatalf("expected no inbound message, but got one: %q (id=%q)", msg.Content, msg.Context.MessageID) + } + case <-time.After(200 * time.Millisecond): + // No message within the window — replay suppression held. This is the + // success path. + } +} + +// TestReplaySuppression_PersistedEventNotRedelivered covers the core restart +// scenario: an event handled in run 1 is recorded to the persistent store; a +// fresh channel loaded from the same store path must not redeliver it. +func TestReplaySuppression_PersistedEventNotRedelivered(t *testing.T) { + seenPath := filepath.Join(t.TempDir(), "seen.json") + + // Run 1: a fresh channel handles one event. + ch1, msgBus1, _ := newTestBuzzChannel(t, seenPath) + senderSK := nostr.GeneratePrivateKey() + evt := makeEvent(t, senderSK, "test-channel", "hello from the past", nostr.Now()) + + ch1.onEvent(evt) + expectInbound(t, msgBus1, "hello from the past") + + // The store file must exist on disk after handling. + if _, err := os.Stat(seenPath); err != nil { + t.Fatalf("stat seen file: %v", err) + } + + // Run 2: a brand-new channel instance reading the same store must not + // redeliver the same event ID, simulating a service restart. + ch2, msgBus2, _ := newTestBuzzChannel(t, seenPath) + ch2.onEvent(evt) + expectNoInbound(t, msgBus2) + + // High-water mark must reflect the handled event's CreatedAt so the + // subscription Since filter would skip older history on restart. + if hw := ch2.seen.highWaterMark(); hw != evt.CreatedAt { + t.Fatalf("high-water mark = %d, want %d", hw, evt.CreatedAt) + } +} + +// TestReplaySuppression_FreshEventStillHandled ensures the seen store does not +// suppress events that were never handled before — the live delivery path is +// preserved. +func TestReplaySuppression_FreshEventStillHandled(t *testing.T) { + seenPath := filepath.Join(t.TempDir(), "seen.json") + ch, msgBus, _ := newTestBuzzChannel(t, seenPath) + senderSK := nostr.GeneratePrivateKey() + + first := makeEvent(t, senderSK, "test-channel", "first message", nostr.Now()) + ch.onEvent(first) + expectInbound(t, msgBus, "first message") + + // A different event ID must still be delivered. + second := makeEvent(t, senderSK, "test-channel", "second message", nostr.Now()+1) + ch.onEvent(second) + expectInbound(t, msgBus, "second message") + + // Replaying the first event again is still suppressed within the same run + // (idempotency, not just restart protection). + ch.onEvent(first) + expectNoInbound(t, msgBus) +} + +// TestReplaySuppression_SelfMessageNotRecorded confirms self-message filtering +// still works and that such messages do not pollute the seen store (so a future +// legitimate event with the same hypothetical ID — impossible in practice but +// the invariant matters — would not be wrongly suppressed). +func TestReplaySuppression_SelfMessageNotRecorded(t *testing.T) { + seenPath := filepath.Join(t.TempDir(), "seen.json") + ch, msgBus, _ := newTestBuzzChannel(t, seenPath) + + // Build an event signed with the bot's own key so PubKey == c.publicKey. + selfEvt := makeEvent(t, ch.secretKey, "test-channel", "self echo", nostr.Now()) + ch.onEvent(selfEvt) + expectNoInbound(t, msgBus) + + // The store should have no entry for this channel because self-messages + // return before markHandled. + if ch.seen.seen("test-channel", selfEvt.ID, selfEvt.CreatedAt) { + t.Fatal("self-message was recorded in the seen store; it should be skipped before recording") + } +} + +// TestSeenStoreHighWaterMarkAcrossChannels verifies the high-water mark is the +// max across all channels, so the subscription Since filter uses a single +// global cutoff that skips history for every channel. +func TestSeenStoreHighWaterMarkAcrossChannels(t *testing.T) { + s := newSeenStore(filepath.Join(t.TempDir(), "seen.json"), "test") + + if hw := s.highWaterMark(); hw != 0 { + t.Fatalf("empty store high-water = %d, want 0", hw) + } + + if err := s.markHandled("chan-a", "id-a1", 100); err != nil { + t.Fatalf("markHandled a: %v", err) + } + if err := s.markHandled("chan-b", "id-b1", 150); err != nil { + t.Fatalf("markHandled b: %v", err) + } + if err := s.markHandled("chan-a", "id-a2", 120); err != nil { + t.Fatalf("markHandled a2: %v", err) + } + + if hw := s.highWaterMark(); hw != 150 { + t.Fatalf("high-water = %d, want 150", hw) + } + + // Reload from disk to confirm persistence. + s2 := newSeenStore(s.path, "test") + if hw := s2.highWaterMark(); hw != 150 { + t.Fatalf("after reload, high-water = %d, want 150", hw) + } + if !s2.seen("chan-a", "id-a1", 100) || !s2.seen("chan-b", "id-b1", 150) { + t.Fatal("reloaded store lost a recorded event ID") + } +} + +func TestReplaySuppression_EventsOlderThanSeededHighWaterAreDropped(t *testing.T) { + seenPath := filepath.Join(t.TempDir(), "seen.json") + ch, msgBus, _ := newTestBuzzChannel(t, seenPath) + if err := ch.seen.markHandled("test-channel", "seed", nostr.Now()); err != nil { + t.Fatalf("seed high-water: %v", err) + } + + oldEvent := makeEvent( + t, + nostr.GeneratePrivateKey(), + "test-channel", + "replayed history", + nostr.Now()-60, + ) + ch.onEvent(oldEvent) + expectNoInbound(t, msgBus) +} + +// TestSeenStoreRingTrims verifies the recent-ID ring does not grow unbounded, +// keeping only the most recent recentIDWindow entries per channel. +func TestSeenStoreRingTrims(t *testing.T) { + s := newSeenStore(filepath.Join(t.TempDir(), "seen.json"), "test") + + // Insert more IDs than the window; old IDs should be evicted. + for i := 0; i < recentIDWindow+50; i++ { + id := "id-" + strings.Repeat("x", i%4) + // Use unique IDs so they are not deduped by the ring's duplicate check. + id = "id-" + nip19Letter(i) + if err := s.markHandled("chan", id, nostr.Timestamp(i)); err != nil { + t.Fatalf("markHandled %d: %v", i, err) + } + } + + hw, ok := s.highWater["chan"] + if !ok { + t.Fatal("channel entry missing") + } + if len(hw.RecentIDs) != recentIDWindow { + t.Fatalf("ring length = %d, want %d", len(hw.RecentIDs), recentIDWindow) + } + // The first 50 IDs must have been evicted; the oldest remaining must be id + // number 50 (0-indexed). + if hw.RecentIDs[0] != "id-"+nip19Letter(50) { + t.Fatalf("oldest remaining ID = %q, want id for index 50", hw.RecentIDs[0]) + } +} + +// nip19Letter converts n to a short distinct string for unique IDs in the ring +// test. Using a deterministic suffix keeps IDs unique and short. +func nip19Letter(n int) string { + const digits = "0123456789abcdefghijklmnopqrstuvwxyz" + var out []byte + if n == 0 { + return "0" + } + for n > 0 { + out = append([]byte{digits[n%36]}, out...) + n /= 36 + } + return string(out) +} diff --git a/pkg/channels/buzz/handler.go b/pkg/channels/buzz/handler.go index c8c9b511..c7ad1f70 100644 --- a/pkg/channels/buzz/handler.go +++ b/pkg/channels/buzz/handler.go @@ -48,6 +48,15 @@ func (c *BuzzChannel) onEvent(evt *nostr.Event) { return } + // Replay suppression: drop events we have already handed to + // HandleInboundContext in a prior run. The seen store persists per-channel + // high-water state; this guard survives restarts even if the relay + // redelivers stored history (it ignores the Since filter or sends events + // at the inclusive Since boundary). + if c.seen != nil && c.seen.seen(chatID, evt.ID, evt.CreatedAt) { + return + } + sender := bus.SenderInfo{ Platform: "buzz", PlatformID: evt.PubKey, @@ -74,6 +83,19 @@ func (c *BuzzChannel) onEvent(evt *nostr.Event) { return } + // Record this event as handled before dispatching so a crash between here + // and HandleInboundContext does not cause a redelivery on the next start. + // The store is persisted to disk; the recent-ID ring absorbs the + // inclusive-Since boundary redelivery on restart. + if c.seen != nil { + if err := c.seen.markHandled(chatID, evt.ID, evt.CreatedAt); err != nil { + logger.WarnCF("buzz", "Failed to persist seen state", map[string]any{ + "error": err.Error(), + "event_id": evt.ID, + }) + } + } + inboundCtx := bus.InboundContext{ Channel: "buzz", ChatID: chatID, diff --git a/pkg/channels/buzz/seen_store.go b/pkg/channels/buzz/seen_store.go new file mode 100644 index 00000000..72079991 --- /dev/null +++ b/pkg/channels/buzz/seen_store.go @@ -0,0 +1,185 @@ +package buzz + +import ( + "encoding/json" + "errors" + "os" + "path/filepath" + "sync" + + "github.com/nbd-wtf/go-nostr" + + "github.com/sipeed/picoclaw/pkg/config" +) + +// recentIDWindow is how many recently-handled event IDs we keep per channel to +// survive the inclusive-Since boundary (NIP-01 redelivers events with +// created_at >= Since) and relays that silently ignore the Since filter. +// A few hundred is ample: it covers the burst of events at the high-water +// timestamp plus any stragglers a misbehaving relay re-sends. +const recentIDWindow = 256 + +// channelHighWater records the highest seen event created_at and a bounded +// ring of recently handled event IDs for a single Buzz channel ("h" tag). +type channelHighWater struct { + MaxCreatedAt int64 `json:"max_created_at"` + RecentIDs []string `json:"recent_ids,omitempty"` +} + +// seenStore persists per-channel high-water state so a service restart does +// not redeliver already-handled kind:9 events to HandleInboundContext. +// +// The store tracks, per "h" tag value: +// - MaxCreatedAt: the largest event.CreatedAt we have handled, used to set a +// Since filter on resubscribe so the relay skips most stored history. +// - RecentIDs: a bounded ring of the most recently handled event IDs. This +// is the authoritative dedup guard: Since is inclusive in NIP-01, so events +// at exactly the high-water timestamp are redelivered, and some relays +// ignore Since entirely. The ring absorbs both cases without unbounded +// growth. +// +// The store is JSON-backed under ~/.picoclaw/buzz/seen-.json, +// matching the persistence pattern used by the wecom reqID store. +type seenStore struct { + mu sync.Mutex + path string + highWater map[string]*channelHighWater +} + +// newSeenStore opens (or creates) the persistent seen-state store at path. +// If path is empty it defaults to ~/.picoclaw/buzz/seen-.json. +// A missing file is not an error; the store starts empty. +func newSeenStore(path, defaultName string) *seenStore { + if path == "" { + path = defaultSeenStorePath(defaultName) + } + s := &seenStore{ + path: path, + highWater: make(map[string]*channelHighWater), + } + _ = s.load() + return s +} + +func defaultSeenStorePath(channelName string) string { + home, err := os.UserHomeDir() + if err != nil || home == "" { + return filepath.Join(os.TempDir(), "picoclaw-buzz-seen-"+channelName+".json") + } + name := channelName + if name == "" { + name = config.ChannelBuzz + } + return filepath.Join(home, ".picoclaw", "buzz", "seen-"+name+".json") +} + +// highWaterMark returns the maximum seen CreatedAt across all channels, or +// zero if nothing has been recorded. Callers use it to set the subscription +// Since filter so the relay skips events older than what we have handled. +func (s *seenStore) highWaterMark() nostr.Timestamp { + s.mu.Lock() + defer s.mu.Unlock() + var max int64 + for _, hw := range s.highWater { + if hw.MaxCreatedAt > max { + max = hw.MaxCreatedAt + } + } + return nostr.Timestamp(max) +} + +// seen reports whether an event has already been handled for chatID. Events +// older than the channel's high-water timestamp are necessarily replayed +// history. Events at the inclusive high-water boundary require an ID lookup. +func (s *seenStore) seen(chatID, eventID string, createdAt nostr.Timestamp) bool { + if eventID == "" { + return false + } + s.mu.Lock() + defer s.mu.Unlock() + hw, ok := s.highWater[chatID] + if !ok { + return false + } + if int64(createdAt) < hw.MaxCreatedAt { + return true + } + for _, id := range hw.RecentIDs { + if id == eventID { + return true + } + } + return false +} + +// markHandled records that eventID (with createdAt) has been handled for +// chatID and persists the store. It updates MaxCreatedAt when createdAt is +// newer and trims the RecentIDs ring to recentIDWindow entries. +func (s *seenStore) markHandled(chatID, eventID string, createdAt nostr.Timestamp) error { + if chatID == "" || eventID == "" { + return nil + } + s.mu.Lock() + defer s.mu.Unlock() + hw, ok := s.highWater[chatID] + if !ok { + hw = &channelHighWater{} + s.highWater[chatID] = hw + } + ts := int64(createdAt) + if ts > hw.MaxCreatedAt { + hw.MaxCreatedAt = ts + } + // Avoid duplicate entries in the ring. + for _, id := range hw.RecentIDs { + if id == eventID { + return s.saveLocked() + } + } + hw.RecentIDs = append(hw.RecentIDs, eventID) + if len(hw.RecentIDs) > recentIDWindow { + // Drop the oldest entries, keeping only the most recent window. + drop := len(hw.RecentIDs) - recentIDWindow + hw.RecentIDs = hw.RecentIDs[drop:] + } + return s.saveLocked() +} + +// load reads the persisted high-water state. A missing file is not an error. +func (s *seenStore) load() error { + s.mu.Lock() + defer s.mu.Unlock() + + data, err := os.ReadFile(s.path) + if err != nil { + if errors.Is(err, os.ErrNotExist) { + return nil + } + return err + } + var hw map[string]*channelHighWater + if err := json.Unmarshal(data, &hw); err != nil { + return err + } + s.highWater = hw + if s.highWater == nil { + s.highWater = make(map[string]*channelHighWater) + } + return nil +} + +// saveLocked writes the high-water state to disk. Caller must hold s.mu. +func (s *seenStore) saveLocked() error { + if err := os.MkdirAll(filepath.Dir(s.path), 0o700); err != nil { + return err + } + data, err := json.MarshalIndent(s.highWater, "", " ") + if err != nil { + return err + } + tmp := s.path + ".tmp" + if err := os.WriteFile(tmp, data, 0o600); err != nil { + return err + } + return os.Rename(tmp, s.path) +} diff --git a/pkg/config/config.go b/pkg/config/config.go index 532bcad4..32d84377 100644 --- a/pkg/config/config.go +++ b/pkg/config/config.go @@ -721,6 +721,11 @@ type BuzzSettings struct { // ReplyInThread emits an "e" tag referencing the triggering event so replies // thread under it rather than appearing as top-level channel messages. ReplyInThread bool `json:"reply_in_thread,omitempty" yaml:"-" env:"PICOCLAW_CHANNELS_BUZZ_REPLY_IN_THREAD"` + // SeenStatePath overrides the on-disk location of the replay-suppression + // store. When empty it defaults to ~/.picoclaw/buzz/seen-.json. + // The store records the highest handled event timestamp per channel so a + // service restart does not redeliver already-handled kind:9 events. + SeenStatePath string `json:"seen_state_path,omitempty" yaml:"-" env:"PICOCLAW_CHANNELS_BUZZ_SEEN_STATE_PATH"` } type VKSettings struct {