chore: add apps from go.sour.is/ev

app/peerfinder/assets/peerfinder.css

@@ -0,0 +1,58 @@
+/* Space out content a bit */
+body {
+  padding-top: 20px;
+  padding-bottom: 20px;
+}
+
+/* Everything but the jumbotron gets side spacing for mobile first views */
+.header,
+.footer {
+  padding-right: 15px;
+  padding-left: 15px;
+}
+
+/* Custom page header */
+.header {
+  padding-bottom: 20px;
+  border-bottom: 1px solid #e5e5e5;
+}
+/* Make the masthead heading the same height as the navigation */
+.header h3 {
+  margin-top: 0;
+  margin-bottom: 0;
+  line-height: 40px;
+}
+
+/* Custom page footer */
+.footer {
+  padding-top: 19px;
+  color: #777;
+  border-top: 1px solid #e5e5e5;
+}
+
+.panel-heading a { 
+  color: white;
+  font-weight: bold;
+}
+
+.container-narrow > hr {
+  margin: 30px 0;
+}
+
+.table tbody tr th {
+  width: 70%
+}
+
+@media (prefers-color-scheme: dark) {
+  body, .panel-body {
+    color: white;
+    background-color: #121212;
+  }
+  .table-striped > tbody > tr:nth-of-type(2n+1) {
+    background-color: darkslategray;
+  }
+}
+
+@media (prefers-color-scheme: light) {
+
+}

app/peerfinder/ev-info.go

@@ -0,0 +1,56 @@
+package peerfinder
+
+import (
+	"bytes"
+
+	"github.com/tj/go-semver"
+
+	"go.sour.is/ev/event"
+)
+
+type Info struct {
+	ScriptVersion string `json:"script_version"`
+
+	event.IsAggregate
+}
+
+var _ event.Aggregate = (*Info)(nil)
+
+func (a *Info) ApplyEvent(lis ...event.Event) {
+	for _, e := range lis {
+		switch e := e.(type) {
+		case *VersionChanged:
+			a.ScriptVersion = e.ScriptVersion
+		}
+	}
+}
+func (a *Info) MarshalEnviron() ([]byte, error) {
+	var b bytes.Buffer
+
+	b.WriteString("SCRIPT_VERSION=")
+	b.WriteString(a.ScriptVersion)
+	b.WriteRune('\n')
+
+	return b.Bytes(), nil
+}
+func (a *Info) OnUpsert() error {
+	if a.StreamVersion() == 0 {
+		event.Raise(a, &VersionChanged{ScriptVersion: initVersion})
+	}
+	current, _ := semver.Parse(initVersion)
+	previous, _ := semver.Parse(a.ScriptVersion)
+
+	if current.Compare(previous) > 0 {
+		event.Raise(a, &VersionChanged{ScriptVersion: initVersion})
+	}
+
+	return nil
+}
+
+type VersionChanged struct {
+	ScriptVersion string `json:"script_version"`
+
+	event.IsEvent
+}
+
+var _ event.Event = (*VersionChanged)(nil)

app/peerfinder/ev-peer.go

@@ -0,0 +1,111 @@
+package peerfinder
+
+import (
+	"net"
+	"strconv"
+	"strings"
+	"time"
+
+	"github.com/keys-pub/keys/json"
+	"go.sour.is/pkg/set"
+
+	"go.sour.is/ev/event"
+)
+
+type Time time.Time
+
+func (t *Time) UnmarshalJSON(b []byte) error {
+	time, err := time.Parse(`"2006-01-02 15:04:05"`, string(b))
+	*t = Time(time)
+	return err
+}
+func (t *Time) MarshalJSON() ([]byte, error) {
+	if t == nil {
+		return nil, nil
+	}
+	i := *t
+	return time.Time(i).MarshalJSON()
+}
+
+type ipFamily string
+
+const (
+	ipFamilyV4   ipFamily = "IPv4"
+	ipFamilyV6   ipFamily = "IPv6"
+	ipFamilyBoth ipFamily = "both"
+	ipFamilyNone ipFamily = "none"
+)
+
+func (t *ipFamily) UnmarshalJSON(b []byte) error {
+	i, err := strconv.Atoi(strings.Trim(string(b), `"`))
+	switch i {
+	case 1:
+		*t = ipFamilyV4
+	case 2:
+		*t = ipFamilyV6
+	case 3:
+		*t = ipFamilyBoth
+	default:
+		*t = ipFamilyNone
+	}
+	return err
+}
+
+type peerType []string
+
+func (t *peerType) UnmarshalJSON(b []byte) error {
+	var bs string
+	json.Unmarshal(b, &bs)
+	*t = strings.Split(bs, ",")
+	return nil
+}
+
+type Peer struct {
+	ID      string   `json:"peer_id,omitempty"`
+	Owner   string   `json:"peer_owner"`
+	Nick    string   `json:"peer_nick"`
+	Name    string   `json:"peer_name"`
+	Country string   `json:"peer_country"`
+	Note    string   `json:"peer_note"`
+	Family  ipFamily `json:"peer_family"`
+	Type    peerType `json:"peer_type"`
+	Created Time     `json:"peer_created"`
+}
+
+func (p *Peer) CanSupport(ip string) bool {
+	addr := net.ParseIP(ip)
+	if addr == nil {
+		return false
+	}
+	if !(addr.IsGlobalUnicast() || addr.IsLoopback() || addr.IsPrivate()) {
+		return false
+	}
+
+	switch p.Family {
+	case ipFamilyV4:
+		return addr.To4() != nil
+	case ipFamilyV6:
+		return addr.To16() != nil
+	case ipFamilyNone:
+		return false
+	}
+
+	return true
+}
+
+type PeerResults struct {
+	set.Set[string]
+	event.IsAggregate
+}
+
+func (p *PeerResults) ApplyEvent(lis ...event.Event) {
+	for _, e := range lis {
+		switch e := e.(type) {
+		case *ResultSubmitted:
+			if p.Set == nil {
+				p.Set = set.New[string]()
+			}
+			p.Set.Add(e.RequestID)
+		}
+	}
+}

