package buzz import ( "strings" "github.com/nbd-wtf/go-nostr" "github.com/sipeed/picoclaw/pkg/bus" "github.com/sipeed/picoclaw/pkg/identity" "github.com/sipeed/picoclaw/pkg/logger" ) // consume drains the subscription until the channel context is cancelled. func (c *BuzzChannel) consume(sub *nostr.Subscription) { for { select { case <-c.ctx.Done(): return case reason, ok := <-sub.ClosedReason: if !ok { return } logger.WarnCF("buzz", "Relay closed subscription", map[string]any{"reason": reason}) return case evt, ok := <-sub.Events: if !ok { return } c.onEvent(evt) } } } // onEvent converts a kind:9 relay event into an inbound bus message. func (c *BuzzChannel) onEvent(evt *nostr.Event) { if evt == nil || evt.Kind != kindStreamMessage { return } // Ignore our own messages, otherwise the agent replies to itself. if evt.PubKey == c.publicKey { return } // The "h" tag scopes the event to a channel; without it there is nowhere to reply. chatID := firstTagValue(evt.Tags, "h") if chatID == "" { 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, CanonicalID: identity.BuildCanonicalID("buzz", evt.PubKey), Username: shortPubkey(evt.PubKey), DisplayName: shortPubkey(evt.PubKey), } if !c.IsAllowedSender(sender) { return } // Mentions are "p" tags carrying the mentioned pubkey. isMentioned := hasPTag(evt.Tags, c.publicKey) content := evt.Content respond, cleaned := c.ShouldRespondInGroup(isMentioned, content) if !respond { return } content = cleaned if strings.TrimSpace(content) == "" { 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 we cannot persist the state, we must NOT dispatch: a successful but // un-persisted mark would let the same event be redelivered (and re-handled) // on the next restart, violating at-most-once inbound delivery. Skip this // event and let the relay redeliver it later, when persistence may succeed. if c.seen != nil { if err := c.seen.markHandled(chatID, evt.ID, evt.CreatedAt); err != nil { logger.WarnCF("buzz", "Failed to persist seen state; skipping dispatch", map[string]any{ "error": err.Error(), "event_id": evt.ID, "chat_id": chatID, }) return } logger.DebugCF("buzz", "Persisted seen state for event", map[string]any{ "chat_id": chatID, "event_id": evt.ID, "created_at": int64(evt.CreatedAt), "high_water": int64(c.seen.highWaterMark()), "seen_path": c.seen.path, }) } inboundCtx := bus.InboundContext{ Channel: "buzz", ChatID: chatID, ChatType: "channel", SenderID: evt.PubKey, MessageID: evt.ID, Mentioned: isMentioned, Raw: map[string]string{ "platform": "buzz", "relay": c.config.RelayURL, "channel": chatID, }, } if err := c.HandleInboundContext(c.ctx, chatID, content, nil, inboundCtx, sender); err != nil { logger.WarnCF("buzz", "Inbound handling failed", map[string]any{ "error": err.Error(), "event_id": evt.ID, }) } } // firstTagValue returns the value of the first tag with the given key. func firstTagValue(tags nostr.Tags, key string) string { for _, t := range tags { if len(t) >= 2 && t[0] == key { return t[1] } } return "" } // hasPTag reports whether any "p" tag references the given pubkey. func hasPTag(tags nostr.Tags, pubkey string) bool { for _, t := range tags { if len(t) >= 2 && t[0] == "p" && strings.EqualFold(t[1], pubkey) { return true } } return false } // shortPubkey renders a pubkey as a readable handle for display purposes. func shortPubkey(pubkey string) string { if len(pubkey) <= 12 { return pubkey } return pubkey[:8] + "…" + pubkey[len(pubkey)-4:] }