Compare commits
3 commits
feat/buzz-
...
feat/buzz-
| Author | SHA1 | Date | |
|---|---|---|---|
|
|
696d767fb4 | ||
|
|
5acf91a136 | ||
|
|
a3dd7fca24 |
1 changed files with 126 additions and 13 deletions
|
|
@ -10,6 +10,7 @@ import (
|
||||||
"fmt"
|
"fmt"
|
||||||
"strings"
|
"strings"
|
||||||
"sync"
|
"sync"
|
||||||
|
"time"
|
||||||
|
|
||||||
"github.com/nbd-wtf/go-nostr"
|
"github.com/nbd-wtf/go-nostr"
|
||||||
"github.com/nbd-wtf/go-nostr/nip19"
|
"github.com/nbd-wtf/go-nostr/nip19"
|
||||||
|
|
@ -23,6 +24,10 @@ import (
|
||||||
// kindStreamMessage is the Buzz chat message kind (NIP-29 style).
|
// kindStreamMessage is the Buzz chat message kind (NIP-29 style).
|
||||||
const kindStreamMessage = 9
|
const kindStreamMessage = 9
|
||||||
|
|
||||||
|
// subscribeTimeout bounds how long Start waits for the relay to accept or
|
||||||
|
// reject a subscription before giving up.
|
||||||
|
const subscribeTimeout = 15 * time.Second
|
||||||
|
|
||||||
// BuzzChannel implements the Channel interface for a Buzz relay.
|
// BuzzChannel implements the Channel interface for a Buzz relay.
|
||||||
type BuzzChannel struct {
|
type BuzzChannel struct {
|
||||||
*channels.BaseChannel
|
*channels.BaseChannel
|
||||||
|
|
@ -117,28 +122,17 @@ func (c *BuzzChannel) Start(ctx context.Context) error {
|
||||||
}
|
}
|
||||||
c.relay = relay
|
c.relay = relay
|
||||||
|
|
||||||
// NIP-42: the relay challenges, we sign the auth event with the bot identity.
|
|
||||||
// Buzz relays reject subscriptions from unauthenticated clients, so a failure
|
|
||||||
// here is fatal rather than advisory.
|
|
||||||
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)
|
channelIDs := []string(c.config.Channels)
|
||||||
filters := nostr.Filters{{
|
filters := nostr.Filters{{
|
||||||
Kinds: []int{kindStreamMessage},
|
Kinds: []int{kindStreamMessage},
|
||||||
Tags: nostr.TagMap{"h": channelIDs},
|
Tags: nostr.TagMap{"h": channelIDs},
|
||||||
}}
|
}}
|
||||||
|
|
||||||
sub, err := relay.Subscribe(c.ctx, filters)
|
sub, err := c.subscribeAuthed(c.ctx, filters)
|
||||||
if err != nil {
|
if err != nil {
|
||||||
_ = relay.Close()
|
_ = relay.Close()
|
||||||
c.cancel()
|
c.cancel()
|
||||||
return fmt.Errorf("buzz subscribe failed: %w", err)
|
return err
|
||||||
}
|
}
|
||||||
c.sub = sub
|
c.sub = sub
|
||||||
|
|
||||||
|
|
@ -157,6 +151,84 @@ func (c *BuzzChannel) Start(ctx context.Context) error {
|
||||||
return nil
|
return nil
|
||||||
}
|
}
|
||||||
|
|
||||||
|
// subscribeAuthed subscribes, performing the NIP-42 handshake if the relay
|
||||||
|
// demands it.
|
||||||
|
//
|
||||||
|
// The handshake MUST be driven by the relay's rejection rather than attempted
|
||||||
|
// eagerly after connect. go-nostr stores the relay's challenge on an unexported
|
||||||
|
// field populated by its reader goroutine when the AUTH envelope arrives;
|
||||||
|
// calling Auth() straight after RelayConnect signs an auth event with an empty
|
||||||
|
// challenge tag, which every relay rejects. Waiting for "auth-required"
|
||||||
|
// guarantees the challenge has been read — the CLOSED envelope is processed
|
||||||
|
// after the AUTH envelope on that same goroutine, and receiving it over a
|
||||||
|
// channel establishes the happens-before edge that makes the read safe.
|
||||||
|
func (c *BuzzChannel) subscribeAuthed(
|
||||||
|
ctx context.Context,
|
||||||
|
filters nostr.Filters,
|
||||||
|
) (*nostr.Subscription, error) {
|
||||||
|
sub, err := c.relay.Subscribe(ctx, filters)
|
||||||
|
if err != nil {
|
||||||
|
return nil, fmt.Errorf("buzz subscribe failed: %w", err)
|
||||||
|
}
|
||||||
|
|
||||||
|
select {
|
||||||
|
case <-ctx.Done():
|
||||||
|
return nil, ctx.Err()
|
||||||
|
|
||||||
|
case <-sub.EndOfStoredEvents:
|
||||||
|
// Relay accepted the subscription without requiring authentication.
|
||||||
|
return sub, nil
|
||||||
|
|
||||||
|
case reason := <-sub.ClosedReason:
|
||||||
|
if !strings.HasPrefix(reason, "auth-required") {
|
||||||
|
return nil, fmt.Errorf("buzz relay closed subscription: %s", reason)
|
||||||
|
}
|
||||||
|
sub.Unsub()
|
||||||
|
|
||||||
|
if err := c.relay.Auth(ctx, func(evt *nostr.Event) error {
|
||||||
|
return evt.Sign(c.secretKey)
|
||||||
|
}); err != nil {
|
||||||
|
return nil, fmt.Errorf("buzz NIP-42 auth failed: %w", err)
|
||||||
|
}
|
||||||
|
logger.InfoCF("buzz", "Authenticated to relay", map[string]any{
|
||||||
|
"relay": c.config.RelayURL,
|
||||||
|
"pubkey": c.publicKey,
|
||||||
|
})
|
||||||
|
|
||||||
|
// postAuthGrace is how long we wait after re-subscribing for either
|
||||||
|
// EOSE (relay sends end-of-stored-events) or a rejection. If neither
|
||||||
|
// arrives, the relay accepted the subscription but doesn't send EOSE —
|
||||||
|
// which is fine, we just start consuming events.
|
||||||
|
const postAuthGrace = 3 * time.Second
|
||||||
|
|
||||||
|
authed, err := c.relay.Subscribe(ctx, filters)
|
||||||
|
if err != nil {
|
||||||
|
return nil, fmt.Errorf("buzz subscribe after auth failed: %w", err)
|
||||||
|
}
|
||||||
|
select {
|
||||||
|
case <-ctx.Done():
|
||||||
|
return nil, ctx.Err()
|
||||||
|
case <-authed.EndOfStoredEvents:
|
||||||
|
return authed, nil
|
||||||
|
case reason := <-authed.ClosedReason:
|
||||||
|
// A second rejection means the identity is not permitted, not that
|
||||||
|
// the handshake was mistimed — retrying would loop forever.
|
||||||
|
return nil, fmt.Errorf("buzz relay rejected subscription after auth: %s", reason)
|
||||||
|
case <-time.After(postAuthGrace):
|
||||||
|
// No EOSE and no rejection within the grace period — the relay
|
||||||
|
// accepted the subscription. Some relays (including Buzz) don't
|
||||||
|
// send EOSE after auth, so this is the expected success path.
|
||||||
|
logger.InfoCF("buzz", "Subscription accepted after auth (no EOSE)", map[string]any{
|
||||||
|
"relay": c.config.RelayURL,
|
||||||
|
})
|
||||||
|
return authed, nil
|
||||||
|
}
|
||||||
|
|
||||||
|
case <-time.After(subscribeTimeout):
|
||||||
|
return nil, fmt.Errorf("buzz timed out waiting for relay to accept subscription")
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
// Stop closes the subscription and disconnects from the relay.
|
// Stop closes the subscription and disconnects from the relay.
|
||||||
func (c *BuzzChannel) Stop(ctx context.Context) error {
|
func (c *BuzzChannel) Stop(ctx context.Context) error {
|
||||||
logger.InfoC("buzz", "Stopping Buzz channel")
|
logger.InfoC("buzz", "Stopping Buzz channel")
|
||||||
|
|
@ -220,6 +292,47 @@ func (c *BuzzChannel) Send(ctx context.Context, msg bus.OutboundMessage) ([]stri
|
||||||
return []string{evt.ID}, nil
|
return []string{evt.ID}, nil
|
||||||
}
|
}
|
||||||
|
|
||||||
|
// kindReaction is the NIP-25 reaction event kind.
|
||||||
|
const kindReaction = 7
|
||||||
|
|
||||||
|
// ReactToMessage implements channels.ReactionCapable.
|
||||||
|
// It publishes a NIP-25 kind:7 reaction event with content 👀 scoped to the
|
||||||
|
// same channel as the reacted message. The undo function is a no-op because
|
||||||
|
// NIP-25 does not define a standard way to remove a reaction.
|
||||||
|
func (c *BuzzChannel) ReactToMessage(ctx context.Context, chatID, messageID string) (func(), error) {
|
||||||
|
if !c.IsRunning() {
|
||||||
|
return func() {}, channels.ErrNotRunning
|
||||||
|
}
|
||||||
|
if messageID == "" {
|
||||||
|
return func() {}, nil
|
||||||
|
}
|
||||||
|
|
||||||
|
evt := nostr.Event{
|
||||||
|
PubKey: c.publicKey,
|
||||||
|
CreatedAt: nostr.Now(),
|
||||||
|
Kind: kindReaction,
|
||||||
|
Tags: nostr.Tags{
|
||||||
|
nostr.Tag{"e", messageID},
|
||||||
|
nostr.Tag{"h", chatID},
|
||||||
|
},
|
||||||
|
Content: "👀",
|
||||||
|
}
|
||||||
|
if err := evt.Sign(c.secretKey); err != nil {
|
||||||
|
return func() {}, fmt.Errorf("buzz reaction sign failed: %w", err)
|
||||||
|
}
|
||||||
|
|
||||||
|
if err := c.relay.Publish(ctx, evt); err != nil {
|
||||||
|
return func() {}, fmt.Errorf("buzz reaction publish failed: %w", err)
|
||||||
|
}
|
||||||
|
|
||||||
|
logger.DebugCF("buzz", "Reaction sent", map[string]any{
|
||||||
|
"channel": chatID,
|
||||||
|
"event_id": evt.ID,
|
||||||
|
"target_id": messageID,
|
||||||
|
})
|
||||||
|
return func() {}, nil
|
||||||
|
}
|
||||||
|
|
||||||
// errJoin wraps err so that errors.Is(result, sentinel) holds for the sentinel
|
// errJoin wraps err so that errors.Is(result, sentinel) holds for the sentinel
|
||||||
// while preserving the underlying cause in the message.
|
// while preserving the underlying cause in the message.
|
||||||
func errJoin(err, sentinel error) error {
|
func errJoin(err, sentinel error) error {
|
||||||
|
|
|
||||||
Loading…
Reference in a new issue