// Package gh is a small read-only GitHub REST client authenticated as a GitHub App. package gh import ( "context" "encoding/json" "errors" "fmt" "io" "net/http" "net/url" "strconv" "strings" "sync" "sync/atomic" "time" "github.com/lukaszraczylo/ghtracker/internal/model" ) const ( acceptHeader = "application/vnd.github+json" apiVersion = "2022-11-28" userAgent = "ghtracker" perPage = 100 maxBody = 16 << 20 maxRetries = 2 // rateReserve is the request budget left untouched so the app never exhausts its hourly quota. rateReserve = 100 // secondaryBackoff applies when GitHub signals abuse limits without a Retry-After. secondaryBackoff = time.Minute ) // APIError is a non-2xx GitHub response. type APIError struct { Message string Status int } func (e *APIError) Error() string { return fmt.Sprintf("github: HTTP %d: %s", e.Status, e.Message) } func apiMessage(body []byte) string { var m struct { Message string `json:"message"` } if json.Unmarshal(body, &m) == nil && m.Message != "" { return m.Message } s := strings.TrimSpace(string(body)) if len(s) > 200 { s = s[:200] } return s } func isNotFound(err error) bool { var ae *APIError return errors.As(err, &ae) && ae.Status == http.StatusNotFound } // TokenSource yields a bearer token for requests against repo. type TokenSource interface { Token(ctx context.Context, repo string) (string, error) } type Client struct { // freshHC opens a new connection for every request. freshHC *http.Client rateReset time.Time blockedUntil time.Time auth TokenSource hc *http.Client // sleep is replaced in tests. sleep func(context.Context, time.Duration) error now func() time.Time cache map[string]cachedResponse base string rateRemaining atomic.Int64 maxItems int mu sync.Mutex } // cachedResponse backs conditional requests: a 304 reply does not count against the primary rate limit. type cachedResponse struct { etag string link string body []byte } func NewClient(baseURL string, auth TokenSource, hc *http.Client, maxItems int) *Client { c := &Client{base: strings.TrimRight(baseURL, "/"), hc: hc, freshHC: freshClient(hc), auth: auth, maxItems: maxItems, sleep: sleepCtx, now: time.Now, cache: map[string]cachedResponse{}} c.rateRemaining.Store(-1) return c } // RateLimitRemaining returns the last reported remaining request budget. func (c *Client) RateLimitRemaining() (int, bool) { v := c.rateRemaining.Load() return int(v), v >= 0 } func sleepCtx(ctx context.Context, d time.Duration) error { t := time.NewTimer(d) defer t.Stop() select { case <-ctx.Done(): return ctx.Err() case <-t.C: return nil } } // do performs one authenticated GET, retrying transient failures, and returns the body and Link header. // do performs one authenticated GET, retrying transient failures. fresh sends it on a new connection // and skips the response cache, so a stale GitHub replica behind a reused connection cannot answer it. func (c *Client) do(ctx context.Context, repo, target string, fresh bool) ([]byte, string, error) { var lastErr error for attempt := 0; attempt <= maxRetries; attempt++ { if attempt > 0 { if err := c.sleep(ctx, time.Duration(attempt)*time.Second); err != nil { return nil, "", err } } body, link, retry, err := c.once(ctx, repo, target, fresh) if err == nil { return body, link, nil } lastErr = err if !retry { break } } return nil, "", lastErr } func (c *Client) once(ctx context.Context, repo, target string, fresh bool) (body []byte, link string, retry bool, err error) { if gerr := c.guard(); gerr != nil { return nil, "", false, gerr } tok, err := c.auth.Token(ctx, repo) if err != nil { return nil, "", false, err } req, err := http.NewRequestWithContext(ctx, http.MethodGet, target, nil) if err != nil { return nil, "", false, err } req.Header.Set("Authorization", "Bearer "+tok) req.Header.Set("Accept", acceptHeader) req.Header.Set("X-GitHub-Api-Version", apiVersion) req.Header.Set("User-Agent", userAgent) c.mu.Lock() cached, haveCache := c.cache[target] c.mu.Unlock() haveCache = haveCache && !fresh if haveCache { req.Header.Set("If-None-Match", cached.etag) } hc := c.hc if fresh { hc = c.freshHC } resp, err := hc.Do(req) if err != nil { return nil, "", ctx.Err() == nil, err } defer func() { _ = resp.Body.Close() }() c.noteRate(resp.Header) if resp.StatusCode == http.StatusNotModified && haveCache { return cached.body, cached.link, false, nil } data, err := io.ReadAll(io.LimitReader(resp.Body, maxBody)) if err != nil { return nil, "", true, err } if resp.StatusCode/100 != 2 { ae := &APIError{Status: resp.StatusCode, Message: apiMessage(data)} if c.noteLimited(resp, ae) { return nil, "", false, fmt.Errorf("%w: %v", model.ErrRateLimited, ae) } return nil, "", resp.StatusCode >= 500, ae } link = resp.Header.Get("Link") if etag := resp.Header.Get("ETag"); etag != "" && !fresh { c.mu.Lock() c.cache[target] = cachedResponse{etag: etag, body: data, link: link} c.mu.Unlock() } return data, link, false, nil } // guard refuses requests while a limit block is active or the remaining budget is at the reserve. func (c *Client) guard() error { c.mu.Lock() defer c.mu.Unlock() now := c.now() if now.Before(c.blockedUntil) { return fmt.Errorf("%w: blocked until %s", model.ErrRateLimited, c.blockedUntil.Format(time.RFC3339)) } if rem := c.rateRemaining.Load(); rem >= 0 && rem < rateReserve && now.Before(c.rateReset) { return fmt.Errorf("%w: %d requests left, resets %s", model.ErrRateLimited, rem, c.rateReset.Format(time.RFC3339)) } return nil } func (c *Client) noteRate(h http.Header) { if v, err := strconv.ParseInt(h.Get("X-RateLimit-Remaining"), 10, 64); err == nil { c.rateRemaining.Store(v) } if v, err := strconv.ParseInt(h.Get("X-RateLimit-Reset"), 10, 64); err == nil { c.mu.Lock() c.rateReset = time.Unix(v, 0) c.mu.Unlock() } } // noteLimited records a block when the response is a primary or secondary rate-limit rejection. func (c *Client) noteLimited(resp *http.Response, ae *APIError) bool { if resp.StatusCode != http.StatusForbidden && resp.StatusCode != http.StatusTooManyRequests { return false } now := c.now() var until time.Time switch secs, err := strconv.Atoi(resp.Header.Get("Retry-After")); { case err == nil: until = now.Add(time.Duration(secs) * time.Second) case resp.Header.Get("X-RateLimit-Remaining") == "0": c.mu.Lock() until = c.rateReset c.mu.Unlock() case strings.Contains(strings.ToLower(ae.Message), "rate limit"): until = now.Add(secondaryBackoff) default: return false } c.mu.Lock() c.blockedUntil = until c.mu.Unlock() return true } func (c *Client) url(path string, q url.Values) string { u := c.base + path if len(q) > 0 { u += "?" + q.Encode() } return u } func (c *Client) getJSON(ctx context.Context, repo, path string, q url.Values, out any) error { return c.getJSONMode(ctx, repo, path, q, out, false) } // getJSONFresh is getJSON on a new connection without the response cache. func (c *Client) getJSONFresh(ctx context.Context, repo, path string, q url.Values, out any) error { return c.getJSONMode(ctx, repo, path, q, out, true) } func (c *Client) getJSONMode(ctx context.Context, repo, path string, q url.Values, out any, fresh bool) error { body, _, err := c.do(ctx, repo, c.url(path, q), fresh) if err != nil { return err } return json.Unmarshal(body, out) } // listAll follows Link pagination, collecting at most limit items. func listAll[T any](ctx context.Context, c *Client, repo, path string, q url.Values, limit int, unwrap func([]byte) ([]T, error)) ([]T, error) { if q == nil { q = url.Values{} } q.Set("per_page", strconv.Itoa(perPage)) next := c.url(path, q) var all []T for next != "" && len(all) < limit { body, link, err := c.do(ctx, repo, next, false) if err != nil { return all, err } page, err := unwrap(body) if err != nil { return all, err } all = append(all, page...) next = nextLink(link) } if len(all) > limit { all = all[:limit] } return all, nil } func unwrapArray[T any](b []byte) ([]T, error) { var out []T return out, json.Unmarshal(b, &out) } // nextLink extracts rel="next" from a Link header. func nextLink(h string) string { for part := range strings.SplitSeq(h, ",") { target, params, found := strings.Cut(part, ";") if found && strings.TrimSpace(params) == `rel="next"` { return strings.Trim(strings.TrimSpace(target), "<>") } } return "" } // freshClient copies hc but turns off connection reuse. Custom transports that are not an // *http.Transport are used as they are. func freshClient(hc *http.Client) *http.Client { tr, ok := hc.Transport.(*http.Transport) if !ok { if hc.Transport != nil { return hc } tr, _ = http.DefaultTransport.(*http.Transport) } clone := tr.Clone() clone.DisableKeepAlives = true return &http.Client{Transport: clone, Timeout: hc.Timeout} }