// The Eleven connect broker: the one place OAuth client_id + client_secret
// pairs live (one per integration service), so instances (and their bot
// sidecars) never hold a secret — the same move as the push relay for the APNs
// .p8. See docs/connected-services.md + docs/integrations.md.
//
// It does two jobs, both for CONFIDENTIAL OAuth clients (DNSimple offers no
// public/PKCE client; GitHub's exchange requires the secret even with PKCE):
//
//  1. Redirect relay (GET /oauth/{service}/callback) — each service's single
//     registered redirect_uri for the whole fleet. The provider redirects here
//     with a one-time ?code; the broker bounces it back to the originating
//     instance (read from an allowlisted `state`). It sees only a single-use
//     code — no secret, no token — so a relay compromise leaks nothing usable.
//
//  2. Token exchange (POST /api/exchange) — a bot sidecar authenticates with a
//     per-instance bearer token and sends {service, code, state}; the broker
//     swaps it for an access token using the client_secret it holds, and
//     returns the token. It stores nothing; the sidecar seals the token to the
//     member. This is the one place that holds secrets and transiently sees a
//     user token.
package main

import (
	"crypto/subtle"
	"encoding/base64"
	"encoding/json"
	"flag"
	"fmt"
	"io"
	"log"
	"net"
	"net/http"
	"net/url"
	"os"
	"strconv"
	"strings"
	"sync"
	"time"

	"lchat/integrations"
)

const maxBodyBytes = 8 << 10

// oauthApp is one provider's confidential OAuth client — the secret this
// broker exists to guard, plus the public endpoints of its flow. One per
// service; DNSimple keeps its legacy env names, every other service is
// configured generically (loadOAuthApps).
type oauthApp struct {
	service      string
	clientID     string
	clientSecret string
	redirectURI  string // the broker's own registered callback; sent verbatim at exchange
	authorizeURL string // where the browser sends the user to consent (public)
	tokenURL     string // the provider's code→token endpoint (overridable for tests)
}

type Server struct {
	db        *DB
	adminHash string // sha256 of LCHAT_ADMIN_TOKEN; constant-time compared

	apps map[string]*oauthApp // service id → its confidential client
	http *http.Client

	perInstance *rateLimiter
	global      *rateLimiter
}

func main() {
	addr := flag.String("addr", envOr("ADDR", ":1750"), "listen address (env: ADDR)")
	dbPath := flag.String("db", envOr("LCHAT_BROKER_DB", "broker.db"), "SQLite path (env: LCHAT_BROKER_DB)")
	flag.Parse()

	// Fail loud: a misconfigured broker breaks connecting services fleet-wide.
	// DNSimple stays required (the legacy contract of the deployed unit); other
	// services are additive — configured when their env pair is present.
	admin := os.Getenv("LCHAT_ADMIN_TOKEN")
	apps := loadOAuthApps()
	if admin == "" || apps["dnsimple"] == nil {
		log.Fatal("broker: LCHAT_ADMIN_TOKEN, LCHAT_DNSIMPLE_CLIENT_ID, LCHAT_DNSIMPLE_CLIENT_SECRET, LCHAT_DNSIMPLE_REDIRECT_URI are required")
	}

	db, err := openDB(*dbPath)
	if err != nil {
		log.Fatalf("open db: %v", err)
	}
	s := &Server{
		db:          db,
		adminHash:   hashToken(admin),
		apps:        apps,
		http:        &http.Client{Timeout: 15 * time.Second},
		perInstance: newRateLimiter(60, 10),
		global:      newRateLimiter(600, 100),
	}
	services := make([]string, 0, len(apps))
	for id := range apps {
		services = append(services, id)
	}
	log.Printf("connect broker listening on %s (db: %s, oauth apps: %s)", *addr, *dbPath, strings.Join(services, ", "))
	if err := serve(*addr, s.handler()); err != nil {
		log.Fatal(err)
	}
}

