<commit_before>package main

import (
	"bytes"
	"crypto/rand"
	"encoding/json"
	"fmt"
	"io"
	"log"
	"net/http"
	"net/http/httputil"
	"net/url"
	"os"
	"path/filepath"
	"strconv"
	"strings"
	"sync"
	"time"

	"github.com/BurntSushi/toml"
	"github.com/julienschmidt/httprouter"
	"github.com/robustirc/robustirc/config"
	"github.com/robustirc/robustirc/ircserver"
	"github.com/robustirc/robustirc/robusthttp"
	"github.com/robustirc/robustirc/types"
	"github.com/stapelberg/glog"

	"github.com/hashicorp/raft"
)

var (
	GetMessageRequests    = make(map[string]GetMessageStats)
	getMessagesRequestsMu sync.Mutex

	configMu sync.Mutex

	// applyMu guards calls to raft.Apply(). We need to lock them because
	// otherwise we cannot guarantee that multiple goroutines will write
	// strictly monotonically increasing timestamps.
	applyMu sync.Mutex

	// To avoid setting up a new proxy on every request, we cache the proxies
	// for each node (since the current leader might change abruptly).
	nodeProxies   = make(map[string]*httputil.ReverseProxy)
	nodeProxiesMu sync.RWMutex
)

type GetMessageStats struct {
	Session types.RobustId
	Nick    string
	Started time.Time
}

func (stats GetMessageStats) StartedAndRelative() string {
	return stats.Started.Format("2006-01-02 15:04:05 -07:00") + " (" +
		time.Now().Round(time.Second).Sub(stats.Started.Round(time.Second)).String() + " ago)"
}

type nopCloser struct {
	io.Reader
}

func (nopCloser) Close() error {
	return nil
}

func maybeProxyToLeader(w http.ResponseWriter, r *http.Request, body io.ReadCloser) {
	leader := node.Leader()
	if leader == nil {
		http.Error(w, fmt.Sprintf("No leader known. Please try another server."),
			http.StatusInternalServerError)
		return
	}

	nodeProxiesMu.RLock()
	p, ok := nodeProxies[leader.String()]
	nodeProxiesMu.RUnlock()

	if !ok {
		u, err := url.Parse("https://" + leader.String())
		if err != nil {
			http.Error(w, fmt.Sprintf("url.Parse(): %v", err), http.StatusInternalServerError)
			return
		}
		p = httputil.NewSingleHostReverseProxy(u)
		p.Transport = robusthttp.Transport()

		// Races are okay, i.e. overwriting the proxy a different goroutine set up.
		nodeProxiesMu.Lock()
		nodeProxies[leader.String()] = p
		nodeProxiesMu.Unlock()
	}

	location := *r.URL
	location.Host = leader.String()
	w.Header().Set("Content-Location", location.String())
	log.Printf("Proxying request (%q) to leader %q\n", r.URL.Path, leader.String())
	r.Body = body
	p.ServeHTTP(w, r)
}

func session(r *http.Request, ps httprouter.Params) (types.RobustId, error) {
	var sessionid types.RobustId

	id, err := strconv.ParseInt(ps[0].Value, 0, 64)
	if err != nil {
		return sessionid, fmt.Errorf("invalid session: %v", err)
	}

	header := r.Header.Get("X-Session-Auth")
	if header == "" {
		return sessionid, fmt.Errorf("no X-Session-Auth header set")
	}

	auth, err := ircServer.GetAuth(types.RobustId{Id: id})
	if err != nil {
		return sessionid, err
	}
	if header != auth {
		return sessionid, fmt.Errorf("invalid X-Session-Auth header")
	}

	sessionid.Id = id

	return sessionid, nil
}

func sessionOrProxy(w http.ResponseWriter, r *http.Request, ps httprouter.Params) (types.RobustId, error) {
	sessionid, err := session(r, ps)
	if err == ircserver.ErrSessionNotYetSeen && node.State() != raft.Leader {
		// The session might exist on the leader, so we must proxy.
		maybeProxyToLeader(w, r, r.Body)
		return sessionid, err
	}

	if err != nil {
		http.Error(w, err.Error(), http.StatusNotFound)
	}
	return sessionid, err
}

