<commit_before>package api

import (
	"context"
	"encoding/json"
	"fmt"
	"log"
	"net/http"
	"strconv"
	"strings"
	"time"

	"github.com/hashicorp/raft"
	"github.com/robustirc/robustirc/internal/ircserver"
	"github.com/robustirc/robustirc/internal/outputstream"
	"github.com/robustirc/robustirc/internal/robust"
	"github.com/stapelberg/glog"
)

func (api *HTTP) updateLastContact() {
	r := api.raftNode
	// node.LastContact() is only updated when we receive heartbeats from the
	// leader, i.e. only when we are a follower.
	if r.State() == raft.Follower && !r.LastContact().IsZero() {
		lastContact = r.LastContact()
	} else if r.State() == raft.Leader {
		lastContact = time.Now()
	}
}

func (api *HTTP) pingMessage() *robust.Message {
	peers, err := api.peerStore.Peers()
	if err != nil {
		log.Fatalf("Could not get peers: %v (Peer file corrupted on disk?)", err)
	}
	return &robust.Message{
		Type:    robust.Ping,
		Servers: peers,
	}
}

func parseLastSeen(lastSeenStr string) (first uint64, last uint64, err error) {
	parts := strings.Split(lastSeenStr, ".")
	if got, want := len(parts), 2; got != want {
		return 0, 0, fmt.Errorf("Unexpected number of parts: got %d, want %d", got, want)
	}
	first, err = strconv.ParseUint(parts[0], 0, 64)
	if err != nil {
		return 0, 0, err
	}
	last, err = strconv.ParseUint(parts[1], 0, 64)
	if err != nil {
		return 0, 0, err
	}
	return first, last, nil
}

func (api *HTTP) pingTicker(ctx context.Context, msgschan chan<- []*robust.Message) {
	ticker := time.NewTicker(pingInterval)
	defer ticker.Stop()
	for {
		select {
		case <-ticker.C:
			msgschan <- []*robust.Message{api.pingMessage()}
		case <-ctx.Done():
			return
		}
	}
}

func outputToRobustMessages(msgs []outputstream.Message) []*robust.Message {
	result := make([]*robust.Message, len(msgs))
	for idx, msg := range msgs {
		result[idx] = &robust.Message{
			Id:             msg.Id,
			Type:           robust.IRCToClient,
			Data:           msg.Data,
			InterestingFor: msg.InterestingFor,
		}
	}
	return result
}

func (api *HTTP) getMessages(ctx context.Context, lastSeen robust.Id, msgschan chan<- []*robust.Message) {
	var msgs []outputstream.Message

	// With the following output messages stored for a session:
	// Id                  Reply
	// 1431542836610113945.1
	// 1431542836610113945.2     |lastSeen|
	// 1431542836610113945.3
	// 1431542836691955391.1
	// …when resuming, GetNext(1431542836610113945.2) will return
	// 1431542836691955391.*, skipping the remaining messages with
	// Id=1431542836610113945.
	// Hence, we need to Get(1431542836610113945.2) to send
	// 1431542836610113945.3 and following to the client.
	if msgs, ok := api.output.Get(lastSeen); ok && int(lastSeen.Reply) < len(msgs) {
		select {
		case <-ctx.Done():
			return
		case msgschan <- outputToRobustMessages(msgs[lastSeen.Reply:]):
		}
	}

	for {
		if msgs = api.output.GetNext(ctx, lastSeen); len(msgs) == 0 {
			if ctx.Err() != nil {
				return
			}
			continue
		}
		// This check prevents replaying old messages in the scenario where
		// the client has seen newer messages than the server, for example
		// because the server is currently recovering from a snapshot after
		// being restarted, or because the server’s network connection to
		// the rest of the network is currently slow.
		if msgs[0].Id.Id < lastSeen.Id {
			glog.Warningf("lastSeen (%d) more recent than GetNext() result %d\n", lastSeen.Id, msgs[0].Id.Id)
			glog.Warningf("This should only happen while the server is recovering from a snapshot\n")
			glog.Warningf("The message in question is %v\n", msgs[0])
			// Prevent busylooping while new messages are applied.
			time.Sleep(250 * time.Millisecond)
			continue
		}

		lastSeen = msgs[0].Id
		select {
		case <-ctx.Done():
			return
		case msgschan <- outputToRobustMessages(msgs):
		}
	}
}

