Request ID: {{ $req.RequestID }}
diff --git a/app/peerfinder/pages/req.tpl b/app/peerfinder/pages/req.tpl
index d67341c..bc396d0 100644
--- a/app/peerfinder/pages/req.tpl
+++ b/app/peerfinder/pages/req.tpl
@@ -15,7 +15,7 @@
{{.Country}} :: {{.Name}} :: {{.Nick}}
@@ -38,7 +38,7 @@
| {{.Name}} |
{{.Country}} |
- {{printf "%0.3f ms" .Latency}} |
+ {{ if eq .Latency 0.0 }}—{{ else }}{{printf "%0.3f ms" .Latency}}{{ end }} |
{{ if eq .Jitter 0.0 }}—{{ else }}{{ printf "%0.3f ms" .Jitter }}{{ end }} |
{{end}}
diff --git a/app/peerfinder/service.go b/app/peerfinder/service.go
index b7ddec6..502259e 100644
--- a/app/peerfinder/service.go
+++ b/app/peerfinder/service.go
@@ -2,31 +2,14 @@ package peerfinder
import (
"context"
- "embed"
- "encoding/json"
- "errors"
"fmt"
- "html/template"
- "io"
- "io/fs"
- "log"
- "net"
- "net/http"
- "sort"
- "strconv"
- "strings"
"time"
- "github.com/oklog/ulid"
- contentnegotiation "gitlab.com/jamietanna/content-negotiation-go"
- "go.opentelemetry.io/otel/attribute"
-
"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"
- "github.com/sour-is/ev/pkg/math"
- "github.com/sour-is/ev/pkg/set"
+ "go.uber.org/multierr"
)
const (
@@ -39,21 +22,17 @@ const (
func aggRequest(id string) string { return "pf-request-" + id }
func aggPeer(id string) string { return "pf-peer-" + id }
-var (
- //go:embed pages/* layouts/* assets/*
- files embed.FS
- templates map[string]*template.Template
-)
-
type service struct {
es *es.EventStore
statusURL string
state *locker.Locked[state]
+ stop func()
}
type state struct {
- peers map[string]*Peer
+ peers map[string]*Peer
+ requests map[string]*Request
}
func New(ctx context.Context, es *es.EventStore, statusURL string) (*service, error) {
@@ -67,621 +46,109 @@ func New(ctx context.Context, es *es.EventStore, statusURL string) (*service, er
return nil, err
}
- svc := &service{es: es, statusURL: statusURL, state: locker.New(&state{peers: make(map[string]*Peer)})}
+ 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) RegisterHTTP(mux *http.ServeMux) {
- a, _ := fs.Sub(files, "assets")
- assets := http.StripPrefix("/peers/assets/", http.FileServer(http.FS(a)))
+func (s *service) loadResult(ctx context.Context, uuid string) (*Request, error) {
+ request := &Request{}
+ request.SetStreamID(aggRequest(uuid))
+ err := s.es.Load(ctx, request)
+ if err != nil {
+ return nil, err
+ }
- mux.Handle("/peers/assets/", lg.Htrace(assets, "peer-assets"))
- mux.Handle("/peers/", lg.Htrace(s, "peers"))
+ return request, s.state.Modify(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) ServeHTTP(w http.ResponseWriter, r *http.Request) {
- ctx := r.Context()
+func (s *service) Run(ctx context.Context) (err error) {
+ var errs error
ctx, span := lg.Span(ctx)
defer span.End()
- r = r.WithContext(ctx)
+ ctx, s.stop = context.WithCancel(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
+ subReq, e := s.es.EventStream().Subscribe(ctx, queueRequests, 0)
+ errs = multierr.Append(errs, e)
- case strings.HasPrefix(r.URL.Path, "/peers/req/"):
- s.getResultsForRequest(w, r, strings.TrimPrefix(r.URL.Path, "/peers/req/"))
- return
+ subRes, e := s.es.EventStream().Subscribe(ctx, queueResults, 0)
+ errs = multierr.Append(errs, e)
- case strings.HasPrefix(r.URL.Path, "/peers/status"):
- s.state.Modify(r.Context(), func(ctx context.Context, state *state) error {
- for id, p := range state.peers {
- fmt.Fprintln(w, "PEER: ", id, p.Owner, p.Name)
- }
+ defer func() {
+ ctx, cancel := context.WithTimeout(context.Background(), 1*time.Second)
+ defer cancel()
- return nil
- })
+ err = multierr.Combine(subReq.Close(ctx), subRes.Close(ctx), err)
+ }()
- 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, uuid string) {
- ctx, span := lg.Span(r.Context())
- defer span.End()
-
- span.SetAttributes(
- attribute.String("uuid", uuid),
- )
-
- var peer *Peer
- err := s.state.Modify(ctx, func(ctx context.Context, state *state) error {
- var ok bool
- if peer, ok = state.peers[uuid]; !ok {
- return fmt.Errorf("peer not found: %s", uuid)
- }
-
- return nil
- })
- if err != nil {
- span.RecordError(err)
- w.WriteHeader(http.StatusNotFound)
- return
+ if errs != nil {
+ return errs
}
- info, err := es.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
- }
-
- responses, err := s.es.Read(ctx, aggPeer(uuid), -1, -30)
- if err != nil {
- span.RecordError(err)
- w.WriteHeader(http.StatusInternalServerError)
- return
- }
-
- span.AddEvent(fmt.Sprintf("req = %d, res = %d", len(requests), len(responses)))
-
- req := filter(peer, 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(),
+ 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)
}
}
- err = json.NewEncoder(w).Encode(out)
+
+ s.state.Modify(ctx, func(ctx context.Context, state *state) error {
+ return state.ApplyEvents(events)
+ })
+ events = events[:0]
}
- 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
- }
+func (s *service) Stop(ctx context.Context) (err error) {
+ defer func() {
+ if p := recover(); p != nil {
+ err = fmt.Errorf("PANIC: %v", p)
}
- }
+ }()
- args := requestArgs(r)
- args.Requests = requests
-
- s.state.Modify(ctx, func(ctx context.Context, state *state) error {
- args.CountPeers = len(state.peers)
- return nil
- })
-
- t := templates["home.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),
- )
-
- 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.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, 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
- }
-
- peerID := r.Form.Get("peer_id")
- err := s.state.Modify(ctx, func(ctx context.Context, state *state) error {
- var ok bool
- if _, ok = state.peers[peerID]; !ok {
- // fmt.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
- }
-
- var unreach bool
- latency, err := strconv.ParseFloat(r.Form.Get("res_latency"), 64)
- if err != nil {
- unreach = true
- }
-
- req := &ResultSubmitted{
- RequestID: id,
- 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),
- )
-
- s.state.Modify(ctx, func(ctx context.Context, state *state) error {
-
- return nil
- })
-
- idx, err := s.es.LastIndex(ctx, aggRequest(id))
- if err != nil {
- w.WriteHeader(http.StatusInternalServerError)
- return
- }
- if idx == 0 {
- w.WriteHeader(http.StatusNotFound)
- return
- }
-
- s.es.Append(ctx, queueResults, event.NewEvents(req))
-}
-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
- }
-
- span.AddEvent(fmt.Sprintf("processed %d peers", len(peers)))
-
- err = s.state.Modify(ctx, func(ctx context.Context, t *state) error {
- for _, peer := range peers {
- t.peers[peer.ID] = peer
- }
-
- return nil
- })
- span.RecordError(err)
+ s.stop()
return err
}
-func (s *service) CleanJob(ctx context.Context, now time.Time) error {
- ctx, span := lg.Span(ctx)
- defer span.End()
- fmt.Println("clear peerfinder requests")
- span.AddEvent("clear peerfinder requests")
-
- endRequestID, 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.Modify(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
- }
- newFirst := math.Max(math.Max(int64(last-30), int64(first)), 1)
- if last == 0 || newFirst == int64(first) {
- // fmt.Println("SKIP", streamID, first, newFirst, last)
- span.AddEvent(fmt.Sprint("SKIP", streamID, first, newFirst, last))
- continue
- }
- // 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) (string, error) {
- ctx, span := lg.Span(ctx)
- defer span.End()
-
- var streamIDs []string
- var endPosition uint64
- var endRequestID string
-
- last, err := s.es.LastIndex(ctx, queueRequests)
- if err != nil {
- return "", err
- }
-
-end:
- for {
- events, err := s.es.Read(ctx, queueRequests, int64(endPosition), 1000) // read 1000 from the top each loop.
- if err != nil && !errors.Is(err, es.ErrNotFound) {
- span.RecordError(err)
- return "", err
- }
-
- if len(events) == 0 {
- break
- }
-
- endPosition = events.Last().EventMeta().ActualPosition
- for _, event := range events {
- switch e := event.(type) {
- case *RequestSubmitted:
- if e.eventMeta.ActualPosition < last-30 || e.Created().Before(now.Add(-24*time.Hour)) {
- streamIDs = append(streamIDs, aggRequest(e.RequestID()))
- } else {
- endRequestID = e.RequestID()
- endPosition = e.eventMeta.ActualPosition
- break end
- }
+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{}
}
- }
- }
-
- // 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 {
- last, err := s.es.LastIndex(ctx, streamID)
- if err != nil {
- return "", err
- }
- 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 endRequestID, nil
-}
-func (s *service) cleanResults(ctx context.Context, endRequestID string) error {
- ctx, span := lg.Span(ctx)
- defer span.End()
-
- var endPosition uint64
-
- done := false
- for !done {
- events, err := s.es.Read(ctx, queueResults, int64(endPosition), 1000) // read 30 from the top each loop.
- if err != nil {
- return err
- }
-
- if len(events) == 0 {
- done = true
- continue
- }
-
- endPosition = events.Last().EventMeta().ActualPosition
-
- for _, event := range events {
- switch e := event.(type) {
- case *ResultSubmitted:
- if e.RequestID == endRequestID {
- done = true
- endPosition = e.eventMeta.ActualPosition
- }
+ 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)
}
}
- // truncate all reqs to found end position
- // fmt.Println("TRUNC", queueResults, int64(endPosition), last)
- span.AddEvent(fmt.Sprint("TRUNC", queueResults, int64(endPosition)))
- err := s.es.Truncate(ctx, queueResults, int64(endPosition))
- if err != nil {
- return err
- }
- return nil
-}
-
-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
-}
-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/*.tpl")
- if err != nil {
- log.Println(err)
-
- return err
- }
- templates[tmpl.Name()] = pt
- }
return nil
}
@@ -707,130 +174,3 @@ func Projector(e event.Event) []event.Event {
}
return nil
}
-
-type Args struct {
- RemoteIP string
- Requests []*Request
- CountPeers int
-}
-
-func requestArgs(r *http.Request) Args {
- remoteIP, _, _ := strings.Cut(r.RemoteAddr, ":")
- if s := r.Header.Get("X-Forwarded-For"); s != "" {
- remoteIP = s
- }
- return Args{
- RemoteIP: remoteIP,
- }
-}
-
-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)
-}
-
-type ListResponse []Response
-
-func (lis ListResponse) Len() int {
- return len(lis)
-}
-func (lis ListResponse) Less(i, j int) bool {
- return lis[i].Latency < lis[j].Latency
-}
-func (lis ListResponse) Swap(i, j int) {
- lis[i], lis[j] = lis[j], lis[i]
-}
-
-func fnOrderByPeer(rq *Request) any {
- type peerResult struct {
- Name string
- Country string
- Latency float64
- Jitter float64
- }
- type peer struct {
- Name string
- Note string
- Nick string
- Country string
- Latency float64
- Jitter float64
- VPNTypes []string
-
- Results []peerResult
- }
-
- peers := make(map[string]peer)
- sort.Sort(ListResponse(rq.Responses))
- for _, rs := range rq.Responses {
- p, ok := peers[rs.Peer.Owner]
-
- if !ok {
- 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{
- Name: rs.Peer.Name,
- Country: rs.Peer.Country,
- Latency: rs.Latency,
- Jitter: rs.Jitter,
- })
-
- peers[rs.Peer.Owner] = p
- }
-
- return peers
-}
-func fnCountResponses(rq *Request) int {
- count := 0
- for _, res := range rq.Responses {
- if !res.Unreachable {
- count++
- }
- }
- return count
-}
-
-var funcMap = map[string]any{
- "orderByPeer": fnOrderByPeer,
- "countResponses": fnCountResponses,
-}
-
-func (s *service) loadResult(ctx context.Context, uuid string) (*Request, error) {
- request := &Request{}
- request.SetStreamID(aggRequest(uuid))
- err := s.es.Load(ctx, request)
- if err != nil {
- return nil, err
- }
-
- return request, s.state.Modify(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
- })
-}
diff --git a/app/salty/service.go b/app/salty/service.go
index ad98c25..28d9e7f 100644
--- a/app/salty/service.go
+++ b/app/salty/service.go
@@ -51,7 +51,7 @@ var saltyKey = contextKey{"salty"}
type SaltyResolver interface {
CreateSaltyUser(ctx context.Context, nick string, pub string) (*SaltyUser, error)
SaltyUser(ctx context.Context, nick string) (*SaltyUser, error)
- RegisterHTTP(mux *http.ServeMux)
+ IsResolver()
}
func New(ctx context.Context, es *es.EventStore, baseURL string) (*service, error) {
@@ -111,6 +111,8 @@ func New(ctx context.Context, es *es.EventStore, baseURL string) (*service, erro
return svc, errs
}
+func (s *service) IsResolver() {}
+
func (s *service) BaseURL() string {
if s == nil {
return "http://missing.context/"
diff --git a/internal/graph/generated/generated.go b/internal/graph/generated/generated.go
index 82bd7be..d4375ee 100644
--- a/internal/graph/generated/generated.go
+++ b/internal/graph/generated/generated.go
@@ -600,6 +600,14 @@ directive @goTag(
key: String!
value: String
) on INPUT_FIELD_DEFINITION | FIELD_DEFINITION`, BuiltIn: false},
+ {Name: "../../../app/mercury/mercury.graphqls", Input: `# extend type Query{
+# keys(namespace: String!) [String!]!
+# get(namespace: String! keys: [String!]) [String]!
+# }
+
+# extend type Mutation{
+# set(namespace: String! key: String! value: String): Bool!
+# }`, BuiltIn: false},
{Name: "../../../app/msgbus/msgbus.graphqls", Input: `extend type Query {
posts(streamID: String! paging: PageInput): Connection!
}
diff --git a/main.go b/main.go
index 8b876c9..40e0cfe 100644
--- a/main.go
+++ b/main.go
@@ -2,6 +2,7 @@ package main
import (
"context"
+ "errors"
"log"
"net/http"
"net/url"
@@ -28,6 +29,7 @@ import (
resolvelinks "github.com/sour-is/ev/pkg/es/driver/resolve-links"
"github.com/sour-is/ev/pkg/es/driver/streamer"
"github.com/sour-is/ev/pkg/es/event"
+ "github.com/sour-is/ev/pkg/gql/resolver"
"github.com/sour-is/ev/pkg/set"
)
@@ -35,18 +37,21 @@ func main() {
ctx, cancel := signal.NotifyContext(context.Background(), os.Interrupt, os.Kill)
go func() {
<-ctx.Done()
- defer cancel()
+ defer cancel() // restore interrupt function
}()
+ // Initialize logger
ctx, stop := lg.Init(ctx, appName)
defer stop()
- if err := run(ctx); err != nil && err != http.ErrServerClosed {
+ // Run application
+ if err := run(ctx); err != nil && !errors.Is(err, http.ErrServerClosed) {
log.Fatal(err)
}
}
func run(ctx context.Context) error {
g, ctx := errgroup.WithContext(ctx)
+ stop := &stopFns{}
cron := cron.New(cron.DefaultGranularity)
@@ -96,8 +101,9 @@ func run(ctx context.Context) error {
enable := set.New(strings.Fields(env("EV_ENABLE", "salty msgbus gql peers"))...)
var svcs []interface{ RegisterHTTP(*http.ServeMux) }
+ var res []resolver.IsResolver
- svcs = append(svcs, es)
+ res = append(res, es)
if enable.Has("salty") {
span.AddEvent("Enable Salty")
@@ -113,6 +119,7 @@ func run(ctx context.Context) error {
return err
}
svcs = append(svcs, salty)
+ res = append(res, salty)
}
if enable.Has("msgbus") {
@@ -123,6 +130,7 @@ func run(ctx context.Context) error {
return err
}
svcs = append(svcs, msgbus)
+ res = append(res, msgbus)
}
if enable.Has("peers") {
@@ -137,11 +145,15 @@ func run(ctx context.Context) error {
cron.NewJob("0,15,30,45", peers.RefreshJob)
cron.Once(ctx, peers.CleanJob)
cron.NewJob("0 1", peers.CleanJob)
+ g.Go(func() error {
+ return peers.Run(ctx)
+ })
+ stop.add(peers.Stop)
}
if enable.Has("gql") {
span.AddEvent("Enable GraphQL")
- gql, err := gql.New(ctx, svcs...)
+ gql, err := resolver.New(ctx, &gql.Resolver{}, res...)
if err != nil {
span.RecordError(err)
return err
@@ -164,18 +176,23 @@ func run(ctx context.Context) error {
Mup.Add(ctx, 1)
g.Go(s.ListenAndServe)
-
- g.Go(func() error {
- <-ctx.Done()
- ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
- defer cancel()
- return s.Shutdown(ctx)
- })
+ stop.add(s.Shutdown)
span.End()
}
- g.Go(func() error { return cron.Run(ctx) })
+ g.Go(func() error {
+ <-ctx.Done()
+ // shutdown jobs
+
+ ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
+ defer cancel()
+
+ return stop.stop(ctx)
+ })
+ g.Go(func() error {
+ return cron.Run(ctx)
+ })
if err := g.Wait(); err != nil && err != http.ErrServerClosed {
return err
@@ -203,3 +220,21 @@ var appName, version = func() (string, string) {
return "sour.is-ev", "(devel)"
}()
+
+type stopFns struct {
+ fns []func(context.Context) error
+}
+
+func (s *stopFns) add(fn func(context.Context) error) {
+ s.fns = append(s.fns, fn)
+}
+func (s *stopFns) stop(ctx context.Context) error {
+ g, _ := errgroup.WithContext(ctx)
+ for i := range s.fns {
+ fn := s.fns[i]
+ g.Go(func() error {
+ return fn(ctx)
+ })
+ }
+ return g.Wait()
+}
diff --git a/pkg/es/driver/driver.go b/pkg/es/driver/driver.go
index 382a952..10ac089 100644
--- a/pkg/es/driver/driver.go
+++ b/pkg/es/driver/driver.go
@@ -28,7 +28,7 @@ type EventLogWithUpdate interface {
}
type Subscription interface {
- Recv(context.Context) bool
+ Recv(context.Context) <-chan bool
Events(context.Context) (event.Events, error)
Close(context.Context) error
}
diff --git a/pkg/es/driver/streamer/streamer.go b/pkg/es/driver/streamer/streamer.go
index 13b3552..b8b1f7f 100644
--- a/pkg/es/driver/streamer/streamer.go
+++ b/pkg/es/driver/streamer/streamer.go
@@ -4,6 +4,7 @@ package streamer
import (
"context"
"fmt"
+ "sync"
"go.opentelemetry.io/otel/attribute"
"go.opentelemetry.io/otel/trace"
@@ -67,6 +68,7 @@ func (s *streamer) Subscribe(ctx context.Context, streamID string, start int64)
if err != nil {
return nil, err
}
+
sub := &subscription{topic: streamID, events: events}
sub.position = locker.New(&position{
idx: start,
@@ -86,12 +88,14 @@ func (s *streamer) Send(ctx context.Context, streamID string, events event.Event
return s.state.Modify(ctx, func(ctx context.Context, state *state) error {
ctx, span := lg.Span(ctx)
defer span.End()
+
span.AddEvent(fmt.Sprint("subscribers=", len(state.subscribers[streamID])))
for _, sub := range state.subscribers[streamID] {
err := sub.position.Modify(ctx, func(ctx context.Context, position *position) error {
ctx, span := lg.Span(ctx)
defer span.End()
+
span.SetAttributes(
attribute.String("streamID", streamID),
attribute.Int64("actualPosition", int64(events.Last().EventMeta().ActualPosition)),
@@ -106,6 +110,7 @@ func (s *streamer) Send(ctx context.Context, streamID string, events event.Event
position.link = trace.LinkFromContext(ctx, attribute.String("src", "event"))
position.wait = nil
}
+
return nil
})
if err != nil {
@@ -210,56 +215,65 @@ type subscription struct {
events driver.EventLog
unsub func(context.Context) error
+ once sync.Once
}
-func (s *subscription) Recv(ctx context.Context) bool {
+func (s *subscription) Recv(ctx context.Context) <-chan bool {
ctx, span := lg.Span(ctx)
defer span.End()
- var wait func(context.Context) bool
+ done := make(chan bool)
- err := s.position.Modify(ctx, func(ctx context.Context, position *position) error {
- _, span := lg.Span(ctx)
- defer span.End()
+ go func() {
+ var wait func(context.Context) bool
+ defer close(done)
- if position.size == es.AllEvents {
- return nil
- }
- if position.size == 0 {
- position.wait = make(chan struct{})
- wait = func(ctx context.Context) bool {
- ctx, span := lg.Span(ctx)
- defer span.End()
+ err := s.position.Modify(ctx, func(ctx context.Context, position *position) error {
+ _, span := lg.Span(ctx)
+ defer span.End()
- select {
- case <-position.wait:
- if position.link.SpanContext.IsValid() {
- _, span := lg.Span(ctx, trace.WithLinks(position.link))
- span.AddEvent("recv event")
- span.End()
- position.link = trace.Link{}
+ if position.size == es.AllEvents {
+ return nil
+ }
+ if position.size == 0 {
+ position.wait = make(chan struct{})
+ wait = func(ctx context.Context) bool {
+ ctx, span := lg.Span(ctx)
+ defer span.End()
+
+ select {
+ case <-position.wait:
+ if position.link.SpanContext.IsValid() {
+ _, span := lg.Span(ctx, trace.WithLinks(position.link))
+ span.AddEvent("recv event")
+ span.End()
+ position.link = trace.Link{}
+ }
+ return true
+
+ case <-ctx.Done():
+ return false
}
-
- return true
- case <-ctx.Done():
-
- return false
}
}
+ position.idx += position.size
+ position.size = 0
+ return nil
+ })
+ if err != nil {
+ done <- false
+ return
}
- position.idx += position.size
- position.size = 0
- return nil
- })
- if err != nil {
- return false
- }
- if wait != nil {
- return wait(ctx)
- }
+ if wait != nil {
+ done <- wait(ctx)
+ return
+ }
- return true
+ done <- true
+ }()
+
+ return done
}
func (s *subscription) Events(ctx context.Context) (event.Events, error) {
ctx, span := lg.Span(ctx)
@@ -293,5 +307,12 @@ func (s *subscription) Close(ctx context.Context) error {
ctx, span := lg.Span(ctx)
defer span.End()
- return s.unsub(ctx)
+ if s == nil || s.unsub == nil {
+ return nil
+ }
+
+ var err error
+ s.once.Do(func() { err = s.unsub(ctx) })
+
+ return err
}
diff --git a/pkg/es/graph.go b/pkg/es/graph.go
index a8301ad..e1329d7 100644
--- a/pkg/es/graph.go
+++ b/pkg/es/graph.go
@@ -23,6 +23,7 @@ type contextKey struct {
var esKey = contextKey{"event-store"}
+func (es *EventStore) IsResolver() {}
func (es *EventStore) Events(ctx context.Context, streamID string, paging *gql.PageInput) (*gql.Connection, error) {
ctx, span := lg.Span(ctx)
defer span.End()
@@ -51,10 +52,10 @@ func (es *EventStore) Events(ctx context.Context, streamID string, paging *gql.P
return &gql.Connection{
Paging: &gql.PageInfo{
- Next: lis.Last().EventMeta().Position < last,
- Prev: lis.First().EventMeta().Position > first,
- Begin: lis.First().EventMeta().Position,
- End: lis.Last().EventMeta().Position,
+ Next: lis.Last().EventMeta().ActualPosition < last,
+ Prev: lis.First().EventMeta().ActualPosition > first,
+ Begin: lis.First().EventMeta().ActualPosition,
+ End: lis.Last().EventMeta().ActualPosition,
},
Edges: edges,
}, nil
@@ -91,7 +92,7 @@ func (e *EventStore) EventAdded(ctx context.Context, streamID string, after int6
}()
}
- for sub.Recv(ctx) {
+ for <-sub.Recv(ctx) {
events, err := sub.Events(ctx)
if err != nil {
span.RecordError(err)
@@ -120,7 +121,6 @@ func (es *EventStore) TruncateStream(ctx context.Context, streamID string, index
err := es.Truncate(ctx, streamID, index)
return err == nil, err
}
-func (*EventStore) RegisterHTTP(*http.ServeMux) {}
func (e *EventStore) GetMiddleware() func(http.Handler) http.Handler {
return func(next http.Handler) http.Handler {
return http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
diff --git a/app/gql/graphiql/playground.go b/pkg/gql/graphiql/playground.go
similarity index 100%
rename from app/gql/graphiql/playground.go
rename to pkg/gql/graphiql/playground.go
diff --git a/app/gql/playground/playground.go b/pkg/gql/playground/playground.go
similarity index 100%
rename from app/gql/playground/playground.go
rename to pkg/gql/playground/playground.go
diff --git a/pkg/gql/resolver/resolver.go b/pkg/gql/resolver/resolver.go
new file mode 100644
index 0000000..a7289fc
--- /dev/null
+++ b/pkg/gql/resolver/resolver.go
@@ -0,0 +1,141 @@
+package resolver
+
+import (
+ "context"
+ "fmt"
+ "net/http"
+ "os"
+ "reflect"
+ "runtime/debug"
+ "time"
+
+ "github.com/99designs/gqlgen/graphql"
+ "github.com/99designs/gqlgen/graphql/handler"
+ "github.com/99designs/gqlgen/graphql/handler/extension"
+ "github.com/99designs/gqlgen/graphql/handler/lru"
+ "github.com/99designs/gqlgen/graphql/handler/transport"
+ "github.com/vektah/gqlparser/v2/gqlerror"
+
+ "github.com/gorilla/websocket"
+ "github.com/ravilushqa/otelgqlgen"
+
+ "github.com/sour-is/ev/internal/lg"
+ "github.com/sour-is/ev/pkg/gql/graphiql"
+ "github.com/sour-is/ev/pkg/gql/playground"
+)
+
+type BaseResolver interface {
+ ExecutableSchema() graphql.ExecutableSchema
+ BaseResolver() IsResolver
+}
+
+type Resolver[T BaseResolver] struct {
+ res T
+}
+type IsResolver interface {
+ IsResolver()
+}
+
+func New[T BaseResolver](ctx context.Context, base T, resolvers ...IsResolver) (*Resolver[T], error) {
+ _, span := lg.Span(ctx)
+ defer span.End()
+
+ v := reflect.ValueOf(base)
+ v = reflect.Indirect(v)
+ noop := reflect.ValueOf(base.BaseResolver())
+
+outer:
+ for _, idx := range reflect.VisibleFields(v.Type()) {
+ field := v.FieldByIndex(idx.Index)
+
+ for i := range resolvers {
+ rs := reflect.ValueOf(resolvers[i])
+
+ if field.IsNil() && rs.Type().Implements(field.Type()) {
+ span.AddEvent(fmt.Sprint("found ", field.Type().Name()))
+ field.Set(rs)
+ continue outer
+ }
+ }
+
+ span.AddEvent(fmt.Sprint("default ", field.Type().Name()))
+ field.Set(noop)
+ }
+
+ return &Resolver[T]{base}, nil
+}
+
+func (r *Resolver[T]) Resolver() T {
+ return r.res
+}
+
+// ChainMiddlewares will check all embeded resolvers for a GetMiddleware func and add to handler.
+func (r *Resolver[T]) ChainMiddlewares(h http.Handler) http.Handler {
+ v := reflect.ValueOf(r) // Get reflected value of *Resolver
+ v = reflect.Indirect(v) // Get the pointed value (returns a zero value on nil)
+ n := v.NumField() // Get number of fields to iterate over.
+ for i := 0; i < n; i++ {
+ f := v.Field(i)
+ if !f.CanInterface() { // Skip non-interface types.
+ continue
+ }
+ if iface, ok := f.Interface().(interface {
+ GetMiddleware() func(http.Handler) http.Handler
+ }); ok {
+ h = iface.GetMiddleware()(h) // Append only items that fulfill the interface.
+ }
+ }
+
+ return h
+}
+
+func (r *Resolver[T]) RegisterHTTP(mux *http.ServeMux) {
+ gql := NewServer(r.res.ExecutableSchema())
+ gql.SetRecoverFunc(NoopRecover)
+ gql.Use(otelgqlgen.Middleware())
+ mux.Handle("/graphiql", graphiql.Handler("GraphiQL playground", "/gql"))
+ mux.Handle("/gql", lg.Htrace(r.ChainMiddlewares(gql), "gql"))
+ mux.Handle("/playground", playground.Handler("GraphQL playground", "/gql"))
+}
+
+func NoopRecover(ctx context.Context, err interface{}) error {
+ if err, ok := err.(string); ok && err == "not implemented" {
+ return gqlerror.Errorf("not implemented")
+ }
+ fmt.Fprintln(os.Stderr, err)
+ fmt.Fprintln(os.Stderr)
+ debug.PrintStack()
+
+ return gqlerror.Errorf("internal system error")
+}
+
+func NewServer(es graphql.ExecutableSchema) *handler.Server {
+ srv := handler.New(es)
+
+ srv.AddTransport(transport.Websocket{
+ Upgrader: websocket.Upgrader{
+ CheckOrigin: func(r *http.Request) bool {
+ switch r.Header.Get("Origin") {
+ case "https://ev.sour.is", "https://www.graphqlbin.com", "http://localhost:8080":
+ return true
+ default:
+ return false
+ }
+ },
+ },
+ KeepAlivePingInterval: 10 * time.Second,
+ })
+ srv.AddTransport(transport.Options{})
+ srv.AddTransport(transport.GET{})
+ srv.AddTransport(transport.POST{})
+ srv.AddTransport(transport.MultipartForm{})
+
+ srv.SetQueryCache(lru.New(1000))
+
+ srv.Use(extension.Introspection{})
+ srv.Use(extension.AutomaticPersistedQuery{
+ Cache: lru.New(100),
+ })
+
+ return srv
+}