// handlePostMessage is called by the robustirc-brigde whenever a message should be
// posted. The handler blocks until either the data was written or an error
// occurred. If successful, it returns the unique id of the message.
func handlePostMessage(w http.ResponseWriter, r *http.Request, ps httprouter.Params) {
	session, err := sessionOrProxy(w, r, ps)
	if err != nil {
		return
	}

	// Don’t throttle server-to-server connections (services)
	if netConfig.PostMessageCooloff > 0 {
		until := ircServer.ThrottleUntil(session, time.Duration(netConfig.PostMessageCooloff))
		time.Sleep(until.Sub(time.Now()))
	}

	type postMessageRequest struct {
		Data            string
		ClientMessageId uint64
	}

	var req postMessageRequest

	// We limit the amount of bytes read to 1024 to prevent reading overly long
	// requests in the first place. The IRC line length limit is 512 bytes, so
	// with 1024 bytes we have plenty of headroom to encode 512 bytes in JSON.
	//
	// We save a copy of the request in case we need to proxy it to the leader.
	var body bytes.Buffer
	rd := io.TeeReader(http.MaxBytesReader(w, r.Body, 1024), &body)
	if err := json.NewDecoder(rd).Decode(&req); err != nil {
		http.Error(w, err.Error(), http.StatusBadRequest)
		return
	}

	// If we have already seen this message, we just reply with a canned response.
	if id, reply := ircServer.LastPostMessage(session); id == req.ClientMessageId && reply != nil {
		w.Header().Set("Content-Type", "application/json")
		w.Write(reply)
		return
	}

	applyMu.Lock()
	defer applyMu.Unlock()
	msg := ircServer.NewRobustMessage(types.RobustIRCFromClient, session, req.Data)
	msg.ClientMessageId = req.ClientMessageId
	msgbytes, err := json.Marshal(msg)
	if err != nil {
		http.Error(w, fmt.Sprintf("Could not store message, cannot encode it as JSON: %v", err),
			http.StatusBadRequest)
		return
	}

	f := node.Apply(msgbytes, 10*time.Second)
	if err := f.Error(); err != nil {
		if err == raft.ErrNotLeader {
			maybeProxyToLeader(w, r, nopCloser{&body})
			return
		}
		http.Error(w, fmt.Sprintf("Apply(): %v", err), http.StatusInternalServerError)
		return
	}
}