// loadOAuthApps builds the per-service confidential clients from env.
// DNSimple reads its legacy LCHAT_DNSIMPLE_* names. Every other OAuth-capable
// registry service reads LCHAT_OAUTH_<ID>_CLIENT_ID / _CLIENT_SECRET (unset →
// that service simply isn't offered), with the provider endpoints defaulting
// to the manifest's auth.oauth block and the redirect defaulting to this
// broker's own /oauth/<id>/callback (derived from the DNSimple redirect's
// origin, or set explicitly with LCHAT_OAUTH_<ID>_REDIRECT_URI).
func loadOAuthApps() map[string]*oauthApp {
	apps := map[string]*oauthApp{}
	dnsRedirect := os.Getenv("LCHAT_DNSIMPLE_REDIRECT_URI") // e.g. https://connect.elevenmessenger.com/oauth/dnsimple/callback
	if id, secret := os.Getenv("LCHAT_DNSIMPLE_CLIENT_ID"), os.Getenv("LCHAT_DNSIMPLE_CLIENT_SECRET"); id != "" && secret != "" && dnsRedirect != "" {
		apps["dnsimple"] = &oauthApp{
			service:      "dnsimple",
			clientID:     id,
			clientSecret: secret,
			redirectURI:  dnsRedirect,
			authorizeURL: envOr("LCHAT_DNSIMPLE_AUTHORIZE_URL", "https://dnsimple.com/oauth/authorize"),
			tokenURL:     envOr("LCHAT_DNSIMPLE_TOKEN_URL", "https://api.dnsimple.com/v2/oauth/access_token"),
		}
	}
	// The broker's public origin, for deriving other services' redirect URIs.
	selfOrigin := ""
	if u, err := url.Parse(dnsRedirect); err == nil && u.Host != "" {
		selfOrigin = u.Scheme + "://" + u.Host
	}
	for _, m := range integrations.MustRegistry().List() {
		if m.ID == "dnsimple" || m.Auth.OAuth == nil || (m.Auth.Kind != "oauth" && m.Auth.Kind != "both") {
			continue
		}
		prefix := "LCHAT_OAUTH_" + strings.ToUpper(strings.ReplaceAll(m.ID, "-", "_"))
		id, secret := os.Getenv(prefix+"_CLIENT_ID"), os.Getenv(prefix+"_CLIENT_SECRET")
		if id == "" || secret == "" {
			continue
		}
		redirect := os.Getenv(prefix + "_REDIRECT_URI")
		if redirect == "" && selfOrigin != "" {
			redirect = selfOrigin + "/oauth/" + m.ID + "/callback"
		}
		if redirect == "" {
			log.Printf("broker: %s has a client id but no derivable redirect uri — skipped", m.ID)
			continue
		}
		apps[m.ID] = &oauthApp{
			service:      m.ID,
			clientID:     id,
			clientSecret: secret,
			redirectURI:  redirect,
			authorizeURL: envOr(prefix+"_AUTHORIZE_URL", m.Auth.OAuth.AuthorizeURL),
			tokenURL:     envOr(prefix+"_TOKEN_URL", m.Auth.OAuth.TokenURL),
		}
	}
	return apps
}

// serve listens on addr: a "/"-prefixed addr is a unix socket (how the router
// fronts this service); otherwise a TCP address. Mirrors the push relay.
func serve(addr string, h http.Handler) error {
	// ReadHeaderTimeout caps how long a client may dribble request headers
	// (Slowloris): the TCP branch binds straight to public TCP on a container
	// platform. Same 10s the messenger's own listeners use (main.go).
	srv := &http.Server{Handler: h, ReadHeaderTimeout: 10 * time.Second}
	if !strings.HasPrefix(addr, "/") {
		srv.Addr = addr
		return srv.ListenAndServe()
	}
	if c, err := net.DialTimeout("unix", addr, 200*time.Millisecond); err == nil {
		c.Close()
		return fmt.Errorf("socket %s is already in use by another process", addr)
	}
	if err := os.Remove(addr); err != nil && !os.IsNotExist(err) {
		return fmt.Errorf("remove stale socket %s: %w", addr, err)
	}
	ln, err := net.Listen("unix", addr)
	if err != nil {
		return err
	}
	defer ln.Close()
	if err := os.Chmod(addr, 0o600); err != nil {
		return fmt.Errorf("chmod socket %s: %w", addr, err)
	}
	return srv.Serve(ln)
}