app/peerfinder/ev-request.go

@@ -0,0 +1,230 @@
+package peerfinder
+
+import (
+	"bytes"
+	"encoding/json"
+	"fmt"
+	"net/netip"
+	"strconv"
+	"time"
+
+	"github.com/oklog/ulid"
+	"go.sour.is/ev/event"
+	"go.sour.is/pkg/set"
+)
+
+type Request struct {
+	event.IsAggregate
+
+	RequestID string    `json:"req_id"`
+	RequestIP string    `json:"req_ip"`
+	Hidden    bool      `json:"hide,omitempty"`
+	Created   time.Time `json:"req_created"`
+	Family    int       `json:"family"`
+
+	Responses []*Response `json:"responses"`
+	peers     set.Set[string]
+	initial   *RequestSubmitted
+}
+
+var _ event.Aggregate = (*Request)(nil)
+
+func (a *Request) ApplyEvent(lis ...event.Event) {
+	for _, e := range lis {
+		switch e := e.(type) {
+		case *RequestSubmitted:
+			a.RequestID = e.EventMeta().EventID.String()
+			a.RequestIP = e.RequestIP
+			a.Hidden = e.Hidden
+			a.Created = ulid.Time(e.EventMeta().EventID.Time())
+			a.Family = e.Family()
+
+			a.initial = e
+		case *ResultSubmitted:
+			if a.peers == nil {
+				a.peers = set.New[string]()
+			}
+			if a.peers.Has(e.PeerID) {
+				continue
+			}
+
+			a.peers.Add(e.PeerID)
+			a.Responses = append(a.Responses, &Response{
+				PeerID:        e.PeerID,
+				ScriptVersion: e.PeerVersion,
+				Latency:       e.Latency,
+				Jitter:        e.Jitter,
+				MinRTT:        e.MinRTT,
+				MaxRTT:        e.MaxRTT,
+				Sent:          e.Sent,
+				Received:      e.Received,
+				Unreachable:   e.Unreachable,
+				Created:       ulid.Time(e.EventMeta().EventID.Time()),
+			})
+		}
+	}
+}
+
+func (a *Request) MarshalEnviron() ([]byte, error) {
+	return a.initial.MarshalEnviron()
+}
+func (a *Request) CreatedString() string {
+	return a.Created.Format("2006-01-02 15:04:05")
+}
+
+type ListRequest []*Request
+
+func (lis ListRequest) Len() int {
+	return len(lis)
+}
+func (lis ListRequest) Less(i, j int) bool {
+	return lis[i].Created.Before(lis[j].Created)
+}
+func (lis ListRequest) Swap(i, j int) {
+	lis[i], lis[j] = lis[j], lis[i]
+}
+
+type Response struct {
+	Peer          *Peer  `json:"peer"`
+	PeerID        string `json:"-"`
+	ScriptVersion string `json:"peer_scriptver"`
+
+	Latency     float64 `json:"res_latency"`
+	Jitter      float64 `json:"res_jitter,omitempty"`
+	MaxRTT      float64 `json:"res_maxrtt,omitempty"`
+	MinRTT      float64 `json:"res_minrtt,omitempty"`
+	Sent        int     `json:"res_sent,omitempty"`
+	Received    int     `json:"res_recv,omitempty"`
+	Unreachable bool    `json:"unreachable,omitempty"`
+
+	Created time.Time `json:"res_created"`
+}
+
+type ListResponse []*Response
+
+func (lis ListResponse) Len() int {
+	return len(lis)
+}
+func (lis ListResponse) Less(i, j int) bool {
+	if lis[j].Latency == 0.0 && lis[i].Latency > 0.0 {
+		return true
+	}
+	if lis[i].Latency == 0.0 && lis[j].Latency > 0.0 {
+		return false
+	}
+
+	return lis[j].Latency >= lis[i].Latency
+}
+func (lis ListResponse) Swap(i, j int) {
+	lis[i], lis[j] = lis[j], lis[i]
+}
+
+type RequestSubmitted struct {
+	event.IsEvent
+
+	RequestIP string `json:"req_ip"`
+	Hidden    bool   `json:"hide,omitempty"`
+}
+
+func (r *RequestSubmitted) StreamID() string {
+	return r.EventMeta().GetEventID()
+}
+func (r *RequestSubmitted) RequestID() string {
+	return r.EventMeta().GetEventID()
+}
+func (r *RequestSubmitted) Created() time.Time {
+	return r.EventMeta().Created()
+}
+func (r *RequestSubmitted) CreatedString() string {
+	return r.Created().Format("2006-01-02 15:04:05")
+}
+func (r *RequestSubmitted) Family() int {
+	if r == nil {
+		return 0
+	}
+
+	ip, err := netip.ParseAddr(r.RequestIP)
+	switch {
+	case err != nil:
+		return 0
+	case ip.Is4():
+		return 1
+	default:
+		return 2
+	}
+}
+func (r *RequestSubmitted) String() string {
+	return fmt.Sprint(r.EventMeta().EventID, r.RequestIP, r.Hidden, r.CreatedString())
+}
+
+var _ event.Event = (*RequestSubmitted)(nil)
+
+func (e *RequestSubmitted) MarshalBinary() (text []byte, err error) {
+	return json.Marshal(e)
+}
+func (e *RequestSubmitted) UnmarshalBinary(b []byte) error {
+	return json.Unmarshal(b, e)
+}
+func (e *RequestSubmitted) MarshalEnviron() ([]byte, error) {
+	if e == nil {
+		return nil, nil
+	}
+
+	var b bytes.Buffer
+	b.WriteString("REQ_ID=")
+	b.WriteString(e.RequestID())
+	b.WriteRune('\n')
+
+	b.WriteString("REQ_IP=")
+	b.WriteString(e.RequestIP)
+	b.WriteRune('\n')
+
+	b.WriteString("REQ_FAMILY=")
+	if family := e.Family(); family > 0 {
+		b.WriteString(strconv.Itoa(family))
+	}
+	b.WriteRune('\n')
+
+	b.WriteString("REQ_CREATED=")
+	b.WriteString(e.CreatedString())
+	b.WriteRune('\n')
+
+	return b.Bytes(), nil
+}
+
+type ResultSubmitted struct {
+	event.IsEvent
+
+	RequestID   string  `json:"req_id"`
+	PeerID      string  `json:"peer_id"`
+	PeerVersion string  `json:"peer_version"`
+	Latency     float64 `json:"latency,omitempty"`
+	Jitter      float64 `json:"jitter,omitempty"`
+	MaxRTT      float64 `json:"maxrtt,omitempty"`
+	MinRTT      float64 `json:"minrtt,omitempty"`
+	Sent        int     `json:"res_sent,omitempty"`
+	Received    int     `json:"res_recv,omitempty"`
+	Unreachable bool    `json:"unreachable,omitempty"`
+}
+
+func (r *ResultSubmitted) Created() time.Time {
+	return r.EventMeta().Created()
+}
+
+var _ event.Event = (*ResultSubmitted)(nil)
+
+func (e *ResultSubmitted) String() string {
+	return fmt.Sprintf("id: %s\npeer: %s\nversion: %s\nlatency: %0.4f", e.RequestID, e.PeerID, e.PeerVersion, e.Latency)
+}
+
+type RequestTruncated struct {
+	RequestID string
+
+	event.IsEvent
+}
+
+var _ event.Event = (*RequestTruncated)(nil)
+
+func (e *RequestTruncated) String() string {
+	return fmt.Sprintf("request truncated id: %s\n", e.RequestID)
+}

