picoclaw/pkg/channels/buzz/handler.go

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:]
}