func (s *Server) handler() http.Handler {
	mux := http.NewServeMux()
	mux.HandleFunc("GET /healthz", func(w http.ResponseWriter, r *http.Request) { w.Write([]byte("ok")) })
	mux.HandleFunc("GET /", s.root)

	// Public config: the non-secret OAuth params an instance needs to start a
	// flow (client_id, authorize URL, the broker's redirect_uri), per service —
	// so instances need NO provider config of their own; they just ask the
	// broker. ?service defaults to dnsimple (the pre-multi-service clients).
	mux.HandleFunc("GET /config", s.configHandler)

	// The public integrations registry (docs/integrations.md): the reviewed
	// manifests, so every instance's web catalog can grow between deploys.
	mux.HandleFunc("GET /registry", s.registryHandler)

	// Redirect relay: each service's single registered redirect target for the
	// whole fleet.
	mux.HandleFunc("GET /oauth/{service}/callback", s.callback)
	// Token exchange: a sidecar swaps a code for a token. Bearer instance token.
	mux.HandleFunc("POST /api/exchange", s.exchange)
	// Token refresh: a sidecar rotates an expiring pair (GitHub Apps). Same
	// auth, same trust shape — nothing persisted.
	mux.HandleFunc("POST /api/refresh", s.refresh)

	// Admin: mint / list / revoke per-instance tokens (rotation needs no redeploy).
	mux.HandleFunc("POST /api/admin/tokens", s.admin(s.mintHandler))
	mux.HandleFunc("GET /api/admin/tokens", s.admin(s.listHandler))
	mux.HandleFunc("DELETE /api/admin/tokens/{id}", s.admin(s.revokeHandler))

	return securityHeaders(mux)
}

func (s *Server) root(w http.ResponseWriter, r *http.Request) {
	if r.URL.Path != "/" {
		http.NotFound(w, r)
		return
	}
	w.Header().Set("Content-Type", "text/plain; charset=utf-8")
	w.Write([]byte("Eleven connect broker. Holds the OAuth app secret; relays codes; exchanges tokens. See docs/connected-services.md.\n"))
}

// configHandler serves one service's public OAuth params (no secret) so an
// instance can start a flow without holding any provider config of its own. A
// service the broker has no app for is a 404 — the instance renders that as
// "OAuth not offered" and falls back to paste-a-token.
func (s *Server) configHandler(w http.ResponseWriter, r *http.Request) {
	service := r.URL.Query().Get("service")
	if service == "" {
		service = "dnsimple" // pre-multi-service clients never send the param
	}
	app := s.apps[service]
	if app == nil {
		http.Error(w, "unknown service", http.StatusNotFound)
		return
	}
	out := map[string]any{
		"service":       app.service,
		"client_id":     app.clientID,
		"authorize_url": app.authorizeURL,
		"redirect_uri":  app.redirectURI,
	}
	// Manifest-declared provider scopes ride along (public data), so the client
	// can put the cap-matched scope set on the authorize URL: scopes for a
	// read-only connection, scopes_write for read & write. Absent for services
	// that scope elsewhere (GitHub Apps at install; DNSimple not at all).
	if m := integrations.MustRegistry().Get(app.service); m != nil && m.Auth.OAuth != nil {
		if len(m.Auth.OAuth.Scopes) > 0 {
			out["scopes"] = m.Auth.OAuth.Scopes
		}
		if len(m.Auth.OAuth.ScopesWrite) > 0 {
			out["scopes_write"] = m.Auth.OAuth.ScopesWrite
		}
	}
	writeJSON(w, http.StatusOK, out)
}

// registryHandler serves the embedded integrations registry. It is public,
// reviewed data (no secrets live in a manifest) and deliberately CORS-open —
// instances' web clients may fetch the catalog cross-origin.
func (s *Server) registryHandler(w http.ResponseWriter, r *http.Request) {
	w.Header().Set("Access-Control-Allow-Origin", "*")
	writeJSON(w, http.StatusOK, map[string]any{"schema": 1, "services": integrations.MustRegistry().List()})
}

// ---- redirect relay --------------------------------------------------------

