ev/app/peerfinder/service.go

package peerfinder

import (
	"context"
	"embed"
	"encoding/json"
	"html/template"
	"io"
	"io/fs"
	"log"
	"net"
	"net/http"
	"strconv"
	"strings"

	contentnegotiation "gitlab.com/jamietanna/content-negotiation-go"
	"go.opentelemetry.io/otel/attribute"

	"github.com/oklog/ulid"
	"github.com/sour-is/ev/internal/lg"
	"github.com/sour-is/ev/pkg/es"
	"github.com/sour-is/ev/pkg/es/event"
	"github.com/sour-is/ev/pkg/locker"
)

const (
	aggInfo        = "pf-info"
	queueRequests  = "pf-requests"
	queueResponses = "pf-request-"
	queuePeers     = "pf-peer-"
	initVersion    = "1.1.0"
)

var (
	//go:embed pages/* layouts/* assets/*
	files     embed.FS
	templates map[string]*template.Template
)

type service struct {
	es *es.EventStore

	State locker.Locked[state]
}

type state struct {
	Version  string
	Requests []Request
}

func New(ctx context.Context, es *es.EventStore) (*service, error) {
	ctx, span := lg.Span(ctx)
	defer span.End()

	if err := event.Register(ctx, &RequestSubmitted{}, &ResultSubmitted{}, &VersionChanged{}); err != nil {
		span.RecordError(err)
		return nil, err
	}

	svc := &service{es: es}

	return svc, nil
}
func (s *service) RegisterHTTP(mux *http.ServeMux) {
	loadTemplates()
	a, err := fs.Sub(files, "assets")
	log.Println(err)
	assets := http.StripPrefix("/peers/assets/", http.FileServer(http.FS(a)))

	mux.Handle("/peers/assets/", lg.Htrace(assets, "peer-assets"))
	mux.Handle("/peers/", lg.Htrace(s, "peers"))
}
func (s *service) ServeHTTP(w http.ResponseWriter, r *http.Request) {
	ctx := r.Context()

	_, span := lg.Span(ctx)
	defer span.End()

	r = r.WithContext(ctx)

	switch r.Method {
	case http.MethodGet:
		switch {
		case strings.HasPrefix(r.URL.Path, "/peers/pending/"):
			s.getPending(w, r, strings.TrimPrefix(r.URL.Path, "/peers/pending/"))
			return

		case strings.HasPrefix(r.URL.Path, "/peers/req/"):
			s.getResults(w, r, strings.TrimPrefix(r.URL.Path, "/peers/req/"))
			return

		default:
			t := templates["home.tpl"]
			t.Execute(w, nil)
			return
		}
	case http.MethodPost:
		switch {
		case strings.HasPrefix(r.URL.Path, "/peers/req/"):
			s.postResult(w, r, strings.TrimPrefix(r.URL.Path, "/peers/req/"))
			return

		case strings.HasPrefix(r.URL.Path, "/peers/req"):
			s.postRequest(w, r)
			return

		default:
			w.WriteHeader(http.StatusNotFound)
			return
		}
	default:
		w.WriteHeader(http.StatusMethodNotAllowed)
		return
	}
}

