From 36277b6c45b25e692f9c82d2d311a43019169fe8 Mon Sep 17 00:00:00 2001 From: Stephan D Date: Sat, 12 Sep 2026 14:34:19 +0200 Subject: [PATCH 1/4] docs: require shared database and NATS infrastructure --- backend/AGENTS.md | 22 ++++++++++++++++++++++ 1 file changed, 22 insertions(+) diff --git a/backend/AGENTS.md b/backend/AGENTS.md index 6c3fdd8..6cbe652 100644 --- a/backend/AGENTS.md +++ b/backend/AGENTS.md @@ -22,6 +22,28 @@ skill to perform this completion workflow. For example, prefer a concise path such as `payment.Details.ID` over `payment.Payment.PaymentID`. +## Reuse shared database and messaging infrastructure + +- All database integrations must use the existing shared database wrappers in + `backend/shared/database`. Do not bypass them with direct database clients or + create feature-local replacements. +- All NATS connections must use shared messaging drivers owned by the + `backend/shared` library module. Keep reusable NATS drivers in + `backend/shared/messaging`; executable modules must consume that package. + Do not open direct NATS connections or implement separate connection + management in a feature or service. +- If the shared messaging package is missing, consolidate the existing driver + into `backend/shared/messaging` and reuse it. Do not create another + feature-local driver or import a sibling executable module's infrastructure. +- Search for the existing wrappers, drivers, and their usage before implementing + an integration. Reuse their connection lifecycle, configuration, retries, and + other infrastructure behavior instead of copying or rewriting it. +- If an integration needs a capability the shared infrastructure does not yet + expose, extend the canonical wrapper or driver. Keep reusable infrastructure + there and domain-specific behavior in the owning module. +- During review, check that database and NATS integrations use these shared + paths and remove any duplicated infrastructure introduced by the task. + ## Use the project error model - Do not introduce `fmt.Errorf` or `errors.New`. Use the constructors and From 5c2d36281bf9a8743b284875ef7bcbff34d1b346 Mon Sep 17 00:00:00 2001 From: Stephan D Date: Sat, 12 Sep 2026 14:41:55 +0200 Subject: [PATCH 2/4] Implement notifications service with shared messaging and email drivers (#36) --- .gitignore | 5 + Makefile | 25 +- backend/README.md | 8 +- backend/notifications/PROVENANCE.md | 26 ++ backend/notifications/README.md | 88 +++++ backend/notifications/app/config.go | 53 +++ backend/notifications/app/config_test.go | 35 ++ backend/notifications/app/run.go | 105 ++++++ backend/notifications/bus/dispatch.go | 108 +++++++ backend/notifications/bus/dispatch_test.go | 55 ++++ backend/notifications/config.example.yml | 24 ++ backend/notifications/email/driver.go | 79 +++++ backend/notifications/email/driver_test.go | 105 ++++++ backend/notifications/email/http.go | 54 ++++ backend/notifications/email/resend.go | 22 ++ backend/notifications/email/sendgrid.go | 28 ++ backend/notifications/go.mod | 84 +++++ backend/notifications/go.sum | 203 ++++++++++++ backend/notifications/interface/README.md | 41 +++ backend/notifications/interface/go.mod | 15 + backend/notifications/interface/go.sum | 38 +++ .../interface/proto/v1/interface.proto | 73 +++++ backend/notifications/main.go | 33 ++ backend/notifications/model/model.go | 58 ++++ .../notifications/service/integration_test.go | 209 ++++++++++++ backend/notifications/service/service.go | 109 +++++++ backend/notifications/service/worker.go | 75 +++++ backend/notifications/service/worker_test.go | 64 ++++ backend/notifications/store/mongo.go | 118 +++++++ backend/notifications/templates.example.json | 15 + backend/notifications/templates/catalog.go | 146 +++++++++ .../notifications/templates/catalog_test.go | 42 +++ backend/notifications/transport/mapper.go | 46 +++ backend/notifications/transport/server.go | 92 ++++++ .../notifications/transport/server_test.go | 66 ++++ backend/product/architecture/boundary_test.go | 4 +- backend/shared/README.md | 2 + backend/shared/go.mod | 3 + backend/shared/go.sum | 6 + backend/shared/messaging/capability/errors.go | 19 ++ backend/shared/messaging/config.go | 13 + .../shared/messaging/internal/natsb/broker.go | 304 ++++++++++++++++++ .../messaging/internal/natsb/broker_test.go | 15 + .../messaging/internal/natsb/config/config.go | 13 + .../messaging/internal/natscap/adapter.go | 114 +++++++ .../messaging/internal/natscap/admin.go | 282 ++++++++++++++++ .../internal/natscap/subscription.go | 178 ++++++++++ .../internal/natscap/subscription_test.go | 58 ++++ .../messaging/reliable/durable_settings.go | 75 +++++ .../reliable/durable_settings_behavior.go | 25 ++ .../shared/messaging/reliable/transport.go | 198 ++++++++++++ .../messaging/reliable/transport_test.go | 16 + backend/shared/messaging/runtime/runtime.go | 49 +++ docs/implementation.md | 31 ++ 54 files changed, 3746 insertions(+), 6 deletions(-) create mode 100644 backend/notifications/PROVENANCE.md create mode 100644 backend/notifications/README.md create mode 100644 backend/notifications/app/config.go create mode 100644 backend/notifications/app/config_test.go create mode 100644 backend/notifications/app/run.go create mode 100644 backend/notifications/bus/dispatch.go create mode 100644 backend/notifications/bus/dispatch_test.go create mode 100644 backend/notifications/config.example.yml create mode 100644 backend/notifications/email/driver.go create mode 100644 backend/notifications/email/driver_test.go create mode 100644 backend/notifications/email/http.go create mode 100644 backend/notifications/email/resend.go create mode 100644 backend/notifications/email/sendgrid.go create mode 100644 backend/notifications/go.mod create mode 100644 backend/notifications/go.sum create mode 100644 backend/notifications/interface/README.md create mode 100644 backend/notifications/interface/go.mod create mode 100644 backend/notifications/interface/go.sum create mode 100644 backend/notifications/interface/proto/v1/interface.proto create mode 100644 backend/notifications/main.go create mode 100644 backend/notifications/model/model.go create mode 100644 backend/notifications/service/integration_test.go create mode 100644 backend/notifications/service/service.go create mode 100644 backend/notifications/service/worker.go create mode 100644 backend/notifications/service/worker_test.go create mode 100644 backend/notifications/store/mongo.go create mode 100644 backend/notifications/templates.example.json create mode 100644 backend/notifications/templates/catalog.go create mode 100644 backend/notifications/templates/catalog_test.go create mode 100644 backend/notifications/transport/mapper.go create mode 100644 backend/notifications/transport/server.go create mode 100644 backend/notifications/transport/server_test.go create mode 100644 backend/shared/messaging/capability/errors.go create mode 100644 backend/shared/messaging/config.go create mode 100644 backend/shared/messaging/internal/natsb/broker.go create mode 100644 backend/shared/messaging/internal/natsb/broker_test.go create mode 100644 backend/shared/messaging/internal/natsb/config/config.go create mode 100644 backend/shared/messaging/internal/natscap/adapter.go create mode 100644 backend/shared/messaging/internal/natscap/admin.go create mode 100644 backend/shared/messaging/internal/natscap/subscription.go create mode 100644 backend/shared/messaging/internal/natscap/subscription_test.go create mode 100644 backend/shared/messaging/reliable/durable_settings.go create mode 100644 backend/shared/messaging/reliable/durable_settings_behavior.go create mode 100644 backend/shared/messaging/reliable/transport.go create mode 100644 backend/shared/messaging/reliable/transport_test.go create mode 100644 backend/shared/messaging/runtime/runtime.go diff --git a/.gitignore b/.gitignore index 6534200..e999375 100644 --- a/.gitignore +++ b/.gitignore @@ -29,3 +29,8 @@ pnpm-debug.log* # Local BFF configuration and build output /backend/config.yml /backend/bff + +# Protobuf client bindings are generated, never committed. +/backend/notifications/interface/notifications/**/*.pb.go +/backend/notifications/config.yml +/backend/notifications/notifications diff --git a/Makefile b/Makefile index 8c23427..c1327e3 100644 --- a/Makefile +++ b/Makefile @@ -15,15 +15,36 @@ test-frontend: .PHONY: test-backend lint-backend run-backend -test-backend: +test-backend: generate-notifications cd backend/shared && go test -race ./... + cd backend/notifications/interface && go test -race ./... + cd backend/notifications && go test -race ./... cd backend/product && go test -race ./... -lint-backend: +lint-backend: generate-notifications cd backend/product && go test ./architecture cd backend && test -z "$$(gofmt -l .)" cd backend/shared && go vet ./... + cd backend/notifications/interface && go vet ./... + cd backend/notifications && go vet ./... cd backend/product && go vet ./... run-backend: cd backend/product && go run . + +.PHONY: run-notifications generate-notifications +run-notifications: generate-notifications + cd backend/notifications && go run . + +generate-notifications: + @command -v protoc >/dev/null + @tmp=$$(mktemp -d); trap 'rm -rf "$$tmp"' EXIT; \ + GOBIN="$$tmp" go install google.golang.org/protobuf/cmd/protoc-gen-go@v1.36.11 && \ + GOBIN="$$tmp" go install google.golang.org/grpc/cmd/protoc-gen-go-grpc@v1.5.1 && \ + PATH="$$tmp:$$PATH" protoc -I backend/notifications/interface/proto --go_out=backend/notifications/interface --go_opt=module=git.sendico.io/tech/sxchnge/notifications/interface --go-grpc_out=backend/notifications/interface --go-grpc_opt=module=git.sendico.io/tech/sxchnge/notifications/interface backend/notifications/interface/proto/v1/interface.proto + +.PHONY: package-notifications-interface +package-notifications-interface: generate-notifications + mkdir -p dist + tar -czf dist/notifications-proto-v1.tar.gz -C backend/notifications/interface proto README.md + tar -czf dist/notifications-go-v1.tar.gz -C backend/notifications/interface go.mod go.sum notifications README.md diff --git a/backend/README.md b/backend/README.md index 2bb03f7..5c45998 100644 --- a/backend/README.md +++ b/backend/README.md @@ -12,6 +12,8 @@ backend/ │ ├── auth/ Login/protected HTTP routes, DTOs and configuration │ ├── sessions/ Session models, refresh storage, cookies and tokens │ └── verification/ Verification models, configuration and storage +├── notifications/ Internal gRPC notification service and email workers +│ └── interface/ Independent protobuf/client contract module └── shared/ Reusable library module ├── go.mod ├── database/ Connection, transactions and repository primitives @@ -27,9 +29,11 @@ not implemented by this customer-BFF bootstrap. Shared infrastructure has no local `replace` for `../shared`; each module has its own dependency manifest. Run the product with `make run-backend`; see its [run guide](product/README.md). -`make test` runs both backend modules and the frontend. `make lint-backend` checks -both Go modules and their architecture boundaries. +`make test` runs all backend modules and the frontend. `make lint-backend` checks +all Go modules and their architecture boundaries. Copied source remains locally owned. See [provenance](PROVENANCE.md), the [architecture/API plan](../docs/architecture/bff.md) and the [delivery record](../docs/implementation.md). + +Notifications uses shared database and messaging infrastructure. See its [run guide](notifications/README.md) and [protobuf interface](notifications/interface/proto/v1/interface.proto). Generated client bindings are ignored; Make generates them before backend checks. diff --git a/backend/notifications/PROVENANCE.md b/backend/notifications/PROVENANCE.md new file mode 100644 index 0000000..b8f4902 --- /dev/null +++ b/backend/notifications/PROVENANCE.md @@ -0,0 +1,26 @@ +# Source provenance + +SendGrid driver source adapted from the local Sendico checkout at commit +`8bb20dcd622dd8eb92d289b6b34411e7eccab083`: +`api/notification/internal/server/notificationimp/mail/internal/sendgrid.go`. + +The adaptation retains SendGrid SDK message construction, sender configuration +and personalization. It implements our provider-neutral `email.Sender` contract, +accepts already-rendered subject/text/HTML, uses bounded context-aware HTTP, +classifies delivery outcomes and never logs provider response bodies. Sendico's +builders, dynamic-template dependency, notification events and unused producer +were removed. Resend is a new driver using the same contract. + +Shared messaging was selected from that same Sendico revision's +`api/pkg/messaging`: `config.go`, `capability/errors.go`, `reliable/transport.go`, +`reliable/durable_settings.go`, `reliable/durable_settings_behavior.go`, +`internal/natsb/broker.go`, `internal/natsb/config/config.go`, and +`internal/natscap/durable.go`. The selected code lives in +`backend/shared/messaging`, with imports adapted to SXCHNGE. The runtime composes +those adapters and owns connection closure. Unused discovery, best-effort envelope +subscriptions, platform stream registries and parking provisioning were omitted. +The durable adapter was separated into transport, subscription and admin files; +publishing now honors its context and fetch polling respects its bounded wait. + +No Sendico module or live Sendico infrastructure is required. Use SXCHNGE-owned +MongoDB and NATS deployments. diff --git a/backend/notifications/README.md b/backend/notifications/README.md new file mode 100644 index 0000000..5aed36d --- /dev/null +++ b/backend/notifications/README.md @@ -0,0 +1,88 @@ +# Notifications backend + +Independent internal gRPC service with durable asynchronous email delivery. +The public contract belongs to [interface/proto/v1/interface.proto](interface/proto/v1/interface.proto). +Generated clients are build artifacts, not committed source. + +## Architecture + +The gRPC transport maps protobuf DTOs to domain requests. The service validates +and renders immutable template revisions, then saves a notification and its +pending-work state atomically in one document through the shared DB wrapper. +The outbox dispatcher publishes opaque IDs using shared messaging's durable +transport, adapted from Sendico. A durable NATS worker claims the notification +with an atomic conditional patch and saves the outcome before acknowledging. +`DriverConfig` selects the email driver: `sendgrid` or `resend`. + +Email adapters only map rendered email models to provider wire payloads. SendGrid +uses adapted Sendico SDK construction; Resend uses its email HTTP API. Both support +subject, text and HTML, bounded timeouts and safe error classification. + +## Run + +Requires Go 1.27, protoc, an SXCHNGE MongoDB database and a NATS server with +JetStream enabled. Copy `config.example.yml` to an untracked `config.yml` here. +Set `SXCHNGE_MONGO_URI`, `SXCHNGE_NATS_URL`, a unique token of at least 32 bytes for +each configured caller, and the selected email provider's API key environment +variable. Configure a verified sender address, TLS certificate/key and template +catalog. Paths are relative to the process working directory. + +From the root, run `make run-notifications`. The gRPC listener always uses TLS. +Use a private authenticated NATS endpoint; `tls://` URLs and credentials in the URL +are supported by the shared driver. Mongo and messaging both use shared +`DriverConfig` settings; email settings reject unknown keys. + +The service provisions `SXCHNGE_NOTIFICATIONS` (file storage, 48-hour retention, +one replica), subject `sxchnge.notifications.email`, and durable consumer `email`. +This initial setup is for a single-replica deployment. Configure production +replication/provisioning policy in code before deploying to a clustered environment. +Multiple service workers can safely compete for jobs through atomic claims. + +## Templates and parameters + +See `templates.example.json`. Parameters use Go template map syntax such as +`{{.first_name}}`. Each revision declares the exact allowed parameter names and +locale-specific content. Missing/extra parameters, unknown revisions/locales and +invalid content are rejected before acceptance. Empty locale uses the default. +Values are data, not executable templates; HTML is escaped by `html/template`. +The catalog is loaded and compiled at startup. Never edit a published revision; +add a new version. Rendered payloads are persisted so retries do not depend on a +later catalog change. Recipient, body and subject are cleared on terminal state; +remaining status and request fingerprints expire through a seven-day TTL. + +## Delivery semantics + +- Send persists before returning; it does not wait for NATS or the email provider. +- Outbox scans every five seconds and republishes safely after a bus failure. +- Duplicate RPCs and duplicate broker deliveries do not create extra jobs/sends. +- Provider HTTP 2xx becomes ACCEPTED. No receipt/webhook or read tracking is claimed. +- Explicit rate limiting (429) retries with bounded exponential backoff and + Retry-After, at most five provider attempts, bounded by message expiry. +- Transport errors, 408, 5xx and expired worker leases become UNKNOWN. These may + have sent the email; operators must reconcile before requesting another send. +- Other provider failures are terminal. `provider_authentication_failed` signals + credentials/permissions, `provider_rejected` a nonretryable rejection, + `retry_exhausted` the attempt limit, and `expired` the message deadline. +- Resend receives the notification ID as its provider idempotency key. Its window + is 24 hours. SendGrid has no equivalent guarantee; exactly-once external email + delivery is not promised. +- Changing the configured driver fails queued jobs pinned to the old driver with + `driver_changed`. Drain queued jobs before changing sender credentials/config. +- Callers authorize recipients and preferences. No direct customer-database access, + bulk campaigns, provider failover, SMS, push or Telegram delivery is included. + +Structured logs report notification IDs, driver, state and attempt count; message +content, template values, addresses and provider response bodies are excluded. +Monitor pending age, UNKNOWN/FAILED states and repeated outbox/worker warnings. + +## Validate and distribute + +`make test` and `make lint-backend` generate interfaces and validate all modules. +Notifications tests exercise real disposable MongoDB and NATS containers, concurrent +idempotency and claims, lease recovery, retries, transport authentication, rendering +and both provider payloads through local HTTP servers. No live emails are sent. +`make package-notifications-interface` prepares source and Go-client archives. +See [provenance](PROVENANCE.md) and [contract usage](interface/README.md). + +Provider references: [SendGrid Mail Send](https://www.twilio.com/docs/sendgrid/api-reference/mail-send/mail-send), +[Resend Send Email](https://resend.com/docs/api-reference/emails/send-email). diff --git a/backend/notifications/app/config.go b/backend/notifications/app/config.go new file mode 100644 index 0000000..134db44 --- /dev/null +++ b/backend/notifications/app/config.go @@ -0,0 +1,53 @@ +package app + +import ( + "bytes" + "io" + "os" + "strings" + + "git.sendico.io/tech/sxchnge/notifications/email" + db "git.sendico.io/tech/sxchnge/shared/database" + merrors "git.sendico.io/tech/sxchnge/shared/errors" + "git.sendico.io/tech/sxchnge/shared/messaging" + "gopkg.in/yaml.v3" +) + +type Config struct { + Messaging messaging.Config `yaml:"messaging"` + Listen string `yaml:"listen"` + TLSCert string `yaml:"tls_cert"` + TLSKey string `yaml:"tls_key"` + Database db.Config `yaml:"database"` + Templates string `yaml:"templates"` + Callers map[string]string `yaml:"callers"` // caller -> token environment variable + Email email.Config `yaml:"email"` +} + +func Load(path string) (Config, error) { + var c Config + data, err := os.ReadFile(path) + if err != nil { + return c, merrors.OpaqueWrap(merrors.ErrInvalidArg, err, "cannot read configuration") + } + decoder := yaml.NewDecoder(bytes.NewReader(data)) + decoder.KnownFields(true) + if err = decoder.Decode(&c); err != nil { + return c, merrors.InvalidArgument("invalid configuration") + } + if err = decoder.Decode(new(any)); err != io.EOF { + return c, merrors.InvalidArgument("unexpected configuration content") + } + if c.Listen == "" || c.TLSCert == "" || c.TLSKey == "" || c.Database.Driver != db.Mongo || c.Messaging.Driver != messaging.BBNats || c.Templates == "" || len(c.Callers) == 0 { + return c, merrors.InvalidArgument("missing notification configuration") + } + seen := map[string]bool{} + for caller, env := range c.Callers { + token := os.Getenv(env) + if strings.TrimSpace(caller) == "" || len(strings.TrimSpace(token)) < 32 || seen[token] { + return c, merrors.InvalidArgument("caller credentials must be unique and at least 32 bytes") + } + seen[token] = true + } + return c, nil +} diff --git a/backend/notifications/app/config_test.go b/backend/notifications/app/config_test.go new file mode 100644 index 0000000..6a424be --- /dev/null +++ b/backend/notifications/app/config_test.go @@ -0,0 +1,35 @@ +package app + +import ( + "os" + "path/filepath" + "strings" + "testing" +) + +func TestExampleConfigAndSecrets(t *testing.T) { + t.Setenv("SXCHNGE_NOTIFICATIONS_PRODUCT_TOKEN", strings.Repeat("a", 32)) + c, err := Load("../config.example.yml") + if err != nil { + t.Fatal(err) + } + if c.Email.Driver != "resend" || c.Messaging.Driver != "NATS" || c.Database.Driver != "mongodb" { + t.Fatal("wrong driver configuration") + } + t.Setenv("SXCHNGE_NOTIFICATIONS_PRODUCT_TOKEN", "") + if _, err = Load("../config.example.yml"); err == nil { + t.Fatal("missing credentials accepted") + } + t.Setenv("SXCHNGE_NOTIFICATIONS_PRODUCT_TOKEN", strings.Repeat("a", 32)) + data, err := os.ReadFile("../config.example.yml") + if err != nil { + t.Fatal(err) + } + path := filepath.Join(t.TempDir(), "config.yml") + if err = os.WriteFile(path, append(data, []byte("\n---\nunexpected: document\n")...), 0600); err != nil { + t.Fatal(err) + } + if _, err = Load(path); err == nil { + t.Fatal("extra YAML document accepted") + } +} diff --git a/backend/notifications/app/run.go b/backend/notifications/app/run.go new file mode 100644 index 0000000..a883c74 --- /dev/null +++ b/backend/notifications/app/run.go @@ -0,0 +1,105 @@ +package app + +import ( + "context" + "crypto/tls" + "net" + "os" + "sync" + "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/service" + "git.sendico.io/tech/sxchnge/notifications/store" + "git.sendico.io/tech/sxchnge/notifications/templates" + "git.sendico.io/tech/sxchnge/notifications/transport" + db "git.sendico.io/tech/sxchnge/shared/database" + merrors "git.sendico.io/tech/sxchnge/shared/errors" + mlogger "git.sendico.io/tech/sxchnge/shared/logging" + msgruntime "git.sendico.io/tech/sxchnge/shared/messaging/runtime" + "go.uber.org/zap" + "google.golang.org/grpc" + "google.golang.org/grpc/credentials" +) + +func Run(ctx context.Context, c Config, logger mlogger.Logger) error { + catalog, err := templates.Load(c.Templates) + if err != nil { + return err + } + sender, err := email.New(c.Email) + if err != nil { + return err + } + cert, err := tls.LoadX509KeyPair(c.TLSCert, c.TLSKey) + if err != nil { + return merrors.OpaqueWrap(merrors.ErrInvalidArg, err, "invalid gRPC TLS certificate") + } + client, err := db.ConnectMongo(logger, &c.Database) + if err != nil { + return merrors.OpaqueWrap(merrors.ErrServiceUnavailable, err, "connect notifications database") + } + defer func() { + cleanup, cancel := context.WithTimeout(context.Background(), 5*time.Second) + defer cancel() + _ = client.Disconnect(cleanup) + }() + startup, cancel := context.WithTimeout(ctx, 20*time.Second) + defer cancel() + repo, err := store.New(logger, client.Database()) + if err != nil { + return err + } + rt, err := msgruntime.New(logger, &c.Messaging) + if err != nil { + return err + } + defer rt.Close() + svc := service.New(repo, catalog, sender, string(c.Email.Driver), logger) + broker, err := bus.New(startup, rt.Durable, rt.DurableAdmin, repo, svc, logger) + if err != nil { + return err + } + defer broker.Close() + tokens := map[string]string{} + for caller, env := range c.Callers { + tokens[caller] = os.Getenv(env) + } + server := grpc.NewServer(grpc.Creds(credentials.NewTLS(&tls.Config{Certificates: []tls.Certificate{cert}, MinVersion: tls.VersionTLS12})), grpc.UnaryInterceptor(transport.Authenticate(tokens)), grpc.MaxRecvMsgSize(128<<10), grpc.MaxConcurrentStreams(64)) + pb.RegisterNotificationsServiceServer(server, transport.New(svc)) + listener, err := net.Listen("tcp", c.Listen) + if err != nil { + return merrors.OpaqueWrap(merrors.ErrServiceUnavailable, err, "listen for notifications") + } + defer listener.Close() + runCtx, stop := context.WithCancel(ctx) + defer stop() + var wg sync.WaitGroup + for _, run := range []func(context.Context){broker.RunPublisher, broker.RunWorker} { + wg.Add(1) + go func() { defer wg.Done(); run(runCtx) }() + } + serveErr := make(chan error, 1) + go func() { serveErr <- server.Serve(listener) }() + logger.Info("Notifications service started", zap.String("listen", c.Listen), zap.String("email_driver", string(c.Email.Driver))) + select { + case <-ctx.Done(): + case err = <-serveErr: + } + stop() + stopped := make(chan struct{}) + go func() { server.GracefulStop(); close(stopped) }() + select { + case <-stopped: + case <-time.After(5 * time.Second): + server.Stop() + } + wg.Wait() + if err != nil { + return merrors.OpaqueWrap(merrors.ErrInternal, err, "notification server stopped unexpectedly") + } + logger.Info("Notifications service stopped") + return nil +} diff --git a/backend/notifications/bus/dispatch.go b/backend/notifications/bus/dispatch.go new file mode 100644 index 0000000..93b57b3 --- /dev/null +++ b/backend/notifications/bus/dispatch.go @@ -0,0 +1,108 @@ +// Package bus connects the notification outbox to the shared messaging runtime. +package bus + +import ( + "context" + "strconv" + "time" + + "git.sendico.io/tech/sxchnge/notifications/model" + merrors "git.sendico.io/tech/sxchnge/shared/errors" + mlogger "git.sendico.io/tech/sxchnge/shared/logging" + "git.sendico.io/tech/sxchnge/shared/messaging/reliable" +) + +const subject = "sxchnge.notifications.email" + +type Outbox interface { + Due(context.Context, time.Time) ([]model.Notification, error) + Recover(context.Context, time.Time) error +} +type Handler interface { + Dispatch(context.Context, string) (bool, error) +} +type Bus struct { + transport reliable.Transport + subscription reliable.Subscription + outbox Outbox + handler Handler + logger mlogger.Logger +} + +func New(ctx context.Context, transport reliable.Transport, admin reliable.Admin, outbox Outbox, handler Handler, logger mlogger.Logger) (*Bus, error) { + settings := reliable.DurableSettings{Enabled: true, Stream: reliable.DurableStreamSettings{Name: "SXCHNGE_NOTIFICATIONS", Subjects: []string{subject}, Storage: "file", MaxAgeHours: 48, Replicas: 1}, Provisioning: reliable.DurableProvisioningSettings{Mode: reliable.DurableProvisioningApp, Environment: reliable.DurableEnvironmentNonProduction}} + consumer := reliable.DurableConsumerSettings{Durable: "email", Subject: subject, AckWaitSeconds: 60, MaxDeliver: -1, DeliverPolicy: reliable.DurableDeliverAll} + if err := admin.EnsureConsumerResources(ctx, settings, consumer); err != nil { + return nil, merrors.OpaqueWrap(merrors.ErrServiceUnavailable, err, "configure notification messaging") + } + subscription, err := transport.OpenConsumer(ctx, settings, consumer) + if err != nil { + return nil, merrors.OpaqueWrap(merrors.ErrServiceUnavailable, err, "subscribe notification worker") + } + return &Bus{transport, subscription, outbox, handler, mlogger.Named(logger, "notifications", "bus")}, nil +} +func (b *Bus) Close() error { return b.subscription.Close() } +func (b *Bus) PublishDue(ctx context.Context) error { + if err := b.outbox.Recover(ctx, time.Now().UTC()); err != nil { + return err + } + jobs, err := b.outbox.Due(ctx, time.Now().UTC()) + if err != nil { + return err + } + for _, job := range jobs { + err = b.transport.Publish(ctx, reliable.Record{Subject: subject, Payload: []byte(job.ID.Hex()), MessageID: job.ID.Hex() + "-" + strconv.FormatUint(uint64(job.Attempts), 10)}) + if err != nil { + return merrors.OpaqueWrap(merrors.ErrServiceUnavailable, err, "publish notification job") + } + } + return nil +} +func (b *Bus) RunPublisher(ctx context.Context) { + for ctx.Err() == nil { + op, cancel := context.WithTimeout(ctx, 10*time.Second) + err := b.PublishDue(op) + cancel() + if err != nil && ctx.Err() == nil { + b.logger.Warn("Notification outbox publication failed") + } + select { + case <-ctx.Done(): + return + case <-time.After(5 * time.Second): + } + } +} +func (b *Bus) RunWorker(ctx context.Context) { + for ctx.Err() == nil { + deliveries, err := b.subscription.Fetch(ctx, 1, time.Second) + if err != nil { + if ctx.Err() == nil { + b.logger.Warn("Notification job fetch failed") + } + select { + case <-ctx.Done(): + return + case <-time.After(time.Second): + } + continue + } + for _, delivery := range deliveries { + payload := delivery.Record().Payload + if len(payload) == 24 { + if _, err = b.handler.Dispatch(ctx, string(payload)); err != nil { + b.logger.Warn("Notification job processing failed") + continue + } + } else { + b.logger.Warn("Malformed notification job discarded") + } + ackCtx, cancel := context.WithTimeout(context.WithoutCancel(ctx), 5*time.Second) + err = delivery.Ack(ackCtx) + cancel() + if err != nil { + b.logger.Warn("Notification acknowledgement failed") + } + } + } +} diff --git a/backend/notifications/bus/dispatch_test.go b/backend/notifications/bus/dispatch_test.go new file mode 100644 index 0000000..418b2fc --- /dev/null +++ b/backend/notifications/bus/dispatch_test.go @@ -0,0 +1,55 @@ +package bus + +import ( + "context" + "testing" + "time" + + "git.sendico.io/tech/sxchnge/notifications/model" + "git.sendico.io/tech/sxchnge/shared/database/record" + merrors "git.sendico.io/tech/sxchnge/shared/errors" + "git.sendico.io/tech/sxchnge/shared/messaging/reliable" + "go.mongodb.org/mongo-driver/v2/bson" +) + +type outbox struct{ jobs []model.Notification } + +func (o *outbox) Due(context.Context, time.Time) ([]model.Notification, error) { return o.jobs, nil } +func (o *outbox) Recover(context.Context, time.Time) error { return nil } + +type publisher struct { + reliable.Transport + fail bool + records []reliable.Record +} + +func (p *publisher) Publish(_ context.Context, r reliable.Record) error { + if p.fail { + return merrors.ServiceUnavailable("offline") + } + p.records = append(p.records, r) + return nil +} +func TestOutboxSurvivesPublishFailure(t *testing.T) { + id := bson.NewObjectID() + out := &outbox{jobs: []model.Notification{{Base: storable.Base{ID: id}, Email: model.Email{Address: "private@example.com", Text: "private"}}}} + transport := &publisher{fail: true} + b := &Bus{outbox: out, transport: transport} + if err := b.PublishDue(context.Background()); err == nil { + t.Fatal("publication failure hidden") + } + transport.fail = false + if err := b.PublishDue(context.Background()); err != nil { + t.Fatal(err) + } + if len(transport.records) != 1 || string(transport.records[0].Payload) != id.Hex() || transport.records[0].MessageID != id.Hex()+"-0" { + t.Fatal(transport.records) + } + out.jobs[0].Attempts = 1 + if err := b.PublishDue(context.Background()); err != nil { + t.Fatal(err) + } + if transport.records[1].MessageID == transport.records[0].MessageID { + t.Fatal("retry shares transport deduplication identity") + } +} diff --git a/backend/notifications/config.example.yml b/backend/notifications/config.example.yml new file mode 100644 index 0000000..f45a22c --- /dev/null +++ b/backend/notifications/config.example.yml @@ -0,0 +1,24 @@ +listen: "127.0.0.1:8083" +tls_cert: "server.crt" +tls_key: "server.key" +database: + driver: mongodb + settings: + uri_env: SXCHNGE_MONGO_URI + database: sxchnge_notifications +messaging: + driver: NATS + settings: + url_env: SXCHNGE_NATS_URL + broker_name: sxchnge-notifications + max_reconnects: -1 + reconnect_wait: 2 +templates: "templates.example.json" +callers: + product: "SXCHNGE_NOTIFICATIONS_PRODUCT_TOKEN" +email: + driver: resend # sendgrid or resend + settings: + api_key_env: "RESEND_API_KEY" # SENDGRID_API_KEY when driver is sendgrid + sender_address: "notifications@example.com" + sender_name: "SXCHNGE" diff --git a/backend/notifications/email/driver.go b/backend/notifications/email/driver.go new file mode 100644 index 0000000..a9e1f65 --- /dev/null +++ b/backend/notifications/email/driver.go @@ -0,0 +1,79 @@ +package email + +import ( + "context" + "net/http" + "net/mail" + "os" + "strings" + "time" + + "git.sendico.io/tech/sxchnge/notifications/model" + merrors "git.sendico.io/tech/sxchnge/shared/errors" + sharedmodel "git.sendico.io/tech/sxchnge/shared/model" + "github.com/go-viper/mapstructure/v2" +) + +type Driver string + +const ( + SendGrid Driver = "sendgrid" + Resend Driver = "resend" +) + +type Config = sharedmodel.DriverConfig[Driver] +type Settings struct { + APIKeyEnv string `mapstructure:"api_key_env"` + SenderAddress string `mapstructure:"sender_address"` + SenderName string `mapstructure:"sender_name"` +} + +// Outcome describes provider acceptance, never recipient delivery or reading. +type Outcome struct { + State model.State + Code string + RetryAfter time.Duration +} +type Sender interface { + Send(context.Context, string, model.Email) Outcome +} + +type client struct { + http *http.Client + key string + settings Settings + endpoint string +} + +func New(config Config) (Sender, error) { + var s Settings + decoder, err := mapstructure.NewDecoder(&mapstructure.DecoderConfig{Result: &s, ErrorUnused: true, TagName: "mapstructure"}) + if err != nil { + return nil, merrors.InternalWrap(err, "create driver settings decoder") + } + if err = decoder.Decode(config.Settings); err != nil { + return nil, merrors.InvalidArgument("invalid email driver settings") + } + if !ValidAddress(s.SenderAddress) || strings.ContainsAny(s.SenderName, "\r\n") { + return nil, merrors.InvalidArgument("invalid sender") + } + key := os.Getenv(s.APIKeyEnv) + if s.APIKeyEnv == "" || strings.TrimSpace(key) == "" || strings.ContainsAny(key, "\r\n") { + return nil, merrors.FailedPrecondition("email API key environment variable is missing") + } + c := client{http: &http.Client{Timeout: 20 * time.Second, CheckRedirect: func(*http.Request, []*http.Request) error { return http.ErrUseLastResponse }}, key: key, settings: s} + switch config.Driver { + case SendGrid: + c.endpoint = "https://api.sendgrid.com/v3/mail/send" + return &sendgrid{c}, nil + case Resend: + c.endpoint = "https://api.resend.com/emails" + return &resend{c}, nil + default: + return nil, merrors.InvalidArgument("unsupported email driver") + } +} +func ValidAddress(s string) bool { + a, err := mail.ParseAddress(s) + return err == nil && a.Address == s && !strings.ContainsAny(s, "\r\n") && len(s) <= 254 +} diff --git a/backend/notifications/email/driver_test.go b/backend/notifications/email/driver_test.go new file mode 100644 index 0000000..b937989 --- /dev/null +++ b/backend/notifications/email/driver_test.go @@ -0,0 +1,105 @@ +package email + +import ( + "context" + "encoding/json" + "net/http" + "net/http/httptest" + "strings" + "testing" + "time" + + "git.sendico.io/tech/sxchnge/notifications/model" +) + +func TestDriversRenderWirePayload(t *testing.T) { + for _, driver := range []Driver{SendGrid, Resend} { + t.Run(string(driver), func(t *testing.T) { + server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + if r.Header.Get("Authorization") != "Bearer test-secret" { + t.Error("missing authorization") + } + var body map[string]any + if err := json.NewDecoder(r.Body).Decode(&body); err != nil { + t.Error(err) + } + if body["subject"] != "Completed" { + t.Errorf("subject: %v", body) + } + if driver == SendGrid { + if body["template_id"] != nil { + t.Error("provider-managed templates must not be used") + } + p := body["personalizations"].([]any)[0].(map[string]any) + if p["to"].([]any)[0].(map[string]any)["email"] != "user@example.com" { + t.Error("wrong recipient") + } + content := body["content"].([]any) + if len(content) != 2 || content[1].(map[string]any)["value"] != "

