diff --git a/app/gql/resolver.go b/app/gql/resolver.go index a62737d..4dcba3f 100644 --- a/app/gql/resolver.go +++ b/app/gql/resolver.go @@ -7,9 +7,16 @@ import ( "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/gorilla/websocket" "github.com/ravilushqa/otelgqlgen" + "github.com/sour-is/ev/app/gql/graphiql" "github.com/sour-is/ev/app/gql/playground" "github.com/sour-is/ev/app/msgbus" "github.com/sour-is/ev/app/salty" @@ -86,11 +93,13 @@ func (r *Resolver) ChainMiddlewares(h http.Handler) http.Handler { } func (r *Resolver) RegisterHTTP(mux *http.ServeMux) { - gql := handler.NewDefaultServer(generated.NewExecutableSchema(generated.Config{Resolvers: r})) + gql := NewServer(generated.NewExecutableSchema(generated.Config{Resolvers: r})) + gql.SetRecoverFunc(NoopRecover) gql.Use(otelgqlgen.Middleware()) - mux.Handle("/", playground.Handler("GraphQL playground", "/gql")) + 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")) } type noop struct{} @@ -130,3 +139,34 @@ func (*noop) EventAdded(ctx context.Context, streamID string, after int64) (<-ch } func (*noop) RegisterHTTP(*http.ServeMux) {} + +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": + 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 +} diff --git a/app/peerfinder/peer.go b/app/peerfinder/peer.go index 78954e6..333ca3e 100644 --- a/app/peerfinder/peer.go +++ b/app/peerfinder/peer.go @@ -11,26 +11,58 @@ import ( "github.com/sour-is/ev/pkg/es/event" ) -type Request struct { +type Request struct{ + event.AggregateRoot + + RequestIP string `json:"req_ip"` + Hidden bool `json:"hide,omitempty"` + + Responses []Response `json:"responses"` +} + +var _ event.Aggregate = (*Request)(nil) +func (a *Request) ApplyEvent(lis ...event.Event) { + for _, e := range lis { + switch e:=e.(type) { + case *RequestSubmitted: + a.RequestIP = e.RequestIP + a.Hidden = e.Hidden + case *ResultSubmitted: + a.Responses = append(a.Responses, Response{ + PeerID: e.PeerID, + PeerVersion: e.PeerVersion, + Latency: e.Latency, + }) + } + } +} + +type Response struct { + PeerID string `json:"peer_id"` + PeerVersion string `json:"peer_version"` + Latency float64 `json:"latency,omitempty"` +} + +type RequestSubmitted struct { eventMeta event.Meta RequestIP string `json:"req_ip"` Hidden bool `json:"hide,omitempty"` } -func (r *Request) StreamID() string { +func (r *RequestSubmitted) StreamID() string { return r.EventMeta().GetEventID() } -func (r *Request) RequestID() string { +func (r *RequestSubmitted) RequestID() string { return r.EventMeta().GetEventID() } -func (r *Request) Created() time.Time { +func (r *RequestSubmitted) Created() time.Time { return r.EventMeta().Created() } -func (r *Request) CreatedString() string { +func (r *RequestSubmitted) CreatedString() string { return r.Created().Format("2006-01-02 15:04:05") } -func (r *Request) Family() int { +func (r *RequestSubmitted) Family() int { if r == nil { return 0 } @@ -46,26 +78,26 @@ func (r *Request) Family() int { } } -var _ event.Event = (*Request)(nil) +var _ event.Event = (*RequestSubmitted)(nil) -func (e *Request) EventMeta() event.Meta { +func (e *RequestSubmitted) EventMeta() event.Meta { if e == nil { return event.Meta{} } return e.eventMeta } -func (e *Request) SetEventMeta(m event.Meta) { +func (e *RequestSubmitted) SetEventMeta(m event.Meta) { if e != nil { e.eventMeta = m } } -func (e *Request) MarshalBinary() (text []byte, err error) { +func (e *RequestSubmitted) MarshalBinary() (text []byte, err error) { return json.Marshal(e) } -func (e *Request) UnmarshalBinary(b []byte) error { +func (e *RequestSubmitted) UnmarshalBinary(b []byte) error { return json.Unmarshal(b, e) } -func (e *Request) MarshalEnviron() ([]byte, error) { +func (e *RequestSubmitted) MarshalEnviron() ([]byte, error) { if e == nil { return nil, nil } @@ -92,7 +124,7 @@ func (e *Request) MarshalEnviron() ([]byte, error) { return b.Bytes(), nil } -type Result struct { +type ResultSubmitted struct { eventMeta event.Meta RequestID string `json:"req_id"` @@ -101,30 +133,30 @@ type Result struct { Latency float64 `json:"latency,omitempty"` } -func (r *Result) Created() time.Time { +func (r *ResultSubmitted) Created() time.Time { return r.eventMeta.Created() } -var _ event.Event = (*Result)(nil) +var _ event.Event = (*ResultSubmitted)(nil) -func (e *Result) EventMeta() event.Meta { +func (e *ResultSubmitted) EventMeta() event.Meta { if e == nil { return event.Meta{} } return e.eventMeta } -func (e *Result) SetEventMeta(m event.Meta) { +func (e *ResultSubmitted) SetEventMeta(m event.Meta) { if e != nil { e.eventMeta = m } } -func (e *Result) MarshalBinary() (text []byte, err error) { +func (e *ResultSubmitted) MarshalBinary() (text []byte, err error) { return json.Marshal(e) } -func (e *Result) UnmarshalBinary(b []byte) error { +func (e *ResultSubmitted) UnmarshalBinary(b []byte) error { return json.Unmarshal(b, e) } -func (e *Result) String() string { +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) } diff --git a/app/peerfinder/service.go b/app/peerfinder/service.go index eaa4627..fdc086d 100644 --- a/app/peerfinder/service.go +++ b/app/peerfinder/service.go @@ -2,8 +2,12 @@ package peerfinder import ( "context" + "embed" "encoding/json" + "html/template" "io" + "io/fs" + "log" "net" "net/http" "strconv" @@ -16,24 +20,39 @@ import ( "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 ( - queueRequests = "pf-requests" - queueResponses = "pf-response-" 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, &Request{}, &Result{}, &VersionChanged{}); err != nil { + if err := event.Register(ctx, &RequestSubmitted{}, &ResultSubmitted{}, &VersionChanged{}); err != nil { span.RecordError(err) return nil, err } @@ -43,8 +62,13 @@ func New(ctx context.Context, es *es.EventStore) (*service, error) { return svc, nil } func (s *service) RegisterHTTP(mux *http.ServeMux) { - mux.Handle("/peers/", lg.Htrace(s, "peers")) + 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() @@ -66,7 +90,8 @@ func (s *service) ServeHTTP(w http.ResponseWriter, r *http.Request) { return default: - w.WriteHeader(http.StatusNotFound) + t := templates["home.tpl"] + t.Execute(w, nil) return } case http.MethodPost: @@ -113,7 +138,7 @@ func (s *service) getPending(w http.ResponseWriter, r *http.Request, uuid string return } - responses, err := s.es.Read(ctx, queueResponses+uuid, -1, -30) + responses, err := s.es.Read(ctx, queuePeers+uuid, -1, -30) if err != nil { span.RecordError(err) w.WriteHeader(http.StatusInternalServerError) @@ -199,19 +224,17 @@ func (s *service) postRequest(w http.ResponseWriter, r *http.Request) { w.WriteHeader(http.StatusBadRequest) return } - - req := &Request{ + 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), ) - if hidden, err := strconv.ParseBool(r.Form.Get("req_hidden")); err != nil { - req.Hidden = hidden - } - s.es.Append(ctx, queueRequests, event.NewEvents(req)) } func (s *service) postResult(w http.ResponseWriter, r *http.Request, id string) { @@ -238,25 +261,36 @@ func (s *service) postResult(w http.ResponseWriter, r *http.Request, id string) return } - req := &Result{ + req := &ResultSubmitted{ RequestID: id, PeerID: r.Form.Get("peer_id"), PeerVersion: r.Form.Get("peer_version"), Latency: latency, } + span.SetAttributes( attribute.Stringer("result", req), ) - s.es.Append(ctx, queueResponses+id, event.NewEvents(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) *Request { +func filter(requests, responses event.Events) *RequestSubmitted { have := make(map[string]struct{}, len(responses)) - for _, res := range toList[Result](responses...) { + for _, res := range toList[ResultSubmitted](responses...) { have[res.RequestID] = struct{}{} } - for _, req := range reverse(toList[Request](requests...)...) { + for _, req := range reverse(toList[RequestSubmitted](requests...)...) { if _, ok := have[req.RequestID()]; !ok { return req } @@ -297,3 +331,50 @@ func encodeTo(w io.Writer, fns ...func() ([]byte, error)) (int, error) { } 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 +} diff --git a/go.mod b/go.mod index 599e50c..2314b08 100644 --- a/go.mod +++ b/go.mod @@ -55,7 +55,7 @@ require ( github.com/logrusorgru/aurora v2.0.3+incompatible // indirect github.com/matttproud/golang_protobuf_extensions v1.0.2-0.20181231171920-c182affec369 // indirect github.com/mitchellh/mapstructure v1.5.0 // indirect - github.com/oklog/ulid v1.3.1 // indirect + github.com/oklog/ulid v1.3.1 github.com/onsi/ginkgo v1.14.0 // indirect github.com/onsi/gomega v1.10.3 // indirect github.com/petermattis/goid v0.0.0-20220331194723-8ee3e6ded87a // indirect diff --git a/httpmux.go b/httpmux.go index a86cd18..811ac58 100644 --- a/httpmux.go +++ b/httpmux.go @@ -34,3 +34,8 @@ func newMux() *mux { return mux } + +type RegisterHTTP func(*http.ServeMux) +func (fn RegisterHTTP) RegisterHTTP(mux *http.ServeMux) { + fn(mux) +} \ No newline at end of file diff --git a/main.go b/main.go index 8c1adbe..9893f57 100644 --- a/main.go +++ b/main.go @@ -22,6 +22,7 @@ import ( diskstore "github.com/sour-is/ev/pkg/es/driver/disk-store" memstore "github.com/sour-is/ev/pkg/es/driver/mem-store" "github.com/sour-is/ev/pkg/es/driver/projecter" + 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/set" @@ -61,10 +62,15 @@ func run(ctx context.Context) error { } es, err := es.Open( - ctx, - env("EV_DATA", "mem:"), - streamer.New(ctx), - projecter.New(ctx, projecter.DefaultProjection), + ctx, + env("EV_DATA", "mem:"), + streamer.New(ctx), + projecter.New( + ctx, + projecter.DefaultProjection, + peerfinder.Projector, + ), + resolvelinks.New(), ) if err != nil { span.RecordError(err) @@ -129,7 +135,9 @@ func run(ctx context.Context) error { } svcs = append(svcs, gql) } - svcs = append(svcs, lg.NewHTTP(ctx)) + svcs = append(svcs, lg.NewHTTP(ctx), RegisterHTTP(func(mux *http.ServeMux) { + mux.Handle("/", http.RedirectHandler("/playground", http.StatusTemporaryRedirect)) + })) s.Handler = httpMux(svcs...) diff --git a/pkg/es/driver/disk-store/disk-store.go b/pkg/es/driver/disk-store/disk-store.go index 0dd165d..020e10d 100644 --- a/pkg/es/driver/disk-store/disk-store.go +++ b/pkg/es/driver/disk-store/disk-store.go @@ -94,7 +94,7 @@ func (d *diskStore) Open(ctx context.Context, dsn string) (driver.Driver, error) _, span := lg.Span(ctx) defer span.End() - l.Modify(ctx, func(w *wal.Log) error { + l.Modify(ctx, func(ctx context.Context, w *wal.Log) error { _, span := lg.Span(ctx) defer span.End() @@ -128,7 +128,7 @@ func (d *diskStore) EventLog(ctx context.Context, streamID string) (driver.Event el := &eventLog{streamID: streamID, diskStore: d} - return el, d.openlogs.Modify(ctx, func(openlogs *openlogs) error { + return el, d.openlogs.Modify(ctx, func(ctx context.Context, openlogs *openlogs) error { _, span := lg.Span(ctx) defer span.End() @@ -166,7 +166,7 @@ func (e *eventLog) Append(ctx context.Context, events event.Events, version uint event.SetStreamID(e.streamID, events...) var count uint64 - err := e.events.Modify(ctx, func(l *wal.Log) error { + err := e.events.Modify(ctx, func(ctx context.Context, l *wal.Log) error { _, span := lg.Span(ctx) defer span.End() @@ -215,7 +215,7 @@ func (e *eventLog) Read(ctx context.Context, after, count int64) (event.Events, var events event.Events - err := e.events.Modify(ctx, func(stream *wal.Log) error { + err := e.events.Modify(ctx, func(ctx context.Context, stream *wal.Log) error { _, span := lg.Span(ctx) defer span.End() @@ -244,17 +244,7 @@ func (e *eventLog) Read(ctx context.Context, after, count int64) (event.Events, span.AddEvent(fmt.Sprintf("read event %d of %d", i, len(events))) // --- - var b []byte - b, err = stream.Read(start) - if err != nil { - if errors.Is(err, wal.ErrNotFound) || errors.Is(err, wal.ErrOutOfRange) { - err = fmt.Errorf("%w: empty", es.ErrNotFound) - } - - span.RecordError(err) - return err - } - events[i], err = event.UnmarshalBinary(ctx, b, start) + events[i], err = readStream(ctx, stream, start) if err != nil { span.RecordError(err) return err @@ -283,6 +273,21 @@ func (e *eventLog) Read(ctx context.Context, after, count int64) (event.Events, return events, nil } +func (e *eventLog) ReadN(ctx context.Context, index ...uint64) (event.Events, error) { + _, span := lg.Span(ctx) + defer span.End() + + var events event.Events + err := e.events.Modify(ctx, func(ctx context.Context, stream *wal.Log) error { + var err error + + events, err = readStreamN(ctx, stream, index...) + + return err + }) + + return events, err +} func (e *eventLog) FirstIndex(ctx context.Context) (uint64, error) { _, span := lg.Span(ctx) defer span.End() @@ -290,7 +295,7 @@ func (e *eventLog) FirstIndex(ctx context.Context) (uint64, error) { var idx uint64 var err error - err = e.events.Modify(ctx, func(events *wal.Log) error { + err = e.events.Modify(ctx, func(ctx context.Context, events *wal.Log) error { idx, err = events.FirstIndex() return err }) @@ -304,7 +309,7 @@ func (e *eventLog) LastIndex(ctx context.Context) (uint64, error) { var idx uint64 var err error - err = e.events.Modify(ctx, func(events *wal.Log) error { + err = e.events.Modify(ctx, func(ctx context.Context, events *wal.Log) error { idx, err = events.LastIndex() return err }) @@ -314,3 +319,50 @@ func (e *eventLog) LastIndex(ctx context.Context) (uint64, error) { func (e *eventLog) LoadForUpdate(ctx context.Context, a event.Aggregate, fn func(context.Context, event.Aggregate) error) (uint64, error) { panic("not implemented") } +func readStream(ctx context.Context, stream *wal.Log, index uint64) (event.Event, error) { + _, span := lg.Span(ctx) + defer span.End() + + var b []byte + var err error + b, err = stream.Read(index) + if err != nil { + if errors.Is(err, wal.ErrNotFound) || errors.Is(err, wal.ErrOutOfRange) { + err = fmt.Errorf("%w: empty", es.ErrNotFound) + } + + span.RecordError(err) + return nil, err + } + e, err := event.UnmarshalBinary(ctx, b, index) + if err != nil { + span.RecordError(err) + return nil, err + } + return e, err +} +func readStreamN(ctx context.Context, stream *wal.Log, index ...uint64) (event.Events, error) { + _, span := lg.Span(ctx) + defer span.End() + + var b []byte + var err error + events := make(event.Events, len(index)) + for i, idx := range index { + b, err = stream.Read(idx) + if err != nil { + if errors.Is(err, wal.ErrNotFound) || errors.Is(err, wal.ErrOutOfRange) { + err = fmt.Errorf("%w: empty", es.ErrNotFound) + } + + span.RecordError(err) + return nil, err + } + events[i], err = event.UnmarshalBinary(ctx, b, idx) + if err != nil { + span.RecordError(err) + return nil, err + } + } + return events, err +} \ No newline at end of file diff --git a/pkg/es/driver/driver.go b/pkg/es/driver/driver.go index acdf108..96deccd 100644 --- a/pkg/es/driver/driver.go +++ b/pkg/es/driver/driver.go @@ -14,6 +14,7 @@ type Driver interface { type EventLog interface { Read(ctx context.Context, after, count int64) (event.Events, error) + ReadN(ctx context.Context, index ...uint64) (event.Events, error) Append(ctx context.Context, events event.Events, version uint64) (uint64, error) FirstIndex(context.Context) (uint64, error) LastIndex(context.Context) (uint64, error) diff --git a/pkg/es/driver/mem-store/mem-store.go b/pkg/es/driver/mem-store/mem-store.go index c9af2c4..eeaf78e 100644 --- a/pkg/es/driver/mem-store/mem-store.go +++ b/pkg/es/driver/mem-store/mem-store.go @@ -49,7 +49,7 @@ func (m *memstore) EventLog(ctx context.Context, streamID string) (driver.EventL el := &eventLog{streamID: streamID} - err := m.state.Modify(ctx, func(state *state) error { + err := m.state.Modify(ctx, func(ctx context.Context, state *state) error { _, span := lg.Span(ctx) defer span.End() @@ -76,7 +76,7 @@ func (m *eventLog) Append(ctx context.Context, events event.Events, version uint event.SetStreamID(m.streamID, events...) - return uint64(len(events)), m.events.Modify(ctx, func(stream *event.Events) error { + return uint64(len(events)), m.events.Modify(ctx, func(ctx context.Context, stream *event.Events) error { _, span := lg.Span(ctx) defer span.End() @@ -111,6 +111,23 @@ func (m *eventLog) Append(ctx context.Context, events event.Events, version uint }) } +// ReadOne implements readone +func (m *eventLog) ReadN(ctx context.Context, index ...uint64) (event.Events, error) { + _, span := lg.Span(ctx) + defer span.End() + + var events event.Events + err := m.events.Modify(ctx, func(ctx context.Context, stream *event.Events) error { + var err error + + events, err = readStreamN(ctx, stream, index...) + + return err + }) + + return events, err +} + // Read implements driver.EventStore func (m *eventLog) Read(ctx context.Context, after int64, count int64) (event.Events, error) { ctx, span := lg.Span(ctx) @@ -118,7 +135,7 @@ func (m *eventLog) Read(ctx context.Context, after int64, count int64) (event.Ev var events event.Events - err := m.events.Modify(ctx, func(stream *event.Events) error { + err := m.events.Modify(ctx, func(ctx context.Context, stream *event.Events) error { _, span := lg.Span(ctx) defer span.End() @@ -141,12 +158,8 @@ func (m *eventLog) Read(ctx context.Context, after int64, count int64) (event.Ev span.AddEvent(fmt.Sprintf("read event %d of %d", i, math.Abs(count))) // --- clone event - e := (*stream)[start-1] - b, err := event.MarshalBinary(e) - if err != nil { - return err - } - events[i], err = event.UnmarshalBinary(ctx, b, e.EventMeta().Position) + var err error + events[i], err = readStream(ctx, stream, start) if err != nil { return err } @@ -194,3 +207,42 @@ func (m *eventLog) LastIndex(ctx context.Context) (uint64, error) { func (m *eventLog) LoadForUpdate(ctx context.Context, a event.Aggregate, fn func(context.Context, event.Aggregate) error) (uint64, error) { panic("not implemented") } + +func readStream(ctx context.Context, stream *event.Events, index uint64) (event.Event, error) { + _, span := lg.Span(ctx) + defer span.End() + + var b []byte + var err error + e := (*stream)[index-1] + b, err = event.MarshalBinary(e) + if err != nil { + return nil, err + } + e, err = event.UnmarshalBinary(ctx, b, e.EventMeta().Position) + if err != nil { + return nil, err + } + return e, err +} +func readStreamN(ctx context.Context, stream *event.Events, index ...uint64) (event.Events, error) { + _, span := lg.Span(ctx) + defer span.End() + + var b []byte + var err error + + events := make(event.Events, len(index)) + for i, index := range index { + e := (*stream)[index-1] + b, err = event.MarshalBinary(e) + if err != nil { + return nil, err + } + events[i], err = event.UnmarshalBinary(ctx, b, e.EventMeta().Position) + if err != nil { + return nil, err + } + } + return events, err +} diff --git a/pkg/es/driver/projecter/projecter.go b/pkg/es/driver/projecter/projecter.go index 8a4b7ce..8809ab9 100644 --- a/pkg/es/driver/projecter/projecter.go +++ b/pkg/es/driver/projecter/projecter.go @@ -53,6 +53,12 @@ func (w *wrapper) Read(ctx context.Context, after int64, count int64) (event.Eve return w.up.Read(ctx, after, count) } +func (w *wrapper) ReadN(ctx context.Context, index ...uint64) (event.Events, error) { + ctx, span := lg.Span(ctx) + defer span.End() + + return w.up.ReadN(ctx, index...) +} func (w *wrapper) Append(ctx context.Context, events event.Events, version uint64) (uint64, error) { ctx, span := lg.Span(ctx) defer span.End() @@ -123,10 +129,11 @@ func (w *wrapper) LoadForUpdate(ctx context.Context, a event.Aggregate, fn func( } func DefaultProjection(e event.Event) []event.Event { - eventType := event.TypeOf(e) m := e.EventMeta() streamID := m.StreamID streamPos := m.Position + eventType := event.TypeOf(e) + pkg, _, _ := strings.Cut(eventType, ".") e1 := event.NewPtr(streamID, streamPos) event.SetStreamID("$all", e1) @@ -135,7 +142,6 @@ func DefaultProjection(e event.Event) []event.Event { event.SetStreamID("$type-"+eventType, e2) e3 := event.NewPtr(streamID, streamPos) - pkg, _, _ := strings.Cut(eventType, ".") event.SetStreamID("$pkg-"+pkg, e3) return []event.Event{e1, e2, e3} diff --git a/pkg/es/driver/projecter/projector_test.go b/pkg/es/driver/projecter/projector_test.go index c9deec2..942b621 100644 --- a/pkg/es/driver/projecter/projector_test.go +++ b/pkg/es/driver/projecter/projector_test.go @@ -39,6 +39,7 @@ type mockEventLog struct { onFirstIndex func(context.Context) (uint64, error) onLastIndex func(context.Context) (uint64, error) onRead func(context.Context, int64, int64) (event.Events, error) + onReadN func(context.Context, ...uint64) (event.Events, error) } // Append implements driver.EventLog @@ -73,6 +74,13 @@ func (m *mockEventLog) Read(ctx context.Context, pos int64, count int64) (event. panic("unimplemented") } +func (m *mockEventLog) ReadN(ctx context.Context, index ...uint64) (event.Events, error) { + if m.onReadN != nil { + return m.onReadN(ctx, index...) + } + + panic("unimplemented") +} var _ driver.EventLog = (*mockEventLog)(nil) diff --git a/pkg/es/driver/streamer/streamer.go b/pkg/es/driver/streamer/streamer.go index 206b417..5865b43 100644 --- a/pkg/es/driver/streamer/streamer.go +++ b/pkg/es/driver/streamer/streamer.go @@ -73,7 +73,7 @@ func (s *streamer) Subscribe(ctx context.Context, streamID string, start int64) }) sub.unsub = s.delete(streamID, sub) - return sub, s.state.Modify(ctx, func(state *state) error { + return sub, s.state.Modify(ctx, func(ctx context.Context, state *state) error { state.subscribers[streamID] = append(state.subscribers[streamID], sub) return nil }) @@ -82,13 +82,13 @@ func (s *streamer) Send(ctx context.Context, streamID string, events event.Event ctx, span := lg.Span(ctx) defer span.End() - return s.state.Modify(ctx, func(state *state) error { + 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(position *position) error { + err := sub.position.Modify(ctx, func(ctx context.Context, position *position) error { _, span := lg.Span(ctx) defer span.End() @@ -116,7 +116,7 @@ func (s *streamer) delete(streamID string, sub *subscription) func(context.Conte if err := ctx.Err(); err != nil { return err } - return s.state.Modify(ctx, func(state *state) error { + return s.state.Modify(ctx, func(ctx context.Context, state *state) error { _, span := lg.Span(ctx) defer span.End() @@ -148,6 +148,12 @@ func (w *wrapper) Read(ctx context.Context, after int64, count int64) (event.Eve return w.up.Read(ctx, after, count) } +func (w *wrapper) ReadN(ctx context.Context, index ...uint64) (event.Events, error) { + ctx, span := lg.Span(ctx) + defer span.End() + + return w.up.ReadN(ctx, index...) +} func (w *wrapper) Append(ctx context.Context, events event.Events, version uint64) (uint64, error) { ctx, span := lg.Span(ctx) defer span.End() @@ -215,7 +221,7 @@ func (s *subscription) Recv(ctx context.Context) bool { var wait func(context.Context) bool - err := s.position.Modify(ctx, func(position *position) error { + err := s.position.Modify(ctx, func(ctx context.Context, position *position) error { _, span := lg.Span(ctx) defer span.End() @@ -263,7 +269,7 @@ func (s *subscription) Events(ctx context.Context) (event.Events, error) { defer span.End() var events event.Events - return events, s.position.Modify(ctx, func(position *position) error { + return events, s.position.Modify(ctx, func(ctx context.Context, position *position) error { ctx, span := lg.Span(ctx) defer span.End() diff --git a/pkg/es/es.go b/pkg/es/es.go index 45bc69d..9e3b10c 100644 --- a/pkg/es/es.go +++ b/pkg/es/es.go @@ -53,7 +53,7 @@ func Register(ctx context.Context, name string, d driver.Driver) error { ctx, span := lg.Span(ctx) defer span.End() - return drivers.Modify(ctx, func(c *config) error { + return drivers.Modify(ctx, func(ctx context.Context, c *config) error { if _, set := c.drivers[name]; set { return fmt.Errorf("driver %s already set", name) } @@ -189,6 +189,30 @@ func (es *EventStore) Read(ctx context.Context, streamID string, after, count in return events, err } +func (es *EventStore) ReadN(ctx context.Context, streamID string, index ...uint64) (event.Events, error) { + ctx, span := lg.Span(ctx) + defer span.End() + + lis := make([]int64, len(index)) + for i,j :=range index { + lis[i] = int64(j) + } + + span.SetAttributes( + attribute.String("streamID", streamID), + attribute.Int64Slice("index", lis), + ) + + l, err := es.Driver.EventLog(ctx, streamID) + if err != nil { + return nil, err + } + + events, err := l.ReadN(ctx, index...) + Mes_read.Add(ctx, events.Count()) + + return events, err +} func (es *EventStore) Append(ctx context.Context, streamID string, events event.Events) (uint64, error) { ctx, span := lg.Span(ctx) defer span.End() diff --git a/pkg/es/event/events.go b/pkg/es/event/events.go index 87f65ef..a3afd13 100644 --- a/pkg/es/event/events.go +++ b/pkg/es/event/events.go @@ -137,7 +137,7 @@ func (m Meta) Created() time.Time { func (m Meta) GetEventID() string { return m.EventID.String() } func Init(ctx context.Context) error { - return Register(ctx, NilEvent, &eventPtr{}) + return Register(ctx, NilEvent, &EventPtr{}) } type nilEvent struct{} @@ -156,40 +156,40 @@ func (e *nilEvent) UnmarshalBinary(b []byte) error { return json.Unmarshal(b, e) } -type eventPtr struct { - streamID string - pos uint64 +type EventPtr struct { + StreamID string `json:"stream_id"` + Pos uint64 `json:"pos"` eventMeta Meta } -var _ Event = (*eventPtr)(nil) +var _ Event = (*EventPtr)(nil) -func NewPtr(streamID string, pos uint64) *eventPtr { - return &eventPtr{streamID: streamID, pos: pos} +func NewPtr(streamID string, pos uint64) *EventPtr { + return &EventPtr{StreamID: streamID, Pos: pos} } // MarshalBinary implements Event -func (e *eventPtr) MarshalBinary() (data []byte, err error) { - return []byte(fmt.Sprintf("%s@%d", e.streamID, e.pos)), nil +func (e *EventPtr) MarshalBinary() (data []byte, err error) { + return []byte(fmt.Sprintf("%s@%d", e.StreamID, e.Pos)), nil } // UnmarshalBinary implements Event -func (e *eventPtr) UnmarshalBinary(data []byte) error { +func (e *EventPtr) UnmarshalBinary(data []byte) error { s := string(data) idx := strings.LastIndex(s, "@") if idx == -1 { return fmt.Errorf("missing @ in: %s", s) } - e.streamID = s[:idx] + e.StreamID = s[:idx] var err error - e.pos, err = strconv.ParseUint(s[idx+1:], 10, 64) + e.Pos, err = strconv.ParseUint(s[idx+1:], 10, 64) return err } // EventMeta implements Event -func (e *eventPtr) EventMeta() Meta { +func (e *EventPtr) EventMeta() Meta { if e == nil { return Meta{} } @@ -197,19 +197,19 @@ func (e *eventPtr) EventMeta() Meta { } // SetEventMeta implements Event -func (e *eventPtr) SetEventMeta(m Meta) { +func (e *EventPtr) SetEventMeta(m Meta) { if e == nil { return } e.eventMeta = m } -func (e *eventPtr) Values() any { +func (e *EventPtr) Values() any { return struct { StreamID string `json:"stream_id"` Pos uint64 `json:"pos"` }{ - e.streamID, - e.pos, + e.StreamID, + e.Pos, } } diff --git a/pkg/es/event/reflect.go b/pkg/es/event/reflect.go index 85a5094..7922178 100644 --- a/pkg/es/event/reflect.go +++ b/pkg/es/event/reflect.go @@ -106,7 +106,7 @@ func RegisterName(ctx context.Context, name string, e Event) error { span.AddEvent("register: " + name) - if err := eventTypes.Modify(ctx, func(c *config) error { + if err := eventTypes.Modify(ctx, func(ctx context.Context, c *config) error { _, span := lg.Span(ctx) defer span.End() @@ -124,7 +124,7 @@ func GetContainer(ctx context.Context, s string) Event { var e Event - eventTypes.Modify(ctx, func(c *config) error { + eventTypes.Modify(ctx, func(ctx context.Context,c *config) error { _, span := lg.Span(ctx) defer span.End() diff --git a/pkg/es/graph.go b/pkg/es/graph.go index 9d5b9b7..74faf2b 100644 --- a/pkg/es/graph.go +++ b/pkg/es/graph.go @@ -160,7 +160,7 @@ func (e *GQLEvent) Linked(ctx context.Context) (*GQLEvent, error) { return nil, nil } - events, err := gql.FromContext[contextKey, *EventStore](ctx, esKey).Read(ctx, streamID, int64(pos)-1, 1) + events, err := gql.FromContext[contextKey, *EventStore](ctx, esKey).ReadN(ctx, streamID, pos) return &GQLEvent{e: events.First()}, err } func (e *GQLEvent) IsEdge() {} diff --git a/pkg/locker/locker.go b/pkg/locker/locker.go index 521b596..b9ef6de 100644 --- a/pkg/locker/locker.go +++ b/pkg/locker/locker.go @@ -19,7 +19,7 @@ func New[T any](initial *T) *Locked[T] { } // Modify will call the function with the locked value -func (s *Locked[T]) Modify(ctx context.Context, fn func(*T) error) error { +func (s *Locked[T]) Modify(ctx context.Context, fn func(context.Context, *T) error) error { _, span := lg.Span(ctx) defer span.End() @@ -30,7 +30,7 @@ func (s *Locked[T]) Modify(ctx context.Context, fn func(*T) error) error { select { case state := <-s.state: defer func() { s.state <- state }() - return fn(state) + return fn(ctx, state) case <-ctx.Done(): return ctx.Err() } @@ -40,7 +40,7 @@ func (s *Locked[T]) Modify(ctx context.Context, fn func(*T) error) error { func (s *Locked[T]) Copy(ctx context.Context) (T, error) { var t T - err := s.Modify(ctx, func(c *T) error { + err := s.Modify(ctx, func(ctx context.Context, c *T) error { if c != nil { t = *c } diff --git a/pkg/locker/locker_test.go b/pkg/locker/locker_test.go index bad312b..387c348 100644 --- a/pkg/locker/locker_test.go +++ b/pkg/locker/locker_test.go @@ -22,7 +22,7 @@ func TestLocker(t *testing.T) { ctx, cancel := context.WithCancel(context.Background()) defer cancel() - err := value.Modify(ctx, func(c *config) error { + err := value.Modify(ctx, func(ctx context.Context, c *config) error { c.Value = "one" c.Counter++ return nil @@ -37,7 +37,7 @@ func TestLocker(t *testing.T) { wait := make(chan struct{}) - go value.Modify(ctx, func(c *config) error { + go value.Modify(ctx, func(ctx context.Context, c *config) error { c.Value = "two" c.Counter++ close(wait) @@ -47,7 +47,7 @@ func TestLocker(t *testing.T) { <-wait cancel() - err = value.Modify(ctx, func(c *config) error { + err = value.Modify(ctx, func(ctx context.Context, c *config) error { c.Value = "three" c.Counter++ return nil