feat: update deps + custom topic (#4)

.github/workflows/build.yaml

@@ -17,7 +17,7 @@ jobs:
           retention-days: 1
   test:
     runs-on: ubuntu-latest
-    container: golang:1.19-alpine
+    container: golang:1.21-alpine
     services:
       nats:
         image: skydev/nats-jetstream:latest
@@ -44,6 +44,8 @@ jobs:
           flags: unittests
       - run: cat /root/test.json
         if: always()
+      - run: wget https://github.com/mfridman/tparse/releases/latest/download/tparse_linux_x86_64 -O tparse && chmod 777 tparse && ./tparse -all -file=/root/test.json
+        if: always()
       - uses: guyarb/[email protected]
         if: always()
         with:

Makefile

@@ -0,0 +1,15 @@
+.PHONY: lint
+lint:
+	golangci-lint run
+
+.PHONY: lint-dev
+lint-dev:
+	golangci-lint run --tests=false
+
+.PHONY: generate
+generate:
+	go generate ./...
+
+.PHONY: test
+test:
+	AUTO_CREATE_CI_DB=true go test ./...

consumer/nats.go

@@ -7,9 +7,9 @@ import (
 	"sync"
 	"time"
 
+	"github.com/cockroachdb/errors"
 	"github.com/gammazero/workerpool"
 	"github.com/nats-io/nats.go"
-	"github.com/pkg/errors"
 	"github.com/rs/zerolog"
 	"github.com/rs/zerolog/log"
 

go.mod

@@ -1,42 +1,46 @@
 module github.com/skynet2/eventsourcing
 
-go 1.19
+go 1.21
 
 require (
+	github.com/cockroachdb/errors v1.11.1
 	github.com/gammazero/workerpool v1.1.3
 	github.com/google/uuid v1.3.0
-	github.com/nats-io/nats.go v1.20.0
-	github.com/pkg/errors v0.9.1
-	github.com/rs/zerolog v1.28.0
-	github.com/stretchr/testify v1.8.1
-	go.elastic.co/apm/module/apmhttp/v2 v2.2.0
-	go.elastic.co/apm/v2 v2.2.0
+	github.com/nats-io/nats.go v1.33.1
+	github.com/rs/zerolog v1.32.0
+	github.com/stretchr/testify v1.8.4
+	go.elastic.co/apm/module/apmhttp/v2 v2.5.0
+	go.elastic.co/apm/v2 v2.5.0
 )
 
 require (
 	github.com/armon/go-radix v1.0.0 // indirect
+	github.com/cockroachdb/logtags v0.0.0-20230118201751-21c54148d20b // indirect
+	github.com/cockroachdb/redact v1.1.5 // indirect
 	github.com/davecgh/go-spew v1.1.1 // indirect
-	github.com/elastic/go-licenser v0.4.1 // indirect
-	github.com/elastic/go-sysinfo v1.8.1 // indirect
+	github.com/elastic/go-sysinfo v1.13.1 // indirect
 	github.com/elastic/go-windows v1.0.1 // indirect
 	github.com/gammazero/deque v0.2.1 // indirect
-	github.com/golang/protobuf v1.5.2 // indirect
-	github.com/jcchavezs/porto v0.4.0 // indirect
+	github.com/getsentry/sentry-go v0.27.0 // indirect
+	github.com/gogo/protobuf v1.3.2 // indirect
 	github.com/joeshaw/multierror v0.0.0-20140124173710-69b34d4ec901 // indirect
+	github.com/klauspost/compress v1.17.7 // indirect
+	github.com/kr/pretty v0.3.1 // indirect
+	github.com/kr/text v0.2.0 // indirect
 	github.com/mattn/go-colorable v0.1.13 // indirect
-	github.com/mattn/go-isatty v0.0.16 // indirect
-	github.com/nats-io/nats-server/v2 v2.9.3 // indirect
-	github.com/nats-io/nkeys v0.3.0 // indirect
+	github.com/mattn/go-isatty v0.0.20 // indirect
+	github.com/nats-io/nkeys v0.4.7 // indirect
 	github.com/nats-io/nuid v1.0.1 // indirect
+	github.com/pkg/errors v0.9.1 // indirect
 	github.com/pmezard/go-difflib v1.0.0 // indirect
-	github.com/prometheus/procfs v0.8.0 // indirect
-	github.com/santhosh-tekuri/jsonschema v1.2.4 // indirect
-	go.elastic.co/fastjson v1.1.0 // indirect
-	golang.org/x/crypto v0.3.0 // indirect
-	golang.org/x/mod v0.7.0 // indirect
-	golang.org/x/sys v0.2.0 // indirect
-	golang.org/x/tools v0.3.0 // indirect
-	google.golang.org/protobuf v1.28.1 // indirect
+	github.com/prometheus/procfs v0.13.0 // indirect
+	github.com/rogpeppe/go-internal v1.12.0 // indirect
+	go.elastic.co/fastjson v1.3.0 // indirect
+	golang.org/x/crypto v0.21.0 // indirect
+	golang.org/x/net v0.22.0 // indirect
+	golang.org/x/sys v0.18.0 // indirect
+	golang.org/x/text v0.14.0 // indirect
+	golang.org/x/tools v0.19.0 // indirect
 	gopkg.in/yaml.v3 v3.0.1 // indirect
-	howett.net/plist v1.0.0 // indirect
+	howett.net/plist v1.0.1 // indirect
 )

module/apmelastic/publisher_interceptor.go

@@ -23,9 +23,9 @@ var PublisherElasticApmInterceptor = func() publisher.UnaryPublisherInterceptorF
 				)
 
 				span.Context.SetLabel("destination", req.GetDestination())
-				span.Context.SetDestinationService(apm.DestinationServiceSpanContext{
-					Name:     req.GetDestinationType(),
-					Resource: req.GetDestination(),
+				span.Context.SetServiceTarget(apm.ServiceTargetSpanContext{
+					Name: req.GetDestinationType(),
+					Type: req.GetDestination(),
 				})
 
 				defer span.End()

publisher/nats.go

@@ -5,8 +5,8 @@ import (
 	"encoding/json"
 	"fmt"
 
+	"github.com/cockroachdb/errors"
 	"github.com/nats-io/nats.go"
-	"github.com/pkg/errors"
 
 	"github.com/skynet2/eventsourcing/common"
 )
@@ -33,7 +33,7 @@ func (n *NatsPublisher[T]) Publish(
 	ctx context.Context,
 	record T,
 	meta common.MetaData,
-	headers map[string][]string,
+	publishOptions *PublishOptions,
 ) error {
 	data, err := json.Marshal(event[T]{
 		Record:   record,
@@ -44,17 +44,22 @@ func (n *NatsPublisher[T]) Publish(
 		return errors.WithStack(err)
 	}
 
+	subject := n.subject
+	if publishOptions != nil && publishOptions.CustomSubject != "" {
+		subject = publishOptions.CustomSubject
+	}
+
 	m := &nats.Msg{
-		Subject: n.subject,
+		Subject: subject,
 		Data:    data,
 		Header: map[string][]string{
 			"co":  {fmt.Sprint(meta.CrudOperation)},
 			"cor": {fmt.Sprint(meta.CrudOperationReason)},
 		},
 	}
 
-	if len(headers) > 0 {
-		for k, v := range headers {
+	if publishOptions != nil && len(publishOptions.Headers) > 0 {
+		for k, v := range publishOptions.Headers {
 			m.Header[k] = v
 		}
 	}

publisher/nats_test.go

@@ -68,8 +68,10 @@ func TestNatsPublisher(t *testing.T) {
 			}
 		})
 
-	assert.NoError(t, pub.Publish(context.TODO(), record, meta, map[string][]string{
-		"header1": {"value1"},
+	assert.NoError(t, pub.Publish(context.TODO(), record, meta, &publisher.PublishOptions{
+		Headers: map[string][]string{
+			"header1": {"value1"},
+		},
 	}))
 
 	con.Close()

publisher/types.go

@@ -11,7 +11,7 @@ type Publisher[T any] interface {
 		ctx context.Context,
 		record T,
 		meta common.MetaData,
-		headers map[string][]string,
+		headers *PublishOptions,
 	) error
 }
 
@@ -28,5 +28,10 @@ type event[T any] struct {
 	MetaData common.MetaData `json:"m"`
 }
 
+type PublishOptions struct {
+	Headers       map[string][]string
+	CustomSubject string
+}
+
 type UnaryPublisherInterceptorFunc = func(next UnaryPublisherFunc) UnaryPublisherFunc
 type UnaryPublisherFunc = func(ctx context.Context, event AnyEvent)