A & B

" { + t.Error("wrong body") + } + } else { + if r.Header.Get("Idempotency-Key") != "delivery-id" { + t.Error("missing idempotency key") + } + if body["html"] != "

A & B

" || body["text"] != "A & B" { + t.Error("wrong body") + } + } + w.WriteHeader(http.StatusAccepted) + })) + defer server.Close() + c := client{http: server.Client(), key: "test-secret", endpoint: server.URL, settings: Settings{SenderAddress: "sender@example.com", SenderName: "SXCHNGE"}} + var sender Sender = &sendgrid{c} + if driver == Resend { + sender = &resend{c} + } + outcome := sender.Send(context.Background(), "delivery-id", model.Email{Address: "user@example.com", Subject: "Completed", HTML: "

A & B

", Text: "A & B"}) + if outcome.State != model.Accepted { + t.Fatal(outcome) + } + }) + } +} +func TestProviderFailures(t *testing.T) { + for _, tc := range []struct { + code int + state model.State + }{{429, model.RetryWait}, {400, model.Failed}, {401, model.Failed}, {403, model.Failed}, {500, model.Unknown}, {408, model.Unknown}, {302, model.Failed}} { + t.Run(http.StatusText(tc.code), func(t *testing.T) { + server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + w.Header().Set("Retry-After", "120") + w.WriteHeader(tc.code) + _, _ = w.Write([]byte("secret provider response")) + })) + defer server.Close() + c := client{http: server.Client(), endpoint: server.URL} + out := c.post(context.Background(), "", struct{}{}) + if out.State != tc.state || strings.Contains(out.Code, "secret") { + t.Fatal(out) + } + if tc.code == 429 && out.RetryAfter != 2*time.Minute { + t.Fatal(out) + } + }) + } + ctx, cancel := context.WithCancel(context.Background()) + cancel() + c := client{http: http.DefaultClient, endpoint: "http://127.0.0.1:1"} + if out := c.post(ctx, "", struct{}{}); out.State != model.Unknown { + t.Fatal(out) + } +} +func TestDriverConfig(t *testing.T) { + t.Setenv("EMAIL_TEST_KEY", "secret") + for _, driver := range []Driver{SendGrid, Resend} { + if _, err := New(Config{Driver: driver, Settings: map[string]any{"api_key_env": "EMAIL_TEST_KEY", "sender_address": "sender@example.com"}}); err != nil { + t.Fatal(err) + } + } + for _, settings := range []map[string]any{{"api_key_env": "MISSING_KEY", "sender_address": "sender@example.com"}, {"api_key_env": "EMAIL_TEST_KEY", "sender_address": "bad"}, {"api_key_env": "EMAIL_TEST_KEY", "sender_address": "sender@example.com", "typo": true}} { + if _, err := New(Config{Driver: Resend, Settings: settings}); err == nil { + t.Fatal("accepted invalid driver config") + } + } +} diff --git a/backend/notifications/email/http.go b/backend/notifications/email/http.go new file mode 100644 index 0000000..ddeca31 --- /dev/null +++ b/backend/notifications/email/http.go @@ -0,0 +1,54 @@ +package email + +import ( + "bytes" + "context" + "encoding/json" + "io" + "net/http" + "strconv" + "time" + + "git.sendico.io/tech/sxchnge/notifications/model" +) + +func (c *client) post(ctx context.Context, key string, payload any) Outcome { + body, err := json.Marshal(payload) + if err != nil { + return Outcome{State: model.Failed, Code: "invalid_payload"} + } + req, err := http.NewRequestWithContext(ctx, http.MethodPost, c.endpoint, bytes.NewReader(body)) + if err != nil { + return Outcome{State: model.Failed, Code: "invalid_provider_config"} + } + req.Header.Set("Authorization", "Bearer "+c.key) + req.Header.Set("Content-Type", "application/json") + if key != "" { + req.Header.Set("Idempotency-Key", key) + } + resp, err := c.http.Do(req) + if err != nil { + return Outcome{State: model.Unknown, Code: "provider_outcome_unknown"} + } + defer resp.Body.Close() + // Discard a bounded response without logging provider-controlled content. + _, _ = io.Copy(io.Discard, io.LimitReader(resp.Body, 64<<10)) + switch { + case resp.StatusCode >= 200 && resp.StatusCode < 300: + return Outcome{State: model.Accepted} + case resp.StatusCode == 429: + delay := time.Minute + if n, e := strconv.Atoi(resp.Header.Get("Retry-After")); e == nil && n > 0 && n <= 86400 { + delay = time.Duration(n) * time.Second + } else if date, e := http.ParseTime(resp.Header.Get("Retry-After")); e == nil && time.Until(date) > 0 { + delay = time.Until(date) + } + return Outcome{State: model.RetryWait, Code: "provider_rate_limited", RetryAfter: delay} + case resp.StatusCode >= 500 || resp.StatusCode == 408: + return Outcome{State: model.Unknown, Code: "provider_outcome_unknown"} + case resp.StatusCode == 401 || resp.StatusCode == 403: + return Outcome{State: model.Failed, Code: "provider_authentication_failed"} + default: + return Outcome{State: model.Failed, Code: "provider_rejected"} + } +} diff --git a/backend/notifications/email/resend.go b/backend/notifications/email/resend.go new file mode 100644 index 0000000..808ef86 --- /dev/null +++ b/backend/notifications/email/resend.go @@ -0,0 +1,22 @@ +package email + +import ( + "context" + "net/mail" + + "git.sendico.io/tech/sxchnge/notifications/model" +) + +type resend struct{ client } +type resendRequest struct { + From string `json:"from"` + To []string `json:"to"` + Subject string `json:"subject"` + HTML string `json:"html,omitempty"` + Text string `json:"text,omitempty"` +} + +func (s *resend) Send(ctx context.Context, key string, email model.Email) Outcome { + from := (&mail.Address{Name: s.settings.SenderName, Address: s.settings.SenderAddress}).String() + return s.post(ctx, key, resendRequest{From: from, To: []string{email.Address}, Subject: email.Subject, HTML: email.HTML, Text: email.Text}) +} diff --git a/backend/notifications/email/sendgrid.go b/backend/notifications/email/sendgrid.go new file mode 100644 index 0000000..facb196 --- /dev/null +++ b/backend/notifications/email/sendgrid.go @@ -0,0 +1,28 @@ +package email + +// Adapted from Sendico's SendGrid notifier (see PROVENANCE.md). The SDK mail +// construction is retained; templates, retries and logging belong to our service. +import ( + "context" + + "git.sendico.io/tech/sxchnge/notifications/model" + "github.com/sendgrid/sendgrid-go/helpers/mail" +) + +type sendgrid struct{ client } + +func (s *sendgrid) Send(ctx context.Context, _ string, email model.Email) Outcome { + m := mail.NewV3Mail() + m.SetFrom(mail.NewEmail(s.settings.SenderName, s.settings.SenderAddress)) + m.Subject = email.Subject + p := mail.NewPersonalization() + p.AddTos(mail.NewEmail("", email.Address)) + m.AddPersonalizations(p) + if email.Text != "" { + m.AddContent(mail.NewContent("text/plain", email.Text)) + } + if email.HTML != "" { + m.AddContent(mail.NewContent("text/html", email.HTML)) + } + return s.post(ctx, "", m) +} diff --git a/backend/notifications/go.mod b/backend/notifications/go.mod new file mode 100644 index 0000000..05d9f84 --- /dev/null +++ b/backend/notifications/go.mod @@ -0,0 +1,84 @@ +module git.sendico.io/tech/sxchnge/notifications + +go 1.27.0 + +require ( + dario.cat/mergo v1.0.2 // indirect + github.com/Azure/go-ansiterm v0.0.0-20250102033503-faa5f7b0171c // indirect + github.com/Microsoft/go-winio v0.6.2 // indirect + github.com/cenkalti/backoff/v4 v4.3.0 // indirect + github.com/cespare/xxhash/v2 v2.3.0 // indirect + github.com/containerd/errdefs v1.0.0 // indirect + github.com/containerd/errdefs/pkg v0.3.0 // indirect + github.com/containerd/log v0.1.0 // indirect + github.com/containerd/platforms v0.2.1 // indirect + github.com/cpuguy83/dockercfg v0.3.2 // indirect + github.com/distribution/reference v0.6.0 // indirect + github.com/docker/go-connections v0.7.0 // indirect + github.com/docker/go-units v0.5.0 // indirect + github.com/ebitengine/purego v0.10.1 // indirect + github.com/felixge/httpsnoop v1.1.0 // indirect + github.com/go-logr/logr v1.4.3 // indirect + github.com/go-logr/stdr v1.2.2 // indirect + github.com/go-ole/go-ole v1.3.0 // indirect + github.com/google/uuid v1.6.0 // indirect + github.com/klauspost/compress v1.19.2 // indirect + github.com/lufia/plan9stats v0.0.0-20260330125221-c963978e514e // indirect + github.com/magiconair/properties v1.8.10 // indirect + github.com/moby/docker-image-spec v1.3.1 // indirect + github.com/moby/go-archive v0.2.0 // indirect + github.com/moby/moby/api v1.55.0 // indirect + github.com/moby/moby/client v0.5.0 // indirect + github.com/moby/patternmatcher v0.6.1 // indirect + github.com/moby/sys/sequential v0.7.0 // indirect + github.com/moby/sys/user v0.4.0 // indirect + github.com/moby/sys/userns v0.1.0 // indirect + github.com/moby/term v0.5.2 // indirect + github.com/nats-io/nats.go v1.53.1 // indirect + github.com/nats-io/nkeys v0.4.15 // indirect + github.com/nats-io/nuid v1.0.1 // indirect + github.com/opencontainers/go-digest v1.0.0 // indirect + github.com/opencontainers/image-spec v1.1.1 // indirect + github.com/power-devops/perfstat v0.0.0-20240221224432-82ca36839d55 // indirect + github.com/shirou/gopsutil/v4 v4.26.6 // indirect + github.com/sirupsen/logrus v1.9.4 // indirect + github.com/stretchr/testify v1.12.1 // indirect + github.com/tklauser/go-sysconf v0.4.0 // indirect + github.com/tklauser/numcpus v0.12.0 // indirect + github.com/xdg-go/pbkdf2 v1.0.0 // indirect + github.com/xdg-go/scram v1.2.0 // indirect + github.com/xdg-go/stringprep v1.0.4 // indirect + github.com/youmark/pkcs8 v0.0.0-20240726163527-a2c0da244d78 // indirect + github.com/yusufpapurcu/wmi v1.2.4 // indirect + go.opentelemetry.io/auto/sdk v1.2.1 // indirect + go.opentelemetry.io/contrib/instrumentation/net/http/otelhttp v0.69.0 // indirect + go.opentelemetry.io/otel v1.44.0 // indirect + go.opentelemetry.io/otel/metric v1.44.0 // indirect + go.opentelemetry.io/otel/trace v1.44.0 // indirect + go.uber.org/multierr v1.10.0 // indirect + go.yaml.in/yaml/v3 v3.0.5 // indirect + golang.org/x/crypto v0.56.0 // indirect + golang.org/x/net v0.58.0 // indirect + golang.org/x/sync v0.22.0 // indirect + golang.org/x/sys v0.47.0 // indirect + golang.org/x/text v0.41.0 // indirect + google.golang.org/genproto/googleapis/rpc v0.0.0-20260526163538-3dc84a4a5aaa // indirect +) + +replace git.sendico.io/tech/sxchnge/shared => ../shared + +require ( + git.sendico.io/tech/sxchnge/notifications/interface v0.0.0 + git.sendico.io/tech/sxchnge/shared v0.0.0 + github.com/go-viper/mapstructure/v2 v2.5.0 + github.com/sendgrid/sendgrid-go v3.16.1+incompatible + github.com/testcontainers/testcontainers-go v0.44.0 + github.com/testcontainers/testcontainers-go/modules/mongodb v0.44.0 + go.mongodb.org/mongo-driver/v2 v2.9.0 + go.uber.org/zap v1.28.0 + google.golang.org/grpc v1.83.2 + google.golang.org/protobuf v1.36.12 + gopkg.in/yaml.v3 v3.0.1 +) + +replace git.sendico.io/tech/sxchnge/notifications/interface => ./interface diff --git a/backend/notifications/go.sum b/backend/notifications/go.sum new file mode 100644 index 0000000..fb3bc71 --- /dev/null +++ b/backend/notifications/go.sum @@ -0,0 +1,203 @@ +dario.cat/mergo v1.0.2 h1:85+piFYR1tMbRrLcDwR18y4UKJ3aH1Tbzi24VRW1TK8= +dario.cat/mergo v1.0.2/go.mod h1:E/hbnu0NxMFBjpMIE34DRGLWqDy0g5FuKDhCb31ngxA= +github.com/AdaLogics/go-fuzz-headers v0.0.0-20240806141605-e8a1dd7889d6 h1:He8afgbRMd7mFxO99hRNu+6tazq8nFF9lIwo9JFroBk= +github.com/AdaLogics/go-fuzz-headers v0.0.0-20240806141605-e8a1dd7889d6/go.mod h1:8o94RPi1/7XTJvwPpRSzSUedZrtlirdB3r9Z20bi2f8= +github.com/Azure/go-ansiterm v0.0.0-20250102033503-faa5f7b0171c h1:udKWzYgxTojEKWjV8V+WSxDXJ4NFATAsZjh8iIbsQIg= +github.com/Azure/go-ansiterm v0.0.0-20250102033503-faa5f7b0171c/go.mod h1:xomTg63KZ2rFqZQzSB4Vz2SUXa1BpHTVz9L5PTmPC4E= +github.com/Microsoft/go-winio v0.6.2 h1:F2VQgta7ecxGYO8k3ZZz3RS8fVIXVxONVUPlNERoyfY= +github.com/Microsoft/go-winio v0.6.2/go.mod h1:yd8OoFMLzJbo9gZq8j5qaps8bJ9aShtEA8Ipt1oGCvU= +github.com/cenkalti/backoff/v4 v4.3.0 h1:MyRJ/UdXutAwSAT+s3wNd7MfTIcy71VQueUuFK343L8= +github.com/cenkalti/backoff/v4 v4.3.0/go.mod h1:Y3VNntkOUPxTVeUxJ/G5vcM//AlwfmyYozVcomhLiZE= +github.com/cespare/xxhash/v2 v2.3.0 h1:UL815xU9SqsFlibzuggzjXhog7bL6oX9BbNZnL2UFvs= +github.com/cespare/xxhash/v2 v2.3.0/go.mod h1:VGX0DQ3Q6kWi7AoAeZDth3/j3BFtOZR5XLFGgcrjCOs= +github.com/containerd/errdefs v1.0.0 h1:tg5yIfIlQIrxYtu9ajqY42W3lpS19XqdxRQeEwYG8PI= +github.com/containerd/errdefs v1.0.0/go.mod h1:+YBYIdtsnF4Iw6nWZhJcqGSg/dwvV7tyJ/kCkyJ2k+M= +github.com/containerd/errdefs/pkg v0.3.0 h1:9IKJ06FvyNlexW690DXuQNx2KA2cUJXx151Xdx3ZPPE= +github.com/containerd/errdefs/pkg v0.3.0/go.mod h1:NJw6s9HwNuRhnjJhM7pylWwMyAkmCQvQ4GpJHEqRLVk= +github.com/containerd/log v0.1.0 h1:TCJt7ioM2cr/tfR8GPbGf9/VRAX8D2B4PjzCpfX540I= +github.com/containerd/log v0.1.0/go.mod h1:VRRf09a7mHDIRezVKTRCrOq78v577GXq3bSa3EhrzVo= +github.com/containerd/platforms v0.2.1 h1:zvwtM3rz2YHPQsF2CHYM8+KtB5dvhISiXh5ZpSBQv6A= +github.com/containerd/platforms v0.2.1/go.mod h1:XHCb+2/hzowdiut9rkudds9bE5yJ7npe7dG/wG+uFPw= +github.com/cpuguy83/dockercfg v0.3.2 h1:DlJTyZGBDlXqUZ2Dk2Q3xHs/FtnooJJVaad2S9GKorA= +github.com/cpuguy83/dockercfg v0.3.2/go.mod h1:sugsbF4//dDlL/i+S+rtpIWp+5h0BHJHfjj5/jFyUJc= +github.com/creack/pty v1.1.24 h1:bJrF4RRfyJnbTJqzRLHzcGaZK1NeM5kTC9jGgovnR1s= +github.com/creack/pty v1.1.24/go.mod h1:08sCNb52WyoAwi2QDyzUCTgcvVFhUzewun7wtTfvcwE= +github.com/davecgh/go-spew v1.1.1 h1:vj9j/u1bqnvCEfJOwUhtlOARqs3+rkHYY13jYWTU97c= +github.com/davecgh/go-spew v1.1.1/go.mod h1:J7Y8YcW2NihsgmVo/mv3lAwl/skON4iLHjSsI+c5H38= +github.com/distribution/reference v0.6.0 h1:0IXCQ5g4/QMHHkarYzh5l+u8T3t73zM5QvfrDyIgxBk= +github.com/distribution/reference v0.6.0/go.mod h1:BbU0aIcezP1/5jX/8MP0YiH4SdvB5Y4f/wlDRiLyi3E= +github.com/docker/go-connections v0.7.0 h1:6SsRfJddP22WMrCkj19x9WKjEDTB+ahsdiGYf0mN39c= +github.com/docker/go-connections v0.7.0/go.mod h1:no1qkHdjq7kLMGUXYAduOhYPSJxxvgWBh7ogVvptn3Q= +github.com/docker/go-units v0.5.0 h1:69rxXcBk27SvSaaxTtLh/8llcHD8vYHT7WSdRZ/jvr4= +github.com/docker/go-units v0.5.0/go.mod h1:fgPhTUdO+D/Jk86RDLlptpiXQzgHJF7gydDDbaIK4Dk= +github.com/ebitengine/purego v0.10.1 h1:dewVBCBT2GaMu1SrNTYxQhgQBethzfhiwvZiLGP/qyY= +github.com/ebitengine/purego v0.10.1/go.mod h1:iIjxzd6CiRiOG0UyXP+V1+jWqUXVjPKLAI0mRfJZTmQ= +github.com/felixge/httpsnoop v1.1.0 h1:3YtUj32ZZkqZtt3sZZsClsymw/QDuVfpNhoA31zeORc= +github.com/felixge/httpsnoop v1.1.0/go.mod h1:Zqxgdd+1Rkcz8euOqdr7lqgCRJztwr5hp9vDSi5UZCE= +github.com/go-logr/logr v1.2.2/go.mod h1:jdQByPbusPIv2/zmleS9BjJVeZ6kBagPoEUsqbVz/1A= +github.com/go-logr/logr v1.4.3 h1:CjnDlHq8ikf6E492q6eKboGOC0T8CDaOvkHCIg8idEI= +github.com/go-logr/logr v1.4.3/go.mod h1:9T104GzyrTigFIr8wt5mBrctHMim0Nb2HLGrmQ40KvY= +github.com/go-logr/stdr v1.2.2 h1:hSWxHoqTgW2S2qGc0LTAI563KZ5YKYRhT3MFKZMbjag= +github.com/go-logr/stdr v1.2.2/go.mod h1:mMo/vtBO5dYbehREoey6XUKy/eSumjCCveDpRre4VKE= +github.com/go-ole/go-ole v1.2.6/go.mod h1:pprOEPIfldk/42T2oK7lQ4v4JSDwmV0As9GaiUsvbm0= +github.com/go-ole/go-ole v1.3.0 h1:Dt6ye7+vXGIKZ7Xtk4s6/xVdGDQynvom7xCFEdWr6uE= +github.com/go-ole/go-ole v1.3.0/go.mod h1:5LS6F96DhAwUc7C+1HLexzMXY1xGRSryjyPPKW6zv78= +github.com/go-viper/mapstructure/v2 v2.5.0 h1:vM5IJoUAy3d7zRSVtIwQgBj7BiWtMPfmPEgAXnvj1Ro= +github.com/go-viper/mapstructure/v2 v2.5.0/go.mod h1:oJDH3BJKyqBA2TXFhDsKDGDTlndYOZ6rGS0BRZIxGhM= +github.com/golang/protobuf v1.5.4 h1:i7eJL8qZTpSEXOPTxNKhASYpMn+8e5Q6AdndVa1dWek= +github.com/golang/protobuf v1.5.4/go.mod h1:lnTiLA8Wa4RWRcIUkrtSVa5nRhsEGBg48fD6rSs7xps= +github.com/google/go-cmp v0.7.0 h1:wk8382ETsv4JYUZwIsn6YpYiWiBsYLSJiTsyBybVuN8= +github.com/google/go-cmp v0.7.0/go.mod h1:pXiqmnSA92OHEEa9HXL2W4E7lf9JzCmGVUdgjX3N/iU= +github.com/google/uuid v1.6.0 h1:NIvaJDMOsjHA8n1jAhLSgzrAzy1Hgr+hNrb57e+94F0= +github.com/google/uuid v1.6.0/go.mod h1:TIyPZe4MgqvfeYDBFedMoGGpEw/LqOeaOT+nhxU+yHo= +github.com/klauspost/compress v1.19.2 h1:hMRETovs/pu/dVWN7zIT1PGG8t509MwT6bO7XSi26R8= +github.com/klauspost/compress v1.19.2/go.mod h1:cwPg85FWrGar70rWktvGQj8/hthj3wpl0PGDogxkrSQ= +github.com/kr/pretty v0.3.1 h1:flRD4NNwYAUpkphVc1HcthR4KEIFJ65n8Mw5qdRn3LE= +github.com/kr/pretty v0.3.1/go.mod h1:hoEshYVHaxMs3cyo3Yncou5ZscifuDolrwPKZanG3xk= +github.com/kr/text v0.2.0 h1:5Nx0Ya0ZqY2ygV366QzturHI13Jq95ApcVaJBhpS+AY= +github.com/kr/text v0.2.0/go.mod h1:eLer722TekiGuMkidMxC/pM04lWEeraHUUmBw8l2grE= +github.com/lufia/plan9stats v0.0.0-20260330125221-c963978e514e h1:Q6MvJtQK/iRcRtzAscm/zF23XxJlbECiGPyRicsX+Ak= +github.com/lufia/plan9stats v0.0.0-20260330125221-c963978e514e/go.mod h1:autxFIvghDt3jPTLoqZ9OZ7s9qTGNAWmYCjVFWPX/zg= +github.com/magiconair/properties v1.8.10 h1:s31yESBquKXCV9a/ScB3ESkOjUYYv+X0rg8SYxI99mE= +github.com/magiconair/properties v1.8.10/go.mod h1:Dhd985XPs7jluiymwWYZ0G4Z61jb3vdS329zhj2hYo0= +github.com/moby/docker-image-spec v1.3.1 h1:jMKff3w6PgbfSa69GfNg+zN/XLhfXJGnEx3Nl2EsFP0= +github.com/moby/docker-image-spec v1.3.1/go.mod h1:eKmb5VW8vQEh/BAr2yvVNvuiJuY6UIocYsFu/DxxRpo= +github.com/moby/go-archive v0.2.0 h1:zg5QDUM2mi0JIM9fdQZWC7U8+2ZfixfTYoHL7rWUcP8= +github.com/moby/go-archive v0.2.0/go.mod h1:mNeivT14o8xU+5q1YnNrkQVpK+dnNe/K6fHqnTg4qPU= +github.com/moby/moby/api v1.55.0 h1:2/sexvQyqIWS8pRSCFddBfpW2qE7vR7FCL+vN8pxwMc= +github.com/moby/moby/api v1.55.0/go.mod h1:+RQ6wluLwtYaTd1WnPLykIDPekkuyD/ROWQClE83pzs= +github.com/moby/moby/client v0.5.0 h1:5XhyPk2fuOWf6RlSFa3MkIIgDZkF25xToXW8Q/BH7cc= +github.com/moby/moby/client v0.5.0/go.mod h1:rcVpF8ncl9vo5gaIBdol6CnbEtSj1uxMvEV/UrykF/s= +github.com/moby/patternmatcher v0.6.1 h1:qlhtafmr6kgMIJjKJMDmMWq7WLkKIo23hsrpR3x084U= +github.com/moby/patternmatcher v0.6.1/go.mod h1:hDPoyOpDY7OrrMDLaYoY3hf52gNCR/YOUYxkhApJIxc= +github.com/moby/sys/sequential v0.7.0 h1:ASQNGNROJSuOO6LL6bPHbKvuZu6NU8P4ldPWk31zj/8= +github.com/moby/sys/sequential v0.7.0/go.mod h1:NfSTAp6V3fw4tmkD62PEcOKeZKquXT8VKCkf7aVR79o= +github.com/moby/sys/user v0.4.0 h1:jhcMKit7SA80hivmFJcbB1vqmw//wU61Zdui2eQXuMs= +github.com/moby/sys/user v0.4.0/go.mod h1:bG+tYYYJgaMtRKgEmuueC0hJEAZWwtIbZTB+85uoHjs= +github.com/moby/sys/userns v0.1.0 h1:tVLXkFOxVu9A64/yh59slHVv9ahO9UIev4JZusOLG/g= +github.com/moby/sys/userns v0.1.0/go.mod h1:IHUYgu/kao6N8YZlp9Cf444ySSvCmDlmzUcYfDHOl28= +github.com/moby/term v0.5.2 h1:6qk3FJAFDs6i/q3W/pQ97SX192qKfZgGjCQqfCJkgzQ= +github.com/moby/term v0.5.2/go.mod h1:d3djjFCrjnB+fl8NJux+EJzu0msscUP+f8it8hPkFLc= +github.com/nats-io/nats.go v1.53.1 h1:Otsq3uLc/kLdjmkNHkXH0jBqwUquwdKFoe3fq6/3/Xo= +github.com/nats-io/nats.go v1.53.1/go.mod h1:26HypzazeOkyO3/mqd1zZd53STJN0EjCYF9Uy2ZOBno= +github.com/nats-io/nkeys v0.4.15 h1:JACV5jRVO9V856KOapQ7x+EY8Jo3qw1vJt/9Jpwzkk4= +github.com/nats-io/nkeys v0.4.15/go.mod h1:CpMchTXC9fxA5zrMo4KpySxNjiDVvr8ANOSZdiNfUrs= +github.com/nats-io/nuid v1.0.1 h1:5iA8DT8V7q8WK2EScv2padNa/rTESc1KdnPw4TC2paw= +github.com/nats-io/nuid v1.0.1/go.mod h1:19wcPz3Ph3q0Jbyiqsd0kePYG7A95tJPxeL+1OSON2c= +github.com/opencontainers/go-digest v1.0.0 h1:apOUWs51W5PlhuyGyz9FCeeBIOUDA/6nW8Oi/yOhh5U= +github.com/opencontainers/go-digest v1.0.0/go.mod h1:0JzlMkj0TRzQZfJkVvzbP0HBR3IKzErnv2BNG4W4MAM= +github.com/opencontainers/image-spec v1.1.1 h1:y0fUlFfIZhPF1W537XOLg0/fcx6zcHCJwooC2xJA040= +github.com/opencontainers/image-spec v1.1.1/go.mod h1:qpqAh3Dmcf36wStyyWU+kCeDgrGnAve2nCC8+7h8Q0M= +github.com/power-devops/perfstat v0.0.0-20240221224432-82ca36839d55 h1:o4JXh1EVt9k/+g42oCprj/FisM4qX9L3sZB3upGN2ZU= +github.com/power-devops/perfstat v0.0.0-20240221224432-82ca36839d55/go.mod h1:OmDBASR4679mdNQnz2pUhc2G8CO2JrUAVFDRBDP/hJE= +github.com/rogpeppe/go-internal v1.14.1 h1:UQB4HGPB6osV0SQTLymcB4TgvyWu6ZyliaW0tI/otEQ= +github.com/rogpeppe/go-internal v1.14.1/go.mod h1:MaRKkUm5W0goXpeCfT7UZI6fk/L7L7so1lCWt35ZSgc= +github.com/sendgrid/sendgrid-go v3.16.1+incompatible h1:zWhTmB0Y8XCDzeWIm2/BIt1GjJohAA0p6hVEaDtHWWs= +github.com/sendgrid/sendgrid-go v3.16.1+incompatible/go.mod h1:QRQt+LX/NmgVEvmdRw0VT/QgUn499+iza2FnDca9fg8= +github.com/shirou/gopsutil/v4 v4.26.6 h1:Mzr/npDtQC/xpeEuQKHZt8Zo9CmPvhTj8nkR8w5TLDs= +github.com/shirou/gopsutil/v4 v4.26.6/go.mod h1:LZ6ewCSkBqUpvSOf+LsTGnRinC6iaNUNMGBtDkJBaLQ= +github.com/sirupsen/logrus v1.9.4 h1:TsZE7l11zFCLZnZ+teH4Umoq5BhEIfIzfRDZ1Uzql2w= +github.com/sirupsen/logrus v1.9.4/go.mod h1:ftWc9WdOfJ0a92nsE2jF5u5ZwH8Bv2zdeOC42RjbV2g= +github.com/stretchr/objx v0.5.3 h1:jmXUvGomnU1o3W/V5h2VEradbpJDwGrzugQQvL0POH4= +github.com/stretchr/objx v0.5.3/go.mod h1:rDQraq+vQZU7Fde9LOZLr8Tax6zZvy4kuNKF+QYS+U0= +github.com/stretchr/testify v1.12.1 h1:EuwCh5fleGS7H32xRwO3wRGT7DxrDhLAT6FF8MpWDWE= +github.com/stretchr/testify v1.12.1/go.mod h1:MDEgiDPPsNp5cuIrHPPCyornHKgEVbtFUmoNlxoYthg= +github.com/testcontainers/testcontainers-go v0.44.0 h1:/Fwh6HY1mIikhnm9e7HwoxGycx0lzRAE0f5VQpjFxzI= +github.com/testcontainers/testcontainers-go v0.44.0/go.mod h1:IcnwQrYTO86xHXu5bvMaBH7ATlbS3Qn1M1QWW3c66rE= +github.com/testcontainers/testcontainers-go/modules/mongodb v0.44.0 h1:VSPDFiumAtt0CkZEVbmAkEmYVRvsJpKJy9oF3exRKYg= +github.com/testcontainers/testcontainers-go/modules/mongodb v0.44.0/go.mod h1:kHfzrY1cYP/zr9H4TdqAxbP836A1C2fyUojlHidhFGI= +github.com/tklauser/go-sysconf v0.4.0 h1:7H0uAN+7RkwWRaxhYXDLqa5V3LPrJeV8wmD9dRUgPQU= +github.com/tklauser/go-sysconf v0.4.0/go.mod h1:8mTNWyog7H+MpKijp4VmKJAd2bbYQ2zuUwkYRbUArPI= +github.com/tklauser/numcpus v0.12.0 h1:NR85qdvHA9pFse3x3weVZ0r0ST8R6l5RHbZrlRaqob4= +github.com/tklauser/numcpus v0.12.0/go.mod h1:ABHeXzJnr/qqwguhClkZKT1/8VABcYrsyUiUGobwWJg= +github.com/xdg-go/pbkdf2 v1.0.0 h1:Su7DPu48wXMwC3bs7MCNG+z4FhcyEuz5dlvchbq0B0c= +github.com/xdg-go/pbkdf2 v1.0.0/go.mod h1:jrpuAogTd400dnrH08LKmI/xc1MbPOebTwRqcT5RDeI= +github.com/xdg-go/scram v1.2.0 h1:bYKF2AEwG5rqd1BumT4gAnvwU/M9nBp2pTSxeZw7Wvs= +github.com/xdg-go/scram v1.2.0/go.mod h1:3dlrS0iBaWKYVt2ZfA4cj48umJZ+cAEbR6/SjLA88I8= +github.com/xdg-go/stringprep v1.0.4 h1:XLI/Ng3O1Atzq0oBs3TWm+5ZVgkq2aqdlvP9JtoZ6c8= +github.com/xdg-go/stringprep v1.0.4/go.mod h1:mPGuuIYwz7CmR2bT9j4GbQqutWS1zV24gijq1dTyGkM= +github.com/youmark/pkcs8 v0.0.0-20240726163527-a2c0da244d78 h1:ilQV1hzziu+LLM3zUTJ0trRztfwgjqKnBWNtSRkbmwM= +github.com/youmark/pkcs8 v0.0.0-20240726163527-a2c0da244d78/go.mod h1:aL8wCCfTfSfmXjznFBSZNN13rSJjlIOI1fUNAtF7rmI= +github.com/yuin/goldmark v1.4.13/go.mod h1:6yULJ656Px+3vBD8DxQVa3kxgyrAnzto9xy5taEt/CY= +github.com/yusufpapurcu/wmi v1.2.4 h1:zFUKzehAFReQwLys1b/iSMl+JQGSCSjtVqQn9bBrPo0= +github.com/yusufpapurcu/wmi v1.2.4/go.mod h1:SBZ9tNy3G9/m5Oi98Zks0QjeHVDvuK0qfxQmPyzfmi0= +go.mongodb.org/mongo-driver/v2 v2.9.0 h1:e2mQdOmbkiYz+dj3faM7lVDwl7WdnRD+g5VicafMhL0= +go.mongodb.org/mongo-driver/v2 v2.9.0/go.mod h1:SHKN0IWkKmEVGHLjXnni6s4wPKX4v86FTgOeJJFuXcA= +go.opentelemetry.io/auto/sdk v1.2.1 h1:jXsnJ4Lmnqd11kwkBV2LgLoFMZKizbCi5fNZ/ipaZ64= +go.opentelemetry.io/auto/sdk v1.2.1/go.mod h1:KRTj+aOaElaLi+wW1kO/DZRXwkF4C5xPbEe3ZiIhN7Y= +go.opentelemetry.io/contrib/instrumentation/net/http/otelhttp v0.69.0 h1:8tvICD4vSTOOsNrsI4Ljf6C+6UKvpTEH5XY3JMoyPoo= +go.opentelemetry.io/contrib/instrumentation/net/http/otelhttp v0.69.0/go.mod h1:z9+yiacE0IHRqM4qFfkbt/JYlmYXgss8GY/jXoNuPJI= +go.opentelemetry.io/otel v1.44.0 h1:JjwHmHpA4iZ3wBxluu2fbbE7j4kqlE8jXyAyPXH7HqU= +go.opentelemetry.io/otel v1.44.0/go.mod h1:BMgjTHL9WPRlRjL2oZCBTL4whCGtXch2H4BhOPIAyYc= +go.opentelemetry.io/otel/metric v1.44.0 h1:1w0gILTcHdr3YI+ixLyjemwrVnsMURbTZFrSYCdDdmc= +go.opentelemetry.io/otel/metric v1.44.0/go.mod h1:8O7hanEPBNgEMmybD3s2VBKcgWOCsA6tzHBPODAiquo= +go.opentelemetry.io/otel/sdk v1.44.0 h1:nHYwb9lK+fJPU/dnT6s7W7Z8itMWyqrnVfbheVYrZ58= +go.opentelemetry.io/otel/sdk v1.44.0/go.mod h1:Osuydd3Se74nqjAKxid74N5eC+jfEqfTegHRnq58oK0= +go.opentelemetry.io/otel/sdk/metric v1.44.0 h1:3LlKgI+VjbVsjNRFZJZAJ30WjXC5VkNRks6si09iEfI= +go.opentelemetry.io/otel/sdk/metric v1.44.0/go.mod h1:5B5pMARnXxKhltooO4xUuCBorl65a4EpnTalObqOigA= +go.opentelemetry.io/otel/trace v1.44.0 h1:jxF5CsGYCe74MCRx2X4g7WsY/VBKRqqpNvXlX/6gtIk= +go.opentelemetry.io/otel/trace v1.44.0/go.mod h1:oLl1jrMQAVo6v3GAggN+1VH9VIz9iUSvW53sW1Q8PIE= +go.uber.org/goleak v1.3.0 h1:2K3zAYmnTNqV73imy9J1T3WC+gmCePx2hEGkimedGto= +go.uber.org/goleak v1.3.0/go.mod h1:CoHD4mav9JJNrW/WLlf7HGZPjdw8EucARQHekz1X6bE= +go.uber.org/multierr v1.10.0 h1:S0h4aNzvfcFsC3dRF1jLoaov7oRaKqRGC/pUEJ2yvPQ= +go.uber.org/multierr v1.10.0/go.mod h1:20+QtiLqy0Nd6FdQB9TLXag12DsQkrbs3htMFfDN80Y= +go.uber.org/zap v1.28.0 h1:IZzaP1Fv73/T/pBMLk4VutPl36uNC+OSUh3JLG3FIjo= +go.uber.org/zap v1.28.0/go.mod h1:rDLpOi171uODNm/mxFcuYWxDsqWSAVkFdX4XojSKg/Q= +go.yaml.in/yaml/v3 v3.0.5 h1:N6y/pJk8buWs9NY5ERU2HSMfm+IuD/OtfdAnq6kESPw= +go.yaml.in/yaml/v3 v3.0.5/go.mod h1:HVTZu1O7/Vkt2N+BFy8Zza+lnLsABggaTM2ZpNIGuKg= +golang.org/x/crypto v0.0.0-20190308221718-c2843e01d9a2/go.mod h1:djNgcEr1/C05ACkg1iLfiJU5Ep61QUkGW8qpdssI0+w= +golang.org/x/crypto v0.0.0-20210921155107-089bfa567519/go.mod h1:GvvjBRRGRdwPK5ydBHafDWAxML/pGHZbMvKqRZ5+Abc= +golang.org/x/crypto v0.56.0 h1:GUh5Ii4J5jtcseSMiRqr1jXCNHoxjeV9Fmekc2oLy6Y= +golang.org/x/crypto v0.56.0/go.mod h1:OMW5y6CY9l38uPLmxU6l6pwcXp1obtLo3e6gT7gQR2I= +golang.org/x/mod v0.6.0-dev.0.20220419223038-86c51ed26bb4/go.mod h1:jJ57K6gSWd91VN4djpZkiMVwK6gcyfeH4XE8wZrZaV4= +golang.org/x/net v0.0.0-20190620200207-3b0461eec859/go.mod h1:z5CRVTTTmAJ677TzLLGU+0bjPO0LkuOLi4/5GtJWs/s= +golang.org/x/net v0.0.0-20210226172049-e18ecbb05110/go.mod h1:m0MpNAwzfU5UDzcl9v0D8zg8gWTRqZa9RBIspLL5mdg= +golang.org/x/net v0.0.0-20220722155237-a158d28d115b/go.mod h1:XRhObCWvk6IyKnWLug+ECip1KBveYUHfp+8e9klMJ9c= +golang.org/x/net v0.58.0 h1:ynWG7rqYi4ccpTEuPZ2QGWHktVEM9DMCj9yzDE0Q7To= +golang.org/x/net v0.58.0/go.mod h1:YwCddHnFlT7eLQqVprV19OnhLGtc5xOKgE0RyqgfWAU= +golang.org/x/sync v0.0.0-20190423024810-112230192c58/go.mod h1:RxMgew5VJxzue5/jJTE5uejpjVlOe/izrB70Jof72aM= +golang.org/x/sync v0.0.0-20220722155255-886fb9371eb4/go.mod h1:RxMgew5VJxzue5/jJTE5uejpjVlOe/izrB70Jof72aM= +golang.org/x/sync v0.22.0 h1:SZjpbeLmrCk4xhRSZFNZW5gFUeCeFgjekvI/+gfScek= +golang.org/x/sync v0.22.0/go.mod h1:9xrNwdLfx4jkKbNva9FpL6vEN7evnE43NNNJQ2LF3+0= +golang.org/x/sys v0.0.0-20190215142949-d0b11bdaac8a/go.mod h1:STP8DvDyc/dI5b8T5hshtkjS+E42TnysNCUPdjciGhY= +golang.org/x/sys v0.0.0-20190916202348-b4ddaad3f8a3/go.mod h1:h1NjWce9XRLGQEsW7wpKNCjG9DtNlClVuFLEZdDNbEs= +golang.org/x/sys v0.0.0-20201119102817-f84b799fce68/go.mod h1:h1NjWce9XRLGQEsW7wpKNCjG9DtNlClVuFLEZdDNbEs= +golang.org/x/sys v0.0.0-20201204225414-ed752295db88/go.mod h1:h1NjWce9XRLGQEsW7wpKNCjG9DtNlClVuFLEZdDNbEs= +golang.org/x/sys v0.0.0-20210615035016-665e8c7367d1/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg= +golang.org/x/sys v0.0.0-20210616094352-59db8d763f22/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg= +golang.org/x/sys v0.0.0-20220520151302-bc2c85ada10a/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg= +golang.org/x/sys v0.0.0-20220722155257-8c9f86f7a55f/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg= +golang.org/x/sys v0.1.0/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg= +golang.org/x/sys v0.47.0 h1:o7XGOvZQCADBQQ4Y7VNq2dRWQR7JmOUW8Kxx4ZsNgWs= +golang.org/x/sys v0.47.0/go.mod h1:4GL1E5IUh+htKOUEOaiffhrAeqysfVGipDYzABqnCmw= +golang.org/x/term v0.0.0-20201126162022-7de9c90e9dd1/go.mod h1:bj7SfCRtBDWHUb9snDiAeCFNEtKQo2Wmx5Cou7ajbmo= +golang.org/x/term v0.0.0-20210927222741-03fcf44c2211/go.mod h1:jbD1KX2456YbFQfuXm/mYQcufACuNUgVhRMnK/tPxf8= +golang.org/x/term v0.45.0 h1:NwWyBmoJCbfTHpxrWoZ9C6/VxOf7ic219I8xZZFdrf0= +golang.org/x/term v0.45.0/go.mod h1:9aqxs0blBcrm/n0L9QW0aRVD+ktan8ssZromtqJC43w= +golang.org/x/text v0.3.0/go.mod h1:NqM8EUOU14njkJ3fqMW+pc6Ldnwhi/IjpwHt7yyuwOQ= +golang.org/x/text v0.3.3/go.mod h1:5Zoc/QRtKVWzQhOtBMvqHzDpF6irO9z98xDceosuGiQ= +golang.org/x/text v0.3.7/go.mod h1:u+2+/6zg+i71rQMx5EYifcz6MCKuco9NR6JIITiCfzQ= +golang.org/x/text v0.3.8/go.mod h1:E6s5w1FMmriuDzIBO73fBruAKo1PCIq6d2Q6DHfQ8WQ= +golang.org/x/text v0.41.0 h1:vz/seA0lnX87Othu2f/0L24RcgrXD9/YFTSuGjj3rH8= +golang.org/x/text v0.41.0/go.mod h1:jvf1O8ajNzZqhSrQBPbutR/EB83Cc0CFrezNQIwbb5M= +golang.org/x/tools v0.0.0-20180917221912-90fa682c2a6e/go.mod h1:n7NCudcB/nEzxVGmLbDWY5pfWTLqBcC2KZ6jyYvM4mQ= +golang.org/x/tools v0.0.0-20191119224855-298f0cb1881e/go.mod h1:b+2E5dAYhXwXZwtnZ6UAqBI28+e2cm9otk0dWdXHAEo= +golang.org/x/tools v0.1.12/go.mod h1:hNGJHUnrk76NpqgfD5Aqm5Crs+Hm0VOH/i9J2+nxYbc= +golang.org/x/xerrors v0.0.0-20190717185122-a985d3407aa7/go.mod h1:I/5z698sn9Ka8TeJc9MKroUUfqBBauWjQqLJ2OPfmY0= +gonum.org/v1/gonum v0.17.0 h1:VbpOemQlsSMrYmn7T2OUvQ4dqxQXU+ouZFQsZOx50z4= +gonum.org/v1/gonum v0.17.0/go.mod h1:El3tOrEuMpv2UdMrbNlKEh9vd86bmQ6vqIcDwxEOc1E= +google.golang.org/genproto/googleapis/rpc v0.0.0-20260526163538-3dc84a4a5aaa h1:mZHHdPZl0dbGHCflZgAq/Q468DWVFcU2whhB2KAo8fk= +google.golang.org/genproto/googleapis/rpc v0.0.0-20260526163538-3dc84a4a5aaa/go.mod h1:4Hqkh8ycfw05ld/3BWL7rJOSfebL2Q+DVDeRgYgxUU8= +google.golang.org/grpc v1.83.2 h1:EManeRomTObA0BU7I8vXgg/78uE5MJ9M8B39EX2WscU= +google.golang.org/grpc v1.83.2/go.mod h1:YPI1hK3kDked6iHvgX3tR0y+nX/qpMFKhPgFsokw1S8= +google.golang.org/protobuf v1.36.12 h1:pJOKDDOyeXErUroCihFAd5LQuwXBSpVnKGrj5o/fwxc= +google.golang.org/protobuf v1.36.12/go.mod h1:HTf+CrKn2C3g5S8VImy6tdcUvCska2kB7j23XfzDpco= +gopkg.in/check.v1 v0.0.0-20161208181325-20d25e280405/go.mod h1:Co6ibVJAznAaIkqp8huTwlJQCZ016jof/cbN4VW5Yz0= +gopkg.in/check.v1 v1.0.0-20201130134442-10cb98267c6c h1:Hei/4ADfdWqJk1ZMxUNpqntNwaWcugrBjAiHlqqRiVk= +gopkg.in/check.v1 v1.0.0-20201130134442-10cb98267c6c/go.mod h1:JHkPIbrfpd72SG/EVd6muEfDQjcINNoR0C8j2r3qZ4Q= +gopkg.in/yaml.v3 v3.0.1 h1:fxVm/GzAzEWqLHuvctI91KS9hhNmmWOoWu0XTYJS7CA= +gopkg.in/yaml.v3 v3.0.1/go.mod h1:K4uyk7z7BCEPqu6E+C64Yfv1cQ7kz7rIZviUmN+EgEM= +gotest.tools/v3 v3.5.2 h1:7koQfIKdy+I8UTetycgUqXWSDwpgv193Ka+qRsmBY8Q= +gotest.tools/v3 v3.5.2/go.mod h1:LtdLGcnqToBH83WByAAi/wiwSFCArdFIUV/xxN4pcjA= +pgregory.net/rapid v1.2.0 h1:keKAYRcjm+e1F0oAuU5F5+YPAWcyxNNRK2wud503Gnk= +pgregory.net/rapid v1.2.0/go.mod h1:PY5XlDGj0+V1FCq0o192FdRhpKHGTRIWBgqjDBTrq04= diff --git a/backend/notifications/interface/README.md b/backend/notifications/interface/README.md new file mode 100644 index 0000000..52d2de9 --- /dev/null +++ b/backend/notifications/interface/README.md @@ -0,0 +1,41 @@ +# Notifications service contract + +This independent module owns the public internal-service contract. The canonical +source is `proto/v1/interface.proto`; it defines `Send`, `Get`, requests, statuses +and email/SMS/push/Telegram targets. The executable currently implements email. + +From the repository root, `make generate-notifications` produces Go/gRPC client +bindings in `notifications/v1`. Generated `.pb.go` files are ignored and must not +be committed. Both backend validation targets generate bindings before compiling. +The generators are pinned in the Makefile; install `protoc` separately. + +`make package-notifications-interface` creates a distributable source archive and +a standalone Go client-module archive under `dist/`. Client consumers can generate +bindings for their own language from the source archive. The Go archive contains +the generated bindings and its own go.mod/go.sum, with no executable or shared +infrastructure dependency. Archive generation does not upload to a registry. + +Internal Go consumers use the module +`git.sendico.io/tech/sxchnge/notifications/interface` and import +`git.sendico.io/tech/sxchnge/notifications/interface/notifications/v1`. Within this +checkout, use a local replace to this directory and generate before building. + +Calls require TLS and `authorization: Bearer ` metadata. The service +maps tokens to configured caller identities. Callers authorize user/destination +ownership and preferences before sending; this contract is not a public user API. + +Parameters are string values keyed by template parameter name. The entire request +is validated before persistence. An identical idempotency key/request returns the +original resource, even after completion or expiry. A changed request conflicts. +Keys are scoped to the authenticated caller and retained at least seven days; +Mongo TTL deletion is asynchronous. Expiry defaults to 24 hours and cannot exceed +24 hours. One email target is accepted; other declared channels return UNIMPLEMENTED. + +ACCEPTED means provider acceptance, not delivery or reading. UNKNOWN means the +provider may have sent the message and automatic replay is unsafe. DELIVERED is +reserved for a future receipt implementation. Get never exposes recipient addresses, +rendered content, parameters, provider credentials or raw provider responses. + +Keep published template revisions and field numbers stable. Reserve removed +protobuf field/enum names and numbers. Compatible additions belong in v1; +incompatible changes require a new version. diff --git a/backend/notifications/interface/go.mod b/backend/notifications/interface/go.mod new file mode 100644 index 0000000..5b55da1 --- /dev/null +++ b/backend/notifications/interface/go.mod @@ -0,0 +1,15 @@ +module git.sendico.io/tech/sxchnge/notifications/interface + +go 1.27.0 + +require ( + google.golang.org/grpc v1.83.2 + google.golang.org/protobuf v1.36.12 +) + +require ( + golang.org/x/net v0.58.0 // indirect + golang.org/x/sys v0.47.0 // indirect + golang.org/x/text v0.41.0 // indirect + google.golang.org/genproto/googleapis/rpc v0.0.0-20260526163538-3dc84a4a5aaa // indirect +) diff --git a/backend/notifications/interface/go.sum b/backend/notifications/interface/go.sum new file mode 100644 index 0000000..1371e12 --- /dev/null +++ b/backend/notifications/interface/go.sum @@ -0,0 +1,38 @@ +github.com/cespare/xxhash/v2 v2.3.0 h1:UL815xU9SqsFlibzuggzjXhog7bL6oX9BbNZnL2UFvs= +github.com/cespare/xxhash/v2 v2.3.0/go.mod h1:VGX0DQ3Q6kWi7AoAeZDth3/j3BFtOZR5XLFGgcrjCOs= +github.com/go-logr/logr v1.4.3 h1:CjnDlHq8ikf6E492q6eKboGOC0T8CDaOvkHCIg8idEI= +github.com/go-logr/logr v1.4.3/go.mod h1:9T104GzyrTigFIr8wt5mBrctHMim0Nb2HLGrmQ40KvY= +github.com/go-logr/stdr v1.2.2 h1:hSWxHoqTgW2S2qGc0LTAI563KZ5YKYRhT3MFKZMbjag= +github.com/go-logr/stdr v1.2.2/go.mod h1:mMo/vtBO5dYbehREoey6XUKy/eSumjCCveDpRre4VKE= +github.com/golang/protobuf v1.5.4 h1:i7eJL8qZTpSEXOPTxNKhASYpMn+8e5Q6AdndVa1dWek= +github.com/golang/protobuf v1.5.4/go.mod h1:lnTiLA8Wa4RWRcIUkrtSVa5nRhsEGBg48fD6rSs7xps= +github.com/google/go-cmp v0.7.0 h1:wk8382ETsv4JYUZwIsn6YpYiWiBsYLSJiTsyBybVuN8= +github.com/google/go-cmp v0.7.0/go.mod h1:pXiqmnSA92OHEEa9HXL2W4E7lf9JzCmGVUdgjX3N/iU= +github.com/google/uuid v1.6.0 h1:NIvaJDMOsjHA8n1jAhLSgzrAzy1Hgr+hNrb57e+94F0= +github.com/google/uuid v1.6.0/go.mod h1:TIyPZe4MgqvfeYDBFedMoGGpEw/LqOeaOT+nhxU+yHo= +go.opentelemetry.io/auto/sdk v1.2.1 h1:jXsnJ4Lmnqd11kwkBV2LgLoFMZKizbCi5fNZ/ipaZ64= +go.opentelemetry.io/auto/sdk v1.2.1/go.mod h1:KRTj+aOaElaLi+wW1kO/DZRXwkF4C5xPbEe3ZiIhN7Y= +go.opentelemetry.io/otel v1.44.0 h1:JjwHmHpA4iZ3wBxluu2fbbE7j4kqlE8jXyAyPXH7HqU= +go.opentelemetry.io/otel v1.44.0/go.mod h1:BMgjTHL9WPRlRjL2oZCBTL4whCGtXch2H4BhOPIAyYc= +go.opentelemetry.io/otel/metric v1.44.0 h1:1w0gILTcHdr3YI+ixLyjemwrVnsMURbTZFrSYCdDdmc= +go.opentelemetry.io/otel/metric v1.44.0/go.mod h1:8O7hanEPBNgEMmybD3s2VBKcgWOCsA6tzHBPODAiquo= +go.opentelemetry.io/otel/sdk v1.44.0 h1:nHYwb9lK+fJPU/dnT6s7W7Z8itMWyqrnVfbheVYrZ58= +go.opentelemetry.io/otel/sdk v1.44.0/go.mod h1:Osuydd3Se74nqjAKxid74N5eC+jfEqfTegHRnq58oK0= +go.opentelemetry.io/otel/sdk/metric v1.44.0 h1:3LlKgI+VjbVsjNRFZJZAJ30WjXC5VkNRks6si09iEfI= +go.opentelemetry.io/otel/sdk/metric v1.44.0/go.mod h1:5B5pMARnXxKhltooO4xUuCBorl65a4EpnTalObqOigA= +go.opentelemetry.io/otel/trace v1.44.0 h1:jxF5CsGYCe74MCRx2X4g7WsY/VBKRqqpNvXlX/6gtIk= +go.opentelemetry.io/otel/trace v1.44.0/go.mod h1:oLl1jrMQAVo6v3GAggN+1VH9VIz9iUSvW53sW1Q8PIE= +golang.org/x/net v0.58.0 h1:ynWG7rqYi4ccpTEuPZ2QGWHktVEM9DMCj9yzDE0Q7To= +golang.org/x/net v0.58.0/go.mod h1:YwCddHnFlT7eLQqVprV19OnhLGtc5xOKgE0RyqgfWAU= +golang.org/x/sys v0.47.0 h1:o7XGOvZQCADBQQ4Y7VNq2dRWQR7JmOUW8Kxx4ZsNgWs= +golang.org/x/sys v0.47.0/go.mod h1:4GL1E5IUh+htKOUEOaiffhrAeqysfVGipDYzABqnCmw= +golang.org/x/text v0.41.0 h1:vz/seA0lnX87Othu2f/0L24RcgrXD9/YFTSuGjj3rH8= +golang.org/x/text v0.41.0/go.mod h1:jvf1O8ajNzZqhSrQBPbutR/EB83Cc0CFrezNQIwbb5M= +gonum.org/v1/gonum v0.17.0 h1:VbpOemQlsSMrYmn7T2OUvQ4dqxQXU+ouZFQsZOx50z4= +gonum.org/v1/gonum v0.17.0/go.mod h1:El3tOrEuMpv2UdMrbNlKEh9vd86bmQ6vqIcDwxEOc1E= +google.golang.org/genproto/googleapis/rpc v0.0.0-20260526163538-3dc84a4a5aaa h1:mZHHdPZl0dbGHCflZgAq/Q468DWVFcU2whhB2KAo8fk= +google.golang.org/genproto/googleapis/rpc v0.0.0-20260526163538-3dc84a4a5aaa/go.mod h1:4Hqkh8ycfw05ld/3BWL7rJOSfebL2Q+DVDeRgYgxUU8= +google.golang.org/grpc v1.83.2 h1:EManeRomTObA0BU7I8vXgg/78uE5MJ9M8B39EX2WscU= +google.golang.org/grpc v1.83.2/go.mod h1:YPI1hK3kDked6iHvgX3tR0y+nX/qpMFKhPgFsokw1S8= +google.golang.org/protobuf v1.36.12 h1:pJOKDDOyeXErUroCihFAd5LQuwXBSpVnKGrj5o/fwxc= +google.golang.org/protobuf v1.36.12/go.mod h1:HTf+CrKn2C3g5S8VImy6tdcUvCska2kB7j23XfzDpco= diff --git a/backend/notifications/interface/proto/v1/interface.proto b/backend/notifications/interface/proto/v1/interface.proto new file mode 100644 index 0000000..e8821de --- /dev/null +++ b/backend/notifications/interface/proto/v1/interface.proto @@ -0,0 +1,73 @@ +syntax = "proto3"; +package sxchnge.notifications.v1; +option go_package = "git.sendico.io/tech/sxchnge/notifications/interface/notifications/v1;notificationsv1"; +import "google/protobuf/timestamp.proto"; + +// Internal service: authenticate callers over TLS. Send durably queues work. +service NotificationsService { + rpc Send(SendRequest) returns (SendResponse); + rpc Get(GetRequest) returns (GetResponse); +} +message SendRequest { + string idempotency_key = 1; // Required; scoped to authenticated caller, retained seven days. + string user_id = 2; // Caller resolves and authorizes recipient ownership. + string template_id = 3; + string template_version = 4; + map parameters = 5; + string locale = 6; + repeated Target targets = 7; // V1 accepts exactly one email target. + google.protobuf.Timestamp expires_at = 8; // Default/max 24h from acceptance. +} +message Target { + oneof destination { + EmailTarget email = 1; + SmsTarget sms = 2; + PushTarget push = 3; + TelegramTarget telegram = 4; + } +} +message EmailTarget { string address = 1; } +message SmsTarget { string phone_number = 1; } +message PushTarget { string endpoint_id = 1; } +message TelegramTarget { string chat_id = 1; } +message SendResponse { Notification notification = 1; } +message GetRequest { string id = 1; } +message GetResponse { Notification notification = 1; } +message Notification { + string id = 1; + string user_id = 2; + string template_id = 3; + string template_version = 4; + repeated Delivery deliveries = 5; + google.protobuf.Timestamp created_at = 6; + google.protobuf.Timestamp updated_at = 7; + google.protobuf.Timestamp expires_at = 8; +} +message Delivery { + string id = 1; + Channel channel = 2; + DeliveryState state = 3; + uint32 attempt_count = 4; + Failure failure = 5; + google.protobuf.Timestamp next_attempt_at = 6; + google.protobuf.Timestamp updated_at = 7; +} +message Failure { string code = 1; string message = 2; } +enum Channel { + CHANNEL_UNSPECIFIED = 0; + CHANNEL_EMAIL = 1; + CHANNEL_SMS = 2; + CHANNEL_PUSH = 3; + CHANNEL_TELEGRAM = 4; +} +enum DeliveryState { + DELIVERY_STATE_UNSPECIFIED = 0; + DELIVERY_STATE_QUEUED = 1; + DELIVERY_STATE_SENDING = 2; + DELIVERY_STATE_RETRY_WAIT = 3; + DELIVERY_STATE_ACCEPTED = 4; + DELIVERY_STATE_DELIVERED = 5; // Reserved for future verified receipts. + DELIVERY_STATE_FAILED = 6; + DELIVERY_STATE_EXPIRED = 7; + DELIVERY_STATE_UNKNOWN = 8; // May have sent; no automatic replay. +} diff --git a/backend/notifications/main.go b/backend/notifications/main.go new file mode 100644 index 0000000..732365f --- /dev/null +++ b/backend/notifications/main.go @@ -0,0 +1,33 @@ +package main + +import ( + "context" + "flag" + "os" + "os/signal" + "syscall" + + "git.sendico.io/tech/sxchnge/notifications/app" + "go.uber.org/zap" +) + +func main() { + path := flag.String("config.file", "config.yml", "notification configuration") + flag.Parse() + logger, err := zap.NewProduction() + if err != nil { + os.Exit(1) + } + ctx, cancel := signal.NotifyContext(context.Background(), os.Interrupt, syscall.SIGTERM) + defer cancel() + config, err := app.Load(*path) + if err == nil { + err = app.Run(ctx, config, logger) + } + if err != nil { + logger.Error("Notifications service terminated", zap.Error(err)) + _ = logger.Sync() + os.Exit(1) + } + _ = logger.Sync() +} diff --git a/backend/notifications/model/model.go b/backend/notifications/model/model.go new file mode 100644 index 0000000..daf0611 --- /dev/null +++ b/backend/notifications/model/model.go @@ -0,0 +1,58 @@ +package model + +import ( + "git.sendico.io/tech/sxchnge/shared/database/record" + "time" +) + +type State string + +const ( + Queued State = "queued" + Sending State = "sending" + RetryWait State = "retry_wait" + Accepted State = "accepted" + Failed State = "failed" + Expired State = "expired" + Unknown State = "unknown" +) + +type Request struct { + Key string + UserID string + TemplateID string + TemplateVersion string + Parameters map[string]string + Locale string + Address string + ExpiresAt time.Time +} + +// Email is a rendered immutable snapshot, independent of provider wire formats. +type Email struct { + Address string + Subject string + HTML string + Text string +} + +type Notification struct { + storable.Base `bson:",inline"` + Caller string `bson:"caller"` + Key string `bson:"key"` + Fingerprint string `bson:"fingerprint"` + UserID string `bson:"user_id"` + TemplateID string `bson:"template_id"` + TemplateVersion string `bson:"template_version"` + Email Email `bson:"email"` + Driver string `bson:"driver"` + State State `bson:"state"` + Attempts uint32 `bson:"attempts"` + FailureCode string `bson:"failure_code"` + ExpiresAt time.Time `bson:"expires_at"` + NextAttemptAt time.Time `bson:"next_attempt_at"` + LeaseUntil time.Time `bson:"lease_until"` + PurgeAt time.Time `bson:"purge_at"` +} + +func (*Notification) Collection() string { return "notifications" } diff --git a/backend/notifications/service/integration_test.go b/backend/notifications/service/integration_test.go new file mode 100644 index 0000000..402de93 --- /dev/null +++ b/backend/notifications/service/integration_test.go @@ -0,0 +1,209 @@ +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: "