func (s *service) getPending(w http.ResponseWriter, r *http.Request, uuid string) {
	ctx, span := lg.Span(r.Context())
	defer span.End()

	span.SetAttributes(
		attribute.String("uuid", uuid),
	)

	info, err := es.Upsert(ctx, s.es, "pf-info", func(ctx context.Context, agg *Info) error {
		return agg.OnCreate() // initialize if not exists
	})
	if err != nil {
		span.RecordError(err)
		w.WriteHeader(http.StatusInternalServerError)
		return
	}

	requests, err := s.es.Read(ctx, queueRequests, -1, -30)
	if err != nil {
		span.RecordError(err)
		w.WriteHeader(http.StatusInternalServerError)
		return
	}

	responses, err := s.es.Read(ctx, queuePeers+uuid, -1, -30)
	if err != nil {
		span.RecordError(err)
		w.WriteHeader(http.StatusInternalServerError)
		return
	}

	req := filter(requests, responses)

	negotiator := contentnegotiation.NewNegotiator("application/json", "text/environment", "text/plain", "text/html")
	negotiated, _, err := negotiator.Negotiate(r.Header.Get("Accept"))
	if err != nil {
		span.RecordError(err)
		w.WriteHeader(http.StatusNotAcceptable)
		return
	}

	span.AddEvent(negotiated.String())
	mime := negotiated.String()
	switch mime {
	case "text/environment":
		w.Header().Set("content-type", negotiated.String())
		_, err = encodeTo(w, info.MarshalEnviron, req.MarshalEnviron)
	case "application/json":
		w.Header().Set("content-type", negotiated.String())
		var out interface{} = info
		if req != nil {
			out = struct {
				ScriptVersion string `json:"script_version"`
				RequestID     string `json:"req_id"`
				RequestIP     string `json:"req_ip"`
				Family        string `json:"req_family"`
				Created       string `json:"req_created"`
			}{
				info.ScriptVersion,
				req.RequestID(),
				req.RequestIP,
				strconv.Itoa(req.Family()),
				req.CreatedString(),
			}
		}
		err = json.NewEncoder(w).Encode(out)
	}
	span.RecordError(err)
}
func (s *service) getResults(w http.ResponseWriter, r *http.Request, uuid string) {
	ctx, span := lg.Span(r.Context())
	defer span.End()

	span.SetAttributes(
		attribute.String("uuid", uuid),
	)

	responses, err := s.es.Read(ctx, queueResponses+uuid, -1, es.AllEvents)
	if err != nil {
		w.WriteHeader(http.StatusInternalServerError)
		return
	}

	negotiator := contentnegotiation.NewNegotiator("application/json", "text/environment", "text/csv", "text/plain", "text/html")
	negotiated, _, err := negotiator.Negotiate("application/json")
	if err != nil {
		w.WriteHeader(http.StatusNotAcceptable)
		return
	}
	switch negotiated.String() {
	// case "text/environment":
	// 	encodeTo(w, responses.MarshalBinary)
	case "application/json":
		json.NewEncoder(w).Encode(responses)
	}
}
func (s *service) postRequest(w http.ResponseWriter, r *http.Request) {
	ctx, span := lg.Span(r.Context())
	defer span.End()

	if err := r.ParseForm(); err != nil {
		w.WriteHeader(http.StatusUnprocessableEntity)
		return
	}

	ip := net.ParseIP(r.Form.Get("req_ip"))
	if ip == nil {
		w.WriteHeader(http.StatusBadRequest)
		return
	}
	req := &RequestSubmitted{
		RequestIP: ip.String(),
	}
	if hidden, err := strconv.ParseBool(r.Form.Get("req_hidden")); err != nil {
		req.Hidden = hidden
	}

	span.SetAttributes(
		attribute.Stringer("req_ip", ip),
	)

	s.es.Append(ctx, queueRequests, event.NewEvents(req))
}
func (s *service) postResult(w http.ResponseWriter, r *http.Request, id string) {
	ctx, span := lg.Span(r.Context())
	defer span.End()

	span.SetAttributes(
		attribute.String("id", id),
	)

	if _, err := ulid.ParseStrict(id); err != nil {
		w.WriteHeader(http.StatusNotFound)
		return
	}

	if err := r.ParseForm(); err != nil {
		w.WriteHeader(http.StatusUnprocessableEntity)
		return
	}

	latency, err := strconv.ParseFloat(r.Form.Get("res_latency"), 64)
	if err != nil {
		w.WriteHeader(http.StatusUnprocessableEntity)
		return
	}

	req := &ResultSubmitted{
		RequestID:   id,
		PeerID:      r.Form.Get("peer_id"),
		PeerVersion: r.Form.Get("peer_version"),
		Latency:     latency,
	}

	span.SetAttributes(
		attribute.Stringer("result", req),
	)

	idx, err := s.es.LastIndex(ctx, queueResponses+id)
	if err != nil {
		w.WriteHeader(http.StatusInternalServerError)
		return
	}
	if idx == 0 {
		w.WriteHeader(http.StatusNotFound)
		return
	}

	s.es.Append(ctx, queueRequests, event.NewEvents(req))
}

func filter(requests, responses event.Events) *RequestSubmitted {
	have := make(map[string]struct{}, len(responses))
	for _, res := range toList[ResultSubmitted](responses...) {
		have[res.RequestID] = struct{}{}
	}
	for _, req := range reverse(toList[RequestSubmitted](requests...)...) {
		if _, ok := have[req.RequestID()]; !ok {
			return req
		}
	}
	return nil
}
func toList[E any, T es.PE[E]](lis ...event.Event) []T {
	newLis := make([]T, 0, len(lis))
	for i := range lis {
		if e, ok := lis[i].(T); ok {
			newLis = append(newLis, e)
		}
	}
	return newLis
}
func reverse[T any](s ...T) []T {
	first, last := 0, len(s)-1
	for first < last {
		s[first], s[last] = s[last], s[first]
		first++
		last--
	}
	return s
}
func encodeTo(w io.Writer, fns ...func() ([]byte, error)) (int, error) {
	i := 0
	for _, fn := range fns {
		b, err := fn()
		if err != nil {
			return i, err
		}

		j, err := w.Write(b)
		i += j
		if err != nil {
			return i, err
		}
	}
	return i, nil
}
func loadTemplates() error {
	if templates != nil {
		return nil
	}
	templates = make(map[string]*template.Template)
	tmplFiles, err := fs.ReadDir(files, "pages")
	if err != nil {
		return err
	}

	for _, tmpl := range tmplFiles {
		if tmpl.IsDir() {
			continue
		}
		log.Println(tmpl.Name())
		pt, err := template.ParseFS(files, "pages/"+tmpl.Name(), "layouts/*.tpl")
		if err != nil {
			return err
		}

		templates[tmpl.Name()] = pt
	}
	return nil
}

func Projector(e event.Event) []event.Event {
	m := e.EventMeta()
	streamID := m.StreamID
	streamPos := m.Position

	switch e := e.(type) {
	case *RequestSubmitted:
		e1 := event.NewPtr(streamID, streamPos)
		event.SetStreamID(queueResponses+e.RequestID(), e1)

		return []event.Event{e1}
	case *ResultSubmitted:
		e1 := event.NewPtr(streamID, streamPos)
		event.SetStreamID(queueResponses+e.RequestID, e1)

		e2 := event.NewPtr(streamID, streamPos)
		event.SetStreamID(queuePeers+e.PeerID, e2)

		return []event.Event{e1, e2}
	}
	return nil
}