<commit_before>package web

import (
	"bytes"
	"compress/gzip"
	"context"
	"encoding/json"
	"fmt"
	"math"
	"net/http"
	"net/url"
	"sort"
	"sync"
	"time"

	"github.com/dustin/go-humanize"
	"github.com/getlantern/zenodb/core"
	"github.com/getlantern/zenodb/encoding"
	"github.com/getlantern/zenodb/sql"
	"github.com/gorilla/mux"
	"github.com/retailnext/hllpp"
)

const (
	nanosPerMilli = 1000000

	pauseTime    = 250 * time.Millisecond
	shortTimeout = 5 * time.Second
	longTimeout  = 1000 * time.Hour
)

type QueryResult struct {
	SQL                string
	Permalink          string
	TS                 int64
	TSCardinality      uint64
	Fields             []string
	FieldCardinalities []uint64
	Dims               []string
	DimCardinalities   []uint64
	Rows               []*ResultRow
}

type ResultRow struct {
	TS   int64
	Key  map[string]interface{}
	Vals []float64
}

type query struct {
	sqlString string
	parsed    *sql.Query
	ce        cacheEntry
}

func (h *handler) runQuery(resp http.ResponseWriter, req *http.Request) {
	h.sqlQuery(resp, req, longTimeout)
}

func (h *handler) asyncQuery(resp http.ResponseWriter, req *http.Request) {
	h.sqlQuery(resp, req, shortTimeout)
}

func (h *handler) cachedQuery(resp http.ResponseWriter, req *http.Request) {
	if !h.authenticate(resp, req) {
		resp.WriteHeader(http.StatusForbidden)
		return
	}

	log.Debug(req.URL)
	permalink := mux.Vars(req)["permalink"]
	ce, err := h.cache.getByPermalink(permalink)
	if ce == nil {
		http.NotFound(resp, req)
		return
	}
	h.respondWithCacheEntry(resp, req, ce, err, shortTimeout)
}

func (h *handler) sqlQuery(resp http.ResponseWriter, req *http.Request, timeout time.Duration) {
	if !h.authenticate(resp, req) {
		resp.WriteHeader(http.StatusForbidden)
		return
	}

	log.Debug(req.URL)
	sqlString, _ := url.QueryUnescape(req.URL.RawQuery)

	ce, err := h.query(req, sqlString)
	h.respondWithCacheEntry(resp, req, ce, err, timeout)
}

func (h *handler) respondWithCacheEntry(resp http.ResponseWriter, req *http.Request, ce cacheEntry, err error, timeout time.Duration) {
	limit := int(timeout / pauseTime)
	for i := 0; i < limit; i++ {
		if err != nil {
			log.Error(err)
			resp.WriteHeader(http.StatusInternalServerError)
			fmt.Fprint(resp, err.Error())
			return
		}
		switch ce.status() {
		case statusSuccess:
			h.respondSuccess(resp, req, ce)
			return
		case statusError:
			h.respondError(resp, req, ce)
			return
		case statusPending:
			// Pause a little bit and try again
			time.Sleep(pauseTime)
			ce, err = h.cache.getByPermalink(ce.permalink())
		}
	}
	// Let the client know that we're still working on it
	resp.WriteHeader(http.StatusAccepted)
	fmt.Fprintf(resp, "/cached/%v", ce.permalink())
}

func (h *handler) respondSuccess(resp http.ResponseWriter, req *http.Request, ce cacheEntry) {
	resp.Header().Set("Content-Type", "application/json")
	resp.Header().Set("Expires", "0")
	resp.Header().Set("Cache-control", "no-cache, no-store, must-revalidate")
	resp.Header().Set("Content-Encoding", "gzip")
	resp.WriteHeader(http.StatusOK)
	resp.Write(ce.data())
}

func (h *handler) respondError(resp http.ResponseWriter, req *http.Request, ce cacheEntry) {
	resp.WriteHeader(http.StatusInternalServerError)
	resp.Write(ce.error())
}

