<commit_before>package batch

import (
	"bufio"
	"errors"
	"io"
	"io/ioutil"
	"mime"
	"mime/multipart"
	"net/http"
	"net/textproto"
	"strings"
	"time"

	"github.com/8legd/RRP/processors"
)

func MultipartMixed(w http.ResponseWriter, r *http.Request) {
	var batch []*http.Request
	ct, params, err := mime.ParseMediaType(r.Header.Get("Content-Type"))
	if err != nil {
		http.Error(w, err.Error(), http.StatusBadRequest)
		return
	}
	if ct != "multipart/mixed" {
		err = errors.New("unsupported content type, expected multipart/mixed")
		http.Error(w, err.Error(), http.StatusUnsupportedMediaType)
		return
	}
	// check for optional timeout header

	tm := r.Header.Get("x-batchproxy-timeout")
	var timeout time.Duration
	if tm != "" {
		timeout, err = time.ParseDuration(tm + "s")
		if err != nil {
			http.Error(w, "invalid value for x-batchproxy-timeout header, expected number of seconds", http.StatusBadRequest)
			return
		}
	} else {
		timeout = time.Duration(20) * time.Second // Default timeout is 20 seconds
	}
	boundary, ok := params["boundary"]
	if !ok {
		err = errors.New("missing multipart boundary")
		http.Error(w, err.Error(), http.StatusBadRequest)
		return
	}
	mr := multipart.NewReader(r.Body, boundary)
	for {
		p, err := mr.NextPart()
		if err == io.EOF {
			if p == nil {
				err = errors.New("invalid multipart content")
				http.Error(w, err.Error(), http.StatusBadRequest)
				return
			}
			break
		}
		if err != nil {
			http.Error(w, err.Error(), http.StatusInternalServerError)
			return
		}
		pct, _, err := mime.ParseMediaType(p.Header.Get("Content-Type"))
		if err != nil {
			http.Error(w, err.Error(), http.StatusInternalServerError)
			return
		}
		if pct != "application/http" {
			err = errors.New("unsupported content type for multipart/mixed content, expected each part to be application/http")
			http.Error(w, err.Error(), http.StatusBadRequest)
			return
		}
		r, err := http.ReadRequest(bufio.NewReader(p))
		// We need to get the protocol from a header in the part's request
		protocol := r.Header.Get("Forwarded")
		if protocol == "" || !strings.Contains(protocol, "proto=http") { // proto must be `http` or `https`
			err = errors.New("missing header in multipart/mixed content, expected each part to contain a Forwarded header with a valid proto value (proto=http or proto=https)")
			http.Error(w, err.Error(), http.StatusBadRequest)
			return
		}
		parts := strings.Split(protocol, "proto=")
		if len(parts) < 2 || (parts[1] != "http" && parts[1] != "https") {
			err = errors.New("invalid proto value in Forwarded header, expected proto=http or proto=https")
			http.Error(w, err.Error(), http.StatusBadRequest)
			return
		}
		protocol = parts[1]
		url := protocol + "://" + r.Host + r.RequestURI
		request, err := http.NewRequest(r.Method, url, r.Body)
		if err != nil {
			http.Error(w, err.Error(), http.StatusInternalServerError)
		}
		batch = append(batch, request)
	}
	responses, err := processors.ProcessBatch(batch, timeout)
	if err != nil {
		http.Error(w, err.Error(), http.StatusInternalServerError)
	}

	mw := multipart.NewWriter(w)
	defer mw.Close()
	w.Header().Set("Content-Type", "multipart/mixed; boundary="+mw.Boundary())

	var pw io.Writer
	var pb []byte

	for _, next := range responses {
		ph := make(textproto.MIMEHeader)
		ph.Set("Content-Type", "application/http")
		pw, err = mw.CreatePart(ph)
		if err != nil {
			http.Error(w, err.Error(), http.StatusInternalServerError)
		}
		io.WriteString(pw, next.Proto+" "+next.Status+"\n")
		if next.Header != nil {
			next.Header.Write(pw)
			io.WriteString(pw, "\n")
		}
		if next.Body != nil {
			pb, err = ioutil.ReadAll(next.Body)
			if err != nil {
				http.Error(w, err.Error(), http.StatusInternalServerError)
			}
			pw.Write(pb)
			io.WriteString(pw, "\n")
		}
	}
}<commit_msg>Continued work on initial release<commit_after>package batch

import (
	"bufio"
	"errors"
	"io"
	"io/ioutil"
	"log"
	"mime"
	"mime/multipart"
	"net/http"
	"net/textproto"
	"strings"
	"time"

	"github.com/8legd/RRP/processors"
)