app/peerfinder/http.go

@@ -0,0 +1,713 @@
+package peerfinder
+
+import (
+	"context"
+	"embed"
+	"encoding/json"
+	"errors"
+	"fmt"
+	"html/template"
+	"io"
+	"io/fs"
+	"log"
+	"net"
+	"net/http"
+	"sort"
+	"strconv"
+	"strings"
+
+	"github.com/oklog/ulid/v2"
+	contentnegotiation "gitlab.com/jamietanna/content-negotiation-go"
+	"go.opentelemetry.io/otel/attribute"
+	"go.sour.is/pkg/lg"
+
+	"go.sour.is/ev"
+	"go.sour.is/ev/event"
+)
+
+var (
+	//go:embed pages/* layouts/* assets/*
+	files     embed.FS
+	templates map[string]*template.Template
+)
+
+// Args passed to templates
+type Args struct {
+	RemoteIP   string
+	Requests   []*Request
+	CountPeers int
+}
+
+// requestArgs builds args from http.Request
+func requestArgs(r *http.Request) Args {
+	remoteIP, _, _ := strings.Cut(r.RemoteAddr, ":")
+	if s := r.Header.Get("X-Forwarded-For"); s != "" {
+		s, _, _ = strings.Cut(s, ", ")
+		remoteIP = s
+	}
+	return Args{
+		RemoteIP: remoteIP,
+	}
+}
+
+// RegisterHTTP adds handler paths to the ServeMux
+func (s *service) RegisterHTTP(mux *http.ServeMux) {
+	a, _ := fs.Sub(files, "assets")
+	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) Setup() error {
+	return nil
+}
+
+// ServeHTTP handle HTTP requests
+func (s *service) ServeHTTP(w http.ResponseWriter, r *http.Request) {
+	ctx := r.Context()
+
+	ctx, span := lg.Span(ctx)
+	defer span.End()
+
+	r = r.WithContext(ctx)
+
+	if !s.up.Load() {
+		w.WriteHeader(http.StatusFailedDependency)
+		fmt.Fprint(w, "Starting up...")
+		return
+	}
+
+	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.getResultsForRequest(w, r, strings.TrimPrefix(r.URL.Path, "/peers/req/"))
+			return
+
+		case strings.HasPrefix(r.URL.Path, "/peers/status"):
+			var pickID string
+			if strings.HasPrefix(r.URL.Path, "/peers/status/") {
+				pickID = strings.TrimPrefix(r.URL.Path, "/peers/status/")
+			}
+
+			var requests []*Request
+
+			s.state.Use(r.Context(), func(ctx context.Context, state *state) error {
+				for id, p := range state.peers {
+					fmt.Fprintln(w, "PEER:", id[24:], p.Owner, p.Name)
+				}
+
+				if pickID != "" {
+					if rq, ok := state.requests[pickID]; ok {
+						requests = append(requests, rq)
+					}
+				} else {
+					requests = make([]*Request, 0, len(state.requests))
+					for i := range state.requests {
+						rq := state.requests[i]
+						requests = append(requests, rq)
+					}
+				}
+
+				for i := range requests {
+					rq := requests[i]
+					for i := range rq.Responses {
+						res := rq.Responses[i]
+						if peer, ok := state.peers[res.PeerID]; ok {
+							res.Peer = peer
+							res.Peer.ID = ""
+						}
+					}
+				}
+
+				return nil
+			})
+
+			for i, rq := range requests {
+				fmt.Fprintln(w, "REQ: ", i, rq.RequestIP, len(rq.Responses))
+				for i, peer := range fnOrderByPeer(rq) {
+					fmt.Fprintln(w, "  PEER: ", i, peer.RequestID, peer.Name, peer.Latency, peer.Jitter)
+					for i, res := range peer.Results {
+						fmt.Fprintln(w, "    RES: ", i, res.RequestID, res.Latency, res.Jitter)
+					}
+				}
+			}
+
+		default:
+			s.getResults(w, r)
+			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, peerID string) {
+	ctx, span := lg.Span(r.Context())
+	defer span.End()
+
+	span.SetAttributes(
+		attribute.String("peerID", peerID),
+	)
+
+	var peer *Peer
+	err := s.state.Use(ctx, func(ctx context.Context, state *state) error {
+		var ok bool
+		if peer, ok = state.peers[peerID]; !ok {
+			return fmt.Errorf("peer not found: %s", peerID)
+		}
+
+		return nil
+	})
+	if err != nil {
+		span.RecordError(err)
+		w.WriteHeader(http.StatusNotFound)
+		return
+	}
+
+	info, err := ev.Upsert(ctx, s.es, aggInfo, func(ctx context.Context, agg *Info) error {
+		return agg.OnUpsert() // 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
+	}
+
+	peerResults := &PeerResults{}
+	peerResults.SetStreamID(aggPeer(peerID))
+	err = s.es.Load(ctx, peerResults)
+	if err != nil && !errors.Is(err, ev.ErrNotFound) {
+		span.RecordError(fmt.Errorf("peer not found: %w", err))
+		w.WriteHeader(http.StatusNotFound)
+	}
+
+	var req *Request
+	for _, e := range requests {
+		r := &Request{}
+		r.ApplyEvent(e)
+
+		if !peerResults.Has(r.RequestID) {
+			if !peer.CanSupport(r.RequestIP) {
+				continue
+			}
+			req = r
+		}
+	}
+	if req == nil {
+		span.RecordError(fmt.Errorf("request not found"))
+		w.WriteHeader(http.StatusNoContent)
+	}
+
+	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) {
+	ctx, span := lg.Span(r.Context())
+	defer span.End()
+
+	// events, err := s.es.Read(ctx, queueRequests, -1, -30)
+	// if err != nil {
+	// 	span.RecordError(err)
+	// 	w.WriteHeader(http.StatusInternalServerError)
+	// 	return
+	// }
+
+	// requests := make([]*Request, len(events))
+	// for i, req := range events {
+	// 	if req, ok := req.(*RequestSubmitted); ok {
+	// 		requests[i], err = s.loadResult(ctx, req.RequestID())
+	// 		if err != nil {
+	// 			span.RecordError(err)
+	// 			w.WriteHeader(http.StatusInternalServerError)
+	// 			return
+	// 		}
+	// 	}
+	// }
+
+	var requests ListRequest
+	s.state.Use(ctx, func(ctx context.Context, state *state) error {
+		requests = make([]*Request, 0, len(state.requests))
+
+		for _, req := range state.requests {
+			if req.RequestID == "" {
+				continue
+			}
+			if req.Hidden {
+				continue
+			}
+
+			requests = append(requests, req)
+		}
+
+		return nil
+	})
+	sort.Sort(sort.Reverse(requests))
+
+	args := requestArgs(r)
+	args.Requests = requests[:maxResults]
+
+	s.state.Use(ctx, func(ctx context.Context, state *state) error {
+		args.CountPeers = len(state.peers)
+		return nil
+	})
+
+	t := templates["home.go.tpl"]
+	t.Execute(w, args)
+}
+func (s *service) getResultsForRequest(w http.ResponseWriter, r *http.Request, uuid string) {
+	ctx, span := lg.Span(r.Context())
+	defer span.End()
+
+	span.SetAttributes(
+		attribute.String("uuid", uuid),
+	)
+
+	var request *Request
+	err := s.state.Use(ctx, func(ctx context.Context, state *state) error {
+		request = state.requests[uuid]
+
+		return nil
+	})
+	if err != nil {
+		w.WriteHeader(http.StatusNotFound)
+		return
+	}
+
+	request, err = s.loadResult(ctx, request)
+
+	// request, err := s.loadResult(ctx, uuid)
+	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(r.Header.Get("Accept"))
+	if err != nil {
+		w.WriteHeader(http.StatusNotAcceptable)
+		return
+	}
+	span.AddEvent(negotiated.String())
+	switch negotiated.String() {
+	// case "text/environment":
+	// 	encodeTo(w, responses.MarshalBinary)
+	case "application/json":
+		json.NewEncoder(w).Encode(request)
+		return
+	default:
+		args := requestArgs(r)
+		args.Requests = append(args.Requests, request)
+		span.AddEvent(fmt.Sprint(args))
+		err := renderTo(w, "req.go.tpl", args)
+		span.RecordError(err)
+
+		return
+	}
+}
+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
+	}
+
+	args := requestArgs(r)
+	requestIP := args.RemoteIP
+
+	if ip := r.Form.Get("req_ip"); ip != "" {
+		requestIP = ip
+	}
+
+	ip := net.ParseIP(requestIP)
+	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))
+
+	http.Redirect(w, r, "/peers/req/"+req.RequestID(), http.StatusSeeOther)
+}
+func (s *service) postResult(w http.ResponseWriter, r *http.Request, reqID string) {
+	ctx, span := lg.Span(r.Context())
+	defer span.End()
+
+	span.SetAttributes(
+		attribute.String("id", reqID),
+	)
+
+	if _, err := ulid.ParseStrict(reqID); err != nil {
+		w.WriteHeader(http.StatusNotFound)
+		return
+	}
+
+	if err := r.ParseForm(); err != nil {
+		w.WriteHeader(http.StatusUnprocessableEntity)
+		return
+	}
+
+	form := make([]string, 0, len(r.Form))
+	for k, vals := range r.Form {
+		for _, v := range vals {
+			form = append(form, fmt.Sprint(k, v))
+		}
+	}
+	span.SetAttributes(
+		attribute.StringSlice("form", form),
+	)
+
+	peerID := r.Form.Get("peer_id")
+
+	err := s.state.Use(ctx, func(ctx context.Context, state *state) error {
+		var ok bool
+		if _, ok = state.peers[peerID]; !ok {
+			log.Printf("peer not found: %s\n", peerID)
+			return fmt.Errorf("peer not found: %s", peerID)
+		}
+
+		return nil
+	})
+	if err != nil {
+		span.RecordError(err)
+		w.WriteHeader(http.StatusNotFound)
+		return
+	}
+
+	peerResults := &PeerResults{}
+	peerResults.SetStreamID(aggPeer(peerID))
+	err = s.es.Load(ctx, peerResults)
+	if err != nil {
+		span.RecordError(fmt.Errorf("peer not found: %w", err))
+		w.WriteHeader(http.StatusNotFound)
+	}
+
+	if peerResults.Has(reqID) {
+		span.RecordError(fmt.Errorf("request previously recorded: req=%v peer=%v", reqID, peerID))
+		w.WriteHeader(http.StatusAlreadyReported)
+		return
+	}
+
+	var unreach bool
+	latency, err := strconv.ParseFloat(r.Form.Get("res_latency"), 64)
+	if err != nil {
+		unreach = true
+	}
+
+	req := &ResultSubmitted{
+		RequestID:   reqID,
+		PeerID:      r.Form.Get("peer_id"),
+		PeerVersion: r.Form.Get("peer_version"),
+		Latency:     latency,
+		Unreachable: unreach,
+	}
+
+	if jitter, err := strconv.ParseFloat(r.Form.Get("res_jitter"), 64); err == nil {
+		req.Jitter = jitter
+	} else {
+		span.RecordError(err)
+	}
+	if minrtt, err := strconv.ParseFloat(r.Form.Get("res_minrtt"), 64); err == nil {
+		req.MinRTT = minrtt
+	} else {
+		span.RecordError(err)
+	}
+	if maxrtt, err := strconv.ParseFloat(r.Form.Get("res_maxrtt"), 64); err == nil {
+		req.MaxRTT = maxrtt
+	} else {
+		span.RecordError(err)
+	}
+
+	span.SetAttributes(
+		attribute.Stringer("result", req),
+	)
+
+	log.Printf("record result: %v", req)
+	s.es.Append(ctx, queueResults, event.NewEvents(req))
+}
+
+func renderTo(w io.Writer, name string, args any) (err error) {
+	defer func() {
+		if p := recover(); p != nil {
+			err = fmt.Errorf("panic: %s", p)
+		}
+		if err != nil {
+			fmt.Fprint(w, err)
+		}
+	}()
+
+	t, ok := templates[name]
+	if !ok || t == nil {
+		return fmt.Errorf("missing template")
+	}
+	return t.Execute(w, args)
+}
+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
+		}
+		pt := template.New(tmpl.Name())
+		pt.Funcs(funcMap)
+		pt, err = pt.ParseFS(files, "pages/"+tmpl.Name(), "layouts/*.go.tpl")
+		if err != nil {
+			log.Println(err)
+
+			return err
+		}
+		templates[tmpl.Name()] = pt
+	}
+	return nil
+}
+
+var funcMap = map[string]any{
+	"orderByPeer":    fnOrderByPeer,
+	"countResponses": fnCountResponses,
+}
+
+type peerResult struct {
+	RequestID string
+	Name      string
+	Country   string
+	Latency   float64
+	Jitter    float64
+}
+type peer struct {
+	RequestID string
+	Name      string
+	Note      string
+	Nick      string
+	Country   string
+	Latency   float64
+	Jitter    float64
+	VPNTypes  []string
+
+	Results peerResults
+}
+type listPeer []peer
+
+func (lis listPeer) Len() int {
+	return len(lis)
+}
+func (lis listPeer) Less(i, j int) bool {
+	if lis[j].Latency == 0.0 && lis[i].Latency > 0.0 {
+		return true
+	}
+	if lis[i].Latency == 0.0 && lis[j].Latency > 0.0 {
+		return false
+	}
+
+	return lis[i].Latency < lis[j].Latency
+}
+func (lis listPeer) Swap(i, j int) {
+	lis[i], lis[j] = lis[j], lis[i]
+}
+
+type peerResults []peerResult
+
+func (lis peerResults) Len() int {
+	return len(lis)
+}
+func (lis peerResults) Less(i, j int) bool {
+	if lis[j].Latency == 0.0 && lis[i].Latency > 0.0 {
+		return true
+	}
+	if lis[i].Latency == 0.0 && lis[j].Latency > 0.0 {
+		return false
+	}
+
+	return lis[i].Latency < lis[j].Latency
+}
+func (lis peerResults) Swap(i, j int) {
+	lis[i], lis[j] = lis[j], lis[i]
+}
+
+func fnOrderByPeer(rq *Request) listPeer {
+	peers := make(map[string]peer)
+	for i := range rq.Responses {
+		if rq.Responses[i] == nil || rq.Responses[i].Peer == nil {
+			continue
+		}
+		rs := rq.Responses[i]
+
+		p, ok := peers[rs.Peer.Owner]
+
+		if !ok {
+			p.RequestID = rq.RequestID
+			p.Country = rs.Peer.Country
+			p.Name = rs.Peer.Name
+			p.Nick = rs.Peer.Nick
+			p.Note = rs.Peer.Note
+			p.Latency = rs.Latency
+			p.Jitter = rs.Jitter
+			p.VPNTypes = rs.Peer.Type
+		}
+
+		p.Results = append(p.Results, peerResult{
+			RequestID: rq.RequestID,
+			Name:      rs.Peer.Name,
+			Country:   rs.Peer.Country,
+			Latency:   rs.Latency,
+			Jitter:    rs.Jitter,
+		})
+
+		peers[rs.Peer.Owner] = p
+	}
+
+	peerList := make(listPeer, 0, len(peers))
+	for i := range peers {
+		v := peers[i]
+		sort.Sort(v.Results)
+
+		v.Name = v.Results[0].Name
+		v.Country = v.Results[0].Country
+		v.Latency = v.Results[0].Latency
+		v.Jitter = v.Results[0].Jitter
+
+		peerList = append(peerList, v)
+	}
+
+	sort.Sort(peerList)
+	return peerList
+}
+func fnCountResponses(rq *Request) int {
+	count := 0
+	for _, res := range rq.Responses {
+		if !res.Unreachable {
+			count++
+		}
+	}
+	return count
+}
+
+// func filter(peer *Peer, 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 {
+// 			if !peer.CanSupport(req.RequestIP) {
+// 				continue
+// 			}
+
+// 			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
+// }

