package twitter import ( "context" "encoding/json" "fmt" "io" "net/http" "net/url" "strconv" "strings" "time" log "github.com/nacorid/logger" ) func parseTwitterTime(s string) time.Time { t, err := time.Parse(time.RFC3339, s) if err != nil { log.Warn("Failed to parse Twitter timestamp", "timestamp", s, "error", err) return time.Time{} } return t } const apiBase = "https://api.x.com/2" // httpTimeout bounds a single HTTP request so a stalled X backend cannot block // a request indefinitely. const httpTimeout = 60 * time.Second const userAgent = "naco-api/1.0" type Client struct { httpClient *http.Client bearerToken string } func NewTwitterClient(bearerToken string) *Client { return &Client{ httpClient: &http.Client{Timeout: httpTimeout}, bearerToken: bearerToken, } } type Post struct { ID string Author string Content string CreatedAt time.Time Likes int Reposts int Quotes int Replies int } type userResponse struct { Data struct { ID string `json:"id"` Name string `json:"name"` Username string `json:"username"` } `json:"data"` Errors []struct { Title string `json:"title"` Detail string `json:"detail"` } `json:"errors"` } type tweetMetrics struct { RetweetCount int `json:"retweet_count"` ReplyCount int `json:"reply_count"` LikeCount int `json:"like_count"` QuoteCount int `json:"quote_count"` } type tweetData struct { ID string `json:"id"` Text string `json:"text"` CreatedAt string `json:"created_at"` PublicMetrics tweetMetrics `json:"public_metrics"` } type tweetsResponse struct { Data []tweetData `json:"data"` Includes struct { Users []struct { Username string `json:"username"` } `json:"users"` } `json:"includes"` Errors []struct { Title string `json:"title"` Detail string `json:"detail"` } `json:"errors"` } type singleTweetResponse struct { Data struct { ID string `json:"id"` Text string `json:"text"` CreatedAt string `json:"created_at"` PublicMetrics tweetMetrics `json:"public_metrics"` } `json:"data"` Includes struct { Users []struct { Username string `json:"username"` } `json:"users"` } `json:"includes"` Errors []struct { Title string `json:"title"` Detail string `json:"detail"` } `json:"errors"` } const ( maxRetries = 4 maxRateLimitWait = 30 * time.Second // Never stall a request longer than 30s ) var defaultBackoff = 500 * time.Millisecond func (c *Client) doGet(ctx context.Context, endpoint string) ([]byte, error) { backoff := defaultBackoff for attempt := 0; ; attempt++ { req, err := http.NewRequestWithContext(ctx, http.MethodGet, endpoint, nil) if err != nil { return nil, fmt.Errorf("failed to build request: %w", err) } req.Header.Set("Authorization", "Bearer "+c.bearerToken) req.Header.Set("User-Agent", userAgent) resp, err := c.httpClient.Do(req) if err != nil { // Transport-level failure (connection refused/reset, timeout). if attempt >= maxRetries { return nil, fmt.Errorf("twitter request failed after %d attempts: %w", maxRetries+1, err) } log.WarnWithContext(ctx, "twitter transport error, waiting to retry", "attempt", attempt+1, "error", err) if werr := waitForRetry(ctx, backoff); werr != nil { return nil, werr } backoff *= 2 continue } // Handle 429 Too Many Requests if resp.StatusCode == http.StatusTooManyRequests { // Drain & close body before retrying. _, _ = io.Copy(io.Discard, resp.Body) _ = resp.Body.Close() if attempt >= maxRetries { return nil, fmt.Errorf("twitter rate limit exceeded, max retries reached") } wait := parseRateLimitWait(resp.Header, backoff) // Protect against 15-minute thread freezes if wait > maxRateLimitWait { return nil, fmt.Errorf("twitter rate limited: reset wait (%s) exceeds maximum allowed (%s)", wait, maxRateLimitWait) } log.WarnWithContext(ctx, "twitter rate limited, waiting to retry", "attempt", attempt+1, "wait", wait.String()) if werr := waitForRetry(ctx, wait); werr != nil { return nil, werr } backoff *= 2 continue } body, err := io.ReadAll(io.LimitReader(resp.Body, 2*1024*1024)) _ = resp.Body.Close() if err != nil { return nil, fmt.Errorf("failed to read response: %w", err) } // Retry transient server-side failures (5xx) with backoff. 4xx/3xx fail // fast below. if resp.StatusCode >= http.StatusInternalServerError && attempt < maxRetries { log.WarnWithContext(ctx, "twitter server error, waiting to retry", "attempt", attempt+1, "status", resp.Status) if werr := waitForRetry(ctx, backoff); werr != nil { return nil, werr } backoff *= 2 continue } if resp.StatusCode != http.StatusOK { errMsg := strings.TrimSpace(string(body)) if len(errMsg) > 512 { errMsg = errMsg[:512] + "... (truncated)" } return nil, fmt.Errorf("twitter error (%s): %s", resp.Status, errMsg) } return body, nil } } // waitForRetry sleeps for wait, aborting early if the context is canceled. // A timer is used (not time.After) so the timer can be stopped to free its // resource when the caller abandons the wait. func waitForRetry(ctx context.Context, wait time.Duration) error { timer := time.NewTimer(wait) defer timer.Stop() select { case <-timer.C: return nil case <-ctx.Done(): return ctx.Err() } } // parseRateLimitWait extracts x-rate-limit-* or Retry-After, falling back to backoff. func parseRateLimitWait(header http.Header, fallback time.Duration) time.Duration { // 1. Check x-rate-limit-reset (Unix timestamp) if remStr := header.Get("x-rate-limit-remaining"); remStr != "" { if rem, err := strconv.Atoi(remStr); err == nil && rem <= 0 { if rstStr := header.Get("x-rate-limit-reset"); rstStr != "" { if rst, err := strconv.ParseInt(rstStr, 10, 64); err == nil { // Add 1s clock skew buffer wait := time.Until(time.Unix(rst, 0)) + (1 * time.Second) if wait > 0 { return wait } } } } } // 2. Check standard Retry-After (seconds) if ra := header.Get("Retry-After"); ra != "" { if secs, err := strconv.Atoi(ra); err == nil && secs > 0 { return time.Duration(secs)*time.Second + (1 * time.Second) } } return fallback } func (c *Client) resolveUser(ctx context.Context, username string) (id, handle string, err error) { endpoint := fmt.Sprintf("%s/users/by/username/%s", apiBase, url.PathEscape(username)) body, err := c.doGet(ctx, endpoint) if err != nil { return "", "", err } var ur userResponse if err := json.Unmarshal(body, &ur); err != nil { return "", "", fmt.Errorf("failed to unmarshal user response: %w", err) } if len(ur.Errors) > 0 { return "", "", fmt.Errorf("twitter error: %s: %s", ur.Errors[0].Title, ur.Errors[0].Detail) } if ur.Data.ID == "" { return "", "", fmt.Errorf("twitter user %q not found", username) } return ur.Data.ID, ur.Data.Username, nil } func (c *Client) GetLatestNPosts(ctx context.Context, actor string, limit int) ([]*Post, error) { if c.bearerToken == "" { return nil, fmt.Errorf("twitter bearer token not configured") } username := strings.TrimPrefix(actor, "@") userID, handle, err := c.resolveUser(ctx, username) if err != nil { return nil, err } if limit > 100 { limit = 100 } endpoint := fmt.Sprintf("%s/users/%s/tweets?max_results=%d&tweet.fields=created_at,public_metrics&exclude=retweets,replies", apiBase, url.PathEscape(userID), limit) return c.fetchTweets(ctx, handle, endpoint) } func (c *Client) GetPostsInTimeframe(ctx context.Context, actor string, start, end time.Time, limit int) ([]*Post, error) { if c.bearerToken == "" { return nil, fmt.Errorf("twitter bearer token not configured") } username := strings.TrimPrefix(actor, "@") userID, handle, err := c.resolveUser(ctx, username) if err != nil { return nil, err } if limit > 100 { limit = 100 } endpoint := fmt.Sprintf("%s/users/%s/tweets?start_time=%s&end_time=%s&max_results=%d&tweet.fields=created_at,public_metrics&exclude=retweets,replies", apiBase, url.PathEscape(userID), start.Format(time.RFC3339), end.Format(time.RFC3339), limit) return c.fetchTweets(ctx, handle, endpoint) } // fetchTweets performs the request and converts the tweet payload into Posts, // skipping entries without an ID so downstream dedupe keys are never empty. func (c *Client) fetchTweets(ctx context.Context, username, endpoint string) ([]*Post, error) { body, err := c.doGet(ctx, endpoint) if err != nil { return nil, err } var tr tweetsResponse if err := json.Unmarshal(body, &tr); err != nil { return nil, fmt.Errorf("failed to unmarshal tweets response: %w", err) } if len(tr.Errors) > 0 { return nil, fmt.Errorf("twitter error: %s: %s", tr.Errors[0].Title, tr.Errors[0].Detail) } posts := make([]*Post, 0, len(tr.Data)) for _, t := range tr.Data { if t.ID == "" { continue } posts = append(posts, &Post{ ID: t.ID, Author: username, Content: t.Text, CreatedAt: parseTwitterTime(t.CreatedAt), Likes: t.PublicMetrics.LikeCount, Reposts: t.PublicMetrics.RetweetCount, Quotes: t.PublicMetrics.QuoteCount, Replies: t.PublicMetrics.ReplyCount, }) } return posts, nil } func (c *Client) GetPost(ctx context.Context, id string) (*Post, error) { if c.bearerToken == "" { return nil, fmt.Errorf("twitter bearer token not configured") } endpoint := fmt.Sprintf("%s/tweets/%s?tweet.fields=created_at,public_metrics&expansions=author_id&user.fields=username", apiBase, url.PathEscape(id)) body, err := c.doGet(ctx, endpoint) if err != nil { return nil, err } var tr singleTweetResponse if err := json.Unmarshal(body, &tr); err != nil { return nil, fmt.Errorf("failed to unmarshal tweet response: %w", err) } if len(tr.Errors) > 0 { return nil, fmt.Errorf("twitter error: %s: %s", tr.Errors[0].Title, tr.Errors[0].Detail) } if tr.Data.ID == "" { return nil, fmt.Errorf("tweet %q not found", id) } author := "unknown" if len(tr.Includes.Users) > 0 { author = tr.Includes.Users[0].Username } return &Post{ ID: tr.Data.ID, Author: author, Content: tr.Data.Text, CreatedAt: parseTwitterTime(tr.Data.CreatedAt), Likes: tr.Data.PublicMetrics.LikeCount, Reposts: tr.Data.PublicMetrics.RetweetCount, Quotes: tr.Data.PublicMetrics.QuoteCount, Replies: tr.Data.PublicMetrics.ReplyCount, }, nil }