func MultipartMixed(w http.ResponseWriter, r *http.Request) {
	var batch []*http.Request
	ct, params, err := mime.ParseMediaType(r.Header.Get("Content-Type"))
	if err != nil {
		log.Println(err)
		http.Error(w, err.Error(), http.StatusBadRequest)
		return
	}
	if ct != "multipart/mixed" {
		err = errors.New("unsupported content type, expected multipart/mixed")
		log.Println(err)
		http.Error(w, err.Error(), http.StatusUnsupportedMediaType)
		return
	}
	// check for optional timeout header

	tm := r.Header.Get("x-batchproxy-timeout")
	var timeout time.Duration
	if tm != "" {
		timeout, err = time.ParseDuration(tm + "s")
		if err != nil {
			log.Println(err)
			http.Error(w, "invalid value for x-batchproxy-timeout header, expected number of seconds", http.StatusBadRequest)
			return
		}
	} else {
		timeout = time.Duration(20) * time.Second // Default timeout is 20 seconds
	}
	boundary, ok := params["boundary"]
	if !ok {
		err = errors.New("missing multipart boundary")
		log.Println(err)
		http.Error(w, err.Error(), http.StatusBadRequest)
		return
	}
	mr := multipart.NewReader(r.Body, boundary)
	for {
		p, err := mr.NextPart()
		if err == io.EOF {
			if p == nil {
				err = errors.New("invalid multipart content")
				log.Println(err)
				http.Error(w, err.Error(), http.StatusBadRequest)
				return
			}
			break
		}
		if err != nil {
			log.Println(err)
			http.Error(w, err.Error(), http.StatusInternalServerError)
			return
		}
		pct, _, err := mime.ParseMediaType(p.Header.Get("Content-Type"))
		if err != nil {
			log.Println(err)
			http.Error(w, err.Error(), http.StatusInternalServerError)
			return
		}
		if pct != "application/http" {
			err = errors.New("unsupported content type for multipart/mixed content, expected each part to be application/http")
			log.Println(err)
			http.Error(w, err.Error(), http.StatusBadRequest)
			return
		}
		r, err := http.ReadRequest(bufio.NewReader(p))
		// We need to get the protocol from a header in the part's request
		protocol := r.Header.Get("Forwarded")
		if protocol == "" || !strings.Contains(protocol, "proto=http") { // proto must be `http` or `https`
			err = errors.New("missing header in multipart/mixed content, expected each part to contain a Forwarded header with a valid proto value (proto=http or proto=https)")
			log.Println(err)
			http.Error(w, err.Error(), http.StatusBadRequest)
			return
		}
		parts := strings.Split(protocol, "proto=")
		if len(parts) < 2 || (parts[1] != "http" && parts[1] != "https") {
			err = errors.New("invalid proto value in Forwarded header, expected proto=http or proto=https")
			log.Println(err)
			http.Error(w, err.Error(), http.StatusBadRequest)
			return
		}
		protocol = parts[1]
		url := protocol + "://" + r.Host + r.RequestURI
		request, err := http.NewRequest(r.Method, url, r.Body)
		if err != nil {
			log.Println(err)
			http.Error(w, err.Error(), http.StatusInternalServerError)
		}
		batch = append(batch, request)
	}
	responses, err := processors.ProcessBatch(batch, timeout)
	if err != nil {
		log.Println(err)
		http.Error(w, err.Error(), http.StatusInternalServerError)
	}

	mw := multipart.NewWriter(w)
	defer mw.Close()
	w.Header().Set("Content-Type", "multipart/mixed; boundary="+mw.Boundary())

	var pw io.Writer
	var pb []byte

	for _, next := range responses {
		ph := make(textproto.MIMEHeader)
		ph.Set("Content-Type", "application/http")
		pw, err = mw.CreatePart(ph)
		if err != nil {
			log.Println(err)
			http.Error(w, err.Error(), http.StatusInternalServerError)
		}
		io.WriteString(pw, next.Proto+" "+next.Status+"\n")
		if next.Header != nil {
			next.Header.Write(pw)
			io.WriteString(pw, "\n")
		}
		if next.Body != nil {
			pb, err = ioutil.ReadAll(next.Body)
			if err != nil {
				log.Println(err)
				http.Error(w, err.Error(), http.StatusInternalServerError)
			}
			pw.Write(pb)
			io.WriteString(pw, "\n")
		}
	}
}
{'fraction_non_alphanumeric': 0.15374889366544442, 'fraction_numerical': 0.0022758882286003286, 'mean_word_length': 3.7111375818939845, 'pattern_counts': {'":': 2, '<': 5, '<?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': '905858', 'n_tokens_mistral': 3061, 'n_tokens_neox': 2765, 'n_words': 661}