app/peerfinder/jobs.go

@@ -0,0 +1,216 @@
+package peerfinder
+
+import (
+	"context"
+	"encoding/json"
+	"errors"
+	"fmt"
+	"log"
+	"net/http"
+	"time"
+
+	"go.sour.is/ev"
+	"go.sour.is/ev/event"
+	"go.sour.is/pkg/lg"
+	"go.sour.is/pkg/set"
+)
+
+// RefreshJob retrieves peer info from the peerdb
+func (s *service) RefreshJob(ctx context.Context, _ time.Time) error {
+	ctx, span := lg.Span(ctx)
+	defer span.End()
+
+	req, err := http.NewRequestWithContext(ctx, http.MethodGet, s.statusURL, nil)
+	span.RecordError(err)
+	if err != nil {
+		return err
+	}
+	req.Header.Set("Accept", "application/json")
+
+	res, err := http.DefaultClient.Do(req)
+	span.RecordError(err)
+	if err != nil {
+		return err
+	}
+
+	defer res.Body.Close()
+	var peers []*Peer
+	err = json.NewDecoder(res.Body).Decode(&peers)
+	span.RecordError(err)
+	if err != nil {
+		return err
+	}
+
+	err = s.state.Use(ctx, func(ctx context.Context, t *state) error {
+		for _, peer := range peers {
+			t.peers[peer.ID] = peer
+		}
+
+		return nil
+	})
+	span.RecordError(err)
+	if err != nil {
+		return err
+	}
+
+	log.Printf("processed %d peers", len(peers))
+	span.AddEvent(fmt.Sprintf("processed %d peers", len(peers)))
+
+	s.up.Store(true)
+
+	err = s.cleanPeerJobs(ctx)
+
+	span.RecordError(err)
+	return err
+}
+
+const maxResults = 30
+
+// CleanJob truncates streams old request data
+func (s *service) CleanJob(ctx context.Context, now time.Time) error {
+	ctx, span := lg.Span(ctx)
+	defer span.End()
+
+	span.AddEvent("clear peerfinder requests")
+
+	err := s.cleanRequests(ctx, now)
+	if err != nil {
+		return err
+	}
+	// if err = s.cleanResults(ctx, endRequestID); err != nil {
+	// 	return err
+	// }
+
+	return s.cleanPeerJobs(ctx)
+}
+func (s *service) cleanPeerJobs(ctx context.Context) error {
+	ctx, span := lg.Span(ctx)
+	defer span.End()
+
+	peers := set.New[string]()
+	err := s.state.Use(ctx, func(ctx context.Context, state *state) error {
+		for id := range state.peers {
+			peers.Add(id)
+		}
+		return nil
+	})
+	if err != nil {
+		return err
+	}
+
+	// trunctate all the peer streams to last 30
+	for streamID := range peers {
+		streamID = aggPeer(streamID)
+		first, err := s.es.FirstIndex(ctx, streamID)
+		if err != nil {
+			return err
+		}
+		last, err := s.es.LastIndex(ctx, streamID)
+		if err != nil {
+			return err
+		}
+		if last-first < maxResults {
+			fmt.Println("SKIP", streamID, first, last)
+			continue
+		}
+
+		newFirst := int64(last - 30)
+		// fmt.Println("TRUNC", streamID, first, newFirst, last)
+		span.AddEvent(fmt.Sprint("TRUNC", streamID, first, newFirst, last))
+		err = s.es.Truncate(ctx, streamID, int64(newFirst))
+		if err != nil {
+			return err
+		}
+	}
+
+	return nil
+}
+func (s *service) cleanRequests(ctx context.Context, now time.Time) error {
+	ctx, span := lg.Span(ctx)
+	defer span.End()
+
+	var streamIDs []string
+	var startPosition, endPosition int64
+
+	first, err := s.es.FirstIndex(ctx, queueRequests)
+	if err != nil {
+		return err
+	}
+	last, err := s.es.LastIndex(ctx, queueRequests)
+	if err != nil {
+		return err
+	}
+
+	if last-first < maxResults {
+		// fmt.Println("SKIP", queueRequests, first, last)
+		return nil
+	}
+
+	startPosition = int64(first - 1)
+	endPosition = int64(last - maxResults)
+
+	for {
+		events, err := s.es.Read(ctx, queueRequests, startPosition, 1000) // read 1000 from the top each loop.
+		if err != nil && !errors.Is(err, ev.ErrNotFound) {
+			span.RecordError(err)
+			return err
+		}
+
+		if len(events) == 0 {
+			break
+		}
+
+		startPosition = int64(events.Last().EventMeta().ActualPosition)
+		for _, event := range events {
+			switch e := event.(type) {
+			case *RequestSubmitted:
+				if e.EventMeta().ActualPosition < last-maxResults {
+					streamIDs = append(streamIDs, e.RequestID())
+				}
+			}
+		}
+	}
+
+	// truncate all reqs to found end position
+	// fmt.Println("TRUNC", queueRequests, int64(endPosition), last)
+	span.AddEvent(fmt.Sprint("TRUNC", queueRequests, int64(endPosition), last))
+	err = s.es.Truncate(ctx, queueRequests, int64(endPosition))
+	if err != nil {
+		return err
+	}
+
+	// truncate all the request streams
+	for _, streamID := range streamIDs {
+		s.state.Use(ctx, func(ctx context.Context, state *state) error {
+			return state.ApplyEvents(event.NewEvents(&RequestTruncated{
+				RequestID: streamID,
+			}))
+		})
+
+		err := s.cleanResult(ctx, streamID)
+		if err != nil {
+			return err
+		}
+	}
+
+	return nil
+}
+func (s *service) cleanResult(ctx context.Context, requestID string) error {
+	ctx, span := lg.Span(ctx)
+	defer span.End()
+
+	streamID := aggRequest(requestID)
+
+	last, err := s.es.LastIndex(ctx, streamID)
+	if err != nil {
+		return err
+	}
+	// truncate all reqs to found end position
+	// fmt.Println("TRUNC", streamID, last)
+	span.AddEvent(fmt.Sprint("TRUNC", streamID, last))
+	err = s.es.Truncate(ctx, streamID, int64(last))
+	if err != nil {
+		return err
+	}
+	return nil
+}

