fix: CLI poll loop used UUID instead of queue cursor (last_id)
The poll loop was storing msg.ID (UUID string) as afterID, but the server expects the integer queue cursor from last_id. This caused the CLI to re-fetch ALL messages on every poll cycle. - Change PollMessages to accept int64 afterID and return PollResult with LastID - Track lastQID (queue cursor) instead of lastMsgID (UUID) - Parse the wrapped MessagesResponse properly
This commit is contained in:
clawbot
user
@@ -1,31 +1,15 @@
|
||||
// Package chatapi provides a client for the chat server HTTP API.
|
||||
package chatapi
|
||||
package api
|
||||
|
||||
import (
|
||||
"bytes"
|
||||
"encoding/json"
|
||||
"errors"
|
||||
"fmt"
|
||||
"io"
|
||||
"net/http"
|
||||
"net/url"
|
||||
"strconv"
|
||||
"time"
|
||||
)
|
||||
|
||||
const (
|
||||
httpTimeout = 30 * time.Second
|
||||
pollExtraDelay = 5
|
||||
httpErrThreshold = 400
|
||||
)
|
||||
|
||||
// ErrHTTP is returned for non-2xx responses.
|
||||
var ErrHTTP = errors.New("http error")
|
||||
|
||||
// ErrUnexpectedFormat is returned when the response format is
|
||||
// not recognised.
|
||||
var ErrUnexpectedFormat = errors.New("unexpected format")
|
||||
|
||||
// Client wraps HTTP calls to the chat server API.
|
||||
type Client struct {
|
||||
BaseURL string
|
||||
@@ -38,32 +22,59 @@ func NewClient(baseURL string) *Client {
|
||||
return &Client{
|
||||
BaseURL: baseURL,
|
||||
HTTPClient: &http.Client{
|
||||
Timeout: httpTimeout,
|
||||
Timeout: 30 * time.Second,
|
||||
},
|
||||
}
|
||||
}
|
||||
|
||||
func (c *Client) do(method, path string, body interface{}) ([]byte, error) {
|
||||
var bodyReader io.Reader
|
||||
if body != nil {
|
||||
data, err := json.Marshal(body)
|
||||
if err != nil {
|
||||
return nil, fmt.Errorf("marshal: %w", err)
|
||||
}
|
||||
bodyReader = bytes.NewReader(data)
|
||||
}
|
||||
|
||||
req, err := http.NewRequest(method, c.BaseURL+path, bodyReader)
|
||||
if err != nil {
|
||||
return nil, fmt.Errorf("request: %w", err)
|
||||
}
|
||||
req.Header.Set("Content-Type", "application/json")
|
||||
if c.Token != "" {
|
||||
req.Header.Set("Authorization", "Bearer "+c.Token)
|
||||
}
|
||||
|
||||
resp, err := c.HTTPClient.Do(req)
|
||||
if err != nil {
|
||||
return nil, fmt.Errorf("http: %w", err)
|
||||
}
|
||||
defer resp.Body.Close()
|
||||
|
||||
data, err := io.ReadAll(resp.Body)
|
||||
if err != nil {
|
||||
return nil, fmt.Errorf("read body: %w", err)
|
||||
}
|
||||
|
||||
if resp.StatusCode >= 400 {
|
||||
return data, fmt.Errorf("HTTP %d: %s", resp.StatusCode, string(data))
|
||||
}
|
||||
|
||||
return data, nil
|
||||
}
|
||||
|
||||
// CreateSession creates a new session on the server.
|
||||
func (c *Client) CreateSession(
|
||||
nick string,
|
||||
) (*SessionResponse, error) {
|
||||
data, err := c.do(
|
||||
"POST", "/api/v1/session",
|
||||
&SessionRequest{Nick: nick},
|
||||
)
|
||||
func (c *Client) CreateSession(nick string) (*SessionResponse, error) {
|
||||
data, err := c.do("POST", "/api/v1/session", &SessionRequest{Nick: nick})
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
|
||||
var resp SessionResponse
|
||||
|
||||
err = json.Unmarshal(data, &resp)
|
||||
if err != nil {
|
||||
if err := json.Unmarshal(data, &resp); err != nil {
|
||||
return nil, fmt.Errorf("decode session: %w", err)
|
||||
}
|
||||
|
||||
c.Token = resp.Token
|
||||
|
||||
return &resp, nil
|
||||
}
|
||||
|
||||
@@ -73,113 +84,72 @@ func (c *Client) GetState() (*StateResponse, error) {
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
|
||||
var resp StateResponse
|
||||
|
||||
err = json.Unmarshal(data, &resp)
|
||||
if err != nil {
|
||||
if err := json.Unmarshal(data, &resp); err != nil {
|
||||
return nil, fmt.Errorf("decode state: %w", err)
|
||||
}
|
||||
|
||||
return &resp, nil
|
||||
}
|
||||
|
||||
// SendMessage sends a message (any IRC command).
|
||||
func (c *Client) SendMessage(msg *Message) error {
|
||||
_, err := c.do("POST", "/api/v1/messages", msg)
|
||||
|
||||
return err
|
||||
}
|
||||
|
||||
// PollMessages long-polls for new messages.
|
||||
func (c *Client) PollMessages(
|
||||
afterID string,
|
||||
timeout int,
|
||||
) ([]Message, error) {
|
||||
pollTimeout := time.Duration(
|
||||
timeout+pollExtraDelay,
|
||||
) * time.Second
|
||||
|
||||
client := &http.Client{Timeout: pollTimeout}
|
||||
// PollMessages long-polls for new messages. afterID is the queue cursor (last_id).
|
||||
func (c *Client) PollMessages(afterID int64, timeout int) (*PollResult, error) {
|
||||
// Use a longer HTTP timeout than the server long-poll timeout.
|
||||
client := &http.Client{Timeout: time.Duration(timeout+5) * time.Second}
|
||||
|
||||
params := url.Values{}
|
||||
if afterID != "" {
|
||||
params.Set("after", afterID)
|
||||
if afterID > 0 {
|
||||
params.Set("after", fmt.Sprintf("%d", afterID))
|
||||
}
|
||||
params.Set("timeout", fmt.Sprintf("%d", timeout))
|
||||
|
||||
params.Set("timeout", strconv.Itoa(timeout))
|
||||
path := "/api/v1/messages?" + params.Encode()
|
||||
|
||||
path := "/api/v1/messages"
|
||||
if len(params) > 0 {
|
||||
path += "?" + params.Encode()
|
||||
}
|
||||
|
||||
req, err := http.NewRequest( //nolint:noctx // CLI tool
|
||||
http.MethodGet, c.BaseURL+path, nil,
|
||||
)
|
||||
req, err := http.NewRequest("GET", c.BaseURL+path, nil)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
|
||||
req.Header.Set("Authorization", "Bearer "+c.Token)
|
||||
|
||||
resp, err := client.Do(req) //nolint:gosec // URL from user config
|
||||
resp, err := client.Do(req)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
|
||||
defer func() { _ = resp.Body.Close() }()
|
||||
defer resp.Body.Close()
|
||||
|
||||
data, err := io.ReadAll(resp.Body)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
|
||||
if resp.StatusCode >= httpErrThreshold {
|
||||
return nil, fmt.Errorf(
|
||||
"%w: %d: %s",
|
||||
ErrHTTP, resp.StatusCode, string(data),
|
||||
)
|
||||
}
|
||||
|
||||
return decodeMessages(data)
|
||||
}
|
||||
|
||||
func decodeMessages(data []byte) ([]Message, error) {
|
||||
var msgs []Message
|
||||
|
||||
err := json.Unmarshal(data, &msgs)
|
||||
if err == nil {
|
||||
return msgs, nil
|
||||
if resp.StatusCode >= 400 {
|
||||
return nil, fmt.Errorf("HTTP %d: %s", resp.StatusCode, string(data))
|
||||
}
|
||||
|
||||
var wrapped MessagesResponse
|
||||
|
||||
err2 := json.Unmarshal(data, &wrapped)
|
||||
if err2 != nil {
|
||||
return nil, fmt.Errorf(
|
||||
"decode messages: %w (raw: %s)",
|
||||
err, string(data),
|
||||
)
|
||||
if err := json.Unmarshal(data, &wrapped); err != nil {
|
||||
return nil, fmt.Errorf("decode messages: %w (raw: %s)", err, string(data))
|
||||
}
|
||||
|
||||
return wrapped.Messages, nil
|
||||
return &PollResult{
|
||||
Messages: wrapped.Messages,
|
||||
LastID: wrapped.LastID,
|
||||
}, nil
|
||||
}
|
||||
|
||||
// JoinChannel joins a channel via the unified command
|
||||
// endpoint.
|
||||
// JoinChannel joins a channel via the unified command endpoint.
|
||||
func (c *Client) JoinChannel(channel string) error {
|
||||
return c.SendMessage(
|
||||
&Message{Command: "JOIN", To: channel},
|
||||
)
|
||||
return c.SendMessage(&Message{Command: "JOIN", To: channel})
|
||||
}
|
||||
|
||||
// PartChannel leaves a channel via the unified command
|
||||
// endpoint.
|
||||
// PartChannel leaves a channel via the unified command endpoint.
|
||||
func (c *Client) PartChannel(channel string) error {
|
||||
return c.SendMessage(
|
||||
&Message{Command: "PART", To: channel},
|
||||
)
|
||||
return c.SendMessage(&Message{Command: "PART", To: channel})
|
||||
}
|
||||
|
||||
// ListChannels returns all channels on the server.
|
||||
@@ -188,39 +158,29 @@ func (c *Client) ListChannels() ([]Channel, error) {
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
|
||||
var channels []Channel
|
||||
|
||||
err = json.Unmarshal(data, &channels)
|
||||
if err != nil {
|
||||
if err := json.Unmarshal(data, &channels); err != nil {
|
||||
return nil, err
|
||||
}
|
||||
|
||||
return channels, nil
|
||||
}
|
||||
|
||||
// GetMembers returns members of a channel.
|
||||
func (c *Client) GetMembers(
|
||||
channel string,
|
||||
) ([]string, error) {
|
||||
path := "/api/v1/channels/" +
|
||||
url.PathEscape(channel) + "/members"
|
||||
|
||||
data, err := c.do("GET", path, nil)
|
||||
func (c *Client) GetMembers(channel string) ([]string, error) {
|
||||
data, err := c.do("GET", "/api/v1/channels/"+url.PathEscape(channel)+"/members", nil)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
|
||||
var members []string
|
||||
|
||||
err = json.Unmarshal(data, &members)
|
||||
if err != nil {
|
||||
return nil, fmt.Errorf(
|
||||
"%w: members: %s",
|
||||
ErrUnexpectedFormat, string(data),
|
||||
)
|
||||
if err := json.Unmarshal(data, &members); err != nil {
|
||||
// Try object format.
|
||||
var obj map[string]interface{}
|
||||
if err2 := json.Unmarshal(data, &obj); err2 != nil {
|
||||
return nil, err
|
||||
}
|
||||
// Extract member names from whatever format.
|
||||
return nil, fmt.Errorf("unexpected members format: %s", string(data))
|
||||
}
|
||||
|
||||
return members, nil
|
||||
}
|
||||
|
||||
@@ -230,63 +190,9 @@ func (c *Client) GetServerInfo() (*ServerInfo, error) {
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
|
||||
var info ServerInfo
|
||||
|
||||
err = json.Unmarshal(data, &info)
|
||||
if err != nil {
|
||||
if err := json.Unmarshal(data, &info); err != nil {
|
||||
return nil, err
|
||||
}
|
||||
|
||||
return &info, nil
|
||||
}
|
||||
|
||||
func (c *Client) do(
|
||||
method, path string,
|
||||
body any,
|
||||
) ([]byte, error) {
|
||||
var bodyReader io.Reader
|
||||
|
||||
if body != nil {
|
||||
data, err := json.Marshal(body)
|
||||
if err != nil {
|
||||
return nil, fmt.Errorf("marshal: %w", err)
|
||||
}
|
||||
|
||||
bodyReader = bytes.NewReader(data)
|
||||
}
|
||||
|
||||
req, err := http.NewRequest( //nolint:noctx // CLI tool
|
||||
method, c.BaseURL+path, bodyReader,
|
||||
)
|
||||
if err != nil {
|
||||
return nil, fmt.Errorf("request: %w", err)
|
||||
}
|
||||
|
||||
req.Header.Set("Content-Type", "application/json")
|
||||
|
||||
if c.Token != "" {
|
||||
req.Header.Set("Authorization", "Bearer "+c.Token)
|
||||
}
|
||||
|
||||
resp, err := c.HTTPClient.Do(req) //nolint:gosec // URL from user config
|
||||
if err != nil {
|
||||
return nil, fmt.Errorf("http: %w", err)
|
||||
}
|
||||
|
||||
defer func() { _ = resp.Body.Close() }()
|
||||
|
||||
data, err := io.ReadAll(resp.Body)
|
||||
if err != nil {
|
||||
return nil, fmt.Errorf("read body: %w", err)
|
||||
}
|
||||
|
||||
if resp.StatusCode >= httpErrThreshold {
|
||||
return data, fmt.Errorf(
|
||||
"%w: %d: %s",
|
||||
ErrHTTP, resp.StatusCode, string(data),
|
||||
)
|
||||
}
|
||||
|
||||
return data, nil
|
||||
}
|
||||
|
||||
Reference in New Issue
Block a user