| 1 | package edge |
| 2 | |
| 3 | import ( |
| 4 | "bufio" |
| 5 | "fmt" |
| 6 | "net" |
| 7 | "net/http" |
| 8 | "sync" |
| 9 | |
| 10 | "github.com/carlosframework/platform/internal/registry" |
| 11 | "github.com/carlosframework/platform/internal/release" |
| 12 | "github.com/carlosframework/platform/internal/usage" |
| 13 | ) |
| 14 | |
| 15 | // UsageMeter accumulates per-route request/byte counters between flushes. |
| 16 | // Record sits on the serving path: one mutex'd map update per request — |
| 17 | // no allocation, no I/O. Drain swaps the map out for the usage flush loop |
| 18 | // (internal/host), so a slow bucket write never blocks serving. |
| 19 | type UsageMeter struct { |
| 20 | mu sync.Mutex |
| 21 | m map[string]usage.EdgeRow |
| 22 | } |
| 23 | |
| 24 | // NewUsageMeter returns an empty meter. |
| 25 | func NewUsageMeter() *UsageMeter { |
| 26 | return &UsageMeter{m: map[string]usage.EdgeRow{}} |
| 27 | } |
| 28 | |
| 29 | // Record adds one served exchange for rt. hopped says the request arrived |
| 30 | // over a verified backhaul hop from a sibling edge (homemap.go) — the same |
| 31 | // exchange is then ALSO added to the row's hop counters, which are a subset |
| 32 | // annotation inside Requests/BytesOut and never a second exchange (see |
| 33 | // usage.EdgeRow). One method rather than a separate RecordHop: two methods |
| 34 | // for one exchange invite calling both, which is exactly the double count |
| 35 | // this arrangement exists to make impossible. |
| 36 | func (u *UsageMeter) Record(rt registry.Route, requests, bytes int64, hopped bool) { |
| 37 | u.mu.Lock() |
| 38 | row := u.m[rt.Host] |
| 39 | row.Account = release.RouteAccount(rt) |
| 40 | row.App = rt.App |
| 41 | row.Requests += requests |
| 42 | row.BytesOut += bytes |
| 43 | if hopped { |
| 44 | row.HopRequests += requests |
| 45 | row.HopBytesOut += bytes |
| 46 | } |
| 47 | u.m[rt.Host] = row |
| 48 | u.mu.Unlock() |
| 49 | } |
| 50 | |
| 51 | // Drain returns the accumulated rows and resets the meter. |
| 52 | func (u *UsageMeter) Drain() map[string]usage.EdgeRow { |
| 53 | u.mu.Lock() |
| 54 | out := u.m |
| 55 | u.m = map[string]usage.EdgeRow{} |
| 56 | u.mu.Unlock() |
| 57 | return out |
| 58 | } |
| 59 | |
| 60 | // countingWriter counts response body bytes and OBSERVES the final status |
| 61 | // and Content-Type (spec 2026-09-07 §3.3): it is the one wrapper installed |
| 62 | // at a single site for every response shape — cache hit, plain proxy, |
| 63 | // hibernating, upgrade — so it is where the edge learns that an upstream |
| 64 | // answered 5xx (an edge error) or served HTML (a pageview, step 2). It |
| 65 | // sits below the compressor, so Content-Type is the synced value |
| 66 | // gzipWriter.syncHeader copied down before calling this WriteHeader. |
| 67 | // Deliberately does NOT implement io.ReaderFrom: io.Copy falls back to |
| 68 | // Write, which counts. Hijack/Flush forward via the interfaces the wrapped |
| 69 | // writer provides, mirroring recordingResponseWriter (edge.go). |
| 70 | type countingWriter struct { |
| 71 | http.ResponseWriter |
| 72 | n *int64 |
| 73 | status int // final status (>= 200); 0 until one is written |
| 74 | ctype string // Content-Type at the final WriteHeader |
| 75 | wroteHdr bool |
| 76 | } |
| 77 | |
| 78 | // WriteHeader latches the first final status. 1xx are forwarded and never |
| 79 | // latched (recordingResponseWriter.WriteHeader's reasoning). |
| 80 | func (c *countingWriter) WriteHeader(status int) { |
| 81 | if status < 200 { |
| 82 | c.ResponseWriter.WriteHeader(status) |
| 83 | return |
| 84 | } |
| 85 | if !c.wroteHdr { |
| 86 | c.wroteHdr = true |
| 87 | c.status = status |
| 88 | c.ctype = c.ResponseWriter.Header().Get("Content-Type") |
| 89 | } |
| 90 | c.ResponseWriter.WriteHeader(status) |
| 91 | } |
| 92 | |
| 93 | func (c *countingWriter) Write(p []byte) (int, error) { |
| 94 | if !c.wroteHdr { |
| 95 | c.wroteHdr = true |
| 96 | c.status = http.StatusOK |
| 97 | c.ctype = c.ResponseWriter.Header().Get("Content-Type") |
| 98 | } |
| 99 | n, err := c.ResponseWriter.Write(p) |
| 100 | *c.n += int64(n) |
| 101 | return n, err |
| 102 | } |
| 103 | |
| 104 | // Hijack forwards to the underlying ResponseWriter's http.Hijacker, when it |
| 105 | // implements one — the same optional-interface pattern the stdlib itself |
| 106 | // uses, and the same forwarding recordingResponseWriter provides. |
| 107 | func (c *countingWriter) Hijack() (net.Conn, *bufio.ReadWriter, error) { |
| 108 | hj, ok := c.ResponseWriter.(http.Hijacker) |
| 109 | if !ok { |
| 110 | return nil, nil, fmt.Errorf("edge: underlying ResponseWriter does not support http.Hijacker") |
| 111 | } |
| 112 | return hj.Hijack() |
| 113 | } |
| 114 | |
| 115 | // Flush forwards to the underlying ResponseWriter's http.Flusher, when it |
| 116 | // implements one — needed so a metered (non-upgrade) response can still |
| 117 | // stream incrementally to the real client, e.g. Server-Sent Events, instead |
| 118 | // of buffering until the handler returns. |
| 119 | func (c *countingWriter) Flush() { |
| 120 | if f, ok := c.ResponseWriter.(http.Flusher); ok { |
| 121 | f.Flush() |
| 122 | } |
| 123 | } |
| 124 | |
amadan