func (api *HTTP) handleGetMessages(w http.ResponseWriter, r *http.Request, sessionId string) {
	// Avoid sessionOrProxy() because GetMessages can be answered on any raft
	// node, it’s a read-only request.
	session, err := api.session(r, sessionId)
	if err != nil {
		if err == ircserver.ErrSessionNotYetSeen {
			http.Error(w, err.Error(), http.StatusInternalServerError)
		} else {
			http.Error(w, err.Error(), http.StatusNotFound)
		}
		return
	}

	lastSeen := session
	if ls := r.FormValue("lastseen"); ls != "0.0" && ls != "" {
		first, last, err := parseLastSeen(ls)
		if err != nil {
			log.Printf("cannot parse %q: %v\n", ls, err)
			http.Error(w, fmt.Sprintf("Malformed lastseen value (%q)", ls),
				http.StatusInternalServerError)
			return
		}
		lastSeen = robust.Id{
			Id:    first,
			Reply: last,
		}
		log.Printf("Trying to resume at %v\n", lastSeen)
	}

	enc := json.NewEncoder(w)
	flushTimer := time.NewTimer(1 * time.Second)
	flushTimer.Stop()
	var lastFlush time.Time
	willFlush := false
	msgschan := make(chan []*robust.Message)

	sessionId = strconv.FormatUint(session.Id, 10)
	ctx, cancel := context.WithCancel(r.Context())
	// Cancel the helper goroutines we are about to start when this
	// request handler returns.
	wasSuperseded := false
	cancelAll := func(superseded bool) {
		if superseded {
			// cancelAll will be called after the context cancellation
			// via defer, so save that this GetMessages request was
			// superseded.
			wasSuperseded = true
		}
		cancel()
		// Wake up the getMessages goroutine from its GetNext call
		// to quickly free up resources.
		api.output.InterruptGetNext()

		if !wasSuperseded {
			api.deleteGetMessagesRequests(sessionId)
		}
	}<commit_msg>api: /messages: acknowledge request immediately<commit_after>package api

import (
	"context"
	"encoding/json"
	"fmt"
	"log"
	"net/http"
	"strconv"
	"strings"
	"time"

	"github.com/hashicorp/raft"
	"github.com/robustirc/robustirc/internal/ircserver"
	"github.com/robustirc/robustirc/internal/outputstream"
	"github.com/robustirc/robustirc/internal/robust"
	"github.com/stapelberg/glog"
)

func (api *HTTP) updateLastContact() {
	r := api.raftNode
	// node.LastContact() is only updated when we receive heartbeats from the
	// leader, i.e. only when we are a follower.
	if r.State() == raft.Follower && !r.LastContact().IsZero() {
		lastContact = r.LastContact()
	} else if r.State() == raft.Leader {
		lastContact = time.Now()
	}
}

func (api *HTTP) pingMessage() *robust.Message {
	peers, err := api.peerStore.Peers()
	if err != nil {
		log.Fatalf("Could not get peers: %v (Peer file corrupted on disk?)", err)
	}
	return &robust.Message{
		Type:    robust.Ping,
		Servers: peers,
	}
}

func parseLastSeen(lastSeenStr string) (first uint64, last uint64, err error) {
	parts := strings.Split(lastSeenStr, ".")
	if got, want := len(parts), 2; got != want {
		return 0, 0, fmt.Errorf("Unexpected number of parts: got %d, want %d", got, want)
	}
	first, err = strconv.ParseUint(parts[0], 0, 64)
	if err != nil {
		return 0, 0, err
	}
	last, err = strconv.ParseUint(parts[1], 0, 64)
	if err != nil {
		return 0, 0, err
	}
	return first, last, nil
}

func (api *HTTP) pingTicker(ctx context.Context, msgschan chan<- []*robust.Message) {
	ticker := time.NewTicker(pingInterval)
	defer ticker.Stop()
	for {
		select {
		case <-ticker.C:
			msgschan <- []*robust.Message{api.pingMessage()}
		case <-ctx.Done():
			return
		}
	}
}

func outputToRobustMessages(msgs []outputstream.Message) []*robust.Message {
	result := make([]*robust.Message, len(msgs))
	for idx, msg := range msgs {
		result[idx] = &robust.Message{
			Id:             msg.Id,
			Type:           robust.IRCToClient,
			Data:           msg.Data,
			InterestingFor: msg.InterestingFor,
		}
	}
	return result
}

func (api *HTTP) getMessages(ctx context.Context, lastSeen robust.Id, msgschan chan<- []*robust.Message) {
	var msgs []outputstream.Message

	// With the following output messages stored for a session:
	// Id                  Reply
	// 1431542836610113945.1
	// 1431542836610113945.2     |lastSeen|
	// 1431542836610113945.3
	// 1431542836691955391.1
	// …when resuming, GetNext(1431542836610113945.2) will return
	// 1431542836691955391.*, skipping the remaining messages with
	// Id=1431542836610113945.
	// Hence, we need to Get(1431542836610113945.2) to send
	// 1431542836610113945.3 and following to the client.
	if msgs, ok := api.output.Get(lastSeen); ok && int(lastSeen.Reply) < len(msgs) {
		select {
		case <-ctx.Done():
			return
		case msgschan <- outputToRobustMessages(msgs[lastSeen.Reply:]):
		}
	}

	for {
		if msgs = api.output.GetNext(ctx, lastSeen); len(msgs) == 0 {
			if ctx.Err() != nil {
				return
			}
			continue
		}
		// This check prevents replaying old messages in the scenario where
		// the client has seen newer messages than the server, for example
		// because the server is currently recovering from a snapshot after
		// being restarted, or because the server’s network connection to
		// the rest of the network is currently slow.
		if msgs[0].Id.Id < lastSeen.Id {
			glog.Warningf("lastSeen (%d) more recent than GetNext() result %d\n", lastSeen.Id, msgs[0].Id.Id)
			glog.Warningf("This should only happen while the server is recovering from a snapshot\n")
			glog.Warningf("The message in question is %v\n", msgs[0])
			// Prevent busylooping while new messages are applied.
			time.Sleep(250 * time.Millisecond)
			continue
		}

		lastSeen = msgs[0].Id
		select {
		case <-ctx.Done():
			return
		case msgschan <- outputToRobustMessages(msgs):
		}
	}
}

func (api *HTTP) handleGetMessages(w http.ResponseWriter, r *http.Request, sessionId string) {
	// Avoid sessionOrProxy() because GetMessages can be answered on any raft
	// node, it’s a read-only request.
	session, err := api.session(r, sessionId)
	if err != nil {
		if err == ircserver.ErrSessionNotYetSeen {
			http.Error(w, err.Error(), http.StatusInternalServerError)
		} else {
			http.Error(w, err.Error(), http.StatusNotFound)
		}
		return
	}

	lastSeen := session
	if ls := r.FormValue("lastseen"); ls != "0.0" && ls != "" {
		first, last, err := parseLastSeen(ls)
		if err != nil {
			log.Printf("cannot parse %q: %v\n", ls, err)
			http.Error(w, fmt.Sprintf("Malformed lastseen value (%q)", ls),
				http.StatusInternalServerError)
			return
		}
		lastSeen = robust.Id{
			Id:    first,
			Reply: last,
		}
		log.Printf("Trying to resume at %v\n", lastSeen)
	}

	// Acknowledge the request even if no messages need to be sent right now:
	w.Header().Set("Content-Type", "text/plain; charset=utf-8")
	w.WriteHeader(http.StatusOK)
	if f, ok := w.(http.Flusher); ok {
		f.Flush()
	}

	enc := json.NewEncoder(w)
	flushTimer := time.NewTimer(1 * time.Second)
	flushTimer.Stop()
	var lastFlush time.Time
	willFlush := false
	msgschan := make(chan []*robust.Message)

	sessionId = strconv.FormatUint(session.Id, 10)
	ctx, cancel := context.WithCancel(r.Context())
	// Cancel the helper goroutines we are about to start when this
	// request handler returns.
	wasSuperseded := false
	cancelAll := func(superseded bool) {
		if superseded {
			// cancelAll will be called after the context cancellation
			// via defer, so save that this GetMessages request was
			// superseded.
			wasSuperseded = true
		}
		cancel()
		// Wake up the getMessages goroutine from its GetNext call
		// to quickly free up resources.
		api.output.InterruptGetNext()

		if !wasSuperseded {
			api.deleteGetMessagesRequests(sessionId)
		}
	}
{'fraction_non_alphanumeric': 0.13297402367909525, 'fraction_numerical': 0.03719738469694292, 'mean_word_length': 3.5113591072140293, 'pattern_counts': {'":': 0, '<': 25, '<?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': '5600766', 'n_tokens_mistral': 4577, 'n_tokens_neox': 3981, 'n_words': 1079}