func handleJoin(w http.ResponseWriter, r *http.Request) {
	log.Println("Join request from", r.RemoteAddr)
	if node.State() != raft.Leader {
		maybeProxyToLeader(w, r, r.Body)
		return<commit_msg>Unlock early to avoid deadlocks.<commit_after>package main

import (
	"bytes"
	"crypto/rand"
	"encoding/json"
	"fmt"
	"io"
	"log"
	"net/http"
	"net/http/httputil"
	"net/url"
	"os"
	"path/filepath"
	"strconv"
	"strings"
	"sync"
	"time"

	"github.com/BurntSushi/toml"
	"github.com/julienschmidt/httprouter"
	"github.com/robustirc/robustirc/config"
	"github.com/robustirc/robustirc/ircserver"
	"github.com/robustirc/robustirc/robusthttp"
	"github.com/robustirc/robustirc/types"
	"github.com/stapelberg/glog"

	"github.com/hashicorp/raft"
)

var (
	GetMessageRequests    = make(map[string]GetMessageStats)
	getMessagesRequestsMu sync.Mutex

	configMu sync.Mutex

	// applyMu guards calls to raft.Apply(). We need to lock them because
	// otherwise we cannot guarantee that multiple goroutines will write
	// strictly monotonically increasing timestamps.
	applyMu sync.Mutex

	// To avoid setting up a new proxy on every request, we cache the proxies
	// for each node (since the current leader might change abruptly).
	nodeProxies   = make(map[string]*httputil.ReverseProxy)
	nodeProxiesMu sync.RWMutex
)

type GetMessageStats struct {
	Session types.RobustId
	Nick    string
	Started time.Time
}

func (stats GetMessageStats) StartedAndRelative() string {
	return stats.Started.Format("2006-01-02 15:04:05 -07:00") + " (" +
		time.Now().Round(time.Second).Sub(stats.Started.Round(time.Second)).String() + " ago)"
}

type nopCloser struct {
	io.Reader
}

func (nopCloser) Close() error {
	return nil
}

func maybeProxyToLeader(w http.ResponseWriter, r *http.Request, body io.ReadCloser) {
	leader := node.Leader()
	if leader == nil {
		http.Error(w, fmt.Sprintf("No leader known. Please try another server."),
			http.StatusInternalServerError)
		return
	}

	nodeProxiesMu.RLock()
	p, ok := nodeProxies[leader.String()]
	nodeProxiesMu.RUnlock()

	if !ok {
		u, err := url.Parse("https://" + leader.String())
		if err != nil {
			http.Error(w, fmt.Sprintf("url.Parse(): %v", err), http.StatusInternalServerError)
			return
		}
		p = httputil.NewSingleHostReverseProxy(u)
		p.Transport = robusthttp.Transport()

		// Races are okay, i.e. overwriting the proxy a different goroutine set up.
		nodeProxiesMu.Lock()
		nodeProxies[leader.String()] = p
		nodeProxiesMu.Unlock()
	}

	location := *r.URL
	location.Host = leader.String()
	w.Header().Set("Content-Location", location.String())
	log.Printf("Proxying request (%q) to leader %q\n", r.URL.Path, leader.String())
	r.Body = body
	p.ServeHTTP(w, r)
}

func session(r *http.Request, ps httprouter.Params) (types.RobustId, error) {
	var sessionid types.RobustId

	id, err := strconv.ParseInt(ps[0].Value, 0, 64)
	if err != nil {
		return sessionid, fmt.Errorf("invalid session: %v", err)
	}

	header := r.Header.Get("X-Session-Auth")
	if header == "" {
		return sessionid, fmt.Errorf("no X-Session-Auth header set")
	}

	auth, err := ircServer.GetAuth(types.RobustId{Id: id})
	if err != nil {
		return sessionid, err
	}
	if header != auth {
		return sessionid, fmt.Errorf("invalid X-Session-Auth header")
	}

	sessionid.Id = id

	return sessionid, nil
}

func sessionOrProxy(w http.ResponseWriter, r *http.Request, ps httprouter.Params) (types.RobustId, error) {
	sessionid, err := session(r, ps)
	if err == ircserver.ErrSessionNotYetSeen && node.State() != raft.Leader {
		// The session might exist on the leader, so we must proxy.
		maybeProxyToLeader(w, r, r.Body)
		return sessionid, err
	}

	if err != nil {
		http.Error(w, err.Error(), http.StatusNotFound)
	}
	return sessionid, err
}

// handlePostMessage is called by the robustirc-brigde whenever a message should be
// posted. The handler blocks until either the data was written or an error
// occurred. If successful, it returns the unique id of the message.
func handlePostMessage(w http.ResponseWriter, r *http.Request, ps httprouter.Params) {
	session, err := sessionOrProxy(w, r, ps)
	if err != nil {
		return
	}

	// Don’t throttle server-to-server connections (services)
	if netConfig.PostMessageCooloff > 0 {
		until := ircServer.ThrottleUntil(session, time.Duration(netConfig.PostMessageCooloff))
		time.Sleep(until.Sub(time.Now()))
	}

	type postMessageRequest struct {
		Data            string
		ClientMessageId uint64
	}

	var req postMessageRequest

	// We limit the amount of bytes read to 1024 to prevent reading overly long
	// requests in the first place. The IRC line length limit is 512 bytes, so
	// with 1024 bytes we have plenty of headroom to encode 512 bytes in JSON.
	//
	// We save a copy of the request in case we need to proxy it to the leader.
	var body bytes.Buffer
	rd := io.TeeReader(http.MaxBytesReader(w, r.Body, 1024), &body)
	if err := json.NewDecoder(rd).Decode(&req); err != nil {
		http.Error(w, err.Error(), http.StatusBadRequest)
		return
	}

	// If we have already seen this message, we just reply with a canned response.
	if id, reply := ircServer.LastPostMessage(session); id == req.ClientMessageId && reply != nil {
		w.Header().Set("Content-Type", "application/json")
		w.Write(reply)
		return
	}

	applyMu.Lock()
	msg := ircServer.NewRobustMessage(types.RobustIRCFromClient, session, req.Data)
	msg.ClientMessageId = req.ClientMessageId
	msgbytes, err := json.Marshal(msg)
	if err != nil {
		applyMu.Unlock()
		http.Error(w, fmt.Sprintf("Could not store message, cannot encode it as JSON: %v", err),
			http.StatusBadRequest)
		return
	}

	f := node.Apply(msgbytes, 10*time.Second)
	applyMu.Unlock()
	if err := f.Error(); err != nil {
		if err == raft.ErrNotLeader {
			maybeProxyToLeader(w, r, nopCloser{&body})
			return
		}
		http.Error(w, fmt.Sprintf("Apply(): %v", err), http.StatusInternalServerError)
		return
	}
}

func handleJoin(w http.ResponseWriter, r *http.Request) {
	log.Println("Join request from", r.RemoteAddr)
	if node.State() != raft.Leader {
		maybeProxyToLeader(w, r, r.Body)
		return
{'fraction_non_alphanumeric': 0.12906538856555974, 'fraction_numerical': 0.007702841492639507, 'mean_word_length': 4.06941431670282, 'pattern_counts': {'":': 0, '<': 3, '<?xml version=': 0, '>': 5, 'https://': 2, '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': '1123316', 'n_tokens_mistral': 4290, 'n_tokens_neox': 4031, 'n_words': 1085}