// callback is the fleet's single registered redirect_uri per service. The
// provider sends a one-time ?code (or ?error) + the ?state we minted. `state`
// is base64url(origin) + "." + nonce; we decode the origin, refuse it unless
// an admin has registered it (originAllowed — so this can never be an open
// redirector), and 302 the code+state on to that instance's own callback page,
// which hands it to the app. We hold nothing and see only a single-use code.
func (s *Server) callback(w http.ResponseWriter, r *http.Request) {
	service := r.PathValue("service")
	if s.apps[service] == nil {
		http.Error(w, "unknown service", http.StatusNotFound)
		return
	}
	q := r.URL.Query()
	state := q.Get("state")
	dot := strings.IndexByte(state, '.')
	if dot <= 0 {
		// No (or malformed) state usually means GitHub sent someone here after
		// an app INSTALL made on github.com itself ("request OAuth during
		// installation" bounces to the registered callback with a code but no
		// state, since our flow never started). The install worked; a human is
		// looking at this page — say something humane instead of "bad state".
		w.Header().Set("Content-Security-Policy", "default-src 'none'; style-src 'unsafe-inline'")
		w.Header().Set("Content-Type", "text/html; charset=utf-8")
		w.WriteHeader(http.StatusOK)
		w.Write([]byte(`<!doctype html><meta charset="utf-8"><title>All set</title>` +
			`<body style="font:16px/1.5 system-ui;max-width:34em;margin:4rem auto;color:#333">` +
			`<h2>All set ✓</h2><p>If you were installing or updating an app on the provider's site, ` +
			`that worked. You can close this tab and head back to your chat — reconnect or refresh there if needed.</p>`))
		return
	}
	originBytes, err := base64.RawURLEncoding.DecodeString(state[:dot])
	if err != nil {
		http.Error(w, "bad state", http.StatusBadRequest)
		return
	}
	origin := string(originBytes)
	if !s.db.originAllowed(origin) {
		// Never redirect to an unregistered origin.
		http.Error(w, "unknown instance", http.StatusBadRequest)
		return
	}
	dest, err := url.Parse(origin + "/oauth/" + service + "/callback")
	if err != nil {
		http.Error(w, "bad origin", http.StatusBadRequest)
		return
	}
	// A billing-flow state (rest = "billing.<nonce>") asks the broker to finish
	// the dance itself: exchange the code right here, keep NOTHING, and send
	// the instance only the connected-account id (Stripe's stripe_user_id).
	// The admin's browser then links that id under its own admin session — the
	// broker can't write to spaces, and no token from this exchange ever
	// leaves (or outlives) this handler. Auth shape: the code is the proof of
	// a real consent (single-use, provider-minted), the origin is
	// registry-gated like every relay, and the exchange is rate-limited.
	if strings.HasPrefix(state[dot+1:], "billing.") {
		fwd := url.Values{}
		fwd.Set("state", state)
		code := q.Get("code")
		switch {
		case q.Get("error") != "":
			fwd.Set("error", q.Get("error"))
		case code == "":
			fwd.Set("error", "no code")
		case !s.perInstance.allow(origin) || !s.global.allow("*"):
			fwd.Set("error", "rate limited")
		default:
			pair, err := s.tokenRequest(s.apps[service], url.Values{
				"grant_type":   {"authorization_code"},
				"code":         {code},
				"redirect_uri": {s.apps[service].redirectURI},
				"state":        {state},
			})
			acct, _ := pair["stripe_user_id"].(string)
			if err != nil {
				// Log for us (never the code or token), generic error onward.
				log.Printf("billing exchange[%s %s]: %v", origin, service, err)
				fwd.Set("error", "exchange failed")
			} else if acct == "" {
				fwd.Set("error", "the provider didn't return an account id")
			} else {
				fwd.Set("billing_account", acct)
				if lm, ok := pair["livemode"].(bool); ok {
					fwd.Set("billing_livemode", strconv.FormatBool(lm))
				}
			}
		}
		dest.RawQuery = fwd.Encode()
		http.Redirect(w, r, dest.String(), http.StatusFound)
		return
	}
	fwd := url.Values{}
	fwd.Set("state", state)
	if code := q.Get("code"); code != "" {
		fwd.Set("code", code)
	}
	if e := q.Get("error"); e != "" { // relay a user cancellation/denial too
		fwd.Set("error", e)
	}
	dest.RawQuery = fwd.Encode()
	http.Redirect(w, r, dest.String(), http.StatusFound)
}

// ---- token exchange --------------------------------------------------------

func (s *Server) exchange(w http.ResponseWriter, r *http.Request) {
	origin, ok := s.db.authInstance(bearerToken(r))
	if !ok {
		http.Error(w, "unauthorized", http.StatusUnauthorized)
		return
	}
	if !s.perInstance.allow(origin) || !s.global.allow("*") {
		http.Error(w, "rate limited", http.StatusTooManyRequests)
		return
	}
	var body struct {
		Service string `json:"service"`
		Code    string `json:"code"`
		State   string `json:"state"`
	}
	if err := json.NewDecoder(http.MaxBytesReader(w, r.Body, maxBodyBytes)).Decode(&body); err != nil || body.Code == "" {
		http.Error(w, "bad request", http.StatusBadRequest)
		return
	}
	if body.Service == "" {
		body.Service = "dnsimple" // pre-multi-service sidecars never send it
	}
	app := s.apps[body.Service]
	if app == nil {
		http.Error(w, "unknown service", http.StatusBadRequest)
		return
	}

	pair, err := s.tokenRequest(app, url.Values{
		"grant_type":   {"authorization_code"},
		"code":         {body.Code},
		"redirect_uri": {app.redirectURI},
		"state":        {body.State},
	})
	if err != nil {
		// Log the failure for us (never the code or token), return a generic error.
		log.Printf("exchange[%s %s]: %v", origin, body.Service, err)
		http.Error(w, "exchange failed", http.StatusBadGateway)
		return
	}
	// The tokens are returned to the caller and never persisted or logged.
	writeJSON(w, http.StatusOK, pair)
}

