Enhance relay testing and event handling
- Updated TestRelay to include a wait mechanism for relay readiness, improving test reliability. - Refactored startTestRelay to return the assigned port, allowing dynamic port assignment. - Added timestamp validation in HandleEvent to reject events with timestamps more than one hour in the future. - Introduced channels for handling OK and COUNT messages in the Client struct, improving message processing. - Updated tests to reflect changes in event timestamp handling and increased wait times for event processing. - Bumped version to v0.20.6 to reflect these enhancements.
This commit is contained in:
@@ -18,6 +18,8 @@ type Client struct {
|
||||
url string
|
||||
mu sync.Mutex
|
||||
subs map[string]chan []byte
|
||||
okCh chan []byte // Channel for OK messages
|
||||
countCh chan []byte // Channel for COUNT messages
|
||||
ctx context.Context
|
||||
cancel context.CancelFunc
|
||||
}
|
||||
@@ -34,11 +36,13 @@ func NewClient(url string) (c *Client, err error) {
|
||||
return
|
||||
}
|
||||
c = &Client{
|
||||
conn: conn,
|
||||
url: url,
|
||||
subs: make(map[string]chan []byte),
|
||||
ctx: ctx,
|
||||
cancel: cancel,
|
||||
conn: conn,
|
||||
url: url,
|
||||
subs: make(map[string]chan []byte),
|
||||
okCh: make(chan []byte, 100),
|
||||
countCh: make(chan []byte, 100),
|
||||
ctx: ctx,
|
||||
cancel: cancel,
|
||||
}
|
||||
go c.readLoop()
|
||||
return
|
||||
@@ -106,11 +110,22 @@ func (c *Client) readLoop() {
|
||||
if subID, ok := raw[1].(string); ok {
|
||||
if ch, exists := c.subs[subID]; exists {
|
||||
close(ch)
|
||||
delete(c.subs, subID)
|
||||
}
|
||||
}
|
||||
}
|
||||
case "OK":
|
||||
// OK messages are handled by WaitForOK
|
||||
// Route OK messages to okCh for WaitForOK
|
||||
select {
|
||||
case c.okCh <- msg:
|
||||
default:
|
||||
}
|
||||
case "COUNT":
|
||||
// Route COUNT messages to countCh for Count
|
||||
select {
|
||||
case c.countCh <- msg:
|
||||
default:
|
||||
}
|
||||
case "NOTICE":
|
||||
// Notice messages are logged
|
||||
case "CLOSED":
|
||||
@@ -140,7 +155,15 @@ func (c *Client) Subscribe(subID string, filters []interface{}) (ch chan []byte,
|
||||
func (c *Client) Unsubscribe(subID string) error {
|
||||
c.mu.Lock()
|
||||
if ch, exists := c.subs[subID]; exists {
|
||||
close(ch)
|
||||
// Channel might already be closed by EOSE, so use recover to handle gracefully
|
||||
func() {
|
||||
defer func() {
|
||||
if recover() != nil {
|
||||
// Channel was already closed, ignore
|
||||
}
|
||||
}()
|
||||
close(ch)
|
||||
}()
|
||||
delete(c.subs, subID)
|
||||
}
|
||||
c.mu.Unlock()
|
||||
@@ -149,10 +172,7 @@ func (c *Client) Unsubscribe(subID string) error {
|
||||
|
||||
// Publish sends an EVENT message to the relay.
|
||||
func (c *Client) Publish(ev *event.E) (err error) {
|
||||
evJSON, err := json.Marshal(ev.Serialize())
|
||||
if err != nil {
|
||||
return errorf.E("failed to marshal event: %w", err)
|
||||
}
|
||||
evJSON := ev.Serialize()
|
||||
var evMap map[string]interface{}
|
||||
if err = json.Unmarshal(evJSON, &evMap); err != nil {
|
||||
return errorf.E("failed to unmarshal event: %w", err)
|
||||
@@ -169,21 +189,14 @@ func (c *Client) WaitForOK(eventID []byte, timeout time.Duration) (accepted bool
|
||||
select {
|
||||
case <-ctx.Done():
|
||||
return false, "", errorf.E("timeout waiting for OK response")
|
||||
default:
|
||||
}
|
||||
var msg []byte
|
||||
_, msg, err = c.conn.ReadMessage()
|
||||
if err != nil {
|
||||
return false, "", errorf.E("connection closed: %w", err)
|
||||
}
|
||||
var raw []interface{}
|
||||
if err = json.Unmarshal(msg, &raw); err != nil {
|
||||
continue
|
||||
}
|
||||
if len(raw) < 3 {
|
||||
continue
|
||||
}
|
||||
if typ, ok := raw[0].(string); ok && typ == "OK" {
|
||||
case msg := <-c.okCh:
|
||||
var raw []interface{}
|
||||
if err = json.Unmarshal(msg, &raw); err != nil {
|
||||
continue
|
||||
}
|
||||
if len(raw) < 3 {
|
||||
continue
|
||||
}
|
||||
if id, ok := raw[1].(string); ok && id == idStr {
|
||||
accepted, _ = raw[2].(bool)
|
||||
if len(raw) > 3 {
|
||||
@@ -208,23 +221,16 @@ func (c *Client) Count(filters []interface{}) (count int64, err error) {
|
||||
select {
|
||||
case <-ctx.Done():
|
||||
return 0, errorf.E("timeout waiting for COUNT response")
|
||||
default:
|
||||
}
|
||||
_, msg, err := c.conn.ReadMessage()
|
||||
if err != nil {
|
||||
return 0, errorf.E("connection closed: %w", err)
|
||||
}
|
||||
var raw []interface{}
|
||||
if err = json.Unmarshal(msg, &raw); err != nil {
|
||||
continue
|
||||
}
|
||||
if len(raw) >= 3 {
|
||||
if typ, ok := raw[0].(string); ok && typ == "COUNT" {
|
||||
case msg := <-c.countCh:
|
||||
var raw []interface{}
|
||||
if err = json.Unmarshal(msg, &raw); err != nil {
|
||||
continue
|
||||
}
|
||||
if len(raw) >= 3 {
|
||||
if subID, ok := raw[1].(string); ok && subID == "count-sub" {
|
||||
if countObj, ok := raw[2].(map[string]interface{}); ok {
|
||||
if c, ok := countObj["count"].(float64); ok {
|
||||
return int64(c), nil
|
||||
}
|
||||
// COUNT response format: ["COUNT", "subscription-id", count, approximate?]
|
||||
if cnt, ok := raw[2].(float64); ok {
|
||||
return int64(cnt), nil
|
||||
}
|
||||
}
|
||||
}
|
||||
@@ -234,12 +240,9 @@ func (c *Client) Count(filters []interface{}) (count int64, err error) {
|
||||
|
||||
// Auth sends an AUTH message with the signed event.
|
||||
func (c *Client) Auth(ev *event.E) error {
|
||||
evJSON, err := json.Marshal(ev.Serialize())
|
||||
if err != nil {
|
||||
return errorf.E("failed to marshal event: %w", err)
|
||||
}
|
||||
evJSON := ev.Serialize()
|
||||
var evMap map[string]interface{}
|
||||
if err = json.Unmarshal(evJSON, &evMap); err != nil {
|
||||
if err := json.Unmarshal(evJSON, &evMap); err != nil {
|
||||
return errorf.E("failed to unmarshal event: %w", err)
|
||||
}
|
||||
return c.Send([]interface{}{"AUTH", evMap})
|
||||
|
||||
@@ -91,7 +91,8 @@ func CreateEphemeralEvent(signer *p256k.Signer, kindNum uint16, content string)
|
||||
func CreateDeleteEvent(signer *p256k.Signer, eventIDs [][]byte, reason string) (ev *event.E, err error) {
|
||||
tags := tag.NewS()
|
||||
for _, id := range eventIDs {
|
||||
tags.Append(tag.NewFromBytesSlice([]byte("e"), id))
|
||||
// e tags must contain hex-encoded event IDs
|
||||
tags.Append(tag.NewFromBytesSlice([]byte("e"), []byte(hex.Enc(id))))
|
||||
}
|
||||
if reason != "" {
|
||||
tags.Append(tag.NewFromBytesSlice([]byte("content"), []byte(reason)))
|
||||
|
||||
@@ -260,7 +260,7 @@ func testRejectFutureEvent(client *Client, key1, key2 *KeyPair) (result TestResu
|
||||
if err != nil {
|
||||
return TestResult{Pass: false, Info: fmt.Sprintf("failed to create event: %v", err)}
|
||||
}
|
||||
ev.CreatedAt = time.Now().Unix() + 3600 // 1 hour in the future
|
||||
ev.CreatedAt = time.Now().Unix() + 3601 // More than 1 hour in the future (should be rejected)
|
||||
// Re-sign with new timestamp
|
||||
if err = ev.Sign(key1.Secret); err != nil {
|
||||
return TestResult{Pass: false, Info: fmt.Sprintf("failed to re-sign: %v", err)}
|
||||
@@ -327,12 +327,14 @@ func testReplaceableEvents(client *Client, key1, key2 *KeyPair) (result TestResu
|
||||
if err != nil || !accepted {
|
||||
return TestResult{Pass: false, Info: "second event not accepted"}
|
||||
}
|
||||
time.Sleep(200 * time.Millisecond)
|
||||
// Wait longer for replacement to complete
|
||||
time.Sleep(500 * time.Millisecond)
|
||||
filter := map[string]interface{}{
|
||||
"kinds": []int{int(kind.ProfileMetadata.K)},
|
||||
"authors": []string{hex.Enc(key1.Pubkey)},
|
||||
"limit": 2, // Set limit > 1 to get multiple versions of replaceable events
|
||||
}
|
||||
events, err := client.GetEvents("test-replaceable", []interface{}{filter}, 2*time.Second)
|
||||
events, err := client.GetEvents("test-replaceable", []interface{}{filter}, 3*time.Second)
|
||||
if err != nil {
|
||||
return TestResult{Pass: false, Info: fmt.Sprintf("failed to get events: %v", err)}
|
||||
}
|
||||
@@ -419,7 +421,8 @@ func testDeletionEvents(client *Client, key1, key2 *KeyPair) (result TestResult)
|
||||
if err != nil || !accepted {
|
||||
return TestResult{Pass: false, Info: "target event not accepted"}
|
||||
}
|
||||
time.Sleep(200 * time.Millisecond)
|
||||
// Wait longer for event to be indexed
|
||||
time.Sleep(500 * time.Millisecond)
|
||||
// Now create deletion event
|
||||
deleteEv, err := CreateDeleteEvent(key1.Secret, [][]byte{targetEv.ID}, "deletion reason")
|
||||
if err != nil {
|
||||
|
||||
Reference in New Issue
Block a user