app/peerfinder/layouts/main.go.tpl

@@ -0,0 +1,38 @@
+{{define "main"}}
+<!DOCTYPE html>
+
+<html>
+<head>
+    <meta charset="UTF-8">
+	{{template "meta" .}}
+    <title>DN42 PingFinder</title>
+
+    <link href="/peers/assets/bootstrap.min.css" rel="stylesheet" integrity="sha384-1q8mTJOASx8j1Au+a5WDVnPi2lkFfwwEAa8hDDdjZlpLegxhjVME1fgjWPGmkzs7" crossorigin="anonymous">
+    <link href="/peers/assets/peerfinder.css" rel="stylesheet" crossorigin="anonymous">
+</head>
+
+<body>
+
+<div class="container-fluid">
+      <div class="header clearfix">
+        <nav>
+          <ul class="nav nav-pills pull-right">
+            <li role="presentation"><a href="/peers">Home</a></li>
+            <!--
+              <li role="presentation"><a href="/peers/status">Status</a></li>
+            -->
+            <li role="presentation"><a href="//util.sour.is/peer">Sign up/Manage</a></li>
+            <li role="presentation"><a href="https://git.dn42.dev/dn42/pingfinder/src/branch/master/clients">Scripts</a></li>
+          </ul>
+        </nav>
+        <h3 class="text-muted">DN42 PeerFinder</h3>
+      </div>
+</div>
+
+<div class=container>
+	{{template "content" .}}
+</div>
+
+</body>
+</html>
+{{end}}