{{.name}}

"}}}}) + 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 != "

A & B

" { + 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) + } +} diff --git a/backend/notifications/service/service.go b/backend/notifications/service/service.go new file mode 100644 index 0000000..f88d710 --- /dev/null +++ b/backend/notifications/service/service.go @@ -0,0 +1,109 @@ +package service + +import ( + "context" + "crypto/sha256" + "encoding/hex" + "encoding/json" + "errors" + "strings" + "time" + + "git.sendico.io/tech/sxchnge/notifications/email" + "git.sendico.io/tech/sxchnge/notifications/model" + "git.sendico.io/tech/sxchnge/notifications/templates" + "git.sendico.io/tech/sxchnge/shared/database/record" + merrors "git.sendico.io/tech/sxchnge/shared/errors" + mlogger "git.sendico.io/tech/sxchnge/shared/logging" + "go.mongodb.org/mongo-driver/v2/bson" + "go.uber.org/zap" +) + +type Repository interface { + Find(context.Context, string, string) (*model.Notification, error) + Get(context.Context, string, string) (*model.Notification, error) + Create(context.Context, *model.Notification) (*model.Notification, error) + Claim(context.Context, string, time.Time) (*model.Notification, error) + Finish(context.Context, *model.Notification) error + Recover(context.Context, time.Time) error +} +type Service struct { + repo Repository + catalog *templates.Catalog + sender email.Sender + driver string + logger mlogger.Logger +} + +func New(repo Repository, catalog *templates.Catalog, sender email.Sender, driver string, logger mlogger.Logger) *Service { + return &Service{repo, catalog, sender, driver, mlogger.Named(logger, "notifications")} +} +func (s *Service) Send(ctx context.Context, caller string, r model.Request) (*model.Notification, error) { + if caller == "" { + return nil, merrors.Unauthorized("missing caller") + } + if r.Key == "" || len(r.Key) > 128 || r.UserID == "" || len(r.UserID) > 128 || !email.ValidAddress(r.Address) || len(r.TemplateID) > 128 || len(r.TemplateVersion) > 64 || len(r.Locale) > 64 || strings.ContainsRune(r.TemplateID+r.TemplateVersion, 0) { + return nil, merrors.InvalidArgument("invalid notification request") + } + if len(r.Parameters) > 64 { + return nil, merrors.InvalidArgument("too many parameters") + } + for k, v := range r.Parameters { + if k == "" || len(k) > 64 || len(v) > 4096 { + return nil, merrors.InvalidArgument("invalid template parameter") + } + } + payload, err := json.Marshal(r) + if err != nil { + return nil, merrors.InvalidArgument("invalid request") + } + if len(payload) > 64<<10 { + return nil, merrors.InvalidArgument("request too large") + } + sum := sha256.Sum256(payload) + fingerprint := hex.EncodeToString(sum[:]) + existing, err := s.repo.Find(ctx, caller, r.Key) + if err == nil { + return same(existing, fingerprint) + } + if !errors.Is(err, merrors.ErrNoData) { + return nil, err + } + now := time.Now().UTC() + expiry := r.ExpiresAt + if expiry.IsZero() { + expiry = now.Add(24 * time.Hour) + } + if !expiry.After(now) || expiry.After(now.Add(24*time.Hour)) { + return nil, merrors.InvalidArgument("expiry must be within the next 24 hours") + } + rendered, err := s.catalog.Render(r) + if err != nil { + return nil, err + } + n := &model.Notification{Base: storable.Base{ID: bson.NewObjectID(), CreatedAt: now, UpdatedAt: now}, Caller: caller, Key: r.Key, Fingerprint: fingerprint, UserID: r.UserID, TemplateID: r.TemplateID, TemplateVersion: r.TemplateVersion, Email: rendered, Driver: s.driver, State: model.Queued, ExpiresAt: expiry, NextAttemptAt: now, PurgeAt: now.Add(7 * 24 * time.Hour)} + stored, err := s.repo.Create(ctx, n) + if err != nil { + return nil, err + } + result, err := same(stored, fingerprint) + if err == nil { + s.logger.Info("Notification accepted", zap.String("notification_id", stored.ID.Hex()), zap.String("caller", caller)) + } + return result, err +} +func same(n *model.Notification, fingerprint string) (*model.Notification, error) { + if n.Fingerprint != fingerprint { + return nil, merrors.DataConflict("idempotency key already used with a different request") + } + return n, nil +} +func (s *Service) Get(ctx context.Context, caller, id string) (*model.Notification, error) { + if caller == "" { + return nil, merrors.Unauthorized("missing caller") + } + if _, err := bson.ObjectIDFromHex(id); err != nil { + return nil, merrors.InvalidArgument("invalid notification ID") + } + return s.repo.Get(ctx, caller, id) +} diff --git a/backend/notifications/service/worker.go b/backend/notifications/service/worker.go new file mode 100644 index 0000000..44fadd9 --- /dev/null +++ b/backend/notifications/service/worker.go @@ -0,0 +1,75 @@ +package service + +import ( + "context" + "errors" + "time" + + "git.sendico.io/tech/sxchnge/notifications/model" + merrors "git.sendico.io/tech/sxchnge/shared/errors" + "go.uber.org/zap" +) + +// Dispatch consumes an opaque job ID from JetStream. Duplicate messages are safe. +func (s *Service) Dispatch(ctx context.Context, id string) (bool, error) { + now := time.Now().UTC() + if err := s.repo.Recover(ctx, now); err != nil { + return false, err + } + n, err := s.repo.Claim(ctx, id, now) + if errors.Is(err, merrors.ErrNoData) { + return false, nil + } + if err != nil { + return false, err + } + switch { + case !n.ExpiresAt.After(now): + n.State = model.Expired + n.FailureCode = "expired" + case n.Driver != s.driver: + n.State = model.Failed + n.FailureCode = "driver_changed" + default: + deadline := now.Add(20 * time.Second) + if n.ExpiresAt.Before(deadline) { + deadline = n.ExpiresAt + } + sendCtx, cancel := context.WithDeadline(ctx, deadline) + outcome := s.sender.Send(sendCtx, n.ID.Hex(), n.Email) + cancel() + n.State = outcome.State + n.FailureCode = outcome.Code + if n.State == model.RetryWait { + delay := time.Duration(1< delay { + delay = outcome.RetryAfter + } + n.NextAttemptAt = time.Now().UTC().Add(delay) + if n.Attempts >= 5 { + n.State = model.Failed + n.FailureCode = "retry_exhausted" + } else if !n.NextAttemptAt.Before(n.ExpiresAt) { + n.State = model.Expired + n.FailureCode = "expired" + } + } else if n.State != model.Accepted && n.State != model.Failed && n.State != model.Unknown { + n.State = model.Unknown + n.FailureCode = "invalid_driver_outcome" + } + } + n.UpdatedAt = time.Now().UTC() + // Persist the result even when shutdown canceled the provider request. + finishCtx, cancel := context.WithTimeout(context.WithoutCancel(ctx), 5*time.Second) + defer cancel() + if err = s.repo.Finish(finishCtx, n); err != nil { + return true, err + } + fields := []zap.Field{zap.String("notification_id", n.ID.Hex()), zap.String("driver", n.Driver), zap.String("state", string(n.State)), zap.Uint32("attempts", n.Attempts), zap.String("failure_code", n.FailureCode)} + if n.State == model.Accepted { + s.logger.Info("Email accepted by provider", fields...) + } else { + s.logger.Warn("Email delivery attempt finished", fields...) + } + return true, nil +} diff --git a/backend/notifications/service/worker_test.go b/backend/notifications/service/worker_test.go new file mode 100644 index 0000000..2c8e084 --- /dev/null +++ b/backend/notifications/service/worker_test.go @@ -0,0 +1,64 @@ +package service + +import ( + "context" + "testing" + "time" + + "git.sendico.io/tech/sxchnge/notifications/email" + "git.sendico.io/tech/sxchnge/notifications/model" + "go.uber.org/zap" +) + +type workerRepo struct { + Repository + job model.Notification + saved *model.Notification +} + +func (r *workerRepo) Recover(context.Context, time.Time) error { return nil } +func (r *workerRepo) Claim(context.Context, string, time.Time) (*model.Notification, error) { + n := r.job + return &n, nil +} +func (r *workerRepo) Finish(_ context.Context, n *model.Notification) error { r.saved = n; return nil } + +type workerSender struct { + calls int + outcome email.Outcome +} + +func (s *workerSender) Send(context.Context, string, model.Email) email.Outcome { + s.calls++ + return s.outcome +} +func TestWorkerTerminalAndRetryOutcomes(t *testing.T) { + for _, tc := range []struct { + name string + expiry time.Time + driver string + attempts uint32 + outcome model.State + want model.State + calls int + }{ + {"expired", time.Now().Add(-time.Minute), "resend", 1, model.Accepted, model.Expired, 0}, + {"driver changed", time.Now().Add(time.Hour), "sendgrid", 1, model.Accepted, model.Failed, 0}, + {"retry exhausted", time.Now().Add(time.Hour), "resend", 5, model.RetryWait, model.Failed, 1}, + {"retry expiry", time.Now().Add(time.Second), "resend", 1, model.RetryWait, model.Expired, 1}, + {"ambiguous", time.Now().Add(time.Hour), "resend", 1, model.Unknown, model.Unknown, 1}, + {"invalid driver outcome", time.Now().Add(time.Hour), "resend", 1, model.Queued, model.Unknown, 1}, + } { + t.Run(tc.name, func(t *testing.T) { + repo := &workerRepo{job: model.Notification{Driver: tc.driver, ExpiresAt: tc.expiry, Attempts: tc.attempts}} + sender := &workerSender{outcome: email.Outcome{State: tc.outcome}} + svc := New(repo, nil, sender, "resend", zap.NewNop()) + if _, err := svc.Dispatch(context.Background(), "id"); err != nil { + t.Fatal(err) + } + if repo.saved == nil || repo.saved.State != tc.want || sender.calls != tc.calls { + t.Fatal(repo.saved, sender.calls) + } + }) + } +} diff --git a/backend/notifications/store/mongo.go b/backend/notifications/store/mongo.go new file mode 100644 index 0000000..de86cfd --- /dev/null +++ b/backend/notifications/store/mongo.go @@ -0,0 +1,118 @@ +package store + +import ( + "context" + "errors" + "time" + + "git.sendico.io/tech/sxchnge/notifications/model" + "git.sendico.io/tech/sxchnge/shared/database/repository" + "git.sendico.io/tech/sxchnge/shared/database/repository/builder" + ri "git.sendico.io/tech/sxchnge/shared/database/repository/index" + "git.sendico.io/tech/sxchnge/shared/database/template" + merrors "git.sendico.io/tech/sxchnge/shared/errors" + mlogger "git.sendico.io/tech/sxchnge/shared/logging" + "go.mongodb.org/mongo-driver/v2/bson" + "go.mongodb.org/mongo-driver/v2/mongo" +) + +type Store struct { + template.DBImp[*model.Notification] +} + +func New(logger mlogger.Logger, db *mongo.Database) (*Store, error) { + s := &Store{*template.Create[*model.Notification](mlogger.Named(logger, "store"), "notifications", db)} + ttl := int32(0) + for _, index := range []*ri.Definition{ + {Name: "unique_caller_key", Keys: []ri.Key{{Field: "caller", Sort: ri.Asc}, {Field: "key", Sort: ri.Asc}}, Unique: true}, + {Name: "pending_dispatch", Keys: []ri.Key{{Field: "state", Sort: ri.Asc}, {Field: "next_attempt_at", Sort: ri.Asc}}}, + {Name: "expired_lease", Keys: []ri.Key{{Field: "state", Sort: ri.Asc}, {Field: "lease_until", Sort: ri.Asc}}}, + {Name: "purge_notifications", Keys: []ri.Key{{Field: "purge_at", Sort: ri.Asc}}, TTL: &ttl}, + } { + if err := s.Repository.CreateIndex(index); err != nil { + return nil, wrap(err) + } + } + return s, nil +} +func wrap(err error) error { + if err == nil { + return nil + } + if errors.Is(err, merrors.ErrNoData) || errors.Is(err, merrors.ErrDataConflict) { + return err + } + return merrors.OpaqueWrap(merrors.ErrServiceUnavailable, err, "notification storage unavailable") +} +func (s *Store) Find(ctx context.Context, caller, key string) (*model.Notification, error) { + var n model.Notification + err := s.FindOne(ctx, repository.Query().And(repository.Filter("caller", caller), repository.Filter("key", key)), &n) + return &n, wrap(err) +} +func (s *Store) Get(ctx context.Context, caller, id string) (*model.Notification, error) { + oid, err := bson.ObjectIDFromHex(id) + if err != nil { + return nil, merrors.InvalidArgument("invalid notification ID") + } + var n model.Notification + err = s.FindOne(ctx, repository.Query().And(repository.Filter("caller", caller), repository.IDFilter(oid)), &n) + return &n, wrap(err) +} +func (s *Store) Create(ctx context.Context, n *model.Notification) (*model.Notification, error) { + err := s.DBImp.Create(ctx, n) + if errors.Is(err, merrors.ErrDataConflict) { + return s.Find(ctx, n.Caller, n.Key) + } + return n, wrap(err) +} +func pending(now time.Time) builder.Query { + return repository.Query().And(repository.Query().In(repository.Field("state"), model.Queued, model.RetryWait), repository.Query().Comparison(repository.Field("next_attempt_at"), builder.Lte, now)) +} + +// A lost worker may already have sent its message. Do not automatically replay it. +func (s *Store) Recover(ctx context.Context, now time.Time) error { + query := repository.Query().And(repository.Filter("state", model.Sending), repository.Query().Comparison(repository.Field("lease_until"), builder.Lte, now)) + patch := repository.Patch().Set(repository.Field("state"), model.Unknown).Set(repository.Field("failure_code"), "worker_outcome_unknown").Set(repository.Field("updatedAt"), now).Set(repository.Field("email"), model.Email{}) + _, err := s.PatchMany(ctx, query, patch) + return wrap(err) +} +func (s *Store) Claim(ctx context.Context, id string, now time.Time) (*model.Notification, error) { + oid, err := bson.ObjectIDFromHex(id) + if err != nil { + return nil, merrors.NoData("invalid job ID") + } + var n model.Notification + query := repository.Query().And(repository.IDFilter(oid), pending(now)) + patch := repository.Patch().Set(repository.Field("state"), model.Sending).Set(repository.Field("updatedAt"), now).Set(repository.Field("lease_until"), now.Add(time.Minute)).Inc(repository.Field("attempts"), 1) + matched, err := s.Repository.PatchOneByFilter(ctx, query, patch, &n) + if err != nil { + return nil, wrap(err) + } + if !matched { + return nil, merrors.NoData("job is not pending") + } + return &n, nil +} +func (s *Store) Finish(ctx context.Context, n *model.Notification) error { + query := repository.Query().And(repository.IDFilter(n.ID), repository.Filter("state", model.Sending), repository.Filter("attempts", n.Attempts), repository.Filter("lease_until", n.LeaseUntil)) + patch := repository.Patch().Set(repository.Field("state"), n.State).Set(repository.Field("failure_code"), n.FailureCode).Set(repository.Field("updatedAt"), n.UpdatedAt).Set(repository.Field("next_attempt_at"), n.NextAttemptAt) + if n.State != model.RetryWait { + patch.Set(repository.Field("email"), model.Email{}) + } + count, err := s.PatchMany(ctx, query, patch) + if err != nil { + return wrap(err) + } + if count != 1 { + return merrors.DataConflict("delivery lease lost") + } + return nil +} + +// Acceptance and outbox work are one document, so no dual-write transaction is needed. +func (s *Store) Due(ctx context.Context, now time.Time) ([]model.Notification, error) { + limit := int64(100) + var jobs []model.Notification + err := s.Repository.FindAllByFilter(ctx, pending(now).Sort(repository.Field("next_attempt_at"), true).Limit(&limit), &jobs) + return jobs, wrap(err) +} diff --git a/backend/notifications/templates.example.json b/backend/notifications/templates.example.json new file mode 100644 index 0000000..0e5922c --- /dev/null +++ b/backend/notifications/templates.example.json @@ -0,0 +1,15 @@ +[ + { + "id": "exchange.completed", + "version": "1", + "default_locale": "en", + "parameters": ["first_name", "amount", "currency"], + "locales": { + "en": { + "subject": "Your exchange is complete", + "html": "