// refresh rotates an expiring access token (providers with expiring
// user tokens — GitHub Apps). Same trust shape as exchange: the caller is an
// authenticated sidecar, the secret stays here, nothing is persisted. The
// provider rotates the refresh token too (GitHub's are single-use), so the
// whole new pair goes back for the sidecar to reseal to the member.
func (s *Server) refresh(w http.ResponseWriter, r *http.Request) {
	origin, ok := s.db.authInstance(bearerToken(r))
	if !ok {
		http.Error(w, "unauthorized", http.StatusUnauthorized)
		return
	}
	if !s.perInstance.allow(origin) || !s.global.allow("*") {
		http.Error(w, "rate limited", http.StatusTooManyRequests)
		return
	}
	var body struct {
		Service      string `json:"service"`
		RefreshToken string `json:"refresh_token"`
	}
	if err := json.NewDecoder(http.MaxBytesReader(w, r.Body, maxBodyBytes)).Decode(&body); err != nil || body.RefreshToken == "" {
		http.Error(w, "bad request", http.StatusBadRequest)
		return
	}
	app := s.apps[body.Service]
	if app == nil {
		http.Error(w, "unknown service", http.StatusBadRequest)
		return
	}
	pair, err := s.tokenRequest(app, url.Values{
		"grant_type":    {"refresh_token"},
		"refresh_token": {body.RefreshToken},
	})
	if err != nil {
		log.Printf("refresh[%s %s]: %v", origin, body.Service, err)
		http.Error(w, "refresh failed", http.StatusBadGateway)
		return
	}
	writeJSON(w, http.StatusOK, pair)
}

// tokenRequest runs one confidential token-endpoint request (exchange or
// refresh — the grant fields differ, the client pair and parsing don't) and
// returns the token pair. Accept: application/json matters: GitHub answers
// form-encoded without it; DNSimple answers JSON either way.
func (s *Server) tokenRequest(app *oauthApp, form url.Values) (map[string]any, error) {
	form.Set("client_id", app.clientID)
	form.Set("client_secret", app.clientSecret)
	if v := form.Get("state"); v == "" {
		form.Del("state")
	}
	req, err := http.NewRequest("POST", app.tokenURL, strings.NewReader(form.Encode()))
	if err != nil {
		return nil, err
	}
	req.Header.Set("Content-Type", "application/x-www-form-urlencoded")
	req.Header.Set("Accept", "application/json")
	resp, err := s.http.Do(req)
	if err != nil {
		return nil, err
	}
	defer resp.Body.Close()
	raw, _ := io.ReadAll(io.LimitReader(resp.Body, maxBodyBytes))
	var out struct {
		AccessToken  string `json:"access_token"`
		RefreshToken string `json:"refresh_token"`
		ExpiresIn    int64  `json:"expires_in"`
		StripeUserID string `json:"stripe_user_id"`    // Stripe Connect: the connected acct_… — the billing flow's whole product
		Livemode     *bool  `json:"livemode"`          // Stripe Connect: live vs sandbox consent
		Message      string `json:"message"`           // DNSimple's error shape
		Error        string `json:"error"`             // GitHub's error shape
		ErrorDesc    string `json:"error_description"` // (a 200 can still carry an error)
	}
	_ = json.Unmarshal(raw, &out)
	if resp.StatusCode != http.StatusOK || out.AccessToken == "" {
		for _, msg := range []string{out.Message, out.ErrorDesc, out.Error} {
			if msg != "" {
				return nil, fmt.Errorf("provider %d: %s", resp.StatusCode, msg)
			}
		}
		return nil, fmt.Errorf("provider %d", resp.StatusCode)
	}
	// Only the fields the sidecar reseals — a provider answering with extras
	// (scope, token_type) doesn't ride along.
	pair := map[string]any{"access_token": out.AccessToken}
	if out.RefreshToken != "" {
		pair["refresh_token"] = out.RefreshToken
	}
	if out.ExpiresIn > 0 {
		pair["expires_in"] = out.ExpiresIn
	}
	if out.StripeUserID != "" {
		pair["stripe_user_id"] = out.StripeUserID
	}
	if out.Livemode != nil {
		pair["livemode"] = *out.Livemode
	}
	return pair, nil
}