app/peerfinder/pages/home.go.tpl

@@ -0,0 +1,65 @@
+{{template "main" .}}
+
+{{define "meta"}}
+    <meta http-equiv="refresh" content="30">
+{{end}}
+
+{{define "content"}}
+  <h2>What is this?</h2>
+
+  <p>This tool allows you to find "good" peerings
+  for <a href="https://dn42.net">dn42</a>, by measuring the latency from
+  various points in the network towards you.</p>
+
+  <p>If you don't know what dn42 is,
+  read <a href="https://dn42.net/Home">the website</a> and in particular
+  the <a href="https://dn42.net/Getting-started-with-dn42">Getting Started
+  guide</a>.</p>
+
+  <h2>How does it work?</h2>
+
+  <p>
+    <ol>
+      <li>You enter your (Internet) IP address</li>
+      <li>Various routers participating in dn42 will ping you over the Internet</li>
+      <li>After a short while, you get back all the latency results</li>
+      <li>You can then peer with people close to you (low latency)</li>
+    </ol>
+  </p>
+
+  <form class="form-inline" method="POST" action="/peers/req">
+        <label>Ping IP Address [Check Hidden?]:</label>
+        <div class="input-group input-group-sm">
+            <input class="form-control" type="text" name="req_ip" placeholder="{{ .RemoteIP }}">
+            <span class="input-group-addon">
+              <input type="checkbox" name="req_hidden" value=1 aria-label="Hidden?">
+            </span>
+        </div>
+        <button class="btn btn-default" type="submit">Submit</button>
+  </form>
+
+  <p>If you mark your measurement as hidden, it will not be displayed on the
+  page below.  Note that the IP addresses of the target will be shown alongside the result.</p>
+
+
+  <div class=row>
+  <h2>Results</h2>
+  {{ with $args := . }}
+  {{ range $req := .Requests }}
+  {{ if ne $req.RequestID "" }}
+      <div class="panel panel-primary">
+          <div class="panel-heading">
+            <a href="/peers/req/{{ $req.RequestID }}">
+              {{ $req.RequestIP }} on {{ $req.Created.Format "02 Jan 06 15:04 MST" }}
+            </a> &mdash; <b>Request ID:</b> {{ $req.RequestID }}
+            <div style='float:right'>
+              <a href="/peers/req/{{ $req.RequestID }}" class='btn btn-success'>{{ countResponses $req }} / {{ $args.CountPeers }} </a>
+            </div>
+          </div>
+      </div>
+  {{end}}
+  {{end}}
+  {{end}}
+  </div>
+
+{{end}}