func (h *handler) query(req *http.Request, sqlString string) (ce cacheEntry, err error) {
	parsed, parseErr := sql.Parse(sqlString)
	if parseErr != nil {
		return nil, parseErr
	}

	if req.Header.Get("Cache-control") == "no-cache" {
		ce, err = h.cache.begin(sqlString)
		if err != nil {
			return
		}
	} else {
		var created bool
		ce, created, err = h.cache.getOrBegin(sqlString)
		if err != nil || !created {
			return
		}
		if ce.status() != statusPending {
			log.Debugf("Found results for %v in cache", sqlString)
			return
		}
	}

	// Request query to run in background
	h.queries <- &query{sqlString, parsed, ce}

	return
}

func (h *handler) coalesceQueries() {
	for q := range h.queries {
		coalescedQueries := append([]*query(nil), q)
		var remainingQueries []*query
		table := q.parsed.From
	coalesceLoop:
		for {
			select {
			case query := <-h.queries:
				if table == query.parsed.From {
					coalescedQueries = append(coalescedQueries)
				} else {
					remainingQueries = append(remainingQueries, query)
				}
			case <-time.After(15 * time.Second):
				break coalesceLoop
			}
		}

		log.Debugf("Coalesced %d queries to %v", len(coalescedQueries), table)
		// re-queue queries that weren't included in this run
		for _, query := range remainingQueries {
			h.queries <- query
		}
		h.coalescedQueries <- coalescedQueries
	}
}