// ---- admin -----------------------------------------------------------------

func (s *Server) admin(next http.HandlerFunc) http.HandlerFunc {
	return func(w http.ResponseWriter, r *http.Request) {
		if subtle.ConstantTimeCompare([]byte(hashToken(bearerToken(r))), []byte(s.adminHash)) != 1 {
			http.Error(w, "unauthorized", http.StatusUnauthorized)
			return
		}
		next(w, r)
	}
}

func (s *Server) mintHandler(w http.ResponseWriter, r *http.Request) {
	var body struct {
		Label string `json:"label"` // the instance ORIGIN, e.g. https://test.example.com
	}
	if json.NewDecoder(http.MaxBytesReader(w, r.Body, 1<<10)).Decode(&body) != nil || body.Label == "" {
		http.Error(w, "label (instance origin) required", http.StatusBadRequest)
		return
	}
	id, token, err := s.db.mintToken(body.Label)
	if err != nil {
		http.Error(w, "db error", http.StatusInternalServerError)
		return
	}
	writeJSON(w, http.StatusOK, map[string]any{"id": id, "label": body.Label, "token": token})
}

func (s *Server) listHandler(w http.ResponseWriter, r *http.Request) {
	tokens, err := s.db.listTokens()
	if err != nil {
		http.Error(w, "db error", http.StatusInternalServerError)
		return
	}
	writeJSON(w, http.StatusOK, map[string]any{"tokens": tokens})
}

func (s *Server) revokeHandler(w http.ResponseWriter, r *http.Request) {
	deleted, err := s.db.revokeToken(r.PathValue("id"))
	if err != nil {
		http.Error(w, "db error", http.StatusInternalServerError)
		return
	}
	if !deleted {
		http.Error(w, "no such token", http.StatusNotFound)
		return
	}
	writeJSON(w, http.StatusOK, map[string]any{"ok": true})
}

// ---- middleware & helpers --------------------------------------------------

func securityHeaders(next http.Handler) http.Handler {
	return http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
		h := w.Header()
		h.Set("X-Content-Type-Options", "nosniff")
		h.Set("X-Frame-Options", "DENY")
		h.Set("Referrer-Policy", "no-referrer")
		h.Set("Strict-Transport-Security", "max-age=31536000")
		next.ServeHTTP(w, r)
	})
}

func bearerToken(r *http.Request) string {
	const p = "Bearer "
	h := r.Header.Get("Authorization")
	if len(h) > len(p) && h[:len(p)] == p {
		return h[len(p):]
	}
	return ""
}

func writeJSON(w http.ResponseWriter, status int, v any) {
	w.Header().Set("Content-Type", "application/json")
	w.WriteHeader(status)
	if err := json.NewEncoder(w).Encode(v); err != nil {
		log.Printf("write json: %v", err)
	}
}

func envOr(key, def string) string {
	if v := os.Getenv(key); v != "" {
		return v
	}
	return def
}

// ---- rate limiting (per-key token bucket, copied from push/main.go) --------

type rateLimiter struct {
	mu      sync.Mutex
	buckets map[string]*bucket
	burst   float64
	perSec  float64
}

type bucket struct {
	tokens float64
	last   time.Time
}

func newRateLimiter(burst, perSec float64) *rateLimiter {
	return &rateLimiter{buckets: map[string]*bucket{}, burst: burst, perSec: perSec}
}

func (rl *rateLimiter) allow(key string) bool {
	rl.mu.Lock()
	defer rl.mu.Unlock()
	now := time.Now()
	if len(rl.buckets) > 10000 {
		for k, b := range rl.buckets {
			if b.tokens >= rl.burst || now.Sub(b.last) > time.Hour {
				delete(rl.buckets, k)
			}
		}
	}
	b, ok := rl.buckets[key]
	if !ok {
		b = &bucket{tokens: rl.burst, last: now}
		rl.buckets[key] = b
	}
	b.tokens += now.Sub(b.last).Seconds() * rl.perSec
	if b.tokens > rl.burst {
		b.tokens = rl.burst
	}
	b.last = now
	if b.tokens < 1 {
		return false
	}
	b.tokens--
	return true
}