278 lines
9.8 KiB
Go
278 lines
9.8 KiB
Go
package service_test
|
|
|
|
import (
|
|
"context"
|
|
"errors"
|
|
"strings"
|
|
"sync"
|
|
"sync/atomic"
|
|
"testing"
|
|
"time"
|
|
|
|
"git.sendico.io/tech/sxchnge/notifications/bus"
|
|
"git.sendico.io/tech/sxchnge/notifications/email"
|
|
pb "git.sendico.io/tech/sxchnge/notifications/interface/notifications/v1"
|
|
"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"
|
|
"git.sendico.io/tech/sxchnge/shared/messaging/reliable"
|
|
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
|
|
mu sync.Mutex
|
|
messages map[string]model.Email
|
|
state model.State
|
|
}
|
|
|
|
func (s *sender) Send(_ context.Context, id string, message model.Email) email.Outcome {
|
|
s.mu.Lock()
|
|
if s.messages == nil {
|
|
s.messages = map[string]model.Email{}
|
|
}
|
|
s.messages[id] = message
|
|
s.mu.Unlock()
|
|
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:latest")
|
|
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)
|
|
}
|
|
// Expire both queued and retrying notifications without a connected broker.
|
|
for _, state := range []model.State{model.Queued, model.RetryWait} {
|
|
request.Key = "offline-expiry-" + string(state)
|
|
expired, e := svc.Send(ctx, "product", request)
|
|
if e != nil {
|
|
t.Fatal(e)
|
|
}
|
|
patch := repository.Patch().Set(repository.Field("state"), state).Set(repository.Field("expires_at"), time.Now().Add(-time.Minute)).Set(repository.Field("next_attempt_at"), time.Now().Add(time.Hour))
|
|
if e = repo.DBImp.Patch(ctx, expired.ID, patch); e != nil {
|
|
t.Fatal(e)
|
|
}
|
|
if e = repo.Recover(ctx, time.Now()); e != nil {
|
|
t.Fatal(e)
|
|
}
|
|
expired, e = svc.Get(ctx, "product", expired.ID.Hex())
|
|
if e != nil || expired.State != model.Expired || expired.Email.Address != "" {
|
|
t.Fatal("offline expiry failed", expired, e)
|
|
}
|
|
}
|
|
// Verify the real JetStream publisher/consumer path with an isolated broker.
|
|
ncContainer, err := testcontainers.GenericContainer(ctx, testcontainers.GenericContainerRequest{ContainerRequest: testcontainers.ContainerRequest{Image: "nats:latest", 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(), bus.DefaultConfig())
|
|
if err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
defer broker.Close()
|
|
// Production ops mode validates existing resources without silently rewriting drift.
|
|
queue := bus.DefaultConfig()
|
|
settings := reliable.DurableSettings{Stream: queue.Stream, Provisioning: reliable.DurableProvisioningSettings{Mode: reliable.DurableProvisioningOps, Environment: reliable.DurableEnvironmentProduction}}
|
|
if err = rt.DurableAdmin.EnsureProducerResources(ctx, settings); err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
settings.Stream.MaxAgeHours = 72
|
|
if err = rt.DurableAdmin.EnsureProducerResources(ctx, settings); !errors.Is(err, merrors.ErrDataConflict) {
|
|
t.Fatal("retention drift accepted", err)
|
|
}
|
|
settings.Stream.MaxAgeHours = 48
|
|
if err = rt.DurableAdmin.EnsureProducerResources(ctx, settings); err != nil {
|
|
t.Fatal("ops mode mutated the stream", err)
|
|
}
|
|
canceled, stopAdmin := context.WithCancel(ctx)
|
|
stopAdmin()
|
|
if err = rt.DurableAdmin.EnsureProducerResources(canceled, settings); err == nil {
|
|
t.Fatal("canceled provisioning succeeded")
|
|
}
|
|
|
|
client := grpcClient(t, svc)
|
|
rpcCtx := ctx
|
|
input := &pb.SendRequest{IdempotencyKey: "jetstream", UserId: "u", TemplateId: "test", TemplateVersion: "1", Parameters: map[string]string{"name": "End to end & escaped"}, Targets: []*pb.Target{{Destination: &pb.Target_Email{Email: &pb.EmailTarget{Address: "user@example.com"}}}}}
|
|
sent, err := client.Send(rpcCtx, input)
|
|
if err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
replay, err := client.Send(rpcCtx, input)
|
|
if err != nil || replay.Notification.Id != sent.Notification.Id {
|
|
t.Fatal("RPC idempotency failed", err)
|
|
}
|
|
if _, e := client.Get(ctx, &pb.GetRequest{Id: sent.Notification.Id}); e != nil {
|
|
t.Fatal("RPC retrieval without credentials failed", e)
|
|
}
|
|
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 {
|
|
result, e := client.Get(rpcCtx, &pb.GetRequest{Id: sent.Notification.Id})
|
|
if e != nil {
|
|
t.Fatal(e)
|
|
}
|
|
if strings.Contains(result.String(), "user@example.com") || strings.Contains(result.String(), "escaped") {
|
|
t.Fatal("response exposes email content")
|
|
}
|
|
if result.Notification.Deliveries[0].State == pb.DeliveryState_DELIVERY_STATE_ACCEPTED {
|
|
break
|
|
}
|
|
if time.Now().After(deadline) {
|
|
t.Fatal("JetStream job did not complete", result.Notification.Deliveries[0].State)
|
|
}
|
|
time.Sleep(20 * time.Millisecond)
|
|
}
|
|
driver.mu.Lock()
|
|
delivered := driver.messages[sent.Notification.Id]
|
|
driver.mu.Unlock()
|
|
if delivered.HTML != "<p>End to end & escaped</p>" || delivered.Subject != "Hello" || delivered.Address != "user@example.com" {
|
|
t.Fatal("parameters did not reach driver", delivered)
|
|
}
|
|
}
|