From 063dca997f26c1f921933f0a1383d6d910a26cbf Mon Sep 17 00:00:00 2001 From: Jon Lundy Date: Tue, 6 Sep 2022 10:35:14 -0600 Subject: [PATCH] feat: add projection layer to eventstore --- Makefile | 1 + README.md | 6 +- app/gql/resolver.go | 6 +- app/msgbus/service.go | 54 ++++++----- app/peerfinder/service.go | 10 +- app/salty/salty-addr.go | 6 +- app/salty/service.go | 16 ++-- env | 5 - httpmux.go | 37 +++++++ internal/{logz => lg}/init.go | 2 +- internal/{logz => lg}/logger.go | 4 +- internal/{logz => lg}/metric.go | 2 +- internal/{logz => lg}/tracer.go | 23 ++++- main.go | 93 +++++------------- pkg/es/driver/disk-store/disk-store.go | 28 +++--- pkg/es/driver/mem-store/mem-store.go | 30 +++--- pkg/es/driver/projecter/projecter.go | 127 +++++++++++++++++++++++++ pkg/es/driver/streamer/streamer.go | 71 +++++++++----- pkg/es/es.go | 40 ++++---- pkg/es/event/events.go | 64 +++++++++++++ pkg/es/event/reflect.go | 22 +++-- pkg/es/graph.go | 26 ++--- pkg/locker/locker.go | 5 + pkg/set/set.go | 35 +++++++ 24 files changed, 493 insertions(+), 220 deletions(-) delete mode 100644 env create mode 100644 httpmux.go rename internal/{logz => lg}/init.go (99%) rename internal/{logz => lg}/logger.go (95%) rename internal/{logz => lg}/metric.go (99%) rename internal/{logz => lg}/tracer.go (84%) create mode 100644 pkg/es/driver/projecter/projecter.go create mode 100644 pkg/set/set.go diff --git a/Makefile b/Makefile index 6dc96ed..f0d6d54 100644 --- a/Makefile +++ b/Makefile @@ -2,6 +2,7 @@ export PATH:=$(shell go env GOPATH)/bin:$(PATH) export EV_DATA=mem: export EV_HTTP=:8080 export EV_TRACE_SAMPLE=always +export EV_TRACE_ENDPOINT=localhost:4318 -include local.mk air: gen diff --git a/README.md b/README.md index 8b9a00c..d102203 100644 --- a/README.md +++ b/README.md @@ -1,3 +1,5 @@ -# ev +# ev - Eventsourcing Playground -Eventsourcing Playground +This project is learnings in building an eventstore and applications around it. + +Feel free to explore. \ No newline at end of file diff --git a/app/gql/resolver.go b/app/gql/resolver.go index 7b701af..a62737d 100644 --- a/app/gql/resolver.go +++ b/app/gql/resolver.go @@ -14,7 +14,7 @@ import ( "github.com/sour-is/ev/app/msgbus" "github.com/sour-is/ev/app/salty" "github.com/sour-is/ev/internal/graph/generated" - "github.com/sour-is/ev/internal/logz" + "github.com/sour-is/ev/internal/lg" "github.com/sour-is/ev/pkg/es" "github.com/sour-is/ev/pkg/gql" "github.com/vektah/gqlparser/v2/gqlerror" @@ -27,7 +27,7 @@ type Resolver struct { } func New(ctx context.Context, resolvers ...interface{ RegisterHTTP(*http.ServeMux) }) (*Resolver, error) { - _, span := logz.Span(ctx) + _, span := lg.Span(ctx) defer span.End() r := &Resolver{} @@ -90,7 +90,7 @@ func (r *Resolver) RegisterHTTP(mux *http.ServeMux) { gql.SetRecoverFunc(NoopRecover) gql.Use(otelgqlgen.Middleware()) mux.Handle("/", playground.Handler("GraphQL playground", "/gql")) - mux.Handle("/gql", logz.Htrace(r.ChainMiddlewares(gql), "gql")) + mux.Handle("/gql", lg.Htrace(r.ChainMiddlewares(gql), "gql")) } type noop struct{} diff --git a/app/msgbus/service.go b/app/msgbus/service.go index b07c58b..70653a6 100644 --- a/app/msgbus/service.go +++ b/app/msgbus/service.go @@ -12,7 +12,7 @@ import ( "time" "github.com/gorilla/websocket" - "github.com/sour-is/ev/internal/logz" + "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/gql" @@ -35,7 +35,7 @@ type MsgbusResolver interface { } func New(ctx context.Context, es *es.EventStore) (*service, error) { - ctx, span := logz.Span(ctx) + ctx, span := lg.Span(ctx) defer span.End() if err := event.Register(ctx, &PostEvent{}); err != nil { @@ -45,7 +45,7 @@ func New(ctx context.Context, es *es.EventStore) (*service, error) { return nil, err } - m := logz.Meter(ctx) + m := lg.Meter(ctx) svc := &service{es: es} @@ -72,11 +72,11 @@ var upgrader = websocket.Upgrader{ } func (s *service) RegisterHTTP(mux *http.ServeMux) { - mux.Handle("/inbox/", logz.Htrace(http.StripPrefix("/inbox/", s), "inbox")) + mux.Handle("/inbox/", lg.Htrace(http.StripPrefix("/inbox/", s), "inbox")) } func (s *service) ServeHTTP(w http.ResponseWriter, r *http.Request) { ctx := r.Context() - ctx, span := logz.Span(ctx) + ctx, span := lg.Span(ctx) defer span.End() r = r.WithContext(ctx) @@ -97,7 +97,7 @@ func (s *service) ServeHTTP(w http.ResponseWriter, r *http.Request) { // Posts is the resolver for the events field. func (r *service) Posts(ctx context.Context, streamID string, paging *gql.PageInput) (*gql.Connection, error) { - ctx, span := logz.Span(ctx) + ctx, span := lg.Span(ctx) defer span.End() r.Mresolver_posts.Add(ctx, 1) @@ -143,7 +143,7 @@ func (r *service) Posts(ctx context.Context, streamID string, paging *gql.PageIn } func (r *service) PostAdded(ctx context.Context, streamID string, after int64) (<-chan *PostEvent, error) { - ctx, span := logz.Span(ctx) + ctx, span := lg.Span(ctx) defer span.End() r.Mresolver_post_added.Add(ctx, 1) @@ -162,15 +162,19 @@ func (r *service) PostAdded(ctx context.Context, streamID string, after int64) ( ch := make(chan *PostEvent) go func() { - ctx, span := logz.Span(ctx) + ctx, span := lg.Span(ctx) defer span.End() - defer func() { - ctx, cancel := context.WithTimeout(context.Background(), 1*time.Second) - defer cancel() - err := sub.Close(ctx) - span.RecordError(err) - }() + { + ctx, span := lg.Fork(ctx) + defer func() { + defer span.End() + ctx, cancel := context.WithTimeout(ctx, 1*time.Second) + defer cancel() + err := sub.Close(ctx) + span.RecordError(err) + }() + } for sub.Recv(ctx) { events, err := sub.Events(ctx) @@ -199,7 +203,7 @@ func (r *service) PostAdded(ctx context.Context, streamID string, after int64) ( func (s *service) get(w http.ResponseWriter, r *http.Request) { ctx := r.Context() - ctx, span := logz.Span(ctx) + ctx, span := lg.Span(ctx) defer span.End() name, _, _ := strings.Cut(r.URL.Path, "/") @@ -253,7 +257,7 @@ func (s *service) get(w http.ResponseWriter, r *http.Request) { func (s *service) post(w http.ResponseWriter, r *http.Request) { ctx := r.Context() - ctx, span := logz.Span(ctx) + ctx, span := lg.Span(ctx) defer span.End() name, tags, _ := strings.Cut(r.URL.Path, "/") @@ -320,7 +324,7 @@ func (s *service) post(w http.ResponseWriter, r *http.Request) { } func (s *service) websocket(w http.ResponseWriter, r *http.Request) { ctx := r.Context() - ctx, span := logz.Span(ctx) + ctx, span := lg.Span(ctx) defer span.End() name, _, _ := strings.Cut(r.URL.Path, "/") @@ -381,12 +385,16 @@ func (s *service) websocket(w http.ResponseWriter, r *http.Request) { span.RecordError(err) return } - defer func() { - ctx, cancel := context.WithTimeout(context.Background(), 1*time.Second) - defer cancel() - span.AddEvent("stop ws") - sub.Close(ctx) - }() + { + ctx, span := lg.Fork(ctx) + defer func() { + defer span.End() + ctx, cancel := context.WithTimeout(ctx, 1*time.Second) + defer cancel() + err := sub.Close(ctx) + span.RecordError(err) + }() + } span.AddEvent("start ws") for sub.Recv(ctx) { diff --git a/app/peerfinder/service.go b/app/peerfinder/service.go index 1f1bb2b..366b209 100644 --- a/app/peerfinder/service.go +++ b/app/peerfinder/service.go @@ -11,7 +11,7 @@ import ( ulid "github.com/oklog/ulid/v2" contentnegotiation "gitlab.com/jamietanna/content-negotiation-go" - "github.com/sour-is/ev/internal/logz" + "github.com/sour-is/ev/internal/lg" "github.com/sour-is/ev/pkg/es" "github.com/sour-is/ev/pkg/es/event" ) @@ -28,7 +28,7 @@ type service struct { } func New(ctx context.Context, es *es.EventStore) (*service, error) { - ctx, span := logz.Span(ctx) + ctx, span := lg.Span(ctx) defer span.End() if err := event.Register(ctx, &Request{}, &Result{}, &VersionChanged{}); err != nil { @@ -41,13 +41,13 @@ func New(ctx context.Context, es *es.EventStore) (*service, error) { return svc, nil } func (s *service) RegisterHTTP(mux *http.ServeMux) { - mux.Handle("/peers/", logz.Htrace(s, "peers")) + mux.Handle("/peers/", lg.Htrace(s, "peers")) } func (s *service) ServeHTTP(w http.ResponseWriter, r *http.Request) { ctx := r.Context() - _, span := logz.Span(ctx) + _, span := lg.Span(ctx) defer span.End() r = r.WithContext(ctx) @@ -90,7 +90,7 @@ func (s *service) ServeHTTP(w http.ResponseWriter, r *http.Request) { func (s *service) getPending(w http.ResponseWriter, r *http.Request, uuid string) { ctx := r.Context() - _, span := logz.Span(ctx) + _, span := lg.Span(ctx) defer span.End() info, err := es.Upsert(ctx, s.es, "pf-info", func(ctx context.Context, agg *Info) error { diff --git a/app/salty/salty-addr.go b/app/salty/salty-addr.go index 62d2dcf..0fb6d24 100644 --- a/app/salty/salty-addr.go +++ b/app/salty/salty-addr.go @@ -10,7 +10,7 @@ import ( "strings" "github.com/keys-pub/keys" - "github.com/sour-is/ev/internal/logz" + "github.com/sour-is/ev/internal/lg" ) // Config represents a Salty Config for a User which at a minimum is required @@ -79,7 +79,7 @@ func (a *Addr) DiscoveredDomain() string { } func (a *Addr) Refresh(ctx context.Context) error { - ctx, span := logz.Span(ctx) + ctx, span := lg.Span(ctx) defer span.End() span.AddEvent(fmt.Sprintf("Looking up SRV record for _salty._tcp.%s", a.Domain)) @@ -118,7 +118,7 @@ func (a *Addr) Refresh(ctx context.Context) error { } func fetchConfig(ctx context.Context, addr string) (config Config, cap Capabilities, err error) { - ctx, span := logz.Span(ctx) + ctx, span := lg.Span(ctx) defer span.End() var req *http.Request diff --git a/app/salty/service.go b/app/salty/service.go index 3bc626e..c442ee9 100644 --- a/app/salty/service.go +++ b/app/salty/service.go @@ -12,7 +12,7 @@ import ( "strings" "github.com/keys-pub/keys" - "github.com/sour-is/ev/internal/logz" + "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/gql" @@ -49,7 +49,7 @@ type SaltyResolver interface { } func New(ctx context.Context, es *es.EventStore, baseURL string) (*service, error) { - ctx, span := logz.Span(ctx) + ctx, span := lg.Span(ctx) defer span.End() if err := event.Register(ctx, &UserRegistered{}); err != nil { @@ -59,7 +59,7 @@ func New(ctx context.Context, es *es.EventStore, baseURL string) (*service, erro return nil, err } - m := logz.Meter(ctx) + m := lg.Meter(ctx) svc := &service{baseURL: baseURL, es: es, dns: net.DefaultResolver} @@ -93,7 +93,7 @@ func (s *service) BaseURL() string { return s.baseURL } func (s *service) RegisterHTTP(mux *http.ServeMux) { - mux.Handle("/.well-known/salty/", logz.Htrace(s, "lookup")) + mux.Handle("/.well-known/salty/", lg.Htrace(s, "lookup")) } func (s *service) RegisterAPIv1(mux *http.ServeMux) { mux.HandleFunc("/ping", s.apiv1) @@ -103,7 +103,7 @@ func (s *service) RegisterAPIv1(mux *http.ServeMux) { } func (s *service) ServeHTTP(w http.ResponseWriter, r *http.Request) { ctx := r.Context() - ctx, span := logz.Span(ctx) + ctx, span := lg.Span(ctx) defer span.End() addr := "saltyuser-" + strings.TrimPrefix(r.URL.Path, "/.well-known/salty/") @@ -137,7 +137,7 @@ func (s *service) ServeHTTP(w http.ResponseWriter, r *http.Request) { } } func (s *service) CreateSaltyUser(ctx context.Context, nick string, pub string) (*SaltyUser, error) { - ctx, span := logz.Span(ctx) + ctx, span := lg.Span(ctx) defer span.End() s.m_create_user.Add(ctx, 1) @@ -167,7 +167,7 @@ func (s *service) CreateSaltyUser(ctx context.Context, nick string, pub string) return a, nil } func (s *service) SaltyUser(ctx context.Context, nick string) (*SaltyUser, error) { - ctx, span := logz.Span(ctx) + ctx, span := lg.Span(ctx) defer span.End() s.m_get_user.Add(ctx, 1) @@ -200,7 +200,7 @@ func (s *service) GetMiddleware() func(http.Handler) http.Handler { func (s *service) apiv1(w http.ResponseWriter, r *http.Request) { ctx := r.Context() - ctx, span := logz.Span(ctx) + ctx, span := lg.Span(ctx) defer span.End() switch r.Method { diff --git a/env b/env deleted file mode 100644 index 660c109..0000000 --- a/env +++ /dev/null @@ -1,5 +0,0 @@ -SCRIPT_VERSION=1.1.0 -REQ_ID=01GB0GDHCEAPMZFVES2KBCN7KF -REQ_IP=1.1.1.1 -REQ_FAMILY=1 -REQ_CREATED=2022-08-21 09:30:22 diff --git a/httpmux.go b/httpmux.go new file mode 100644 index 0000000..a19f204 --- /dev/null +++ b/httpmux.go @@ -0,0 +1,37 @@ +package main + +import ( + "net/http" + + "github.com/rs/cors" +) + +type mux struct { + *http.ServeMux + api *http.ServeMux +} + +func httpMux(fns ...interface{ RegisterHTTP(*http.ServeMux) }) http.Handler { + mux := newMux() + for _, fn := range fns { + fn.RegisterHTTP(mux.ServeMux) + + if fn, ok := fn.(interface{ RegisterAPIv1(*http.ServeMux) }); ok { + fn.RegisterAPIv1(mux.api) + } + } + + return cors.AllowAll().Handler(mux) +} +func newMux() *mux { + mux := &mux{ + api: http.NewServeMux(), + ServeMux: http.NewServeMux(), + } + mux.Handle("/api/v1/", http.StripPrefix("/api/v1/", mux.api)) + + return mux +} +func (m mux) HandleAPIv1(pattern string, handler http.Handler) { + m.api.Handle(pattern, handler) +} diff --git a/internal/logz/init.go b/internal/lg/init.go similarity index 99% rename from internal/logz/init.go rename to internal/lg/init.go index f2a097f..9a2a572 100644 --- a/internal/logz/init.go +++ b/internal/lg/init.go @@ -1,4 +1,4 @@ -package logz +package lg import ( "context" diff --git a/internal/logz/logger.go b/internal/lg/logger.go similarity index 95% rename from internal/logz/logger.go rename to internal/lg/logger.go index 3c1a7c9..80cb813 100644 --- a/internal/logz/logger.go +++ b/internal/lg/logger.go @@ -1,4 +1,4 @@ -package logz +package lg import ( "bytes" @@ -72,7 +72,7 @@ func initLogger(name string) func() error { l, err := logzio.New( token.Secret(), // logzio.SetDebug(os.Stderr), - logzio.SetUrl(env("LOGZIO_LOG_URL", "https://listener.logz.io:8071")), + logzio.SetUrl(env("LOGZIO_LOG_URL", "https://listener.lg.io:8071")), logzio.SetDrainDuration(time.Second*5), logzio.SetTempDirectory(env("LOGZIO_DIR", os.TempDir())), logzio.SetCheckDiskSpace(true), diff --git a/internal/logz/metric.go b/internal/lg/metric.go similarity index 99% rename from internal/logz/metric.go rename to internal/lg/metric.go index e72b2cb..4a1d36f 100644 --- a/internal/logz/metric.go +++ b/internal/lg/metric.go @@ -1,4 +1,4 @@ -package logz +package lg import ( "context" diff --git a/internal/logz/tracer.go b/internal/lg/tracer.go similarity index 84% rename from internal/logz/tracer.go rename to internal/lg/tracer.go index 6188140..a5c9639 100644 --- a/internal/logz/tracer.go +++ b/internal/lg/tracer.go @@ -1,4 +1,4 @@ -package logz +package lg import ( "context" @@ -33,10 +33,10 @@ func Tracer(ctx context.Context) trace.Tracer { return otel.Tracer("") } -func Span(ctx context.Context, opts ...trace.SpanStartOption) (context.Context, trace.Span) { +func attrs(ctx context.Context) (string, []attribute.KeyValue) { var attrs []attribute.KeyValue var name string - if pc, file, line, ok := runtime.Caller(1); ok { + if pc, file, line, ok := runtime.Caller(2); ok { if fn := runtime.FuncForPC(pc); fn != nil { name = fn.Name() } @@ -47,12 +47,29 @@ func Span(ctx context.Context, opts ...trace.SpanStartOption) (context.Context, attribute.String("name", name), ) } + return name, attrs +} + +func Span(ctx context.Context, opts ...trace.SpanStartOption) (context.Context, trace.Span) { + name, attrs := attrs(ctx) ctx, span := Tracer(ctx).Start(ctx, name, opts...) span.SetAttributes(attrs...) return ctx, span } +func Fork(ctx context.Context, opts ...trace.SpanStartOption) (context.Context, trace.Span) { + name, attrs := attrs(ctx) + childCTX, childSpan := Tracer(ctx).Start(context.Background(), name, append(opts, trace.WithLinks(trace.LinkFromContext(ctx)))...) + childSpan.SetAttributes(attrs...) + + _, span := Tracer(ctx).Start(ctx, name, append(opts, trace.WithLinks(trace.LinkFromContext(childCTX)))...) + span.SetAttributes(attrs...) + defer span.End() + + return childCTX, childSpan +} + type SampleRate string const ( diff --git a/main.go b/main.go index 3e863e5..f837534 100644 --- a/main.go +++ b/main.go @@ -1,9 +1,7 @@ package main import ( - "bytes" "context" - "fmt" "log" "net/http" "os" @@ -12,18 +10,21 @@ import ( "strings" "time" - "github.com/rs/cors" + "go.uber.org/multierr" "golang.org/x/sync/errgroup" "github.com/sour-is/ev/app/gql" "github.com/sour-is/ev/app/msgbus" "github.com/sour-is/ev/app/peerfinder" "github.com/sour-is/ev/app/salty" - "github.com/sour-is/ev/internal/logz" + "github.com/sour-is/ev/internal/lg" "github.com/sour-is/ev/pkg/es" 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" "github.com/sour-is/ev/pkg/es/driver/streamer" + "github.com/sour-is/ev/pkg/es/event" + "github.com/sour-is/ev/pkg/set" ) const AppName string = "sour.is-ev" @@ -35,7 +36,7 @@ func main() { defer cancel() }() - ctx, stop := logz.Init(ctx, AppName) + ctx, stop := lg.Init(ctx, AppName) defer stop() if err := run(ctx); err != nil && err != http.ErrServerClosed { @@ -46,12 +47,20 @@ func run(ctx context.Context) error { g, ctx := errgroup.WithContext(ctx) { - ctx, span := logz.Span(ctx) + ctx, span := lg.Span(ctx) - diskstore.Init(ctx) - memstore.Init(ctx) + err := multierr.Combine( + es.Init(ctx), + event.Init(ctx), + diskstore.Init(ctx), + memstore.Init(ctx), + ) + if err != nil { + span.RecordError(err) + return err + } - es, err := es.Open(ctx, env("EV_DATA", "file:data"), streamer.New(ctx)) + es, err := es.Open(ctx, env("EV_DATA", "mem:"), streamer.New(ctx), projecter.New(ctx)) if err != nil { span.RecordError(err) return err @@ -65,7 +74,7 @@ func run(ctx context.Context) error { s.Addr = "[::]" + s.Addr } - enable := set(strings.Fields(env("EV_ENABLE", "salty msgbus gql peers"))...) + enable := set.New(strings.Fields(env("EV_ENABLE", "salty msgbus gql peers"))...) var svcs []interface{ RegisterHTTP(*http.ServeMux) } svcs = append(svcs, es) @@ -109,14 +118,14 @@ func run(ctx context.Context) error { } svcs = append(svcs, gql) } - svcs = append(svcs, logz.NewHTTP(ctx)) + svcs = append(svcs, lg.NewHTTP(ctx)) s.Handler = httpMux(svcs...) log.Print("Listen on ", s.Addr) - span.AddEvent("begin listen and serve") + span.AddEvent("begin listen and serve on " + s.Addr) - Mup, err := logz.Meter(ctx).SyncInt64().UpDownCounter("up") + Mup, err := lg.Meter(ctx).SyncInt64().UpDownCounter("up") if err != nil { return err } @@ -149,61 +158,3 @@ func env(name, defaultValue string) string { log.Println("#", name, "=", defaultValue, "(default)") return defaultValue } -func httpMux(fns ...interface{ RegisterHTTP(*http.ServeMux) }) http.Handler { - mux := newMux() - for _, fn := range fns { - fn.RegisterHTTP(mux.ServeMux) - - if fn, ok := fn.(interface{ RegisterAPIv1(*http.ServeMux) }); ok { - fn.RegisterAPIv1(mux.api) - } - } - - return cors.AllowAll().Handler(mux) -} - -type Set[T comparable] map[T]struct{} - -func set[T comparable](items ...T) Set[T] { - s := make(map[T]struct{}, len(items)) - for i := range items { - s[items[i]] = struct{}{} - } - return s -} -func (s Set[T]) Has(v T) bool { - _, ok := (s)[v] - return ok -} -func (s Set[T]) String() string { - if s == nil { - return "set()" - } - lis := make([]string, 0, len(s)) - for k := range s { - lis = append(lis, fmt.Sprint(k)) - } - - var b bytes.Buffer - b.WriteString("set(") - b.WriteString(strings.Join(lis, ",")) - b.WriteString(")") - return b.String() -} - -type mux struct { - *http.ServeMux - api *http.ServeMux -} -func newMux() *mux { - mux := &mux{ - api: http.NewServeMux(), - ServeMux: http.NewServeMux(), - } - mux.Handle("/api/v1/", http.StripPrefix("/api/v1/", mux.api)) - - return mux -} -func (m mux) HandleAPIv1(pattern string, handler http.Handler) { - m.api.Handle(pattern, handler) -} \ No newline at end of file diff --git a/pkg/es/driver/disk-store/disk-store.go b/pkg/es/driver/disk-store/disk-store.go index 43e7057..874fd53 100644 --- a/pkg/es/driver/disk-store/disk-store.go +++ b/pkg/es/driver/disk-store/disk-store.go @@ -11,7 +11,7 @@ import ( "go.opentelemetry.io/otel/metric/instrument/syncint64" "go.uber.org/multierr" - "github.com/sour-is/ev/internal/logz" + "github.com/sour-is/ev/internal/lg" "github.com/sour-is/ev/pkg/cache" "github.com/sour-is/ev/pkg/es" "github.com/sour-is/ev/pkg/es/driver" @@ -38,10 +38,10 @@ const AppendOnly = es.AppendOnly const AllEvents = es.AllEvents func Init(ctx context.Context) error { - _, span := logz.Span(ctx) + _, span := lg.Span(ctx) defer span.End() - m := logz.Meter(ctx) + m := lg.Meter(ctx) var err, errs error Mdisk_open, err := m.SyncInt64().Counter("disk_open") @@ -61,7 +61,7 @@ func Init(ctx context.Context) error { var _ driver.Driver = (*diskStore)(nil) func (d *diskStore) Open(ctx context.Context, dsn string) (driver.Driver, error) { - ctx, span := logz.Span(ctx) + ctx, span := lg.Span(ctx) defer span.End() d.Mdisk_open.Add(ctx, 1) @@ -82,11 +82,11 @@ func (d *diskStore) Open(ctx context.Context, dsn string) (driver.Driver, error) } } c, err := cache.NewWithEvict(CachSize, func(ctx context.Context, s string, l *lockedWal) { - _, span := logz.Span(ctx) + _, span := lg.Span(ctx) defer span.End() l.Modify(ctx, func(w *wal.Log) error { - _, span := logz.Span(ctx) + _, span := lg.Span(ctx) defer span.End() d.Mdisk_evict.Add(ctx, 1) @@ -112,13 +112,13 @@ func (d *diskStore) Open(ctx context.Context, dsn string) (driver.Driver, error) }, nil } func (d *diskStore) EventLog(ctx context.Context, streamID string) (driver.EventLog, error) { - _, span := logz.Span(ctx) + _, span := lg.Span(ctx) defer span.End() el := &eventLog{streamID: streamID} return el, d.openlogs.Modify(ctx, func(openlogs *openlogs) error { - _, span := logz.Span(ctx) + _, span := lg.Span(ctx) defer span.End() if events, ok := openlogs.logs.Get(streamID); ok { @@ -145,14 +145,14 @@ type eventLog struct { var _ driver.EventLog = (*eventLog)(nil) func (e *eventLog) Append(ctx context.Context, events event.Events, version uint64) (uint64, error) { - _, span := logz.Span(ctx) + _, span := lg.Span(ctx) defer span.End() event.SetStreamID(e.streamID, events...) var count uint64 err := e.events.Modify(ctx, func(l *wal.Log) error { - _, span := logz.Span(ctx) + _, span := lg.Span(ctx) defer span.End() last, err := l.LastIndex() @@ -190,13 +190,13 @@ func (e *eventLog) Append(ctx context.Context, events event.Events, version uint return count, err } func (e *eventLog) Read(ctx context.Context, pos, count int64) (event.Events, error) { - _, span := logz.Span(ctx) + _, span := lg.Span(ctx) defer span.End() var events event.Events err := e.events.Modify(ctx, func(stream *wal.Log) error { - _, span := logz.Span(ctx) + _, span := lg.Span(ctx) defer span.End() first, err := stream.FirstIndex() @@ -259,7 +259,7 @@ func (e *eventLog) Read(ctx context.Context, pos, count int64) (event.Events, er return events, nil } func (e *eventLog) FirstIndex(ctx context.Context) (uint64, error) { - _, span := logz.Span(ctx) + _, span := lg.Span(ctx) defer span.End() var idx uint64 @@ -273,7 +273,7 @@ func (e *eventLog) FirstIndex(ctx context.Context) (uint64, error) { return idx, err } func (e *eventLog) LastIndex(ctx context.Context) (uint64, error) { - _, span := logz.Span(ctx) + _, span := lg.Span(ctx) defer span.End() var idx uint64 diff --git a/pkg/es/driver/mem-store/mem-store.go b/pkg/es/driver/mem-store/mem-store.go index 70f4a72..b4378ab 100644 --- a/pkg/es/driver/mem-store/mem-store.go +++ b/pkg/es/driver/mem-store/mem-store.go @@ -4,7 +4,7 @@ import ( "context" "fmt" - "github.com/sour-is/ev/internal/logz" + "github.com/sour-is/ev/internal/lg" "github.com/sour-is/ev/pkg/es" "github.com/sour-is/ev/pkg/es/driver" "github.com/sour-is/ev/pkg/es/event" @@ -26,30 +26,30 @@ type memstore struct { const AppendOnly = es.AppendOnly const AllEvents = es.AllEvents -func Init(ctx context.Context) { - ctx, span := logz.Span(ctx) +func Init(ctx context.Context) error { + ctx, span := lg.Span(ctx) defer span.End() - es.Register(ctx, "mem", &memstore{}) + return es.Register(ctx, "mem", &memstore{}) } var _ driver.Driver = (*memstore)(nil) func (memstore) Open(ctx context.Context, name string) (driver.Driver, error) { - _, span := logz.Span(ctx) + _, span := lg.Span(ctx) defer span.End() s := &state{streams: make(map[string]*locker.Locked[event.Events])} return &memstore{locker.New(s)}, nil } func (m *memstore) EventLog(ctx context.Context, streamID string) (driver.EventLog, error) { - ctx, span := logz.Span(ctx) + ctx, span := lg.Span(ctx) defer span.End() el := &eventLog{streamID: streamID} err := m.state.Modify(ctx, func(state *state) error { - _, span := logz.Span(ctx) + _, span := lg.Span(ctx) defer span.End() l, ok := state.streams[streamID] @@ -70,16 +70,16 @@ var _ driver.EventLog = (*eventLog)(nil) // Append implements driver.EventStore func (m *eventLog) Append(ctx context.Context, events event.Events, version uint64) (uint64, error) { - ctx, span := logz.Span(ctx) + ctx, span := lg.Span(ctx) defer span.End() event.SetStreamID(m.streamID, events...) return uint64(len(events)), m.events.Modify(ctx, func(stream *event.Events) error { - _, span := logz.Span(ctx) + _, span := lg.Span(ctx) defer span.End() - span.AddEvent(fmt.Sprintf(" %s %#v %d", m.streamID, stream, len(*stream))) + span.AddEvent(fmt.Sprintf(" %s %d", m.streamID, len(*stream))) last := uint64(len(*stream)) if version != AppendOnly && version != last { @@ -100,16 +100,16 @@ func (m *eventLog) Append(ctx context.Context, events event.Events, version uint // Read implements driver.EventStore func (m *eventLog) Read(ctx context.Context, pos int64, count int64) (event.Events, error) { - ctx, span := logz.Span(ctx) + ctx, span := lg.Span(ctx) defer span.End() var events event.Events err := m.events.Modify(ctx, func(stream *event.Events) error { - _, span := logz.Span(ctx) + _, span := lg.Span(ctx) defer span.End() - span.AddEvent(fmt.Sprintf(" %s %#v %d", m.streamID, stream, len(*stream))) + span.AddEvent(fmt.Sprintf("%s %d", m.streamID, len(*stream))) first := stream.First().EventMeta().Position last := stream.Last().EventMeta().Position @@ -154,7 +154,7 @@ func (m *eventLog) Read(ctx context.Context, pos int64, count int64) (event.Even // FirstIndex for the streamID func (m *eventLog) FirstIndex(ctx context.Context) (uint64, error) { - _, span := logz.Span(ctx) + _, span := lg.Span(ctx) defer span.End() events, err := m.events.Copy(ctx) @@ -163,7 +163,7 @@ func (m *eventLog) FirstIndex(ctx context.Context) (uint64, error) { // LastIndex for the streamID func (m *eventLog) LastIndex(ctx context.Context) (uint64, error) { - _, span := logz.Span(ctx) + _, span := lg.Span(ctx) defer span.End() events, err := m.events.Copy(ctx) diff --git a/pkg/es/driver/projecter/projecter.go b/pkg/es/driver/projecter/projecter.go new file mode 100644 index 0000000..28a5309 --- /dev/null +++ b/pkg/es/driver/projecter/projecter.go @@ -0,0 +1,127 @@ +package projecter + +import ( + "context" + "strings" + + "github.com/sour-is/ev/internal/lg" + "github.com/sour-is/ev/pkg/es" + "github.com/sour-is/ev/pkg/es/driver" + "github.com/sour-is/ev/pkg/es/event" +) + +type projector struct { + up driver.Driver +} + +func New(ctx context.Context) *projector { + return &projector{} +} +func (p *projector) Apply(e *es.EventStore) { + p.up = e.Driver + e.Driver = p +} +func (s *projector) Unwrap() driver.Driver { + return s.up +} +func (s *projector) Open(ctx context.Context, dsn string) (driver.Driver, error) { + ctx, span := lg.Span(ctx) + defer span.End() + + return s.up.Open(ctx, dsn) +} +func (s *projector) EventLog(ctx context.Context, streamID string) (driver.EventLog, error) { + ctx, span := lg.Span(ctx) + defer span.End() + + l, err := s.up.EventLog(ctx, streamID) + return &wrapper{l, s}, err +} + +type wrapper struct { + up driver.EventLog + projector *projector +} + +var _ driver.EventLog = (*wrapper)(nil) + +func (w *wrapper) Read(ctx context.Context, pos int64, count int64) (event.Events, error) { + ctx, span := lg.Span(ctx) + defer span.End() + + return w.up.Read(ctx, pos, count) +} +func (w *wrapper) Append(ctx context.Context, events event.Events, version uint64) (uint64, error) { + ctx, span := lg.Span(ctx) + defer span.End() + + i, err := w.up.Append(ctx, events, version) + if err != nil { + return i, err + } + + { + ctx, span := lg.Fork(ctx) + + go func() { + defer span.End() + + var pevents []event.Event + + for i := range events { + e := events[i] + eventType := event.TypeOf(e) + streamID := e.EventMeta().StreamID + streamPos := e.EventMeta().Position + + e1 := event.NewPtr(streamID, streamPos) + event.SetStreamID("$all", e1) + + e2 := event.NewPtr(streamID, streamPos) + event.SetStreamID("$type-"+eventType, e2) + + e3 := event.NewPtr(streamID, streamPos) + pkg, _, _ := strings.Cut(eventType, ".") + event.SetStreamID("$pkg-"+pkg, e3) + + pevents = append( + pevents, + e1, + e2, + e3, + ) + } + + for i := range pevents { + e := pevents[i] + l, err := w.projector.up.EventLog(ctx, event.StreamID(e).StreamID()) + if err != nil { + span.RecordError(err) + continue + } + _, err = l.Append(ctx, event.NewEvents(e), es.AppendOnly) + span.RecordError(err) + } + }() + } + + return i, err +} +func (w *wrapper) FirstIndex(ctx context.Context) (uint64, error) { + ctx, span := lg.Span(ctx) + defer span.End() + + return w.up.FirstIndex(ctx) +} +func (w *wrapper) LastIndex(ctx context.Context) (uint64, error) { + ctx, span := lg.Span(ctx) + defer span.End() + + return w.up.LastIndex(ctx) +} +func (w *wrapper) LoadForUpdate(ctx context.Context, a event.Aggregate, fn func(context.Context, event.Aggregate) error) (uint64, error) { + ctx, span := lg.Span(ctx) + defer span.End() + + return w.up.LoadForUpdate(ctx, a, fn) +} diff --git a/pkg/es/driver/streamer/streamer.go b/pkg/es/driver/streamer/streamer.go index ccc6507..5faa065 100644 --- a/pkg/es/driver/streamer/streamer.go +++ b/pkg/es/driver/streamer/streamer.go @@ -2,12 +2,15 @@ package streamer import ( "context" + "fmt" - "github.com/sour-is/ev/internal/logz" + "github.com/sour-is/ev/internal/lg" "github.com/sour-is/ev/pkg/es" "github.com/sour-is/ev/pkg/es/driver" "github.com/sour-is/ev/pkg/es/event" "github.com/sour-is/ev/pkg/locker" + "go.opentelemetry.io/otel/attribute" + "go.opentelemetry.io/otel/trace" ) type state struct { @@ -20,7 +23,7 @@ type streamer struct { } func New(ctx context.Context) *streamer { - ctx, span := logz.Span(ctx) + _, span := lg.Span(ctx) defer span.End() return &streamer{state: locker.New(&state{subscribers: map[string][]*subscription{}})} @@ -39,13 +42,13 @@ func (s *streamer) Unwrap() driver.Driver { var _ driver.Driver = (*streamer)(nil) func (s *streamer) Open(ctx context.Context, dsn string) (driver.Driver, error) { - ctx, span := logz.Span(ctx) + ctx, span := lg.Span(ctx) defer span.End() return s.up.Open(ctx, dsn) } func (s *streamer) EventLog(ctx context.Context, streamID string) (driver.EventLog, error) { - ctx, span := logz.Span(ctx) + ctx, span := lg.Span(ctx) defer span.End() l, err := s.up.EventLog(ctx, streamID) @@ -55,7 +58,7 @@ func (s *streamer) EventLog(ctx context.Context, streamID string) (driver.EventL var _ driver.EventStream = (*streamer)(nil) func (s *streamer) Subscribe(ctx context.Context, streamID string, start int64) (driver.Subscription, error) { - ctx, span := logz.Span(ctx) + ctx, span := lg.Span(ctx) defer span.End() events, err := s.up.EventLog(ctx, streamID) @@ -75,22 +78,24 @@ func (s *streamer) Subscribe(ctx context.Context, streamID string, start int64) }) } func (s *streamer) Send(ctx context.Context, streamID string, events event.Events) error { - ctx, span := logz.Span(ctx) + ctx, span := lg.Span(ctx) defer span.End() return s.state.Modify(ctx, func(state *state) error { - ctx, span := logz.Span(ctx) + 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 { - _, span := logz.Span(ctx) + _, span := lg.Span(ctx) defer span.End() position.size = int64(events.Last().EventMeta().Position - uint64(position.idx)) if position.wait != nil { close(position.wait) + position.link = trace.LinkFromContext(ctx, attribute.String("src", "event")) position.wait = nil } return nil @@ -102,17 +107,16 @@ func (s *streamer) Send(ctx context.Context, streamID string, events event.Event return nil }) } - func (s *streamer) delete(streamID string, sub *subscription) func(context.Context) error { return func(ctx context.Context) error { - ctx, span := logz.Span(ctx) + ctx, span := lg.Span(ctx) defer span.End() if err := ctx.Err(); err != nil { return err } return s.state.Modify(ctx, func(state *state) error { - _, span := logz.Span(ctx) + _, span := lg.Span(ctx) defer span.End() lis := state.subscribers[streamID] @@ -138,44 +142,53 @@ type wrapper struct { var _ driver.EventLog = (*wrapper)(nil) func (w *wrapper) Read(ctx context.Context, pos int64, count int64) (event.Events, error) { - ctx, span := logz.Span(ctx) + ctx, span := lg.Span(ctx) defer span.End() return w.up.Read(ctx, pos, count) } - func (w *wrapper) Append(ctx context.Context, events event.Events, version uint64) (uint64, error) { - ctx, span := logz.Span(ctx) + ctx, span := lg.Span(ctx) defer span.End() i, err := w.up.Append(ctx, events, version) if err != nil { return i, err } - return i, w.streamer.Send(ctx, w.topic, events) -} + ctx, span = lg.Fork(ctx) + go func() { + defer span.End() + + err := w.streamer.Send(ctx, w.topic, events) + span.RecordError(err) + }() + + return i, nil +} func (w *wrapper) FirstIndex(ctx context.Context) (uint64, error) { - ctx, span := logz.Span(ctx) + ctx, span := lg.Span(ctx) defer span.End() return w.up.FirstIndex(ctx) } - func (w *wrapper) LastIndex(ctx context.Context) (uint64, error) { - ctx, span := logz.Span(ctx) + ctx, span := lg.Span(ctx) defer span.End() return w.up.LastIndex(ctx) } - func (w *wrapper) LoadForUpdate(ctx context.Context, a event.Aggregate, fn func(context.Context, event.Aggregate) error) (uint64, error) { + ctx, span := lg.Span(ctx) + defer span.End() + return w.up.LoadForUpdate(ctx, a, fn) } type position struct { size int64 idx int64 + link trace.Link wait chan struct{} } @@ -189,13 +202,13 @@ type subscription struct { } func (s *subscription) Recv(ctx context.Context) bool { - ctx, span := logz.Span(ctx) + ctx, span := lg.Span(ctx) defer span.End() var wait func(context.Context) bool err := s.position.Modify(ctx, func(position *position) error { - _, span := logz.Span(ctx) + _, span := lg.Span(ctx) defer span.End() if position.size == es.AllEvents { @@ -204,11 +217,17 @@ func (s *subscription) Recv(ctx context.Context) bool { if position.size == 0 { position.wait = make(chan struct{}) wait = func(ctx context.Context) bool { - ctx, span := logz.Span(ctx) + 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(): @@ -232,12 +251,12 @@ func (s *subscription) Recv(ctx context.Context) bool { return true } func (s *subscription) Events(ctx context.Context) (event.Events, error) { - ctx, span := logz.Span(ctx) + ctx, span := lg.Span(ctx) defer span.End() var events event.Events return events, s.position.Modify(ctx, func(position *position) error { - ctx, span := logz.Span(ctx) + ctx, span := lg.Span(ctx) defer span.End() var err error @@ -253,7 +272,7 @@ func (s *subscription) Events(ctx context.Context) (event.Events, error) { }) } func (s *subscription) Close(ctx context.Context) error { - ctx, span := logz.Span(ctx) + ctx, span := lg.Span(ctx) defer span.End() return s.unsub(ctx) diff --git a/pkg/es/es.go b/pkg/es/es.go index 78e2ef3..e5604eb 100644 --- a/pkg/es/es.go +++ b/pkg/es/es.go @@ -6,7 +6,7 @@ import ( "fmt" "strings" - "github.com/sour-is/ev/internal/logz" + "github.com/sour-is/ev/internal/lg" "github.com/sour-is/ev/pkg/es/driver" "github.com/sour-is/ev/pkg/es/event" "github.com/sour-is/ev/pkg/locker" @@ -25,11 +25,8 @@ var ( drivers = locker.New(&config{drivers: make(map[string]driver.Driver)}) ) -func Register(ctx context.Context, name string, d driver.Driver) error { - ctx, span := logz.Span(ctx) - defer span.End() - - m := logz.Meter(ctx) +func Init(ctx context.Context) error { + m := lg.Meter(ctx) var err, errs error Mes_open, err = m.SyncInt64().Counter("es_open") @@ -47,15 +44,20 @@ func Register(ctx context.Context, name string, d driver.Driver) error { Mes_append, err = m.SyncInt64().Counter("es_append") errs = multierr.Append(errs, err) - err = drivers.Modify(ctx, func(c *config) error { + return errs +} + +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 { if _, set := c.drivers[name]; set { return fmt.Errorf("driver %s already set", name) } c.drivers[name] = d return nil }) - - return multierr.Append(errs, err) } type EventStore struct { @@ -71,7 +73,7 @@ var ( ) func Open(ctx context.Context, dsn string, options ...Option) (*EventStore, error) { - ctx, span := logz.Span(ctx) + ctx, span := lg.Span(ctx) defer span.End() Mes_open.Add(ctx, 1) @@ -107,7 +109,7 @@ type Option interface { } func (es *EventStore) Save(ctx context.Context, agg event.Aggregate) (uint64, error) { - ctx, span := logz.Span(ctx) + ctx, span := lg.Span(ctx) defer span.End() events := agg.Events(true) @@ -131,7 +133,7 @@ func (es *EventStore) Save(ctx context.Context, agg event.Aggregate) (uint64, er return count, err } func (es *EventStore) Load(ctx context.Context, agg event.Aggregate) error { - ctx, span := logz.Span(ctx) + ctx, span := lg.Span(ctx) defer span.End() Mes_load.Add(ctx, 1) @@ -151,7 +153,7 @@ func (es *EventStore) Load(ctx context.Context, agg event.Aggregate) error { return nil } func (es *EventStore) Read(ctx context.Context, streamID string, pos, count int64) (event.Events, error) { - ctx, span := logz.Span(ctx) + ctx, span := lg.Span(ctx) defer span.End() Mes_read.Add(ctx, 1) @@ -163,7 +165,7 @@ func (es *EventStore) Read(ctx context.Context, streamID string, pos, count int6 return l.Read(ctx, pos, count) } func (es *EventStore) Append(ctx context.Context, streamID string, events event.Events) (uint64, error) { - ctx, span := logz.Span(ctx) + ctx, span := lg.Span(ctx) defer span.End() Mes_append.Add(ctx, 1) @@ -175,7 +177,7 @@ func (es *EventStore) Append(ctx context.Context, streamID string, events event. return l.Append(ctx, events, AppendOnly) } func (es *EventStore) FirstIndex(ctx context.Context, streamID string) (uint64, error) { - ctx, span := logz.Span(ctx) + ctx, span := lg.Span(ctx) defer span.End() l, err := es.Driver.EventLog(ctx, streamID) @@ -185,7 +187,7 @@ func (es *EventStore) FirstIndex(ctx context.Context, streamID string) (uint64, return l.FirstIndex(ctx) } func (es *EventStore) LastIndex(ctx context.Context, streamID string) (uint64, error) { - ctx, span := logz.Span(ctx) + ctx, span := lg.Span(ctx) defer span.End() l, err := es.Driver.EventLog(ctx, streamID) @@ -231,7 +233,7 @@ type PE[T any] interface { // Create uses fn to create a new aggregate and store in db. func Create[A any, T PA[A]](ctx context.Context, es *EventStore, streamID string, fn func(context.Context, T) error) (agg T, err error) { - ctx, span := logz.Span(ctx) + ctx, span := lg.Span(ctx) defer span.End() agg = new(A) @@ -261,7 +263,7 @@ func Create[A any, T PA[A]](ctx context.Context, es *EventStore, streamID string // Update uses fn to update an exsisting aggregate and store in db. func Update[A any, T PA[A]](ctx context.Context, es *EventStore, streamID string, fn func(context.Context, T) error) (agg T, err error) { - ctx, span := logz.Span(ctx) + ctx, span := lg.Span(ctx) defer span.End() agg = new(A) @@ -288,7 +290,7 @@ func Update[A any, T PA[A]](ctx context.Context, es *EventStore, streamID string // Update uses fn to update an exsisting aggregate and store in db. func Upsert[A any, T PA[A]](ctx context.Context, es *EventStore, streamID string, fn func(context.Context, T) error) (agg T, err error) { - ctx, span := logz.Span(ctx) + ctx, span := lg.Span(ctx) defer span.End() agg = new(A) diff --git a/pkg/es/event/events.go b/pkg/es/event/events.go index be42a2d..b81a57f 100644 --- a/pkg/es/event/events.go +++ b/pkg/es/event/events.go @@ -1,11 +1,13 @@ package event import ( + "context" "crypto/rand" "encoding" "encoding/json" "fmt" "io" + "strconv" "strings" "sync" "time" @@ -129,6 +131,10 @@ 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{}) +} + type nilEvent struct{} func (*nilEvent) EventMeta() Meta { @@ -144,3 +150,61 @@ func (e *nilEvent) MarshalBinary() ([]byte, error) { func (e *nilEvent) UnmarshalBinary(b []byte) error { return json.Unmarshal(b, e) } + +type eventPtr struct { + streamID string + pos uint64 + + eventMeta Meta +} + +var _ Event = (*eventPtr)(nil) + +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 +} + +// UnmarshalBinary implements Event +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] + var err error + e.pos, err = strconv.ParseUint(s[idx+1:], 10, 64) + + return err +} + +// EventMeta implements Event +func (e *eventPtr) EventMeta() Meta { + if e == nil { + return Meta{} + } + return e.eventMeta +} + +// SetEventMeta implements Event +func (e *eventPtr) SetEventMeta(m Meta) { + if e == nil { + return + } + e.eventMeta = m +} + +func (e *eventPtr) Values() any { + return struct { + StreamID string `json:"stream_id"` + Pos uint64 `json:"pos"` + }{ + e.streamID, + e.pos, + } +} diff --git a/pkg/es/event/reflect.go b/pkg/es/event/reflect.go index c1813bb..2610d57 100644 --- a/pkg/es/event/reflect.go +++ b/pkg/es/event/reflect.go @@ -9,7 +9,7 @@ import ( "reflect" "strings" - "github.com/sour-is/ev/internal/logz" + "github.com/sour-is/ev/internal/lg" "github.com/sour-is/ev/pkg/locker" ) @@ -67,7 +67,7 @@ func (u *UnknownEvent) MarshalBinary() ([]byte, error) { // Register a type container for Unmarshalling values into. The type must implement Event and not be a nil value. func Register(ctx context.Context, lis ...Event) error { - _, span := logz.Span(ctx) + _, span := lg.Span(ctx) defer span.End() for _, e := range lis { @@ -84,7 +84,7 @@ func Register(ctx context.Context, lis ...Event) error { return nil } func RegisterName(ctx context.Context, name string, e Event) error { - _, span := logz.Span(ctx) + _, span := lg.Span(ctx) defer span.End() if e == nil { @@ -107,7 +107,7 @@ func RegisterName(ctx context.Context, name string, e Event) error { span.AddEvent("register: " + name) if err := eventTypes.Modify(ctx, func(c *config) error { - _, span := logz.Span(ctx) + _, span := lg.Span(ctx) defer span.End() c.eventTypes[name] = typ @@ -119,13 +119,13 @@ func RegisterName(ctx context.Context, name string, e Event) error { return nil } func GetContainer(ctx context.Context, s string) Event { - _, span := logz.Span(ctx) + _, span := lg.Span(ctx) defer span.End() var e Event eventTypes.Modify(ctx, func(c *config) error { - _, span := logz.Span(ctx) + _, span := lg.Span(ctx) defer span.End() typ, ok := c.eventTypes[s] @@ -176,7 +176,7 @@ func MarshalBinary(e Event) (txt []byte, err error) { } func UnmarshalBinary(ctx context.Context, txt []byte, pos uint64) (e Event, err error) { - _, span := logz.Span(ctx) + _, span := lg.Span(ctx) defer span.End() sp := bytes.SplitN(txt, []byte{'\t'}, 4) @@ -262,7 +262,13 @@ func Values(e Event) map[string]any { } field := v.FieldByIndex(idx.Index) - m[idx.Name] = field.Interface() + + name := idx.Name + if n, ok := idx.Tag.Lookup("json"); ok { + name = n + } + + m[name] = field.Interface() } return m } diff --git a/pkg/es/graph.go b/pkg/es/graph.go index 03c3173..68f4ded 100644 --- a/pkg/es/graph.go +++ b/pkg/es/graph.go @@ -6,7 +6,7 @@ import ( "net/http" "time" - "github.com/sour-is/ev/internal/logz" + "github.com/sour-is/ev/internal/lg" "github.com/sour-is/ev/pkg/es/event" "github.com/sour-is/ev/pkg/gql" ) @@ -17,7 +17,7 @@ type EventResolver interface { } func (es *EventStore) Events(ctx context.Context, streamID string, paging *gql.PageInput) (*gql.Connection, error) { - ctx, span := logz.Span(ctx) + ctx, span := lg.Span(ctx) defer span.End() lis, err := es.Read(ctx, streamID, paging.GetIdx(0), paging.GetCount(30)) @@ -53,7 +53,7 @@ func (es *EventStore) Events(ctx context.Context, streamID string, paging *gql.P }, nil } func (e *EventStore) EventAdded(ctx context.Context, streamID string, after int64) (<-chan *GQLEvent, error) { - ctx, span := logz.Span(ctx) + ctx, span := lg.Span(ctx) defer span.End() es := e.EventStream() @@ -70,15 +70,19 @@ func (e *EventStore) EventAdded(ctx context.Context, streamID string, after int6 ch := make(chan *GQLEvent) go func() { - ctx, span := logz.Span(ctx) + ctx, span := lg.Span(ctx) defer span.End() - defer func() { - ctx, cancel := context.WithTimeout(context.Background(), 1*time.Second) - defer cancel() - err := sub.Close(ctx) - span.RecordError(err) - }() + { + ctx, span := lg.Fork(ctx) + defer func() { + defer span.End() + ctx, cancel := context.WithTimeout(ctx, 1*time.Second) + defer cancel() + err := sub.Close(ctx) + span.RecordError(err) + }() + } for sub.Recv(ctx) { events, err := sub.Events(ctx) @@ -109,7 +113,7 @@ type GQLEvent struct { } func (e *GQLEvent) ID() string { - return "Event/" + e.e.EventMeta().GetEventID() + return fmt.Sprint(e.e.EventMeta().StreamID, "@", e.e.EventMeta().Position) } func (e *GQLEvent) EventID() string { return e.e.EventMeta().GetEventID() diff --git a/pkg/locker/locker.go b/pkg/locker/locker.go index b7bf629..521b596 100644 --- a/pkg/locker/locker.go +++ b/pkg/locker/locker.go @@ -2,6 +2,8 @@ package locker import ( "context" + + "github.com/sour-is/ev/internal/lg" ) type Locked[T any] struct { @@ -18,6 +20,9 @@ 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 { + _, span := lg.Span(ctx) + defer span.End() + if ctx.Err() != nil { return ctx.Err() } diff --git a/pkg/set/set.go b/pkg/set/set.go new file mode 100644 index 0000000..cbd238b --- /dev/null +++ b/pkg/set/set.go @@ -0,0 +1,35 @@ +package set + +import ( + "fmt" + "strings" +) + +type Set[T comparable] map[T]struct{} + +func New[T comparable](items ...T) Set[T] { + s := make(map[T]struct{}, len(items)) + for i := range items { + s[items[i]] = struct{}{} + } + return s +} +func (s Set[T]) Has(v T) bool { + _, ok := (s)[v] + return ok +} +func (s Set[T]) String() string { + if s == nil { + return "set()" + } + lis := make([]string, 0, len(s)) + for k := range s { + lis = append(lis, fmt.Sprint(k)) + } + + var b strings.Builder + b.WriteString("set(") + b.WriteString(strings.Join(lis, ",")) + b.WriteString(")") + return b.String() +}