func (h *handler) processQueries() {
	for queries := range h.coalescedQueries {
		for _, query := range queries {
			sqlString := query.sqlString
			ce := query.ce
			result, err := h.doQuery(sqlString, ce.permalink())
			if err != nil {
				err = fmt.Errorf("Unable to query: %v", err)
				log.Error(err)<commit_msg>Cleaned up web coalescing<commit_after>package web

import (
	"bytes"
	"compress/gzip"
	"context"
	"encoding/json"
	"fmt"
	"math"
	"net/http"
	"net/url"
	"sort"
	"sync"
	"time"

	"github.com/dustin/go-humanize"
	"github.com/getlantern/zenodb/core"
	"github.com/getlantern/zenodb/encoding"
	"github.com/getlantern/zenodb/sql"
	"github.com/gorilla/mux"
	"github.com/retailnext/hllpp"
)

const (
	nanosPerMilli = 1000000

	pauseTime    = 250 * time.Millisecond
	shortTimeout = 5 * time.Second
	longTimeout  = 1000 * time.Hour
)

type QueryResult struct {
	SQL                string
	Permalink          string
	TS                 int64
	TSCardinality      uint64
	Fields             []string
	FieldCardinalities []uint64
	Dims               []string
	DimCardinalities   []uint64
	Rows               []*ResultRow
}

type ResultRow struct {
	TS   int64
	Key  map[string]interface{}
	Vals []float64
}

type query struct {
	sqlString string
	parsed    *sql.Query
	ce        cacheEntry
}

func (h *handler) runQuery(resp http.ResponseWriter, req *http.Request) {
	h.sqlQuery(resp, req, longTimeout)
}

func (h *handler) asyncQuery(resp http.ResponseWriter, req *http.Request) {
	h.sqlQuery(resp, req, shortTimeout)
}

func (h *handler) cachedQuery(resp http.ResponseWriter, req *http.Request) {
	if !h.authenticate(resp, req) {
		resp.WriteHeader(http.StatusForbidden)
		return
	}

	log.Debug(req.URL)
	permalink := mux.Vars(req)["permalink"]
	ce, err := h.cache.getByPermalink(permalink)
	if ce == nil {
		http.NotFound(resp, req)
		return
	}
	h.respondWithCacheEntry(resp, req, ce, err, shortTimeout)
}

func (h *handler) sqlQuery(resp http.ResponseWriter, req *http.Request, timeout time.Duration) {
	if !h.authenticate(resp, req) {
		resp.WriteHeader(http.StatusForbidden)
		return
	}

	log.Debug(req.URL)
	sqlString, _ := url.QueryUnescape(req.URL.RawQuery)

	ce, err := h.query(req, sqlString)
	h.respondWithCacheEntry(resp, req, ce, err, timeout)
}

func (h *handler) respondWithCacheEntry(resp http.ResponseWriter, req *http.Request, ce cacheEntry, err error, timeout time.Duration) {
	limit := int(timeout / pauseTime)
	for i := 0; i < limit; i++ {
		if err != nil {
			log.Error(err)
			resp.WriteHeader(http.StatusInternalServerError)
			fmt.Fprint(resp, err.Error())
			return
		}
		switch ce.status() {
		case statusSuccess:
			h.respondSuccess(resp, req, ce)
			return
		case statusError:
			h.respondError(resp, req, ce)
			return
		case statusPending:
			// Pause a little bit and try again
			time.Sleep(pauseTime)
			ce, err = h.cache.getByPermalink(ce.permalink())
		}
	}
	// Let the client know that we're still working on it
	resp.WriteHeader(http.StatusAccepted)
	fmt.Fprintf(resp, "/cached/%v", ce.permalink())
}

func (h *handler) respondSuccess(resp http.ResponseWriter, req *http.Request, ce cacheEntry) {
	resp.Header().Set("Content-Type", "application/json")
	resp.Header().Set("Expires", "0")
	resp.Header().Set("Cache-control", "no-cache, no-store, must-revalidate")
	resp.Header().Set("Content-Encoding", "gzip")
	resp.WriteHeader(http.StatusOK)
	resp.Write(ce.data())
}

func (h *handler) respondError(resp http.ResponseWriter, req *http.Request, ce cacheEntry) {
	resp.WriteHeader(http.StatusInternalServerError)
	resp.Write(ce.error())
}

func (h *handler) query(req *http.Request, sqlString string) (ce cacheEntry, err error) {
	parsed, parseErr := sql.Parse(sqlString)
	if parseErr != nil {
		return nil, parseErr
	}

	if req.Header.Get("Cache-control") == "no-cache" {
		ce, err = h.cache.begin(sqlString)
		if err != nil {
			return
		}
	} else {
		var created bool
		ce, created, err = h.cache.getOrBegin(sqlString)
		if err != nil || !created {
			return
		}
		if ce.status() != statusPending {
			log.Debugf("Found results for %v in cache", sqlString)
			return
		}
	}

	// Request query to run in background
	h.queries <- &query{sqlString, parsed, ce}

	return
}

func (h *handler) coalesceQueries() {
	for q := range h.queries {
		coalescedQueries := append([]*query(nil), q)
		var remainingQueries []*query
		table := q.parsed.From
	coalesceLoop:
		for {
			select {
			case query := <-h.queries:
				if table == query.parsed.From {
					coalescedQueries = append(coalescedQueries, query)
				} else {
					remainingQueries = append(remainingQueries, query)
				}
			case <-time.After(15 * time.Second):
				break coalesceLoop
			}
		}

		log.Debugf("Coalesced %d queries to %v", len(coalescedQueries), table)
		// re-queue queries that weren't included in this run
		for _, query := range remainingQueries {
			h.queries <- query
		}
		h.coalescedQueries <- coalescedQueries
	}
}

func (h *handler) processQueries() {
	for queries := range h.coalescedQueries {
		for _, query := range queries {
			sqlString := query.sqlString
			ce := query.ce
			result, err := h.doQuery(sqlString, ce.permalink())
			if err != nil {
				err = fmt.Errorf("Unable to query: %v", err)
				log.Error(err)
{'fraction_non_alphanumeric': 0.1342493638676845, 'fraction_numerical': 0.006310432569974555, 'mean_word_length': 3.6283560998586903, 'pattern_counts': {'":': 0, '<': 15, '<?xml version=': 0, '>': 3, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 0, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '1591417', 'n_tokens_mistral': 3785, 'n_tokens_neox': 3484, 'n_words': 809}