app/peerfinder/pages/req.go.tpl

@@ -0,0 +1,50 @@
+{{template "main" .}}
+
+{{define "meta"}}
+  <meta http-equiv="refresh" content="30">
+{{end}}
+
+
+{{define "content"}}
+  {{range .Requests}}
+    <h2>Results to {{.RequestIP}}{{if .Hidden}} 👁️{{end}}</h2>
+
+    {{range orderByPeer .}}  
+      <div class="panel panel-primary" id="peer-{{.Nick}}">
+        <div class="panel-heading">
+          <b> {{.Country}} :: {{.Name}} :: {{.Nick}} </b>
+          <div style='float:right'>
+            <a class='btn btn-success' href="#peer-{{.Nick}}">{{ if eq .Latency 0.0 }}&mdash;{{ else }}{{printf "%0.3f ms" .Latency}}{{ end }}</a>
+          </div>
+        </div>
+        <div class="panel-body">
+          <b>Note:</b> {{.Note}}<br/>
+          <b>VPN Types:</b> {{range .VPNTypes}} {{.}} {{end}}<br/>
+          <b>IRC:</b> {{.Nick}}
+          <h4>Other Results</h4>
+
+          <table class="table table-striped">
+            <thead>
+              <tr>
+                <th>Peer Name</th>
+                <th>Country</th>
+                <th>Latency</th>
+                <th>Jitter</th>
+              </tr>
+            </thead>
+            <tbody>
+              {{range .Results}}
+                <tr>
+                  <th>{{.Name}}</th>
+                  <td>{{.Country}}</td>
+                  <td>{{ if eq .Latency 0.0 }}&mdash;{{ else }}{{printf "%0.3f ms" .Latency}}{{ end }}</td>
+                  <td>{{ if eq .Jitter 0.0 }}&mdash;{{ else }}{{ printf "%0.3f ms" .Jitter }}{{ end }}</td>
+                </tr>
+              {{end}}
+            </tbody>
+          </table>
+        </div>
+      </div>
+    {{end}}
+  {{end}}
+{{end}}

