registerLiveAPI wires the live-state read endpoints + the session-kill and per-user kill-switch actions. fetcher (may be nil) aggregates / fans out per-CP in-memory state across replicas; users (may be nil) persists the disabled flag for the disable/enable endpoints.
(r *gin.RouterGroup, live LiveInfo, fetcher PeerFetcher, users UserAdmin)
| 164 | // org/user has no live sessions on this replica. |
| 165 | KillUserSessions(orgID, username string) int |
| 166 | // RecentErrors returns up to limit of THIS replica's most recent redacted |
| 167 | // query errors, newest first. Each error is captured on the CP that owned the |
| 168 | // failing connection, so the handler fans out and concatenates across replicas |
| 169 | // (no cross-CP dedup needed — an error belongs to exactly one CP). |
| 170 | RecentErrors(limit int) []ErrorEntry |
| 171 | } |
| 172 | |
| 173 | // UserAdmin persists the per-user kill switch (the disabled flag) and forces a |
| 174 | // config-snapshot reload so a flip takes effect immediately rather than one poll |
| 175 | // interval later. *configstore.ConfigStore satisfies it. nil disables the |
| 176 | // disable/enable endpoints (they 503). |
| 177 | type UserAdmin interface { |
| 178 | SetOrgUserDisabled(orgID, username string, disabled bool) error |
| 179 | ReloadSnapshot() error |
| 180 | } |
| 181 | |
| 182 | // registerLiveAPI wires the live-state read endpoints + the session-kill and |
| 183 | // per-user kill-switch actions. fetcher (may be nil) aggregates / fans out |
| 184 | // per-CP in-memory state across replicas; users (may be nil) persists the |
| 185 | // disabled flag for the disable/enable endpoints. |
| 186 | func registerLiveAPI(r *gin.RouterGroup, live LiveInfo, fetcher PeerFetcher, users UserAdmin) { |
| 187 | if live == nil { |
| 188 | return |
| 189 | } |
| 190 | |
| 191 | // userActionPath builds the peer fan-out path for a per-user action, escaping |
| 192 | // org/user so names with slashes or reserved chars route correctly. |
| 193 | userActionPath := func(org, user, action string) string { |
| 194 | return "/api/v1/orgs/" + url.PathEscape(org) + "/users/" + url.PathEscape(user) + "/" + action |
| 195 | } |
| 196 | // sumKilled parses peer {"killed":N} bodies, returning the total and how many |
| 197 | // peers responded with a parseable 200. |
| 198 | sumKilled := func(bodies [][]byte) (killed, responders int) { |
| 199 | for _, b := range bodies { |
| 200 | var e struct { |
| 201 | Killed int `json:"killed"` |
| 202 | } |
| 203 | if json.Unmarshal(b, &e) == nil { |
| 204 | killed += e.Killed |
| 205 | } |
| 206 | responders++ |
| 207 | } |
| 208 | return |
| 209 | } |
| 210 | r.GET("/queries", func(c *gin.Context) { |
| 211 | queries := live.RunningQueries() |
| 212 | responders, total := 1, 1 |
| 213 | // Aggregate every other CP's in-memory view (a query lives on exactly |
| 214 | // one CP, so the union is disjoint; dedupeBy makes it idempotent anyway). |
| 215 | if !localScope(c) && fetcher != nil { |
| 216 | peerResult := fetcher.FetchPeers(c.Request.Context(), "/api/v1/queries") |
| 217 | type env struct { |
| 218 | Queries []QueryStatus `json:"queries"` |
| 219 | } |
| 220 | coverage := peerReadCoverage(peerResult, mergePeer(&queries, peerResult.Bodies, func(e env) []QueryStatus { return e.Queries })) |
| 221 | responders, total = coverage.Responders, coverage.Total |
| 222 | queries = dedupeBy(queries, func(q QueryStatus) int { return q.WorkerID }) |
| 223 | } |