fix(buzz): suppress replayed events after restart

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()

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)
+}

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,

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-<channel-name>.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-<defaultName>.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)
+}

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-<channel-name>.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 {