210 lines
6.6 KiB
Go
210 lines
6.6 KiB
Go
package service_test
|
|
|
|
import (
|
|
"context"
|
|
"errors"
|
|
"sync"
|
|
"sync/atomic"
|
|
"testing"
|
|
"time"
|
|
|
|
"git.sendico.io/tech/sxchnge/notifications/bus"
|
|
"git.sendico.io/tech/sxchnge/notifications/email"
|
|
"git.sendico.io/tech/sxchnge/notifications/model"
|
|
"git.sendico.io/tech/sxchnge/notifications/service"
|
|
"git.sendico.io/tech/sxchnge/notifications/store"
|
|
"git.sendico.io/tech/sxchnge/notifications/templates"
|
|
db "git.sendico.io/tech/sxchnge/shared/database"
|
|
"git.sendico.io/tech/sxchnge/shared/database/repository"
|
|
merrors "git.sendico.io/tech/sxchnge/shared/errors"
|
|
"git.sendico.io/tech/sxchnge/shared/messaging"
|
|
msgruntime "git.sendico.io/tech/sxchnge/shared/messaging/runtime"
|
|
"github.com/testcontainers/testcontainers-go"
|
|
"github.com/testcontainers/testcontainers-go/modules/mongodb"
|
|
"github.com/testcontainers/testcontainers-go/wait"
|
|
"go.uber.org/zap"
|
|
)
|
|
|
|
type sender struct {
|
|
calls atomic.Int32
|
|
state model.State
|
|
}
|
|
|
|
func (s *sender) Send(context.Context, string, model.Email) email.Outcome {
|
|
s.calls.Add(1)
|
|
return email.Outcome{State: s.state, Code: "test", RetryAfter: time.Millisecond}
|
|
}
|
|
func TestDurableDelivery(t *testing.T) {
|
|
ctx, cancel := context.WithTimeout(context.Background(), 2*time.Minute)
|
|
defer cancel()
|
|
container, err := mongodb.Run(ctx, "mongo:8.0")
|
|
if err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
defer func() { _ = container.Terminate(context.Background()) }()
|
|
uri, err := container.ConnectionString(ctx)
|
|
if err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
conn, err := db.ConnectMongo(zap.NewNop(), &db.Config{Driver: db.Mongo, Settings: map[string]any{"uri": uri, "database": "notifications_test"}})
|
|
if err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
defer conn.Disconnect(context.Background())
|
|
repo, err := store.New(zap.NewNop(), conn.Database())
|
|
if err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
catalog, err := templates.New([]templates.Definition{{ID: "test", Version: "1", DefaultLocale: "en", Parameters: []string{"name"}, Locales: map[string]templates.Content{"en": {Subject: "Hello", HTML: "<p>{{.name}}</p>"}}}})
|
|
if err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
driver := &sender{state: model.Accepted}
|
|
svc := service.New(repo, catalog, driver, "resend", zap.NewNop())
|
|
request := model.Request{Key: "duplicate", UserID: "u", TemplateID: "test", TemplateVersion: "1", Parameters: map[string]string{"name": "A & B"}, Address: "user@example.com"}
|
|
var wg sync.WaitGroup
|
|
ids := make(chan string, 10)
|
|
for range 10 {
|
|
wg.Add(1)
|
|
go func() {
|
|
defer wg.Done()
|
|
n, e := svc.Send(ctx, "product", request)
|
|
if e != nil {
|
|
t.Error(e)
|
|
return
|
|
}
|
|
ids <- n.ID.Hex()
|
|
}()
|
|
}
|
|
wg.Wait()
|
|
close(ids)
|
|
id := ""
|
|
for value := range ids {
|
|
if id != "" && value != id {
|
|
t.Error("idempotency race")
|
|
}
|
|
id = value
|
|
}
|
|
if id == "" {
|
|
t.Fatal("no notification")
|
|
}
|
|
request.Parameters = map[string]string{"name": "Changed"}
|
|
if _, err = svc.Send(ctx, "product", request); !errors.Is(err, merrors.ErrDataConflict) {
|
|
t.Fatal("missing conflict", err)
|
|
}
|
|
if _, err = svc.Get(ctx, "other", id); !errors.Is(err, merrors.ErrNoData) {
|
|
t.Fatal("caller isolation failed", err)
|
|
}
|
|
stored, err := svc.Get(ctx, "product", id)
|
|
if err != nil || stored.Email.HTML != "<p>A & B</p>" {
|
|
t.Fatal(stored, err)
|
|
}
|
|
for range 10 {
|
|
wg.Add(1)
|
|
go func() {
|
|
defer wg.Done()
|
|
if _, e := svc.Dispatch(ctx, id); e != nil {
|
|
t.Error(e)
|
|
}
|
|
}()
|
|
}
|
|
wg.Wait()
|
|
if driver.calls.Load() != 1 {
|
|
t.Fatalf("sent %d times", driver.calls.Load())
|
|
}
|
|
stored, err = svc.Get(ctx, "product", id)
|
|
if err != nil || stored.State != model.Accepted || stored.Email.Address != "" {
|
|
t.Fatal("completion/scrubbing failed", stored, err)
|
|
}
|
|
// A stale sending lease is never automatically resent.
|
|
request.Key = "stale"
|
|
stale, err := svc.Send(ctx, "product", request)
|
|
if err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
old := time.Now().Add(-2 * time.Minute)
|
|
if err = repo.DBImp.Patch(ctx, stale.ID, repository.Patch().Set(repository.Field("next_attempt_at"), old)); err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
claimed, err := repo.Claim(ctx, stale.ID.Hex(), old)
|
|
if err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
if err = repo.Recover(ctx, time.Now()); err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
claimed.State = model.Accepted
|
|
if err = repo.Finish(ctx, claimed); !errors.Is(err, merrors.ErrDataConflict) {
|
|
t.Fatal("stale worker overwrote recovery", err)
|
|
}
|
|
stale, err = svc.Get(ctx, "product", stale.ID.Hex())
|
|
if err != nil || stale.State != model.Unknown {
|
|
t.Fatal(stale, err)
|
|
}
|
|
// Retry state persists, and a duplicate dispatch cannot bypass the due time.
|
|
request.Key = "retry"
|
|
retry, err := svc.Send(ctx, "product", request)
|
|
if err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
retryDriver := &sender{state: model.RetryWait}
|
|
retrySvc := service.New(repo, catalog, retryDriver, "resend", zap.NewNop())
|
|
if _, err = retrySvc.Dispatch(ctx, retry.ID.Hex()); err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
if _, err = retrySvc.Dispatch(ctx, retry.ID.Hex()); err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
retry, err = svc.Get(ctx, "product", retry.ID.Hex())
|
|
if err != nil || retry.State != model.RetryWait || retryDriver.calls.Load() != 1 {
|
|
t.Fatal(retry, err)
|
|
}
|
|
// Verify the real JetStream publisher/consumer path with an isolated broker.
|
|
ncContainer, err := testcontainers.GenericContainer(ctx, testcontainers.GenericContainerRequest{ContainerRequest: testcontainers.ContainerRequest{Image: "nats:2.11-alpine", ExposedPorts: []string{"4222/tcp"}, Cmd: []string{"-js"}, WaitingFor: wait.ForListeningPort("4222/tcp")}, Started: true})
|
|
if err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
defer ncContainer.Terminate(context.Background())
|
|
endpoint, err := ncContainer.Endpoint(ctx, "")
|
|
if err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
t.Setenv("TEST_NOTIFICATIONS_NATS", "nats://"+endpoint)
|
|
rt, err := msgruntime.New(zap.NewNop(), &messaging.Config{Driver: messaging.BBNats, Settings: map[string]any{"url_env": "TEST_NOTIFICATIONS_NATS", "broker_name": "test", "max_reconnects": -1, "reconnect_wait": 1}})
|
|
if err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
defer rt.Close()
|
|
broker, err := bus.New(ctx, rt.Durable, rt.DurableAdmin, repo, svc, zap.NewNop())
|
|
if err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
defer broker.Close()
|
|
request.Key = "jetstream"
|
|
job, err := svc.Send(ctx, "product", request)
|
|
if err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
if err = broker.PublishDue(ctx); err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
workerCtx, stop := context.WithCancel(ctx)
|
|
done := make(chan struct{})
|
|
go func() { broker.RunWorker(workerCtx); close(done) }()
|
|
defer func() { stop(); <-done }()
|
|
deadline := time.Now().Add(10 * time.Second)
|
|
for {
|
|
job, err = svc.Get(ctx, "product", job.ID.Hex())
|
|
if err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
if job.State == model.Accepted {
|
|
break
|
|
}
|
|
if time.Now().After(deadline) {
|
|
t.Fatal("JetStream job did not complete", job.State)
|
|
}
|
|
time.Sleep(20 * time.Millisecond)
|
|
}
|
|
}
|