Hi {{.first_name}}, your exchange of {{.amount}} {{.currency}} is complete.

", + "text": "Hi {{.first_name}}, your exchange of {{.amount}} {{.currency}} is complete." + } + } + } +] diff --git a/backend/notifications/templates/catalog.go b/backend/notifications/templates/catalog.go new file mode 100644 index 0000000..c4ddcf8 --- /dev/null +++ b/backend/notifications/templates/catalog.go @@ -0,0 +1,146 @@ +package templates + +import ( + "bytes" + "encoding/json" + html "html/template" + "io" + "os" + "strings" + text "text/template" + + "git.sendico.io/tech/sxchnge/notifications/model" + merrors "git.sendico.io/tech/sxchnge/shared/errors" +) + +type Content struct { + Subject string `json:"subject"` + HTML string `json:"html"` + Text string `json:"text"` +} +type Definition struct { + ID string `json:"id"` + Version string `json:"version"` + DefaultLocale string `json:"default_locale"` + Parameters []string `json:"parameters"` + Locales map[string]Content `json:"locales"` +} +type compiled struct { + definition Definition + locales map[string]renderers +} +type renderers struct { + subject, text *text.Template + html *html.Template +} +type Catalog struct{ entries map[string]compiled } + +func Load(path string) (*Catalog, error) { + f, err := os.Open(path) + if err != nil { + return nil, merrors.OpaqueWrap(merrors.ErrInvalidArg, err, "cannot open template catalog") + } + defer f.Close() + var definitions []Definition + dec := json.NewDecoder(io.LimitReader(f, 4<<20)) + dec.DisallowUnknownFields() + if err = dec.Decode(&definitions); err != nil { + return nil, merrors.OpaqueWrap(merrors.ErrInvalidArg, err, "invalid template catalog") + } + if err = dec.Decode(new(any)); err != io.EOF { + return nil, merrors.InvalidArgument("unexpected template catalog content") + } + return New(definitions) +} +func New(definitions []Definition) (*Catalog, error) { + c := &Catalog{entries: map[string]compiled{}} + if len(definitions) == 0 { + return nil, merrors.InvalidArgument("empty template catalog") + } + for _, d := range definitions { + key := d.ID + "\x00" + d.Version + if d.ID == "" || d.Version == "" || strings.ContainsRune(d.ID+d.Version, 0) { + return nil, merrors.InvalidArgument("invalid template identity") + } + if _, ok := c.entries[key]; ok { + return nil, merrors.InvalidArgument("duplicate template revision") + } + if _, ok := d.Locales[d.DefaultLocale]; !ok || d.DefaultLocale == "" { + return nil, merrors.InvalidArgument("missing default locale") + } + params := map[string]string{} + for _, p := range d.Parameters { + if p == "" { + return nil, merrors.InvalidArgument("empty parameter") + } + if _, ok := params[p]; ok { + return nil, merrors.InvalidArgument("duplicate parameter") + } + params[p] = "sample" + } + entry := compiled{definition: d, locales: map[string]renderers{}} + for locale, content := range d.Locales { + s, err := text.New("subject").Option("missingkey=error").Parse(content.Subject) + if err != nil { + return nil, merrors.InvalidArgument("invalid subject template") + } + t, err := text.New("text").Option("missingkey=error").Parse(content.Text) + if err != nil { + return nil, merrors.InvalidArgument("invalid text template") + } + h, err := html.New("html").Option("missingkey=error").Parse(content.HTML) + if err != nil { + return nil, merrors.InvalidArgument("invalid HTML template") + } + entry.locales[locale] = renderers{s, t, h} + } + c.entries[key] = entry + for locale := range d.Locales { + if _, err := c.Render(model.Request{TemplateID: d.ID, TemplateVersion: d.Version, Locale: locale, Parameters: params}); err != nil { + return nil, err + } + } + } + return c, nil +} + +type boundedBuffer struct{ bytes.Buffer } + +func (b *boundedBuffer) Write(p []byte) (int, error) { + if b.Len()+len(p) > 256<<10 { + return 0, merrors.InvalidArgument("rendered content too large") + } + return b.Buffer.Write(p) +} +func (c *Catalog) Render(r model.Request) (model.Email, error) { + entry, ok := c.entries[r.TemplateID+"\x00"+r.TemplateVersion] + if !ok { + return model.Email{}, merrors.NoData("template revision not found") + } + if len(r.Parameters) != len(entry.definition.Parameters) { + return model.Email{}, merrors.InvalidArgument("template parameters do not match") + } + for _, p := range entry.definition.Parameters { + if _, ok := r.Parameters[p]; !ok { + return model.Email{}, merrors.InvalidArgument("missing template parameter") + } + } + locale := r.Locale + if locale == "" { + locale = entry.definition.DefaultLocale + } + render, ok := entry.locales[locale] + if !ok { + return model.Email{}, merrors.InvalidArgument("unsupported template locale") + } + var subject, body, plain boundedBuffer + for _, run := range []func() error{func() error { return render.subject.Execute(&subject, r.Parameters) }, func() error { return render.html.Execute(&body, r.Parameters) }, func() error { return render.text.Execute(&plain, r.Parameters) }} { + if err := run(); err != nil { + return model.Email{}, merrors.OpaqueWrap(merrors.ErrInvalidArg, err, "template rendering failed") + } + } + if subject.Len() == 0 || subject.Len() > 998 || strings.ContainsAny(subject.String(), "\r\n") || (body.Len() == 0 && plain.Len() == 0) { + return model.Email{}, merrors.InvalidArgument("invalid rendered email") + } + return model.Email{Address: r.Address, Subject: subject.String(), HTML: body.String(), Text: plain.String()}, nil +} diff --git a/backend/notifications/templates/catalog_test.go b/backend/notifications/templates/catalog_test.go new file mode 100644 index 0000000..9d033b7 --- /dev/null +++ b/backend/notifications/templates/catalog_test.go @@ -0,0 +1,42 @@ +package templates + +import ( + "strings" + "testing" + + "git.sendico.io/tech/sxchnge/notifications/model" +) + +func TestRenderAndValidate(t *testing.T) { + c, err := New([]Definition{{ID: "test", Version: "1", DefaultLocale: "en", Parameters: []string{"name"}, Locales: map[string]Content{"en": {Subject: "Hello {{.name}}", HTML: "

{{.name}}

", Text: "{{.name}}"}}}}) + if err != nil { + t.Fatal(err) + } + r := model.Request{TemplateID: "test", TemplateVersion: "1", Parameters: map[string]string{"name": ""}} + result, err := c.Render(r) + if err != nil { + t.Fatal(err) + } + if strings.Contains(result.HTML, "