package server import ( "encoding/json" "log" "net/http" "strconv" "time" "git.unkin.net/unkin/logviewer/web" ) type Config struct { CHURL string CHUser string CHPassword string Version string } type Server struct { ch *chClient version string now func() time.Time mux *http.ServeMux } func New(cfg Config) *Server { s := &Server{ ch: newCHClient(cfg.CHURL, cfg.CHUser, cfg.CHPassword), version: cfg.Version, now: time.Now, } s.routes() return s } func (s *Server) routes() { s.mux = http.NewServeMux() s.mux.HandleFunc("GET /api/query", s.handleQuery) s.mux.HandleFunc("GET /api/tail", s.handleTail) s.mux.HandleFunc("GET /api/facets", s.handleFacets) s.mux.HandleFunc("GET /healthz", s.handleHealthz) s.mux.Handle("GET /", http.FileServerFS(web.FS)) } func (s *Server) ServeHTTP(w http.ResponseWriter, r *http.Request) { s.mux.ServeHTTP(w, r) } 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 response: %v", err) } } func writeError(w http.ResponseWriter, status int, err error) { writeJSON(w, status, map[string]string{"error": err.Error()}) } func (s *Server) handleQuery(w http.ResponseWriter, r *http.Request) { form := r.URL.Query() tr, err := parseTimeRange(form.Get("since"), form.Get("until"), s.now()) if err != nil { writeError(w, http.StatusBadRequest, err) return } limit, err := parseLimit(form.Get("limit"), defaultLimit) if err != nil { writeError(w, http.StatusBadRequest, err) return } offset, err := parseOffset(form.Get("offset")) if err != nil { writeError(w, http.StatusBadRequest, err) return } params := map[string]string{} where, err := whereClause(tr, form, params, false) if err != nil { writeError(w, http.StatusBadRequest, err) return } res, err := s.ch.query(r.Context(), buildQuerySQL(where, limit, offset), params) if err != nil { writeError(w, http.StatusBadGateway, err) return } writeJSON(w, http.StatusOK, map[string]any{ "rows": res.Data, "count": len(res.Data), "since": tr.since.UTC().Format(time.RFC3339Nano), "until": tr.until.UTC().Format(time.RFC3339Nano), "limit": limit, "offset": offset, }) } func (s *Server) handleTail(w http.ResponseWriter, r *http.Request) { form := r.URL.Query() now := s.now() // The cursor is the exclusive lower bound in ms. It is clamped to the // default window so a stale client can never trigger a wide scan. cursor := now.Add(-10 * time.Second).UnixMilli() if c := form.Get("cursor"); c != "" { n, err := strconv.ParseInt(c, 10, 64) if err != nil { writeError(w, http.StatusBadRequest, err) return } cursor = n } if min := now.Add(-defaultWindow).UnixMilli(); cursor < min { cursor = min } // The tail bound is exclusive of the cursor so rows never repeat. tr := timeRange{since: time.UnixMilli(cursor), until: now.Add(time.Second)} params := map[string]string{} where, err := whereClause(tr, form, params, true) if err != nil { writeError(w, http.StatusBadRequest, err) return } res, err := s.ch.query(r.Context(), buildTailSQL(where), params) if err != nil { writeError(w, http.StatusBadGateway, err) return } next := cursor for _, row := range res.Data { if ts, ok := numField(row, "ts_ms"); ok && ts > next { next = ts } } writeJSON(w, http.StatusOK, map[string]any{ "rows": res.Data, "count": len(res.Data), "cursor": next, }) } // numField reads an int64-ish field that ClickHouse may return as a JSON // number or (with 64-bit quoting) a string. func numField(row map[string]any, key string) (int64, bool) { switch v := row[key].(type) { case float64: return int64(v), true case string: n, err := strconv.ParseInt(v, 10, 64) return n, err == nil case json.Number: n, err := v.Int64() return n, err == nil } return 0, false } func (s *Server) handleFacets(w http.ResponseWriter, r *http.Request) { form := r.URL.Query() tr, err := parseTimeRange(form.Get("since"), form.Get("until"), s.now()) if err != nil { writeError(w, http.StatusBadRequest, err) return } params := map[string]string{} // Facets only take the time window; drop filters so dropdowns always // show the full set for the window. where, err := whereClause(tr, nil, params, false) if err != nil { writeError(w, http.StatusBadRequest, err) return } res, err := s.ch.query(r.Context(), buildFacetsSQL(where), params) if err != nil { writeError(w, http.StatusBadGateway, err) return } out := map[string][]map[string]any{ "namespaces": {}, "apps": {}, "hosts": {}, } keys := map[string]string{"namespace": "namespaces", "app": "apps", "host": "hosts"} for _, row := range res.Data { facet, _ := row["facet"].(string) key, ok := keys[facet] if !ok { continue } n, _ := numField(row, "n") out[key] = append(out[key], map[string]any{"value": row["value"], "count": n}) } writeJSON(w, http.StatusOK, out) } func (s *Server) handleHealthz(w http.ResponseWriter, r *http.Request) { if err := s.ch.ping(r.Context()); err != nil { writeError(w, http.StatusServiceUnavailable, err) return } writeJSON(w, http.StatusOK, map[string]string{"status": "ok", "version": s.version}) }