app/peerfinder/service.go

@@ -0,0 +1,179 @@
+package peerfinder
+
+import (
+	"context"
+	"fmt"
+	"sync/atomic"
+	"time"
+
+	"go.sour.is/pkg/lg"
+	"go.sour.is/pkg/locker"
+	"go.uber.org/multierr"
+
+	"go.sour.is/ev"
+	"go.sour.is/ev/event"
+)
+
+const (
+	aggInfo       = "pf-info"
+	queueRequests = "pf-requests"
+	queueResults  = "pf-results"
+	initVersion   = "1.2.1"
+)
+
+func aggRequest(id string) string { return "pf-request-" + id }
+func aggPeer(id string) string    { return "pf-peer-" + id }
+
+type service struct {
+	es        *ev.EventStore
+	statusURL string
+
+	state *locker.Locked[*state]
+	up    atomic.Bool
+	stop  func()
+}
+
+type state struct {
+	peers    map[string]*Peer
+	requests map[string]*Request
+}
+
+func New(ctx context.Context, es *ev.EventStore, statusURL string) (*service, error) {
+	ctx, span := lg.Span(ctx)
+	defer span.End()
+
+	loadTemplates()
+
+	if err := event.Register(ctx, &RequestSubmitted{}, &ResultSubmitted{}, &VersionChanged{}); err != nil {
+		span.RecordError(err)
+		return nil, err
+	}
+
+	svc := &service{
+		es:        es,
+		statusURL: statusURL,
+		state: locker.New(&state{
+			peers:    make(map[string]*Peer),
+			requests: make(map[string]*Request),
+		})}
+
+	return svc, nil
+}
+func (s *service) loadResult(ctx context.Context, request *Request) (*Request, error) {
+	if request == nil {
+		return request, nil
+	}
+
+	return request, s.state.Use(ctx, func(ctx context.Context, t *state) error {
+
+		for i := range request.Responses {
+			res := request.Responses[i]
+			if peer, ok := t.peers[res.PeerID]; ok {
+				res.Peer = peer
+				res.Peer.ID = ""
+			}
+		}
+
+		return nil
+	})
+}
+func (s *service) Run(ctx context.Context) (err error) {
+	var errs error
+
+	ctx, span := lg.Span(ctx)
+	defer span.End()
+
+	ctx, s.stop = context.WithCancel(ctx)
+
+	subReq, e := s.es.EventStream().Subscribe(ctx, queueRequests, 0)
+	errs = multierr.Append(errs, e)
+
+	subRes, e := s.es.EventStream().Subscribe(ctx, queueResults, 0)
+	errs = multierr.Append(errs, e)
+
+	defer func() {
+		ctx, cancel := context.WithTimeout(context.Background(), 1*time.Second)
+		defer cancel()
+
+		err = multierr.Combine(subReq.Close(ctx), subRes.Close(ctx), err)
+	}()
+
+	if errs != nil {
+		return errs
+	}
+
+	for {
+		var events event.Events
+		select {
+		case <-ctx.Done():
+			return nil
+		case ok := <-subReq.Recv(ctx):
+			if ok {
+				events, err = subReq.Events(ctx)
+			}
+		case ok := <-subRes.Recv(ctx):
+			if ok {
+				events, err = subRes.Events(ctx)
+			}
+		}
+
+		s.state.Use(ctx, func(ctx context.Context, state *state) error {
+			return state.ApplyEvents(events)
+		})
+		events = events[:0]
+	}
+}
+func (s *service) Stop(ctx context.Context) (err error) {
+	defer func() {
+		if p := recover(); p != nil {
+			err = fmt.Errorf("PANIC: %v", p)
+		}
+	}()
+
+	s.stop()
+	return err
+}
+
+func (s *state) ApplyEvents(events event.Events) error {
+	for _, e := range events {
+		switch e := e.(type) {
+		case *RequestSubmitted:
+			if _, ok := s.requests[e.RequestID()]; !ok {
+				s.requests[e.RequestID()] = &Request{}
+			}
+			s.requests[e.RequestID()].ApplyEvent(e)
+		case *ResultSubmitted:
+			if _, ok := s.requests[e.RequestID]; !ok {
+				s.requests[e.RequestID] = &Request{}
+			}
+			s.requests[e.RequestID].ApplyEvent(e)
+		case *RequestTruncated:
+			delete(s.requests, e.RequestID)
+		}
+	}
+
+	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(aggRequest(e.RequestID()), e1)
+
+		return []event.Event{e1}
+	case *ResultSubmitted:
+		e1 := event.NewPtr(streamID, streamPos)
+		event.SetStreamID(aggRequest(e.RequestID), e1)
+
+		e2 := event.NewPtr(streamID, streamPos)
+		event.SetStreamID(aggPeer(e.PeerID), e2)
+
+		return []event.Event{e1, e2}
+	}
+	return nil
+}